使用说明
mcp-django服务器
Django的完整MCP（模型上下文协议）服务器实现。使用简单的装饰器将工具和资源暴露给AI代理。
特性
- 🚀 简单的装饰师 -
@mcp_tool,@mcp_resource,@mcp_prompt - 🔄 自动发现 -自动查找
mcp_tools.py在所有Django应用程序中 - 📝 类型安全 -从Python类型提示生成JSON模式
- 🎯 符合MCP标准 -完整的JSON-RPC 2.0实现（规范2025-06-18）
- 🔒 自定义身份验证 -插入任何基于令牌的身份验证后端
- 👤 用户上下文 -自动将经过身份验证的用户注入工具
- 🔧 零配置 -开箱即用
安装
pip install mcp-django-server
快速开始
1.添加到INSTALLED_APPS
# settings.py
INSTALLED_APPS = [
...
'mcp_server',
]
# Optional configuration
MCP_SERVER_NAME = "My Django MCP Server"
MCP_SERVER_VERSION = "1.0.0"
# Auth backend (see Authentication section)
# MCP_AUTH_BACKEND = 'myapp.mcp.auth.TokenAuth'
# Public methods — skip auth (default: initialize + notifications/initialized)
# MCP_PUBLIC_METHODS = ['initialize', 'notifications/initialized']
# Rate limiting via Django cache (calls per period in seconds)
# MCP_RATE_LIMIT = {"calls": 100, "period": 60}
2.包含URL
# urls.py
from django.urls import path, include
urlpatterns = [
...
path('', include('mcp_server.urls')),
]
3.创建工具
创建 mcp_tools.py 在任何Django应用程序中：
# myapp/mcp_tools.py
from mcp_server import mcp_tool, mcp_resource
from .models import Product
@mcp_tool(
name="search_products",
description="Search products by name and category"
)
def search_products(query: str, category: str = None, limit: int = 10):
"""Search products - automatically exposed via /mcp/ endpoint."""
qs = Product.objects.filter(name__icontains=query)
if category:
qs = qs.filter(category=category)
return [
{"id": p.id, "name": p.name, "price": str(p.price)}
for p in qs[:limit]
]
@mcp_resource(
uri="catalog://products/{id}",
name="Product",
description="Full product details",
mime_type="application/json"
)
def get_product(id: int):
"""Get product by ID - exposed as MCP resource."""
p = Product.objects.get(pk=id)
return {
"id": p.id,
"name": p.name,
"description": p.description,
"price": str(p.price)
}
4.测试一下
# Initialize
curl -X POST http://localhost:8000/mcp/ \
-H "Content-Type: application/json" \
-d '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18","clientInfo":{"name":"test"}}}'
# List tools
curl -X POST http://localhost:8000/mcp/ \
-d '{"jsonrpc":"2.0","id":2,"method":"tools/list","params":{}}'
# Call tool
curl -X POST http://localhost:8000/mcp/ \
-d '{"jsonrpc":"2.0","id":3,"method":"tools/call","params":{"name":"search_products","arguments":{"query":"laptop"}}}'
认证
自定义身份验证后端
集 MCP_AUTH_BACKEND 在 settings.py 要求对每个MCP请求进行令牌认证。
# settings.py
MCP_AUTH_BACKEND = 'myapp.mcp.auth.TokenAuth'
类必须实现 authenticate(token: str) -> User | None:
# myapp/mcp/auth.py
class TokenAuth:
def authenticate(self, token):
from django.contrib.auth import get_user_model
User = get_user_model()
try:
return User.objects.get(mcp_token=token)
except User.DoesNotExist:
return None
配置后，每个请求都必须包括 Authorization: Bearer <token>.
缺少或无效的令牌返回HTTP 401。
工具中的用户上下文
如果你的工具函数声明了 user 参数，自动注入经过身份验证的用户。这 user 参数被排除在MCP输入模式之外（客户端不可见）。
@mcp_tool(description="Get my listings")
def get_my_listings(user):
return list(Listing.objects.filter(owner=user).values())
有条件的工具
使用 condition= 仅当用户的谓词满足时才公开工具。该工具已隐藏 tools/list 并被封锁 tools/call 当条件恢复时 False.
@mcp_tool(
description="Publish a listing",
condition=lambda user: hasattr(user, 'seller_profile')
)
def publish_listing(user, listing_id: int):
...
钩子
以（权力）否决 MCPView 在不接触核心调度的情况下添加自定义逻辑的方法：
# myapp/views.py
from mcp_server.views import MCPView
import logging
logger = logging.getLogger(__name__)
class AuditedMCPView(MCPView):
def before_dispatch(self, method, user):
logger.info("MCP call: method=%s user=%s", method, getattr(user, 'pk', None))
def after_tool_call(self, user, tool_name, result, duration_ms):
logger.info("Tool %s completed in %.1fms for user %s", tool_name, duration_ms, user)
# urls.py
from myapp.views import AuditedMCPView
urlpatterns = [
path('mcp/', AuditedMCPView.as_view()),
]
速率限制
通过Django的缓存启用每个用户（或匿名的每个IP）的速率限制：
# settings.py
MCP_RATE_LIMIT = {"calls": 100, "period": 60} # 100 requests per 60 seconds
当超过限制时，返回带有JSON-RPC错误的HTTP 429。
API 参考
@mcp_tool
将函数注册为MCP工具。
@mcp_tool(name="tool_name", description="Tool description")
def my_tool(param1: str, param2: int = 10):
return {"result": "value"}
参数：
name（str，可选）：工具名称。默认为函数名。description（str）：AI代理的工具描述。condition（可调用，可选）：(user) -> bool设置后，只有当当前用户满足条件时，该工具才会列出/调用。
类型提示：
- 函数类型提示会自动转换为JSON模式
- 支持的类型：
str,int,float,bool,list,dict - 没有默认值的参数标记为必需
- 这
user参数（如果存在）由身份验证层注入，并从模式中排除
@mcp资源
将函数注册为MCP资源。
@mcp_resource(
uri="catalog://items/{id}",
name="Item",
description="Item details",
mime_type="application/json"
)
def get_item(id: int):
return {"id": id, "data": "..."}
参数：
uri（str）：资源URI模板{param}占位符name（str，可选）：资源名称。默认为函数名。description（str）：资源描述mime_type（str）：MIME类型。默认值：“application/json”
@mcp-prompt
将函数注册为MCP提示模板。
@mcp_prompt(
name="analyze",
description="Analyze data",
arguments=[
{"name": "data_id", "description": "Data ID", "required": True}
]
)
def analyze_prompt(data_id: int):
return f"Analyze data with ID: {data_id}"
MCP端点
安装后，Django应用程序将公开：
POST /mcp/-主MCP JSON-RPC 2.0端点
支持的方法：
initialize-初始化MCP会话tools/list-列出所有已注册的工具tools/call-执行工具resources/list-列出所有资源resources/read-阅读资源prompts/list-列出所有提示prompts/get-获得提示
与django-mcp发现集成
如果你有 django-mcp-discovery 安装后，此程序包会自动更新 /.well-known/mcp-server 使用注册的工具清单。
例子
Django ORM工具
@mcp_tool(description="Get user by email")
def get_user(email: str):
from django.contrib.auth import get_user_model
User = get_user_model()
user = User.objects.get(email=email)
return {
"id": user.id,
"email": user.email,
"name": user.get_full_name()
}
外部API工具
@mcp_tool(description="Get weather forecast")
def get_weather(city: str):
import requests
response = requests.get(f"https://api.weather.com/{city}")
return response.json()
文件资源
@mcp_resource(
uri="files://{path}",
description="Read file content",
mime_type="text/plain"
)
def read_file(path: str):
with open(path, 'r') as f:
return f.read()
许可证
麻省理工学院
