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

在 Model Context Protocol（MCP）的 Python SDK 中，工具（Tool）的参数来自模型，而其余的一切——你正在处理的请求、你所在的服务器、一条通回客户端的通道——都来自同一个对象：Context。本文基于官方文档 i18n/de/pages/handlers/context.md（英文原版见 docs/handlers/context.md），深入讲解Context的注入机制、它对模型不可见的原理、它提供的全部能力（读资源、报进度、elicitation、回写会话），并结合仓库源码验证其底层实现。读完你将能够：在自己的 MCP 服务器中通过类型注解声明式地使用Context，让工具读取服务器自有资源，并在运行时动态注册工具后主动通知客户端刷新列表。

Context：你不构造它，你请求它

Context的设计哲学是声明式注入：你既不构造它，也不配置它，只需在函数签名里"开口要"即可。SDK 会为每一个请求构建一个全新的Context并传入。

熟悉 FastAPI 的开发者会立刻认出这种模式：声明一个以框架自身类型（FastAPI 中是Request，这里是Context）注解的参数，框架负责注入。无需注册、无需配置——类型注解本身就是全部机制。

从源码看，SDK 中Context是一个 PydanticBaseModel（见 src/mcp/server/mcpserver/context.py#L32），它内部持有ServerRequestContext（每请求的原始记录，定义于 src/mcp/server/context.py#L30-L49）与MCPServer实例的引用，从而把会话、请求元数据、lifespan 上下文等能力统一暴露给处理函数。

请求 Context：注解即机制

向任意工具添加一个以Context注解的参数即可。以下示例取自 docs_src/context/tutorial001.py：

from 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并注入，请求之间互不共享。
参数名无关紧要：ctx、context、c都可以，SDK 通过注解（annotation）而非参数名来识别它。源码中的文档注释也明确说明："The context parameter name can be anything as long as it's annotated with Context"（src/mcp/server/mcpserver/context.py#L61-L62）。
资源（Resource）和提示（Prompt）处理器同样可以声明Context，方式完全一致。
ctx.request_id是你当前函数正在处理的请求的 ID。源码中它被转换为字符串返回（src/mcp/server/mcpserver/context.py#L291-L294）。

对模型不可见

这是最需要内化的部分。下面是tools/list为search_books报告的输入模式（input schema）：

{ "type": "object", "properties": { "query": {"title": "Query", "type": "string"} }, "required": ["query"], "title": "search_booksArguments" }

只有一个属性。ctx不是参数：它永远不会出现在模式中，模型永远不会被告知它的存在，也没有任何客户端能够填写它。它是你与 SDK 之间的契约，在线上（wire）不可见。

从实现上可以印证这一点：Context通过类型注解被识别并在调用前被剥离（injected），不会进入函数参数的模式推导过程；工具参数的 JSON Schema 只由真正的业务参数生成。

动手试一下

使用 MCP Inspector 启动服务器：

uv run mcp dev server.py

search_books的表单只有一个query字段。用dune调用它：

[request 3] Found 3 books matching 'dune'.

这个数字恰好是这次请求的编号。再次调用工具，数字会变化：每个请求都有自己的Context。

Context 提供什么

注入的对象很小巧。除了request_id，它还包括：

await ctx.read_resource(uri)：在工具内部读取服务器自己的资源（见下文专节）。
await ctx.report_progress(progress, total, message)：在长时间调用期间向调用方流式回报进度。完整故事见 Fortschritt（进度）（英文版 docs/handlers/progress.md）。源码中它转发到会话的report_progress（src/mcp/server/mcpserver/context.py#L113-L121）。
await ctx.elicit(message, schema)与await ctx.elicit_url(...)：暂停工具，向 Host 端的人提出一个问题。这是 Elicitation（elicitation）（英文版 docs/handlers/elicitation.md）的主题。源码中elicit通过elicit_with_validation实现，elicit_url则引导用户跳转到外部 URL 完成带外交互（如敏感凭据收集、OAuth 授权、支付流程），并在完成后通过ctx.session.send_elicit_complete(elicitation_id)通知客户端（src/mcp/server/mcpserver/context.py#L189-L255）。
ctx.session：与当前客户端的会话的服务端一侧。你发送给客户端的通知都在这里；最后一节会用到它。
ctx.headers：传输层携带的请求头，stdio 下为None。读取自定义头用(ctx.headers or {}).get("x-...")。头部是客户端提供的输入——适合用于 locale 或 feature flag，永远不能当作身份（identity）。源码中的headers属性正是从传输层请求对象上取出的（src/mcp/server/mcpserver/context.py#L281-L289）。
ctx.request_context：原始的每请求数据记录。你最常取用的字段是lifespan_context，即你的启动代码（startup code）通过 yield 交付的对象（见 Lifespan（生命周期）（英文版 docs/handlers/lifespan.md））。该字段定义于ServerRequestContext.lifespan_context（src/mcp/server/context.py#L41）。

