深度剖析 MCP SDK 最新版： Streamable HTTP 模式正式發佈，爲你實測揭祕

最近，MCP SDK 新版本更新發布（最新爲 v1.9.0），其中最大的更新莫過於終於提供了新版協議中的傳輸模式 — streamable HTTP。不過由於 MCP SDK 的文檔一直以來” 語焉不詳 “的風格，很多開發者知其然卻不知其所以然，很容易在應用中踩坑。

本文將對這種模式進行全面剖析與實測，幫助大家深入認識這種新的模式。

• 快速上手：開啓 streamable HTTP

• 深入兩個核心參數

• 認識 session-id

• 樣例實測與驗證

• Low-Level API 開發服務端

• 解讀多應用實例模式

01

快速上手：開啓 streamable HTTP

從版本 SDK 1.8.0 開始，MCP 支持三種傳輸模式：stdio、sse 與 streamable-http。這三種模式的服務端與客戶端必須匹配使用，即 streamable HTTP 的客戶端無法與 sse 模式服務端交互，所以在一些客戶端工具中配置 MCP 時，不要盲目採用新模式。

【服務端】

服務端開啓 streamable HTTP 模式仍然建議藉助 FastMCP 高層 API。方法如下：

# 創建FastMCP實例
app = FastMCP(
    ,
    port=5050,
    stateless_http=False,
    json_response=False,
    streamable_http_path="/mcp"
)

....工具....

if __name__ == '__main__':
    app.run(transport='streamable-http')

主要變化來自三個參數：其中 transport 參數增加了 streamable-http 選項；而 stateless_http 和 json_response 則控制了不同的工作模式（默認都爲 False）。

【客戶端】

對應的客戶端代碼修改如下，注意這裏的 server_url 端點默認爲 / mcp，比如：

http://localhost:5050/mcp

try:
        async with streamablehttp_client(url=server_url) as (read, write, get_session_id):
            async with ClientSession(read, write) as session:
                print(f"連接成功!")
                
                # 初始化會話
                await session.initialize()
                print("會話初始化完成")

                # 獲取會話ID
                session_id = get_session_id()
                print(f"會話ID: {session_id}")
     ......

客戶端的主要變化包括：

• 需要使用 streamablehttp_client 這一客戶端傳輸模塊

• 新增了可回調的 get_session_id 方法，用來獲取 session-id

經過 FastMCP 的精心 “包裝”，streamable HTTP 的使用看上去挺簡潔，但背後的實現要比 SSE 更復雜（這也讓很多人質疑爲什麼不直接使用 WebSocket）。

讓我們來逐步深入。

02

深入兩個核心參數

首先看到 stateless_http 與 json_response 這兩個服務端的核心控制參數。

還記得 SSE 模式中的 “僞雙工” 通信嗎：Post 通道用作客戶端到服務端的 JSON-RPC 消息傳輸，SSE 通道則用作服務端到客戶端的 JSON-RPC 消息傳輸：

在 streamable HTTP 模式下的規則變爲：

• Post 通道：用作客戶端到服務端的所有消息傳輸；同時服務端對客戶端的響應消息也通過 Post 通道直接返回（基本類似 Restful HTTP Post 交互）。響應形式可能是 SSE 流，也可能是 JSON 數據。

• SSE 通道：變爲可選的通道，僅用作服務端發往客戶端的通知（Notification，比如進度更新）、請求（Request，比如 Sampling 請求）。很顯然這裏的形式只能是 SSE 流。（實際還有一種 “會話恢復” 功能會用到 SSE 通道）

那麼什麼時候會有 SSE 通道；Post 的響應形式怎麼確定？這就是上面兩個控制參數的作用。這兩個參數的不同組合，產生的效果用下圖簡潔的來表示：

關於這兩個參數，你只需瞭解這幾點：

【stateless_http】

• 控制是否會建立長連接的 SSE 通道。爲 False 時則會開啓 SSE 通道。

• 控制是否會管理客戶端的會話。爲 False 時會對客戶端會話進行管理。

【json_response】

• 控制 Post 請求響應形式是否用 JSON。默認爲 False，會使用 SSE 流式響應（注意並不是走 SSE 通道）。

