5.3 流式处理和实时通信

🎯 学习目标：搞清楚 MCP 里"流"到底是什么、有哪几种、各自的边界在哪，并能正确实现进度上报、日志流、请求取消与变更订阅
⏱️ 预计时间：45分钟
📊 难度等级：⭐⭐⭐⭐

📌 内容对齐 MCP 官方规范 2026-07-28。 MCP 没有自己的流式框架——它把流的能力交给了传输层（Streamable HTTP 的 SSE），协议层只定义"什么时候允许发什么通知"。所以本节不讲任何自创的"流管理器"，只讲规范里真实存在的东西。

🌊 先厘清：MCP 里的"流"有两种，别混

这是最容易搞混的地方。MCP 的响应流有两种完全不同的用途，生命周期、作用域、承载的通知都不同：

	请求作用域流	订阅流
由谁开启	客户端发任意业务请求	客户端发 `subscriptions/listen`
活多久	该请求处理完就关	长期存活，直到取消或断开
承载什么	`notifications/progress`、`notifications/message`（只属于这个请求）	`notifications/*/list_changed`、`notifications/resources/updated`
怎么区分	就是该请求的响应流	`_meta` 里的 `io.modelcontextprotocol/subscriptionId`
规范依据	Progress、Logging	Subscriptions

⚠️ 规范明确定义了边界：请求作用域的通知（notifications/progress、notifications/message）必须走它们所关联请求的响应流，不得走 subscriptions/listen 的流。

🏗️ 传输层怎么承载流

流的能力来自传输。两种标准传输的差异很关键：

📟 stdio：一条通道，全靠 ID 分流

stdio 只有一条 stdin/stdout 通道。所有请求的响应、所有通知、所有订阅消息都混在同一条通道上。

这意味着：

客户端必须用 id 匹配响应
订阅通知必须用 _meta.io.modelcontextprotocol/subscriptionId 区分
任何一条消息写错格式，整条通道都会乱——所以日志绝不能写 stdout

🌊 Streamable HTTP：每个请求自带一条流

这是它比旧版 HTTP+SSE 优雅的地方。客户端一条 POST /mcp，服务器可以：

回一个普通 JSON 对象（快操作）
回一条 SSE 流（需要持续推送过程信息，或最后才出结果）

客户端用 Accept: application/json, text/event-stream 声明两种都接受。

POST /mcp HTTP/1.1
Accept: application/json, text/event-stream
MCP-Protocol-Version: 2026-07-28
Mcp-Method: tools/call
Mcp-Name: analyze_dataset

← HTTP/1.1 200 OK
← Content-Type: text/event-stream
←
← event: message
← data: {"jsonrpc":"2.0","method":"notifications/progress","params":{"progressToken":"tok-1","progress":10,"total":100}}
←
← event: message
← data: {"jsonrpc":"2.0","method":"notifications/progress","params":{"progressToken":"tok-1","progress":60,"total":100}}
←
← event: message
← data: {"jsonrpc":"2.0","id":7,"result":{"resultType":"complete","content":[{"type":"text","text":"分析完成"}]}}

流在最后一条响应消息发出后关闭。

📌 没有 Last-Event-ID，没有消息重投。 2026-07-28 明确移除了 SSE 流的可恢复性与重投。流断了，在途请求就丢了——客户端必须用新的请求 ID 重新发起，不能"续传"。

⏳ 进度通知：让长任务可见

客户端希望收到进度，就在请求的 _meta 里放一个 progressToken：

json

{
  "jsonrpc": "2.0",
  "id": 7,
  "method": "tools/call",
  "params": {
    "name": "analyze_dataset",
    "arguments": { "dataset": "sales_2023.csv" },
    "_meta": {
      "io.modelcontextprotocol/protocolVersion": "2026-07-28",
      "io.modelcontextprotocol/clientCapabilities": {},
      "progressToken": "tok-1"
    }
  }
}

服务器随后可以发进度通知：

json

{
  "jsonrpc": "2.0",
  "method": "notifications/progress",
  "params": {
    "progressToken": "tok-1",
    "progress": 50,
    "total": 100,
    "message": "已处理 25000 行 / 50000 行"
  }
}

规范层面的硬性要求：

要求	级别	说明
`progressToken` 必须是字符串或整数	MUST	客户端自选，但在所有活跃请求间必须唯一
`progress` 值必须递增	MUST	即使总量未知也要递增
`progress` / `total` 可以是浮点数	MAY	不要求整数
`message` 应提供人类可读信息	SHOULD	用于展示给用户
只能引用仍然活跃的 token	MUST	请求结束后不得再发

