企业官网建设流程全解析

MCP Python SDK 中间件（Middleware）完全指南：用(ctx, call_next)包裹每一个入站消息

【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址: https://gitcode.com/gh_mirrors/pythonsd/python-sdk

本文围绕 MCP Python SDK（python-sdk 仓库）的中间件机制展开，系统讲解其核心 API、执行顺序、四大能力（观察、拒绝、改写、应答）以及initialize特例与默认内置的 OpenTelemetry 中间件。读完本文，你将掌握如何用async (ctx, call_next)编写计时、日志、限流、鉴权门控等中间件，并理解它与 ASGI 中间件、授权（Authorization）的边界关系。

本指南基于仓库中的 docs/advanced/middleware.md 及其 Hindi 翻译 i18n/hi/pages/advanced/middleware.md 整理扩充，并结合 src/mcp/server 源码与 tests/docs_src/test_middleware.py 测试用例逐条印证。

Middleware 是什么：整个 API 只有三个要素

在 MCP Python SDK 中，middleware（中间件）就是一个 async 函数，它包裹服务器收到的每一条消息。

你把它写成async (ctx, call_next)的形态，然后 append 到server.middleware列表里——这就是全部 API：

• ctx：与你的 handlers 收到的同一个ServerRequestContext。ctx.method是原始 method 字符串；ctx.params是原始 params，在任何校验之前。
• call_next(ctx)：调用链中剩下的部分——参数校验、handler 查找、你的 handler。把它的返回值原样返回，响应就原封不动。
• try/finally是刻意为之：handler 抛出异常时照样会被计时，因为失败会以call_next抛出的异常形式回到你的中间件里。

中间件列表在两种服务器形态下以同一份 list 暴露：

MCPServer的middleware属性返回的正是其内部低层Server的同一份 list（见 MCPServer.middleware 属性定义），所以两种形态共享同一套语义。

!!! warning "provisional 状态" 源码中将 middleware list 标记为provisional：其签名和语义可能在某个 2.x minor release 中发生变化。请用它来观察（计时、日志、tracing）和拒绝消息；不要把它当作你服务器赖以立足的地基。

下面的示例使用低层Server；如果Server(name, on_call_tool=...)对你来说还很陌生，建议先阅读 低级服务器（Low-level Server）指南。

实战：一个计时中间件

一个服务器、一个工具、一个记录每条消息耗时的中间件，完整代码如下（来源：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)

逐行拆解这个中间件：

• start = time.perf_counter()：在进入调用链之前记录开始时间。
• return await call_next(ctx)：把控制权交给链条剩余部分，并原样返回其结果——响应保持"原汁原味"。
• finally块：无论call_next正常返回还是抛出异常，都会执行计时并打印日志。这正是"handler 抛错也要计时"的保证。
• server.middleware.append(...)：注册该中间件。列表最外层优先执行，所以middleware[0]是离 wire（网络层）最近的那个。

从源码层面看，列表的执行逻辑位于 ServerRunner._compose_server_middleware：它对self.server.middleware做reversed遍历并逐层包裹，最终形成的调用链由_on_request与_on_notify共用，保证同一条中间件链能观察到每一条入站消息。

试运行：为什么两次调用打出三行日志

连接一个 client，获取工具列表，再调用一个工具。你的日志里会有三行：

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

你只发起了两次调用，却得到三行日志。第一行server/discover是 client 在你请求任何东西之前、为建立连接而发送的请求。仓库测试 test_middleware_observes_every_inbound_message 精确验证了这一行为：对tutorial001.server依次list_tools()与call_tool()，捕获到的计时记录依次正是["server/discover", "tools/list", "tools/call"]，且末行符合tools/call took \d+\.\d ms格式。

这正是 middleware 的意义所在：它包裹每一条入站消息：

• 连接建立：server/discover；在 legacy session 上则是initialize与notifications/initialized。
• 每条请求和每条通知。对通知而言，ctx.request_id is None，call_next(ctx)返回None，而你的返回值会被丢弃。（在2026-07-28版本的 streamable-HTTP 路径上，client 的通知 POST 会在 transport 层直接以202确认、从不 dispatch，因此也不会到达 middleware；该 revision 在 HTTP 上根本没有定义 client-to-server 通知。）
• 服务器没有对应 handler 的方法：call_next会把MCPError(-32601, "Method not found")穿过你的中间件抛向 client。测试 test_an_unhandled_method_raises_through_the_middleware 验证了这一点：请求resources/read（无 handler）时，spy 中间件先观察到("resources/read", METHOD_NOT_FOUND)再重抛。

中间件内部能做什么：从"保守"到"大胆"的四个层次

按你应该犹豫的程度从低到高排列：

