MCP Sso Tutorial

具有Django JWT身份验证和OAuth2的MCP服务器

这是一个基于FastAPI的MCP（模型上下文协议）服务器，它使用Django后端的JWT令牌以及Allauth和SimpleJWT提供安全身份验证。它支持 OAuth2动态客户端注册 使用自动Google OAuth流。

建筑

┌───────────────┐    OAuth2    ┌───────────────┐   Google    ┌───────────────┐           ┌───────────────┐
│ MCP Client    │────────────▶│ MCP Server    │───────────▶│ Django Backend│ ────────▶│ Google OAuth  │
│ (e.g. GPT)    │ Register+    │ (FastAPI)     │ Redirect    │ (Allauth+DRF) │ OAuth     │ Provider      │
│               │ Authorize    │               │             │               │           │               │
└───────────────┘              └───────────────┘             └───────────────┘           └───────────────┘
        │                             │                             │                         │
        │         JWT Access Token    │      JWT Token Exchange     │    User Info + Session  │
        └─────────────────────────────┼─────────────────────────────┼─────────────────────────┘
                                      │                             │
                                      └─────────────────────────────┘
                                             API Calls with JWT

特性

• OAuth2动态客户端注册：MCP客户端的自动客户端注册
• Google OAuth集成：通过Django的Google OAuth流进行无缝身份验证
• JWT身份验证：使用RS256验证Django后端发出的令牌
• JWKS支持：自动从Django的JWKS端点获取和缓存公钥
• 用户上下文：将经过身份验证的用户信息附加到请求中
• MCP工具：提供对Django后端API的身份验证访问
• 缓存：具有可配置TTL的高效JWKS密钥缓存
• 错误处理：正确的OAuth2错误响应
• 遵从标准：支持PKCE的完整OAuth2授权代码流

设置

1.安装依赖项

cd mcp-auth-test
pip install -e .
# or using uv
uv pip install -e .

2.配置环境

复制示例环境文件并对其进行配置：

cp .env.example .env

编辑 .env 使用Django后端配置：

# Django Backend Configuration
DJANGO_BASE_URL=http://localhost:8000

# JWT Configuration (these should match your Django settings)
JWT_AUDIENCE=cccrm-api
JWT_ISSUER=cccrm-backend

# Optional: Provide static public key (otherwise JWKS will be used)
# JWT_PUBLIC_KEY=-----BEGIN PUBLIC KEY-----\n...\n-----END PUBLIC KEY-----

# Server Configuration
PORT=10000

# Optional: Tavily API for web search tool
TAVILY_API_KEY=your-tavily-api-key

3.Django后端设置

确保你的Django后端在 settings.py:

# JWT Configuration
SIMPLE_JWT = {
    'ALGORITHM': 'RS256',
    'SIGNING_KEY': env('JWT_PRIVATE_KEY'),
    'VERIFYING_KEY': env('JWT_PUBLIC_KEY'),
    'AUDIENCE': env('JWT_AUDIENCE', default='cccrm-api'),
    'ISSUER': env('JWT_ISSUER', default='cccrm-backend'),
    # ... other settings
}

# Expose JWKS endpoint
# Add this URL pattern to expose /.well-known/jwks.json

4.生成JWT密钥（如果需要）

如果你需要为Django后端生成RS256密钥对：

# generate_jwt_keys.py
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.primitives import serialization

# Generate private key
private_key = rsa.generate_private_key(
    public_exponent=65537,
    key_size=2048
)

# Get public key
public_key = private_key.public_key()

## OAuth2 Dynamic Client Registration

The MCP server supports OAuth2 Dynamic Client Registration, allowing clients to automatically register and authenticate users via Google OAuth through Django.

### OAuth2 Flow

1. 客户注册

MCP客户端──────────POST/oauth2/注册──────────▶ MCP服务器 │ ▼ 生成client_id/secret │ ▼ MCP客户端◀──────────client_id+客户端密码───────┘

1. 授权请求\

