具有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-key3.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.json4.生成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
- 客户注册
MCP客户端──────────POST/oauth2/注册──────────▶ MCP服务器 │ ▼ 生成client_id/secret │ ▼ MCP客户端◀──────────client_id+客户端密码───────┘
- 授权请求\
MCP客户端──────────GET/oauth2/授权─────────▶ MCP服务器 │ ▼ 重定向到Django │ ▼ 用户浏览器────────/oauth/谷歌/？next=回调──▶ Django后端 │ ▼ 谷歌OAuth │ ▼ 用户浏览器◀谷歌 │ ▼ 用户身份验证 │ ▼ 用户浏览器─────────使用身份验证码进行回调──────▶ Django后端 │ ▼ Django回调 │ ▼ 用户浏览器─────────重定向到MCP回调─────▶ MCP服务器 │ ▼ 生成身份验证码 │ ▼ MCP客户端◀─────────授权代码─────────────── MCP服务器
- 代币交换
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：令牌没有所需的权限
发展
添加新工具
- 将您的工具功能添加到
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"}- 工具可以自动访问：
- request.state.user：经过身份验证的用户信息 - request.headers.get("Authorization")：后端请求的原始JWT令牌
测试
# Install test dependencies
pip install pytest httpx pytest-asyncio
# Run tests (when available)
pytest tests/配置参考
| 环境变量 | 描述 | 默认值 |
|---|---|---|
DJANGO_BASE_URL | Django后端的基本URL | http://localhost:8000 |
JWT_AUDIENCE | 预计JWT观众人数 | cccrm-api |
JWT_ISSUER | 预期JWT发行人索赔 | cccrm-backend |
JWT_PUBLIC_KEY | 静态RSA公钥（可选） | - |
JWKS_CACHE_TTL | JWKS缓存TTL（秒） | 3600 |
PORT | 服务器端口 | 10000 |
TAVILY_API_KEY | Tavilly API网络搜索密钥 | - |
故障排除
常见问题
- “无法获取JWKS密钥”
- 检查一下 DJANGO_BASE_URL 是正确的 - 确保Django后端正在运行且可访问 - 验证JWKS端点是否可用 /auth/.well-known/jwks.json
- “令牌颁发者/受众无效”
- 确保 JWT_ISSUER 和 JWT_AUDIENCE 匹配Django设置 - 检查令牌是否以正确的声明发放
- “令牌验证失败”
- 验证令牌未过期 - 检查Django和MCP服务器之间的JWT密钥是否匹配 - 确保使用RS256算法
调试
启用调试日志记录：
import logging
logging.basicConfig(level=logging.DEBUG)许可证
\[您的许可证在这里\]