來源：艾逗筆

介紹 MCP 傳輸機制

MCP 傳輸機制（Transport）是 MCP 客戶端與 MCP 服務器通信的一個橋梁，定義了客戶端與服務器通信的細節，幫助客戶端和服務器交換消息。

MCP 協議使用 JSON-RPC 來編碼消息。JSON-RPC 消息必須使用 UTF-8 編碼。

MCP 協議目前定義了三種傳輸機制用于客戶端-服務器通信：

stdio：通過標準輸入和標準輸出進行通信

SSE：通過 HTTP 進行通信，支持流式傳輸。（協議版本 2024-11-05 開始支持，即將廢棄）

Streamble HTTP：通過 HTTP 進行通信，支持流式傳輸。（協議版本 2025-03-26 開始支持，用于替代 SSE）

MCP 協議要求客戶端應盡可能支持 stdio。

MCP 協議的傳輸機制是可插拔的，也就是說，客戶端和服務器不局限于 MCP 協議標準定義的這幾種傳輸機制，也可以通過自定義的傳輸機制來實現通信。

stdio 傳輸

stdio 即 standard input & output（標準輸入 / 輸出）。

是 MCP 協議推薦使用的一種傳輸機制，主要用于本地進程通信。

在 stdio 傳輸中：

客戶端以子進程的形式啟動 MCP 服務器。

服務器從其標準輸入（stdin）讀取 JSON-RPC 消息，并將消息發送到其標準輸出（stdout）。

消息可能是單個 JSON-RPC 請求、通知、響應，或者包含多個請求、通知、響應的 JSON-RPC 批處理。

消息由換行符分隔，且不得包含嵌套的換行符。

服務器可以將其 UTF-8 字符串寫入標準錯誤（stderr）以進行日志記錄。客戶端可以捕獲、轉發或忽略此日志。

服務器不得向 stdout 寫入無效的 MCP 消息內容。

客戶端不得向服務器的 stdin 寫入無效的 MCP 消息內容。

stdio 通信流程

客戶端以子進程的方式啟動服務器

客戶端往服務器的 stdin 寫入消息

服務器從自身的 stdin 讀取消息

服務端往自身的 stdout 寫入消息

客戶端從服務器的 stdout 讀取消息

客戶端終止子進程，關閉服務器的 stdin

服務器關閉自身的 stdout

stdio 傳輸實現

參考 MCP 官方的typescript-sdk來看 stdio 傳輸機制是如何實現的：

啟動 MCP 服務器

以命令行的方式，在本地啟動 MCP 服務器：

npx -y mcp-server-time

創建 stdio 通信管道

MCP 服務器啟動時，會創建 stdio 通信管道（pipeline），用于跟 MCP 客戶端進行消息通信。在 MCP 客戶端發送關閉信號，或者 MCP 服務器異常退出之前，這個通信管道會一直保持，常駐進程。

exportclassStdioServerTransportimplementsTransport {
private_readBuffer: ReadBuffer =newReadBuffer();
private_started =false;

constructor(
 private_stdin: Readable = process.stdin,
 private_stdout: Writable = process.stdout
) {}

 onclose?:()=>void;
 onerror?:(error:Error) =>void;
 onmessage?:(message: JSONRPCMessage) =>void;
}

從 stdin 讀取請求消息

MCP 客戶端把消息發到通信管道。MCP 服務器通過標準輸入 stdin 讀取客戶端發送的消息，以換行符： 作為讀取完成標識。

MCP 服務器讀取到的有效消息，是 JSON-RPC 編碼的結構體。

readMessage(): JSONRPCMessage |null{
 if(!this._buffer) {
  returnnull;
  }

 constindex =this._buffer.indexOf("
");
 if(index ===-1) {
  returnnull;
  }

 constline =this._buffer.toString("utf8",0, index).replace(/
$/,'');
 this._buffer =this._buffer.subarray(index +1);
 returndeserializeMessage(line);
}

往 stdout 寫入響應消息

MCP 服務器運行完內部邏輯，需要給 MCP 客戶端響應消息。

MCP 服務器先用 JSON-RPC 編碼消息，再把消息寫入標準輸出 stdout。

send(message: JSONRPCMessage):Promise {
 returnnewPromise((resolve) =>{
  constjson = serializeMessage(message);
  if(this._stdout.write(json)) {
    resolve();
   }else{
   this._stdout.once("drain", resolve);
   }
  });
 }
}

MCP 客戶端從 MCP 服務器的標準輸出 stdout 讀取消息，獲得 MCP 服務器的響應內容。

關閉 stdio 通信管道

MCP 客戶端退出，給 MCP 服務器發送關閉信號。

MCP 服務器通過 stdio 通信管道讀到客戶端發送的終止信號，或者內部運行錯誤，主動關閉 stdio 通信管道。

stdio 通信管道關閉之后，MCP 客戶端與 MCP 服務器之間不能再相互發送消息，直到再次建立 stdio 通信管道。

asyncclose():Promise {
 // Remove our event listeners first
 this._stdin.off("data",this._ondata);
 this._stdin.off("error",this._onerror);

 // Check if we were the only data listener
 constremainingDataListeners =this._stdin.listenerCount('data');
 if(remainingDataListeners ===0) {
  // Only pause stdin if we were the only listener
  // This prevents interfering with other parts of the application that might be using stdin
  this._stdin.pause();
  }

 // Clear the buffer and notify closure
 this._readBuffer.clear();
 this.onclose?.();
}

stdio 傳輸的利弊

stdio 傳輸機制主要依靠本地進程通信實現。

主要的優勢是：

無外部依賴，實現簡單

無網絡傳輸，通信速度快

本地通信，安全性高

也有一些局限性：

單進程通信，無法并行處理多個客戶端請求

進程通信的資源開銷大，很難在本地運行非常多的服務

stdio 傳輸的適用場景

stdio 傳輸適用于要操作的數據資源位于本地計算機，且不希望暴露外部訪問的場景。

