背景說明

有幾個原因導致 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 tool@mcp1.tool()
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 sync_to_async@mcp1.tool()
# 工具函數，需要定義為 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