企业官网建设流程全解析

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 Protocol（MCP）的请求几乎全部是单向的：由客户端发往服务器。但服务器有时也会反过来向客户端提出请求——向用户提问（elicitation）、用用户侧的模型做采样（sampling）、列出用户允许访问的工作区目录（roots）。本指南以官方文档 docs/client/callbacks.md 为主体，完整讲解如何通过向Client(...)传入回调来应答这类反向请求，并深入剖析“注册回调即声明能力（capability）”这一核心机制。读完本文，你将掌握 elicitation 回调的完整签名与返回约定、mode="legacy"在反向通道（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?", schema=CardHolder) if answer.action == "accept": return f"Card 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(action="accept", content={"name": "Ada Lovelace"}) 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("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(action="accept", content={...})表示接受并给出内容；也可以返回action="decline"或action="cancel"。除这三种外，唯一的“其他选择”是返回ErrorData(...)——它会拒绝请求并让整个调用失败（见下文“故意犯错”一节）。测试 tests/docs_src/test_client_callbacks.py 验证了返回ErrorData(code=INVALID_REQUEST, ...)时call_tool会抛出MCPError。

context：类型为ClientRequestContext，携带当前使用的session、服务器的request_id以及服务器附加的meta。其定义位于 src/mcp/client/session.py：

@dataclass(kw_only=True) 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(type='text', text='Card issued to Ada Lovelace.')]

整个流程是：客户端发出 1 次tools/call→ 服务器折返 1 次elicitation/create→ 客户端的回调函数应答。全部发生在这 1 次工具调用内部。测试 tests/docs_src/test_client_callbacks.py 用进程内（in-process）服务器端到端验证了这一结论。

为什么必须指定mode="legacy"

教程中Client(...)的mode="legacy"并不是摆设。从 src/mcp/client/client.py 的源码看，mode的默认值是"auto"：它会探测server/discover并在旧服务器上回退到 initialize 握手；对于进程内Server/MCPServer则直接分发、不走 JSON-RPC 帧。默认协商出的新协议路径没有“服务器 → 客户端请求”的反向通道（back-channel），因此在回调被调用之前ctx.elicit就会失败。

决定这一点的是协商出的协议版本，而不是传输层（transport）。只要客户端需要应答这类反向请求，就必须显式传入mode="legacy"。测试 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，而这个对象不需要你手写：注册回调这个动作本身就是声明。

这张表在源码中有直接对应实现——src/mcp/client/session.py 的_build_capabilities逐条检查回调是否仍是默认实现（_default_sampling_callback等），从而决定sampling、elicitation、roots是否进入能力声明：

• 注册了elicitation_callback→ 声明ElicitationCapability(form=FormElicitationCapability(), url=UrlElicitationCapability())，即{"form": {}, "url": {}}；
• 注册了sampling_callback→ 声明SamplingCapability()，即"sampling": {}；
• 注册了list_roots_callback→ 声明RootsCapability(list_changed=True)，即"roots": {"listChanged": true}。

唯一的细化选项是采样能力的子能力：如果采样器能够处理tools/tool_choice参数，需要在sampling_callback之外再传入sampling_capabilities=SamplingCapability(tools=SamplingToolsCapability())。服务器只有在看到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?", schema=CardHolder) if answer.action == "accept": return f"Card 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(elicitation=ElicitationCapability()), "sampling": ClientCapabilities(sampling=SamplingCapability()), "roots": ClientCapabilities(roots=RootsCapability()), } 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(code=INVALID_REQUEST, message="Elicitation not supported")。

注意这是协议错误（-32600，invalid 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( role="assistant", content=TextContent(type="text", text="The answer is 42."), model="my-llm", ) async def handle_list_roots(context: ClientRequestContext) -> ListRootsResult: return ListRootsResult(roots=[Root(uri=FileUrl("file:///home/ada/notebooks"), name="notebooks")])

要点：

• 采样回调接收完整的CreateMessageRequestParams（messages、model_preferences、max_tokens），返回CreateMessageResult。真正调用模型的是客户端自己，方式不限，SDK 只负责搬运请求；
• roots 回调不接收任何参数，返回ListRootsResult；
• 两者拒绝请求的方式同样是返回ErrorData(...)；
• 传入Client(...)的方式与elicitation_callback完全一致。

SDK 中的协议定义见 src/mcp/client/session.py（SamplingFnT、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_level="info"（或其他级别）会在每个请求上附带该 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也会送达。

有两个例外不会到达 handler：

• notifications/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接收通知，不声明任何能力；
• 需要应答反向请求时务必使用mode="legacy"，因为默认协商的新协议路径没有反向通道。

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