MCP Python SDK 中介軟體（Middleware）完全指南：攔截、計時與控制每一則入站訊息-平芜编程栈

人工智能
MCP 服务
MCP Clients

【免费下载链接】python-sdk

The official Python SDK for Model Context Protocol servers and clients

项目地址：https://gitcode.com/gh_mirrors/pythonsd/python-sdk

中介軟體（middleware）是 MCP Python SDK（pythonsd/python-sdk）為伺服器提供的一條非同步函式包住每一則入站訊息的掛鉤機制，可用來計時、記錄、追蹤、拒絕甚至改寫訊息，是觀察initialize交握的唯一入口。本文以官方文件為骨架，結合src/mcp/server下的真實實作（lowlevel/server.py、runner.py、context.py、mcpserver/server.py）與完整範例（docs_src/middleware/tutorial001.py），帶你寫出第一個計時中介軟體，並釐清「觀察、拒絕、改寫、回答」四種用法各自的代價與邊界，以及初始化死結、暫定介面等關鍵陷阱。

什麼是中介軟體：一條包住每則訊息的 async 函式

中介軟體的完整 API 只有一句話：把它寫成

async def my_middleware(ctx, call_next): ...

的形式，再附加到伺服器的middleware清單上，就完成了。

它本質上是一條洋蔥式的呼叫鏈：每一則伺服器收到的訊息（server/discover、initialize、一般請求、通知，甚至是伺服器沒有處理函式的方法）都會先進入中介軟體，由中介軟體決定是否呼叫call_next(ctx)把控制權交給鏈上剩下的部分（驗證 → 查找處理函式 → 你的處理函式），再視需要攔截、修改或直接回覆結果。

在 SDK 原始碼中，中介軟體的型別定義在 src/mcp/server/context.py：

CallNext = Callable[["ServerRequestContext[Any, Any]"], Awaitable[HandlerResult]] class ServerMiddleware(Protocol[_MwLifespanT]): async def __call__( self, ctx: ServerRequestContext[_MwLifespanT, Any], call_next: CallNext, ) -> HandlerResult: ...

也就是說，中介軟體是async (ctx, call_next) -> result，call_next以當下的ctx呼叫鏈上其餘部分並回傳結果。HandlerResult允許BaseModel、dict[str, Any]或None三種形式，由ServerRunner統一序列化為回應。

註冊的三種方式

middleware清單可以透過三種路徑取得與修改：

使用情境	註冊方式	對應原始碼
高階`MCPServer`建構時	`MCPServer(name, middleware=[...])`	src/mcp/server/mcpserver/server.py
高階`MCPServer`建構後	`mcp.middleware.append(...)`	同檔案的`middleware`property（L275-L283）
低階`Server`	`server.middleware.append(...)`	src/mcp/server/lowlevel/server.py

高階MCPServer.middleware其實就是低階Server.middleware的同一份清單（return self._lowlevel_server.middleware），所以兩者改的是同一條鏈。如果你對低階Server(name, on_call_tool=...)的寫法還不熟悉，建議先閱讀低階 Server 指南。

從原始碼還可以看到高階MCPServer在初始化時對清單做了三層排列（L241-L244）：

SDK 內建的OpenTelemetryMiddleware()（預設就在清單上，見下文專節）；
RequestStateBoundary（request state 安全邊界，位於 OpenTelemetry 內側）；
你透過middleware=[...]傳入的使用者中介軟體，以給定順序由外而內追加在內建元件內側。

動手寫一個計時中介軟體

完整的可執行範例位於 docs_src/middleware/tutorial001.py，一個伺服器、一個工具、一個中介軟體，記錄每則訊息花了多久：

import logging import time from mcp.server import Server, ServerRequestContext from mcp.server.context import CallNext, HandlerResult from mcp.types import ( CallToolRequestParams, CallToolResult, ListToolsResult, PaginatedRequestParams, TextContent, Tool, ) logger = logging.getLogger(__name__) async def on_list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams | None) -> ListToolsResult: return ListToolsResult( tools=[ Tool( name="search_books", description="Search the catalog by title or author.", input_schema={ "type": "object", "properties": {"query": {"type": "string"}}, "required": ["query"], }, ) ] ) async def on_call_tool(ctx: ServerRequestContext, params: CallToolRequestParams) -> CallToolResult: query = (params.arguments or {})["query"] return CallToolResult(content=[TextContent(type="text", text=f"Found 3 books matching {query!r}.")]) async def log_timing(ctx: ServerRequestContext, call_next: CallNext) -> HandlerResult: start = time.perf_counter() try: return await call_next(ctx) finally: elapsed_ms = (time.perf_counter() - start) * 1000 logger.info("%s took %.1f ms", ctx.method, elapsed_ms) server = Server("Bookshop", on_list_tools=on_list_tools, on_call_tool=on_call_tool) server.middleware.append(log_timing)

