MCP Python SDK 依赖注入实战:用 Resolve 让工具参数对模型不可见
【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址: https://gitcode.com/gh_mirrors/pythonsd/python-sdk
工具(tool)的参数由模型(LLM)提供,但有些值绝不应该来自模型:从你的记录里查出来的价格、只有人能给出的确认、任何模型靠编造就可能搞错的东西。本文介绍 python-sdk(Model Context Protocol 官方 Python SDK)中MCPServer提供的**依赖注入(Dependencies)**机制:用Annotated[T, Resolve(fn)]声明参数由你自己的函数填充,SDK 在工具运行前自动调用该函数。读完本文,你将掌握如何声明单个依赖、让依赖互相嵌套、在必须时向用户提问(Elicit)、以及向客户端请求 LLM 采样与 roots(Sample/ListRoots),并理解这些能力在 2026-07-28 与 2025-11-25 两代协议下的行为差异。
核心概念:什么是 Dependency
MCP 工具的参数来自模型。但有一类值,模型永远不该负责提供——它们一旦被模型"脑补",就会出错:
- 从你的业务记录中查出的价格、库存;
- 只有真人才能给出的确认(如"确定要下这个订单吗?");
- 任何身份、权限类信息。
**Dependency(依赖)**就是由你自己的函数填充的参数:你给参数加上注解、指明函数名,SDK 会在工具 body 执行之前调用该函数,并把返回值注入参数。
如果你用过 FastAPI,这就是它的Depends。同样的思路、同样的理由:函数声明自己需要什么,框架负责提供,所有 wiring 都留在类型注解里,无需任何注册表。
该机制在源码中对应 resolve.py 模块(Resolve、Elicit、Sample、ListRoots四个 marker 均定义于此),并在 test_dependencies.py 中有逐条可验证的测试覆盖。
声明一个依赖:Annotated[T, Resolve(fn)]
把参数的类型包进Annotated[...],并加上Resolve(fn)即可。以书店铺货为例,完整代码见 tutorial001.py:
from typing import Annotated from pydantic import BaseModel from mcp.server import MCPServer from mcp.server.mcpserver import Resolve mcp = MCPServer("Bookshop") INVENTORY = {"Dune": 7, "Neuromancer": 0} class Stock(BaseModel): title: str copies: int async def check_stock(title: str) -> Stock: return Stock(title=title, copies=INVENTORY.get(title, 0)) @mcp.tool() async def reserve_book(title: str, stock: Annotated[Stock, Resolve(check_stock)]) -> str: """Reserve a copy of a book.""" if stock.copies == 0: return f"{title!r} is out of stock." return f"Reserved {title!r} ({stock.copies - 1} copies left)."理解这段代码的三个关键点:
check_stock是resolver:一个普通函数,SDK 会在reserve_book之前运行它,其返回值成为stock参数;- resolver 的
title参数就是工具自己的title参数,按名称匹配。resolver 看到的正是工具 body 将看到的同一份已校验(validated)值; - 工具 body 一开始就拿到一个已存在的
Stock——工具里没有查询代码,没有"万一查不到怎么办"的前置处理。
对模型不可见
这是最值得内化的部分。下面是tools/list为reserve_book报告的输入 schema:
{ "type": "object", "properties": { "title": {"title": "Title", "type": "string"} }, "required": ["title"], "title": "reserve_bookArguments" }只有一个 property。和 Context 一样,被 resolve 的参数是你与 SDK 之间的约定:stock不在 schema 里,模型永远不会被告知它的存在,即使客户端强行发送stock值也会被忽略。resolver 的值是工具唯一能收到的值。
最后这一点正是关键所在:模型无法提供的参数,模型也就无法弄错。测试 test_dependencies.py 中test_a_client_supplied_value_for_a_resolved_parameter_is_ignored验证了这一点——客户端即使传入stock: {"copies": 999},工具收到的仍是 resolver 自己算出的值。
动手尝试
用 MCP Inspector 启动服务器:
uv run mcp dev server.pyreserve_book的表单里只有一个title字段,stock根本不在上面。用Dune调用:
Reserved 'Dune' (6 copies left).工具 body 没有做过任何查询:check_stock先运行,返回的Stock作为参数送达。试试Neuromancer,同一个 resolver 会给工具递上一个零库存。
提示:你也可以在工具 body 里直接调用check_stock(title)。但当一个值值得比"一次辅助调用"更高的待遇时,就把它声明为 dependency:每个需要库存的工具都声明同一个参数,而无论多少工具声明它,SDK 每次调用最多只运行一次该 resolver。接下来的小节会补充其余部分:互相依赖的 resolvers,以及向用户提问的 resolvers。
依赖的依赖:resolver 构成 DAG
resolver 可以用同样的注解声明自己的依赖。见 tutorial002.py:
from typing import Annotated from pydantic import BaseModel from mcp.server import MCPServer from mcp.server.mcpserver import Resolve mcp = MCPServer("Bookshop") INVENTORY = {"Dune": 7, "Neuromancer": 0} class Stock(BaseModel): title: str copies: int async def check_stock(title: str) -> Stock: return Stock(title=title, copies=INVENTORY.get(title, 0)) async def estimate_delivery(stock: Annotated[Stock, Resolve(check_stock)]) -> str: return "tomorrow" if stock.copies > 0 else "in 2-3 weeks" @mcp.tool() async def order_book( title: str, stock: Annotated[Stock, Resolve(check_stock)], delivery: Annotated[str, Resolve(estimate_delivery)], ) -> str: """Order a book from the shop.""" if stock.copies == 0: return f"{title!r} is on backorder; it would arrive {delivery}." return f"Ordered {title!r}; it arrives {delivery}."三个要点:
estimate_delivery依赖check_stock。SDK 按图(graph)顺序执行:先 stock,再 estimate,最后工具;stock和delivery最终都需要check_stock,但它每次调用只运行一次——一次库存查询,两个消费者;- 没有任何需要注册的东西,注解本身就是这张图。
"每次调用一次"不是空话,可以自己验证:在check_stock里放一个print,从 Inspector 调用order_book——每次调用只打一行。两个消费者,一次查询。测试test_a_shared_dependency_runs_once_per_call用计数器库存对象证明了这一点,并且确认记忆化(memoization)是按调用而非按服务器生命周期生效:下一次tools/call会再次运行check_stock。
坏图在注册时失败,而不是调用中途
SDK 在工具注册时分析依赖图,而不是调用时。以下两种情况都会在启动时抛出InvalidSignature:
- 某个参数无法归类——既不是
Context,也不是Resolve(...),也不是某个工具参数的名字; - resolver 之间存在循环依赖。
服务器会在任何客户端连接之前就失败,错误信息中带有出问题的参数或 resolver 的名字。从源码看,这一分析实现在 resolve.py 的build_resolver_plans(第 347 行起):它递归遍历每个Resolve标记引用的函数,用stack检测环(第 365 行),无法分类的参数直接 raiseInvalidSignature(第 392 行)。
resolver 的参数如何解析
resolver 的参数与工具参数完全一样地解析:可以是另一个Resolve(...)、按名称取工具自己的参数,或Context——ctx.headers、lifespan 对象,全部可用。
需要注意一个安全点:在 HTTP transports 上,Context中包含ctx.headers。headers 是客户端提供的输入,与任何工具参数无异:用来传 locale 或 feature flag 没问题,但永远不要用来做身份识别。调用者是谁,应该由你的授权层(见 Authorization)决定,而不是由任何人都能设置的 header 决定。
另外,"每次调用一次"意味着下一次tools/call会重新运行check_stock。需要跨请求存活的资源——数据库连接池、HTTP client——应该放在 Lifespan 中,resolver 通过ctx.request_context.lifespan_context访问它。
只在必要时提问:Elicit
resolver 不一定要知道答案。它可以返回Elicit(message, Model),SDK 会替你运行 Elicitation 机制向用户提问。见 tutorial003.py:
from typing import Annotated from pydantic import BaseModel, Field from mcp.server import MCPServer from mcp.server.mcpserver import Elicit, Resolve mcp = MCPServer("Bookshop") INVENTORY = {"Dune": 7, "Neuromancer": 0} class Stock(BaseModel): title: str copies: int class Backorder(BaseModel): confirm: bool = Field(description="Order anyway and wait?") async def check_stock(title: str) -> Stock: return Stock(title=title, copies=INVENTORY.get(title, 0)) async def confirm_backorder( title: str, stock: Annotated[Stock, Resolve(check_stock)], ) -> Backorder | Elicit[Backorder]: if stock.copies > 0: return Backorder(confirm=True) # in stock: nothing to ask return Elicit(f"{title!r} is out of stock (2-3 weeks). Order anyway?", Backorder) @mcp.tool() async def order_book( title: str, stock: Annotated[Stock, Resolve(check_stock)], backorder: Annotated[Backorder, Resolve(confirm_backorder)], ) -> str: """Order a book from the shop.""" if not backorder.confirm: return "No order placed." if stock.copies == 0: return f"Backordered {title!r}; it ships in 2-3 weeks." return f"Ordered {title!r}."三个关键行为:
- 有库存:
confirm_backorder直接返回Backorder。没有问题,没有往返。用户只在答案真正重要时才被打断; - 无库存:SDK 发送 elicitation,按
Backorderschema 校验答案并注入。你的 resolver 从不接触协议; - 工具像读其他参数一样读
backorder.confirm。回答no也是一种回答:elicitation 以confirm=False被接受,工具运行,但不下单。提问成了前置条件(precondition),而不是工具 body 里的管道代码。
测试 test_dependencies.py 在legacy与auto两种模式下分别验证了:有库存时elicitation_callback绝不会被调用(test_an_in_stock_order_asks_no_question)、无库存时会收到确切的提问文案并按 accept/decline 正确处理。
用户拒绝或取消怎么办
如果用户干脆不回答——decline 或 cancel 问题会怎样?把注解写成Annotated[Backorder, Resolve(...)]时,工具 body 永远不会运行;调用会以模型可读的错误结果失败:
Error executing tool order_book: Resolver for parameter 'backorder' could not resolve: elicitation was decline这是前置条件的正确默认值:没有答案,就没有订单。当 decline 是工具想自行处理的结局时——比如跳过 backorder 但仍然推荐另一本书——应改用ElicitationResult[Backorder]注解,工具会收到完整的 accept/decline/cancel 结果并自行分支。ElicitationResult的三种成员(AcceptedElicitation、DeclinedElicitation、CancelledElicitation)定义在 resolve.py 中,_unwrap(第 645 行)正是产生上面那条错误信息的地方。更多细节见 Elicitation:schema 规则、三种回答、对话的客户端一侧。
提问的传输方式取决于协议版本
框架根据协商出的协议版本选择提问的传输方式,上面的代码在两种版本上完全一致:
- 2026-07-28 及之后:问题搭载在一次多轮往返(multi-round-trip)的
tools/call内部——服务器返回问题,客户端的elicitation_callback作答,Client替你重试调用(见 Multi-round-trip requests); - 2025-11-25 及之前:在调用中途发送一次同步的 elicitation 请求。
每个问题每次调用恰好被问一次——这是关于"问题"的保证,不是关于 resolver 的。在多轮往返形式下,每当调用在某问题之后恢复时,任何 resolver 都可能再次运行,因此return Elicit(...)之前的代码会在每一轮都执行;已记录的答案随后满足重复的问题,而不必再次打扰用户。已记录答案只在 resolver 提问时才会被查阅;一个不提问就作答的 resolver(如check_stock)总是提供自己计算出的值。由于每个答案都会与它的问题匹配回去,进行 elicitation 的 resolver 必须从工具参数和之前的答案确定性地构建问题。每次调用生成的值(如default_factory生成的 id、时间戳)每一轮都会重新生成,绝不能出现在答案要绑定的问题中——由这种易变数据构成的问题会让每个已记录答案都显得过期,于是服务器每轮重问,直到客户端的轮次上限结束调用。
问客户端,而不是问用户:Sample与ListRoots
Elicitation 是 resolver 能提出的三种问题之一,multi-round-trip 流程不允许其他问题。另外两种问题面向客户端而不是用户:
- 返回
Sample(...):通过客户端运行一次 LLM 调用(一次sampling/createMessage请求); - 返回
ListRoots():获取客户端当前的 roots。
两者都没有 accept/decline 结局;消费方直接注解结果类型——CreateMessageResult(当请求携带tools或tool_choice时为CreateMessageResultWithTools)或ListRootsResult。示例见 tutorial004.py:
from typing import Annotated from mcp.server import MCPServer from mcp.server.mcpserver import Resolve, Sample from mcp.types import CreateMessageResult, SamplingMessage, TextContent mcp = MCPServer("Bookshop") def suggest_title(genre: str) -> Sample: prompt = f"Suggest one {genre} book title. Answer with the title only." return Sample( [SamplingMessage(role="user", content=TextContent(type="text", text=prompt))], max_tokens=50, ) @mcp.tool() async def recommend_book( genre: str, suggestion: Annotated[CreateMessageResult, Resolve(suggest_title)], ) -> str: """Recommend a book in the given genre.""" title = suggestion.content.text if suggestion.content.type == "text" else "the classics" return f"Today's {genre} pick: {title}"要点:
- 框架像路由
Elicit一样路由它们:2026-07-28上在 multi-round-triptools/call内部,2025-11-25上通过独立的 server→client 请求。未声明的能力会以-32021协议错误拒绝该调用(sampling、roots、form 模式的elicitation;当请求携带tools或tool_choice时是sampling.tools)。能力校验实现在_require_capability(resolve.py 第 673 行); - 前面 info box 关于问题的所有论述原样适用:
Sample请求按精确渲染与其记录结果匹配,所以要基于工具参数和之前的答案确定性地构建;这样客户端只为每次工具调用付一次 LLM 调用的钱,而不是每轮付一次。记录结果在调用剩余部分随request_state一起传递,因此一个非常大的 completion 会让剩下的每一轮往返都更重; - 独立的 sampling 和 roots特性在 2026-07-28 已弃用(SEP-2577)。需要客户端模型的新服务器通过这个 carrier 提问;不需要的服务器应直接与 LLM provider 集成。
"none"以外的include_context值本身已弃用,应避免使用。
Sample的构造参数(max_tokens、system_prompt、include_context、temperature、stop_sequences、metadata、model_preferences、tools、tool_choice)定义在 resolve.py 第 131 行起,底层封装为CreateMessageRequestParams。
总结
- 工具参数上的
Annotated[T, Resolve(fn)]:SDK 运行fn并注入其返回值; - 被 resolve 的参数对模型不可见,客户端无法提供。模型不该编造的值——价格、身份、权限——就该放在这里;
- resolver 的参数以同样方式解析:
Context、另一个Resolve(...)、或按名称取工具参数。无论有多少消费者,依赖图每轮最多运行每个 resolver 一次;每个问题恰好被问一次,调用在某问题后恢复时任何 resolver 都可能再次运行; - 坏图在注册时以
InvalidSignature失败,而不是在调用中途; - 只在必要时返回
Elicit(message, Model)向用户提问。未包装的注解在 decline 时中止调用;ElicitationResult[T]让工具自行分支; - 返回
Sample(...)或ListRoots()向客户端请求 LLM completion 或 roots 列表;普通结果被直接注入。
服务器在启动时创建一次、并由 handler 访问的状态,见 Lifespan 页面。
【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址: https://gitcode.com/gh_mirrors/pythonsd/python-sdk
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考