生命不息，折腾不止。前三课我们把 Server 造出来、挂上公网、把人的手感补齐了；这一课补上这版协议最实在的两块——Server 想反过来问你，怎么问（MRTR），它的东西变了怎么主动告诉你（subscriptions/listen），最后把前面三课的 Server 真正收尾上线：网关头、限流、可观测、灰度回滚，一份能长期挂着跑的运维清单。

上一篇（第三课）结尾我留了个坑，说这一课要钻到 2026-07-28 这版协议里最实在的变化。写之前我又去核了一遍官方文档，越核越觉得：这两件事其实是同一件事的两面——协议把「会话」拿掉了，随之而来的问题是「Server 怎么还能跟客户端说上话」。答案是两条路：

  1. Server 要东西：不喊了，改成返回一个「我需要输入」的结果，客户端补答案后重发同一个请求。这就是 MRTR（Multi Round-Trip Requests，多轮往返请求）；
  2. Server 有变化要通知：不再是随手推，改成客户端订阅一条流。这就是 subscriptions/listen。

听起来像术语，落到代码上其实就几行。我们一层层拆。

一、先搞明白：2026-07-28 之后，Server 不能再「回头喊」了

要理解 MRTR，得先看它替掉的东西有多别扭。

老协议（2025-11-25 及更早）里，Server 是可以在处理你这次 tools/call 的中途，「顺着那条双向流」回头给客户端发一个请求的：

  • 想让你确认一下？发 elicitation/create；
  • 想借客户端的模型跑一段？发 sampling/createMessage；
  • 想知道客户端的工作目录？发 roots/list。

功能上很爽，代价是：那条连接必须一直开着。于是你的部署模型被迫变成「有状态」——粘性路由、共享 session 存储、Serverless 上根本没法跑（一个长调用把 SSE 流钉死在那儿）。

2026-07-28 这版协议干的第一件大事，就是把协议内核改成无状态的（SEP-2575、SEP-2567）：

旧世界（≤ 2025-11-25） 新世界（2026-07-28）
握手 initialize / initialized 没了，每个请求自带 _meta（协议版本、客户端信息、能力）
会话 Mcp-Session-Id header 没了，请求落在哪个实例都一样
提前知道能力 握手时协商 可选地调一次 server/discover
Server 反过来要东西 顺着长连接发请求 只能返回 InputRequiredResult，客户端重试
变更通知 独立 GET 流 / resources/subscribe subscriptions/listen 一条订阅流
网关怎么路由 得解析 body 看 Mcp-Method / Mcp-Name 头就行

看最后两行就明白这一课的分量了：老的那套「Server 主动发请求」被正式移除，不再是「不推荐」，是 breaking change。协议里那句原话很硬气——Server 只能用 MRTR 模式发这类请求，「之前的模式不再支持」。

所以今天的第一件事：学会「返回」而不是「喊」。

二、MRTR 四条腿：InputRequiredResult、inputRequests、requestState、inputResponses

MRTR（规范编号 SEP-2322）的流程就四步，官方写得比谁都清楚：

  1. 客户端发一个原始请求（通常是 tools/call）；
  2. Server 发现信息不够，不报错，而是返回一个 InputRequiredResult，里面装着它还需要的东西；
  3. 客户端去把答案凑齐（问用户、调模型、列出 roots……）；
  4. 客户端重发同一个请求，把答案挂在 inputResponses 上、把 Server 给的 requestState 原样带回来。

猜猜重发落到哪儿？可以是另一个实例。因为接连性的东西不在服务器的内存里，而在那个 requestState 字符串里。这就是「不需要粘性路由」的全部秘密。

两个字段最关键：

  • inputRequests：Server 还需要什么。一个 map，键是 Server 自己起的名字，值是一个请求对象（ElicitRequest / CreateMessageRequest / ListRootsRequest）；
  • requestState：一个对客户端不透明的令牌。它只负责原样回传，内容是什么只有 Server 自己懂——里面可以塞明文 JSON、base64、加密 JWT，随便你。

线上真实形状长这样（这是规范级的报文，不是某家 SDK 的写法）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
// Server 返回这个：我还没法干完，先问你一句
{
  "inputRequests": {
    "confirm-delete": {
      "type": "elicitation",
      "message": "Delete 42 records? This cannot be undone.",
      "requestedSchema": {
        "type": "object",
        "properties": { "confirm": { "type": "boolean" } },
        "required": ["confirm"]
      }
    }
  },
  "requestState": "eyJzdGVwIjoiYXdhaXQtY29uZmlybSIsImJhdGNoIjo0Mn0="
}

