使用 Streamable HTTP 前後的數據比對

MCP（Model Context Protocol）協議是一個用於 AI 模型和工具之間通信的標準協議。隨着 AI 應用變得越來越複雜並被廣泛部署，原有的通信機制面臨着一系列挑戰。近期 MCP 倉庫的 PR #206 引入了一個全新的 Streamable HTTP 傳輸層替代原有的 HTTP+SSE 傳輸層。本文將詳細分析該協議的技術細節和優勢。

要點速覽

• Streamable HTTP 相比 HTTP + SSE 具有更好的穩定性，在高併發場景下表現更優。

• Streamable HTTP 在性能方面相比 HTTP + SSE 具有明顯優勢，響應時間更短且更穩定。

• Streamable HTTP 客戶端實現相比 HTTP + SSE 更簡單，代碼量更少，維護成本更低。

01 爲什麼選擇 Streamable HTTP?

HTTP + SSE 存在的問題

HTTP+SSE 的傳輸過程實現中，客戶端和服務器通過兩個主要渠道進行通信：

• HTTP 請求 / 響應：客戶端通過標準的 HTTP 請求向服務器發送消息。 

• 服務器發送事件（SSE）：服務器通過專門的 /sse 端點向客戶端推送消息。

這就導致存在下面三個問題：

• 服務器必須維護長連接，在高併發情況下會導致顯著的資源消耗。

• 服務器消息只能通過 SSE 傳遞，造成了不必要的複雜性和開銷。

• 基礎架構兼容性，許多現有的網絡基礎架構可能無法正確處理長期的 SSE 連接。企業防火牆可能會強制終止超時連接，導致服務不可靠。

Streamable HTTP 的改進

Streamable HTTP 是 MCP 協議的一次重要升級，通過下面的改進解決了原有 HTTP + SSE 傳輸方式的多個關鍵問題：

• 統一端點：移除了專門建立連接的 /sse 端點，將所有通信整合到統一的端點。

• 按需流式傳輸：服務器可以靈活選擇返回標準 HTTP 響應或通過 SSE 流式返回。

• 狀態管理：引入 session 機制以支持狀態管理和恢復。

下面通過實際應用場景中穩定性，性能和客戶端複雜度三個角度對比說明 Streamable HTTP 相比 HTTP + SSE 的優勢，AI 網關 Higress 目前已經支持了 Streamable HTTP 協議，通過 MCP 官方 Python SDK 的樣例 Server 部署了一個 HTTP + SSE 協議的 MCP Server，通過 Higress 部署了一個 Streamable HTTP 協議的 MCP Server。

02 穩定性對比

TCP 連接數對比

利用 Python 程序模擬 1000 個用戶同時併發訪問遠程的 MCP Server 並調用獲取工具列表，圖中可以看出 SSE Server 的 SSE 連接無法複用且需要長期維護，高併發的需求也會帶來 TCP 連接數的突增，而 Streamable HTTP 協議則可以直接返回響應，多個請求可以複用同一個 TCP 連接，TCP 連接數最高只到幾十條，並且整體執行時間也只有 SSE Server 的四分之一。 

在 1000 個併發用戶的測試場景下，Higress 部署的 Streamable HTTP 方案的 TCP 連接數明顯低於 HTTP + SSE 方案：

• HTTP + SSE：需要維持大量長連接，TCP 連接數隨時間持續增長。

• Streamable HTTP：按需建立連接，TCP 連接數維持在較低水平。

請求成功率對比

實際應用場景中進程級別通常會限制最大連接數，Linux 默認通常是 1024。利用 Python 程序模擬不同數量的用戶訪問遠程的 MCP Server 並調用獲取工具列表，SSE Server 在併發請求數到達最大連接數限制後，成功率會極速下降，大量的併發請求無法建立新的 SSE 連接而訪問失敗。

在不同併發用戶數下的請求成功率測試中，Higress 部署的 Streamable HTTP 的成功率顯著高於 HTTP + SSE 方案：

• HTTP + SSE：隨着併發用戶數增加，成功率顯著下降

• Streamable HTTP：即使在高併發場景下仍能保持較高的請求成功率

性能對比

社區  Python 版：

https://github.com/modelcontextprotocol/servers/tree/main/src/github

Higress 版：

https://mcp.higress.ai/server/server0011

利用 Python 程序模擬不同數量的用戶同時併發訪問遠程的 MCP Server 並調用獲取工具列表，並統計調用返回響應的時間，圖中給出的響應時間對比爲對數刻度，SSE Server 在併發用戶數量較多時平均響應時間會從 0.0018s 顯著增加到 1.5112s，而 Higress 部署的 Streamable HTTP Server 則依然維持在 0.0075s 的響應時間，也得益於 Higress 生產級的性能相比於 Python Starlette 框架。

性能測試結果顯示，Higress 部署的 Streamable HTTP 在響應時間方面具有明顯優勢：

• Streamable HTTP 的平均響應時間更短，響應時間波動較小，隨併發用戶數增加，響應時間增長更平

• HTTP + SSE 的平均響應時間更長，在高併發場景下響應時間波動較大

03 客戶端複雜度對比

Streamable HTTP 支持無狀態的服務和有狀態的服務，目前的大部分場景無狀態的 Streamable HTTP 的可以解決，通過對比兩種傳輸方案的客戶端實現代碼，可以直觀地看到無狀態的 Streamable HTTP 的客戶端實現簡潔性。

HTTP + SSE 客戶端樣例代碼

class SSEClient:
    def __init__(self, url: str, headers: dict = None):
        self.url = url
        self.headers = headers or {}
        self.event_source = None
        self.endpoint = None

        asyncdef connect(self):
            # 1. 建立 SSE 連接
            asyncwith aiohttp.ClientSession(headers=self.headers) as session:
                self.event_source = await session.get(self.url)

                # 2. 處理連接事件
                print('SSE connection established')

                # 3. 處理消息事件
                asyncfor line in self.event_source.content:
                    if line:
                        message = json.loads(line)
                        await self.handle_message(message)

                        # 4. 處理錯誤和重連
                        if self.event_source.status != 200:
                            print(f'SSE error: {self.event_source.status}')
                            await self.reconnect()

        asyncdef send(self, message: dict):
            # 需要額外的 POST 請求發送消息
            asyncwith aiohttp.ClientSession(headers=self.headers) as session:
                asyncwith session.post(self.endpoint, json=message) as response:
                    returnawait response.json()

                asyncdef handle_message(self, message: dict):
                    # 處理接收到的消息
                    print(f'Received message: {message}')

    asyncdef reconnect(self):
        # 實現重連邏輯
        print('Attempting to reconnect...')
        await self.connect()

Streamable HTTP 客戶端樣例代碼

class StreamableHTTPClient:
    def __init__(self, url: str, headers: dict = None):
        self.url = url
        self.headers = headers or {}
        
    async def send(self, message: dict):
        # 1. 發送 POST 請求
        async with aiohttp.ClientSession(headers=self.headers) as session:
            async with session.post( self.url, json=message,
                headers={'Content-Type': 'application/json'}
            ) as response:
                # 2. 處理響應
                if response.status == 200:
                    return await response.json()
                else:
                    raise Exception(f'HTTP error: {response.status}')

從代碼對比可以看出：

• 複雜度：Streamable HTTP 無需處理連接維護、重連等複雜邏輯。

• 可維護性：Streamable HTTP 代碼結構更清晰，更易於維護和調試。

• 錯誤處理：Streamable HTTP 的錯誤處理更直接，無需考慮連接狀態。

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