机制:工具中途提问的两种模式与客户端回调实现)
MCP Python SDK 中的 Elicitation征询机制工具中途提问的两种模式与客户端回调实现【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址: https://gitcode.com/gh_mirrors/pythonsd/python-sdkElicitation征询是 MCPModel Context Protocol中服务端在工具调用中途向用户提问、并把答案回收到同一个函数调用中的机制。本文以 python-sdk 仓库的官方文档docs/handlers/elicitation.md为主线结合docs_src/elicitation/下的四个完整教程与src/mcp/server/的底层实现系统讲解「表单模式form mode」与「URL 模式url mode」两种提问形态以及「Resolve 解析器resolver」与ctx.elicit()直连两种提问方式并给出可直接运行的服务端与客户端示例代码。什么是 Elicitation一个工具执行到一半只差一个答案就能完成工作时并不一定要以失败告终。Elicitation允许它中途提问在工具调用进行中用户会收到一个问题而用户的回答会回到同一个函数调用中继续执行。Elicitation 存在两种模式表单模式Form mode你需要一个具体的值如一个确认、一个日期、一个数量。你在代码里描述字段客户端据此渲染出表单让用户填写。URL 模式URL mode你需要用户去别处完成某些操作如 OAuth 授权确认页、支付页面。用户在那边做的一切都不经过协议传输完全在带外out of band进行。而提问的方式也有两种其中首选是解析器resolver把问题挂在一个参数上由 SDK 代为提问——这种方式在任何连接上都可用无论客户端使用的是哪个协议版本。直接的提问方式await ctx.elicit(...)则是服务端向客户端发起的请求这种通道只对历史连接legacy connection规范版本 2025-11-25 或更早上的客户端存在。两种方式在本页都有介绍建议优先使用 resolver。用 Resolver 提问推荐方式一个决定整个工具走向的问题——你确定吗三个匹配账户选哪个——可以把提问逻辑从工具函数体内抽离出来放到一个resolver中由框架替你提问。凡是标注为Annotated[T, Resolve(fn)]的参数都会在工具函数体执行之前通过运行fn来填充。当 resolver 已经知道答案时直接返回值当它需要提问时返回Elicit(...)由框架代为提问from typing import Annotated from pydantic import BaseModel from mcp.server import MCPServer from mcp.server.mcpserver import ( AcceptedElicitation, CancelledElicitation, DeclinedElicitation, Elicit, ElicitationResult, Resolve, ) mcp MCPServer(Files) _FOLDERS: dict[str, list[str]] {/tmp/empty: [], /tmp/project: [main.py, README.md]} class Confirm(BaseModel): ok: bool async def confirm_delete(path: str) - Confirm | Elicit[Confirm]: Resolver: ask for confirmation only when the folder is not empty. file_count len(_FOLDERS.get(path, [])) if file_count 0: return Confirm(okTrue) # nothing to confirm, no round-trip to the client return Elicit(f{path} has {file_count} file(s). Delete anyway?, Confirm) mcp.tool() async def delete_folder( path: str, confirm: Annotated[ElicitationResult[Confirm], Resolve(confirm_delete)], ) - str: Delete a folder, asking for confirmation when it is not empty. match confirm: case AcceptedElicitation(dataConfirm(okTrue)): _FOLDERS.pop(path, None) return fdeleted {path} case AcceptedElicitation(): return kept the folder case DeclinedElicitation(): return declined: folder not deleted case CancelledElicitation(): return cancelled: folder not deleted这个例子中有三个值得注意的设计点confirm_delete按名字读取工具自身的path参数、列出文件夹内容并且只在必要的时候才征询——空文件夹直接解析为Confirm(okTrue)与客户端之间零往返。delete_folder将参数标注为ElicitationResult[Confirm]框架注入完整的征询结果工具用match逐一处理所有分支接受并确认删除、接受但保留okFalse、拒绝decline、取消cancel。confirm参数永远不会出现在工具的输入 schema 中——客户端只提供pathconfirm由 resolver 负责填充。当工具不需要分叉处理时也可以直接标注未包装的模型Annotated[Confirm, Resolve(confirm_delete)]接受时工具收到模型实例拒绝或取消时调用直接以错误终止。Resolver 在所有连接上都能工作Resolver 在每一种连接上都能工作。对于历史连接上的客户端SDK 直接把问题发送过去对于2026-07-28规范的连接SDK 改为从调用中返回问题客户端下一次重试时携带回答。你的 resolver 代码完全感知不到这些差异——底层发生的是多轮往返请求Multi-round-trip requests。从源码看这一分派逻辑位于 src/mcp/server/mcpserver/resolve.py_INPUT_REQUIRED_VERSION 2026-07-28是协议分水岭_uses_input_required()根据ctx.protocol_version判断走哪条传输通道在 2026-07-28 及之后多个待决问题会被批量收进InputRequiredResult随request_state跨轮次保留在旧版本上则通过服务端到客户端的请求通道逐个同步询问。文件头部注释明确说明Resolve机制形成依赖 DAGresolver 之间可以互相依赖也可以按名取工具参数、取用Context。提问只是 resolver 能力的一部分。更通用的机制——不提问也能计算的依赖、依赖的依赖、模型能提供什么不能提供什么——详见 依赖Dependencies 页面。从工具内部直接提问工具也可以在自己的函数体中停下来提问。但需要注意ctx.elicit()和ctx.elicit_url()是服务端发往客户端的请求——这个通道只对历史连接规范版本2025-11-25或更早上的客户端存在。在2026-07-28连接上没有服务端主动发起的请求这些调用会失败resolver 则在两种连接上都可用。完整背景见 协议版本。await ctx.elicit()接收一条消息和一个 Pydantic 模型from pydantic import BaseModel, Field from mcp.server import MCPServer from mcp.server.mcpserver import Context mcp MCPServer(Bistro) class AlternativeDate(BaseModel): accept_alternative: bool Field(descriptionTry another date?) date: str Field(default2025-12-26, descriptionAlternative date (YYYY-MM-DD)) mcp.tool() async def book_table(date: str, party_size: int, ctx: Context) - str: Book a table at the bistro. if date ! 2025-12-25: return fBooked a table for {party_size} on {date}. result await ctx.elicit( messagefNo tables for {party_size} on {date}. Would you like to try another date?, schemaAlternativeDate, ) if result.action accept and result.data.accept_alternative: return await book_table(result.data.date, party_size, ctx) return No booking made.这个示例里蕴含了几条重要规则Context参数是ctx.elicit的入口任何工具都可以声明一个Context参数参数名随意只要类型注解是Context。该对象有专门页面Context 对象。AlternativeDate是你期望的回答的schema。工具必须是async def。它必须如此因为工具要在中途停下来等待真人作答。对于其他日期工具立即返回。它只在必要时才提问。用户接受的日期会重新经过book_table本身。回答与普通输入没有区别如果换的日期也被订满了会再次触发提问而不是被盲目确认。客户端会收到什么客户端会收到你的消息以及一个由模型生成的 JSON Schema{ properties: { accept_alternative: { description: Try another date?, title: Accept Alternative, type: boolean }, date: { default: 2025-12-26, description: Alternative date (YYYY-MM-DD), title: Date, type: string } }, required: [accept_alternative], title: AlternativeDate, type: object }这个 schema 就是表单本身。Field(description...)是字段的标签提供默认值default会预填输入框并让该字段变为可选。这套「Pydantic 到 JSON Schema」的机制与工具Tools 页面描述的工具参数生成机制完全相同。表单 schema 的字段限制需要特别留意征询用的 schema 没有工具的输入 schema 那么强表达力。只支持扁平的原生字段str、int、float、bool或字符串Literal会转成enum。如果在模型里再嵌套一个模型ctx.elicit会在向客户端发送任何内容之前抛出异常。此时工具调用会以Error executing tool name失败服务端日志中会记录原因TypeError: Elicitation schema field address rendered as {$ref: #/$defs/Address}, which is not a valid PrimitiveSchemaDefinition从实现看这一校验位于 src/mcp/server/elicitation.pyrender_elicitation_schema()用自定义的_ElicitationJsonSchema生成器渲染 schema会把T | None展平为T、丢弃None默认值随后_validate_rendered_properties()用规范中的PrimitiveSchemaDefinition逐一校验每个字段不合法就直接抛TypeError。你是在打断一个人正在进行的工作如果回答需要嵌套结构那它本应是工具的一个参数。三种回答result.action告诉你用户做了什么恰好有三种可能accept用户提交了表单。result.data是AlternativeDate的实例已经过校验。decline用户拒绝了。cancel用户未做选择就关掉了提问。result.data只在accept时存在这正是示例代码先判断result.action的原因。类型检查器也会强制这个顺序在result.action accept之后result.data才是AlternativeDate在此之前根本没有.data。**拒绝不是错误。**工具自己决定「拒绝」意味着什么在这里是不预订然后正常地回复模型。另外有个安全细节值得记住回答会在你的代码看到它之前按照你的模型进行校验。一个把bool字段发成maybe的客户端不会破坏你的预订ctx.elicit会抛出ValueError调用失败你的if永远不会执行。这一点在src/mcp/server/elicitation.py的elicit_with_validation()中由schema.model_validate(result.content)保证。把用户送去一个 URLURL 模式有些东西绝不能经过模型或客户端凭证、卡号、OAuth 授权。对这类场景你不该索取数据而该请用户去某个地方from mcp.server import MCPServer from mcp.server.mcpserver import Context mcp MCPServer(Bistro) mcp.tool() async def pay_deposit(booking_id: str, ctx: Context) - str: Take the deposit that confirms a booking. result await ctx.elicit_url( messageA 20 EUR deposit confirms your booking., urlfhttps://pay.example.com/deposit/{booking_id}, elicitation_idfdeposit-{booking_id}, ) if result.action accept: return Complete the payment in your browser. return No deposit taken. The booking expires in one hour. mcp.tool() async def confirm_deposit(booking_id: str, ctx: Context) - str: Record a payment reported by the payment provider. await ctx.session.send_elicit_complete(fdeposit-{booking_id}) return fDeposit received for booking {booking_id}.这个例子的要点ctx.elicit_url()接收消息、要访问的URL以及一个由你指定的elicitation_id任意字符串只要能在你的服务器内部唯一标识这次征询即可。返回的结果只包含一个 action没有别的。accept表示用户同意打开该 URL而不是说他已经完成了 URL 那一侧的事情。支付发生在带外在用户浏览器和你的支付服务商之间完成任何内容都不会经由 MCP 回传。再看第二个工具当服务器得知带外流程已经结束通过 webhook、轮询等方式这里用一个工具来模拟ctx.session.send_elicit_complete(...)会发送notifications/elicitation/complete携带同一个elicitation_id。客户端正是靠这条通知才知道可以停止显示等待支付……。没有它客户端只能靠猜。该方法的实现位于 src/mcp/server/session.py 的send_elicit_complete。客户端一侧用 elicitation_callback 应答服务端负责提问客户端通过向Client(...)传入一个回调函数elicitation_callback来应答from mcp import Client from mcp.client import ClientRequestContext from mcp.types import ElicitRequestParams, ElicitRequestURLParams, ElicitResult async def handle_elicitation(context: ClientRequestContext, params: ElicitRequestParams) - ElicitResult: if isinstance(params, ElicitRequestURLParams): print(fOpen this link to continue: {params.url}) return ElicitResult(actionaccept) print(params.message) return ElicitResult(actionaccept, content{accept_alternative: True, date: 2025-12-27}) async def main() - None: async with Client( http://127.0.0.1:8000/mcp, modelegacy, elicitation_callbackhandle_elicitation, ) as client: result await client.call_tool(book_table, {date: 2025-12-25, party_size: 2}) print(result.content)几点说明一个回调同时处理两种模式。params是ElicitRequestFormParams和ElicitRequestURLParams的联合类型用isinstance做分支即可。对 URL 模式把params.url展示给用户返回用户选择的 action。永远不要携带content。对表单模式真正的应用应该渲染params.requested_schema并把用户输入作为content返回。本示例总是用写死的答案回答是这正是测试中你想要的回调行为。传入回调函数本身同时也是能力声明capability declaration服务端正是据此知道该客户端可以被提问。客户端还能为服务端应答的其他内容见客户端回调Client callbacks。回调参数elicitation_callback定义在 src/mcp/client/client.py会话层处理逻辑在 src/mcp/client/session.py含一个_default_elicitation_callback兜底实现。需要说明Elicitation 是服务端到客户端的请求这类请求只存在于经典握手classic-handshake会话上所以客户端传了modelegacy。在2026-07-28连接上工具改为从调用中返回问题那个流程见多轮往返请求Multi-round-trip requests同一个回调会被该机制驱动。动手试一下先用表单模式book_table那个启动server.py跑在 Streamable HTTP 上启动命令见运行你的服务器Running your server 的简介然后运行客户端的main()向book_table询问圣诞节当天。回调会打印它收到的提问No tables for 2 on 2025-12-25. Would you like to try another date?它回答{accept_alternative: True, date: 2025-12-27}而工具——一直在await ctx.elicit(...)里等待——完成预订Booked a table for 2 on 2025-12-27.接着换成 URL 模式的server.py让同一个main()指向pay_deposit同一个回调走另一个分支打印支付链接工具带着Complete the payment in your browser.返回。一次往返发生在调用中途双向都是如此。值得验证的反例把Client上的elicitation_callback移除再次为圣诞节调用book_table。整个调用会以协议错误失败Elicitation not supported没有注册任何回调的客户端从未声明elicitation能力所以根本没有可问的对象。你的工具收到的不是decline而是一个异常。请据此设计代码每次征询都要想清楚如果我问不了怎么办的合理解答。小结标注为Annotated[T, Resolve(fn)]的参数由 resolver 填充resolver 需要提问时返回Elicit(...)。这在所有连接上都可用。schema 是一个扁平的 Pydantic 模型只允许原生字段返回时校验。result.action取值accept、decline或cancelresult.data仅在 accept 时存在。await ctx.elicit(message, schemaModel)从工具体内部提问await ctx.elicit_url(message, url, elicitation_id)用于一切不该经过模型的内容ctx.session.send_elicit_complete(elicitation_id)表示带外部分已完成。两者都是服务端到客户端的请求需要客户端位于历史连接上。客户端用一个elicitation_callback应答按 params 类型分支注册它即是声明能力。在 2026-07-28 连接上服务端改为返回问题而不是推送同一个回调由多轮往返请求Multi-round-trip requests 驱动。在这些返回值之下的一切机制重试循环、保护requestState、手动驱动整个流程都在多轮往返请求Multi-round-trip requests 页面中详解。仓库中tests/docs_src/test_elicitation.py与examples/snippets/下的elicitation.py等文件也提供了可继续研读的配套示例。【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址: https://gitcode.com/gh_mirrors/pythonsd/python-sdk创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考