比如，你希望通過一個聊天客戶端，來總結你的微信消息，微信消息文件存儲在你的本地電腦，外部訪問不了，也不應該訪問。

這種情況，你可以實現一個 MCP 服務器來讀取你電腦上的微信消息文件，通過 stdio 傳輸接收 MCP 客戶端的訪問請求。

如果你要訪問的是一個遠程服務器上的文件，也可以使用 stdio 傳輸，流程會復雜一些：

先寫一個 API 服務，部署在遠程服務器，操作遠程服務器上的資源，暴露公網訪問

寫一個 MCP 服務器，對接遠程 API，再通過 stdio 傳輸與客戶端本地通信

既然 stdio 傳輸訪問遠程資源這么麻煩，是不是應該有一種更適合遠程資源訪問的傳輸機制？

當然有。可以使用 SSE 傳輸。

SSE 傳輸

MCP 協議使用 SSE（Server-Sent Events） 傳輸來解決遠程資源訪問的問題。底層是基于 HTTP 通信，通過類似 API 的方式，讓 MCP 客戶端直接訪問遠程資源，而不用通過 stdio 傳輸做中轉。

在 SSE 傳輸中，服務器作為一個獨立進程運行，可以處理多個客戶端連接。

服務器必須提供兩個端點：

一個 SSE 端點，供客戶端建立連接并從服務器接收消息

一個常規 HTTP POST 端點，供客戶端向服務器發送消息

當客戶端連接時，服務器必須發送一個包含客戶端用于發送消息的 URL 的端點事件。所有后續客戶端消息必須作為 HTTP POST 請求發送到該端點。

服務器消息作為 SSE 消息事件發送，消息內容以 JSON 格式編碼在事件數據中。

SSE 通信流程

客戶端向服務器的 /sse 端點發送請求（一般是 GET 請求），建立 SSE 連接

服務器給客戶端返回一個包含消息端點地址的事件消息

客戶端給消息端點發送消息

服務器給客戶端響應消息已接收狀態碼

服務器給雙方建立的 SSE 連接推送事件消息

客戶端從 SSE 連接讀取服務器發送的事件消息

客戶端關閉 SSE 連接

SSE 傳輸實現

參考 MCP 官方的typescript-sdk來看 SSE 傳輸機制是如何實現的：

啟動 MCP 服務器

系統管理員在遠程服務器（也可以是本地電腦）輸入命令，啟動 MCP 服務器，監聽服務器端口，對外暴露 HTTP 接口。

constserver =newMcpServer({
 name:"example-server",
 version:"1.0.0",
});

constapp = express();

app.get("/sse",async(_: Request, res: Response) => {
consttransport =newSSEServerTransport("/messages", res);

awaitserver.connect(transport);
});

app.post("/messages",async(req: Request, res: Response) => {
awaittransport.handlePostMessage(req, res);
});

app.listen(3001);

在這個示例中，使用了express框架，啟動了一個 HTTP 服務，監聽在 3001 端口，對外暴露了兩個端點：

/sse：GET 請求，用于建立 SSE 連接

/messages：POST 請求，用于接收客戶端發送的消息

解析一個 MCP 客戶端可訪問的域名到 MCP 服務器。比如：abc.mcp.so

建立 SSE 連接

MCP 客戶端請求 MCP 服務器的 URL 地址：https://abc.mcp.so:3001/sse與 MCP 服務器建立連接，MCP 服務器需要給 MCP 客戶端返回一個用于消息通信的地址：

res.writeHead(200, {
"Content-Type":"text/event-stream",
"Cache-Control":"no-cache, no-transform",
 Connection:"keep-alive",
});

constmessagesUrl ="https://abc.mcp.so:3001/messages?sessionId=xxx";

res.write(`event: endpoint
data:${messagesUrl}

`);

MCP 客戶端從 MCP 服務器返回的 endpoint 事件中得到了消息通信的地址，與 MCP 服務器建立 SSE 連接成功。

MCP 客戶端通過 POST 請求把消息發到這個消息通信地址，與 MCP 服務器進行消息交互。

消息交互

MCP 客戶端在與 MCP 服務器建立 SSE 連接之后，開始給 MCP 服務器返回的通信地址發送消息。

MCP 協議中的 SSE 傳輸是雙通道響應機制。也就是說，MCP 服務器在接收到 MCP 客戶端的消息之后，既要給當前的請求回復一個響應，也要給之前建立的 SSE 連接發送一條響應消息。（通知類型的消息，不需要給 SSE 連接發消息）

舉個例子，MCP 客戶端與 MCP 服務器建立 SSE 連接之后，給 MCP 服務器發送的第一條消息，用于初始化階段做能力協商。

MCP 客戶端請求示例：

curl -X POST https://abc.mcp.so/messages?sessionId=xxx 
-H"Content-Type: application/json"
-d '{
"jsonrpc":"2.0",
"id":"1",
"method":"initialize",
"params": {
 "protocolVersion":"1.0",
 "capabilities": {},
 "clientInfo": {
  "name":"mcp-client",
  "version":"1.0.0"
  }
 }
}'

MCP 服務器從 HTTP 請求體里面讀取 MCP 客戶端發送的消息：

asynchandlePostMessage(
 req: IncomingMessage,
 res: ServerResponse,
 parsedBody?: unknown,
):Promise {
if(!this._sseResponse) {
 constmessage ="SSE connection not established";
  res.writeHead(500).end(message);
 thrownewError(message);
 }

letbody:string| unknown;
try{
 constct = contentType.parse(req.headers["content-type"] ??"");
 if(ct.type !=="application/json") {
  thrownewError(`Unsupported content-type:${ct}`);
  }

  body = parsedBody ??awaitgetRawBody(req, {
   limit: MAXIMUM_MESSAGE_SIZE,
   encoding: ct.parameters.charset ??"utf-8",
  });
 }catch(error) {
  res.writeHead(400).end(String(error));
 this.onerror?.(errorasError);
 return;
 }

try{
 awaitthis.handleMessage(typeofbody ==='string'?JSON.parse(body) : body);
 }catch{
  res.writeHead(400).end(`Invalid message:${body}`);
 return;
 }

 res.writeHead(202).end("Accepted");
}

