` 观测、拦截与改写每条入站消息)
MCP Python SDK 中间件Middleware实战用async (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 SDKModel Context Protocol 官方 Python SDK的 Server 中间件机制展开讲解如何用一条形如async (ctx, call_next)的异步函数包裹服务器接收到的每一条消息从而完成耗时统计、日志观测、按调用方拒绝请求、改写参数乃至直接应答等横切能力。读完本文你将掌握低层Server与高层MCPServer两套 API 上注册中间件的方法、中间件链的执行顺序、initialize握手消息的特殊性与陷阱以及 SDK 默认附带的 OpenTelemetry 中间件的内部实现。中间件是什么一条异步函数包裹每条入站消息在 MCP Python SDK 中中间件middleware就是一条异步函数它包裹服务器接收到的每一条消息。它的签名固定为async def middleware(ctx: ServerRequestContext, call_next: CallNext) - HandlerResult: ...你把它追加到server.middleware列表即可生效——这就是中间件的全部 API。这里的ctx与你的处理器handler收到的是同一个ServerRequestContext对象ctx.method是原始方法名字符串ctx.params是尚未经过任何校验的原始参数字典。call_next(ctx)则负责执行链条的剩余部分参数校验、处理器查找、最终调用你的 handler并把结果原样返回。一个关键的设计原则是中间件列表在源码中被明确标记为“临时provisional”。在 src/mcp/server/lowlevel/server.py 中可以看到对应的 TODO 注释其签名与语义可能随 2.x 次版本发布中的 Context/middleware 重构而改变。因此官方建议用它来观测计时、日志、追踪和拒绝消息而不要把它当作服务器赖以运转的地基。两套 APIMCPServer 构造参数与低层 server.middleware中间件列表在 SDK 的两层服务器 API 上以两种方式暴露低层Server在构造后直接通过server.middleware.append(...)追加这也是下文计时示例采用的方式如果你还不熟悉Server(name, on_call_tool...)这种低层写法建议先阅读 低层 Server 指南高层MCPServer在构造时通过MCPServer(name, middleware[...])传入并通过mcp.middleware属性暴露同一份列表。从源码可以确认MCPServer.middleware属性与低层Server.middleware是同一份列表——MCPServer内部维护着一个_lowlevel_server实例其middleware属性直接透传底层列表见 src/mcp/server/mcpserver/server.py。值得注意的还有高层 API 中用户中间件的插入位置。在 src/mcp/server/mcpserver/server.py 中可以看到SDK 内置的中间件OpenTelemetry 追踪、请求状态边界RequestStateBoundary先被追加进列表用户的中间件再按传入顺序追加其后形成“最外层是 SDK 内置件、用户中间件在其内部”的嵌套结构。计时中间件实战一个完整可运行的示例下面是一个“一个服务器 一个工具 一个中间件”的完整示例它记录每条消息的处理耗时。该示例源码位于 docs_src/middleware/tutorial001.pyimport 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( namesearch_books, descriptionSearch 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(typetext, textfFound 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_toolson_list_tools, on_call_toolon_call_tool) server.middleware.append(log_timing)逐行拆解这个示例可以提炼出四条核心语义ctx即ServerRequestContext它由ServerRunner为每条入站消息构造携带连接级ServerSession用于服务器向客户端发起请求与通知、协议版本、原始方法名与原始参数等元数据定义见 src/mcp/server/context.py。call_next(ctx)执行链条剩余部分校验 → 处理器查找 → 你的 handler返回值原样透传响应不被改动。try/finally是刻意为之即便 handler 抛出异常耗时依然会被记录——因为失败会以call_next抛出的异常形式到达你的中间件。server.middleware.append(...)完成注册列表从外向内执行因此middleware[0]是离“线路”wire最近的那一个。运行验证两次调用为何产生三条日志连接一个客户端依次列出工具、调用一个工具你的日志中会出现三行server/discover took 18.3 ms tools/list took 0.1 ms tools/call took 0.1 ms你明明只发起了两次调用却得到三行日志。第一行server/discover是客户端在建立连接阶段自动发送的请求——发生在你提出任何业务请求之前。这正是中间件的意义所在它包裹每一条到达服务器的入站消息包括连接建立流程server/discover在传统会话legacy session上则对应initialize与notifications/initialized每条请求与通知对于通知ctx.request_id is Nonecall_next(ctx)返回None你返回的任何内容都会被丢弃。一个补充细节是在2026-07-28版本的 Streamable HTTP 路径上客户端的通知 POST 会在传输层直接收到202应答而不会被分发因此也到不了中间件——该协议修订版没有定义任何 HTTP 上的客户端到服务器通知服务器没有处理器的未知方法call_next会抛出MCPError(-32601, Method not found)该异常穿过你的中间件一路传回客户端。中间件内可以做什么观察、拒绝、改写、应答按“需要多谨慎”递增的顺序中间件内可以执行四类操作观察Observe计时、计数、打日志——即上文示例所示。这是中间件最主要的用途风险最低。拒绝Refuse不调用call_next(ctx)而是直接抛出一个MCPError那么这一条消息会收到一个 JSON-RPC 错误应答而连接保持存活下一条消息照常通过。这是服务器按调用方控制subscriptions/listen访问权限的惯用手段具体做法可参考 Abonnements订阅页面中的“决定谁可以监听”一节。改写Rewritectx是一个 dataclass因此可以用dataclasses.replace构造一个改写过参数的新上下文await call_next(dataclasses.replace(ctx, params...))这样链条后续部分收到的参数就与客户端实际发送的不同。从 src/mcp/server/runner.py 的实现看_compose_server_middleware是在调用时读取ctx上的method/params的所以中间件对ctx的改写会即时作用于后续链条——这也是这一能力能够成立的原因。但绝不要对initialize做改写客户端拿到的返回结果虽然是由你改写后的参数构建的但服务器握手阶段提交的连接状态却来自线路上的原始参数最终可能导致通信双方对协商结果各执一词。应答Answer不调用call_next(ctx)直接返回一个结果它会作为你的应答发给客户端。此时call_next会交给你“最终在线上传输的形态”而管道不会再改动你返回的内容因此整个信封都由你负责在 2026 时代的连接上这包括_meta中的serverInfo时间戳——SDK 会给 handler 的结果附加该字段但不会替你附加。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对initialize方法直接抛出ValueError因为握手流程归 runner 所有。此外还有一个必须牢记的死锁陷阱initialize是内联处理的——在你的中间件链返回之前服务器不会读取任何其他入站消息。因此如果在处理initialize期间等待一个服务器到客户端的请求ctx.session.send_request(...)例如一次 elicitation 引导会死锁整个连接你等待的响应永远不可能被读取。而“发出即忘”的通知fire-and-forget则没有问题。默认随 SDK 附带的中间件OpenTelemetrySDK 恰好内置了一个中间件并且默认已经在你的服务器中间件列表里即每处理一条消息就发射一个 OpenTelemetry span 的那个。你不需要手动追加它大多数时候甚至感觉不到它的存在——在安装导出器exporter之前它完全是无操作no-op。其实现位于 src/mcp/server/_otel.py类OpenTelemetryMiddleware同样遵循ServerMiddleware协议在call_next的调用点包裹 span 的开启与关闭。在 src/mcp/server/lowlevel/server.py 可以看到它的默认注册方式self.middleware: list[...] [OpenTelemetryMiddleware()]。如果你想关闭它把它从列表中移除即可。完整的配置与导出方式见 OpenTelemetry 指南。与 ASGI/Starlette 中间件的对比如果你写过 ASGI 中间件会对这个形态感到熟悉Starlette 的(scope, receive, send)在这里变成了(ctx, call_next)并且它运行在传输层之后作用于解码后的 MCP 消息而非原始 HTTP 请求。两者可以叠加组合挂在streamable_http_app()上的 Starlette 中间件看到的是 HTTP而这里的中间件看到的是 MCP。在高层MCPServer的源码中也能看到这种分层的痕迹——Starlette 层的认证中间件AuthenticationMiddleware、BearerAuthBackend等用于 HTTP 认证而ctx层的中间件链用于 MCP 消息处理。源码级原理runner 如何组合中间件链从 src/mcp/server/runner.py 的实现可以看到链条的组装方式_compose_server_middleware将server.middleware列表反转遍历把每个中间件通过partial(_apply_middleware, middleware, call)包裹到前一个之上从而保证列表从外向内执行middleware[0]最靠近线路。_on_request与_on_notify共享这条链路因此请求与通知走的是同一条中间件链。另外中间件的异常处理同样被纳入考量链条内部的异常会以call_next抛出的形式让每个外层中间件依次观测到且只在整条链成功返回后才提交连接状态——这意味着中间件“否决”一条消息不会留下任何状态残留。总结中间件的形态是async (ctx, call_next) - result低层通过server.middleware.append(...)注册高层通过MCPServer(middleware[...])传入或追加到mcp.middleware它包裹每一条到达服务器的入站消息server/discover、initialize、普通请求、通知、未知方法并从外向内执行ctx.request_id is None是区分通知与请求的标志不调用call_next而是抛异常可以拒绝单条消息且连接不受影响SDK 自带的 OpenTelemetry 追踪本身就是一个中间件且默认已在列表中参考 OpenTelemetry整套中间件接口目前仍是临时 APIprovisional请用它来观测而不要在其上构建长期依赖。至此我们覆盖了所有“包裹在请求外围”的机制至于一条请求是否有权执行则由 授权Authorization 机制来决定。【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址: https://gitcode.com/gh_mirrors/pythonsd/python-sdk创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
MCP Python SDK 中间件（Middleware）实战：用 `async (ctx, call_next)` 观测、拦截与改写每条入站消息
MCP Python SDK 中间件Middleware实战用async (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 SDKModel Context Protocol 官方 Python SDK的 Server 中间件机制展开讲解如何用一条形如async (ctx, call_next)的异步函数包裹服务器接收到的每一条消息从而完成耗时统计、日志观测、按调用方拒绝请求、改写参数乃至直接应答等横切能力。读完本文你将掌握低层Server与高层MCPServer两套 API 上注册中间件的方法、中间件链的执行顺序、initialize握手消息的特殊性与陷阱以及 SDK 默认附带的 OpenTelemetry 中间件的内部实现。中间件是什么一条异步函数包裹每条入站消息在 MCP Python SDK 中中间件middleware就是一条异步函数它包裹服务器接收到的每一条消息。它的签名固定为async def middleware(ctx: ServerRequestContext, call_next: CallNext) - HandlerResult: ...你把它追加到server.middleware列表即可生效——这就是中间件的全部 API。这里的ctx与你的处理器handler收到的是同一个ServerRequestContext对象ctx.method是原始方法名字符串ctx.params是尚未经过任何校验的原始参数字典。call_next(ctx)则负责执行链条的剩余部分参数校验、处理器查找、最终调用你的 handler并把结果原样返回。一个关键的设计原则是中间件列表在源码中被明确标记为“临时provisional”。在 src/mcp/server/lowlevel/server.py 中可以看到对应的 TODO 注释其签名与语义可能随 2.x 次版本发布中的 Context/middleware 重构而改变。因此官方建议用它来观测计时、日志、追踪和拒绝消息而不要把它当作服务器赖以运转的地基。两套 APIMCPServer 构造参数与低层 server.middleware中间件列表在 SDK 的两层服务器 API 上以两种方式暴露低层Server在构造后直接通过server.middleware.append(...)追加这也是下文计时示例采用的方式如果你还不熟悉Server(name, on_call_tool...)这种低层写法建议先阅读 低层 Server 指南高层MCPServer在构造时通过MCPServer(name, middleware[...])传入并通过mcp.middleware属性暴露同一份列表。从源码可以确认MCPServer.middleware属性与低层Server.middleware是同一份列表——MCPServer内部维护着一个_lowlevel_server实例其middleware属性直接透传底层列表见 src/mcp/server/mcpserver/server.py。值得注意的还有高层 API 中用户中间件的插入位置。在 src/mcp/server/mcpserver/server.py 中可以看到SDK 内置的中间件OpenTelemetry 追踪、请求状态边界RequestStateBoundary先被追加进列表用户的中间件再按传入顺序追加其后形成“最外层是 SDK 内置件、用户中间件在其内部”的嵌套结构。计时中间件实战一个完整可运行的示例下面是一个“一个服务器 一个工具 一个中间件”的完整示例它记录每条消息的处理耗时。该示例源码位于 docs_src/middleware/tutorial001.pyimport 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( namesearch_books, descriptionSearch 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(typetext, textfFound 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_toolson_list_tools, on_call_toolon_call_tool) server.middleware.append(log_timing)逐行拆解这个示例可以提炼出四条核心语义ctx即ServerRequestContext它由ServerRunner为每条入站消息构造携带连接级ServerSession用于服务器向客户端发起请求与通知、协议版本、原始方法名与原始参数等元数据定义见 src/mcp/server/context.py。call_next(ctx)执行链条剩余部分校验 → 处理器查找 → 你的 handler返回值原样透传响应不被改动。try/finally是刻意为之即便 handler 抛出异常耗时依然会被记录——因为失败会以call_next抛出的异常形式到达你的中间件。server.middleware.append(...)完成注册列表从外向内执行因此middleware[0]是离“线路”wire最近的那一个。运行验证两次调用为何产生三条日志连接一个客户端依次列出工具、调用一个工具你的日志中会出现三行server/discover took 18.3 ms tools/list took 0.1 ms tools/call took 0.1 ms你明明只发起了两次调用却得到三行日志。第一行server/discover是客户端在建立连接阶段自动发送的请求——发生在你提出任何业务请求之前。这正是中间件的意义所在它包裹每一条到达服务器的入站消息包括连接建立流程server/discover在传统会话legacy session上则对应initialize与notifications/initialized每条请求与通知对于通知ctx.request_id is Nonecall_next(ctx)返回None你返回的任何内容都会被丢弃。一个补充细节是在2026-07-28版本的 Streamable HTTP 路径上客户端的通知 POST 会在传输层直接收到202应答而不会被分发因此也到不了中间件——该协议修订版没有定义任何 HTTP 上的客户端到服务器通知服务器没有处理器的未知方法call_next会抛出MCPError(-32601, Method not found)该异常穿过你的中间件一路传回客户端。中间件内可以做什么观察、拒绝、改写、应答按“需要多谨慎”递增的顺序中间件内可以执行四类操作观察Observe计时、计数、打日志——即上文示例所示。这是中间件最主要的用途风险最低。拒绝Refuse不调用call_next(ctx)而是直接抛出一个MCPError那么这一条消息会收到一个 JSON-RPC 错误应答而连接保持存活下一条消息照常通过。这是服务器按调用方控制subscriptions/listen访问权限的惯用手段具体做法可参考 Abonnements订阅页面中的“决定谁可以监听”一节。改写Rewritectx是一个 dataclass因此可以用dataclasses.replace构造一个改写过参数的新上下文await call_next(dataclasses.replace(ctx, params...))这样链条后续部分收到的参数就与客户端实际发送的不同。从 src/mcp/server/runner.py 的实现看_compose_server_middleware是在调用时读取ctx上的method/params的所以中间件对ctx的改写会即时作用于后续链条——这也是这一能力能够成立的原因。但绝不要对initialize做改写客户端拿到的返回结果虽然是由你改写后的参数构建的但服务器握手阶段提交的连接状态却来自线路上的原始参数最终可能导致通信双方对协商结果各执一词。应答Answer不调用call_next(ctx)直接返回一个结果它会作为你的应答发给客户端。此时call_next会交给你“最终在线上传输的形态”而管道不会再改动你返回的内容因此整个信封都由你负责在 2026 时代的连接上这包括_meta中的serverInfo时间戳——SDK 会给 handler 的结果附加该字段但不会替你附加。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对initialize方法直接抛出ValueError因为握手流程归 runner 所有。此外还有一个必须牢记的死锁陷阱initialize是内联处理的——在你的中间件链返回之前服务器不会读取任何其他入站消息。因此如果在处理initialize期间等待一个服务器到客户端的请求ctx.session.send_request(...)例如一次 elicitation 引导会死锁整个连接你等待的响应永远不可能被读取。而“发出即忘”的通知fire-and-forget则没有问题。默认随 SDK 附带的中间件OpenTelemetrySDK 恰好内置了一个中间件并且默认已经在你的服务器中间件列表里即每处理一条消息就发射一个 OpenTelemetry span 的那个。你不需要手动追加它大多数时候甚至感觉不到它的存在——在安装导出器exporter之前它完全是无操作no-op。其实现位于 src/mcp/server/_otel.py类OpenTelemetryMiddleware同样遵循ServerMiddleware协议在call_next的调用点包裹 span 的开启与关闭。在 src/mcp/server/lowlevel/server.py 可以看到它的默认注册方式self.middleware: list[...] [OpenTelemetryMiddleware()]。如果你想关闭它把它从列表中移除即可。完整的配置与导出方式见 OpenTelemetry 指南。与 ASGI/Starlette 中间件的对比如果你写过 ASGI 中间件会对这个形态感到熟悉Starlette 的(scope, receive, send)在这里变成了(ctx, call_next)并且它运行在传输层之后作用于解码后的 MCP 消息而非原始 HTTP 请求。两者可以叠加组合挂在streamable_http_app()上的 Starlette 中间件看到的是 HTTP而这里的中间件看到的是 MCP。在高层MCPServer的源码中也能看到这种分层的痕迹——Starlette 层的认证中间件AuthenticationMiddleware、BearerAuthBackend等用于 HTTP 认证而ctx层的中间件链用于 MCP 消息处理。源码级原理runner 如何组合中间件链从 src/mcp/server/runner.py 的实现可以看到链条的组装方式_compose_server_middleware将server.middleware列表反转遍历把每个中间件通过partial(_apply_middleware, middleware, call)包裹到前一个之上从而保证列表从外向内执行middleware[0]最靠近线路。_on_request与_on_notify共享这条链路因此请求与通知走的是同一条中间件链。另外中间件的异常处理同样被纳入考量链条内部的异常会以call_next抛出的形式让每个外层中间件依次观测到且只在整条链成功返回后才提交连接状态——这意味着中间件“否决”一条消息不会留下任何状态残留。总结中间件的形态是async (ctx, call_next) - result低层通过server.middleware.append(...)注册高层通过MCPServer(middleware[...])传入或追加到mcp.middleware它包裹每一条到达服务器的入站消息server/discover、initialize、普通请求、通知、未知方法并从外向内执行ctx.request_id is None是区分通知与请求的标志不调用call_next而是抛异常可以拒绝单条消息且连接不受影响SDK 自带的 OpenTelemetry 追踪本身就是一个中间件且默认已在列表中参考 OpenTelemetry整套中间件接口目前仍是临时 APIprovisional请用它来观测而不要在其上构建长期依赖。至此我们覆盖了所有“包裹在请求外围”的机制至于一条请求是否有权执行则由 授权Authorization 机制来决定。【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址: https://gitcode.com/gh_mirrors/pythonsd/python-sdk创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考