• 客戶端必須在 Post 請求頭中聲明可同時接收兩種形式的響應。即：

  "Accept": "application/json, text/event-stream"

會話恢復功能

此外，當 stateless_http 與 json_response 同時爲 False 時，可具備一項額外的能力：會話恢復。即：服務端返回的事件流具備 “斷點” 續傳的功能。該功能需要自行實現服務端事件的持久化接口，且完善性有待驗證，本文暫不對此深入。

03

認識 session-id

session-id 是 streamable HTTP 模式下服務端用來跟蹤與管理客戶端會話使用。客戶端可以使用連接時獲得的回調方法 get_session_id 來獲得。

關於 session-id，你需要了解的有：   

• 什麼時候有：只有當 stateless_http=False 時纔會有 session-id

• 什麼時候生成：在發起初始化請求時服務端生成並在 headers 中返回

• 客戶端如何使用：後續所有的客戶端發起的請求會自動攜帶該 session-id；你也可開發其他用途，比如用來關聯一組與 MCP 服務的多次交互

• 什麼時候失效：客戶端退出 streamablehttp_client 的上下文區域時會自動觸發一個 HTTP Delete 請求，服務端會刪除會話，session-id 失效

• 在服務端的作用：後續的請求無需每次建立新的連接與會話，而是從一個 “池” 裏查詢出已經創建的會話模塊用來處理請求

04

樣例實測與驗證

我們用一個例子來體驗 streamable HTTP 的效果以加強認識。首先我們在服務端中創建一個簡單的工具，並使用 streamable-http 模式啓動：

@app.tool(name='hello')
async def hello(ctx: Context, name: str) -> str:

    steps = 10
    await ctx.report_progress(0.0, steps, 'MCP Server正在處理請求...')

    # 模擬計算過程的多個步驟
    for step in range(1, steps + 1):
        await asyncio.sleep(1)
        logger.info(f"正在處理第{step}步，發送進度通知...")
        await ctx.report_progress(float(step), float(steps),f'正在處理第{step}步...')

    await ctx.report_progress(steps, steps, 'MCP Server請求處理完成！')

    return f'Hello,{name}'

這個工具模擬一個長時間處理的任務，並在任務過程中定期報告進度，用來模擬服務端發送通知消息（Notification）的過程。

客戶端用一個自己編寫的簡單交互式命令行程序：

【服務端：stateless_http=Fase，json_response=False】

在這種服務端模式下，我們啓動客戶端，調用 “hello” 工具，過程如下圖。

可以看到，能夠獲取到服務端生成的 session-id，而且可以接收到服務端發送的進度通知（這裏用進度條做美化）。這表示該模式下成功建立了 SSE 通道。注意：服務端的通知消息一定是通過 SSE 通道以流的形式發送。

接着調用另外一個工具，過程如下圖。可以看到 session-id 沒有變化：

現在我們選擇主菜單中的 “開啓新的會話” 功能。該功能會退出 streamablehttp_client 管理的上下文，並重新進入：

此時再次測試上面的工具，就會看到 session-id 已經發生變化：

同時觀察服務端的日誌輸出，會發現如下圖所示的信息。圖中信息的含義是：

服務端先接收到了一個 DELETE 請求，終止了一個會話；然後接收到新的 POST 請求（初始化），開始創建了一個新的 session-id 與傳輸模塊（transport）：

【服務端：stateless_http=True，json_response=False】

現在我們開啓服務端的無狀態模式，其他不變。我們測試相同的 “hello” 工具，此時發現，儘管仍能夠獲得最終結果，但無法接收到進度通知：

觀察服務端後臺的日誌，可以看到如下圖的提示信息。該信息表明，沒有找到對應的發送流（stream，服務端的一種內部通信機制。這裏的 "_GET_stream" 是獨立的 SSE 通道使用的流名稱）。這表明無狀態模式下不會建立 SSE 通道。

再次強調：服務端發起的通知與請求消息都只會通過獨立的 SSE 通道進行，而客戶端 Post 請求的響應也不會 “借用”SSE 通道（哪怕是流形式的響應）。

05

完整的通信流程

用下圖來整理 streamable HTTP 模式下客戶端與服務端的交互過程。其中：

• 紅色代表 SSE 通道的交互；其他都是 HTTP Post/Get/Delete。