日志（logging）刻意不在此清单中。服务器的日志应使用 Python 的logging模块，就像任何其他 Python 程序一样。原因见 Logging（日志）（英文版 docs/handlers/logging.md）的简短说明。值得注意的是，Context上的log/info/debug等方法已随 2026-07-28 协议版本废弃（见源码中的@deprecated标记，src/mcp/server/mcpserver/context.py#L257）。

提示：注入只发生在你注册的那个函数上。你的工具调用的辅助函数并不会获得自己的Context；请把ctx作为普通参数向下传递。不存在可以从别处取用的"当前上下文"（ambient current context）。

读取自有资源：工具与客户端共享同一事实来源

服务器的资源不只是给客户端用的，工具同样可以读取：

from 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 f"The catalog is organised into: {contents.content}"

（完整代码见 docs_src/context/tutorial002.py。）

ctx.read_resource通过**服务resources/read的同一个注册表（registry）**来解析 URI，因此工具得到的就是客户端会得到的东西：一个ReadResourceContents的可迭代对象，每个内容块一个。对于这个 URI 只有一个：

contents.content # 'fiction, non-fiction, poetry' contents.mime_type # 'text/plain'

content正是genres()返回的内容。单一事实来源：客户端浏览资源，你的工具消费资源，没有人需要复制这个字符串。
describe_catalog的唯一参数就是Context，因此它的输入模式完全没有属性。模型以{}调用它。

从源码看，ctx.read_resource实际上调用MCPServer.read_resource(uri, ...)，并做了一层额外的保护：如果资源返回了InputRequiredResult（2026-07-28 多轮往返流程），这里会抛出RuntimeError，提示应改用MCPServer.read_resource(uri, context)来接收并转发——因为ctx.read_resource只是内容读取器（src/mcp/server/mcpserver/context.py#L151-L187）。

告知客户端列表已变更：运行时动态注册

服务器能提供什么，并不在导入（import）时就固定。可以在运行时注册工具，然后告知客户端：

from 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 f"In {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."

（完整代码见 docs_src/context/tutorial003.py。）

mcp.add_tool(recommend_book)把一个普通函数注册为工具：名称、描述和模式与@mcp.tool()推导的方式完全一致。从源码看，MCPServer.add_tool委托给工具管理器（见 src/mcp/server/mcpserver/server.py#L609 与 src/mcp/server/mcpserver/tools/tool_manager.py#L39）。
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 连接：订阅流与 notify_* 方法

在 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)

从源码看，这些方法通过SubscriptionBus（订阅总线）发布对应的事件对象（ToolsListChanged、PromptsListChanged、ResourcesListChanged、ResourceUpdated），其中notify_resource_updated以精确字符串匹配每个流的过滤器（src/mcp/server/mcpserver/context.py#L129-L149）。完整故事——包括跨副本水平扩展——见 Abonnements（订阅）（英文版 docs/handlers/subscriptions.md）。

验证：在任何人运行enable_recommendations之前，你所承诺的那个工具并不存在。强行调用它会得到模型可读的错误：
Unknown tool: recommend_book
运行enable_recommendations之后，完全相同的调用就成功了。工具列表是真正动态的：tools/list反映的是当下已注册的内容。

总结

在参数上以Context注解（用于工具、资源或提示），SDK 就会注入它。参数名由你决定。
它对模型不可见：输入模式永远只包含你真正的业务参数。
ctx.request_id标识当前请求；ctx.request_context.lifespan_context是你的启动代码 yield 出来的对象。
await ctx.read_resource(uri)让工具读取服务器自己的资源，与客户端共享同一事实来源。
ctx.session是回到客户端的通道：send_tool_list_changed()及其同族方法会提示客户端重新拉取你变更过的列表；在 2026-07-28 连接上则改用ctx.notify_*系列方法经由订阅流投递。
进度回报（report_progress）与 elicitation（elicit/elicit_url）同样从Context开始，各自有独立章节。