客户端凑完答案，重发原来的调用：

1
2
3
4
5
6
7
8
{
  "name": "delete_records",
  "arguments": { "query": "status = 'archived'" },
  "inputResponses": {
    "confirm-delete": { "confirm": true }
  },
  "requestState": "eyJzdGVwIjoiYXdhaXQtY29uZmlybSIsImJhdGNoIjo0Mn0="
}

规范对客户端有两条 MUST，值得背下来，排障时全靠它俩定位：

  1. 收到带 inputRequests 的结果，必须先把这些输入凑齐，再重发原请求；
  2. 收到 requestState，重发时必须把那个值原样带回来。

顺便记两个小概念：

  • 每个结果现在都带 resultType：正常收尾是 "complete"，要你补输入是 "input_required"。老客户端不认识这个字段时，按「正常完成」处理——这是兼容性设计的巧劲。
  • 一个安全约束：Server 只在处理客户端请求的过程中才能提要求。客户端不会在没人发起动作的时候，突然收到一个 Server 弹出来的确认框。

还有一句特别容易把人绕晕的话，我单独拎出来：

roots、sampling、协议日志这三个能力被 SEP-2577 废弃了，但它们的载荷类型活着——ElicitRequest / CreateMessageRequest / ListRootsRequest 现在被塞进 InputRequiredResult.input_requests 里继续用。

也就是说：废掉的是「Server 主动发这些 RPC」这条老通道，不是这些请求本身。 你以前写过的 elicitation 表单，格式一点没变，只是换了个运载方式。

三、推荐姿势：把问题挂在参数上（Resolve + Elicit）

好，原理讲完。接下来是我最想让你记住的一节：在 Python SDK（v2，稳定版线）里，你基本不用手写 InputRequiredResult。

因为不是所有值都该由模型给。价格、身份、权限、确认——这些模型完全可能一本正经地编一个出来。SDK 给的做法是把这类参数「挂」起来：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from typing import Annotated
from pydantic import BaseModel
from mcp.server import MCPServer
from mcp.server.mcpserver import Resolve

mcp = MCPServer("Bookshop")
INVENTORY = {"Dune": 7, "Neuromancer": 0}

class Stock(BaseModel):
    title: str
    copies: int

async def check_stock(title: str) -> Stock:
    return Stock(title=title, copies=INVENTORY.get(title, 0))

@mcp.tool()
async def reserve_book(title: str, stock: Annotated[Stock, Resolve(check_stock)]) -> str:
    """Reserve a copy of a book."""
    if stock.copies == 0:
        return f"{title!r} is out of stock."
    return f"Reserved {title!r} ({stock.copies - 1} copies left)."

三个要点：

  • Resolve(check_stock) 的意思是：这个参数不向模型要，工具跑之前由 check_stock 填；
  • 它对模型不可见——reserve_book 的输入 schema 里根本没有 stock 这个字段（跟上一课讲的 Context 一个道理）；
  • 就算哪个客户端硬塞一个 stock 参数进来，也会被忽略。工具只能收到 resolver 给的值。

resolver 也不一定知道答案。它可以直接把问题弹给用户——返回 Elicit(message, 模型)，SDK 替你走完后面那一整套：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
from typing import Annotated
from pydantic import BaseModel
from mcp.server import MCPServer
from mcp.server.mcpserver import (
    AcceptedElicitation, CancelledElicitation, DeclinedElicitation,
    Elicit, ElicitationResult, Resolve,
)

mcp = MCPServer("Files")
_FOLDERS: dict[str, list[str]] = {"/tmp/empty": [], "/tmp/project": ["main.py", "README.md"]}

class Confirm(BaseModel):
    ok: bool

async def confirm_delete(path: str) -> Confirm | Elicit[Confirm]:
    """要删非空目录时，先问一句。"""
    file_count = len(_FOLDERS.get(path, []))
    if file_count == 0:
        return Confirm(ok=True)          # 空的，没什么好确认的，不产生往返
    return Elicit(f"{path} 里有 {file_count} 个文件。还要删吗？", Confirm)

@mcp.tool()
async def delete_folder(
    path: str,
    confirm: Annotated[ElicitationResult[Confirm], Resolve(confirm_delete)],
) -> str:
    """Delete a folder, asking for confirmation when it is not empty."""
    match confirm:
        case AcceptedElicitation(data=Confirm(ok=True)):
            _FOLDERS.pop(path, None)
            return f"deleted {path}"
        case AcceptedElicitation():
            return "kept the folder"
        case DeclinedElicitation():
            return "declined: folder not deleted"
        case CancelledElicitation():
            return "cancelled: folder not deleted"