🧩 Node.js：服务器侧上报进度

javascript

// 进度上下文绑定在「这一次请求处理」上，不是连接上 —— 这是无状态模型的必然结果
async function handleToolCall(params, requestContext) {
    const progressToken = params._meta?.progressToken;

    // 没有 token 就说明客户端不想要进度，别发
    const reportProgress = progressToken
        ? async (progress, total, message) => {
              await requestContext.sendNotification({
                  jsonrpc: '2.0',
                  method: 'notifications/progress',
                  params: { progressToken, progress, total, message }
              });
          }
        : null;

    const rows = await loadRows(params.arguments.dataset);
    let processed = 0;
    let lastProgress = -1;

    for await (const row of rows) {
        await processRow(row);
        processed += 1;

        // 节流 + 保证严格递增
        if (reportProgress && processed % 1000 === 0) {
            const pct = Math.floor((processed / rows.length) * 100);
            if (pct > lastProgress) {
                lastProgress = pct;
                await reportProgress(processed, rows.length, `已处理 ${processed} 行`);
            }
        }
    }

    return {
        resultType: 'complete',
        content: [{ type: 'text', text: `分析完成，共处理 ${processed} 行` }],
        isError: false
    };
}

⚠️ 别过度上报。每次通知都是一条独立消息，发太密会拖慢整体吞吐。常见做法是按百分比变化或固定条数间隔节流。

🐍 Python：客户端侧消费进度

python

import json, httpx

async def call_with_progress(endpoint: str, tool: str, args: dict, on_progress):
    """发起可能返回 SSE 流的工具调用，并把进度回调出去。"""
    payload = {
        "jsonrpc": "2.0",
        "id": 7,
        "method": "tools/call",
        "params": {
            "name": tool,
            "arguments": args,
            "_meta": {
                "io.modelcontextprotocol/protocolVersion": "2026-07-28",
                "io.modelcontextprotocol/clientInfo": {"name": "my-client", "version": "1.0.0"},
                "io.modelcontextprotocol/clientCapabilities": {},
                # 打开进度上报的开关
                "progressToken": "tok-1",
            },
        },
    }
    headers = {
        "Content-Type": "application/json",
        "Accept": "application/json, text/event-stream",
        "MCP-Protocol-Version": "2026-07-28",
        "Mcp-Method": "tools/call",
        "Mcp-Name": tool,
    }

    async with httpx.AsyncClient(timeout=None) as client:
        async with client.stream("POST", endpoint, json=payload, headers=headers) as resp:
            resp.raise_for_status()

            if "text/event-stream" not in resp.headers.get("content-type", ""):
                # 普通 JSON 响应：直接就是结果
                return json.loads(await resp.aread()).get("result")

            final_result = None
            async for line in resp.aiter_lines():
                if not line.startswith("data:"):
                    continue
                msg = json.loads(line[5:].strip())

                if msg.get("method") == "notifications/progress":
                    p = msg["params"]
                    await on_progress(p["progress"], p.get("total"), p.get("message"))
                elif "result" in msg or "error" in msg:
                    final_result = msg.get("result")
                    if msg.get("error"):
                        raise RuntimeError(msg["error"]["message"])

            # 流结束后 result 已拿到；resultType 缺失时按 complete 处理
            return final_result

📝 请求级日志：notifications/message

旧的 logging/setLevel 方法已删除。现在日志级别是每个请求自己声明的，通过 _meta.io.modelcontextprotocol/logLevel：

json

{
  "jsonrpc": "2.0",
  "id": 9,
  "method": "tools/call",
  "params": {
    "name": "deploy_service",
    "arguments": { "service": "api" },
    "_meta": {
      "io.modelcontextprotocol/protocolVersion": "2026-07-28",
      "io.modelcontextprotocol/clientCapabilities": {},
      "io.modelcontextprotocol/logLevel": "debug"
    }
  }
}

服务器随之可以在这条请求的响应流里发日志通知：

json

{
  "jsonrpc": "2.0",
  "method": "notifications/message",
  "params": {
    "level": "info",
    "logger": "deploy-tool",
    "data": { "message": "正在拉取镜像", "service": "api", "tag": "v1.4.2" }
  }
}

❗ 一条硬性禁令：服务器不得为没有携带 logLevel 字段的请求发送 notifications/message。这个设计让日志噪音完全由客户端控制——不想看日志就什么都不用做。

💡 官方对 Logging 的整体态度：Logging 能力在 2026-07-28 已被标记废弃。官方建议的长期方案是：stdio 上直接写 stderr，或者引入 OpenTelemetry。新项目不要把 Logging 当作主要可观测性手段。

