企业官网建设流程全解析

MCP Python SDK 依赖注入实战：用Resolve让工具参数脱离模型幻觉

【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址: https://gitcode.com/gh_mirrors/pythonsd/python-sdk

在 MCP（Model Context Protocol）服务端开发中，工具的入参默认全部来自模型。但总有那么一类参数绝不应该由模型决定：从你业务库存里查出的实时价格、只有真人才能给出的审批确认、任何模型靠编造就可能出错的数值。本教程以当前仓库的官方文档 docs/handlers/dependencies.md 为骨架，结合 SDK 源码 src/mcp/server/mcpserver/resolve.py 与配套测试 tests/docs_src/test_dependencies.py，系统讲解 Python SDK 的依赖注入机制：如何用Annotated[...] + Resolve(fn)声明依赖、如何让解析器（resolver）彼此嵌套成依赖图、如何在必要时向用户提问（Elicit），以及如何向客户端请求 LLM 采样（Sample）与根目录列表（ListRoots）。读完你将掌握一套"模型无法伪造的服务端值注入"的完整实战方案。

什么是依赖注入：工具参数的另一个来源

工具的参数来自模型，这是 MCP 的默认模型。docs/handlers/dependencies.md开篇点出核心问题：有些值永远不该来自模型——从你的记录中查出的价格、只有人能给出的确认、任何模型靠编造就可能出错的数值。

依赖（Dependencies）就是由你自己的函数填充的参数：你为参数加上类型注解、指明函数，SDK 会在工具执行之前先调用它，把返回值注入参数。这是 FastAPI 用户非常熟悉的Depends模式在 MCP 世界的对应物——函数声明自己需要什么，框架负责提供，连接关系全部写在类型注解里。

从源码看，这一机制的核心标记定义在 src/mcp/server/mcpserver/resolve.py#L102-L106：

class Resolve: """Marker for `Annotated[T, Resolve(fn)]`: fill the parameter by running `fn`.""" def __init__(self, fn: Callable[..., Any]) -> None: self.fn = fn

Resolve本身只是一个携带函数的标记，真正的工作由 SDK 的解析管线完成：注册时静态分析、调用时按依赖图求值。

声明第一个依赖：Annotated[T, Resolve(fn)]

把参数的类型用Annotated[...]包起来，并追加Resolve(fn)即可。完整的可运行示例见 docs_src/dependencies/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参数。
• check_stock的title参数是工具自身的title参数，按名字匹配。解析器看到的正是工具体稍后将会看到的那个已验证值——两者完全一致。
• 工具体从"已经存在的一个Stock"开始工作：工具里没有查找库存的代码，也没有"万一没有怎么办"的前置判断。职责被干净地拆给了解析器。

值得注意的是，解析器既可以是async def也可以是同步函数：源码 src/mcp/server/mcpserver/resolve.py#L553-L556 显示，同步解析器会被调度到线程池（anyio.to_thread.run_sync）中执行，异步解析器则直接await。

对模型不可见

tools/list为reserve_book报告的输入 schema 是：

{ "type": "object", "properties": { "title": {"title": "Title", "type": "string"} }, "required": ["title"], "title": "reserve_bookArguments" }

只有一个属性。和 Context 对象 一样，被解析的参数是你与 SDK 之间的契约：stock不在 schema 中、模型永远不会被告知它的存在，而且即使某个客户端强行发送一个stock值，也会被忽略。解析器的值是工具唯一能收到的值。

最后这一点才是关键——模型无法提供的参数，就是模型不可能搞错的参数。价格、身份、权限这类"模型必须不能编造"的值，就属于这里。

这一点有测试直接背书。tests/docs_src/test_dependencies.py#L41-L46 的用例test_a_client_supplied_value_for_a_resolved_parameter_is_ignored故意向reserve_book传入{"title": "Dune", "stock": {"title": "Dune", "copies": 999}}，结果工具仍收到解析器算出的真实库存（6 本），伪造的 999 被静默丢弃。

动手尝试：用 MCP Inspector 验证

用 MCP Inspector 启动服务器：

uv run mcp dev server.py

Inspector 中reserve_book的表单只有一个title字段，stock无处可见。用Dune调用它：

Reserved 'Dune' (6 copies left).

工具体什么都没查：check_stock先运行，返回的Stock作为参数抵达。再试Neuromancer，同一个解析器交给工具一个零库存对象，工具返回缺货提示。