先給當前請求，響應一個 HTTP 202 狀態碼，告知 MCP 客戶端，請求已收到。

然后 MCP 服務器運行內部邏輯，完成業務功能。

再給之前與 MCP 客戶端建立的 SSE 連接發送一個事件消息，data 里面放 JSON-RPC 編碼的消息內容：

asyncsend(message: JSONRPCMessage):Promise {
if(!this._sseResponse) {
 thrownewError("Not connected");
 }

this._sseResponse.write(
 `event: message
data:${JSON.stringify(message)}

`,
 );
}

MCP 客戶端根據 MCP 服務器同步響應的 2** 狀態碼，判斷 MCP 服務器已經接到請求，并開始讀取 MCP 服務器發到 SSE 連接的消息內容。

MCP 客戶端從與 MCP 服務器建立的 SSE 連接中讀取event: message事件消息，獲得 MCP 服務器發送的業務數據data。

MCP 客戶端與 MCP 服務器建立的 SSE 連接，應該是 1:1 的。為了防止串數據的問題，在建立 SSE 連接階段，MCP 服務器返回的通信地址，應該為當前連接分配一個唯一標識，叫做sessionId，給 MCP 客戶端返回的通信地址帶上這個標識，比如/messages?sessionId=xxx。

在消息交互階段，MCP 服務器根據 MCP 客戶端請求地址參數里面的 sessionId，找到之前建立的 SSE 連接，并只給這個 SSE 連接發送消息。

斷開 SSE 連接

MCP 服務器與 MCP 客戶端雙方都可能會主動斷開 SSE 連接。

還保持連接的一方，應該加上必要的連接檢測和超時關閉機制。

比如通過 SSE 連接，給對方定時發送一條心跳檢測消息，如果多次無響應，可以認作對方已斷開連接，此時可以主動關閉 SSE 連接，避免資源泄露。

一個用 go 實現的心跳檢測和超時關閉示例：

// Setup heartbeat ticker
heartbeatInterval :=30* time.Second
heartbeatTicker := time.NewTicker(heartbeatInterval)
deferheartbeatTicker.Stop()

// Setup idle timeout
idleTimeout :=5* time.Minute
idleTimer := time.NewTimer(idleTimeout)
deferidleTimer.Stop()

gofunc(){
for{
 select{
 case<-session.Done():
? ? ??return
? ??case?<-heartbeatTicker.C:
? ? ??// Send heartbeat
? ? ??if?err := writer.SendHeartbeat(); err !=?nil?{
? ? ? ? session.Close()
? ? ? ??return
? ? ? }
? ??case?<-idleTimer.C:
? ? ??// Close connection due to inactivity
? ? ? session.Close()
? ? ??return
? ? }
? }
}()

SSE 安全防護

當使用 SSE 傳輸時，服務器一方需要實現一些必要的安全防護措施：

服務器必須驗證所有傳入連接的 Origin 頭，以防止 DNS 重綁定攻擊

在本地運行時，服務器應僅綁定到 localhost（127.0.0.1），而不是所有網絡接口（0.0.0.0）

服務器應對所有連接實施適當的身份驗證

如果沒有這些保護措施，攻擊者可能會使用 DNS 重綁定從遠程網站與本地 MCP 服務器交互。

SSE 傳輸的適用場景

SSE 傳輸適用于 MCP 客戶端與 MCP 服務器不在同一個網絡下的通信場景。

比如，你希望在本地電腦，通過對話的方式，查詢你云服務器上的數據庫。你就可以在你的云服務器上部署一個 MCP 服務器，去讀取數據庫，再跟你本地電腦上的 MCP 客戶端建立連接通信。

當然，所有用 SSE 傳輸實現的 MCP 服務器，理論上都可以通過 stdio 傳輸 + API 的方式實現。

區別在于：

用 SSE 傳輸，MCP 客戶端直接與 MCP 服務器通信，而不用通過本地的 stdio 傳輸調用 API 進行中轉。

用 SSE 傳輸，在 MCP 客戶端只需要一個 URL 即可接入，對本地環境無要求，也無需在本地運行 MCP 服務器，用戶側的使用門檻更低。

SSE 傳輸的利弊

SSE 傳輸主要解決遠程資源訪問的問題，依靠 HTTP 協議實現底層通信。

SSE 傳輸的主要優勢：

支持遠程資源訪問，讓 MCP 客戶端可以直接訪問遠程服務，解決了 stdio 傳輸僅適用于本地資源的局限

基于標準 HTTP 協議實現，兼容性好，便于與現有 Web 基礎設施集成

服務器可作為獨立進程運行，支持處理多個客戶端連接

相比 WebSocket 實現簡單，是普通 HTTP 的擴展，不需要協議升級

SSE 傳輸的主要劣勢與問題：

連接不穩定：在無服務器(serverless)環境中，SSE 連接會隨機、頻繁斷開，影響 AI 代理需要的可靠持久連接

擴展性挑戰：SSE 不是為云原生架構設計的，在擴展平臺時會遇到瓶頸

瀏覽器連接限制：每個瀏覽器和域名的最大打開連接數很低(6 個)，當用戶打開多個標簽頁時會出現問題

代理和防火墻問題：某些代理和防火墻會因為缺少 Content-Length 頭而阻止 SSE 連接，在企業環境部署時造成挑戰

復雜的雙通道響應機制：MCP 中的 SSE 實現要求服務器在接收客戶端消息后，既要給當前請求響應，也要給之前建立的 SSE 連接發送響應消息

無法支持長期的無服務器部署：無服務器架構通常自動擴縮容，不適合長時間連接，而 SSE 需要維持持久連接

需要大量會話管理：需要為每個 SSE 連接分配唯一標識(sessionId)來防止數據混淆，增加了實現復雜度

需要額外的連接檢測和超時關閉機制：需要實現心跳檢測和超時機制來避免資源泄露

