中介軟體

機器翻譯

本頁是從英文說明文件自動翻譯而來，以英文頁面為準。如果哪裡讀起來不對勁，翻譯有說明如何回報。

中介軟體（middleware）是一個非同步函式，包住伺服器收到的每一則訊息。

寫成 async (ctx, call_next) 的形式，再附加到 server.middleware 就好。整個 API 就這樣。

Warning

中介軟體清單在原始碼裡標示為暫定（provisional）：它的簽章和語意可能在 2.x 的小版本中變動。用它來觀察（計時、記錄、追蹤）和拒絕訊息；不要把它當成伺服器賴以運作的基礎。

MCPServer 在建構時接收這份清單（MCPServer(name, middleware=[...])），並以 mcp.middleware 公開；低階的 Server 則以 server.middleware 公開同一份清單。下面的範例使用低階的 Server；如果還沒見過 Server(name, on_call_tool=...)，請先讀 低階 Server。

一個計時中介軟體

一個伺服器、一個工具、一個中介軟體，記錄每則訊息花了多久：

server.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 是原始的方法字串；ctx.params 是原始的參數，尚未經過任何驗證。
call_next(ctx) 會執行鏈上剩下的部分：驗證、查找處理函式、你的處理函式。把它的回傳值原樣回傳，回應就不會被動到。
try/finally 是刻意的：引發例外的處理函式一樣會被計時，因為失敗會以 call_next 拋出的例外形式抵達你的中介軟體。
server.middleware.append(...) 完成註冊。清單由最外層開始執行，所以 middleware[0] 是最靠近線路的那一個。

試試看

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

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

呼叫了兩次，卻得到三行。第一行是 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")，穿過你的中介軟體一路送到用戶端。

並行上限

中介軟體不一定要呼叫 call_next(ctx)。改為引發 MCPError，就等於拒絕了那一則訊息：連線不會斷，下一則訊息照常通過。

假設每次搜尋都會佔用連線池裡的一條連線，而池子裡只有 4 條。這個中介軟體讓 4 個工具呼叫同時執行，第 5 個就拒絕：

server.py

from typing import Any

from mcp import MCPError
from mcp.server import Server, ServerRequestContext
from mcp.server.context import CallNext, HandlerResult, ServerMiddleware
from mcp.types import (
    CallToolRequestParams,
    CallToolResult,
    ListToolsResult,
    PaginatedRequestParams,
    TextContent,
    Tool,
)

# MCP defines no "busy" error, so this server picks its own code.
SERVER_BUSY = 1


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}.")])


def max_concurrent_tool_calls(limit: int) -> ServerMiddleware[Any]:
    running = 0

    async def middleware(ctx: ServerRequestContext, call_next: CallNext) -> HandlerResult:
        nonlocal running
        if ctx.method != "tools/call":
            return await call_next(ctx)
        if running >= limit:
            raise MCPError(code=SERVER_BUSY, message=f"Server busy: tool call limit reached ({limit} in progress).")
        running += 1
        try:
            return await call_next(ctx)
        finally:
            running -= 1

    return middleware


server = Server("Bookshop", on_list_tools=on_list_tools, on_call_tool=on_call_tool)
server.middleware.append(max_concurrent_tool_calls(4))

只計算 tools/call，所以伺服器在拒絕工具呼叫的同時，仍會繼續回應 server/discover 和 tools/list。
MCP 沒有定義「伺服器忙碌」的錯誤碼，所以 SERVER_BUSY 是這個伺服器自己定義的。
拒絕能讓用戶端立刻知道伺服器已經過載。如果寧可讓呼叫端等待，就改在 call_next(ctx) 外面包一層 anyio.CapacityLimiter。

引發的 MCPError 會送到用戶端應用程式，而不是模型。如果要讓模型讀到這則訊息，就改為回傳一個帶有 is_error=True 的工具結果：這就是下面的回答。

在裡面能做什麼

依照該有的猶豫程度，由低到高排列：

觀察。計時、計數、記錄。就是上面的計時中介軟體。
拒絕。不呼叫 call_next(ctx)，改為引發 MCPError，那一則訊息就會以 JSON-RPC 錯誤回應。連線不會斷；下一則訊息照常通過。就是上面的並行上限。伺服器也是這樣依呼叫端控管 subscriptions/listen 的：訂閱頁面的 決定誰可以觀看 有逐步說明。
改寫。ctx 是一個 dataclass：await call_next(dataclasses.replace(ctx, params=...)) 會把和用戶端送來的不同的參數交給鏈上剩下的部分。絕對不要對 initialize 這麼做：用戶端拿到的結果是根據你改寫後的參數建立的，但伺服器提交連線狀態時用的是線路上原本的參數。雙方可能在交握結束時，對彼此協商出的內容認知不一致。
回答。不呼叫 call_next(ctx) 就直接回傳一個結果，它會作為你的回應送到用戶端。call_next 交給你的是完成的線路格式，而管線絕不會修補你回傳的東西，所以整個封包都由你負責：在 2026 世代的連線上，這包括 serverInfo 的 _meta 戳記，SDK 會替處理函式的結果加上它，但不會替你的加。

Check

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

Warning

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

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

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

Info

如果寫過 ASGI 中介軟體，這個形狀你已經認得。Starlette 的 (scope, receive, send) 變成了 (ctx, call_next)，而且它在傳輸之後執行，處理的是解碼後的訊息而不是原始的 HTTP 請求。兩者可以組合：掛在 streamable_http_app() 上的 Starlette 中介軟體看到的是 HTTP；這裡看到的是 MCP。

重點回顧

中介軟體是 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。
整個介面都是暫定的。用它來觀察；不要在它上面蓋東西。

以上就是包住請求的一切。至於請求到底能不能執行，則由授權決定。