MCP Python SDK 依赖注入实战用Resolve让工具参数脱离模型幻觉【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址: https://gitcode.com/gh_mirrors/pythonsd/python-sdk在 MCPModel 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-L106class Resolve: Marker for Annotated[T, Resolve(fn)]: fill the parameter by running fn. def __init__(self, fn: Callable[..., Any]) - None: self.fn fnResolve本身只是一个携带函数的标记真正的工作由 SDK 的解析管线完成注册时静态分析、调用时按依赖图求值。声明第一个依赖Annotated[T, Resolve(fn)]把参数的类型用Annotated[...]包起来并追加Resolve(fn)即可。完整的可运行示例见 docs_src/dependencies/tutorial001.pyfrom 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(titletitle, copiesINVENTORY.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 fReserved {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.pyInspector 中reserve_book的表单只有一个title字段stock无处可见。用Dune调用它Reserved Dune (6 copies left).工具体什么都没查check_stock先运行返回的Stock作为参数抵达。再试Neuromancer同一个解析器交给工具一个零库存对象工具返回缺货提示。提示你当然也可以在工具体内直接调用check_stock(title)。但当一个值值得被多个工具共享时就该把它声明为依赖——每个需要库存信息的工具声明同一个参数无论多少个工具声明它SDK 每次调用最多只运行该解析器一次。依赖的依赖解析器组成的 DAG解析器可以用同样的注解声明自己的依赖。完整示例见 docs_src/dependencies/tutorial002.pyfrom 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(titletitle, copiesINVENTORY.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 fOrdered {title!r}; it arrives {delivery}.estimate_delivery依赖check_stock。SDK 按图序执行先查库存再估算配送最后运行工具。stock和delivery最终都需要check_stock但它每次调用只运行一次。一次库存查询两个消费者。没有任何注册表需要维护。依赖图就是那些类型注解本身。每次调用一次不是口头承诺。源码中_Resolution持有按解析器身份去重的cachesrc/mcp/server/mcpserver/resolve.py#L456而 tests/docs_src/test_dependencies.py#L61-L81 用带计数的CountingInventory验证一次order_book调用只触发一次INVENTORY.get下一次tools/call会再次触发——记忆化按调用生效而不是按服务器进程生效。图在注册时分析坏图直接InvalidSignatureSDK 在工具注册时分析依赖图而不是在工具被调用时。一个无法归类的参数——既不是Context不是Resolve(...)也不是某个工具参数的名字——以及解析器之间的循环依赖都会在启动阶段抛出InvalidSignature。服务器在任何一个客户端连接之前就启动失败报错信息会点名出问题的参数或解析器。这一点在源码中有完整实现build_resolver_planssrc/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(...)、按名字匹配的工具参数、或完整的Contextctx.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.pyfrom 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(descriptionOrder anyway and wait?) async def check_stock(title: str) - Stock: return Stock(titletitle, copiesINVENTORY.get(title, 0)) async def confirm_backorder( title: str, stock: Annotated[Stock, Resolve(check_stock)], ) - Backorder | Elicit[Backorder]: if stock.copies 0: return Backorder(confirmTrue) # 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 fBackordered {title!r}; it ships in 2-3 weeks. return fOrdered {title!r}.三个关键行为有货时confirm_backorder直接返回Backorder。没有问题、没有额外往返。用户只在自己的回答真正重要时才被打扰。缺货时SDK 发送 elicitation 请求把用户回答按Backorder模型校验后注入。你的解析器全程不接触协议细节。工具像读取其他参数一样读取backorder.confirm。回答no也是一种回答elicitation 以confirmFalse被接受工具继续运行订单不生成。询问成为前置条件而不是工具体里的管道设施。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这正是前置条件的正确默认没有回答就没有订单。源码中_unwrapsrc/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_requiredsrc/mcp/server/mcpserver/resolve.py#L664-L670依据协商版本决定走InputRequiredResult批量传输还是老式同步回信道。关于每个问题每次调用只问一次有几点精细的保证这个保证是关于问题的不是关于解析器的。多轮形式下调用在问题之后每次恢复时任何解析器都可能重新运行——因此return Elicit(...)之前的代码会在每一轮都执行。已记录的回答会满足重复的问题而不再向用户追问。已记录的回答只在解析器真正提问时才会被参考像check_stock这样不问就答的解析器永远提供自己算出的值。因为每个回答都要匹配回它对应的问题发起 elicitation 的解析器必须从工具参数和之前的回答中确定性地推导问题。每次调用生成的值一个default_factory生成的 id、一个时间戳会在每一轮重新生成绝不能出现在需要绑定回答的问题里。用这类易变数据构造的问题会让每个已记录回答都显得过期于是服务器每轮都会重新提问直到客户端的轮次上限终止调用。询问客户端而不是用户Sample与ListRootsElicitation 只是解析器能提的三种问题之一而多轮流程不允许其他形式。另外两种面向的是客户端而非用户返回Sample(...)通过客户端运行一次 LLM 调用一次sampling/createMessage请求返回ListRoots()获取客户端当前的根目录roots列表。两者都没有 accept/decline 结果消费者直接在类型注解中写结果类型CreateMessageResult当请求携带tools或tool_choice时用CreateMessageResultWithTools或ListRootsResult。完整示例见 docs_src/dependencies/tutorial004.pyfrom 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 fSuggest one {genre} book title. Answer with the title only. return Sample( [SamplingMessage(roleuser, contentTextContent(typetext, textprompt))], max_tokens50, ) 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 fTodays {genre} pick: {title}行为要点框架像路由Elicit一样路由这两种请求2026-07-28版本走多轮tools/call2025-11-25版本走独立的服务器到客户端请求。如果客户端没有声明相应能力调用会被-32021协议错误拒绝——源码_require_capabilitysrc/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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考