正因為這些問題，MCP 協議已經引入了新的 Streamable HTTP 傳輸機制(2025-03-26 版本)來替代 SSE，并計劃廢棄 SSE 傳輸。新的傳輸機制保留了 HTTP 的基礎，但支持更靈活的連接方式，更適合現代云架構和無服務器環境。

Streamable HTTP 傳輸

Streamable HTTP 傳輸是 MCP 協議在 2025-03-26 版本中引入的新傳輸機制，用于替代之前的 SSE 傳輸。

在 Streamable HTTP 傳輸中，服務器作為一個獨立進程運行，可以處理多個客戶端連接。此傳輸使用 HTTP POST 和 GET 請求，服務器可以選擇使用服務器發送事件（SSE）來流式傳輸多個服務器消息。

服務器必須提供一個同時支持 POST 和 GET 方法的單個 HTTP 端點。例如：https://xyz.mcp.so/mcp。

Streamable HTTP 通信流程

客戶端給服務器的通信端點發消息

服務器給客戶端響應消息

客戶端根據服務器的響應類型，繼續給服務器發消息

服務器繼續響應客戶端消息

跟 SSE 傳輸不同的點在于，Streamable HTTP 傳輸中，客戶端與服務器的消息交互，基本上是“一來一回”的（單通道響應）。而這個“一來一回”的消息交互，可能會有很多種組合類型。

客戶端發送 GET 請求給服務器，服務器返回 SSE 連接

客戶端 POST JSON-RPC 編碼的消息給服務器，服務器返回 JSON-RPC 編碼的消息響應

客戶端 POST JSON-RPC 編碼的消息給服務器，服務器返回一個 SSE 連接

客戶端給 SSE 連接發消息，服務器收到后給 SSE 連接響應消息

服務器響應的消息，可能包含狀態標識：Mcp-Session-Id

客戶端發消息時候需要帶上狀態標識：Mcp-Session-Id

Streamable HTTP 傳輸實現

參考 MCP 官方的typescript-sdk來看 Streamable HTTP 傳輸機制是如何實現的：

啟動服務器

跟 SSE 傳輸機制一樣，Streamable HTTP 傳輸本質上也是基于 HTTP 協議通信，需要先啟動一個 HTTP 服務：

constserver =newMcpServer({
 name:"example-server",
 version:"1.0.0",
});

constapp = express();

app.all("/mcp",async(req: Request, res: Response) => {
consttransport =newStreamableHTTPServerTransport();

awaitserver.connect(transport);

awaittransport.handleMessage(req, res);
});

app.listen(3002);

跟 SSE 傳輸不一樣的點在于，Streamable HTTP 傳輸只需要暴露一個端點（endpoint），來接收各種類型的客戶端請求（GET / POST / DELETE）。

比如在這個例子，使用express框架，啟動了一個 HTTP 服務，監聽在 3002 端口，對外暴露了一個端點：

/mcp：接收客戶端建立連接、交換消息的請求

解析一個 MCP 客戶端可訪問的域名到 MCP 服務器。比如：xyz.mcp.so

消息交互

在 MCP 服務器啟動成功之后，MCP 客戶端可以直接給 MCP 服務器暴露的地址發送消息。

跟 SSE 傳輸不同，Streamable HTTP 傳輸機制中，MCP 客戶端給服務器發送消息，無需先跟一個端點建立 SSE 連接，再給另一個端點發消息。而是單通道模式，即客戶端給服務器發消息，直接獲得服務器的響應內容。

MCP 客戶端可以使用 GET 或者 POST 請求給服務器發消息，每個請求必須設置請求頭Accept，傳遞以下兩個值：

application/json 接收服務器響應的 JSON-RPC 編碼消息

text/event-stream 由服務器開啟流式傳輸通道，客戶端從這個流里面讀取事件消息

MCP 客戶端請求示例：

curl -X POST https://xyz.mcp.so/mcp 
-H "Content-Type: application/json" 
-H "Accept: application/json, text/event-stream" 
-d '{
 "jsonrpc": "2.0",
 "id": "1",
 "method": "initialize",
 "params": {
  "protocolVersion": "1.0",
  "capabilities": {},
  "clientInfo": {
   "name": "mcp-client",
   "version": "1.0.0"
  }
 }
}'

Streamable HTTP 傳輸機制下，MCP 客戶端與服務器通信的幾個要點：

客戶端可以給服務器發送不包含請求體的 GET 請求，用于建立 SSE 連接，讓服務器可以主動給客戶端先發消息

客戶端給服務器發送 JSON-RPC 消息的情況，必須使用 POST 請求

服務器接到客戶端的 GET 請求時，要么返回Contet-Type: text/event-stream開啟 SSE 連接，要么返回 HTTP 405 狀態碼，表示不支持 SSE 連接。

服務器接到客戶端的 POST 請求時，從請求體里面讀取 JSON-RPC 消息，如果是通知消息，就響應 HTTP 202 狀態碼，表示消息已收到。如果是非通知消息，服務器可以選擇返回Content-Type: text/event-stream開啟 SSE 傳輸，或者返回Content-Type: application/json同步響應一條 JSON-RPC 消息。

會話保持

Streamable HTTP 傳輸既支持無狀態的請求：每一次請求都是獨立的，無需記錄狀態。

也支持有狀態的請求：一次新的請求，可能需要同步之前的請求 / 響應信息作為參考。這種情況叫做：會話保持。

如果需要保持會話，MCP 服務器與 MCP 客戶端之間的交互應該遵守以下原則：

使用 Streamable HTTP 傳輸的服務器可以在初始化時分配一個會話 ID，方法是在包含InitializeResult的 HTTP 響應中包含它，放在Mcp-Session-Id頭中。

如果服務器在初始化期間返回了Mcp-Session-Id，使用 Streamable HTTP 傳輸的客戶端必須在所有后續的 HTTP 請求中在Mcp-Session-Id頭中包含它。

服務器可以隨時終止會話，之后它必須使用 HTTP 404 Not Found 響應包含該會話 ID 的請求。

