企业官网建设流程全解析

• 人工智能
• MCP 服务
• MCP Clients

在 Model Context Protocol（MCP）2026-07-28 规范中，工具（tool）不再能"半途"通过回拨（back-channel）向客户端发起 elicitation 或 sampling 请求；当一次tools/call需要用户才能给出的信息——一个选择、一次确认、一份凭证——时，服务器改为返回一个InputRequiredResult，让客户端补全后重试同一次调用。本文基于 python-sdk 官方仓库的 multi-round-trip 文档 与其教程源码，完整讲解这套"返回而非回拨"（return, don't 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.py：

from mcp.server import Server, ServerRequestContext from mcp.types import ( CallToolRequestParams, CallToolResult, ElicitRequest, ElicitRequestFormParams, ElicitResult, InputRequiredResult, ListToolsResult, PaginatedRequestParams, TextContent, Tool, ) ASK_REGION = ElicitRequest( params=ElicitRequestFormParams( message="Which 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( name="provision", description="Provision 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_state="provision-v1") name = (params.arguments or {})["name"] text = f"Provisioned {name!r} in {answer.content['region']}." return CallToolResult(content=[TextContent(type="text", text=text)]) server = Server("Provisioner", on_list_tools=list_tools, on_call_tool=call_tool)

三个关键点：

1. 第一次调用时params.input_responses为None，守卫条件触发，handler 选择"提问"而不是"回答"；
2. 客户端重试后，发送回来的ElicitResult正躺在与input_requests中相同的键（"region"）之下，服务器据此取回答案；
3. 文件里其余部分（显式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.py：

from 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( params=ElicitRequestFormParams( message="Who 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(f"Write 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.py：

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={"region": "eu-west-1"}) async def main() -> None: async with Client("http://127.0.0.1:8000/mcp", elicitation_callback=handle_elicitation) as client: result = await client.call_tool("provision", {"name": "orders"}) print(result.content)

值得注意的设计是回调跨时代复用：这个elicitation_callback正是 pre-2026 服务器反向通道中elicitation/create会触发的那个回调；sampling_callback对应sampling/createMessage，list_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_rounds=10)是默认上限（源码中定义于 src/mcp/client/client.py），服务器若持续返回InputRequiredResult超过上限，call_tool会抛出InputRequiredRoundsExceededError。另外，如果某一轮只携带request_state而没有input_requests（服务器在说"还没准备好"），Client会短暂休眠（50ms 起步、倍增直至 250ms 封顶）再重试，避免对服务器造成 busy-polling。

手动接管循环：client.session.call_tool(..., allow_input_required=True)

单进程客户端用自动循环就够了，但以下场景需要你亲自接管：

• 客户端是分布式的：向用户展示问题的进程不是调用call_tool的进程，重试由另一个 worker 发出。request_state就是你要跨过这道边界的可持久化 token（经你自己的存储），input_responses则是另一端随它一起发回的答案；
• 想逐轮检查：记录或审计每一轮input_requests条目、拒绝某些请求类型、或在各段之间施加自定义退避；
• 想要墙钟（wall-clock）时限而非轮次计数上限：用自己的循环包上anyio.fail_after(...)，而不是依赖input_required_max_rounds。

这时下探到底层 session，allow_input_required=True会把联合类型直接交到你手里。参见 docs_src/mrtr/tutorial002.py：

from 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(f"this client cannot answer a {request.method!r} request") return ElicitResult(action="accept", content={"region": "eu-west-1"}) async def provision(client: Client, name: str) -> CallToolResult: result = await client.session.call_tool("provision", {"name": name}, allow_input_required=True) 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_responses=responses, request_state=result.request_state, allow_input_required=True, ) return result

要点：

