人工智能MCP 服务MCP Clients【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址https://gitcode.com/gh_mirrors/pythonsd/python-sdk点击查看免费下载在 Model Context ProtocolMCP2026-07-28 规范中工具tool不再能半途通过回拨back-channel向客户端发起 elicitation 或 sampling 请求当一次tools/call需要用户才能给出的信息——一个选择、一次确认、一份凭证——时服务器改为返回一个InputRequiredResult让客户端补全后重试同一次调用。本文基于 python-sdk 官方仓库的 multi-round-trip 文档 与其教程源码完整讲解这套返回而非回拨return, dont call back协议的服务端、客户端两侧实现以及 SDK 如何默认对requestState进行加密封印与防重放校验并给出多实例部署下的密钥配置方案。读完你将在高低两层 API 中正确实现多轮交互并掌握RequestStateSecurity的 TTL、principal 绑定、密钥轮换与自定义加密等全部细节。从回拨到返回协议演进的动因在 2026-07-28 规范之前服务器在处理原始请求的中途可以向客户端主动打开一条反向请求如 elicitation、sampling 调用来获取缺失信息。2026-07-28 规范正式退役了这一反向通道back-channel服务器不再呼叫客户端而是返回一个特殊结果由客户端发起新一轮的普通请求继续对话。这套新机制的关键载体就是InputRequiredResult。在 SDK 类型系统中它被定义在 src/mcp-types/mcp_types/_types.py其约束非常明确result_type固定为input_required作为双结果响应联合类型中的判别标签input_requests服务器还需要什么是一个由服务器自选键名组成的 dict每个值是ElicitRequest、CreateMessageRequest或ListRootsRequest之一request_state一个不透明 token客户端在新一轮尝试中原样回传echo只有你的服务器能读懂它校验器强制至少携带input_requests或request_state之一spec MUST否则模型构造即报错。整个流程因此变得非常朴素客户端依次满足input_requests中的每一项请求然后以相同的工具名、相同的参数再次调用原工具把答案放在input_responses里、token 放在request_state里回传。服务器拿到缺失的信息后返回一个普通的CallToolResult。协议的每一段都是一次客户端 → 服务器的普通请求任何时刻都不会有反向流量。服务端实现低层Server与高层mcp.tool()低层Server手写on_call_tool返回联合类型在低层Server中on_call_tool处理器的返回类型被放宽为CallToolResult | InputRequiredResult——返回第二个类型就是服务端侧的全部 API。参见教程 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)三个关键点第一次调用时params.input_responses为None守卫条件触发handler 选择提问而不是回答客户端重试后发送回来的ElicitResult正躺在与input_requests中相同的键region之下服务器据此取回答案文件里其余部分显式input_schema、手拼的CallToolResult都是低层Server的常规内容详见 低层 Server 指南——本页只是在返回值里多了一个类型。高层MCPServer声明式依赖与函数体两种形式在mcp.tool()上你很少手工拼装InputRequiredResult。更常用的做法是声明式依赖声明一个询问用户的Elicit、在客户端 LLM 上做采样的Sample、或列出其 roots 的ListRoots依赖SDK 会自动替你返回InputRequiredResult这部分完整内容见 依赖Dependencies页面。需要特别注意的是两种形式不可混用一次调用只有一个input_responses/request_state通道。因此一个使用Resolve(...)参数的函数不能再从其函数体返回InputRequiredResult已声明返回InputRequiredResult的注册会被拒绝抛出InvalidSignature未声明却返回它则会在运行时使该次调用失败。从源码看这一约束由 src/mcp/server/mcpserver/resolve.py 的returns_input_required检查实现它会递归解析返回类型注解中的联合类型Union/|是否存在input_required分支因为两条流程会互相覆盖同一个通道所以必须互斥。当依赖式不适用时mcp.tool()函数体也可以直接返回InputRequiredResult见下文 prompt/resource 示例的同一模式。不止工具prompts/get与resources/read同样参与tools/call并不特殊在 2026-07-28 下服务器同样可以用这种方式应答prompts/get与resources/read。在高层的MCPServer上mcp.prompt()函数或mcp.resource()模板函数自己返回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回声返回时会被校验与服务器上其他一切状态同等对待静态mcp.resource()函数不参与它们不接收Context因此永远无法读取重试只有模板资源才能提问下文时代规则同样适用在 pre-2026 会话中返回InputRequiredResult会得到与警告中相同的-32603错误。从上下文实现看src/mcp/server/mcpserver/context.py 的Context.input_responses正是从请求参数中解出的InputResponses它是重试轮次中读取答案的唯一入口。客户端侧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正是 pre-2026 服务器反向通道中elicitation/create会触发的那个回调sampling_callback对应sampling/createMessagelist_roots_callback对应roots/list。2026-07-28 下独立的服务器→客户端 RPC 已不存在但完全相同的ElicitRequest/CreateMessageRequest/ListRootsRequestpayload 现在搭载在input_requests内部仍被分派给同样的三个回调——一套回调同时服务两个时代。call_tool对调用方始终返回朴素的CallToolResult中间轮次完全不可见get_prompt与read_resource驱动着同一套循环。check如果漏注册回调循环会在第一轮就失败——SDK 的替代回调会对每次 elicitation 都返回错误call_tool抛出MCPError消息为Elicitation not supported。循环是有上限的Client(..., input_required_max_rounds10)是默认上限源码中定义于 src/mcp/client/client.py服务器若持续返回InputRequiredResult超过上限call_tool会抛出InputRequiredRoundsExceededError。另外如果某一轮只携带request_state而没有input_requests服务器在说还没准备好Client会短暂休眠50ms 起步、倍增直至 250ms 封顶再重试避免对服务器造成 busy-polling。手动接管循环client.session.call_tool(..., allow_input_requiredTrue)单进程客户端用自动循环就够了但以下场景需要你亲自接管客户端是分布式的向用户展示问题的进程不是调用call_tool的进程重试由另一个 worker 发出。request_state就是你要跨过这道边界的可持久化 token经你自己的存储input_responses则是另一端随它一起发回的答案想逐轮检查记录或审计每一轮input_requests条目、拒绝某些请求类型、或在各段之间施加自定义退避想要墙钟wall-clock时限而非轮次计数上限用自己的循环包上anyio.fail_after(...)而不是依赖input_required_max_rounds。这时下探到底层 sessionallow_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 result要点client.session.call_tool(..., allow_input_requiredTrue)把返回类型放宽为CallToolResult | InputRequiredResultisinstance再把它收窄回来session 层默认allow_input_requiredFalse参见 src/mcp/client/session.pyrequest_state现在掌握在你自己手里在各段之间把它落盘对话就能从全新的进程继续对input_requests中的每一项你都要在input_responses中同键放一个InputResponsefulfil就是你接入 UI 的位置每一段都使用相同的工具名、相同的arguments——重试是把原始调用再做一遍而不是一个新的方法。保护requestState默认封印与RequestStateSecurity为什么必须保护request_state是客户端提供的输入以上所有讨论都把request_state当作一个回声echo在网络上它确实只是如此。但客户端会在各段之间持有它——跨进程落盘正是上一节推荐的做法——所以回到服务器的内容实际上是客户端提供的输入它可能被篡改、已过期甚至是从完全不同的调用里偷来的。规范要求只要该状态可能影响授权、资源访问或业务逻辑服务器就必须对其做完整性保护并在校验失败时拒绝该轮次。MCPServer默认就做保护。每个服务器都会在进程启动时生成一把密钥封印所有出站的requestState并校验每一个回声——resolver 状态与手工拼装的状态一视同仁。你无需配置任何东西写明文、读明文网络上永远只出现一个不透明且加密的 token。默认密钥的生命周期单进程之外必须配置默认密钥与进程同生共死——这是部署到单进程之外前唯一必须知道的事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或必须扛过重启时是必需的每个实例都能校验任何兄弟实例铸造的 token。机制完全相同只是用你的密钥替换了生成的密钥想要自己的加密例如 KMS 或既有 token 服务改用RequestStateSecurity(codec...)契约见下文。从源码看RequestStateSecurity定义在 src/mcp/server/request_state.py它要求keys与codec恰好提供其一否则抛ValueError内置 codec 是 AES-256-GCM 的AESGCMRequestStateCodecRequestStateSecurity.ephemeral()正是MCPServer未传request_state_security时安装的策略——os.urandom(32)生成的进程本地密钥。密钥至少 32 字节的校验也在 codec 构造时强制不满足会直接报错并提示用secrets.token_hex(32)生成。封印携带的绑定时间窗、principal、原始请求与问题本身无论默认还是配置网络上的requestState都是一个加密且认证的 token。你的代码永远看不到它的真身handlers 与 resolvers 写明文、读明文ctx.request_stateSDK 在出口封印、入口校验。除完整性之外每个 token 还被绑定到以下四类要素RequestStateBoundary中间件在 src/mcp/server/request_state.py 中实现只对tools/call、prompts/get、resources/read三个多轮载体生效时间窗每一轮都用新的过期时间重新封印因此RequestStateSecurity(ttl...)默认 600 秒限制的是每一轮的思考时间而不是整个流程认证主体principal当请求携带 SDK 校验过的 OAuth 访问令牌时状态被绑定到令牌的 client、issuer 与 subject——为一个用户铸造的状态在另一个用户下必然失败即使两者共享同一个 OAuth client。verifier 不提供 subject 时绑定降级为仅客户端身份在基于 URL 的 client ID 下被该软件的所有用户共享。当认证在 SDK 之外终结前置代理或传输本身未认证时没有 principal 可绑定该校验处于惰性状态——除非用RequestStateSecurity(bind_principal...)从你自己的身份信号提供。无论 verifier 提供哪些组件都必须保持一致时而带 subject、时而不带的 verifier 会在流程中途改变 principal导致在途轮次被拒绝原始请求方法、工具或 prompt 名或资源 URI以及参数的摘要digest。token 被换到不同的工具、不同的参数或不同的方法上重放都会失败——_request_identity对resources/read取 URI对其他方法取namearguments的 SHA-256 摘要src/mcp/server/request_state.py被问的确切问题每一条 resolver 答案都被钉在客户端看到的那条渲染问题上无论它首次到达的那一轮还是之后重放已记录答案时。用改过措辞的消息或改过的 schema 重新部署服务器会重新提问而不是消费过期答案。这个钉扎是双向的消息要由工具的参数派生而不是每次调用都变的数据——用 timestamp 或实时报价拼出来的消息每轮渲染都不同于是每条记录的答案都显得过期服务器会一直重新提问直到客户端的轮次上限终止调用。以上全部是 SDK 的职责不是你的也不是若你自带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 声明因此另一个碰巧共享了同一密钥的服务铸造的 token 照样被拒。该声明的区分度与名字本身相当——所以被赋予显式策略的服务器必须有真实名字或设置RequestStateSecurity(audience...)未命名的服务器在构造时直接抛错。audience也服务于刻意的多服务拓扑一个服务需要接受另一个服务铸造的状态时。零配置默认是豁免的它的密钥从不离开进程audience 声明没有可添加的东西。自带加密RequestStateSecurity(codec...)RequestStateSecurity(codec...)接受任何满足seal(bytes) - str与unseal(str) - bytes的对象且unseal对任何非自己铸造的 token 抛出InvalidRequestState。经典形态是对 KMS 的信封加密在启动时解开一次数据密钥data key随后每个 token 的加密都在本地完成。参见 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())))TTL、principal 绑定与请求绑定不是codec 的职责SDK 会在seal之前把它们盖进 payload并在unseal之后重新校验对所有 codec 一律如此。codec 的唯一义务是完整性被篡改就抛错与理想情况下机密性。协议契约还要求unseal(seal(payload))必须往返一致两个方法都是同步的因此要缓存密钥材料而不是每个 token 调一次 KMStoken 永远不标明自己的算法用版本前缀并绑在认证标签下遵循 RFC 8725比较必须是常数时间的。校验失败时的统一响应无论入站失败是哪一种——被篡改、过期、在不同请求或 principal 下重放、或封印在某个本服务器不认识的密钥下——都得到同一条固定响应{code: -32602, message: Invalid or expired requestState}对所有原因冻结同一条消息是为了让网络永不泄露是哪一项校验失败真正的原因只进服务器日志。tools/call、prompts/get、resources/read上的每一个入站requestState都会被校验包括发给从不铸造状态的 handler 的。实践中最常见的拒绝并不是攻击者——而是进程本地默认密钥遇上了重启前或另一实例来的重试客户端重开流程即可keys[...]就是在意这个场景时的修复。从源码看这一固定错误由_reject以MCPError(codeINVALID_PARAMS, ...)抛出src/mcp/server/request_state.py真实原因仅通过logger.warning记录。手工拼装的状态hand-built state你自己设置的request_state从工具、prompt 或资源模板函数返回InputRequiredResult时会走与 resolver 状态完全相同的封印与校验机制零代码改动写明文、读明文上文所有绑定全部生效。SDK 唯一无法替你钉住即使配置了的是问题身份它不知道你状态里的一条答案属于你哪个问题。如果你按问题索引存储答案就把自己的问题标识符放进状态里并在重试时核对。低层Server是不带电池的层级与MCPServer不同在你亲自附加边界之前什么都不会封印你的request_state会原样过网。这一行式的 opt-in 见 低层 Server 指南的其他 handlers一节。时代规则这是 2026-07-28 才有的结果InputRequiredResult只在协议版本2026-07-28中存在。Client默认的modeauto会在任何连接上自动发现它连接之后client.protocol_version会告诉你实际协商到的版本。warningpre-2026 会话没有地方安放InputRequiredResult。在modelegacy连接上从 handler 返回它runner 无法把它序列化进协商好的版本——客户端会收到-32603Handler returned an invalid result错误。一个同时服务两个时代的服务器必须在诉诸它之前检查ctx.protocol_version。infoURL 模式 elicitation在 2026 连接上正是借助这一机制input_requests中的条目是一个 params 为ElicitRequestURLParams的ElicitRequest用户带外完成流程后你的客户端重试该调用。同一个循环没有新 API。高层服务端那一半见 Elicitation 页面。要点回顾在 2026-07-28 下需要中途输入的服务器返回一个InputRequiredResult永远不会向客户端打开请求input_requests是它还需要的东西request_state是只有服务器读的不透明恢复 tokenClient替你跑重试循环注册elicitation_callback/sampling_callback/list_roots_callbackcall_tool就返回朴素的CallToolResult。input_required_max_rounds默认 10限定轮数想检查或持久化各轮用client.session.call_tool(..., allow_input_requiredTrue)自己掌控while isinstance(result, InputRequiredResult)循环在mcp.tool()上一个询问用户的依赖会替你产出该结果见 Dependencies 页面低层Server是手工形式prompt 与资源同样参与mcp.prompt()或模板mcp.resource()函数自己返回InputRequiredResult并在重试时读取ctx.input_responsesrequestState会作为客户端提供的输入回来所以MCPServer默认封印它——resolver 状态与手工状态一视同仁——用一把进程本地密钥多实例部署传入RequestStateSecurity(keys[...])或自定义 codec让每个实例都能校验兄弟实例铸造的状态。封印把每个 token 绑定到时间窗、原始请求以及当请求携带 SDK 校验过的认证或bind_principal提供你自己的身份信号时的认证主体详见上文 Protegendo/ProtectingrequestState一节。这套机制正是取代服务器发起的 sampling 与其余 push 风格反向通道的方案相关退役内容见 已废弃功能Deprecated features。赞分享人工智能MCP 服务MCP Clients【免费下载链接】python-sdkThe official Python SDK for Model Context Protocol servers and clients项目地址https://gitcode.com/gh_mirrors/pythonsd/python-sdk点击查看免费下载相关推荐Python SDK 多轮往返Multi-Round-Trip请求实战从 InputRequiredResult 到 requestState 保护Python SDK 多轮往返Multi Round Trip请求实战从 InputRequiredResult 到 requestState 保护 导读人工智能MCP 服务MCP ClientsMCP Python SDK 多轮往返请求Multi-round-trip实战从 InputRequiredResult 到 requestState 安全机制MCP Python SDK 多轮往返请求Multi round trip实战从 InputRequiredResult 到 requestState 安人工智能MCP 服务MCP Clients免费开源水下机器人仿真5分钟掌握UUV Simulator终极指南免费开源水下机器人仿真5分钟掌握UUV Simulator终极指南 UUV Simulator是一款基于Gazebo和ROS的免费开源水下机器人仿真平台专为人工智能MCP 服务MCP Clients上一篇Vuescroll 插件架构解析深入理解组件设计原理与核心实现下一篇Apache ECharts终极指南5个简单技巧打造惊艳数据可视化图表创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考