🛑 取消：两种传输两种做法

取消的语义由传输定义，这一点很关键：

传输	取消方式	服务器必须做什么
Streamable HTTP	客户端关闭该请求的响应流	必须把客户端断开视为该请求被取消；不需要也不期望收到取消通知
stdio	客户端发 `notifications/cancelled`	收到后终止对应请求的处理

stdio 上的取消通知长这样：

json

{
  "jsonrpc": "2.0",
  "method": "notifications/cancelled",
  "params": {
    "requestId": "123",
    "reason": "用户取消了操作"
  }
}

❗ 服务器侧有一条严格限制：服务器只有一种情况可以发 notifications/cancelled——拆掉某条 subscriptions/listen 订阅流时，用 requestId 指向那条订阅请求。其他任何场景都不得发送。

🧩 Node.js：实现可取消的长任务

javascript

// 每个在途请求维护一个 AbortController，取消时触发
const inFlight = new Map();   // requestId -> AbortController

async function handleRequest(id, method, params, requestContext) {
    const controller = new AbortController();
    inFlight.set(id, controller);

    // Streamable HTTP：客户端关流时 requestContext 会感知断开
    requestContext.onDisconnect(() => controller.abort());

    // stdio：监听取消通知
    requestContext.onCancellation((cancelledId, reason) => {
        if (cancelledId !== id) return;
        console.error(`请求 ${id} 被取消：${reason}`);
        controller.abort();
    });

    try {
        return await executeWithSignal(method, params, controller.signal);
    } catch (err) {
        if (err.name === 'AbortError') {
            // 取消不是错误，别返回 error 响应——流已经没了或调用方已放弃
            return null;
        }
        throw err;
    } finally {
        inFlight.delete(id);
    }
}

📌 超时也要设。规范建议：所有发出的请求都应该设超时，超时后应该取消该请求并停止等待，避免连接挂死和资源耗尽。

📡 订阅流：subscriptions/listen

这是 2026-07-28 中最"流式"的部分，也取代了旧版的 HTTP GET 端点、resources/subscribe、resources/unsubscribe。

开启订阅

json

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "subscriptions/listen",
  "params": {
    "notifications": {
      "toolsListChanged": true,
      "promptsListChanged": true,
      "resourcesListChanged": true,
      "resourceSubscriptions": ["file:///project/config.json"]
    },
    "_meta": {
      "io.modelcontextprotocol/protocolVersion": "2026-07-28",
      "io.modelcontextprotocol/clientInfo": { "name": "my-client", "version": "1.0.0" },
      "io.modelcontextprotocol/clientCapabilities": {}
    }
  }
}

过滤器字段：

字段	类型	收到什么
`toolsListChanged`	`boolean`	`notifications/tools/list_changed`
`promptsListChanged`	`boolean`	`notifications/prompts/list_changed`
`resourcesListChanged`	`boolean`	`notifications/resources/list_changed`
`resourceSubscriptions`	`string[]`	这些 URI 的 `notifications/resources/updated`

所有字段都是可选的。省略 = 不订阅。服务器不得发送客户端未显式请求的通知类型。

必须先回确认

流的第一条消息必须是确认，notifications 字段反映服务器实际同意的那部分：

json

{
  "jsonrpc": "2.0",
  "method": "notifications/subscriptions/acknowledged",
  "params": {
    "_meta": { "io.modelcontextprotocol/subscriptionId": 1 },
    "notifications": {
      "toolsListChanged": true,
      "resourceSubscriptions": ["file:///project/config.json"]
    }
  }
}

在确认消息之前，服务器不得在这条订阅上发任何通知。

💡 客户端要核对：把你请求的过滤器和服务器确认的过滤器比一比，被拒的类型要优雅降级，别默默等着永不来的通知。

通知怎么关联回订阅

subscriptionId 就是当初那条 subscriptions/listen 请求的 JSON-RPC id。每条通知都带它：

json

{
  "jsonrpc": "2.0",
  "method": "notifications/resources/updated",
  "params": {
    "_meta": { "io.modelcontextprotocol/subscriptionId": 1 },
    "uri": "file:///project/config.json"
  }
}

在 stdio 上这一点尤其重要——所有消息共用一条通道，客户端必须靠 subscriptionId 才能把通知分派到正确的订阅。

并行订阅与取消

客户端可以同时开多条订阅（比如一条听工具变更、一条听资源更新），靠各自的 subscriptionId 分流。

订阅在三种情况下结束：

客户端取消：HTTP 上关掉 SSE 流；stdio 上发 notifications/cancelled 指向订阅请求的 ID
服务器主动关闭：应该先返回一个完整结果表示优雅结束，再关流
底层传输断开：HTTP 超时、TCP 断连、stdio 子进程退出

