跳到主要内容

MCP、SSE 与 WebSocket：分层原理与协作关系

在 AI 应用开发中，你常会同时遇到 MCP（模型上下文协议）、SSE（服务端推送事件） 和 WebSocket。它们分别解决不同层次的问题：MCP 描述「主机与工具/资源之间如何对话」；SSE 与 WebSocket 描述「字节如何在网络上持续流动」。本文从协议栈、时序与示例代码三条线，说明各自原理、典型组合与工程落地要点。

目录​

1. 为什么要区分「应用协议」和「传输方式」​

  • 传输层关心的是：连接如何建立、数据是否持续、方向是单向还是双向、是否复用 HTTP 基础设施（代理、负载均衡、HTTPS）。
  • 应用层关心的是：消息语义是什么、一次「调用」包含哪些字段、错误如何表示、能力如何发现（例如有哪些工具）。

把概念混在一谈时，容易出现「MCP 是不是就是 WebSocket」这类问题。更准确的说法是：MCP 定义对话内容与过程；SSE 或 WebSocket（以及 stdio、普通 HTTP 等）只是承载这些字节的多种管道之一。

2. 一张图看懂协议栈​

下面用「自顶向下」方式把常见概念摆到同一幅画里：越往上越接近业务与 AI 语义，越往下越接近 TCP/HTTP 与浏览器能力。

读图提示：同一条「用户发一句 → 模型流式回」路径，常见组合是 Chat API（应用 JSON）+ SSE（传输）；而「IDE 调本地工具」常见组合是 MCP（应用）+ stdio（传输）。

3. 快速对照：三者各管哪一层​

概念层次核心能力典型关键词
SSE传输/表示（基于 HTTP）服务器 → 客户端 单向、长连接、文本事件流text/event-stream、断线重连、浏览器 EventSource
WebSocket传输（独立协议）全双工、二进制/文本帧、低延迟双向消息ws:// / wss://、长连接、心跳
MCP应用协议工具/资源/提示等能力的发现与调用、与宿主（IDE、Agent 运行时）协作JSON-RPC 风格消息、多传输适配

4. HTTP 与「一问一答」​

经典 HTTP/1.1：客户端发一个请求，服务器返回完整响应体后，一次事务通常结束（连接可复用，但语义上仍是「一次请求对应一份完整答案」）。这对「整段 JSON 一次性返回」很友好，但对 token 级流式输出不友好：要么轮询，要么使用不结束响应体的 SSE，或改用 WebSocket 长连接。

5. SSE：基于 HTTP 的服务器单向流​

5.1 原理要点​

  • 仍是一次 HTTP 请求：客户端发起 GET（部分场景可用 POST），服务器返回 Content-Type: text/event-stream，并保持响应不结束。
  • 数据按 SSE 事件格式分块写出（data: 行、可选 event:、id:、注释行等），客户端按事件边界解析。
  • 方向以 服务器 → 客户端 为主；客户端若需上行，通常另发普通 HTTP 请求，而不是在同一条 SSE 里双向传。
  • 浏览器有原生 EventSource；Node/网关侧可按格式写入 HTTP 响应流。
  • 与 HTTPS、反向代理兼容度总体较好；HTTP/2 下行为以实际代理与浏览器为准，生产环境建议压测验证。

5.2 时序：谁保持连接不释放​

5.3 事件帧长什么样​

下面是一段最小可读的 SSE 文本（多行 data: 会被拼接为一则消息体，具体解析规则以规范为准）：

HTTP/1.1 200 OK
Content-Type: text/event-stream
Cache-Control: no-cache
Connection: keep-alive

: 心跳或注释行

id: 42
event: token
data: {"delta":"你"}

data: {"delta":"好"}

5.4 核心代码：Express 写 SSE 端点​

要点：禁用缓冲、设置 SSE 头、按行写入 data:，并用 \n\n 结束一则事件。

// Node + express：演示用，生产需鉴权、限流、错误与断开处理
import express from "express";

const app = express();

app.get("/api/chat/stream", (req, res) => {
  res.setHeader("Content-Type", "text/event-stream; charset=utf-8");
  res.setHeader("Cache-Control", "no-cache, no-transform");
  res.setHeader("Connection", "keep-alive");
  res.flushHeaders?.();

  const sendSse = (obj) => {
    res.write(`data: ${JSON.stringify(obj)}\n\n`);
  };

  sendSse({ type: "start" });
  const chunks = ["你", "好", "，", "世", "界"];
  let i = 0;
  const timer = setInterval(() => {
    if (i >= chunks.length) {
      sendSse({ type: "done" });
      clearInterval(timer);
      res.end();
      return;
    }
    sendSse({ type: "token", delta: chunks[i++] });
  }, 120);
});