當客戶端收到對包含Mcp-Session-Id的請求的 HTTP 404 響應時，它必須通過發送一個不帶會話 ID 的新InitializeRequest來啟動一個新會話。

不再需要特定會話的客戶端應該發送一個帶有Mcp-Session-Id頭的 HTTP DELETE 到 MCP 端點，以顯式終止會話。

MCP 服務器驗證會話的一個示例：

/**
* Validates session ID for non-initialization requests
* Returns true if the session is valid, false otherwise
*/
privatevalidateSession(req: IncomingMessage, res: ServerResponse):boolean{
if(!this._initialized) {
 // If the server has not been initialized yet, reject all requests
  res.writeHead(400).end(JSON.stringify({
   jsonrpc:"2.0",
   error: {
    code:-32000,
    message:"Bad Request: Server not initialized"
   },
   id:null
  }));
 returnfalse;
 }
if(this.sessionId ===undefined) {
 // If the session ID is not set, the session management is disabled
 // and we don't need to validate the session ID
 returntrue;
 }
constsessionId = req.headers["mcp-session-id"];

if(!sessionId) {
 // Non-initialization requests without a session ID should return 400 Bad Request
  res.writeHead(400).end(JSON.stringify({
   jsonrpc:"2.0",
   error: {
    code:-32000,
    message:"Bad Request: Mcp-Session-Id header is required"
   },
   id:null
  }));
 returnfalse;
 }elseif(Array.isArray(sessionId)) {
  res.writeHead(400).end(JSON.stringify({
   jsonrpc:"2.0",
   error: {
    code:-32000,
    message:"Bad Request: Mcp-Session-Id header must be a single value"
   },
   id:null
  }));
 returnfalse;
 }
elseif(sessionId !==this.sessionId) {
 // Reject requests with invalid session ID with 404 Not Found
  res.writeHead(404).end(JSON.stringify({
   jsonrpc:"2.0",
   error: {
    code:-32001,
    message:"Session not found"
   },
   id:null
  }));
 returnfalse;
 }

returntrue;
}

連接斷開與重連

Streamable HTTP 傳輸，如果客戶端與服務器使用 SSE 連接通信，斷開連接的方式跟 SSE 傳輸斷開連接的方式一致。

可以由連接的任意一方主動斷開連接。還保持著連接的一方，需要實現心跳檢測和超時機制，以便能及時關閉連接，避免資源泄露。

Streamable HTTP 傳輸比起 SSE 傳輸，做了一些改進，支持恢復已中斷的連接，重新發送可能丟失的消息：

服務器可以在其 SSE 事件中附加一個 ID 字段。如果存在，ID 必須在所有會話所有流中全局唯一。

如果客戶端希望在斷開連接后恢復，它應該向服務器發出 HTTP GET 請求，并包含 Last-Event-ID 頭，告知服務器它接收到的最后一個事件 ID。服務器可以重放在最后一個事件 ID 之后將發送的消息，并從該點恢復流。

支持斷點重連的 Streamable HTTP 傳輸，在消息傳輸方面會比 SSE 傳輸更加可靠。

Streamable HTTP 傳輸的利弊

Streamable HTTP 傳輸機制結合了 SSE 傳輸的遠程訪問能力和無狀態 HTTP 的靈活性，同時解決了 SSE 傳輸中的許多問題。

主要優勢：

兼容無服務器環境，可以在短連接模式下工作

靈活的連接模式，支持簡單的請求-響應和流式傳輸

會話管理更加標準化和清晰

支持斷開連接恢復和消息重傳

保留了 SSE 的流式傳輸能力，同時解決了其穩定性問題

向后兼容，可以支持舊版客戶端和服務器

主要劣勢：

相比單純的 stdio 傳輸實現復雜度更高

仍需處理網絡連接斷開和恢復的邏輯

會話管理需要服務器引入額外的組件（比如用 Redis 來存儲 Session）

Streamable HTTP 傳輸的適用場景

Streamable HTTP 傳輸適用于：

需要遠程訪問服務的場景，特別是云環境和無服務器架構

需要支持流式輸出的 AI 服務

需要服務器主動推送消息給客戶端的場景

大規模部署需要高可靠性和可擴展性的服務

需要在不穩定網絡環境中保持可靠通信的場景

與 SSE 傳輸相比，Streamable HTTP 傳輸是一個更全面、更靈活的解決方案，特別適合現代云原生應用和無服務器環境。

自定義傳輸

MCP 客戶端和 MCP 服務器可以實現額外的自定義傳輸機制以滿足其特定需求。MCP 協議與傳輸無關，可以在支持雙向消息交換的任何通信通道上實現。

選擇支持自定義傳輸的實現者必須確保他們保留由 MCP 定義的 JSON-RPC 消息格式和生命周期要求。自定義傳輸應記錄其特定的連接建立和消息交換模式，以實現互操作性。

用一個實際的例子來說明，如何實現一個自定義傳輸，來滿足特定的需求。

需求分析

目前市面上大部分的 MCP 服務器是使用 stdio 和 SSE 傳輸機制實現的，用戶要使用這些 MCP 服務器，需要拉代碼到本地運行，使用門檻有點高。

我們希望實現一個 MCP 代理服務，在云上部署，讓用戶僅需要配置一個 URL 即可接入。由這個云端部署的 MCP 代理去對接第三方的 MCP 服務器。

大致的流程是：

這個 MCP 代理服務以 HTTP 的形式提供接入，需要支持并發調用。這個 MCP 代理服務作為 stdio 傳輸或者 SSE 傳輸的客戶端，發送消息給后臺的 MCP 服務器。

按照這個方案，MCP 代理服務跟后臺部署的第三方服務器之間的交互存在著一些問題：

如果后臺對接的 MCP 服務器是基于 stdio 傳輸實現的，MCP 代理每接到一個用戶請求，都需要調用 MCP 服務器創建一個獨立的 stdio 通信進程。服務器開銷特別大（進程創建和銷毀、上下文切換，內存隔離等操作，都很消耗服務器資源）。