MCP客户端──────────GET/oauth2/授权─────────▶ MCP服务器 │ ▼ 重定向到Django │ ▼ 用户浏览器────────/oauth/谷歌/？next=回调──▶ Django后端 │ ▼ 谷歌OAuth │ ▼ 用户浏览器◀谷歌 │ ▼ 用户身份验证 │ ▼ 用户浏览器─────────使用身份验证码进行回调──────▶ Django后端 │ ▼ Django回调 │ ▼ 用户浏览器─────────重定向到MCP回调─────▶ MCP服务器 │ ▼ 生成身份验证码 │ ▼ MCP客户端◀─────────授权代码─────────────── MCP服务器

1. 代币交换

MCP客户端──────────POST/oauth2/令牌────────────▶ MCP服务器 （身份验证码+客户端凭据）│ ▼ 为JWT调用Django │ ▼ MCP客户端◀─────────JWT访问令牌──────────────── MCP服务器

### OAuth2 Endpoints

#### 1. Dynamic Client Registration

POST /oauth2/register Content-Type: application/json

{ "client_name": "My MCP Client", "client_uri": "https://example.com", "redirect_uris": ["https://example.com/callback"], "grant_types": ["authorization_code"], "response_types": ["code"], "scope": "openid profile email" }

**答复：**

{ "client_id": "mcp_client_abc123...", "client_secret": "secret_xyz789...", "client_name": "My MCP Client", "redirect_uris": ["https://example.com/callback"], "grant_types": ["authorization_code"], "response_types": ["code"], "scope": "openid profile email", "client_id_issued_at": 1699123456, "client_secret_expires_at": 0 }

#### 2.授权请求

GET /oauth2/authorize?client_id={client_id}&redirect_uri={redirect_uri}&response_type=code&scope=openid%20profile%20email&state={state}&code_challenge={challenge}&code_challenge_method=S256

这将把用户重定向到：

Django: /oauth/google/?next=/mcp/oauth2/callback/?mcp_callback_params=...

#### 3.代币兑换

POST /oauth2/token Content-Type: application/x-www-form-urlencoded

grant_type=authorization_code& client_id={client_id}& client_secret={client_secret}& code={authorization_code}& redirect_uri={redirect_uri}& code_verifier={verifier}

**答复：**

{ "access_token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJSUzI1NiJ9...", "token_type": "Bearer", "expires_in": 3600, "scope": "openid profile email", "refresh_token": "refresh_token_here..." }

### OAuth2发现

#### 授权服务器元数据

GET /.well-known/oauth-authorization-server

**答复：**

{ "issuer": "http://localhost:10000", "authorization_endpoint": "http://localhost:10000/oauth2/authorize", "token_endpoint": "http://localhost:10000/oauth2/token", "registration_endpoint": "http://localhost:10000/oauth2/register", "jwks_uri": "http://localhost:8000/auth/.well-known/jwks.json", "response_types_supported": ["code"], "grant_types_supported": ["authorization_code"], "token_endpoint_auth_methods_supported": ["client_secret_post"], "scopes_supported": ["openid", "profile", "email"], "code_challenge_methods_supported": ["S256", "plain"], "subject_types_supported": ["public"] }

#### 资源服务器元数据

GET /.well-known/oauth-protected-resource/mcp

**答复：**

{ "resource_server_metadata": { "resource_identifier": "cccrm-api", "authorization_servers": ["http://localhost:10000"], "authorization_server_metadata_url": "http://localhost:10000/.well-known/oauth-authorization-server", "jwks_uri": "http://localhost:8000/auth/.well-known/jwks.json", "audience": "cccrm-api", "issuer": "cccrm-backend", "oauth2_dynamic_registration": true, "registration_endpoint": "http://localhost:10000/oauth2/register" } }

# 序列化私钥

private_pem=private_key.private_bytes(
编码=序列化。编码。PEM，
格式=序列化。私人格式。PKCS8，
加密算法=序列化。无加密（）
).decode（）

# 序列化公钥

public_pem=public_key.public_bytes(
编码=序列化。编码。PEM，
格式=序列化。公共格式。主题公钥信息
).decode（）

print（“私钥：”）
打印（private_pem）
print（“\\n公钥：”）
打印（public_pem）

Running the Server

# Using the entry point
mcp-server

# Or directly
python -m src.server

# Or with uvicorn
uvicorn src.server:app --host localhost --port 10000 --reload

认证流程

1.用户身份验证

用户首先通过Django后端进行身份验证：

