ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

MCP Python SDK 客户端回调(Client Callbacks)权威指南:响应服务端发起的请求与能力协商

MCP Python SDK 客户端回调(Client Callbacks)权威指南:响应服务端发起的请求与能力协商 MCP Python SDK 客户端回调Client Callbacks权威指南响应服务端发起的请求与能力协商【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址: https://gitcode.com/gh_mirrors/pythonsd/python-sdk导读Model Context ProtocolMCP的请求几乎全部是单向的由客户端发往服务器。但服务器有时也会反过来向客户端提出请求——向用户提问elicitation、用用户侧的模型做采样sampling、列出用户允许访问的工作区目录roots。本指南以官方文档 docs/client/callbacks.md 为主体完整讲解如何通过向Client(...)传入回调来应答这类反向请求并深入剖析“注册回调即声明能力capability”这一核心机制。读完本文你将掌握 elicitation 回调的完整签名与返回约定、modelegacy在反向通道back-channel上的必要性、旧协议下 sampling/roots 回调的用法以及logging_callback、message_handler两类通知回调的边界并能在实际项目中正确配置可验证的客户端。MCP 中反向请求的定位MCP 的常规请求方向是 client → server例如tools/call、prompts/get、resources/read。但服务器在某些场景下需要向客户端反向求助向用户提问elicitation/create使用用户侧模型进行采样sampling/createMessage查询用户允许操作的工作区目录roots/list。这类请求无法靠客户端主动发起必须由客户端在连接时注册的**回调callback**来应答。SDK 提供的Client(...)构造器为此预留了elicitation_callback、sampling_callback、list_roots_callback、logging_callback、message_handler等参数全部定义于 src/mcp/client/client.py。官方文档将其归类为“客户端回调”专题并配有 4 个可运行的教程示例docs_src/client_callbacks/和对应的自动化测试tests/docs_src/test_client_callbacks.py。第一步让服务器“开口提问”回调的另一端是服务器通过ctx.elicit(...)主动发送elicitation/create请求。以图书馆办卡场景为例服务器的issue_card工具单独无法完成任务——它必须等到有人提供持卡人姓名才能继续from pydantic import BaseModel from mcp.server import MCPServer from mcp.server.mcpserver import Context mcp MCPServer(Library) class CardHolder(BaseModel): name: str mcp.tool() async def issue_card(ctx: Context) - str: Issue a new library card. answer await ctx.elicit(What name should go on the card?, schemaCardHolder) if answer.action accept: return fCard issued to {answer.data.name}. return No card issued.关键点ctx.elicit(...)会把elicitation/create请求发送给客户端并挂起等待在有人表单填写者或客户端的代码提供name之前工具不会返回这是“服务器侧”的行为完整的服务器侧讲解见 docs/handlers/elicitation.md。本文聚焦通信的另一侧——客户端如何应答。第二步用elicitation_callback应答客户端的应答侧代码非常简单from mcp import Client from mcp.client import ClientRequestContext from mcp.types import ElicitRequestParams, ElicitResult async def handle_elicitation( context: ClientRequestContext, params: ElicitRequestParams, ) - ElicitResult: return ElicitResult(actionaccept, content{name: Ada Lovelace}) 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(issue_card) print(result.content)这段代码揭示了回调的完整契约签名async (context, params) - ElicitResult。SDK 在 src/mcp/client/session.py 中以ElicitationFnT协议精确定义了该签名。参数含义params.message是服务器提出的问题文本params.requested_schema是服务器期望答案遵循的 JSON Schema。真实客户端例如渲染表单的 GUI会据此绘制表单而教程里用固定值自动填写。返回值ElicitResult(actionaccept, content{...})表示接受并给出内容也可以返回actiondecline或actioncancel。除这三种外唯一的“其他选择”是返回ErrorData(...)——它会拒绝请求并让整个调用失败见下文“故意犯错”一节。测试 tests/docs_src/test_client_callbacks.py 验证了返回ErrorData(codeINVALID_REQUEST, ...)时call_tool会抛出MCPError。context类型为ClientRequestContext携带当前使用的session、服务器的request_id以及服务器附加的meta。其定义位于 src/mcp/client/session.pydataclass(kw_onlyTrue) class ClientRequestContext: session: ClientSession request_id: RequestId meta: RequestParamsMeta | None None两种模式params是两种 elicitation 模式的联合类型。本示例中params.mode form而url模式的请求不含 schema取而代之的是params.url。官方建议在同一个回调函数里用params.mode分叉处理两种模式完整模式示例见 docs/handlers/elicitation.md。试运行观察一次完整往返调用issue_card后回调收到的是“已解析”的问题对象params.mode # form params.message # What name should go on the card? params.requested_schema # {properties: {name: {title: Name, type: string}}, # required: [name], title: CardHolder, type: object}回调给出答案后工具内部的ctx.elicit(...)恢复执行工具随之完成result.content # [TextContent(typetext, textCard issued to Ada Lovelace.)]整个流程是客户端发出 1 次tools/call→ 服务器折返 1 次elicitation/create→ 客户端的回调函数应答。全部发生在这 1 次工具调用内部。测试 tests/docs_src/test_client_callbacks.py 用进程内in-process服务器端到端验证了这一结论。为什么必须指定modelegacy教程中Client(...)的modelegacy并不是摆设。从 src/mcp/client/client.py 的源码看mode的默认值是auto它会探测server/discover并在旧服务器上回退到 initialize 握手对于进程内Server/MCPServer则直接分发、不走 JSON-RPC 帧。默认协商出的新协议路径没有“服务器 → 客户端请求”的反向通道back-channel因此在回调被调用之前ctx.elicit就会失败。决定这一点的是协商出的协议版本而不是传输层transport。只要客户端需要应答这类反向请求就必须显式传入modelegacy。测试 tests/docs_src/test_client_callbacks.py 精确复现了默认模式下报MCPError: ... no back-channel的行为。协议版本协商的更多细节见 docs/protocol-versions.md。2026-07-28 协议下回调并未消失需要澄清2026-07-28 版本的会话中回调并没有被弃用只是调用方式发生了变化。当工具返回包含ElicitRequest的InputRequiredResult时Client会把其中的条目路由到同一个elicitation_callback并自动重试该调用。这一机制在 src/mcp/client/client.py 的_drive_input_required中实现通过dispatch_input_request将内嵌请求分发给与旧协议相同的回调表从而保证两条路径行为一致。重试轮数由input_required_max_rounds控制默认值见 src/mcp/client/_input_required.py 的DEFAULT_INPUT_REQUIRED_MAX_ROUNDS。完整说明见 docs/handlers/multi-round-trip.md。回调即能力注册即声明你可能从未显式告诉过服务器“客户端能应答 elicitation 请求”——声明这件事的是 SDK。客户端在连接时会像服务器一样声明自己的capabilities而这个对象不需要你手写注册回调这个动作本身就是声明。传入的参数客户端声明的能力elicitation_callbackelicitation: {form: {}, url: {}}sampling_callbacksampling: {}list_roots_callbackroots: {listChanged: true}都不传{}这张表在源码中有直接对应实现——src/mcp/client/session.py 的_build_capabilities逐条检查回调是否仍是默认实现_default_sampling_callback等从而决定sampling、elicitation、roots是否进入能力声明注册了elicitation_callback→ 声明ElicitationCapability(formFormElicitationCapability(), urlUrlElicitationCapability())即{form: {}, url: {}}注册了sampling_callback→ 声明SamplingCapability()即sampling: {}注册了list_roots_callback→ 声明RootsCapability(list_changedTrue)即roots: {listChanged: true}。唯一的细化选项是采样能力的子能力如果采样器能够处理tools/tool_choice参数需要在sampling_callback之外再传入sampling_capabilitiesSamplingCapability(toolsSamplingToolsCapability())。服务器只有在看到sampling.tools被声明后才会发送这些参数。对应参数定义见 src/mcp/client/client.py其在能力构建中的应用见 src/mcp/client/session.py。logging_callback与message_handler不在上表中它们处理的是通知notification而通知不需要声明能力。服务器如何“先问后要”服务器端通过ctx.session.check_client_capability(...)读取客户端声明。教程为服务器新增了一个工具来展示这一点from pydantic import BaseModel from mcp.server import MCPServer from mcp.server.mcpserver import Context from mcp.types import ClientCapabilities, ElicitationCapability, RootsCapability, SamplingCapability mcp MCPServer(Library) class CardHolder(BaseModel): name: str mcp.tool() async def issue_card(ctx: Context) - str: Issue a new library card. answer await ctx.elicit(What name should go on the card?, schemaCardHolder) if answer.action accept: return fCard issued to {answer.data.name}. return No card issued. mcp.tool() def client_features(ctx: Context) - list[str]: Which optional features the connected client declared. declared { elicitation: ClientCapabilities(elicitationElicitationCapability()), sampling: ClientCapabilities(samplingSamplingCapability()), roots: ClientCapabilities(rootsRootsCapability()), } return [name for name, capability in declared.items() if ctx.session.check_client_capability(capability)]check_client_capability的实现位于 src/mcp/server/session.py其核心是比对连接时收到的能力声明。实验三种连接方式# 只传 elicitation_callback result.structured_content # {result: [elicitation]} # 三个回调都传 result.structured_content # {result: [elicitation, sampling, roots]} # 一个都不传 result.structured_content # {result: []}对应测试见 tests/docs_src/test_client_callbacks.py。故意犯错不注册回调会发生什么现在做一次“错误示范”不传elicitation_callback仍去调用issue_card。服务器的elicitation/create请求依然会抵达客户端但由于客户端没有声明可处理它SDK 会代答一个错误并让整个调用失败——call_tool抛出的不是is_error结果而是异常MCPError: Elicitation not supported这个错误来自 SDK 内置的默认回调 src/mcp/client/session.py任何未注册的 elicitation 请求都会得到ErrorData(codeINVALID_REQUEST, messageElicitation not supported)。注意这是协议错误-32600invalid request而不是工具错误——模型读取后没有任何可重试的内容。这正是client_features这类工具的价值所在有礼貌的服务器在请求之前会先检查。对应测试见 tests/docs_src/test_client_callbacks.py。两个已弃用但仍在服役的回调sampling_callback应答sampling/createMessage服务器请求使用客户端侧模型补全内容list_roots_callback应答roots/list服务器询问可操作的工作目录。两者目前都可用也都遵循上文的规则但对应的是在 2026-07-28 规范中被删除的 RPC——新服务器不再在请求中途回调客户端而是把输入需求作为工具结果的一部分返回见 docs/handlers/multi-round-trip.md。回调本身并不会被弃用当InputRequiredResult内嵌CreateMessageRequest或ListRootsRequest时Client的自动重试循环即上文_drive_input_required会把它们分发给这里注册的同一个sampling_callback/list_roots_callback。完整弃用清单见 docs/deprecated.md。与尚未迁移的旧服务器通信时仍需要这些回调。签名示例如下from pydantic import FileUrl from mcp.client import ClientRequestContext from mcp.types import CreateMessageRequestParams, CreateMessageResult, ListRootsResult, Root, TextContent async def handle_sampling( context: ClientRequestContext, params: CreateMessageRequestParams, ) - CreateMessageResult: return CreateMessageResult( roleassistant, contentTextContent(typetext, textThe answer is 42.), modelmy-llm, ) async def handle_list_roots(context: ClientRequestContext) - ListRootsResult: return ListRootsResult(roots[Root(uriFileUrl(file:///home/ada/notebooks), namenotebooks)])要点采样回调接收完整的CreateMessageRequestParamsmessages、model_preferences、max_tokens返回CreateMessageResult。真正调用模型的是客户端自己方式不限SDK 只负责搬运请求roots 回调不接收任何参数返回ListRootsResult两者拒绝请求的方式同样是返回ErrorData(...)传入Client(...)的方式与elicitation_callback完全一致。SDK 中的协议定义见 src/mcp/client/session.pySamplingFnT、ListRootsFnT默认拒绝行为见同文件 src/mcp/client/session.py。测试 tests/docs_src/test_client_callbacks.py 验证了两个回调的返回类型与文档描述一致。通知类回调logging_callback与message_handler最后还有两个回调它们处理通知因此不声明任何能力。logging_callback日志消息logging_callback接收服务器发来的notifications/message参数类型为LoggingMessageNotificationParams包含level、logger、data。协议层面的日志功能本身已在 2026-07-28 规范中弃用替代方案见 docs/handlers/logging.md因此该回调主要为仍在发送日志通知的旧服务器保留。这里有一个重要的版本差异2026 世代连接仅注册回调收不到任何日志。因为 2026 年的服务器只对显式 opt-in 的请求发送日志消息。向Client(...)传入log_levelinfo或其他级别会在每个请求上附带该 opt-in从而收到不低于该级别的日志。对应参数见 src/mcp/client/client.py其注释明确说明log_level会将 opt-in 盖印到每个请求的_meta的io.modelcontextprotocol/logLevel键上2026 年之前的服务器忽略 opt-in沿用传统的logging/setLevel行为。message_handler所有通知的总入口message_handler是“来者不拒”的窗口会话对外呈现的所有服务器通知除各自有专属回调的那些之外都会送达这里在基于流stream的传输上传输层抛出的所有Exception也会送达。有两个例外不会到达 handlernotifications/cancelled不会对外呈现由 SDK 直接处理活动中的listen()流对应的订阅确认应答会被该流自身消费Client.listen详见 src/mcp/client/client.py。参数应使用IncomingMessage标注其定义为ServerNotification | Exception从mcp.client导出类型别名见 src/mcp/client/session.py协议定义见同文件 src/mcp/client/session.py。值得记住的一个模式是if isinstance(message, Exception): raise message这样连接中断时会明确失败而不是悄无声息地消失——便于在日志中暴露网络/传输故障。总结服务器可以向客户端发起请求应答方式是把回调传给Client(...)现行机制是 elicitation 回调async (context, params) - ElicitResult一个函数同时处理 form 与 url 两种模式注册回调就是声明能力。不注册时SDK 会代替客户端拒绝服务器的请求整个调用以MCPError失败服务器在发起请求前应通过ctx.session.check_client_capability(...)检查客户端能力sampling_callback与list_roots_callback工作方式相同但服务于已弃用功能新服务器改用 multi-round-trip 请求logging_callback与message_handler接收通知不声明任何能力需要应答反向请求时务必使用modelegacy因为默认协商的新协议路径没有反向通道。Client(...)的第一个参数决定传输层所有传输类型的完整介绍见 docs/client/transports.md。上述所有行为均有教程代码docs_src/client_callbacks/与自动化测试tests/docs_src/test_client_callbacks.py背书可放心参照落地。【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址: https://gitcode.com/gh_mirrors/pythonsd/python-sdk创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考
RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进