如果后臺對接的 MCP 服務器是基于 SSE 傳輸實現的，從前面對 SSE 傳輸的分析我們知道，MCP 代理需要跟后臺部署的 MCP 服務器之間保持一個 SSE 連接。如果每個用戶請求都需要維持一個 SSE 的長連接，對服務器也是不小的壓力。

為了解決這些問題，對于 MCP 代理和其背后連接的 MCP 服務器，可以想到的一些優化方案：

用 k8s 集群分布式部署，代替單機部署，保持 MCP 代理的彈性擴容能力，支持更高的并發請求

修改第三方服務器的傳輸機制，如果是 stdio 進程通信，可以改成輕量級的協程通信，或者使用 HTTP 通信，支持多路復用

如果第三方服務器的傳輸機制是 SSE，需要把雙通道響應改成單通道響應，并且不應該使用長連接

分布式網絡下，會話保持需要引入一些額外的組件（比如 redis）或者額外的策略（比如用負載均衡的 IP Hash 策略把請求路由到固定的機器），如果 MCP 服務器不是必須要用到會話保持，那就最好改成無狀態的傳輸

基于以上幾點分析，我們可以設計一套自定義的傳輸機制，來改造在分布式集群部署的第三方 MCP 服務器。

這套自定義的傳輸機制應該具備的幾個特性：

基于 HTTP 通信

非流式傳輸

無狀態，無會話保持

短連接

雖然 Streamable HTTP 傳輸通過配置參數也能實現這些功能，但是我們還是希望能自定義一套更簡單，更直接，零配置的傳輸機制。

我們把這個新的傳輸機制取名為：Restful HTTP Transport，簡稱 Rest Transport

因為要改造的主要是 MCP 服務器，接下來主要講解如何實現一個服務器使用的 Rest Server Transport.

實現 Rest Server Transport

參考 MCP 官方實現的 Streamable HTTP Transport，我們來實現這個自定義的 Rest Server Transport。

定義新的傳輸類，實現 MCP 協議的 Transport 接口

/**
* Server transport for Synchronous HTTP: a stateless implementation for direct HTTP responses.
* It supports concurrent requests with no streaming, no SSE, and no persistent connections.
*/
exportclassRestServerTransportimplementsTransport {
// ...
}

定義啟動服務方法，用來啟動一個 HTTP 服務器，處理客戶端的請求

/**
* Start the HTTP server
*/
asyncstartServer():Promise {
if(this._server) {
 thrownewError("Server is already running");
 }

this._server = express();
this._server.post(this._endpoint,(req, res) =>{
 this.handleRequest(req, res, req.body);
 });

returnnewPromise((resolve, reject) =>{
 try{
  this._httpServer =this._server!.listen(this._port,()=>{
   console.log(
    `Server is running on http://localhost:${this._port}${this._endpoint}`
    );
    resolve();
   });

  this._httpServer.on("error",(error) =>{
   console.error("Server error:", error);
   this.onerror?.(error);
   });
  }catch(error) {
   reject(error);
  }
 });
}

跟 Streamable HTTP 傳輸一樣，自定義的傳輸在啟動 HTTP 服務器之后，我們也只暴露一個端點來接收客戶端請求，這個端點和 HTTP 服務器監聽的端口，都可以在啟動服務器的時候自定義。

exportinterfaceRestServerTransportOptions {
 endpoint?:string;
 port?:string|number;
}

接收客戶端請求

跟 Streamable HTTP 傳輸最主要的區別，我們自定義的這個傳輸，僅支持客戶端發起 POST 請求，客戶端也只會收到application/json響應，不會收到text/event-stream響應。

客戶端與服務器使用短連接通信，不會有Connection: "keep-alive"。服務器響應完，與客戶端的連接就會斷開。

客戶端與服務器之間的消息交互是無狀態的，所以客戶端無需傳遞Mcp-Session-Id，服務器也不會判斷客戶端的會話有效性，不會維護任何 session 相關的數據。

Rest Server Transport 處理用戶請求的主要實現邏輯：

/**
* Handles an incoming HTTP request
*/
asynchandleRequest(
 req: IncomingMessage,
 res: ServerResponse,
 parsedBody?: unknown
):Promise {
if(req.method ==="POST") {
 awaitthis.handlePostRequest(req, res, parsedBody);
 }else{
  res.writeHead(405).end(
  JSON.stringify({
    jsonrpc:"2.0",
    error: {
     code:-32000,
     message:"Method not allowed",
    },
    id:null,
   })
  );
 }
}