提示：你当然也可以在工具体内直接调用check_stock(title)。但当一个值值得被多个工具共享时，就该把它声明为依赖——每个需要库存信息的工具声明同一个参数，无论多少个工具声明它，SDK 每次调用最多只运行该解析器一次。

依赖的依赖：解析器组成的 DAG

解析器可以用同样的注解声明自己的依赖。完整示例见 docs_src/dependencies/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 按图序执行：先查库存，再估算配送，最后运行工具。
• stock和delivery最终都需要check_stock，但它每次调用只运行一次。一次库存查询，两个消费者。
• 没有任何注册表需要维护。依赖图就是那些类型注解本身。

"每次调用一次"不是口头承诺。源码中_Resolution持有按解析器身份去重的cache（src/mcp/server/mcpserver/resolve.py#L456），而 tests/docs_src/test_dependencies.py#L61-L81 用带计数的CountingInventory验证：一次order_book调用只触发一次INVENTORY.get，下一次tools/call会再次触发——记忆化按调用生效，而不是按服务器进程生效。

图在注册时分析：坏图直接InvalidSignature

SDK 在工具注册时分析依赖图，而不是在工具被调用时。一个无法归类的参数——既不是Context，不是Resolve(...)，也不是某个工具参数的名字——以及解析器之间的循环依赖，都会在启动阶段抛出InvalidSignature。服务器在任何一个客户端连接之前就启动失败，报错信息会点名出问题的参数或解析器。

这一点在源码中有完整实现：build_resolver_plans（src/mcp/server/mcpserver/resolve.py#L347-L404）递归分析每个解析器的参数，用调用栈检测循环（Resolver 'fn' has a cyclic dependency），对无法分类的参数抛出InvalidSignature。此外，find_resolved_parameters还会拒绝把Resolve(...)埋在 union 里的写法（如Annotated[T, Resolve(f)] | None），要求直接注解为Annotated[T, Resolve(...)]。

解析器的参数与工具参数遵循同一套解析规则：另一个Resolve(...)、按名字匹配的工具参数、或完整的Context（ctx.headers、lifespan 对象，全部可用）。

一个安全警告：ctx.headers是客户端输入

在 HTTP 传输上，Context包含ctx.headers。请求头是客户端提供的输入，和任何工具参数一样：适合用来传递语言区域或特性开关，但永远不能用来做身份认证。调用者是谁，应该来自你的授权层（见 授权），而不是任何人都能设置的请求头。

关于"每次调用一次"的边界

"每次调用一次"意味着下一个tools/call会重新运行check_stock。需要跨请求存活、由服务器在启动时构建一次的资源——数据库连接池、HTTP 客户端——属于 Lifespan 生命周期 的职责范围，解析器可以通过ctx.request_context.lifespan_context访问它。

只在必要时询问用户：Elicit

解析器并不非得知道答案。它可以返回Elicit(message, Model)，由 SDK 去问用户——这正是为你代跑的 Elicitation 信息征询机制。完整示例见 docs_src/dependencies/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 请求，把用户回答按Backorder模型校验后注入。你的解析器全程不接触协议细节。
• 工具像读取其他参数一样读取backorder.confirm。回答no也是一种回答：elicitation 以confirm=False被接受，工具继续运行，订单不生成。"询问"成为前置条件，而不是工具体里的管道设施。

Elicit在源码中定义于 src/mcp/server/mcpserver/resolve.py#L109-L118，它携带message（展示给用户的问题文本）和schema（回答必须符合的模型）。

用户不回答怎么办

如果用户拒绝或取消问题呢？当注解写成Annotated[Backorder, Resolve(...)]（未解包形式）时，工具体根本不会运行，调用以模型可读的错误结果失败：

Error executing tool order_book: Resolver for parameter 'backorder' could not resolve: elicitation was decline

这正是前置条件的正确默认：没有回答，就没有订单。源码中_unwrap（src/mcp/server/mcpserver/resolve.py#L645-L648）在 outcome 不是AcceptedElicitation时抛出ToolError，措辞与文档完全一致。配套测试 tests/docs_src/test_dependencies.py#L126-L140 在 legacy 与 auto 两种模式（对应新旧协议版本）下都验证了这条错误文本。

如果"拒绝"是工具想要自己处理的结果——比如跳过缺货预订但仍推荐另一本书——就把注解改为ElicitationResult[Backorder]，工具将收到完整的 accept/decline/cancel 三态结果自行分支。关于该形式、schema 规则、三种回答以及对话的客户端一侧，Elicitation 页面 有完整说明。

协议版本双轨：多轮tools/call与同步请求