app.listen(3000);

5.5 核心代码：浏览器用 EventSource 消费​

注意：EventSource 默认只用 GET，且不支持自定义请求头（若必须带 Token，常见做法是 query 参数或改用 fetch 读流，见下节）。

const es = new EventSource("/api/chat/stream");

es.onmessage = (ev) => {
  const payload = JSON.parse(ev.data);
  console.log("收到:", payload);
};

es.onerror = () => {
  // 浏览器会自动重连；生产环境应配合后端 id/Last-Event-ID 做断点续传策略
  console.warn("SSE 连接异常或结束");
};

5.6 与 fetch 流式读取的关系​

很多大模型 HTTP 流式接口返回 text/event-stream 或 chunked 文本，浏览器侧也可以用 fetch + ReadableStream 逐块读取并解析（便于自定义 Header、POST body）。解析是否按 SSE 规范分割，取决于上游格式：有的严格 SSE，有的是「每行一个 JSON」的近似流。

async function readOpenAIStyleStream(response) {
  const reader = response.body.getReader();
  const decoder = new TextDecoder();
  let buffer = "";
  while (true) {
    const { value, done } = await reader.read();
    if (done) break;
    buffer += decoder.decode(value, { stream: true });
    // 此处仅示意：真实需按服务端分隔符拆分 data: 行或 JSON 行
    console.log("chunk buffer:", buffer);
  }
}

5.7 在 AI 场景中的典型用途​

  • 大模型流式补全：服务端把模型生成的片段逐段推给前端，实现打字机效果。
  • 日志/进度推送：长时间任务的状态更新，只需服务端往下推。

6. WebSocket：独立 TCP 连接上的全双工通道​

6.1 原理要点​

  • 通过 HTTP Upgrade 握手后，切换为 WebSocket 帧协议，在一条长连接上双方随时互发消息。
  • 适合 高频双向、低延迟场景（协作、信令、实时 Agent 会话控制等）。
  • 需要自行约定应用层消息格式；连接管理（心跳、重连、背压）通常也要自己做。

6.2 时序：从 HTTP Upgrade 到 WS 帧​

6.3 核心代码：Node ws 最小服务端与浏览器客户端​

下面演示「双向 JSON 消息」——真实 AI 项目里可把 type 扩展为 user_message、model_delta、tool_result 等。

服务端（Node，ws 包）：

import { WebSocketServer } from "ws";

const wss = new WebSocketServer({ port: 8080 });

wss.on("connection", (socket) => {
  socket.send(JSON.stringify({ type: "hello", from: "server" }));

  socket.on("message", (data) => {
    const msg = JSON.parse(data.toString());
    if (msg.type === "ping") {
      socket.send(JSON.stringify({ type: "pong", t: Date.now() }));
    }
  });
});

客户端（浏览器）：

const ws = new WebSocket("ws://localhost:8080");

ws.onopen = () => {
  ws.send(JSON.stringify({ type: "ping" }));
};

ws.onmessage = (ev) => {
  console.log("server:", JSON.parse(ev.data));
};

6.4 在 AI 场景中的典型用途​

  • 实时多轮对话 + 打断/取消：客户端随时发新指令，服务端持续推送音频/文本片段。
  • 与实时 API 配套：语音链路、需要双向信令的 Agent 会话。
  • 远程 MCP 或 Agent 网关：部分实现选用 WebSocket 作为传输，便于双向工具调用与状态同步（以具体 SDK/部署为准）。

7. MCP：主机与「工具/资源/提示」之间的标准对话​

7.1 MCP 解决什么问题​

MCP（Model Context Protocol） 描述的是：AI 宿主（编辑器、桌面应用、Agent 运行时）如何连接 MCP 服务器，发现并调用 工具（Tools）、读取 资源（Resources）、使用 提示模板（Prompts） 等。它规定的是消息与能力模型，而不是「必须用哪一种网线协议」。

7.2 一次「发现工具 → 调用工具」的逻辑流​

下图是逻辑阶段示意，消息字段名以你所用的 SDK 与规范版本为准。