这段代码信息量很大，逐条拆：

  • Elicit(message, Model) 里的 pydantic 模型，就是客户端要渲染的表单——Confirm 里有什么字段，界面上就是什么字段；
  • 注解决定你要哪种结局：
    • 写 Annotated[T, Resolve(fn)]（不包 ElicitationResult）→ 只拿解包后的值，用户拒绝或取消，整个调用直接中止；
    • 写 Annotated[ElicitationResult[T], Resolve(fn)] → 拿完整结果，自己在 case 里分支（接受/拒绝/取消三种结局）。
  • 不用手写 InputRequiredResult：SDK 根据协商到的协议版本自己选通道——>= 2026-07-28 走 MRTR（批量把问题带出去，客户端重试时带答案回来）；<= 2025-11-25 就走老的同步 elicitation/create。同一份工具代码，两个年代都能跑，这是 v2 SDK 最省心的地方。

再补两个多轮细节，都是踩过才知道的：

  • 互相独立的 resolver，会在同一轮里批量问（一次往返把所有问题带出去，用户一口气填完）；有依赖的（A 的答案决定了 B 问什么）就排到下一轮问。每个问题只会问一次，答案随 request_state 跨轮携带；
  • resolver 本体每轮都可能重跑，但被记录下来的答案只在「resolver 再次问出同一个问题」时才被采用——也就是说，resolver 自己算出来的值永远优先于客户端回带的内容。

最后是两个必须记住的硬约束：

  1. 两套写法不能混。一次调用只有一条 input_responses / request_state 通道，所以「用了 Resolve(...) 参数的工具」不能再在自己的函数体里返回 InputRequiredResult；
  2. 返回值声明了 InputRequiredResult 却没用，注册期就被拒（InvalidSignature）；没声明就返回，运行时调用失败。这类错误全在启动时报出来，比在线上一半崩一半强多了。

客户端侧呢？简单到有点不真实：

1
2
3
4
5
6
7
8
9
10
11
12
from mcp import Client
from mcp.client import ClientRequestContext
from mcp.types import ElicitRequestParams, ElicitResult

async def handle_elicitation(context: ClientRequestContext, params: ElicitRequestParams) -> ElicitResult:
    # 这里接你的界面：命令行 input()、Web 表单、Telegram 按钮……
    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)

重试循环是 Client 替你跑的，你只要注册回调。轮数上限是 input_required_max_rounds（默认 10 轮），防的就是两边互相「你再说一句」到天荒地老。

四、手工挡：低层 Server 自己返回结果、自己跑循环

@mcp.tool() 是「常用挡」，但你总有需要对整个过程动手的时候——比如自己决定 request_state 里塞什么（把业务进度打包进去，让跨进程续跑成为可能）。这时候用低层的 Server：

1
2
3
4
5
6
7
8
9
10
11
async def call_tool(ctx, params) -> types.CallToolResult | types.InputRequiredResult:
    answer = (params.input_responses or {}).get("region")
    if not isinstance(answer, ElicitResult) or answer.content is None:
        # 第一次调用：input_responses 是 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)])

看出来了吗？服务端 API 就多了一样东西：on_call_tool 的返回类型从 CallToolResult 扩成了 CallToolResult | InputRequiredResult。 返回第二个，就是「我要问你」。

客户端如果不想让 SDK 自动跑循环，也可以自己拿着方向盘：

1
2
3
4
5
6
7
8
9
result = await client.session.call_tool("provision", {"name": name}, allow_input_required=True)
while isinstance(result, InputRequiredResult):
    responses = {k: fulfil(req) for k, req 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,
    )

手动挡的价值在这一行：request_state 现在握在你手里了——两腿之间把它写进磁盘，那么「用户填到一半去开会」这种事就不怕了，换个进程接着来。（这也是无状态部署真正能落地的地方：接续用的东西全在报文里。）

还有一个容易被忽略的扩展：tools/call 不特殊。 在 2026-07-28 里，prompts/get 和 resources/read 也能这么答——@mcp.prompt() 函数、或者模板型的 @mcp.resource() 函数，可以直接返回 InputRequiredResult，然后在重试时从 ctx.input_responses 里读答案：

1
2
3
4
5
6
7
@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']}.")]