框架根据协商出的协议版本选择问题的传输方式，上面的代码在两个时代完全相同：

• 2026-07-28 及之后：问题搭载在多轮往返（multi-round-trip）的tools/call中——服务器返回问题，客户端的elicitation_callback作答，Client替你重试调用（见 多轮请求）。
• 2025-11-25 及之前：调用中途发出一条同步的 elicitation 请求。

源码中_uses_input_required（src/mcp/server/mcpserver/resolve.py#L664-L670）依据协商版本决定走InputRequiredResult批量传输还是老式同步回信道。

关于"每个问题每次调用只问一次"有几点精细的保证：

• 这个保证是关于问题的，不是关于解析器的。多轮形式下，调用在问题之后每次恢复时，任何解析器都可能重新运行——因此return Elicit(...)之前的代码会在每一轮都执行。已记录的回答会满足重复的问题，而不再向用户追问。
• 已记录的回答只在解析器真正提问时才会被参考；像check_stock这样"不问就答"的解析器，永远提供自己算出的值。
• 因为每个回答都要匹配回它对应的问题，发起 elicitation 的解析器必须从工具参数和之前的回答中确定性地推导问题。每次调用生成的值（一个default_factory生成的 id、一个时间戳）会在每一轮重新生成，绝不能出现在需要绑定回答的问题里。用这类易变数据构造的问题会让每个已记录回答都显得过期，于是服务器每轮都会重新提问，直到客户端的轮次上限终止调用。

询问客户端而不是用户：Sample与ListRoots

Elicitation 只是解析器能提的三种问题之一，而多轮流程不允许其他形式。另外两种面向的是客户端而非用户：

• 返回Sample(...)：通过客户端运行一次 LLM 调用（一次sampling/createMessage请求）；
• 返回ListRoots()：获取客户端当前的根目录（roots）列表。

两者都没有 accept/decline 结果；消费者直接在类型注解中写结果类型：CreateMessageResult（当请求携带tools或tool_choice时用CreateMessageResultWithTools）或ListRootsResult。完整示例见 docs_src/dependencies/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版本走多轮tools/call，2025-11-25版本走独立的服务器到客户端请求。如果客户端没有声明相应能力，调用会被-32021协议错误拒绝——源码_require_capability（src/mcp/server/mcpserver/resolve.py#L673-L708）会校验sampling、roots、表单模式的elicitation能力，请求携带tools或tool_choice时还需sampling.tools，并以MISSING_REQUIRED_CLIENT_CAPABILITY错误码拒绝。
• 关于问题的一切规则原样适用：Sample请求按其精确渲染匹配已记录结果，所以要由工具参数和之前的回答确定性地构造它。这样客户端为一次工具调用支付一次 LLM 调用费用，而不是每轮一次。已记录结果在调用剩余部分随request_state携带——因此一个极大的补全结果会让后续每一轮往返都更重。
• Sample的构造参数（max_tokens、system_prompt、temperature、stop_sequences、metadata、model_preferences、tools、tool_choice等）定义在 src/mcp/server/mcpserver/resolve.py#L131-L157，并会校验"带工具使用"的消息合法性。
• 独立的 sampling 与 roots特性自 2026-07-28 起被弃用（SEP-2577）。需要客户端模型的新服务器应通过这一载体提问；不需要的应直接集成 LLM 提供商。除"none"之外的include_context值本身也已弃用，应避免使用。

总结

• 工具参数上的Annotated[T, Resolve(fn)]：SDK 运行fn并注入其返回值。
• 被解析的参数对模型不可见，客户端也无法提供它。模型绝不能编造的值——价格、身份、权限——都属于这里。
• 解析器的参数以同样方式解析：Context、另一个Resolve(...)、或按名字匹配的工具参数。无论有多少消费者，依赖图每轮最多运行每个解析器一次；每个问题恰好被问一次；调用在问题之后恢复时，任何解析器都可能再次运行。
• 坏图在注册时以InvalidSignature失败，而不是在调用中途失败。
• 需要询问用户时（且仅在你必须问时），返回Elicit(message, Model)。未解包的注解在拒绝时中止调用；ElicitationResult[T]则允许工具自行分支。
• 需要向客户端索要 LLM 补全或根目录列表时，返回Sample(...)或ListRoots()；纯结果被直接注入。

服务器在启动时一次性构建的状态、以及处理器如何访问它，是 Lifespan 生命周期 页面的主题。

【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址: https://gitcode.com/gh_mirrors/pythonsd/python-sdk