7.3 传输层：MCP 不绑定某一种传输​

规范与实现中常见：

  • stdio：本地子进程，标准输入输出上传输 JSON 消息，适合 IDE 与本地工具集成。
  • 基于 HTTP 的传输：远程服务时，可与 SSE 或 Streamable HTTP 等模式组合（以具体版本与实现为准）。
  • WebSocket：需要双向、长连接的部署里，也有采用 WebSocket 作为传输的实现。

因此：MCP 与 SSE/WebSocket 不是「三选一」的并列协议，而是 MCP = 应用层契约，SSE/WebSocket/stdio/HTTP = 可选传输与承载方式。

7.4 核心代码：stdio 上挂 MCP 的直觉（与 SDK 文档对照）​

真实项目应使用 @modelcontextprotocol/sdk 等官方库；下面用伪代码说明「进程边界」：宿主启动子进程，stdin/stdout 上往返 JSON，对应你在 MCP开发与应用 里的完整示例。

// 概念示意：宿主进程侧（非完整协议，仅说明 stdio 传输形态）
import { spawn } from "node:child_process";

const child = spawn("node", ["./mcp-server-entry.js"], {
  stdio: ["pipe", "pipe", "inherit"],
});

function sendRpc(jsonLineObject) {
  child.stdin.write(JSON.stringify(jsonLineObject) + "\n");
}

child.stdout.on("data", (buf) => {
  const text = buf.toString();
  console.log("来自 MCP Server 的原始回包:", text);
  // 真实场景：按行缓冲、解析 JSON-RPC、处理并发请求 id
});

// sendRpc({ jsonrpc: "2.0", id: 1, method: "tools/list", params: {} });

要点：stdio = 操作系统管道，不经过 TCP，天然适合「本机 IDE 插件拉起工具进程」；远程场景再考虑 HTTP+SSE 或 WebSocket。

8. 三者如何拼在一起：端到端架构​

8.1 流式聊天：Chat API + SSE​

8.2 MCP：同一套应用消息，多种传输​

8.3 组合视图：页面聊天 + 侧车工具（概念）​

当 BFF 既调大模型又需要 MCP 工具时，常见做法是：模型决定调用哪个工具 → BFF 或 Host 通过 MCP 执行 → 结果再注入对话。下图省略鉴权与错误分支。

9. 选型建议（简表）​

需求更常考虑的选项
只要模型输出流式展示，客户端主要是收SSE（或供应商提供的同类 HTTP 流）
需要强双向、低延迟、长会话WebSocket
统一走 443、防火墙友好、以服务端推送为主优先评估 SSE；若双向都很重再上 WebSocket
在 IDE/本地集成工具、与 CLI 子进程通信stdio 上的 MCP 很常见
标准化「工具/资源」发现与调用，并与多种宿主互通MCP（再选具体传输）

10. 常见误解​

  • 「MCP 替代了 SSE」：不对。MCP 不是为替代 SSE 而生；流式 token 输出仍大量依赖 SSE 类 HTTP 流。
  • 「做 MCP 必须用 WebSocket」：不对。stdio、HTTP 等同样常见，按场景选型。
  • 「SSE 和 WebSocket 都比 HTTP 高级」：它们解决的是不同交互形状（单向流 vs 全双工），不是简单升级关系。

11. 延伸阅读​

12. 简单完整示例：三个可独立运行的小项目​

下面三个示例彼此独立，可按目录分别拷贝到本地运行：不涉及真实大模型 API Key，便于先把「传输形状」跑通；接入真实模型时，只需在服务端把「推送 delta 的循环」换成上游 HTTP 流式解析即可。

12.1 示例在整体链路中的位置​

12.2 场景 A：SSE 流式模拟回复（Express + fetch 读流）​

目标：浏览器用 POST 把用户问题发给本机服务，服务端用 SSE（text/event-stream）按块推送「假模型」字符；前端用 fetch + ReadableStream 解析 data: 行（比 EventSource 更适合带 JSON body 的 POST）。

目录结构：

sse-stream-demo/
├── package.json
├── server.mjs
└── public/
    └── index.html

package.json：

{
  "name": "sse-stream-demo",
  "private": true,
  "type": "module",
  "scripts": {
    "start": "node server.mjs"
  },
  "dependencies": {
    "express": "^4.21.0"
  }
}

