MCP Streamable HTTP Transport 與 SSE:專用通道與通用協議的辨析,內附MCP的示例代碼
一、MCP Streamable HTTP Transport 與 SSE:技術選型辨析
在構建需要實時數據流的AI應用時,MCP 的 Streamable HTTP Transport 和傳統的 Server-Sent Events 是兩種常見的備選方案。雖然它們都旨在實現服務器到客戶端的單向數據推送,但其設計目標、協議層級和應用場景有著本質區別。
核心定義
- SSE:一種 Web 標準協議SSE 是一個基于 HTTP 的輕量級標準。它允許服務器通過一個持久的 HTTP 連接,主動向客戶端(通常是瀏覽器)推送格式化的文本數據流。其核心特點是簡單、易用,是 Web 生態的原生組成部分。
- MCP Streamable HTTP Transport:一個應用層傳輸機制這是 Model Context Protocol 規范中定義的一種傳輸方式,用于在 MCP 客戶端與服務器之間交換結構化的消息(如工具調用、結果、資源列表等)。它不是一個像 SSE 那樣的獨立協議,而是 MCP 這個特定應用協議之上的一個實現約定。
主要區別總結
特性 | MCP Streamable HTTP Transport | Server-Sent Events |
協議層級 | 應用層協議(MCP)的傳輸實現 | 網絡通信標準協議 |
數據格式 | 承載的是結構化的 MCP 消息(JSON-RPC-like)。 | 傳輸的是純文本流,格式遵循 |
設計目標 | 專門為 MCP 客戶端-服務器間雙向通信的“單向流”部分 而設計。例如,流式傳輸一個工具調用的執行日志或進度更新。 | 為 通用的服務器到客戶端(尤其是瀏覽器)事件推送 而設計。 |
連接方向 | 在 MCP 中,它通常用于服務器向客戶端推送流數據,但其底層連接本身可能由客戶端發起的一個請求(如帶 | 嚴格的單向通信 :服務器 -> 客戶端??蛻舳送ㄟ^一個普通的 HTTP GET 請求發起連接,然后服務器持續推送。 |
生態與兼容性 | 與 MCP 生態系統強綁定,主要用于 AI 應用與外部服務(如數據庫、文件系統)的橋接。 | 廣泛支持 ,是所有現代瀏覽器的內置功能,也可用于任何 HTTP 客戶端。通用性極強。 |
結論與選型建議
簡單來說,可以將兩者的關系理解為 “專用車”與“通用零部件”。
- SSE 是一個“通用零部件”:它是一種解決“如何從服務器向客戶端持續發送數據”這個通用問題的標準化方案。你可以在任何需要此功能的 Web 應用或服務中使用它。
- MCP Streamable HTTP Transport 是一輛“專用車”:它是在 MCP 這個特定框架下,為了解決“如何將服務器端的結構化數據(如日志、進度)流式傳輸給 MCP 客戶端”這個具體問題而設計的。它很可能在內部使用了類似 SSE 的機制(例如,通過一個持久的 HTTP 響應流),但其之上承載的數據是 MCP 協議專屬的。
因此,你的選擇取決于你的目標:
- 如果你正在 開發或使用 MCP 服務器,并需要實現流式輸出(如實時日志),那么你需要遵循 MCP Streamable HTTP Transport 的規范。
- 如果你需要在 一個普通的 Web 應用或后端服務 中實現簡單的服務器推送功能(如新聞推送、狀態更新),那么 SSE 是一個成熟、簡單且標準的選擇。
二、MCP Streamable HTTP Transport 示例代碼
創建虛擬環境及安裝mcp依賴包
conda create -n mcp_demo python=3.12
conda activate mcp_demo
pip install uv
uv pip install "mcp[cli]"運行服務端代碼
server_demo.py
"""
Run from the repository root:
uv run examples/snippets/servers/streamable_config.py
"""
from mcp.server.fastmcp import FastMCP
# Stateful server (maintains session state)
mcp = FastMCP("StatefulServer")
# Other configuration options:
# Stateless server (no session persistence)
# mcp = FastMCP("StatelessServer", stateless_http=True)
# Stateless server (no session persistence, no sse stream with supported client)
# mcp = FastMCP("StatelessServer", stateless_http=True, json_response=True)
# Add a simple tool to demonstrate the server
@mcp.tool()
def greet(name: str = "World") -> str:
"""Greet someone by name."""
return f"Hello, {name}!"
# Run server with streamable_http transport
if __name__ == "__main__":
mcp.run(transport="streamable-http")啟動成功!
運行客戶端獲取工具列表:
client_demo.py
"""
Run from the repository root:
uv run examples/snippets/clients/streamable_basic.py
"""
import asyncio
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client
async def main():
# Connect to a streamable HTTP server
async with streamablehttp_client("http://127.0.0.1:8000/mcp") as (
read_stream,
write_stream,
_,
):
# Create a session using the client streams
async with ClientSession(read_stream, write_stream) as session:
# Initialize the connection
await session.initialize()
# List available tools
tools = await session.list_tools()
print(f"Available tools: {[tool.name for tool in tools.tools]}")
if __name__ == "__main__":
asyncio.run(main())獲取到了服務端的工具列表:
圖片





































