好記憶不如爛筆頭,能記下點東西,就記下點,有時間拿出來看看,也會發覺不一樣的感受.
目錄
一、概述
二、快速上手:開啟 Streamable HTTP
服務端開啟
客戶端連接
三、深入兩個核心參數
stateless_http
json_response
參數組合效果
四、認識 session-id
生成與獲取
作用與使用
失效機制
五、樣例實測與驗證
服務端工具實現
客戶端測試
六、完整的通信流程
七、Low - Level API 開發服務端
八、解讀多應用實例模式
九、思考與總結
一、概述
本文主要對 MCP SDK 最新版(v1.9.0)中的 Streamable HTTP 模式進行全面剖析與實測。Streamable HTTP 是 MCP SDK 新增的傳輸模式,具有更復雜的實現和更靈活的配置。它為開發者提供了更多選擇,以滿足不同場景下的需求。
二、快速上手:開啟 Streamable HTTP
服務端開啟
從 MCP SDK 1.8.0 開始,支持三種傳輸模式:stdio、sse 與 streamable-http。服務端開啟 Streamable HTTP 模式推薦使用 FastMCP 高層 API:
# 創建 FastMCP 實例
app = FastMCP(name="SimpleServer",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。
客戶端連接
客戶端代碼需做以下修改:
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("會話初始化完成")# 獲取會話 IDsession_id = get_session_id()print(f"會話 ID: {session_id}")......
主要變化:
-
使用 streamablehttp_client 客戶端傳輸模塊。
-
新增可回調的 get_session_id 方法獲取 session-id。
三、深入兩個核心參數
stateless_http
-
控制是否建立長連接的 SSE 通道,為 False 時開啟 SSE 通道。
-
控制是否管理客戶端會話,為 False 時管理客戶端會話。
json_response
-
控制 Post 請求響應形式是否為 JSON,默認 False,使用 SSE 流式響應。
-
客戶端 Post 請求頭需聲明可接收兩種形式響應:"Accept": "application/json, text/event-stream"。
參數組合效果
不同參數組合產生不同效果,具體如下:
四、認識 session-id
session-id 是服務端跟蹤與管理客戶端會話的關鍵標識。
生成與獲取
-
當 stateless_http=False 時,發起初始化請求,服務端生成 session-id 并在 HTTP 頭中返回。
-
客戶端使用 get_session_id 回調方法獲取 session-id。
作用與使用
-
客戶端后續請求自動攜帶 session-id,也可用于關聯多次交互。
-
服務端從會話池查詢對應會話模塊處理請求,無需每次都建立新連接與會話。
失效機制
客戶端退出 streamablehttp_client 上下文區域時,觸發 HTTP Delete 請求,服務端刪除會話,session-id 失效。
五、樣例實測與驗證
服務端工具實現
創建一個模擬長時間處理任務的工具,定期報告進度:
@app.tool(name='hello')
async def hello(ctx: Context, name: str) -> str:steps = 10await 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}'
客戶端測試
-
有狀態模式(stateless_http=False,json_response=False) :連接成功后獲取 session-id,可接收服務端發送的進度通知,通過 SSE 通道以流形式發送。
-
無狀態模式(stateless_http=True,json_response=False) :無法接收到進度通知,服務端不會建立 SSE 通道。
六、完整的通信流程
-
連接 :客戶端無需事先創建 SSE 連接,直接發起初始化請求。
-
初始化請求 :客戶端發起 Initialize 請求,有狀態模式下服務端返回帶 session-id 的 HTTP 頭。
-
初始化確認 :客戶端發起 Initialized 確認,有狀態模式下客戶端發起 HTTP Get 請求建立獨立 SSE 通道。
-
正常交互 :普通交互通過 Post 通道進行,只有服務端發起的通知與請求、會話恢復事件發送使用 SSE 通道。
七、Low - Level API 開發服務端
使用 Low - Level API 開發服務端更簡潔,借助 SessionManager 模塊管理客戶端會話:
?
...mcp_server = Server(name="example")... 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")
......
八、解讀多應用實例模式
MCP 服務端支持多應用實例模式,可創建多個 FastMCP 應用實例,不同實例采用不同參數,靈活滿足不同場景需求:
......
app = FastMCP(name="SimpleServer",...stateless_http=True,json_response=False
)app2 = FastMCP(name="SimpleServer2",...stateless_http=False,json_response=False
)if __name__ == '__main__':
......@asynccontextmanagerasync def lifespan(server):async with contextlib.AsyncExitStack() as stack:await stack.enter_async_context(app.session_manager.run())await stack.enter_async_context(app2.session_manager.run())yieldserver = 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 應用實例內的 session_manager 必須在啟動時初始化。
-
mount 映射將應用實例掛載到指定路徑,客戶端連接的 URL 要相應變化。
九、思考與總結
Streamable HTTP 模式相比之前的傳輸模式,提供了更高的靈活性和更多的配置選項。開發者可以根據實際需求,選擇有狀態或無狀態模式,決定是否使用長連接的 SSE 通道,以及控制響應的形式。這種靈活性使得 MCP SDK 能夠更好地適應各種復雜的應用場景。
然而,在實際使用過程中,我們也發現了一些問題,如會話恢復功能的完善性有待驗證。希望在后續的版本更新中,這些問題能夠得到解決,進一步提升 Streamable HTTP 模式的穩定性和可靠性。
對于開發者來說,深入理解 Streamable HTTP 模式的原理和配置,熟練掌握其開發方法,將有助于更好地利用 MCP SDK 構建高效、可靠的通信系統,滿足不斷增長的業務需求。