注意这里的分寸：静态 @mcp.resource() 不参与——它连 Context 都没有，压根读不到重试的答案。只有模板型资源能问。

五、必须补的安全课：request_state 是「客户端递回来的输入」

这一节请务必看，因为它是那种「不看也能跑起来，但迟早出事」的东西。

关键认知一句话：request_state 在协议上确实只是个回显，但它中途一直握在客户端手里。 客户端可以改它、可以放到过期、甚至能把另一个调用的状态搬过来。所以它回到你 Server 时，身份是客户端提供的输入，不是你的内存。

规范的要求很明确：只要这个 state 能影响授权、资源访问或业务逻辑，Server 就必须对它做完整性保护，验不过就拒绝这一轮。

好消息是 MCPServer 默认就帮你封好了——你写明文、读明文，线上跑的是一个不透明的加密令牌。每个令牌被绑在四样东西上：

绑定项 效果
时间窗 每轮重签，ttl 默认 600 秒——管的是单轮思考时间，不是整个流程
认证主体 请求带 OAuth 令牌时，state 绑到它的 client / issuer / subject；甲用户的状态在乙用户手里验不过
原始请求 绑方法、工具名（或 prompt 名 / 资源 URI）、参数摘要——换个工具重放，失败
问过的那句话 每个答案钉在客户端当时看到的那段问题原文上；你改了文案重新部署，Server 会重新问而不是消费旧答案

第 4 条有个反直觉的副作用，值得单独警告：

问题的文案要从工具参数推，别从「每次都不一样的数据」拼。 比如消息里带时间戳或者实时汇率，那么每一轮渲染出来都不同，客户端回带的历史答案全都「看起来过期」，Server 就一直重问，直到撞上客户端的轮数上限，整个调用失败。

然后是部署上真正的坑——默认密钥是进程级的：

1
2
3
4
from mcp.server.mcpserver import MCPServer, RequestStateSecurity

# 多实例、或者要扛住重启：给一组共享密钥（每个 ≥ 32 字节）
mcp = MCPServer("fleet", request_state_security=RequestStateSecurity(keys=[key]))
  • 不配置：只适合单进程——stdio，或者只有一个 HTTP worker。重试落到别的 worker、别的实例，或者同一台机器重启之后，都会验不过，客户端必须从头再来一遍；
  • keys=[...]：只要重试可能落到不同实例（多 worker 的 uvicorn、负载均衡后面的多副本），或者需要扛住重启，就必须配。一台签的，任何兄弟实例都能验；
  • 用自己的加密体系（KMS、现成 token 服务）就传 RequestStateSecurity(codec=...) 代替 keys。

轮换密钥别搞反顺序，官方给了三步，关键是「先让大家都会验，再换签发的那个」：

1
2
3
RequestStateSecurity(keys=[OLD, NEW])  # 第一步：所有实例学会验 NEW，还用 OLD 签
RequestStateSecurity(keys=[NEW, OLD])  # 第二步：NEW 开始签，在途的 OLD 还能验
RequestStateSecurity(keys=[NEW])       # 第三步：等第二步全量铺开一个 ttl 之后，退休 OLD

最后两条：

  • 你自己手搓的 request_state（工具/提示词/资源模板直接返回 InputRequiredResult 那种）同样被密封，零改动。但 SDK 没法替你钉「这是哪个问题」——你要是按问题键存答案，就把自己的问题标识一起塞进 state，重试时自己核；
  • 低层 Server 默认什么都不封，request_state 怎么写出去就怎么过线，除非你自己把边界加上去。

六、subscriptions/listen：通知从「随手推」变成「订阅制」

说完「Server 要东西」，说第二面：Server 有变化，怎么告诉客户端。

老办法是两套混着用：resources/subscribe 加上一条独立的 HTTP GET 流。问题跟前面一样——长连接、要会话、断线还能重放（Last-Event-ID），整套东西在无状态世界里全是债。

2026-07-28 换成了 subscriptions/listen（SEP-2575）：客户端发一个请求，这个请求的响应就是那条流。 它一直开着，推客户端点名要的那几类变更，直到客户端取消。

客户端要什么，写在 filter 里：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "subscriptions/listen",
  "params": {
    "_meta": {
      "io.modelcontextprotocol/protocolVersion": "2026-07-28",
      "io.modelcontextprotocol/clientInfo": { "name": "ExampleClient", "version": "1.0.0" },
      "io.modelcontextprotocol/clientCapabilities": {}
    },
    "notifications": {
      "toolsListChanged": true,
      "resourceSubscriptions": ["board://sprint"]
    }
  }
}

