项目介绍
MCPToolKit - 生产就绪的MCP服务器框架
我们解决的问题
1. 生产环境中扩展MCP服务器
标准的FastMCP框架在生产环境中面临重大挑战：
- 状态管理：传统的FastMCP服务器在内存中维护状态，使得水平扩展变得困难
- 无服务器限制：无服务器环境需要无状态架构，而FastMCP并未为此设计
- 多租户支持：在同一服务器上运行多个租户需要复杂的会话管理
2. 委托OAuth支持
管理MCP服务器的身份验证非常复杂：
- 工具级认证：用户应在工具需要时才进行身份验证
- 第三方集成：支持像Notion、Slack等服务的OAuth需要复杂的令牌管理
- 安全性：同时管理多个认证流程并保持安全具有挑战性
我们的解决方案
MCPToolKit提供了一个生产就绪的框架，解决了这些问题，同时完全兼容FastMCP。以下是其工作原理：
graph TD
A[LLM客户端<br/>例如Claude, ChatGPT, Cursor] --> B[负载均衡器]
B --> C[MCP服务器实例1<br/>带有Redis状态]
B --> D[MCP服务器实例2<br/>带有Redis状态]
B --> E[MCP服务器实例N<br/>带有Redis状态]
C --> F[Redis<br/>会话状态及OAuth令牌]
D --> F
E --> F
C --> H[MCP授权服务器<br/>OAuth 2.1 & PKCE]
D --> H
E --> H
H --> G[OAuth提供商<br/>Notion, Slack等]
style H fill:#f9f,stroke:#333,stroke-width:2px
此架构图展示了MCPToolKit如何启用生产就绪的MCP服务器：
- LLM客户端（如Claude, ChatGPT, Cursor）发起对MCP服务器的请求。这些客户端可以是任何需要与MCP工具交互的应用程序。
- 负载均衡器将传入的请求分配给多个MCP服务器实例，实现水平扩展和高可用性。
- MCP服务器实例（从1到N）处理工具执行和资源访问。每个实例：
- 维护自己的Redis状态以持久化会话
- 可独立处理请求
- 共享相同的代码库和配置
- 根据需求进行水平扩展
- Redis作为中央状态存储，提供：
- 跨服务器重启的会话状态持久化
- OAuth令牌存储和管理
- 服务器实例之间的共享状态
- 实现无状态服务器实例
- MCP授权服务器（用粉色突出显示）管理所有与OAuth相关的操作：
- 实现OAuth 2.1与PKCE用于安全认证
- 处理令牌发放和刷新
- 管理同意流程
- 集中管理所有服务器实例的OAuth逻辑
- OAuth提供商（如Notion, Slack）是用户可以进行身份验证的第三方服务。授权服务器安全地管理这些连接。
这种架构实现了：
- 通过无状态服务器实例实现真正的水平扩展
- 集中的OAuth管理
- 通过多个服务器实例实现高可用性
- 安全的令牌管理
- 跨会话的一致用户体验
从FastMCP迁移
从FastMCP迁移到MCPToolKit非常简单。以下是更新现有FastMCP服务器的方法：
# 之前（FastMCP）
- from mcp.server.fastmcp import FastMCP
-
- # 创建一个MCP服务器
- mcp = FastMCP("Demo")
# 之后（MCPToolKit）
+ from mcptoolkit import MCPToolKit
+ import os
+
+ # 创建一个生产就绪的MCP服务器
+ mcp = MCPToolKit(
+ name="Demo",
+ redis_url=os.environ["REDIS_URL"] # 必需：在您的环境中设置REDIS_URL
+ )
# 您的工具和资源保持完全相同
@mcp.tool()
def add(a: int, b: int) -> int:
"""加两个数"""
return a + b
@mcp.resource("greeting://{name}")
def get_greeting(name: str) -> str:
"""获取个性化的问候语"""
return f"你好，{name}！"
迁移只需要几个简单的更改：
- 更改导入语句
- 设置
REDIS_URL环境变量（生产必需） - 就这样！您现有的工具、资源和提示将继续按以前的方式工作
对于本地开发，您可以设置环境变量：
export REDIS_URL="redis://localhost:6379/0"
对于无服务器部署，您还需要更新部署配置：
# 之前（FastMCP）
- # api/index.py
- from mcp.server.fastmcp import FastMCP
-
- mcp = FastMCP("Demo")
- app = mcp.create_fastapi_app()
# 之后（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"] # 必需：在您的环境中设置REDIS_URL
+ )
关键特性：
- 基于Redis的状态：会话状态跨服务器重启和函数调用持久化
- 无服务器就绪：专为Vercel、AWS Lambda和其他无服务器平台设计
- 水平扩展：状态持久化实现真正的水平扩展
- 多租户支持：多个用户可以连接到同一端点，并具有隔离的会话
2. 委托OAuth支持
from mcptoolkit import MCPToolKit, requires_auth
from mcptoolkit.auth.providers import NotionProvider, SlackProvider
# 定义每个提供商的默认范围
default_notion_scopes = [
"read:database",
"write:page",
"read:page"
]
default_slack_scopes = [
"channels:read",
"chat:write",
"reactions:write"
]
server = MCPToolKit(name="授权服务器")
@server.tool()
@requires_auth(provider=NotionProvider(
scopes=default_notion_scopes,
consent_required=True # 需要明确的用户同意
))
def notion_search(query: str, ctx: Context) -> str:
# 访问具有特定范围的已认证Notion客户端
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:
# 访问具有特定范围的已认证Slack客户端
slack = ctx.get_oauth_client("slack")
return slack.post_message(channel, message)
关键特性：
- 延迟认证：用户仅在工具需要时进行认证
- 提供商支持：内置支持常见提供商（Notion, Slack等）
- 令牌管理：自动令牌刷新和存储
- 安全性：安全的令牌存储和传输
- 细粒度范围：对OAuth权限的精细控制
- 同意管理：用户友好的同意屏幕，逻辑权限分组
- 人机循环：高风险操作的可选审批要求
OAuth流程
MCPToolKit实现了安全的OAuth 2.1流程与PKCE：
- 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 # 需要明确的人工批准
))
def delete_database(database_id: str, ctx: Context) -> str:
# 此操作需要明确的人工批准
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="无服务器MCP",
redis_url=os.environ.get("REDIS_URL")
)
快速开始
- 安装MCPToolKit：
pip install mcp-python-sdk
- 创建您的服务器：
from mcptoolkit import MCPToolKit, requires_auth
server = MCPToolKit(
name="我的生产服务器",
redis_url="redis://localhost:6379/0"
)
@server.tool()
def public_tool() -> str:
return "此工具不需要认证"
@server.tool()
@requires_auth(provider="notion")
def notion_tool() -> str:
return "此工具需要Notion认证"
- 部署到您首选的平台（Kubernetes, Vercel等）
要求
- Python 3.9+
- Redis实例（用于会话状态持久化）
- OAuth提供商凭据（如果使用委托认证）
许可证
同MCP Python SDK。