來源：艾逗筆

介紹 MCP 傳輸機制

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

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

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

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

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

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

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

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

stdio 傳輸

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

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

在 stdio 傳輸中：

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

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

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

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

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

服務器不得向 stdout 寫入無效的 MCP 消息內(nèi)容。

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

stdio 通信流程

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

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

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

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

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

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

服務器關閉自身的 stdout

stdio 傳輸實現(xiàn)

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

啟動 MCP 服務器

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

npx -y mcp-server-time

創(chuàng)建 stdio 通信管道

MCP 服務器啟動時，會創(chuàng)建 stdio 通信管道（pipeline），用于跟 MCP 客戶端進行消息通信。在 MCP 客戶端發(fā)送關閉信號，或者 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 客戶端把消息發(fā)到通信管道。MCP 服務器通過標準輸入 stdin 讀取客戶端發(fā)送的消息，以換行符： 作為讀取完成標識。

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 服務器運行完內(nèi)部邏輯，需要給 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 服務器的響應內(nèi)容。

關閉 stdio 通信管道

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

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

stdio 通信管道關閉之后，MCP 客戶端與 MCP 服務器之間不能再相互發(fā)送消息，直到再次建立 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 傳輸?shù)睦?/p>

stdio 傳輸機制主要依靠本地進程通信實現(xiàn)。

主要的優(yōu)勢是：

無外部依賴，實現(xiàn)簡單

無網(wǎng)絡傳輸，通信速度快

本地通信，安全性高

也有一些局限性：

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

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

stdio 傳輸?shù)倪m用場景

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

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

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

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

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

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

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

當然有?？梢允褂?SSE 傳輸。

SSE 傳輸

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

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

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

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

一個常規(guī) HTTP POST 端點，供客戶端向服務器發(fā)送消息

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

服務器消息作為 SSE 消息事件發(fā)送，消息內(nèi)容以 JSON 格式編碼在事件數(shù)據(jù)中。

SSE 通信流程

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

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

客戶端給消息端點發(fā)送消息

服務器給客戶端響應消息已接收狀態(tài)碼

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

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

客戶端關閉 SSE 連接

SSE 傳輸實現(xiàn)

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

啟動 MCP 服務器

系統(tǒng)管理員在遠程服務器（也可以是本地電腦）輸入命令，啟動 MCP 服務器，監(jiān)聽服務器端口，對外暴露 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 服務，監(jiān)聽在 3001 端口，對外暴露了兩個端點：

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

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

解析一個 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 請求把消息發(fā)到這個消息通信地址，與 MCP 服務器進行消息交互。

消息交互

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

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

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

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 客戶端發(fā)送的消息：

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 狀態(tài)碼，告知 MCP 客戶端，請求已收到。

然后 MCP 服務器運行內(nèi)部邏輯，完成業(yè)務功能。

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

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

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

