如何在 Django 中集成 MCP Server

背景说明

有几个原因导致 Django 集成 MCP Server 比较麻烦

• 目前支持的 MCP 服务是 SSE 协议的，需要长连接，但一般来讲 Django 是传统的连接方式
• MCP 的 Python SDK 直接封装了 Starlette 作为 ASGI 服务的提供，和 Django 的框架并不兼容
• 目前无法直接将 Starlette 的服务，作为 Django 某个子路由下的处理函数直接使用，我有看到一个插件项目，在处理这个事情，但目前我自己没有运行成功，欢迎交流

所以最简单的思路是 MCP 跑在另一个端口上，通过 Django ORM 来获取 Django 项目中的数据。

但如果希望集成到一个项目中使用，需要做一些调整：

• 使用 ASGI 来处理服务请求
• 修改 asgi.py 中的应用，指定部分路由由 Starlette 或其他 ASGI 服务处理
• 针对 Django 数据库查询语句，进行异步环境下的使用调整

第一步：使用 ASGI

官方文档推荐了几个 ASGI 服务，可以用来替代原有的 uwsgi，包括 Daphne、Hypercorn、Uvicorn，这里以 Daphne 为例说明，因为 Daphne 提供了和 Django 更好的集成效果

• 原有项目中 wsgi.py 下的相应调整，需要放到 asgy.py 下

• 调整 settings.py 文件

  # settings.py
  INSTALLED_APPS = ['daphne', # 需要放在第一个...
  ]
  ASGI_APPLICATION = '项目名称.asgi.application'
  

• 使用 Daphne 来接受请求，需要运行

  # 提供接口
  daphne -b 127.0.0.1 -p 8001 your_project_name.asgi:application
  # 通过 socket 文件
  daphne -u /path/to/your/project/daphne.sock your_project_name.asgi:application# 调试
  python manage.py runserver
  

• 至于 Daphne 和 Nginx 或 Apache 的配置文件，和之前使用 uwsgi 没有太大差异

第二步：修改 asgi.py 中的应用

主要目的有几个

• 引入 mcp 服务的 starlette app，对部分路径进行处理
• 对于其他路径，仍由 Django 进行处理

这里作为一个参考

import osfrom django.core.asgi import get_asgi_applicationos.environ.setdefault('DJANGO_SETTINGS_MODULE', '项目名称.settings')# 原来的 application
django_asgi_app = get_asgi_application()# ============
# 写一个 mcp 服务，内容也可以挪到其他位置，引入即可
from mcp.server.fastmcp import FastMCP
from asgiref.sync import sync_to_asyncmcp1 = FastMCP("weather")@mcp1.resource("greeting://{name}")
def get_greeting(name: str) -> str:"""Get a personalized greeting"""return f"Hello, {name}!"# Add an addition [e-mail]()
def add(a: int, b: int) -> int:"""Add two numbers"""return a + bmcp1.settings.mount_path = "/star" # 告诉 mcp server 将要处理 /star 路径下的内容
starlette_app = mcp1.sse_app()# ============
# 对 application 进行引流
def application(scope, receive, send):# star 开头的服务全部交给 mcp1 处理if scope['path'].startwith('/star):scope['path'] = scope['path'].replace('/star', '', 1) # 记得把路径做个调整，不然会出错return starlette_app(scope, receive, send)# 其他交给 Django 处理return django_asgi_app(scope, receive, send)

第三步：Django 数据的异步查询

当我们定义一个 mcp 的工具，需要访问 Django 数据库时，初始化的部分需要注意：

• 如果 mcp 的服务就在 Django 的工程目录下，和 Django 一起完成了初始化，基本不需要做数据库和配置的处理
• 如果需要单独使用 Django 的数据库，可以参考之前的文章，完成配置。

由于 mcp 是 ASGI 服务，所以需要把所有的 Django 数据库查询调整为异步模式：

• queryset 是惰性执行的，需要通过某些涉及数据库查询的函数才能触发执行
• 数据库查询的函数是同步的需要调整成异步函数

# 假设我们定义了一个 Model 叫 Entry
from xxx import Entry
# 引入 sync_to_async
from asgiref.sync import [e-mail]()
# 工具函数，需要定义为 async 函数，和 mcp1 放在同一个文件里
async def get_entry(tile: str):"""Get the Entries"""# 方法一：async 循环async for e in Entry.objects.filter(title=name):results.append(e.body)breakreturn results# 方法二：通过 sync_to_async 函数，把同步函数异步执行q = Entry.objects.all()lc = sync_to_async(len)(q) # sync_to_async 会把函数变成一个 coroutinel = await lc # 等待 lc 执行results = q[0].bodyreturn results# 方法三，使用同步函数的异步版本，通常在同步函数前会增加 a，例如 aget、afirstq = await Entry.objects.afirst()results = q.bodyreturn results