MCPToolKit-生产就绪的MCP服务器框架
我们解决的问题
1.在生产中扩展MCP服务器
标准FastMCP框架在生产环境中面临着重大挑战:
- 状态管理:传统的FastMCP服务器在内存中维护状态,这使得横向扩展变得困难
- 无服务器限制:无服务器环境需要无状态架构,而FastMCP不是为此而设计的
- 多租户支持:在同一服务器上运行多个租户需要复杂的会话管理
2.授权OAuth支持
管理MCP服务器的身份验证很复杂:
- 工具级身份验证:用户只应在工具需要时进行身份验证
- 第三方集成:为Notion、Slack等服务支持OAuth需要复杂的令牌管理
- 安全:在维护安全性的同时管理多个身份验证流具有挑战性
我们的解决方案
MCPToolKit提供了一个生产就绪的框架,可以解决这些问题,同时保持与FastMCP的完全兼容性。其工作原理如下:
graph TD
A[LLM Client
e.g. Claude, ChatGPT, Cursor] --> B[Load Balancer]
B --> C[MCP Server Instance 1
with Redis State]
B --> D[MCP Server Instance 2
with Redis State]
B --> E[MCP Server Instance N
with Redis State]
C --> F[Redis
Session State & OAuth Tokens]
D --> F
E --> F
C --> H[MCP Authorization Server
OAuth 2.1 & PKCE]
D --> H
E --> H
H --> G[OAuth Providers
Notion, Slack, etc.]
style H fill:#f9f,stroke:#333,stroke-width:2px此架构图说明了MCPToolKit如何启用生产就绪的MCP服务器:
- LLM客户 (例如,Claude、ChatGPT、Cursor)向MCP服务器发起请求。这些客户端可以是需要与MCP工具交互的任何应用程序。
- 负载均衡器 跨多个MCP服务器实例分发传入请求,实现横向扩展和高可用性。
- MCP服务器实例 (1到N)处理工具执行和资源访问。每个实例:
- 维护自己的Redis状态以实现会话持久性 - 可以独立处理请求 - 共享相同的代码库和配置 - 可根据需求水平扩展
- 雷迪斯 作为中央州立商店,提供:
- 跨服务器重启的会话状态持久性 - OAuth令牌存储和管理 - 服务器实例之间的共享状态 - 启用无状态服务器实例
- MCP授权服务器 (以粉红色突出显示)管理所有与OAuth相关的操作:
- 使用PKCE实现OAuth 2.1以实现安全身份验证 - 处理代币发行和刷新 - 管理同意流程 - 为所有服务器实例集中OAuth逻辑
- OAuth提供者 (例如,Notion、Slack)是用户可以进行身份验证的第三方服务。授权服务器安全地管理这些连接。
该架构实现了:
- 通过无状态服务器实例实现真正的横向扩展
- 集中式OAuth管理
- 通过多个服务器实例实现高可用性
- 安全令牌管理
- 跨会话的一致用户体验
从FastMCP迁移
从FastMCP迁移到MCPToolKit很简单。以下是如何更新现有的FastMCP服务器:
# Before (FastMCP)
- from mcp.server.fastmcp import FastMCP
-
- # Create an MCP server
- mcp = FastMCP("Demo")
# After (MCPToolKit)
+ from mcptoolkit import MCPToolKit
+ import os
+
+ # Create a production-ready MCP server
+ mcp = MCPToolKit(
+ name="Demo",
+ redis_url=os.environ["REDIS_URL"] # Required: Set REDIS_URL in your environment
+ )
# Your tools and resources remain exactly the same
@mcp.tool()
def add(a: int, b: int) -> int:
"""Add two numbers"""
return a + b
@mcp.resource("greeting://{name}")
def get_greeting(name: str) -> str:
"""Get a personalized greeting"""
return f"Hello, {name}!"迁移只需要几个简单的更改:
- 更改导入语句
- 设置
REDIS_URL环境变量(生产所需) - 就是这样!您现有的所有工具、资源和提示将继续与以前完全一样工作
对于本地开发,您可以设置环境变量:
export REDIS_URL="redis://localhost:6379/0"对于无服务器部署,您还需要更新部署配置:
# Before (FastMCP)
- # api/index.py
- from mcp.server.fastmcp import FastMCP
-
- mcp = FastMCP("Demo")
- app = mcp.create_fastapi_app()
# After (MCPToolKit)
+ # api/index.py
+ from mcptoolkit.vercel import create_vercel_app
+ import os
+
+ app = create_vercel_app(
+ name="Demo",
+ redis_url=os.environ["REDIS_URL"] # Required: Set REDIS_URL in your environment
+ )主要特点:
- Redis支持状态:会话状态在服务器重启和函数调用过程中持续存在
- 无服务器就绪:专为Vercel、AWS Lambda和其他无服务器平台设计
- 水平缩放:状态持久性实现了真正的水平扩展
- 多租户支持:多个用户可以通过隔离会话连接到同一端点
2.授权OAuth支持
from mcptoolkit import MCPToolKit, requires_auth
from mcptoolkit.auth.providers import NotionProvider, SlackProvider
# Define default scopes for each provider
default_notion_scopes = [
"read:database",
"write:page",
"read:page"
]
default_slack_scopes = [
"channels:read",
"chat:write",
"reactions:write"
]
server = MCPToolKit(name="Auth Server")
@server.tool()
@requires_auth(provider=NotionProvider(
scopes=default_notion_scopes,
consent_required=True # Require explicit user consent
))
def notion_search(query: str, ctx: Context) -> str:
# Access authenticated Notion client with specific scopes
notion = ctx.get_oauth_client("notion")
return notion.search(query)
@server.tool()
@requires_auth(provider=SlackProvider(
scopes=default_slack_scopes,
consent_required=True
))
def slack_message(channel: str, message: str, ctx: Context) -> str:
# Access authenticated Slack client with specific scopes
slack = ctx.get_oauth_client("slack")
return slack.post_message(channel, message)主要特点:
- 延迟身份验证:用户仅在工具需要时进行身份验证
- 提供商支持:内置对常见提供商(Notion、Slack等)的支持
- 许可证管理:自动令牌刷新和存储
- 安全:安全的令牌存储和传输
- 粒度范围:对OAuth权限进行细粒度控制
- 同意管理:具有逻辑权限分组的用户友好同意屏幕
- 循环中的人类:高风险行动的可选批准要求
OAuth流
MCPToolKit使用PKCE实现了安全的OAuth 2.1流:
- LLM客户端向MCP服务器发起请求
- 服务器返回401未经授权和重定向链接
- 用户登录OAuth提供者并授予请求的作用域
- 服务器向客户端返回身份验证码
- 客户端交换访问+刷新令牌的代码
- 令牌用于后续请求
- MCP服务器调用第三方服务
授权服务器架构
MCPToolKit支持授权服务器的两种部署模型:
- 嵌入式授权服务器
- MCP服务器同时充当身份提供者和依赖方 - 直接处理登录、同意和令牌发放 - 管理令牌生命周期、刷新逻辑和吊销 - 最适合独立应用程序
- 外部授权服务器
- MCP服务器充当依赖方 - 将OAuth流委托给外部服务(例如Stytch) - 专注于工具级访问控制 - 最适合与现有身份基础设施集成
两种型号都支持:
- OAuth 2.1 与 PKCE
- 动态客户端注册
- 授权服务器元数据(RFC 8414)
- 基于资源/操作的自定义范围
- 最终用户同意管理
- 每个提供者的粒度范围定义
- 组织级可见性和控制
- 隐含权限(用户只能授予他们拥有的权限)
同意和访问管理
MCPToolKit提供全面的同意和访问管理:
- 组织级可见性:查看整个组织中授权的所有已连接应用程序
- 精细权限:查看哪些成员已授予访问权限以及他们已授权的范围
- 访问管理:随时撤销特定用户或应用程序的访问权限
- 用户友好同意:在逻辑分组中显示RBAC权限
- 隐含权限:用户只能给应用程序相同的权限
- 循环中的人类:高风险行动需要人工批准
高风险行动保护
@server.tool()
@requires_auth(provider=NotionProvider(
scopes=["delete:database"],
human_approval_required=True # Require explicit human approval
))
def delete_database(database_id: str, ctx: Context) -> str:
# This action will require explicit human approval
notion = ctx.get_oauth_client("notion")
return notion.delete_database(database_id)建筑
部署选项
Kubernetes
apiVersion: apps/v1
kind: Deployment
metadata:
name: mcp-server
spec:
replicas: 3
template:
spec:
containers:
- name: mcp-server
image: your-mcp-server
env:
- name: REDIS_URL
valueFrom:
secretKeyRef:
name: redis-credentials
key: url无服务器(Vercel)
# api/index.py
from mcptoolkit.vercel import create_vercel_app
app = create_vercel_app(
name="Serverless MCP",
redis_url=os.environ.get("REDIS_URL")
)快速开始
- 安装MCPToolKit:
pip install mcp-python-sdk- 创建您的服务器:
from mcptoolkit import MCPToolKit, requires_auth
server = MCPToolKit(
name="My Production Server",
redis_url="redis://localhost:6379/0"
)
@server.tool()
def public_tool() -> str:
return "This tool doesn't require auth"
@server.tool()
@requires_auth(provider="notion")
def notion_tool() -> str:
return "This tool requires Notion auth"- 部署到您喜欢的平台(Kubernetes、Vercel等)
需求
- Python 3.9+
- Redis实例(用于会话状态持久化)
- OAuth提供者凭据(如果使用委托身份验证)
许可证
与MCP Python SDK相同。
