免费获取学习方案
ARTICLE DETAIL

资讯详情

深耕编程基础知识与建站技术分享的一线实战洞察。

MCP Python SDK 深入解析:Context 注入机制与请求级服务端能力

MCP Python SDK 深入解析:Context 注入机制与请求级服务端能力 MCP Python SDK 深入解析Context 注入机制与请求级服务端能力【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址: https://gitcode.com/gh_mirrors/pythonsd/python-sdk导读在 MCP Python SDK官方 Model Context Protocol Python SDK源码位于本仓库src/mcp/中你写的 tool 函数其参数来自模型model而其余一切——你正在服务的请求、你所在的服务器、以及回话给客户端的方式——都来自同一个对象Context。本文以 docs/handlers/context.md 为核心系统讲解如何通过类型注解请求Context、它对模型为何不可见、它提供哪些能力读取资源、上报进度、发起 elicitation、发送变更通知等并结合仓库源码验证其底层实现机制。什么是 Context你不需要构造它只需要向 SDK 索取一个 tool 的 arguments 来自模型。但你在处理这个请求时常常还需要更多信息当前请求的 ID、服务端自己的资源、与客户端通信的通道等。SDK 的设计是——把这些全部塞进一个对象即Context。它有两个关键特性不需要你构造你从不写Context(...)来手动创建它。不需要你配置你也不需要做任何注册或依赖注入声明。你唯一要做的事情是索取——在函数签名里声明一个以Context注解的参数。如何请求 Context注解即一切给任意 tool 添加一个以Context注解的参数即可。完整示例见 docs_src/context/tutorial001.pyfrom mcp.server import MCPServer from mcp.server.mcpserver import Context mcp MCPServer(Bookshop) mcp.tool() def search_books(query: str, ctx: Context) - str: Search the catalog by title or author. return f[request {ctx.request_id}] Found 3 books matching {query!r}.需要注意的机制细节SDK 为每个请求构建全新的Context并注入。每次调用 tool你拿到的都是一个独立的、绑定到本次请求的对象。参数名不重要。叫ctx、context、c都可以——SDK 是根据类型注解annotation来识别的而不是根据参数名。源码中的_is_context_annotation()位于 src/mcp/server/mcpserver/resolve.py会检查参数注解是否为Context或其子类def _is_context_annotation(annotation: Any) - bool: if get_origin(annotation) is Annotated: annotation get_args(annotation)[0] candidates get_args(annotation) if get_origin(annotation) is not None else (annotation,) return any(isinstance(c, type) and issubclass(c, Context) for c in candidates)resources 和 prompts 也可以用同样方式声明Context参数注入机制是通用的。ctx.request_id即你当前正在服务的这条请求的 ID。如果你用过 FastAPI会立刻认出这套模式声明一个框架自有类型的参数那里是Request这里是Context框架负责注入。无需注册、无需配置类型注解本身就是全部机制。Context类的 docstring见 src/mcp/server/mcpserver/context.py也明确说明了这一点工具函数中参数名可以是任何名字只要以Context注解即可不需要 Context 的工具完全可以省略该参数。对模型不可见wire 上的契约只属于你和 SDK这一点值得深入理解。对于上面的search_bookstools/list报告的 input schema 是这样的{ type: object, properties: { query: {title: Query, type: string} }, required: [query], title: search_booksArguments }只有query一个 property。ctx不是一个参数它永远不会出现在 schema 里模型永远不会被告知它的存在任何客户端都无法填充它。它是你与 SDK 之间的一份契约在 wire 上完全不可见。从源码实现看Context参数与工具的真实参数走的是完全不同的注入路径Context由 server.py 在处理请求时构建如context Context(request_contextctx, mcp_serverself, ...)而工具的其他参数才来自模型传来的参数。试一试用 MCP Inspector 验证用 MCP Inspector 启动服务器uv run mcp dev server.pysearch_books的表单里只有query一个字段。用dune调用它[request 3] Found 3 books matching dune.输出中的数字是这次请求碰巧是第几个请求。再次调用 tool这个数字会变化——因为每个请求都获得属于自己的Contextrequest_id是每次请求独立的。Context 给你什么能力清单注入的对象很小。除了request_id之外它提供以下能力全部可对照 src/mcp/server/mcpserver/context.py 的实现能力签名 / 形态用途与备注读取自己的资源await ctx.read_resource(uri)在 tool 内部读取服务器自己注册的 resource下一节详述上报进度await ctx.report_progress(progress, total, message)长调用期间向调用方持续回传进度完整说明见 Progress发起 elicitationawait ctx.elicit(message, schema)与await ctx.elicit_url(...)暂停 tool向用户提问见 Elicitation会话通道ctx.session与当前客户端对话的服务端一侧你发给客户端的 notification 都在这里最后一节会用到请求头ctx.headerstransport 携带的请求 headersstdio 下为None原始请求记录ctx.request_context每个请求的原始记录最常用字段是lifespan_context——你的 startup 代码 yield 出来的对象见 Lifespan补充说明ctx.headers用(ctx.headers or {}).get(x-...)读取自定义 header。源码中headers属性context.py的实现是直接从request_context.request上取headers因此 HTTP 类 transport 会填充它stdio 下为None。headers 是客户端提供的输入——适合做 locale、feature flag 之类的读取但绝不能用来做身份认证identity。ctx.request_context.lifespan_context这是你获取启动时代码 yield 的对象的唯一途径典型的用法是在 lifespan 中初始化共享状态数据库连接、模型客户端等在 tool 里通过它访问。logging刻意不在这个清单里服务器应该用 Python 标准的logging模块打日志就像任何其他 Python 程序一样。原因见 Logging。提示注入只发生在你注册的那个函数上。你的 tool 调用的某个 helper 函数不会自动获得自己的Context——没有所谓的当前上下文ambient current context可以从任何地方取到。你需要把ctx作为普通参数显式往下传。从源码结构还可以看到Context还暴露了其他对高级场景有用的属性例如protocol_version协商后的协议版本、client_capabilities客户端声明的能力等这些在写需要感知客户端能力的逻辑时非常有用。读取你自己的资源tool 与 resource 共享同一套真相服务器注册的 resources 不只是给客户端用的tool 也可以在内部读取它们。完整示例见 docs_src/context/tutorial002.pyfrom mcp.server import MCPServer from mcp.server.mcpserver import Context mcp MCPServer(Bookshop) mcp.resource(catalog://genres) def genres() - str: The genres the catalog is organised into. return fiction, non-fiction, poetry mcp.tool() async def describe_catalog(ctx: Context) - str: Describe how the catalog is organised. [contents] await ctx.read_resource(catalog://genres) return fThe catalog is organised into: {contents.content}关键点ctx.read_resource通过与resources/read相同的 registry解析 URI所以 tool 拿到的东西和客户端拿到的是完全一样的一个ReadResourceContents的 iterable每个 content block 一个元素。对于这个 URI 只有一个contents.content # fiction, non-fiction, poetry contents.mime_type # text/plaincontent就是genres()返回的原样字符串。这保证了单一事实来源one source of truth客户端浏览 resource你的 tool 消费它没有人复制一份字符串副本。describe_catalog的唯一参数就是Context因此它的 input schema一个 property 都没有——模型会以{}调用它。从源码看read_resource最终调用的是MCPServer.read_resource(uri, ...)server.py 附近与resources/read走的是同一条资源解析管线这正是tool 与 client 拿到一致内容的实现保证。告诉客户端列表变了运行时注册 tool 与变更通知服务器提供什么并不在 import 时就被锁死。你可以在运行时注册 tool然后通知客户端。完整示例见 docs_src/context/tutorial003.pyfrom mcp.server import MCPServer from mcp.server.mcpserver import Context mcp MCPServer(Bookshop) def recommend_book(genre: str) - str: Recommend a book in the given genre. return fIn {genre}, try Dune. mcp.tool() async def enable_recommendations(ctx: Context) - str: Switch on the recommendation tool. mcp.add_tool(recommend_book) await ctx.session.send_tool_list_changed() return Recommendations are now available.mcp.add_tool(recommend_book)把一个普通函数注册为 toolname、description 和 schema 的推导方式与mcp.tool()装饰器完全一致。从 server.py 的实现看add_tool支持name、title、description、annotations、icons、meta、structured_output等参数不传时 name 默认取函数名。await ctx.session.send_tool_list_changed()发送notifications/tools/list_changed。收到该通知的客户端会重新调用tools/list从而看到recommend_book。这个家族的兄弟方法还有send_resource_list_changed()——资源列表变化send_prompt_list_changed()——提示词列表变化send_resource_updated(uri)——某个具体资源的内容变化。2026-07-28 协议下的订阅流通知在 2026-07-28 协议连接上客户端只会在自己打开的subscriptions/listen流上收到变更通知所以上面的send_*方法到达不了那些流。Context的 publish 方法可以同时投递到每一个已订阅的流await ctx.notify_tools_changed()await ctx.notify_prompts_changed()await ctx.notify_resources_changed()await ctx.notify_resource_updated(uri)从源码看context.py这些notify_*方法通过SubscriptionBusself._bus.publish(...)把对应事件广播给所有subscriptions/listen订阅者。完整的故事包括跨副本 scale out见 Subscriptions。验证动态工具列表在任何人运行enable_recommendations之前你承诺的那个 tool 并不存在。这时调用它会得到一个模型能读懂的报错Unknown tool: recommend_book运行enable_recommendations之后同样的调用就成功了。这说明工具列表是真正动态的tools/list反映的是此时此刻已注册的工具。小结在 tool、resource 或 prompt 中把某个参数注解为ContextSDK 就会注入它。参数名随你起。它对模型不可见input schema 里永远只有你的真实参数。ctx.request_id标识当前请求ctx.request_context.lifespan_context是你的 startup 代码 yield 出来的对象。await ctx.read_resource(uri)让 tool 读取服务器自己的资源与客户端走同一条资源管线。ctx.session是回到客户端的通道send_tool_list_changed()及其兄弟方法通知客户端重新拉取你改过的列表订阅流场景下则用notify_*系列发布方法。progress reporting 和 elicitation 也从Context出发各自有独立文档Progress、Elicitation。而模型永远看不到、由你自己的函数填充的参数属于 Dependencies 的范畴——那是另一套注入机制与Context互为补充。把两者结合起来你就能写出既干净又强大的 MCP 服务器。【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址: https://gitcode.com/gh_mirrors/pythonsd/python-sdk创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考
返回列表