server.mjs（核心：设置 SSE 头、按块 write，两则 data: 之间用空行分隔）：

import express from "express";
import path from "path";
import { fileURLToPath } from "url";

const __dirname = path.dirname(fileURLToPath(import.meta.url));
const app = express();
const port = 3000;

app.use(express.json());
app.use(express.static(path.join(__dirname, "public")));

app.post("/api/chat", (req, res) => {
  const message = String(req.body?.message ?? "");
  res.setHeader("Content-Type", "text/event-stream; charset=utf-8");
  res.setHeader("Cache-Control", "no-cache, no-transform");
  res.setHeader("Connection", "keep-alive");
  res.flushHeaders?.();

  const writeSse = (obj) => {
    res.write(`data: ${JSON.stringify(obj)}\n\n`);
  };

  const reply = `（模拟）你说的是：${message || "（空）"}`;
  const tokens = reply.split("");

  let i = 0;
  writeSse({ type: "start" });
  const timer = setInterval(() => {
    if (i >= tokens.length) {
      writeSse({ type: "done" });
      clearInterval(timer);
      res.end();
      return;
    }
    writeSse({ type: "token", delta: tokens[i++] });
  }, 40);
});

app.listen(port, () => {
  console.log(`SSE demo: http://localhost:${port}`);
});

public/index.html（核心：按 \n\n 拆事件块，再解析以 data: 开头的行）：

<!DOCTYPE html>
<html lang="zh-Hans">
  <head>
    <meta charset="UTF-8" />
    <title>SSE 流式演示</title>
    <style>
      body { font-family: system-ui, sans-serif; max-width: 640px; margin: 2rem auto; }
      #out { white-space: pre-wrap; border: 1px solid #ccc; padding: 1rem; min-height: 120px; }
    </style>
  </head>
  <body>
    <h1>SSE 流式（POST + fetch 读流）</h1>
    <textarea id="msg" rows="3" style="width: 100%">你好</textarea>
    <p><button id="go">发送</button></p>
    <div id="out"></div>
    <script>
      const out = document.getElementById("out");
      const go = document.getElementById("go");

      async function run() {
        out.textContent = "";
        const res = await fetch("/api/chat", {
          method: "POST",
          headers: { "Content-Type": "application/json" },
          body: JSON.stringify({ message: document.getElementById("msg").value }),
        });
        const reader = res.body.getReader();
        const decoder = new TextDecoder();
        let buffer = "";
        while (true) {
          const { value, done } = await reader.read();
          if (done) break;
          buffer += decoder.decode(value, { stream: true });
          const parts = buffer.split("\n\n");
          buffer = parts.pop() ?? "";
          for (const block of parts) {
            for (const line of block.split("\n")) {
              if (line.startsWith("data: ")) {
                const payload = JSON.parse(line.slice(6));
                if (payload.type === "token") out.textContent += payload.delta;
              }
            }
          }
        }
      }
      go.onclick = run;
    </script>
  </body>
</html>

运行：

cd sse-stream-demo
npm install
npm start

浏览器打开 http://localhost:3000，点击「发送」即可看到打字机效果。

12.3 场景 B：WebSocket 双向回显（HTTP 静态页 + WS）​

目标：同一端口提供静态页面与 WebSocket：用户输入一行，服务端把「回显 + 时间戳」推回，演示 全双工 与自定义 JSON 消息格式（可扩展为 tool_result、audio_chunk 等）。

目录结构：

ws-echo-demo/
├── package.json
├── server.mjs
└── public/
    └── index.html

package.json：

{
  "name": "ws-echo-demo",
  "private": true,
  "type": "module",
  "scripts": {
    "start": "node server.mjs"
  },
  "dependencies": {
    "express": "^4.21.0",
    "ws": "^8.18.0"
  }
}

server.mjs（核心：http.Server 上挂 express 与 WebSocketServer）：

import express from "express";
import http from "http";
import path from "path";
import { fileURLToPath } from "url";
import { WebSocketServer } from "ws";

const __dirname = path.dirname(fileURLToPath(import.meta.url));
const app = express();
const server = http.createServer(app);
const wss = new WebSocketServer({ server });

app.use(express.static(path.join(__dirname, "public")));