# Login to get JWT token
curl -X POST http://localhost:8000/auth/login/ \
  -H "Content-Type: application/json" \
  -d '{"email": "user@example.com", "password": "password"}'

答复：

{
  "access_token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJSUzI1NiJ9...",
  "refresh_token": "...",
  "user": {
    "id": 1,
    "email": "user@example.com"
  }
}

2.MCP客户端连接

MCP客户端使用JWT令牌连接到服务器：

# Example MCP request
curl -X POST http://localhost:10000/ \
  -H "Authorization: Bearer YOUR_JWT_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "id": 1,
    "method": "tools/call",
    "params": {
      "name": "get_user_profile",
      "arguments": {}
    }
  }'

可用工具

1. get_user_profile

获取当前已验证用户的个人资料信息。

{
  "method": "tools/call",
  "params": {
    "name": "get_user_profile",
    "arguments": {}
  }
}

2. list_campaigns

列出已验证用户的所有活动。

{
  "method": "tools/call",
  "params": {
    "name": "list_campaigns",
    "arguments": {}
  }
}

3. create_campaign

为经过身份验证的用户创建新活动。

{
  "method": "tools/call",
  "params": {
    "name": "create_campaign",
    "arguments": {
      "name": "My New Campaign",
      "description": "Campaign description"
    }
  }
}

4. get_leads_count

获取已验证用户的潜在客户总数。

{
  "method": "tools/call",
  "params": {
    "name": "get_leads_count",
    "arguments": {}
  }
}

5. search_web （可选）

使用Tavilly API搜索网页（需要Tavily_API_KEY）。

{
  "method": "tools/call",
  "params": {
    "name": "search_web",
    "arguments": {
      "query": "latest news about AI"
    }
  }
}

6. add_numbers

简单的演示工具，用于添加两个数字。

{
  "method": "tools/call",
  "params": {
    "name": "add_numbers",
    "arguments": {
      "a": 5,
      "b": 3
    }
  }
}

安全特性

JWT验证

• RS256算法：使用非对称加密技术进行安全令牌验证
• JWKS集成：自动从Django的JWKS端点获取公钥
• 索赔验证：验证发行人、受众和到期声明
• 缓存：缓存具有可配置TTL的JWKS密钥以提高性能

请求上下文

中间件将用户信息附加到每个请求中：

request.state.user = {
    "id": payload.get("user_id"),
    "email": payload.get("email"),
    "username": payload.get("username"),
    "token_type": payload.get("token_type"),
    "exp": payload.get("exp"),
    "iat": payload.get("iat"),
    "jti": payload.get("jti"),
}

错误处理

服务器返回正确的OAuth2错误响应：

{
  "error": "invalid_token",
  "error_description": "Token has expired"
}

常见错误类型：

• invalid_token：令牌格式错误、已过期或无效
• insufficient_scope：令牌没有所需的权限

发展

添加新工具

1. 将您的工具功能添加到 src/tools.py:

@mcp.tool
async def my_new_tool(request: Request, param1: str, param2: int = 0) -> Dict:
    """Description of what this tool does"""
    if not hasattr(request.state, 'user') or not request.state.user:
        raise Exception("User not authenticated")
    
    # Your tool logic here
    return {"result": "success"}

1. 工具可以自动访问：

- request.state.user：经过身份验证的用户信息 - request.headers.get("Authorization")：后端请求的原始JWT令牌

测试

# Install test dependencies
pip install pytest httpx pytest-asyncio

# Run tests (when available)
pytest tests/

配置参考

故障排除

常见问题

1. “无法获取JWKS密钥”

- 检查一下 DJANGO_BASE_URL 是正确的 - 确保Django后端正在运行且可访问 - 验证JWKS端点是否可用 /auth/.well-known/jwks.json

1. “令牌颁发者/受众无效”

- 确保 JWT_ISSUER 和 JWT_AUDIENCE 匹配Django设置 - 检查令牌是否以正确的声明发放

1. “令牌验证失败”

- 验证令牌未过期 - 检查Django和MCP服务器之间的JWT密钥是否匹配 - 确保使用RS256算法

调试

启用调试日志记录：

import logging
logging.basicConfig(level=logging.DEBUG)

许可证

\[您的许可证在这里\]