四个字段，都能省略，省略就等于不订阅那一类：

字段 收到什么
toolsListChanged notifications/tools/list_changed
promptsListChanged notifications/prompts/list_changed
resourcesListChanged notifications/resources/list_changed
resourceSubscriptions 这些 URI 的 notifications/resources/updated

三条规矩记牢（这就是「订阅制」跟「随手推」的区别）：

  1. 第一帧必须是确认：Server 必须先发 notifications/subscriptions/acknowledged，带上这条订阅的 id（_meta 里的 io.modelcontextprotocol/subscriptionId，值就是那个 subscriptions/listen 请求的 JSON-RPC id），在它之前一个通知都不许发；
  2. 不许超发：Server 不能发送客户端没有明确订阅的类型——你在确认帧里看到的 filter，就是服务器答应给你的子集（它不支持的类型会被省略，客户端要能优雅处理）；
  3. 每帧都带订阅 id：stdio 上所有消息挤一根管子，客户端必须靠这个字段区分是哪个订阅推的。一个客户端可以同时开多条订阅，各自 demux。

客户端侧在 Python SDK 里就是下面这个样子，我觉得这是整个协议里最好看的一段代码：

1
2
3
4
5
6
7
8
9
10
11
async def follow_board(client: Client) -> None:
    async with client.listen(tools_list_changed=True, resource_subscriptions=[BOARD]) as sub:
        async for event in sub:
            match event:
                case ResourceUpdated(uri=uri):
                    print(await read_board(client, uri))
                case ToolsListChanged():
                    tools = await client.list_tools()
                    print("tools:", [t.name for t in tools.tools])
                case _:
                    pass  # 没订阅的类型根本不会到这儿

四个重点：

  • async with client.listen(...) 进入时就把请求发出、并且等到了确认——所以代码块一开始，流就是活的，从那之后的变化一个都不会漏；
  • 事件是「指示你重新拉取」，不是负载。 ResourceUpdated 只给一个 uri，内容还得你自己 read_resource。事件说「什么变了」，从来不说「变成什么了」；
  • 没有重放：流建立之前发布的变更，就是不补给你。所以「先进订阅、再读一次快照」的顺序很重要——那个快照不可能漏事件；
  • sub.honored 是服务器实际答应的 filter（客户端可以核对），sub.subscription_id 是这条流的 id，每条帧上都盖着它。

流怎么结束，也是普通控制流，别当异常处理：

1
2
3
4
5
6
7
8
9
while True:
    try:
        async with client.listen(resource_subscriptions=["board://sprint"]) as sub:
            print(await read_board(client))     # 重订阅后必须重新拉，没有重放
            async for _event in sub:
                print(await read_board(client))
    except SubscriptionLost:
        pass          # 掉线了：退避一下再订阅，重拉
    await anyio.sleep(backoff)
  • Server 优雅关闭 → async for 自然结束；
  • 异常断开 → 抛 SubscriptionLost；
  • 两种都是「流没了、没重放」，处理办法一样：退避、重新订阅、重新拉取；
  • 进块时还可能抛三种，要分开对待：MCPError（连接挂了，或者服务器根本不提供这个方法）、TimeoutError（确认没等到）、ListenNotSupportedError（对方是 2026-07-28 之前的连接）——最后这个永远不会自己好，别往重试里塞。

七、Server 侧落地：发布变更 + 横向扩容

站在 Server 这一边，写起来比客户端还省事——你只管宣布「什么变了」，SDK 负责盖上订阅 id、按 filter 过滤、管生命周期：

1
2
3
4
5
@mcp.tool()
async def complete_task(board: str, task: str, ctx: Context) -> str:
    BOARDS[board][task] = True
    await ctx.notify_resource_updated(f"board://{board}")   # 订阅了这条 URI 的流都收到
    return f"{task}: done"

兄弟方法还有三个：notify_tools_changed()、notify_prompts_changed()、notify_resources_changed()。

两个好消息：

  • 没有听众就是个空操作。你永远不用先检查「有人在听吗」，直接宣布就行；
  • 跟上一课对上了：想要老客户端也收到「工具列表变了」，得另发一次 ctx.session.send_prompt_list_changed() 那套老接口。新老双发，互不打扰。

不过有个细节得留神：MCPServer 对资源 URI 是精确字符串匹配。订阅了 board://sprint 的流，听不到 board://sprint/tasks/1 的变化——规范其实允许服务器上报「子资源变了」，MCPServer 不做，但客户端是照着收到的来写的，所以你的客户端代码别硬猜是哪个 URI 动了，读事件里的那个。