wss.on("connection", (ws) => {
  ws.send(JSON.stringify({ type: "hello", from: "server" }));
  ws.on("message", (raw) => {
    let msg;
    try {
      msg = JSON.parse(raw.toString());
    } catch {
      ws.send(JSON.stringify({ type: "error", error: "invalid json" }));
      return;
    }
    if (msg.type === "chat") {
      const text = String(msg.text ?? "");
      ws.send(
        JSON.stringify({
          type: "reply",
          text: `回显：${text}`,
          t: Date.now(),
        })
      );
    }
  });
});

const port = 3001;
server.listen(port, () => {
  console.log(`WS demo: http://localhost:${port}`);
});

public/index.html：

<!DOCTYPE html>
<html lang="zh-Hans">
  <head>
    <meta charset="UTF-8" />
    <title>WebSocket 双向演示</title>
  </head>
  <body>
    <h1>WebSocket 回显</h1>
    <input id="line" placeholder="输入一句话" size="40" />
    <button id="send">发送</button>
    <pre id="log"></pre>
    <script>
      const log = document.getElementById("log");
      const ws = new WebSocket(`ws://${location.host}`);

      ws.onmessage = (ev) => {
        log.textContent += ev.data + "\n";
      };
      document.getElementById("send").onclick = () => {
        const text = document.getElementById("line").value;
        ws.send(JSON.stringify({ type: "chat", text }));
      };
    </script>
  </body>
</html>

运行：

cd ws-echo-demo
npm install
npm start

打开 http://localhost:3001，在控制台 Network 里可看到 101 Switching Protocols 的 WebSocket 连接。

12.4 场景 C：MCP stdio 最小服务（@modelcontextprotocol/sdk）​

目标：一个无 HTTP 的进程：通过 stdio 与宿主（如 Cursor、Claude Desktop、自写客户端）通信，对外暴露一个工具。下面与站内 MCP开发与应用 中「天气 MCP」一致，便于你对照「完整协议」章节。

目录结构：

mcp-stdio-minimal/
├── package.json
└── server.cjs

package.json（SDK 示例以 CommonJS 写法最常见；若你项目统一为 ESM，可再改 import 写法）：

{
  "name": "mcp-stdio-minimal",
  "private": true,
  "scripts": {
    "start": "node server.cjs"
  },
  "dependencies": {
    "@modelcontextprotocol/sdk": "^1.12.0",
    "zod": "^3.24.0"
  }
}

server.cjs：

const { McpServer } = require("@modelcontextprotocol/sdk/server/mcp.js");
const { StdioServerTransport } = require("@modelcontextprotocol/sdk/server/stdio.js");
const { z } = require("zod");

const server = new McpServer({
  name: "demo-stdio-server",
  version: "1.0.0",
});

server.registerTool(
  "get_weather",
  {
    title: "天气查询工具",
    description: "获取指定城市的模拟天气",
    inputSchema: { city: z.string().describe("城市名") },
  },
  async ({ city }) => {
    const mockWeather = `【${city}】当前天气：晴朗，25℃（模拟数据）`;
    return { content: { type: "text", text: mockWeather } };
  }
);

const transport = new StdioServerTransport();
server.connect(transport).then(() => {
  console.error("MCP stdio 已就绪（日志请走 stderr，避免污染 JSON-RPC）");
});

若安装后运行报错与 content 字段形状有关，请对照你本机 @modelcontextprotocol/sdk 版本的 registerTool 文档（少数版本要求 content 为块数组而非单个对象）。

运行与说明：

cd mcp-stdio-minimal
npm install
npm start

单独运行终端里看起来没有交互，这是正常的：MCP 在 stdin/stdout 上收发 JSON-RPC 消息，需由 宿主程序 拉起子进程并对接。你可以：

  • 在 Cursor / Claude Desktop 的配置里把 command 指到 node、args 指到本仓库的 server.cjs（具体路径因客户端而异）；
  • 或使用官方 MCP Inspector / 自写客户端，通过 stdio 连接（进阶内容见 MCP开发与应用）。

与场景 A、B 的关系：场景 A、B 在 浏览器侧演示 SSE 与 WebSocket；场景 C 在 进程间演示 MCP 的典型落地形态（stdio）。三者可组合进同一产品：浏览器 → BFF（SSE 或 WS）→ 模型；工具链侧由 Agent 宿主通过 MCP 调本地或远程工具。

一句话记忆：SSE/WebSocket 管「管道形状」，MCP 管「管道里在说什么」；流式聊天多用 SSE，强双向多用 WebSocket，工具与资源的标准化集成多看 MCP。