MCP Python SDK 服务端中间件Middleware实战观察、拒绝、改写每一条入站消息【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址: https://gitcode.com/gh_mirrors/pythonsd/python-sdkMiddleware是 python-sdkModel Context Protocol 官方 Python SDK服务端提供的一个单一而强大的拦截点它就是一个异步函数包裹服务器收到的每一条入站消息。本文以 docs/advanced/middleware.md 为骨架结合仓库源码与示例完整讲解它的 API、执行顺序、能做的四类事情以及它与 OpenTelemetry、ASGI 中间件的关系。读完后你将能写出计时/日志/限权中间件并理解它为什么是观测initialize握手与按调用者放行订阅的唯一钩子。中间件是什么整个 API 只有一行一个中间件就是一个异步函数签名固定为async def my_middleware(ctx, call_next): ...它被追加到服务器的中间件列表上server.middleware.append(my_middleware)这就是全部 API。不需要注册类、装饰器或配置对象——函数即中间件。SDK 分两层暴露这个列表内容完全相同高层MCPServer在构造时传入MCPServer(name, middleware[...])之后通过mcp.middleware属性访问底层Server直接暴露server.middleware列表。两种写法等价MCPServer.middleware属性在源码中就是直接转发到底层服务器的同一份列表见 src/mcp/server/mcpserver/server.py并且用户中间件被追加在 SDK 内置中间件OpenTelemetry、request-state 边界之后、按给定顺序最外层优先执行src/mcp/server/mcpserver/server.py。下面的示例使用底层Server如果Server(name, on_call_tool...)这种构造式处理器对你来说还很陌生建议先阅读 底层 Server 指南。警告provisional中间件列表在源码中明确标记为provisional——签名与语义可能在 2.x minor 版本中变化src/mcp/server/lowlevel/server.py。请用它来观察计时、日志、追踪和拒绝消息不要把它当成服务器赖以运转的根基。一个计时中间件从示例到逐行解读仓库中的完整示例位于 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( 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。ctx.method是原始方法字符串如tools/callctx.params是尚未经过任何校验的原始参数。ServerRequestContext是一个dataclass携带session、lifespan_context、protocol_version、method、params、request_id、meta等字段src/mcp/server/context.py。call_next(ctx)运行链条的其余部分参数校验 → 处理器查找 → 你的处理器。返回它返回的东西响应就原样不动。try/finally是刻意设计的处理器抛出异常时异常会从call_next抛到你的中间件里所以失败的请求也会被计时——finally保证无论成功失败都记录耗时。server.middleware.append(...)注册。列表最外层优先执行即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或者在 legacy 会话上是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)异常穿过你的中间件抛向客户端。中间件内部能做的四件事按你该有多犹豫递增排序中间件的能力阶梯如下1. 观察Observe计时、计数、打日志。上面的log_timing就是标准形态。这是最安全、最推荐的用法。2. 拒绝Refuse不调用call_next(ctx)而是直接raise一个MCPError这条消息会被以 JSON-RPC 错误应答连接保持存活下一条消息照常通过。这就是服务器按调用者限制subscriptions/listen的方式。仓库中的完整门禁示例在 docs_src/subscriptions/tutorial006.py对应文档章节 Deciding who may watchfrom mcp_types import INVALID_REQUEST, SubscriptionsListenRequestParams from mcp.server.auth.middleware.auth_context import get_access_token from mcp.server.context import CallNext, HandlerResult, ServerRequestContext from mcp.server.mcpserver import MCPServer from mcp.shared.exceptions import MCPError # Who may see each file. Replace this table with a database or your RBAC system. ACCESS { files://report.pdf: {alice, bob}, files://payroll.csv: {carol}, } def can_access(user: str | None, uri: str) - bool: return user is not None and user in ACCESS.get(uri, set()) async def gate_subscriptions(ctx: ServerRequestContext, call_next: CallNext) - HandlerResult: if ctx.method subscriptions/listen: params SubscriptionsListenRequestParams.model_validate(ctx.params or {}, by_nameFalse) token get_access_token() user token.subject if token else None if not all(can_access(user, uri) for uri in params.notifications.resource_subscriptions or ()): raise MCPError(INVALID_REQUEST, not permitted to watch the requested resources) return await call_next(ctx) mcp MCPServer(Reports, middleware[gate_subscriptions])注意它用MCPServer(..., middleware[...])的构造式注入等价于 append。只有特定用户能订阅files://report.pdf其他人收到INVALID_REQUEST但连接不断。3. 重写Rewritectx是一个 dataclass因此可以import dataclasses await call_next(dataclasses.replace(ctx, params...))这样链条的其余部分拿到的是与客户端发送不同的参数。源码层面这是被显式支持的ServerRunner._inner在进入查找与处理器之前重新读取ctx.method与ctx.paramssrc/mcp/server/runner.py。但绝不要对initialize做这件事客户端拿到的结果是根据你重写后的参数构建的而服务器提交连接状态用的是原始线缆参数。两边可能以对协商结果的不同理解完成握手。4. 应答Answer不调用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警告死锁风险initialize是内联处理的——在你的中间件链返回之前服务器不会读取后续入站消息。因此在处理initialize时 await 一个服务器到客户端的请求ctx.session.send_request(...)即 elicitation会死锁连接你等待的响应永远不会被读取。fire-and-forget 的通知则没有问题。SDK 默认自带的那一个中间件SDK 恰好内置一个中间件而且它已经在你的服务器列表上了为每条消息发射一个 OpenTelemetry span 的那个。你不需要 append 它大多数时候也不用想它。在没有安装 exporter 之前它完全是无操作no-op并且它有专门的一页文档OpenTelemetry。从源码可以确认这一点底层Server构造时就把OpenTelemetryMiddleware()放进了默认列表src/mcp/server/lowlevel/server.py注释明确写着每个服务器默认发射每消息一个 SERVER span安装 OTel exporter 之前是无操作从列表中移除即可退出。它的实现位于 src/mcp/server/_otel.py会为每条入站消息产生一个以tools/call search_books这种方法目标命名的 span并携带mcp.method.name、mcp.protocol.version、jsonrpc.request.id仅请求等属性。相关链接完整 tracing 说明与关闭方法OpenTelemetry关闭时使用mcp._lowlevel_server.middleware[:] [...]过滤掉OpenTelemetryMiddleware实例与 ASGI 中间件的关系一个看 HTTP一个看 MCP如果你写过 ASGI 中间件这个形态你很熟悉Starlette 的(scope, receive, send)变成了(ctx, call_next)而且它运行在传输层之后、作用于解码后的消息而非原始 HTTP 请求。两者是可组合的挂在streamable_http_app()上的 Starlette 中间件看到的是 HTTP而本文的中间件看到的是 MCP。源码视角链条是如何组合的要理解最外层优先与重写生效看 src/mcp/server/runner.py 的_compose_server_middleware就一清二楚call inner for middleware in reversed(self.server.middleware): call partial(_apply_middleware, middleware, call) return callinner是真正的处理内核校验 → 查找处理器 → 调用 → 序列化请求与通知两条路径共用同一套组合逻辑_on_request与_on_notify所以同一条中间件链观察每一条入站消息反向遍历列表使得middleware[0]成为最外层、最先执行、最靠近网络的那一个组合出的可调用对象在调用时才接收ctxsrc/mcp/server/runner.py所以中间件可以用dataclasses.replace改写后的 ctx 继续下传。另外两处值得注意的实现细节请求路径上中间件短路返回的结果会经_dump_result规范化BaseModel/dict/None皆可src/mcp/server/runner.py且流水线不再修补——你返回什么信封客户端就收到什么信封initialize只在链条整体成功后才提交连接状态src/mcp/server/runner.py所以一个中途raise的中间件可以干净地否决握手而不留下半初始化状态。Recap一个中间件就是async (ctx, call_next) - result通过MCPServer(middleware[...])传入或追加到mcp.middleware在底层Server上追加到server.middleware。它包裹每一条到达服务器的入站消息server/discover、initialize、请求、通知、未知方法并最外层优先执行。ctx.request_id is None是区分通知与请求的方法。不调用call_next而是直接raise即可拒绝单条消息连接存活。SDK 自带的 OpenTelemetry 追踪本身就是一个中间件且已经在列表上详见 OpenTelemetry。整个表面都是 provisional用它来观察别在它之上构建地基。以上就是包装请求的一切。请求能不能真正运行则由 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）实战：观察、拒绝、改写每一条入站消息
MCP Python SDK 服务端中间件Middleware实战观察、拒绝、改写每一条入站消息【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址: https://gitcode.com/gh_mirrors/pythonsd/python-sdkMiddleware是 python-sdkModel Context Protocol 官方 Python SDK服务端提供的一个单一而强大的拦截点它就是一个异步函数包裹服务器收到的每一条入站消息。本文以 docs/advanced/middleware.md 为骨架结合仓库源码与示例完整讲解它的 API、执行顺序、能做的四类事情以及它与 OpenTelemetry、ASGI 中间件的关系。读完后你将能写出计时/日志/限权中间件并理解它为什么是观测initialize握手与按调用者放行订阅的唯一钩子。中间件是什么整个 API 只有一行一个中间件就是一个异步函数签名固定为async def my_middleware(ctx, call_next): ...它被追加到服务器的中间件列表上server.middleware.append(my_middleware)这就是全部 API。不需要注册类、装饰器或配置对象——函数即中间件。SDK 分两层暴露这个列表内容完全相同高层MCPServer在构造时传入MCPServer(name, middleware[...])之后通过mcp.middleware属性访问底层Server直接暴露server.middleware列表。两种写法等价MCPServer.middleware属性在源码中就是直接转发到底层服务器的同一份列表见 src/mcp/server/mcpserver/server.py并且用户中间件被追加在 SDK 内置中间件OpenTelemetry、request-state 边界之后、按给定顺序最外层优先执行src/mcp/server/mcpserver/server.py。下面的示例使用底层Server如果Server(name, on_call_tool...)这种构造式处理器对你来说还很陌生建议先阅读 底层 Server 指南。警告provisional中间件列表在源码中明确标记为provisional——签名与语义可能在 2.x minor 版本中变化src/mcp/server/lowlevel/server.py。请用它来观察计时、日志、追踪和拒绝消息不要把它当成服务器赖以运转的根基。一个计时中间件从示例到逐行解读仓库中的完整示例位于 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( 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。ctx.method是原始方法字符串如tools/callctx.params是尚未经过任何校验的原始参数。ServerRequestContext是一个dataclass携带session、lifespan_context、protocol_version、method、params、request_id、meta等字段src/mcp/server/context.py。call_next(ctx)运行链条的其余部分参数校验 → 处理器查找 → 你的处理器。返回它返回的东西响应就原样不动。try/finally是刻意设计的处理器抛出异常时异常会从call_next抛到你的中间件里所以失败的请求也会被计时——finally保证无论成功失败都记录耗时。server.middleware.append(...)注册。列表最外层优先执行即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或者在 legacy 会话上是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)异常穿过你的中间件抛向客户端。中间件内部能做的四件事按你该有多犹豫递增排序中间件的能力阶梯如下1. 观察Observe计时、计数、打日志。上面的log_timing就是标准形态。这是最安全、最推荐的用法。2. 拒绝Refuse不调用call_next(ctx)而是直接raise一个MCPError这条消息会被以 JSON-RPC 错误应答连接保持存活下一条消息照常通过。这就是服务器按调用者限制subscriptions/listen的方式。仓库中的完整门禁示例在 docs_src/subscriptions/tutorial006.py对应文档章节 Deciding who may watchfrom mcp_types import INVALID_REQUEST, SubscriptionsListenRequestParams from mcp.server.auth.middleware.auth_context import get_access_token from mcp.server.context import CallNext, HandlerResult, ServerRequestContext from mcp.server.mcpserver import MCPServer from mcp.shared.exceptions import MCPError # Who may see each file. Replace this table with a database or your RBAC system. ACCESS { files://report.pdf: {alice, bob}, files://payroll.csv: {carol}, } def can_access(user: str | None, uri: str) - bool: return user is not None and user in ACCESS.get(uri, set()) async def gate_subscriptions(ctx: ServerRequestContext, call_next: CallNext) - HandlerResult: if ctx.method subscriptions/listen: params SubscriptionsListenRequestParams.model_validate(ctx.params or {}, by_nameFalse) token get_access_token() user token.subject if token else None if not all(can_access(user, uri) for uri in params.notifications.resource_subscriptions or ()): raise MCPError(INVALID_REQUEST, not permitted to watch the requested resources) return await call_next(ctx) mcp MCPServer(Reports, middleware[gate_subscriptions])注意它用MCPServer(..., middleware[...])的构造式注入等价于 append。只有特定用户能订阅files://report.pdf其他人收到INVALID_REQUEST但连接不断。3. 重写Rewritectx是一个 dataclass因此可以import dataclasses await call_next(dataclasses.replace(ctx, params...))这样链条的其余部分拿到的是与客户端发送不同的参数。源码层面这是被显式支持的ServerRunner._inner在进入查找与处理器之前重新读取ctx.method与ctx.paramssrc/mcp/server/runner.py。但绝不要对initialize做这件事客户端拿到的结果是根据你重写后的参数构建的而服务器提交连接状态用的是原始线缆参数。两边可能以对协商结果的不同理解完成握手。4. 应答Answer不调用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警告死锁风险initialize是内联处理的——在你的中间件链返回之前服务器不会读取后续入站消息。因此在处理initialize时 await 一个服务器到客户端的请求ctx.session.send_request(...)即 elicitation会死锁连接你等待的响应永远不会被读取。fire-and-forget 的通知则没有问题。SDK 默认自带的那一个中间件SDK 恰好内置一个中间件而且它已经在你的服务器列表上了为每条消息发射一个 OpenTelemetry span 的那个。你不需要 append 它大多数时候也不用想它。在没有安装 exporter 之前它完全是无操作no-op并且它有专门的一页文档OpenTelemetry。从源码可以确认这一点底层Server构造时就把OpenTelemetryMiddleware()放进了默认列表src/mcp/server/lowlevel/server.py注释明确写着每个服务器默认发射每消息一个 SERVER span安装 OTel exporter 之前是无操作从列表中移除即可退出。它的实现位于 src/mcp/server/_otel.py会为每条入站消息产生一个以tools/call search_books这种方法目标命名的 span并携带mcp.method.name、mcp.protocol.version、jsonrpc.request.id仅请求等属性。相关链接完整 tracing 说明与关闭方法OpenTelemetry关闭时使用mcp._lowlevel_server.middleware[:] [...]过滤掉OpenTelemetryMiddleware实例与 ASGI 中间件的关系一个看 HTTP一个看 MCP如果你写过 ASGI 中间件这个形态你很熟悉Starlette 的(scope, receive, send)变成了(ctx, call_next)而且它运行在传输层之后、作用于解码后的消息而非原始 HTTP 请求。两者是可组合的挂在streamable_http_app()上的 Starlette 中间件看到的是 HTTP而本文的中间件看到的是 MCP。源码视角链条是如何组合的要理解最外层优先与重写生效看 src/mcp/server/runner.py 的_compose_server_middleware就一清二楚call inner for middleware in reversed(self.server.middleware): call partial(_apply_middleware, middleware, call) return callinner是真正的处理内核校验 → 查找处理器 → 调用 → 序列化请求与通知两条路径共用同一套组合逻辑_on_request与_on_notify所以同一条中间件链观察每一条入站消息反向遍历列表使得middleware[0]成为最外层、最先执行、最靠近网络的那一个组合出的可调用对象在调用时才接收ctxsrc/mcp/server/runner.py所以中间件可以用dataclasses.replace改写后的 ctx 继续下传。另外两处值得注意的实现细节请求路径上中间件短路返回的结果会经_dump_result规范化BaseModel/dict/None皆可src/mcp/server/runner.py且流水线不再修补——你返回什么信封客户端就收到什么信封initialize只在链条整体成功后才提交连接状态src/mcp/server/runner.py所以一个中途raise的中间件可以干净地否决握手而不留下半初始化状态。Recap一个中间件就是async (ctx, call_next) - result通过MCPServer(middleware[...])传入或追加到mcp.middleware在底层Server上追加到server.middleware。它包裹每一条到达服务器的入站消息server/discover、initialize、请求、通知、未知方法并最外层优先执行。ctx.request_id is None是区分通知与请求的方法。不调用call_next而是直接raise即可拒绝单条消息连接存活。SDK 自带的 OpenTelemetry 追踪本身就是一个中间件且已经在列表上详见 OpenTelemetry。整个表面都是 provisional用它来观察别在它之上构建地基。以上就是包装请求的一切。请求能不能真正运行则由 Authorization授权 决定。【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址: https://gitcode.com/gh_mirrors/pythonsd/python-sdk创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考