横向扩容也是这版协议最有性价比的一块：整条流的状态只在单个进程里，多副本要跨进程广播，就实现一个 SubscriptionBus——两个方法，接 Redis 就行：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from collections.abc import Callable
from redis.asyncio import Redis
from mcp.server.mcpserver import MCPServer
from mcp.server.subscriptions import ServerEvent

class RedisSubscriptionBus:
    def __init__(self, redis: Redis) -> None:
        self._redis = redis
        self._listeners: dict[object, Callable[[ServerEvent], None]] = {}

    async def publish(self, event: ServerEvent) -> None:
        await self._redis.publish("mcp-events", encode(event))   # encode 你自己写

    def subscribe(self, listener: Callable[[ServerEvent], None]) -> Callable[[], None]:
        token = object()
        self._listeners[token] = listener
        def unsubscribe() -> None:
            self._listeners.pop(token, None)
        return unsubscribe

mcp = MCPServer("Sprint Board", subscriptions=RedisSubscriptionBus(redis))

三条实现注意：监听器是同步的、不许抛异常、跑在服务器的事件循环上。另外每个副本还得有个后台任务，把收到的消息解码后喂给所有注册的监听器。

低层的 Server 就是自己拼零件了：InMemorySubscriptionBus() + ListenHandler(bus) + on_subscriptions_listen= 槽位。这时候规范义务就转移到你身上——先确认、每帧盖 id、不超发。还有个别忘的收尾：ListenHandler.close() 会优雅结束每一条流，每条的最后一帧就是那个 listen 请求的结果（等于告诉客户端「是我主动收的，不是你网断了」），它返回时流还没刷完，拆传输之前给它一点时间。

不需要变更通知就别开：MCPServer(subscriptions=False) 会直接声明「我没有通知能力」，也不持有任何流——干净。

最后是订阅这条路上的授权，写法跟普通资源鉴权不是一回事。要在确认之前就把请求卡住，所以得用中间件看 subscriptions/listen 的原始请求：

1
2
3
4
5
6
7
8
async def gate_subscriptions(ctx: ServerRequestContext, call_next: CallNext) -> HandlerResult:
    if ctx.method == "subscriptions/listen":
        params = SubscriptionsListenRequestParams.model_validate(ctx.params or {}, by_name=False)
        token = get_access_token()
        user = token.subject if token else None
        if not all(can_access(user, uri) for uri in params.notifications.resource_subscriptions or ()):
            raise MCPError(INVALID_REQUEST, "not permitted to watch the requested resources")
    return await call_next(ctx)

三个要点：

  • 拒绝要用统一的、不提具体 URI 的话术——否则「拒绝」本身就泄漏了「哪些 URI 是受保护的」；
  • 同一个 can_access(user, uri) 同时服务两处：资源处理器在 resources/read 上问它，中间件在 subscriptions/listen 上问它。换成数据库或你的 RBAC，两边一起同步；
  • 这个决定管整条流的命。规范里没有逐事件复查，所以如果调用方的权限会在流中途失效（比如令牌要过期），你必须在过期时断开那条连接。

八、收尾上线：把前三课的 Server 变成能长期挂着的服务

好，到这里协议给的新工具都用上了。最后这一节解决上一篇文末那句话——「限流、可观测、灰度回滚」，因为无状态化把部署模型变简单的同时，也把过去搭便车的东西一次性抽掉了。以前一个 session id 就能串起一串调用、就能做限流、就能存凭据；现在这些都得你自己显式做。

按顺序来。

1）网关层：三个头，先让路由不解析 body

2026-07-28 的 Streamable HTTP 请求必须带这几个头，网关、WAF、限流器可以照着路由和计费：

头 内容
MCP-Protocol-Version 协议版本
Mcp-Method JSON-RPC 方法名，比如 tools/call
Mcp-Name 具体的工具名 / 资源 URI / 提示词名（tools/call、resources/read、prompts/get 必带）

但后端必须校验头与请求体一致，不一致就拒绝——否则网关按一个方法做的策略，落到后端执行的是另一个方法，等于门形同虚设。

2）列表缓存：ttlMs / cacheScope

tools/list、prompts/list、resources/list 的结果现在带缓存提示：能缓存多久、能不能跨用户共享（配合确定性排序，缓存命中率会好看很多）。工具目录不用每次重连都拉一遍。