這個例子濃縮了四個關鍵知識點：

ctx就是處理函式收到的那個ServerRequestContext。ctx.method是原始的方法字串（如"tools/call"），ctx.params是尚未經過任何驗證的原始參數（可能是 dict，不一定是型別模型）。
call_next(ctx)執行鏈上剩下的部分：驗證、查找處理函式、你的處理函式。把它的回傳值原樣回傳，回應就不會被動到。
try/finally是刻意的。處理函式若拋出例外，失敗會以call_next拋出的例外形式抵達你的中介軟體，因此finally仍會執行計時——失敗的訊息一樣會被記錄。
server.middleware.append(...)完成註冊。清單由最外層開始執行（outermost-first），所以middleware[0]是最靠近線路（最先生效）的那一個。

試試看：連上客戶端後的執行效果

連上一個客戶端，列出工具，呼叫其中一個。記錄裡會有三行：

server/discover took 18.3 ms tools/list took 0.1 ms tools/call took 0.1 ms

你只呼叫了兩次（tools/list與tools/call），卻得到三行。第一行server/discover是客戶端為了建立連線而送出的請求，早在你要求任何東西之前就發生了。

中介軟體包住每一則入站訊息

這就是重點：中介軟體不是只包住工具呼叫，而是包住每一則抵達伺服器的入站訊息：

連線建立階段：server/discover，或在舊版工作階段（session）上的initialize與notifications/initialized。
每一個抵達伺服器的請求與每一則通知。對通知而言，ctx.request_id is None，call_next(ctx)回傳None，而你回傳的任何東西都會被丟棄。值得注意的是：在2026-07-28的 Streamable HTTP 路徑上，客戶端以 POST 送出的通知會在傳輸層直接以202確認收到、從不分派，所以不會抵達中介軟體；該修訂版沒有定義任何透過 HTTP 由客戶端送往伺服器的通知。
連伺服器沒有處理函式的方法也一樣：call_next會引發MCPError(-32601, "Method not found")，穿過你的中介軟體一路送到客戶端——你的中介軟體有機會記錄或改寫這個錯誤。

從原始碼看，這條鏈的組合邏輯位於 src/mcp/server/runner.py 的_compose_server_middleware（L291-L301）：

def _compose_server_middleware(self, inner: CallNext) -> CallNext: call = inner for middleware in reversed(self.server.middleware): call = partial(_apply_middleware, middleware, call) return call

reversed迴圈讓middleware[0]成為最外層（最先生效、最後取得控制權返還），而_on_request與_on_notify共用這條鏈，因此請求與通知走的是同一組中介軟體。ctx是在呼叫時才傳入的（_apply_middleware把call_next綁定給中介軟體），這正是中介軟體可以透過dataclasses.replace(ctx, ...)改寫參數交給鏈上其餘部分的原因。

在中介軟體裡能做什麼

官方文件依照「你應該猶豫的程度」由低到高，列舉了四種用法。

觀察（Observe）：計時、計數、記錄

就是上面的範例。這是中介軟體最安全、最主要的用途。因為它包住每一則訊息、且錯誤也會以例外形式流經鏈上，觀察型中介軟體可以建立完整的訊息級稽核軌跡。

拒絕（Refuse）：引發 MCPError 取代呼叫 call_next

不呼叫call_next(ctx)，改為引發MCPError，那一則訊息就會以 JSON-RPC 錯誤回應。關鍵特性：

連線不會斷，下一則訊息照常通過。
這是伺服器依呼叫端控管subscriptions/listen的方式——訂閱頁面的「決定誰可以觀看」有逐步說明。
從 runner 原始碼看，中介軟體鏈成功返回後才會提交連線狀態（commit），因此中介軟體否決（veto）訊息時不會留下任何狀態變更。

改寫（Rewrite）：用 dataclasses.replace 換參數

