深度剖析 MCP 三大核心通信模式 STDIO、SSE 與 Streamable HTTP

MCP（Model Context Protocol）模型上下文協議提供三種核心的通信模式：STDIO、SSE 和 Streamable HTTP，在不同的場景中爲開發者提供了靈活且高效的數據傳輸方式。今天，我們將一起深入分析這三種模式，探討它們的應用場景與差異。

STDIO 模式

STDIO（Standard Input and Output） 是最基本的輸入輸出方式，廣泛應用於命令行工具、腳本編程以及本地調試過程中。它通過標準輸入、輸出和錯誤流來進行數據的傳輸。

• 標準輸入（stdin）

  ：接收用戶或系統輸入的數據

• 標準輸出（stdout）

  ：將處理後的數據輸出到終端或文件

• 標準錯誤（stderr）

  ：輸出錯誤信息，便於調試

適用場景：面向傳統命令行式交互，強調順序性與確定性，通常用於命令行工具、自動化腳本等短期交互場景。它是最基礎的數據傳輸方式，適合單次的、低延遲的交互任務，適合本地開發與調試。

from mcp.server.fastmcp import FastMCP
# 初始化FastMCP server，創建一個MCP服務器實例
mcp = FastMCP("mcpserver")
# @MCP.tool()裝飾器將這個函數註冊爲MCP工具
@mcp.tool()
def add(x: int, y: int) -> int:
    return x * y
if __name__ == "__main__":
    mcp.run(transport="stdio")

SSE 模式

SSE（Server-Sent Events） 是一種基於 HTTP 協議的單向數據流傳輸方式。它允許服務器主動向客戶端推送實時數據。SSE 通過保持一個持久的 HTTP 連接，將數據流式傳輸到客戶端，特別適合需要持續更新的實時場景。

• 單向傳輸：

   服務器向客戶端推送數據。

• 自動重連：

   如果連接中斷，客戶端會自動重新連接。

• 基於 HTTP 協議：

   與 Web 應用的兼容性良好。

適用場景： SSE 非常適合實時推送應用，如新聞更新、社交媒體通知、股票行情等。它適用於服務器向多個客戶端實時推送數據的場景。

• 1 個長連接：客戶端通過 SSE 建立服務端到客戶端的單向消息流

• 2 個接口：

  • GET /sse：用於建立長連接並接收服務端消息
  • POST /message：客戶端發送請求、響應或通知

from flask import Flask, Response, request
app = Flask(__name__)
# SSE事件流端點 (核心功能)
@app.route('/sse')
def sse_endpoint():
    def event_stream():
        # 1. 發送POST端點地址 (MCP規範要求)
        yield "event: endpoint\ndata: /post\n\n"
        # 2. 發送示例消息
        for i in range(3):
            yield f"data: 消息 {i+1}\n\n"
            time.sleep(1)
    return Response(event_stream(), mimetype='text/event-stream')
# 消息接收端點 (核心功能)
@app.route('/messages', methods=['POST'])
def receive_message():
    print(f"收到客戶端消息: {request.data.decode()}")
    return "OK"
if __name__ == '__main__':
    # 綁定本地地址 (安全要求)
    app.run(host='127.0.0.1', port=5000)

Streamable HTTP 模式

Streamable HTTP（是一種基於 HTTP 協議的流式傳輸技術，專門用於大文件（如視頻、音頻）的分段傳輸。與 SSE 不同，Streamable HTTP 允許文件在傳輸的同時被處理，使客戶端可以邊接收數據邊處理，避免等待整個文件加載完成。

• 分塊傳輸：

   文件被切割成若干小塊，逐步傳輸。

• 適合大文件：

   主要用於視頻、音頻、圖像等多媒體文件的流式傳輸。

• 低延遲：

   數據可以在客戶端接收到部分內容時立即開始處理。

適用場景： Streamable HTTP 非常適合大文件的流式傳輸，尤其在在線視頻、音頻流和其他大數據量內容的傳輸中，能極大地提高用戶體驗。

from flask import Flask, request, Response
import json
app = Flask(__name__)
@app.route('/add', methods=['POST'])
def stream_add():
    # 流式處理函數
    def generate():
        # 1. 逐行讀取流式輸入
        for line in request.stream:
            try:
                # 2. 解析JSON數據
                data = json.loads(line)
                a = data['a']
                b = data['b']
                # 3. 計算並返回結果（流式輸出）
                result = a + b
                yield f"data: {json.dumps({'result': result})}\n\n"
            except:
                yield "data: {\"error\": \"無效輸入\"}\n\n"
    # 4. 返回SSE流響應
    return Response(generate(), mimetype='text/event-stream')
if __name__ == '__main__':
    app.run(host='127.0.0.1', port=5000)

三大通信模式差異

• STDIO 模式：適合快速開發和本地調試，但不適合發佈生產環境
• SSE 模式：適合需要實時推送的遠程服務
• Streamable HTTP 模式：MCP 的未來方向，尤其在雲原生和無狀態架構中具有顯著優勢

本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/QUg4RXMZBlaRQYR03sNeJw