3）限流：键怎么挑，计数器放哪，超限说什么

  • 键：单用 IP 最不靠谱——Agent 多半跑在云上，一堆用户共享出口 IP。稳的做法是以调用方身份为主键（令牌里的 subject 或 client id），再按工具名分一层：同一个用户查一次目录和跑一次全表扫描，配额显然不该同价；
  • 计数器：无状态之后本地计数等于没数——请求会撒到任意实例上，各算各的。老老实实放 Redis 之类的集中缓存，用滑动窗口计数、漏桶这些近似算法就够了，别为了精确度把每个请求变成一次强一致写；
  • 超限返回什么：别只丢一个错误码。调用方是模型驱动的客户端，它需要知道「该等多久」，所以响应里给出明确的重试等待时间，并在错误文本里写清是哪一类配额被打满了。否则 Agent 会立刻重试，把你从限流打成雪崩；
  • 还有一个容易漏的点：区分「单次工具调用」和「一次任务」。长任务走 Tasks 扩展之后可能是一次提交、多次轮询状态——如果限流只数请求数，轮询会把配额吃光。把轮询类归一类，给更宽松的配额，或者干脆按任务数计。

4）可观测：关联 ID 得自己造了

无状态是双刃剑：好处是每个请求自包含，日志不用跨实例拼；坏处是天然的那个关联键没了。补法：

  1. 每个请求打一个关联 ID：客户端在 _meta 里带了的就沿用，没带就在入口生成，并在响应里回带。日志、指标、追踪三处都带上它；
  2. 结构化日志，字段先定死：关联 ID、调用方身份、工具名、入参摘要（脱敏）、耗时、结果状态、错误分类；
  3. 入参不要原样落盘：工具参数里经常混着真实业务数据，直接全量写既是隐私问题也是存储问题。只记结构和长度，敏感字段哈希或截断；
  4. 指标按工具名拆维度：整体 QPS 和 P99 看不出问题，一个 Server 里几十个工具，慢的永远就那两三个；
  5. 协议错误和业务错误分开统计：参数校验失败、鉴权失败、工具内部异常、下游超时——四类处置方式完全不同，混在一个「错误率」里等于没监控；
  6. 留一个兼容性观测口：记录每个请求声明的协议版本。你总得知道旧客户端还剩多少，才敢下兼容层（下面第 6 条要用）；
  7. 追踪方面，规范已经把 OpenTelemetry 的 trace context 键名统一到 _meta 里了，一次调用可以从宿主应用 → MCP 客户端 → MCP 服务 → 下游 API 串成一棵树。

5）幂等：无状态最大的副作用是「重试可能跑两次」

网络一抖，客户端就会重发同一个 POST。能重试是好事，但副作用工具（下单、改库存、扣款）可能已经执行过一次了。两条最低成本的做法：

  • 客户端给带副作用的调用带幂等键，服务端对同一个键只执行一次；
  • 工具本身设计成天然幂等（同样的入参结果一致），副作用操作先查后写。

6）灰度与回滚：双端点并行，别一刀切

旧客户端（SSE 那条路）连新服务会失败——事件流入口都没了，只剩一个 POST 端点。生产的做法是两套端点并行、按客户端分流：