/**
* Handles POST requests containing JSON-RPC messages
*/
privateasynchandlePostRequest(
 req: IncomingMessage,
 res: ServerResponse,
 parsedBody?: unknown
):Promise {
try{
 // validate the Accept header
 constacceptHeader = req.headers.accept;
 if(
   acceptHeader &&
   acceptHeader !=="*/*"&&
   !acceptHeader.includes("application/json")
  ) {
   res.writeHead(406).end(
   JSON.stringify({
     jsonrpc:"2.0",
     error: {
      code:-32000,
      message:"Not Acceptable: Client must accept application/json",
     },
     id:null,
    })
   );
  return;
  }

 constct = req.headers["content-type"];
 if(!ct || !ct.includes("application/json")) {
   res.writeHead(415).end(
   JSON.stringify({
     jsonrpc:"2.0",
     error: {
      code:-32000,
      message:
      "Unsupported Media Type: Content-Type must be application/json",
     },
     id:null,
    })
   );
  return;
  }

 letrawMessage;
 if(parsedBody !==undefined) {
   rawMessage = parsedBody;
  }else{
  constparsedCt = contentType.parse(ct);
  constbody =awaitgetRawBody(req, {
    limit: MAXIMUM_MESSAGE_SIZE,
    encoding: parsedCt.parameters.charset ??"utf-8",
   });
   rawMessage =JSON.parse(body.toString());
  }

 letmessages: JSONRPCMessage[];

 // handle batch and single messages
 if(Array.isArray(rawMessage)) {
   messages = rawMessage.map((msg) =>JSONRPCMessageSchema.parse(msg));
  }else{
   messages = [JSONRPCMessageSchema.parse(rawMessage)];
  }

 // check if it contains requests
 consthasRequests = messages.some(
  (msg) =>"method"inmsg &&"id"inmsg
  );
 consthasOnlyNotifications = messages.every(
  (msg) =>"method"inmsg && !("id"inmsg)
  );

 if(hasOnlyNotifications) {
  // if it only contains notifications, return 202
   res.writeHead(202).end();

  // handle each message
  for(constmessage of messages) {
   this.onmessage?.(message);
   }
  }elseif(hasRequests) {
  // Create a unique identifier for this request batch
  constrequestBatchId = randomUUID();

  // Extract the request IDs that we need to collect responses for
  constrequestIds = messages
    .filter((msg) =>"method"inmsg &&"id"inmsg)
    .map((msg) =>String(msg.id));

  // Set up a promise that will be resolved with all the responses
  constresponsePromise =newPromise((resolve) => {
   this._pendingRequests.set(requestBatchId, {
     resolve,
     responseMessages: [],
     requestIds,
    });
   });

   //Processallmessages
  for(constmessage of messages) {
   this.onmessage?.(message);
   }

   //Waitforresponsesandsendthem
  constresponses=awaitPromise.race([
    responsePromise,
   // 30 second timeout
   newPromise((resolve) =>
     setTimeout(() => resolve([]), 30000)
   ),
   ]);

   //Cleanupthependingrequest
  this._pendingRequests.delete(requestBatchId);

   //Setresponseheaders
  constheaders:Record = {
    "Content-Type": "application/json",
   };

  res.writeHead(200, headers);

   //FormattheresponseaccordingtoJSON-RPCspec
  constresponseBody=responses.length=== 1 ?responses[0] :responses;
  res.end(JSON.stringify(responseBody));
  }
 }catch(error) {
  //returnJSON-RPCformattederror
 res.writeHead(400).end(
  JSON.stringify({
    jsonrpc: "2.0",
    error: {
     code: -32700,
     message: "Parse error",
     data:String(error),
    },
    id:null,
   })
 );
 this.onerror?.(errorasError);
 }
}

服務器發送消息

MCP 服務器接到客戶端請求后，把請求數據發到內部實現的各個功能函數內，得到響應內容后，用 JSON-RPC 編碼，響應給 MCP 客戶端。

Rest Server Transport 發送消息的主要實現邏輯：

asyncsend(message: JSONRPCMessage):Promise {
// Only process response messages
if(!("id"inmessage) || !("result"inmessage ||"error"inmessage)) {
 return;
 }

constmessageId =String(message.id);

// Find the pending request that is waiting for this response
for(const[batchId, pendingRequest] ofthis._pendingRequests.entries()) {
 if(pendingRequest.requestIds.includes(messageId)) {
  // Add this response to the collection
   pendingRequest.responseMessages.push(message);

  // If we've collected all responses for this batch, resolve the promise
  if(
    pendingRequest.responseMessages.length ===
    pendingRequest.requestIds.length
   ) {
    pendingRequest.resolve(pendingRequest.responseMessages);
   }

  break;
  }
 }
}

關閉服務

Streamable HTTP / SSE 以及我們自定義的 Rest Transport，關閉服務的邏輯都是一致的。

因為服務啟動時是啟動了一個 HTTP 服務器，所以服務關閉時，只需要關閉這個 HTTP 服務器即可。

Rest Server Transport 實現的關閉服務的主要邏輯：

asyncstopServer():Promise {
if(this._httpServer) {
 returnnewPromise((resolve, reject) =>{
  this._httpServer!.close((err) =>{
   if(err) {
     reject(err);
    return;
    }
   this._server =null;
   this._httpServer =null;
    resolve();
   });
  });
 }
}

在實現完這個自定義的傳輸后，我們就可以把代碼打包，發布到 npm 公共倉庫。這樣第三方 MCP 服務器都可以使用這個 Transport 來改造自己的實現。

比如，我用 typescript 實現的Rest Server Transport封裝在@chatmcp/sdk這個 npm 包里面，第三方服務器可以這樣引入：

import{ RestServerTransport }from"@chatmcp/sdk/server/index.js";

改造 MCP 服務器

為了把第三方 MCP 服務器部署到云端，支持多租戶并發調用，我們需要對第三方 MCP 服務器做一定的改造。

以Perplexity Ask Server這個 MCP 服務器為例，來演示具體的改造流程：

安裝包含 Rest Server Transport 的 SDK

npm install @chatmcp/sdk

獲取服務啟動參數

通過@chatmcp/sdk提供的getParamValue方法，可以獲得 MCP 服務器啟動時候的參數，從命令行讀取，或者從環境變量讀取：

import { getParamValue } from "@chatmcp/sdk/utils/index.js";

const perplexityApiKey = getParamValue("perplexity_api_key") || "";

const mode = getParamValue("mode") || "stdio";
const port = getParamValue("port") || 9593;
const endpoint = getParamValue("endpoint") || "/rest";

獲取請求參數

MCP 服務器的設計初衷，是讓用戶運行在本地，跟私有數據打交道。

所以目前絕大部分的 MCP 服務器，都是不支持多租戶的（也就是不能讓多個用戶共同使用）

主要的限制在于，在 MCP 服務器內部的實現邏輯里面，用到鑒權參數的地方，是在服務器啟動時獲取的，不是在請求時動態獲取的。

如果要在云端部署 MCP 服務器，支持多租戶使用，我們需要修改 MCP 服務器內部的參數獲取邏輯，改成從請求參數里面動態獲取。

MCP 協議允許在每個請求的 Request 對象通過 _meta 字段傳遞自定義的數據。

那么就可以改造 MCP 服務器，從 request.params._meta 里面獲取客戶端傳遞的 auth 參數。

在@chatmcp/sdk中，實現了一個getAuthValue方法，可以讓 MCP 服務器獲取 MCP 客戶端傳遞的鑒權參數。