• 虛線代表只有在有狀態模式下才會有出現的交互或信息。

1. 連接：在 streamable HTTP 模式下，並沒有和 SSE 模式類似的 “連接” 過程（在 sse_client 調用時），因爲無需事先創建 SSE 連接；

2. 客戶端發起初始化請求（Initialize）。如果是有狀態模式，會在返回消息的 HTTP 頭中攜帶 session-id；

3. 客戶端發起初始化確認（Initialized）。此時如果已有 session-id（有狀態），客戶端會首先發起一次 HTTP Get 請求，以建立獨立的 SSE 通道；

4. 後續正常交互：普通的交互都是通過 Post 通道來進行，只有兩種情況會使用 SSE 通道：服務端發起的通知與請求、以及會話恢復的事件發送。

06

Low-level API 開發服務端

Low-level API 開發服務端要比 SSE 模式下更爲簡潔。這是由於服務端現在引入了 SessionManager 模塊來管理客戶端會話。因此只需要：創建 SessionManager 模塊，並把所有 / mcp 端點的請求都路由到該模塊的 handle_request 方法即可。此處注意 SessionManager 模塊需要首先通過 run 方法初始化。

...
    mcp_server = Server()

    ...call_tool等實現...

    try:

        # 創建會話管理器
        session_manager = StreamableHTTPSessionManager(
            app=mcp_server,
            json_response=True,
            stateless=False
        )
        
        starlette_app = Starlette(
            debug=True,
            routes=[
                Mount("/mcp", app=session_manager.handle_request),
            ],
            lifespan=lambda app: session_manager.run(),
        )

        config = uvicorn.Config(starlette_app, host="127.0.0.1", port=5050)
        server = uvicorn.Server(config)
        await server.serve()
        logger.info("MCP server is running on http://127.0.0.1:5050")
 ......

07

解讀多應用實例模式

現在 MCP 服務端可以支持多應用實例模式：你可以創建多個 FastMCP 的應用實例，不同實例採用不同的參數，這有助靈活的根據不同的場景需求來設計 MCP 服務端，比如可以把企業 API 訪問的工具全部用無狀態模式提供；其他需要長連接的工具使用有狀態模式提供。參考如下實現：

......
app = FastMCP(
    ,...
    stateless_http=True,
    json_response=False
)

app2 = FastMCP(
    ,...
    stateless_http=False,
    json_response=False
)

if __name__ == '__main__':
......
    @asynccontextmanager
    async def lifespan(server):
        asyncwith contextlib.AsyncExitStack() as stack:
            await stack.enter_async_context(app.session_manager.run())
            await stack.enter_async_context(app2.session_manager.run())
            yield

    server = FastAPI(lifespan=lifespan)
    server.mount("/server1", app.streamable_http_app())
    server.mount("/server2", app2.streamable_http_app())
    
    print("Starting FastAPI server on http://localhost:5050")
    print("- App1 available at: http://localhost:5050/server1")
    print("- App2 available at: http://localhost:5050/server2")
    uvicorn.run(server, host="0.0.0.0", port=5050)

多應用實例模式下，你不能直接使用 FastMCP 提供的 run 方法。你只能調用其 streamable_http_app() 函數獲得內部的 Starlette 應用模塊，然後映射到不同的路徑。有兩個問題需要注意：

• 每一個 FastMCP 應用實例內的 session_manager 都必須在啓動時初始化（藉助 lifespan 生命週期管理器）；

• 這裏的 mount 映射是將應用實例掛載到指定路徑；與 streamable HTTP 默認的服務端點 / mcp 不矛盾（類似 FastAPI 中的 @app.post("/mcp"）。因此這種模式下客戶端連接的 URL 要變爲（以上面的代碼中的 server1 爲例）：

    http://xxxx:port/server1/mcp

多應用實例的好處是相互獨立，每一個都有自己的配置和生命週期，又可以同時運行在一個服務器實例中。

以上就是我們對最新 MCP SDK 中推出的 streamable HTTP 通信模式的詳細解析。新的模式在靈活性上有了較大提升，比如你可以選擇徹底退化成無狀態模式，以獲得更好的橫向擴展能力。在實際測試中，也發現了一些 Bug，還有一些功能尚未完善（比如會話恢復），期待後面的更新吧。

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