服务器优雅关闭时，会回应最初那条 subscriptions/listen 请求：

json

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "resultType": "complete",
    "_meta": { "io.modelcontextprotocol/subscriptionId": 1 }
  }
}

收到这个响应 = 订阅正常关闭；没收到就断流 = 异常断开，客户端可以据此决定是否重连。

❗ stdio 重连后必须重发 subscriptions/listen。服务器不跨连接保存任何订阅状态——这是无状态设计的直接推论。

🐍 Python：完整的订阅消费端

python

import asyncio, json, httpx

async def listen_subscriptions(endpoint: str, filters: dict, on_notification):
    """开启一条订阅流并持续消费；正常返回表示优雅关闭，抛错表示异常断开。"""
    payload = {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "subscriptions/listen",
        "params": {
            "notifications": filters,
            "_meta": {
                "io.modelcontextprotocol/protocolVersion": "2026-07-28",
                "io.modelcontextprotocol/clientCapabilities": {},
            },
        },
    }
    headers = {
        "Content-Type": "application/json",
        "Accept": "text/event-stream",
        "MCP-Protocol-Version": "2026-07-28",
        "Mcp-Method": "subscriptions/listen",
    }

    acknowledged = False

    async with httpx.AsyncClient(timeout=None) as client:
        async with client.stream("POST", endpoint, json=payload, headers=headers) as resp:
            resp.raise_for_status()
            async for line in resp.aiter_lines():
                if not line.startswith("data:"):
                    continue
                msg = json.loads(line[5:].strip())

                # 1) 首条必须是确认
                if msg.get("method") == "notifications/subscriptions/acknowledged":
                    granted = msg["params"].get("notifications", {})
                    rejected = set(filters) - set(granted)
                    if rejected:
                        print(f"服务器拒绝了这些订阅类型，需降级处理: {rejected}")
                    acknowledged = True
                    continue

                if not acknowledged:
                    raise RuntimeError("协议违规：确认消息之前就收到了通知")

                # 2) 订阅正常关闭
                if msg.get("id") == 1 and "result" in msg:
                    return

                # 3) 普通通知
                if "method" in msg:
                    sub_id = msg["params"].get("_meta", {}).get(
                        "io.modelcontextprotocol/subscriptionId"
                    )
                    await on_notification(sub_id, msg["method"], msg["params"])

    # 流结束但没收到关闭响应 => 异常断开
    raise ConnectionError("订阅流意外断开，需要重新发起 subscriptions/listen")

🔁 断流与重连的正确姿势

因为可恢复性被移除了，重连策略必须自己设计。推荐做法：

要点：

重发必须用新的请求 ID——规范明确要求，旧 ID 不得复用
不要静默重试非幂等操作（扣款、部署、发消息）。先通过查询类工具确认实际状态
订阅流断开：重发 subscriptions/listen，并重新核对确认结果（服务器可能已变化）
服务器侧应该把状态放在显式句柄里（作为普通工具参数传递），这样客户端重发时能带上，不依赖任何连接状态

💡 这是无状态设计带来的直接好处：因为服务器不依赖连接状态，重发请求天然是安全的——只要操作本身幂等。

🧰 缓存提示：减少不必要的往返

流的反面是"根本不用流"。很多数据变化很慢，让客户端缓存就够了。规范的 CacheableResult 接口要求下列结果必须带缓存提示：

server/discover、tools/list、prompts/list、resources/list、resources/templates/list、resources/read

字段	类型	含义
`ttlMs`	整数	客户端可以把这个结果视为新鲜多久（毫秒）
`cacheScope`	`"public"` / `"private"`	共享中间层能否缓存

json

{
  "resultType": "complete",
  "tools": [],
  "ttlMs": 300000,
  "cacheScope": "public"
}

缓存与变更通知是互补关系，不是二选一：

用 ttlMs 减少轮询（省往返）
用 subscriptions/listen 在数据真的变了时立刻知道（省延迟）

两条规则别踩：

缓存键 = 方法 + 影响结果的参数（resources/read 的 uri、分页列表的 cursor）。参数不同不得复用缓存
通过 MRTR 重试产生的请求（带 inputResponses 或 requestState）不得缓存——它们依赖不在缓存键里的输入

📌 服务器侧建议：tools/list 返回确定顺序（底层工具集没变时顺序保持一致）。这样客户端能可靠缓存，LLM 侧的工具定义提示缓存命中率也会更高。

⚠️ 常见误区

