MCP Python SDK 多轮往返请求：InputRequiredResult、requestState 安全封条与客户端重试循环

MCP Python SDK 多轮往返请求InputRequiredResult、requestState 安全封条与客户端重试循环【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址: https://gitcode.com/gh_mirrors/pythonsd/python-sdk本篇围绕 MCPModel Context Protocol官方 Python SDK 的 2026-07-28 协议版本新增的“多轮往返”multi-round-trip机制展开当一次tools/call或prompts/get、resources/read无法在单轮内完成、必须向用户索取选择、确认或凭据时服务器不再反向调用客户端而是返回一个InputRequiredResult把“缺口”交回客户端。读完后你将掌握三种实现形态——低层级Server手动返回、MCPServer的mcp.prompt()/mcp.resource()模板、以及高层依赖注入——并能正确配置Client的自动重试循环与RequestStateSecurity状态保护把该机制安全地落到单进程到多实例的部署上。背景从“回调”到“返回”在多轮往返机制出现之前协议版本 2026-07-28 之前如果工具执行中途需要用户提供的东西服务器会在处理原始请求的中途反向发起一条请求——一次 elicitation征询、一次 sampling采样调用——这构成了服务器到客户端的“回传通道”back-channel。2026-07-28 规范废除了这条回传通道取而代之的是一句话服务器改为“返回”而不是“回调”。具体来说服务器对tools/call的回答不再是CallToolResult而是一个InputRequiredResult。客户端随后满足其中嵌入的每一条请求携带答案再次调用同一个工具直到拿到普通的CallToolResult。整个协议中每一跳leg都是一条普通的客户端到服务器请求永远不会有数据反向流动。类型定义位于 2026-07-28 版本的线协议类型模块 src/mcp-types/mcp_types/_v2026_07_28/init.pyclass InputRequiredResult(WireModel): An InputRequiredResult sent by the server to indicate that additional input is needed before the request can be completed. At least one of inputRequests or requestState MUST be present. meta: Annotated[ResultMetaObject | None, Field(alias_meta)] None input_requests: Annotated[InputRequests | None, Field(aliasinputRequests)] None request_state: Annotated[str | None, Field(aliasrequestState)] None result_type: Annotated[str, Field(aliasresultType)]其中InputRequest是三种请求的联合类型同文件第 3510 行InputRequest CreateMessageRequest | ListRootsRequest | ElicitRequest两个核心字段InputRequiredResult中真正“干活”的是两个字段input_requests服务器还缺什么。它是一个字典键是服务器自己起的名字值是一组ElicitRequest、CreateMessageRequest或ListRootsRequest。request_state一个不透明opaque令牌。客户端在重试时原样verbatim带回它只有你的服务器会读它客户端只负责搬运。客户端把每一条请求满足掉之后携带答案放在input_responses和令牌放在request_state再次调用同一个工具。此时服务器已拿到它缺少的东西返回一个普通的CallToolResult。这就是全部协议。服务器端两种形态高层形态依赖注入mcp.tool()在mcp.tool()中你很少需要手工构造这个结果声明一个“向用户提问”Elicit、“采样客户端 LLM”Sample或“列出 roots”ListRoots的依赖SDK 会替你返回InputRequiredResult。该形态的完整说明见 依赖Dependencies 页面。两种形态不能混用一次调用只有一条input_responses/request_state通道所以使用了Resolve(...)参数的工具其函数体不能再返回InputRequiredResult。声明式的InputRequiredResult返回值会在注册期被拒绝InvalidSignature未声明的返回值则会让调用在运行期失败。手动形态低层级Server手动形态是低层级Server其on_call_tool处理器被允许返回两种结果类型中的任意一种。示例代码完整收录于 docs_src/mrtr/tutorial001.pyfrom mcp.server import Server, ServerRequestContext from mcp.types import ( CallToolRequestParams, CallToolResult, ElicitRequest, ElicitRequestFormParams, ElicitResult, InputRequiredResult, ListToolsResult, PaginatedRequestParams, TextContent, Tool, ) ASK_REGION ElicitRequest( paramsElicitRequestFormParams( messageWhich region should the database live in?, requested_schema{ type: object, properties: {region: {type: string}}, required: [region], }, ) ) async def list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams | None) - ListToolsResult: return ListToolsResult( tools[ Tool( nameprovision, descriptionProvision a database. Asks which region to put it in., input_schema{ type: object, properties: {name: {type: string}}, required: [name], }, ) ] ) async def call_tool(ctx: ServerRequestContext, params: CallToolRequestParams) - CallToolResult | InputRequiredResult: answer (params.input_responses or {}).get(region) if not isinstance(answer, ElicitResult) or answer.content is None: return InputRequiredResult(input_requests{region: ASK_REGION}, request_stateprovision-v1) name (params.arguments or {})[name] text fProvisioned {name!r} in {answer.content[region]}. return CallToolResult(content[TextContent(typetext, texttext)]) server Server(Provisioner, on_list_toolslist_tools, on_call_toolcall_tool)三个要点on_call_tool的返回类型是CallToolResult | InputRequiredResult。返回后者就是服务器端的全部 API——没有别的按钮要按。首次调用时params.input_responses为None守卫条件触发处理器选择“提问”而不是“回答”于是返回携带input_requests的InputRequiredResult。重试时客户端送回的ElicitResult恰好落在服务器在input_requests里使用的同一个键这里是region之下。该文件其余部分显式的input_schema、手工构造的CallToolResult都属于普通低层级Server的范畴见 低层级 Server。本页只在其上增加第二个返回类型。不止工具prompts 与 resourcestools/call没有什么特殊之处在 2026-07-28 版本中服务器可以用同样的方式回答prompts/get和resources/read。在MCPServer上mcp.prompt()函数——或mcp.resource()的模板template函数——自己返回InputRequiredResult并从上下文读取重试时带来的答案。示例完整收录于 docs_src/mrtr/tutorial004.pyfrom mcp.server.mcpserver import Context, MCPServer from mcp.server.mcpserver.prompts.base import UserMessage from mcp.types import ElicitRequest, ElicitRequestFormParams, ElicitResult, InputRequiredResult mcp MCPServer(Briefing) ASK_AUDIENCE ElicitRequest( paramsElicitRequestFormParams( messageWho is the briefing for?, requested_schema{ type: object, properties: {audience: {type: string}}, required: [audience], }, ) ) mcp.prompt() async def briefing(ctx: Context) - list[UserMessage] | InputRequiredResult: Draft a briefing tuned to its audience. answer (ctx.input_responses or {}).get(audience) if not isinstance(answer, ElicitResult) or answer.content is None: return InputRequiredResult(input_requests{audience: ASK_AUDIENCE}) return [UserMessage(fWrite a briefing for {answer.content[audience]}.)]首轮返回InputRequiredResult重试时ctx.input_responses在相同的键下持有答案函数返回其普通结果——此处是 prompt 消息模板资源则是资源内容。你自己设置的request_state在跨越网络之前会被封条seal加密、回声时会被校验与服务器侧其他一切一致见下文“保护requestState”。mcp.tool()函数同样可以这样直接返回该结果适用于依赖形态不合身的场合。静态mcp.resource()函数不参与它们不接受Context因而永远无法读取重试。只有模板资源能提问。下文“版本边界”一节同样适用于这里在 2026 之前的会话上返回InputRequiredResult会得到警告框描述的同一种-32603错误。客户端侧Client替你跑循环Client会自动执行重试循环。注册服务器可能用到的回调elicitation_callback、sampling_callback、list_roots_callback然后直接调用工具即可。当InputRequiredResult到达时Client把input_requests的每条分派给对应回调带着答案和回声的request_state重试直到CallToolResult回来为止。示例完整收录于 docs_src/mrtr/tutorial003.pyfrom 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{region: eu-west-1}) async def main() - None: async with Client(http://127.0.0.1:8000/mcp, elicitation_callbackhandle_elicitation) as client: result await client.call_tool(provision, {name: orders}) print(result.content)这个elicitation_callback正是 2026 之前的服务器反向通道elicitation/create会命中的同一个回调。sampling_callback之于sampling/createMessage、list_roots_callback之于roots/list同理在 2026-07-28独立的服务器到客户端 RPC 消失了但完全相同的ElicitRequest/CreateMessageRequest/ListRootsRequest载荷现在装进input_requests里分派到同一组三个回调上。一套回调同时服务两代协议。call_tool只返回一个普通的CallToolResult中间的轮次对调用方完全不可见。get_prompt和read_resource驱动的是同一个循环。注意如果省略回调循环会直接在第一轮失败SDK 的替补回调对每次 elicitation 都以错误应答call_tool随即抛出消息为Elicitation not supported的MCPError。循环的边界与退避循环是有界的其驱动逻辑实现在 src/mcp/client/_input_required.pyClient(..., input_required_max_rounds10)是默认上限源码中DEFAULT_INPUT_REQUIRED_MAX_ROUNDS 10见 src/mcp/client/_input_required.py。服务器若在此之后仍持续返回InputRequiredResultcall_tool将抛出InputRequiredRoundsExceededError。如果某一轮只携带request_state而不带input_requestsClient会先短暂休眠再重试从 50 ms 起翻倍直到 250 ms 封顶源码常量_STATE_ONLY_BACKOFF_INITIAL_SECONDS 0.05与_STATE_ONLY_BACKOFF_CAP_SECONDS 0.25见 src/mcp/client/_input_required.py。这样一个只会说“还没完成”的服务器不会被忙轮询busy-poll。任何携带input_requests的轮次都会把退避重置回 50 ms见 驱动函数主体。核心驱动是一个纯函数run_input_required_driver使tools/call、prompts/get、resources/read三个方法可以复用同一套循环async def run_input_required_driver( first: InputRequiredResult, *, dispatch: Callable[[str, InputRequest], Awaitable[InputResponse | ErrorData]], retry: Callable[[InputResponses | None, str | None], Awaitable[ResultT | InputRequiredResult]], max_rounds: int DEFAULT_INPUT_REQUIRED_MAX_ROUNDS, ) - ResultT: ... while isinstance(current, InputRequiredResult): rounds 1 if rounds max_rounds: raise InputRequiredRoundsExceededError(max_rounds) if current.input_requests: state_only_delay _STATE_ONLY_BACKOFF_INITIAL_SECONDS responses await _dispatch_all(current.input_requests, dispatch) else: await anyio.sleep(state_only_delay) state_only_delay min(state_only_delay * 2, _STATE_ONLY_BACKOFF_CAP_SECONDS) responses None current await retry(responses, current.request_state) return current注意两点源码级细节request_state是字节级原样透传、从不被检视的客户端只当它是个搬运物_dispatch_all会并发执行所有input_requests第一条返回ErrorData的任务会经任务组取消兄弟任务被拒绝的输入不必等待慢速的同伴。自己驱动循环自动循环对单进程客户端足够了。以下场景应改为自己接管循环你的客户端是分布式的向用户呈现问题的进程与调用call_tool的进程不是同一个重试由另一个 worker 发出。request_state就是你要穿过这条边界、经由自有存储携带的可持久化令牌input_responses则是另一侧随它带回的内容。你想检视每一轮记录或审计每条input_requests条目、拒绝某些类型的请求、或在自己控制的退避间隔之间执行。你要的是墙钟边界而不是轮数边界用自己的anyio.fail_after(...)包裹循环而不是依赖input_required_max_rounds。做法是下沉到底层会话allow_input_requiredTrue会把联合类型直接交给你。示例完整收录于 docs_src/mrtr/tutorial002.pyfrom mcp import Client from mcp.types import CallToolResult, ElicitRequest, ElicitResult, InputRequest, InputRequiredResult, InputResponse def fulfil(request: InputRequest) - InputResponse: if not isinstance(request, ElicitRequest): raise NotImplementedError(fthis client cannot answer a {request.method!r} request) return ElicitResult(actionaccept, content{region: eu-west-1}) async def provision(client: Client, name: str) - CallToolResult: result await client.session.call_tool(provision, {name: name}, allow_input_requiredTrue) while isinstance(result, InputRequiredResult): responses {key: fulfil(request) for key, request in (result.input_requests or {}).items()} result await client.session.call_tool( provision, {name: name}, input_responsesresponses, request_stateresult.request_state, allow_input_requiredTrue, ) return resultclient.session.call_tool(..., allow_input_requiredTrue)把返回类型放宽为CallToolResult | InputRequiredResult由你的isinstance判断再把它收窄回来。request_state现在完全在你手里。在两步之间把它写下来对话就可以从一个全新进程恢复。对input_requests里的每条你都在input_responses的同一个键下放一个InputResponse。fulfil就是你的 UI 该出现的位置本例把答案写死了。每一跳都用相同的工具名、相同的arguments。重试是“把原始调用再执行一遍”而不是一个新方法。保护requestState以上所有讨论都把request_state当回声用而在线上它确实只是一段回声。但客户端会在两步之间持有它跨进程把它写下来正是上一节所认可的做法所以回来的是客户端提供的输入它可能被篡改、可能已过期、也可能整段取自另一个调用。规范强制要求只要该状态可能影响授权、资源访问或业务逻辑服务器就必须对其做完整性保护并在校验失败时拒绝该轮。默认行为MCPServer自动封条MCPServer默认就保护它。每个服务器都会封条seal出站的requestState、并校验每一个回声——解析器resolver状态与手工构建的状态一视同仁——使用的密钥在进程启动时生成。你什么都不用配置写的是明文、读的也是明文线上永远只流动一个不透明的加密令牌。默认密钥与进程同生共死。这是你从单进程部署跨出去之前唯一需要知道的事。多实例或需跨重启存活的部署传入共享密钥每把至少 32 字节from mcp.server.mcpserver import MCPServer, RequestStateSecurity # Multi-instance or restart-surviving: one or more shared secret keys ( 32 bytes each). mcp MCPServer(fleet, request_state_securityRequestStateSecurity(keys[key]))三种配置策略**默认零配置**适合单进程stdio或恰好一个 HTTP worker。落在不同 worker、负载均衡后的另一实例、或重启后同一服务器上的重试其状态是用一个该进程并不持有的密钥封的条——客户端收到下文那个冻结的拒绝必须从头开始整个流程。keys[...]在重试可能到达另一实例多 worker 的uvicorn、负载均衡的 HTTP或必须跨重启存活时是必需的每个实例都能校验任何兄弟实例铸造的状态。同一套机制只是用你的秘密换掉了生成出来的秘密。对于你自己的加密设施——例如 KMS 或既有令牌服务——改传RequestStateSecurity(codec...)而不是keys契约见下文“自带加密”。策略类的构造函数实现在 src/mcp/server/request_state.py几个约束值得注意def __init__( self, *, keys: Sequence[bytes | bytearray | str] | None None, codec: RequestStateCodec | None None, ttl: float 600.0, bind_principal: Callable[[ServerRequestContext[Any, Any]], str | None] | None authenticated_principal, audience: str | None None, ) - None: if (keys is None) (codec is None): raise ValueError(RequestStateSecurity takes exactly one of keys or codec) ...keys与codec必须且只能二选一ttl默认 600 秒且必须是正有限数bind_principal默认取authenticated_principal从 SDK 校验过的鉴权信息推导主体无配置时MCPServer安装的是RequestStateSecurity.ephemeral()等价物——os.urandom(32)生成的进程内一次性密钥见 ephemeral 类方法。封条里装了什么无论默认还是自定义线上的requestState都是一个加密且经认证的令牌。你的代码永远看不到它处理器和解析器写明文、读明文ctx.request_stateSDK 出站时封条、入站时校验。在完整性之外每个令牌还绑定一个时间窗口。每一轮都用新的过期时间重新封条所以RequestStateSecurity(ttl...)默认 600 秒约束的是每一轮的思考时间而不是整个流程。已认证的主体principal。当请求携带 SDK 校验过的 OAuth 访问令牌时状态绑定到该令牌的 client、issuer 与 subject为用户 A 铸造的状态在用户 B 名下会失败即便两人共享同一个 OAuth client。不提供 subject 的校验器会把绑定降级为仅 client 身份——而在基于 URL 的 client ID 下这个身份被该客户端软件的所有用户共享。当鉴权在 SDK 之外终结前置代理或传输层未认证时没有可绑定的主体该校验不生效除非你用RequestStateSecurity(bind_principal...)从自己的身份信号供给一个。无论你提供的令牌校验器给出哪些组件必须一致地给出某些请求带 subject、另一些不带的校验器会在流程中途改变主体在途轮次会被拒绝。原始请求。方法名、工具或 prompt 名或资源 URI、以及参数的摘要digest。把令牌重放到别的工具、别的参数或别的方法上都会失败。所问的确切问题。每一条解析器答案都钉在“展示给客户端的那份渲染后的问题”上——无论是答案首次到达的那一轮还是之后复用已记录答案的时候。重新部署一条措辞不同或 schema 有变的消息服务器会重新提问而不是消费过期答案。同一钉扎也反过来咬人从工具的参数派生消息而不是从逐调用数据派生。用时间戳或实时汇率构造的消息每轮渲染结果都不同每条已记录答案看起来都像过期的服务器会不停重新提问直到客户端的轮数上限终结这次调用。这些全部是 SDK 的工作不是你的工作也不是你自带 codec 时 codec 的工作。密钥轮换keys[0]负责封条新状态列表中的每一把都参与校验。零停机轮换分三阶段每阶段完整铺开后才进入下一阶段RequestStateSecurity(keys[OLD, NEW]) # 1: every instance learns to verify NEW; OLD still mints RequestStateSecurity(keys[NEW, OLD]) # 2: NEW mints; in-flight OLD state keeps verifying RequestStateSecurity(keys[NEW]) # 3: one ttl after phase 2 is fully out, retire OLD绝不要先提升铸造密钥用一个某些实例尚不会校验的密钥铸造会让在途轮次在部署中途坠落。密钥的作用域限于单个服务封条信封还把服务器名作为 audience受众声明携带因此另一个恰好共享秘密的服务铸造的令牌照样会被拒绝。该声明的独特性与名字的独特性相同——所以被赋予显式策略的服务器必须有一个真实的名字或显式设置RequestStateSecurity(audience...)否则在构造时抛异常。audience也服务于有意的多服务拓扑其中一个服务需要接受另一服务铸造的状态。零配置的默认值被豁免它的密钥从不离开进程audience 声明没有可补充的东西。自带加密BYO codecRequestStateSecurity(codec...)接受任何具备seal(bytes) - str与unseal(str) - bytes、且对非本方铸造的令牌抛出InvalidRequestState的对象。经典形态是背靠 KMS 的信封加密启动时解包一次数据密钥此后每个令牌的加密都在本地完成。示例完整收录于 docs_src/mrtr/tutorial005.pyimport os from cryptography.exceptions import InvalidTag from cryptography.hazmat.primitives.ciphers.aead import AESGCM from mcp.server import MCPServer from mcp.server.mcpserver import InvalidRequestState, RequestStateSecurity PREFIX kms1. # format version; fed to GCM as associated data, so it is bound under the tag def unwrap_data_key() - bytes: One KMS call at process start, kms.decrypt(CiphertextBlob...); every token after that is local crypto. return os.urandom(32) # stand-in for the unwrapped 32-byte data key class EnvelopeCodec: def __init__(self, data_key: bytes) - None: self._aesgcm AESGCM(data_key) def seal(self, payload: bytes) - str: nonce os.urandom(12) return PREFIX (nonce self._aesgcm.encrypt(nonce, payload, PREFIX.encode())).hex() def unseal(self, token: str) - bytes: if not token.startswith(PREFIX): raise InvalidRequestState(unknown token format) body token[len(PREFIX):] try: raw bytes.fromhex(body) if raw.hex() ! body: # only the exact string seal() produced verifies raise ValueError(non-canonical hex) return self._aesgcm.decrypt(raw[:12], raw[12:], PREFIX.encode()) except (ValueError, InvalidTag) as exc: raise InvalidRequestState(token failed verification) from exc mcp MCPServer(Deployer, request_state_securityRequestStateSecurity(codecEnvelopeCodec(unwrap_data_key())))注意契约细节前缀格式版本作为 GCM 的关联数据参与标签因此任何改动都会使标签校验失败非规范十六进制被显式拒绝保证只有seal()产出的确切字符串能通过unseal()。TTL、主体绑定、请求绑定都不是codec 的职责SDK 在seal之前把它们写进载荷在unseal之后对每一种codec 重新校验。codec 唯一的义务是完整性被篡改就抛异常以及理想情况下的保密性。校验失败时任何入站失败——被篡改、已过期、重放到另一请求或另一主体上、或用本服务器不知晓的密钥封条——都收到同一个回答{code: -32602, message: Invalid or expired requestState}对所有原因使用同一条冻结消息线上永不透露是哪一项校验失败真实原因写入服务器日志。到达tools/call、prompts/get、resources/read的每一个入站requestState都会被校验包括给一个从不铸造状态的处理器送来的那种。实践中最常见的拒绝并不是攻击者——而是进程本地的默认密钥遇到了重启前或来自另一实例的重试客户端重开整个流程而当这成为问题时keys[...]就是修复手段。手工构建的状态你自己设置的request_state从工具、prompt 或资源模板函数返回InputRequiredResult时与解析器状态使用同一套封条与校验机制零代码改动写明文、读明文上述所有绑定全部生效。SDK 即使配置了也无法替你钉住的一件事是问题身份它不知道你状态里的某条答案属于你的哪一条问题。如果你按问题索引存储答案就在状态里放入自己的问题标识符并在重试时校验它。低层级Server是“无电池”层级与MCPServer不同在你自己追加边界之前什么都不被封条你的request_state会原样正如所写跨越网络直到你这么做为止。一行启用方式见 低层级 Server 的“其他处理器”一节。版本边界2026-07-28 专属结果InputRequiredResult只存在于协议版本2026-07-28。Client默认的modeauto会在任意连接上探测并发现它。连接建立后client.protocol_version告诉你实际拿到的是什么。警告2026 之前的会话没有地方放InputRequiredResult。在modelegacy连接上从处理器返回它运行器runner无法把它序列化进已协商的版本客户端收到的是-32603Handler returned an invalid result错误。同时服务两代协议的服务器必须先检查ctx.protocol_version再考虑使用它。补充URL 模式的 elicitation在 2026 连接上走的正是这套机制。input_requests中的条目是一个 params 为ElicitRequestURLParams的ElicitRequest用户在带外完成流程你的客户端重试该调用。同一个循环没有新 API。高层服务器一侧的内容见 Elicitation。小结在 2026-07-28一个中途需要输入的服务器返回InputRequiredResult绝不向客户端发起请求。input_requests是它缺什么request_state是只有服务器读取的不透明续接令牌。Client替你跑重试循环注册elicitation_callback/sampling_callback/list_roots_callbackcall_tool就返回一个普通的CallToolResultinput_required_max_rounds默认 10为它设界仅含request_state的轮次有 50 ms→250 ms 的指数退避。要检视或持久化各轮次使用client.session.call_tool(..., allow_input_requiredTrue)自己接管while isinstance(result, InputRequiredResult)循环。在mcp.tool()上询问用户的依赖会替你产生该结果见 Dependencies低层级Server是手动形态。prompt 与资源同样参与mcp.prompt()或模板mcp.resource()函数自己返回InputRequiredResult在重试时读取ctx.input_responses。requestState以客户端提供的输入形式回流所以MCPServer默认对其封条——解析器状态与手工状态一视同仁——密钥进程本地多实例部署传RequestStateSecurity(keys[...])或自定义 codec让每个实例都能校验兄弟实例铸造的状态。封条把每个令牌绑定到时间窗口、原始请求以及在请求携带 SDK 校验过的鉴权或bind_principal提供你自己的身份信号时的已认证主体。这是取代“服务器主动采样”与其余推送式回传通道的机制废弃特性总览见 Deprecated features。【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址: https://gitcode.com/gh_mirrors/pythonsd/python-sdk创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考