1. client.session.call_tool(..., allow_input_required=True)把返回类型放宽为CallToolResult | InputRequiredResult，isinstance再把它收窄回来（session 层默认allow_input_required=False，参见 src/mcp/client/session.py）；
2. request_state现在掌握在你自己手里：在各段之间把它落盘，对话就能从全新的进程继续；
3. 对input_requests中的每一项，你都要在input_responses中同键放一个InputResponse；fulfil就是你接入 UI 的位置；
4. 每一段都使用相同的工具名、相同的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_security=RequestStateSecurity(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 的AESGCMRequestStateCodec；RequestStateSecurity.ephemeral()正是MCPServer未传request_state_security=时安装的策略——os.urandom(32)生成的进程本地密钥。密钥至少 32 字节的校验也在 codec 构造时强制，不满足会直接报错并提示用secrets.token_hex(32)生成。

封印携带的绑定：时间窗、principal、原始请求与问题本身

无论默认还是配置，网络上的requestState都是一个加密且认证的 token。你的代码永远看不到它的真身：handlers 与 resolvers 写明文、读明文（ctx.request_state），SDK 在出口封印、入口校验。除完整性之外，每个 token 还被绑定到以下四类要素（RequestStateBoundary中间件在 src/mcp/server/request_state.py 中实现，只对tools/call、prompts/get、resources/read三个多轮载体生效）：

1. 时间窗：每一轮都用新的过期时间重新封印，因此RequestStateSecurity(ttl=...)（默认 600 秒）限制的是每一轮的思考时间，而不是整个流程；
2. 认证主体（principal）：当请求携带 SDK 校验过的 OAuth 访问令牌时，状态被绑定到令牌的 client、issuer 与 subject——为一个用户铸造的状态在另一个用户下必然失败，即使两者共享同一个 OAuth client。verifier 不提供 subject 时，绑定降级为仅客户端身份（在基于 URL 的 client ID 下被该软件的所有用户共享）。当认证在 SDK 之外终结（前置代理）或传输本身未认证时，没有 principal 可绑定，该校验处于惰性状态——除非用RequestStateSecurity(bind_principal=...)从你自己的身份信号提供。无论 verifier 提供哪些组件，都必须保持一致：时而带 subject、时而不带的 verifier 会在流程中途改变 principal，导致在途轮次被拒绝；
3. 原始请求：方法、工具或 prompt 名（或资源 URI），以及参数的摘要（digest）。token 被换到不同的工具、不同的参数或不同的方法上重放，都会失败——_request_identity对resources/read取 URI，对其他方法取name+arguments的 SHA-256 摘要（src/mcp/server/request_state.py）；
4. 被问的确切问题：每一条 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.py：

import 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_security=RequestStateSecurity(codec=EnvelopeCodec(unwrap_data_key())))

TTL、principal 绑定与请求绑定不是codec 的职责：SDK 会在seal之前把它们盖进 payload，并在unseal之后重新校验，对所有 codec 一律如此。codec 的唯一义务是完整性（被篡改就抛错）与（理想情况下）机密性。协议契约还要求：unseal(seal(payload))必须往返一致；两个方法都是同步的，因此要缓存密钥材料而不是每个 token 调一次 KMS；token 永远不标明自己的算法（用版本前缀并绑在认证标签下，遵循 RFC 8725）；比较必须是常数时间的。

校验失败时的统一响应

无论入站失败是哪一种——被篡改、过期、在不同请求或 principal 下重放、或封印在某个本服务器不认识的密钥下——都得到同一条固定响应：

{"code": -32602, "message": "Invalid or expired requestState"}

对所有原因冻结同一条消息，是为了让网络永不泄露是哪一项校验失败；真正的原因只进服务器日志。tools/call、prompts/get、resources/read上的每一个入站requestState都会被校验，包括发给从不铸造状态的 handler 的。实践中最常见的拒绝并不是攻击者——而是进程本地默认密钥遇上了重启前或另一实例来的重试；客户端重开流程即可，keys=[...]就是在意这个场景时的修复。从源码看，这一固定错误由_reject以MCPError(code=INVALID_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默认的mode="auto"会在任何连接上自动发现它；连接之后，client.protocol_version会告诉你实际协商到的版本。

warning：pre-2026 会话没有地方安放InputRequiredResult。在mode="legacy"连接上从 handler 返回它，runner 无法把它序列化进协商好的版本——客户端会收到-32603"Handler returned an invalid result"错误。一个同时服务两个时代的服务器必须在诉诸它之前检查ctx.protocol_version。

info：URL 模式 elicitation在 2026 连接上正是借助这一机制：input_requests中的条目是一个 params 为ElicitRequestURLParams的ElicitRequest；用户带外完成流程后，你的客户端重试该调用。同一个循环，没有新 API。高层服务端那一半见 Elicitation 页面。

要点回顾

• 在 2026-07-28 下，需要中途输入的服务器返回一个InputRequiredResult，永远不会向客户端打开请求；
• input_requests是它还需要的东西；request_state是只有服务器读的不透明恢复 token；
• Client替你跑重试循环：注册elicitation_callback/sampling_callback/list_roots_callback，call_tool就返回朴素的CallToolResult。input_required_max_rounds（默认 10）限定轮数；
• 想检查或持久化各轮，用client.session.call_tool(..., allow_input_required=True)，自己掌控while isinstance(result, InputRequiredResult)循环；
• 在@mcp.tool()上，一个询问用户的依赖会替你产出该结果（见 Dependencies 页面）；低层Server是手工形式；
• prompt 与资源同样参与：@mcp.prompt()或模板@mcp.resource()函数自己返回InputRequiredResult，并在重试时读取ctx.input_responses；
• requestState会作为客户端提供的输入回来，所以MCPServer默认封印它——resolver 状态与手工状态一视同仁——用一把进程本地密钥；多实例部署传入RequestStateSecurity(keys=[...])（或自定义 codec），让每个实例都能校验兄弟实例铸造的状态。封印把每个 token 绑定到时间窗、原始请求，以及（当请求携带 SDK 校验过的认证或bind_principal=提供你自己的身份信号时的）认证主体（详见上文 Protegendo/ProtectingrequestState一节）。

这套机制正是取代服务器发起的 sampling 与其余 push 风格反向通道的方案；相关退役内容见 已废弃功能（Deprecated features）。

• 人工智能
• MCP 服务
• MCP Clients