常见说法	实际情况（2026-07-28）
"MCP 有 SSE、WebSocket、HTTP Streaming、STDIO Streaming 四种流式方案"	❌ 流的能力来自传输层。标准传输只有 stdio 与 Streamable HTTP，没有 WebSocket
"用 `Stream Manager` / `Buffer Manager` 做背压控制"	❌ MCP 协议里不存在这些概念，是早期文章的臆造
"用 `notifications/tools/stream` 推送流式内容"	❌ 不存在这个方法
"用 `tools/progress` 上报进度"	❌ 正确方法是 `notifications/progress`，通过 `progressToken` 关联
"用 `logging/setLevel` 设置日志级别"	❌ 已删除。改为请求级 `_meta.io.modelcontextprotocol/logLevel`
"用 `resources/subscribe` 订阅资源变更"	❌ 改为 `subscriptions/listen` + `resourceSubscriptions` 过滤器
"用 `Last-Event-ID` 续传断掉的流"	❌ 可恢复性与重投已移除，必须用新请求 ID 重发
"服务器随时可以主动发通知"	❌ 请求作用域通知只能走该请求的响应流；订阅通知只能走订阅流且必须是客户端订阅过的类型
"订阅一开就能收通知"	❌ 必须先收到 `notifications/subscriptions/acknowledged`

🎯 本节小结

✅ 两种流，别混：请求作用域流（进度、日志）vs 订阅流（变更通知）；规范明令禁止互串
✅ 传输决定流的能力：stdio 一条通道靠 ID 分流；Streamable HTTP 每请求一条 SSE 流，关流即取消
✅ 进度：_meta.progressToken 开启 → notifications/progress 上报，progress 必须递增
✅ 日志：请求级 logLevel；没带 logLevel 的请求不得收到 notifications/message
✅ 取消：HTTP 关流；stdio 发 notifications/cancelled。服务器只有拆订阅流时才能发取消通知
✅ 订阅：subscriptions/listen + 过滤器 + 必须先确认 + subscriptionId 关联 + stdio 重连要重发
✅ 断流：没有续传，只能用新 ID 重发；非幂等操作先查状态
✅ 缓存：ttlMs + cacheScope 与变更通知互补，能省掉大量往返

🤔 实践思考

💭 如果你要做一个"导出 10 万行数据"的工具，应该用进度通知、分页、还是让客户端多次调用？各自代价是什么？
🔍 为什么规范要把"请求作用域通知"和"订阅通知"分得这么严格？如果不分，会出什么问题？
🛠️ 在 Streamable HTTP 上，服务器什么时候该回 JSON、什么时候该回 SSE 流？判断依据是什么？
📋 移除 Last-Event-ID 续传能力后，客户端要做"断点续传"式的长任务，该怎么设计？

👉 下一小节：5.4 性能监控

5.3 流式处理和实时通信 ​

🌊 先厘清：MCP 里的"流"有两种，别混 ​

🏗️ 传输层怎么承载流 ​

📟 stdio：一条通道，全靠 ID 分流 ​

🌊 Streamable HTTP：每个请求自带一条流 ​

⏳ 进度通知：让长任务可见 ​

🧩 Node.js：服务器侧上报进度 ​

🐍 Python：客户端侧消费进度 ​

📝 请求级日志：notifications/message ​

🛑 取消：两种传输两种做法 ​

🧩 Node.js：实现可取消的长任务 ​

📡 订阅流：subscriptions/listen ​

开启订阅 ​

必须先回确认 ​

通知怎么关联回订阅 ​

并行订阅与取消 ​

🐍 Python：完整的订阅消费端 ​

🔁 断流与重连的正确姿势 ​

🧰 缓存提示：减少不必要的往返 ​

⚠️ 常见误区 ​

🎯 本节小结 ​

🤔 实践思考 ​

5.3 流式处理和实时通信

🌊 先厘清：MCP 里的"流"有两种，别混

🏗️ 传输层怎么承载流

📟 stdio：一条通道，全靠 ID 分流

🌊 Streamable HTTP：每个请求自带一条流

⏳ 进度通知：让长任务可见

🧩 Node.js：服务器侧上报进度

🐍 Python：客户端侧消费进度

📝 请求级日志：notifications/message

🛑 取消：两种传输两种做法

🧩 Node.js：实现可取消的长任务

📡 订阅流：subscriptions/listen

开启订阅

必须先回确认

通知怎么关联回订阅

并行订阅与取消

🐍 Python：完整的订阅消费端

🔁 断流与重连的正确姿势

🧰 缓存提示：减少不必要的往返

⚠️ 常见误区

🎯 本节小结

🤔 实践思考