`,
 );
}

MCP 客戶端根據(jù) MCP 服務器同步響應的 2** 狀態(tài)碼，判斷 MCP 服務器已經(jīng)接到請求，并開始讀取 MCP 服務器發(fā)到 SSE 連接的消息內(nèi)容。

MCP 客戶端從與 MCP 服務器建立的 SSE 連接中讀取event: message事件消息，獲得 MCP 服務器發(fā)送的業(yè)務數(shù)據(jù)data。

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

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

斷開 SSE 連接

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

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

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

一個用 go 實現(xiàn)的心跳檢測和超時關閉示例：

// 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 傳輸時，服務器一方需要實現(xiàn)一些必要的安全防護措施：

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

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

服務器應對所有連接實施適當?shù)纳矸蒡炞C

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

SSE 傳輸?shù)倪m用場景

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

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

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

區(qū)別在于：

用 SSE 傳輸，MCP 客戶端直接與 MCP 服務器通信，而不用通過本地的 stdio 傳輸調(diào)用 API 進行中轉(zhuǎn)。

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

SSE 傳輸?shù)睦?/p>

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

SSE 傳輸?shù)闹饕獌?yōu)勢：

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

基于標準 HTTP 協(xié)議實現(xiàn)，兼容性好，便于與現(xiàn)有 Web 基礎設施集成

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

相比 WebSocket 實現(xiàn)簡單，是普通 HTTP 的擴展，不需要協(xié)議升級

SSE 傳輸?shù)闹饕觿菖c問題：

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

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

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

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

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

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

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

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

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

Streamable HTTP 傳輸

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

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

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

Streamable HTTP 通信流程

客戶端給服務器的通信端點發(fā)消息

服務器給客戶端響應消息

客戶端根據(jù)服務器的響應類型，繼續(xù)給服務器發(fā)消息

服務器繼續(xù)響應客戶端消息

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

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

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

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

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

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

客戶端發(fā)消息時候需要帶上狀態(tài)標識：Mcp-Session-Id

Streamable HTTP 傳輸實現(xiàn)

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

啟動服務器

跟 SSE 傳輸機制一樣，Streamable HTTP 傳輸本質(zhì)上也是基于 HTTP 協(xié)議通信，需要先啟動一個 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 服務，監(jiān)聽在 3002 端口，對外暴露了一個端點：

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

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

消息交互

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

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

MCP 客戶端可以使用 GET 或者 POST 請求給服務器發(fā)消息，每個請求必須設置請求頭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 客戶端與服務器通信的幾個要點：

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

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

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

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

會話保持

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

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

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

使用 Streamable HTTP 傳輸?shù)姆掌骺梢栽诔跏蓟瘯r分配一個會話 ID，方法是在包含InitializeResult的 HTTP 響應中包含它，放在Mcp-Session-Id頭中。

如果服務器在初始化期間返回了Mcp-Session-Id，使用 Streamable HTTP 傳輸?shù)目蛻舳吮仨氃谒泻罄m(xù)的 HTTP 請求中在Mcp-Session-Id頭中包含它。

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

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

不再需要特定會話的客戶端應該發(fā)送一個帶有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 傳輸斷開連接的方式一致。

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

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

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

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

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

Streamable HTTP 傳輸?shù)睦?/p>

Streamable HTTP 傳輸機制結合了 SSE 傳輸?shù)倪h程訪問能力和無狀態(tài) HTTP 的靈活性，同時解決了 SSE 傳輸中的許多問題。

主要優(yōu)勢：

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

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

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

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

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

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

主要劣勢：

相比單純的 stdio 傳輸實現(xiàn)復雜度更高

仍需處理網(wǎng)絡連接斷開和恢復的邏輯

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

Streamable HTTP 傳輸?shù)倪m用場景

Streamable HTTP 傳輸適用于：

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

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

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

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

需要在不穩(wěn)定網(wǎng)絡環(huán)境中保持可靠通信的場景

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

自定義傳輸

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

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

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

需求分析

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

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

大致的流程是：

這個 MCP 代理服務以 HTTP 的形式提供接入，需要支持并發(fā)調(diào)用。這個 MCP 代理服務作為 stdio 傳輸或者 SSE 傳輸?shù)目蛻舳耍l(fā)送消息給后臺的 MCP 服務器。

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

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

如果后臺對接的 MCP 服務器是基于 SSE 傳輸實現(xiàn)的，從前面對 SSE 傳輸?shù)姆治鑫覀冎?，MCP 代理需要跟后臺部署的 MCP 服務器之間保持一個 SSE 連接。如果每個用戶請求都需要維持一個 SSE 的長連接，對服務器也是不小的壓力。

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

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

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

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

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

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

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

基于 HTTP 通信

非流式傳輸

無狀態(tài)，無會話保持

短連接

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

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

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

實現(xiàn) Rest Server Transport

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

定義新的傳輸類，實現(xiàn) MCP 協(xié)議的 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 服務器監(jiān)聽的端口，都可以在啟動服務器的時候自定義。

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

接收客戶端請求

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

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

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

Rest Server Transport 處理用戶請求的主要實現(xiàn)邏輯：

/**
* 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);
 }
}

服務器發(fā)送消息

MCP 服務器接到客戶端請求后，把請求數(shù)據(jù)發(fā)到內(nèi)部實現(xiàn)的各個功能函數(shù)內(nèi)，得到響應內(nèi)容后，用 JSON-RPC 編碼，響應給 MCP 客戶端。

Rest Server Transport 發(fā)送消息的主要實現(xiàn)邏輯：

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 實現(xiàn)的關閉服務的主要邏輯：

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

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

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

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

改造 MCP 服務器

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

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

安裝包含 Rest Server Transport 的 SDK

npm install @chatmcp/sdk

獲取服務啟動參數(shù)

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

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";

獲取請求參數(shù)

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

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

主要的限制在于，在 MCP 服務器內(nèi)部的實現(xiàn)邏輯里面，用到鑒權參數(shù)的地方，是在服務器啟動時獲取的，不是在請求時動態(tài)獲取的。

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

MCP 協(xié)議允許在每個請求的 Request 對象通過 _meta 字段傳遞自定義的數(shù)據(jù)。

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

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

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

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

// 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 代理，就可以把用戶設置的鑒權參數(shù)：perplexity_api_key，通過這種方式傳給 MCP 服務器了：

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

新增 Rest 傳輸

修改這個 MCP 服務器，在原來僅支持 stdio 傳輸?shù)幕A上，新增一個 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 參數(shù)。

云端調(diào)用

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

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

使用了自定義的 Rest 傳輸，啟動 HTTP 服務器，監(jiān)聽在本地的 8081 端口。通過K8S Service創(chuàng)建一個內(nèi)網(wǎng)訪問域名：perplexity-svc

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

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

總結

在本節(jié)中，我們詳細講解了 MCP 協(xié)議支持的三種標準傳輸機制以及自定義傳輸?shù)膶崿F(xiàn)方式：

stdio 傳輸：

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

優(yōu)勢在于實現(xiàn)簡單、通信速度快、安全性高

主要適用于本地數(shù)據(jù)訪問場景

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

SSE 傳輸（即將廢棄）：

基于 HTTP 協(xié)議，支持遠程資源訪問

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

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

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

Streamable HTTP 傳輸：

替代 SSE 的新傳輸機制，兼容現(xiàn)代云架構

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

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

適合遠程訪問、無服務器環(huán)境和大規(guī)模部署

自定義傳輸機制：

MCP 協(xié)議支持實現(xiàn)自定義傳輸以滿足特定需求

可以針對特定部署環(huán)境和使用場景進行優(yōu)化

示例中實現(xiàn)了一個 Rest Transport，適合無狀態(tài)、短連接場景

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

本地數(shù)據(jù)訪問優(yōu)先選擇 stdio 傳輸

遠程資源訪問優(yōu)先選擇 Streamable HTTP 傳輸

特殊需求場景可考慮實現(xiàn)自定義傳輸

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