1
2
location /mcp/legacy { proxy_pass http://legacy-sse; }   # 老客户端
location /mcp        { proxy_pass http://stateless;  }   # 新客户端

先把无状态端点铺上去，盯错误率、重试率、MRTR 完成率；旧会话自然排空之后再摘掉 legacy。回滚方案（切回 legacy 端点）要随时可执行。

废弃的时间表也记一下，好排期：roots / sampling / 日志这三样，最早也要 2027-07-28 当日或之后发布的第一个规范版本才会移除；HTTP+SSE 是「SEP-2596 转 Final 之后 3 个月」。窗口看着宽裕，但如果你对外提供服务，实际能用的时间要打折——你控制不了别人什么时候升级客户端。先看观测里旧版本的占比和身份，再定兼容层维持多久，比拍脑袋定日期稳妥得多。

顺便说一句正经的未来方向：sampling 废弃之后，Server 需要模型能力就直接调模型提供商的 API。这一步我建议一开始就把出口设计成「一个 base_url + 一个 key」的 OpenAI 兼容形态——我自己是把多家模型挂在 ai.aklibk.com 这个中转上，一把钥匙接 DeepSeek、通义、Kimi 这些国产模型，按量付费、价格本来就便宜；要接 Claude / GPT / Gemini 那些国外模型也一样走这个口子，国内直连、免绑卡、人民币按量付费，切换模型只改一个 model 名，Server 代码不用动。

7）客户端兼容的现实：别把「协议支持」当成「客户端支持」

这一课我核客户端支持时，翻到两个真实的坑，性质一样：协议写着的能力，客户端可能只声明一半，或者声明了却不干活。

  • Claude Code 在 2026-07-28 的握手里，elicitation 能力是空的 {}——按规范这意味着只支持表单模式。所以带 mode: "url" 的 URL 型 elicitation（OAuth 授权、支付、第三方登录那类跳转）永远拿不到，Server 辛辛苦苦走 MRTR 返回的链接，用户界面上根本不出现；
  • Claude 的 Cowork 桌面版更典型：它声明支持 elicitation，但收到 InputRequiredResult 之后走了「打印模式」分支——收到、解析、然后什么都不做。不渲染、不重试、不返回拒绝也不报错，那次 tools/call 就挂在那儿，一直到 180 秒的传输超时才报一个跟 elicitation 毫无关系的超时。

所以我的建议就两条：

  1. 写带 elicitation 的工具之前，先用真实客户端验一次，别拿 Inspector 过了就上线；
  2. 给「客户端不答」设计兜底：调用方超时或者返回 cancel 时，工具要能优雅退化成「只读、不落副作用」的行为，而不是把脏活干了一半。

怎么自己确认客户端到底声明了什么？两个办法：用 uv run mcp dev server.py 的 Inspector 看握手；或者在 Server 和客户端之间放个日志代理，抓 _meta 里的 io.modelcontextprotocol/clientCapabilities，看有没有 elicitation 字段、里面有没有 url。这一招比翻任何文档都准。

排障清单（这一课踩到的坑基本都在里面）：

  1. 工具报 -32603：你在 2026-07-28 连接上返回了 InputRequiredResult，但客户端是 mode="legacy" 或者协议版本太老。用 client.protocol_version 确认协商到什么，再决定用哪种写法；
  2. 客户端不弹表单，直接报超时：先看客户端有没有声明 elicitation（上面第 7 条），再看它是不是在非交互分支里把请求丢了；
  3. URL 型 elicitation 拿不到链接：客户端只声明了表单模式（elicitation: {}）。要么改用表单模式收验证码/短码，要么这条流程走应用自己的跳转，别指望协议；
  4. 校验报 InvalidSignature：给工具参数声明了 Resolve(...)，同时又想从函数体返回 InputRequiredResult。两套写法不能混，二选一；
  5. 跨实例重试失败：request_state 的默认密钥是进程级的。多副本或者滚动重启，必须配 RequestStateSecurity(keys=[...])；
  6. 密文轮换期间丢在途请求：密钥顺序搞反了（先换了签发的那个）。三步走：先让所有实例会验新键，再换 minter，最后退休旧键；
  7. 同一个问题被反复问：问题文案里拼了时间戳、实时价这类「每轮都不一样」的数据，导致记录下的答案永远「过期」；
  8. 客户端提示 ListenNotSupportedError：对面是 2026-07-28 之前的连接，这个方法就不存在，重试无门，只能降级成轮询；
  9. 订阅流看着开了却没有通知：先核对 sub.honored——你请求的类型服务器不一定全答应；再核对 URI，MCPServer 是精确字符串匹配；
  10. 重连后少了中间那次变更：subscriptions/listen 没有重放。正确姿势是「先订阅、再拉一次快照」，而不是先拉快照再订阅；
  11. 限流好像不生效：计数器还在实例本地内存里。无状态之后请求会落到任意实例，计数必须外置；
  12. 切换部署后旧客户端全挂：只保留了一个 /mcp 端点。灰度期必须双端点分流，等观测确认旧流量归零再摘。

一句话收口：MRTR 让 Server 在无状态世界里还能问问题，subscriptions/listen 让它还能说变化，而无状态换来的部署自由，代价是限流、凭据、可观测这些「原来搭便车」的活都得自己重做一遍。 三样齐了，你那个 Server 才算真的能上线挂着。

生命不息，折腾不止。下一篇「MCP 实战第五课」，我们把这条线上最后两块硬骨头啃掉：Tasks 扩展（长任务从核心协议搬进扩展之后，异步提交、状态轮询、结果回收到底怎么写）和 OAuth 授权加固（RFC 9207 的 iss 校验、DCR 到 CIMD 的迁移路径，多用户、多租户怎么收口），最后用一个真实场景把前四课的东西串成一套能交付给别人的完整服务。