MCPToolKit - 生产就绪的 MCP 服务器框架

我们解决的问题

1. 在生产环境中扩展 MCP 服务器

标准的 FastMCP 框架在生产环境中面临重大挑战：

• 状态管理：传统的 FastMCP 服务器将状态保存在内存中，这使得水平扩展变得困难
• 无服务器环境限制：无服务器环境要求无状态架构，而 FastMCP 并不是为此设计的
• 多租户支持：在同一服务器上运行多个租户需要复杂的会话管理

2. 委托 OAuth 支持

为 MCP 服务器管理身份验证是复杂的：

• 工具级认证：用户只有在工具需要时才应进行身份验证
• 第三方集成：支持 Notion、Slack 等服务的 OAuth 需要复杂的令牌管理
• 安全性：在维护安全性的同时管理多种身份验证流程是具有挑战性的

我们的解决方案

MCPToolKit 提供了一个生产就绪的框架，解决了这些问题，同时保持与 FastMCP 的完全兼容。其工作原理如下：

graph TD
    A[LLM Client<br/>e.g. Claude, ChatGPT, Cursor] --> B[Load Balancer]
    B --> C[MCP Server Instance 1<br/>with Redis State]
    B --> D[MCP Server Instance 2<br/>with Redis State]
    B --> E[MCP Server Instance N<br/>with Redis State]
    C --> F[Redis<br/>Session State & OAuth Tokens]
    D --> F
    E --> F
    C --> H[MCP Authorization Server<br/>OAuth 2.1 & PKCE]
    D --> H
    E --> H
    H --> G[OAuth Providers<br/>Notion, Slack, etc.]
    style H fill:#f9f,stroke:#333,stroke-width:2px

该架构图说明了 MCPToolKit 如何实现生产就绪的 MCP 服务器：

1. LLM 客户端（例如 Claude、ChatGPT、Cursor）向 MCP 服务器发起请求。这些客户端可以是任何需要与 MCP 工具交互的应用程序。

2. 负载均衡器将传入请求分发到多个 MCP 服务器实例，从而实现水平扩展和高可用性。

3. MCP 服务器实例（1 到 N）处理工具执行和资源访问。每个实例：

   • 维护自己的 Redis 状态以实现会话持久化
   • 可以独立处理请求
   • 共享相同的代码库和配置
   • 可以根据需求进行水平扩展

4. Redis作为中央状态存储，提供：

   • 跨服务器重启的会话状态持久化
   • OAuth 令牌存储和管理
   • 服务器实例之间的共享状态
   • 使服务器实例无状态

5. MCP 授权服务器（用粉红色突出显示）管理所有与 OAuth 相关的操作：

   • 实现带 PKCE 的 OAuth 2.1 以确保安全认证
   • 处理令牌发放和刷新
   • 管理同意流程
   • 为所有服务器实例集中管理 OAuth 逻辑

6. 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}!"

迁移只需要进行一些简单的更改：

1. 更改导入语句
2. 设置 REDIS_URL 环境变量（生产环境中必需）
3. 就这样！您现有的所有工具、资源和提示将继续像以前一样工作

对于本地开发，您可以设置环境变量：

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 流程：

1. LLM 客户端向 MCP 服务器发起请求
2. 服务器响应 401 未授权，并返回重定向链接
3. 用户登录到 OAuth 提供商并授予请求的权限范围
4. 服务器将授权码返回给客户端
5. 客户端使用授权码交换访问令牌和刷新令牌
6. 使用这些令牌进行后续请求
7. MCP 服务器调用第三方服务

授权服务器架构

MCPToolKit 支持两种授权服务器的部署模式：

1. 嵌入式授权服务器

   • MCP 服务器同时充当身份提供者和依赖方
   • 直接处理登录、同意和令牌发放
   • 管理令牌生命周期、刷新逻辑和撤销
   • 最适合独立应用程序

2. 外部授权服务器

   • MCP 服务器作为依赖方
   • 将 OAuth 流程委托给外部服务（例如 Stytch）
   • 专注于工具级别的访问控制
   • 最适合与现有身份基础设施集成

两种模式都支持：

• 带有 PKCE 的 OAuth 2.1
• 动态客户端注册
• 授权服务器元数据 (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")
)

快速开始

1. 安装 MCPToolKit：

pip install mcp-python-sdk

1. 创建您的服务器：

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"

1. 部署到您选择的平台（Kubernetes、Vercel 等）

要求

• Python 3.9+
• Redis 实例（用于会话状态持久化）
• OAuth 提供者凭据（如果使用委托认证）

许可证

与 MCP Python SDK 相同。