MCP Python SDK 中的 Elicitation(征询)机制:工具中途提问的两种模式与客户端回调实现
2026/9/20 2:09:56 网站建设 项目流程

MCP Python SDK 中的 Elicitation(征询)机制:工具中途提问的两种模式与客户端回调实现

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

Elicitation(征询)是 MCP(Model 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(ok=True) # 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(data=Confirm(ok=True)): _FOLDERS.pop(path, None) return f"deleted {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(ok=True),与客户端之间零往返。
  • delete_folder将参数标注为ElicitationResult[Confirm]:框架注入完整的征询结果,工具用match逐一处理所有分支:接受并确认删除、接受但保留(ok=False)、拒绝(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机制形成依赖 DAG,resolver 之间可以互相依赖,也可以按名取工具参数、取用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(description="Try another date?") date: str = Field(default="2025-12-26", description="Alternative 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 f"Booked a table for {party_size} on {date}." result = await ctx.elicit( message=f"No tables for {party_size} on {date}. Would you like to try another date?", schema=AlternativeDate, ) 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 那么强表达力。只支持扁平的原生字段:strintfloatbool,或字符串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.py:render_elicitation_schema()用自定义的_ElicitationJsonSchema生成器渲染 schema(会把T | None展平为T、丢弃None默认值),随后_validate_rendered_properties()用规范中的PrimitiveSchemaDefinition逐一校验每个字段,不合法就直接抛TypeError。你是在打断一个人正在进行的工作,如果回答需要嵌套结构,那它本应是工具的一个参数。

三种回答

result.action告诉你用户做了什么,恰好有三种可能:

  • "accept":用户提交了表单。result.dataAlternativeDate的实例,已经过校验
  • "decline":用户拒绝了。
  • "cancel":用户未做选择就关掉了提问。

result.data只在"accept"时存在,这正是示例代码先判断result.action的原因。类型检查器也会强制这个顺序:在result.action == "accept"之后,result.data才是AlternativeDate;在此之前根本没有.data

**拒绝不是错误。**工具自己决定「拒绝」意味着什么(在这里是不预订),然后正常地回复模型。

另外有个安全细节值得记住:回答会在你的代码看到它之前按照你的模型进行校验。一个把bool字段发成"maybe"的客户端不会破坏你的预订:ctx.elicit会抛出ValueError,调用失败,你的if永远不会执行。这一点在src/mcp/server/elicitation.pyelicit_with_validation()中由schema.model_validate(result.content)保证。

把用户送去一个 URL(URL 模式)

有些东西绝不能经过模型或客户端:凭证、卡号、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( message="A 20 EUR deposit confirms your booking.", url=f"https://pay.example.com/deposit/{booking_id}", elicitation_id=f"deposit-{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(f"deposit-{booking_id}") return f"Deposit 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(f"Open this link to continue: {params.url}") return ElicitResult(action="accept") print(params.message) return ElicitResult(action="accept", content={"accept_alternative": True, "date": "2025-12-27"}) async def main() -> None: async with Client( "http://127.0.0.1:8000/mcp", mode="legacy", elicitation_callback=handle_elicitation, ) as client: result = await client.call_tool("book_table", {"date": "2025-12-25", "party_size": 2}) print(result.content)

几点说明:

  • 一个回调同时处理两种模式paramsElicitRequestFormParamsElicitRequestURLParams的联合类型,用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)会话上,所以客户端传了mode="legacy"。在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""cancel"result.data仅在 accept 时存在。
  • await ctx.elicit(message, schema=Model)从工具体内部提问;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.pyexamples/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),仅供参考

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询