• 观察（Observe）：计时、计数、打日志。就是上面的示例。
• 拒绝（Refuse）：不调用call_next(ctx)，直接抛一个MCPError，这一条消息就会以 JSON-RPC error 应答。连接保持存活，下一条消息照常通过。这正是服务器按调用方 gatesubscriptions/listen的做法，参见订阅页面的 决定谁可以观看（Deciding who may watch）。测试 test_raising_before_call_next_refuses_the_message 展示了完整闭环：拒绝tools/call后 client 收到INVALID_REQUEST错误，而随后的list_tools依然成功返回。
• 改写（Rewrite）：ctx是一个 dataclass，可以用await call_next(dataclasses.replace(ctx, params=...))让链条其余部分拿到与 client 发送不同的 params。千万不要对initialize这么做：client 拿到的结果由你改写后的 params 构建，但服务器会用原始 wire params 提交连接状态——双方可能带着"对谈判内容的分歧"完成握手。
• 应答（Answer）：不调用call_next(ctx)直接返回 result，它会作为你的响应发给 client。call_next交给你的已是完成态的 wire form，而 pipeline 永远不会 patch 你返回的内容，所以整个 envelope 都是你的：在 2026 一代的连接上，这包含serverInfo的_metastamp——SDK 会把它加进 handler 结果，但不会加进你的结果。

类型层面，src/mcp/server/context.py 中的ServerMiddlewareProtocol 明确定义了该契约：它运行在ServerRunner._on_request/_on_notify的顶部、ctx构建之后、任何校验/查找/握手之前；initialize、pre-init gate、METHOD_NOT_FOUND、params 校验、handler 调用、notifications/initialized全部在call_next(ctx)内部执行。CallNext与HandlerResult类型也定义于此（HandlerResult = BaseModel | dict[str, Any] | None，由ServerRunner序列化为 result dict）。

initialize 特例：你唯一能拿到它的 hook

initialize是 middleware 包裹的对象之一，也是你唯一能 hook 它的入口。尝试用add_request_handler接管它会直接被 SDK 拒绝：

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

这条错误信息在 lowlevel/server.py 的 add_request_handler 中直接硬编码抛出：initialize为 runner 保留（握手归 runner 所有），注册它即抛ValueError。测试 test_initialize_cannot_be_replaced_only_wrapped 断言了这一行为。

!!! warning "initialize 是 inline 处理的"initialize是 inline 处理的：在你的 middleware 链返回之前，服务器不会读取任何后续入站消息。因此，在处理initialize时 await 一个 server-to-client 请求（如ctx.session.send_request(...)、elicitation）会死锁连接：你等待的响应永远不会被读取。Fire-and-forget 的通知则没问题。（context.py 中的 Protocol 文档同样警告了这一点：发送后忘记的通知是安全的。）

另外一个与initialize相关的细节：从 runner.py 的 _on_request 可以看到，握手状态（connection.client_params、connection.protocol_version）只会在链条成功返回后提交，因此中间件在call_next()之前抛错即可否决握手，且不会留下任何已提交的状态。

默认自带的一个中间件：OpenTelemetry

SDK 恰好随附一个中间件，而且它已经在你的服务器列表里了：为每条消息发射一个 OpenTelemetry span 的那个。你不需要 append 它，多数时候甚至不会想起它。在安装 exporter 之前它是 no-op，它有自己独立的页面：OpenTelemetry。

源码位置在 src/mcp/server/lowlevel/server.py#L428-L440：

self.middleware: list[ServerMiddleware[LifespanResultT]] = [OpenTelemetryMiddleware()]

其实现位于 src/mcp/server/_otel.py，从中可以看到它为你做了什么：

• 为每个入站消息发射一个SpanKind.SERVER的 span，span 名为{ctx.method}（有 target 时附加{target}），携带mcp.method.name、mcp.protocol.version属性；有request_id时附加jsonrpc.request.id。
• 对tools/call附加gen_ai.operation.name = "execute_tool"与gen_ai.tool.name；对prompts/get附加gen_ai.prompt.name。
• 通过extract_trace_context(ctx.meta)从请求_meta中提取父 trace context，实现链路延续。
• 对MCPError、ValidationError和任意异常分别记录error.type/rpc.response.status_code并置 span 状态为 ERROR 后重抛；对tools/call的CallToolResult(is_error=True)（或原始 dict 的"isError": True）也会标记tool_error。

另外需要注意：MCPServer构造时，SDK 的 built-in 中间件（OpenTelemetry，然后是 request-state boundary）在内层，你的用户中间件追加在它们外层（见 mcpserver/server.py#L241-L244），按给定的顺序 outermost-first 排列。

!!! info "与 ASGI 中间件的对比" 如果你写过 ASGI 中间件，这个形态你一定似曾相识。Starlette 的(scope, receive, send)在这里变成了(ctx, call_next)，而且它运行在transport 之后，作用于解码后的 message 而非原始 HTTP request。两者可以组合使用：streamable_http_app()上的 Starlette 中间件看的是 HTTP，这里的 middleware 看的是 MCP。

总结

• 一个 middleware 是async (ctx, call_next) -> result，可通过MCPServer(middleware=[...])传入（或 append 到mcp.middleware），在低层Server上 append 到server.middleware。
• 它包裹到达服务器的每一条入站消息（server/discover、initialize、requests、notifications、未知 method），并且 outermost-first 执行。
• ctx.request_id is None是你区分通知与请求的方式。
• 想拒绝某一条消息就抛异常而不是调用call_next；连接会存活下来。
• SDK 自己的 OpenTelemetry tracing 也是一个 middleware，已经在列表里了。参见 OpenTelemetry。
• 整个 surface 是 provisional 的。用它来观察；不要在其上构建根基。

以上就是所有包裹请求的东西。授权（Authorization） 才是决定请求是否允许运行的那一层——middleware 负责"看到"每一条消息，授权负责"放行"它们。

【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址: https://gitcode.com/gh_mirrors/pythonsd/python-sdk