const auth: any = request.params?._meta?.auth;

以這個Perplexity Ask Server為例，我們把讀取啟動時參數，改成讀取請求時參數：

// before: get params from env and set as global params

// after: get params from env or command line, set as global params
import{ getParamValue, getAuthValue }from"@chatmcp/sdk/utils/index.js";

constperplexityApiKey = getParamValue("perplexity_api_key") ||"";

constmode = getParamValue("mode") ||"stdio";
constport = getParamValue("port") ||9593;
constendpoint = getParamValue("endpoint") ||"/rest";

server.setRequestHandler(CallToolRequestSchema,async(request) => {
try{
 // before: use global params

 // after: get auth params from request, use global params if request params not set
 constapiKey =
   getAuthValue(request,"PERPLEXITY_API_KEY") || perplexityApiKey;
 if(!apiKey) {
  thrownewError("PERPLEXITY_API_KEY not set");
  }

 const{ name,arguments: args } = request.params;
 if(!args) {
  thrownewError("No arguments provided");
  }
 switch(name) {
  case"perplexity_ask": {
   if(!Array.isArray(args.messages)) {
    thrownewError(
     "Invalid arguments for perplexity_ask: 'messages' must be an array"
     );
    }

   constmessages = args.messages;

   // before: use global params in every function
   // const result = await performChatCompletion(
   //  messages,
   //  "sonar-pro"
   // );

   // after: pass params to every function
   constresult =awaitperformChatCompletion(
     apiKey,
     messages,
    "sonar-pro"
    );

   return{
     content: [{type:"text", text: result }],
     isError:false,
    };
   }
  // ...
  }
 }catch(error) {
 // Return error details in the response
 return{
   content: [
    {
    type:"text",
     text:`Error:${
      errorinstanceofError? error.message :String(error)
     }`,
    },
   ],
   isError:true,
  };
 }
});

改造完之后，云端的 MCP 代理，就可以把用戶設置的鑒權參數：perplexity_api_key，通過這種方式傳給 MCP 服務器了：

request.params._meta.auth = {
 perplexity_api_key:"xxx",
};

新增 Rest 傳輸

修改這個 MCP 服務器，在原來僅支持 stdio 傳輸的基礎上，新增一個 rest http 傳輸：

import{ RestServerTransport }from"@chatmcp/sdk/server/rest.js";
import{ getParamValue }from"@chatmcp/sdk/utils/index.js";

constmode = getParamValue("mode") ||"stdio";
constport = getParamValue("port") ||9593;
constendpoint = getParamValue("endpoint") ||"/rest";

asyncfunctionrunServer(){
try{
 // after: MCP Server run with rest transport and stdio transport
 if(mode ==="rest") {
  consttransport =newRestServerTransport({
    port,
    endpoint,
   });
  awaitserver.connect(transport);

  awaittransport.startServer();

  return;
  }

 // before: MCP Server only run with stdio transport
 consttransport =newStdioServerTransport();
 awaitserver.connect(transport);
 console.error(
  "Perplexity MCP Server running on stdio with Ask, Research, and Reason tools"
  );
 }catch(error) {
 console.error("Fatal error running server:", error);
  process.exit(1);
 }
}

改造之后，這個 MCP 服務器就支持兩種傳輸機制了。在不同的場景下，使用不同的傳輸機制運行：

使用 MCP 服務器

本地使用

通過源代碼運行：

export PERPLEXITY_API_KEY=xxx && node build/index.js

或者使用二進制運行：

PERPLEXITY_API_KEY=xxx npx -y server-perplexity-ask

或者使用 docker 運行：

docker run -i --rm -e PERPLEXITY_API_KEY=xxx mcp/perplexity-ask

這三種方式，都會在本地啟動 stdio 通信，在服務器啟動的時候傳遞 PERPLEXITY_API_KEY 參數。

云端調用

改造完的 MCP 服務器在云端部署的啟動命令：

node build/index.js --mode=rest --port=8081 --endpoint=/rest

使用了自定義的 Rest 傳輸，啟動 HTTP 服務器，監聽在本地的 8081 端口。通過K8S Service創建一個內網訪問域名：perplexity-svc

MCP 代理把客戶端的請求轉發到http://perplexity-svc:8081/rest，就可以處理多個用戶的并發請求了。

MCP 客戶端僅需要配置一個 MCP 代理的 URL，就可以使用 MCP 服務器了：

總結

在本節中，我們詳細講解了 MCP 協議支持的三種標準傳輸機制以及自定義傳輸的實現方式：

stdio 傳輸：

通過標準輸入/輸出進行本地進程間通信

優勢在于實現簡單、通信速度快、安全性高

主要適用于本地數據訪問場景

局限于單進程通信，資源開銷較大

SSE 傳輸（即將廢棄）：

基于 HTTP 協議，支持遠程資源訪問

使用雙通道響應機制（SSE 連接 + POST 端點）

存在連接不穩定、擴展性差、瀏覽器限制等問題

不適合無服務器(serverless)環境和云原生架構

Streamable HTTP 傳輸：

替代 SSE 的新傳輸機制，兼容現代云架構

更靈活的連接模式，支持簡單請求-響應和流式傳輸

標準化的會話管理和斷點恢復功能

適合遠程訪問、無服務器環境和大規模部署

自定義傳輸機制：

MCP 協議支持實現自定義傳輸以滿足特定需求

可以針對特定部署環境和使用場景進行優化

示例中實現了一個 Rest Transport，適合無狀態、短連接場景

傳輸機制的選擇應基于具體應用場景：

本地數據訪問優先選擇 stdio 傳輸

遠程資源訪問優先選擇 Streamable HTTP 傳輸

特殊需求場景可考慮實現自定義傳輸

MCP 協議的可插拔傳輸架構使其能夠靈活適應不同的部署環境和通信需求，從簡單的本地工具到復雜的云服務均可支持。隨著技術發展，傳輸機制也在不斷優化，以提供更好的性能、可靠性和可擴展性。