ctx是一個 dataclass，所以可以這樣改寫：

await call_next(dataclasses.replace(ctx, params=...))

這會把與客戶端送來不同的參數交給鏈上剩下的部分。注意ctx.params在進入中介軟體時尚未驗證，所以改寫會影響後續的驗證與處理函式呼叫。

絕對不要對initialize這樣做：客戶端拿到的結果是根據你改寫後的參數建立的，但伺服器提交連線狀態時用的是線路上原本的參數。雙方可能在交握結束時，對彼此協商出的內容認知不一致（例如協議版本、能力集合），導致連線處於「雙方各自以為協商成功」的危險狀態。

回答（Answer）：不呼叫 call_next 直接回傳

不呼叫call_next(ctx)就直接回傳一個結果，它會作為你的回應送到客戶端。call_next交給你的是完成的線路格式，而管線絕不會修補你回傳的東西，所以整個封包都由你負責：在 2026 世代的連線上，這包括serverInfo的_meta戳記——SDK 會替處理函式的結果加上它，但不會替你的加。換句話說，直接回覆等於完全接管了回應信封的格式責任。

initialize 的特殊地位：唯一掛鉤點與死結陷阱

中介軟體清單包住了initialize，而且這是你在initialize上唯一的掛鉤點。如果你試圖用add_request_handler接管它，SDK 會直接拒絕：

ValueError: 'initialize' is handled by the server runner and cannot be overridden; use Server.middleware to observe or wrap initialization

這個拒絕在原始碼中位於 src/mcp/server/lowlevel/server.py 的add_request_handler：

if method == "initialize": raise ValueError( "'initialize' is handled by the server runner and cannot be overridden; " "use Server.middleware to observe or wrap initialization" )

死結警告：initialize是就地處理的——在你的中介軟體鏈回傳之前，伺服器不會再讀取任何傳入的訊息。因此，如果在處理initialize時等待一個伺服器對客戶端的請求（ctx.session.send_request(...)、一次徵詢 elicitation），就會讓連線死結：你等的回應永遠讀不到。射後不理（fire-and-forget）的通知則沒問題。

唯一一個預設就啟用的中介軟體：OpenTelemetry

SDK 只附帶一個中介軟體，而且它已經在伺服器的清單上了：為每則訊息發出一個 OpenTelemetry span 的那一個。你不需要自己附加它，大多數時候也不用去想它。在安裝匯出器（exporter）之前它什麼都不做（no-op），它有自己的專頁：OpenTelemetry 指南。

原始碼佐證在 src/mcp/server/lowlevel/server.py：建構時直接初始化為self.middleware: list[ServerMiddleware[LifespanResultT]] = [OpenTelemetryMiddleware()]，並在註解中說明「預設啟用，每則訊息產生一個 SERVER span；安裝 OTel exporter 前是 no-op；若要退出（opt out），從清單中移除它即可」。OpenTelemetryMiddleware的實作位於 src/mcp/server/_otel.py。

中介軟體 vs ASGI 中介軟體：兩個層級，可以組合

如果你寫過 ASGI 中介軟體，這個形狀你已經認得：

Starlette 的(scope, receive, send)變成了(ctx, call_next)；
它在傳輸層之後執行，處理的是解碼後的訊息而不是原始的 HTTP 請求；
兩者可以組合：掛在streamable_http_app()上的 Starlette 中介軟體看到的是 HTTP；這裡（MCP 層）看到的是 MCP 訊息。

換句話說，MCP 中介軟體位於 MCP 協定層級，對傳輸層（HTTP/SSE/stdio）透明。

重點回顧

中介軟體是async (ctx, call_next) -> result，以MCPServer(middleware=[...])傳入（或附加到mcp.middleware），在低階的Server上則附加到server.middleware。
它包住每一則抵達伺服器的入站訊息（server/discover、initialize、請求、通知、未知的方法），並由最外層開始執行。
用ctx.request_id is None區分通知和請求。
不呼叫call_next改為引發例外，就能拒絕一則訊息；連線會存活下來。
SDK 自己的 OpenTelemetry 追蹤也是一個中介軟體，已經在清單上。詳見 OpenTelemetry 指南。
整個介面都是**暫定（provisional）**的：原始碼註解明確標示簽章和語意可能在 2.x 小版本中變動（見 context.py 與 lowlevel/server.py）。用它來觀察；不要在它上面蓋東西。