使用说明
具有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
-
客户注册 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
```http
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
```bash
# 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)
许可证
[您的许可证在这里]
