一、MCP Streamable HTTP Transport 與 SSE：技術(shù)選型辨析

在構(gòu)建需要實(shí)時(shí)數(shù)據(jù)流的AI應(yīng)用時(shí)，MCP 的 Streamable HTTP Transport 和傳統(tǒng)的 Server-Sent Events 是兩種常見(jiàn)的備選方案。雖然它們都旨在實(shí)現(xiàn)服務(wù)器到客戶端的單向數(shù)據(jù)推送，但其設(shè)計(jì)目標(biāo)、協(xié)議層級(jí)和應(yīng)用場(chǎng)景有著本質(zhì)區(qū)別。

核心定義

• SSE：一種 Web 標(biāo)準(zhǔn)協(xié)議SSE 是一個(gè)基于 HTTP 的輕量級(jí)標(biāo)準(zhǔn)。它允許服務(wù)器通過(guò)一個(gè)持久的 HTTP 連接，主動(dòng)向客戶端（通常是瀏覽器）推送格式化的文本數(shù)據(jù)流。其核心特點(diǎn)是簡(jiǎn)單、易用，是 Web 生態(tài)的原生組成部分。
• MCP Streamable HTTP Transport：一個(gè)應(yīng)用層傳輸機(jī)制這是 Model Context Protocol 規(guī)范中定義的一種傳輸方式，用于在 MCP 客戶端與服務(wù)器之間交換結(jié)構(gòu)化的消息（如工具調(diào)用、結(jié)果、資源列表等）。它不是一個(gè)像 SSE 那樣的獨(dú)立協(xié)議，而是 MCP 這個(gè)特定應(yīng)用協(xié)議之上的一個(gè)實(shí)現(xiàn)約定。

主要區(qū)別總結(jié)

結(jié)論與選型建議

簡(jiǎn)單來(lái)說(shuō)，可以將兩者的關(guān)系理解為 “專用車(chē)”與“通用零部件”。

• SSE 是一個(gè)“通用零部件”：它是一種解決“如何從服務(wù)器向客戶端持續(xù)發(fā)送數(shù)據(jù)”這個(gè)通用問(wèn)題的標(biāo)準(zhǔn)化方案。你可以在任何需要此功能的 Web 應(yīng)用或服務(wù)中使用它。
• MCP Streamable HTTP Transport 是一輛“專用車(chē)”：它是在 MCP 這個(gè)特定框架下，為了解決“如何將服務(wù)器端的結(jié)構(gòu)化數(shù)據(jù)（如日志、進(jìn)度）流式傳輸給 MCP 客戶端”這個(gè)具體問(wèn)題而設(shè)計(jì)的。它很可能在內(nèi)部使用了類似 SSE 的機(jī)制（例如，通過(guò)一個(gè)持久的 HTTP 響應(yīng)流），但其之上承載的數(shù)據(jù)是 MCP 協(xié)議專屬的。

因此，你的選擇取決于你的目標(biāo)：

• 如果你正在 開(kāi)發(fā)或使用 MCP 服務(wù)器，并需要實(shí)現(xiàn)流式輸出（如實(shí)時(shí)日志），那么你需要遵循 MCP Streamable HTTP Transport 的規(guī)范。
• 如果你需要在 一個(gè)普通的 Web 應(yīng)用或后端服務(wù) 中實(shí)現(xiàn)簡(jiǎn)單的服務(wù)器推送功能（如新聞推送、狀態(tài)更新），那么 SSE 是一個(gè)成熟、簡(jiǎn)單且標(biāo)準(zhǔn)的選擇。

二、MCP Streamable HTTP Transport 示例代碼

創(chuàng)建虛擬環(huán)境及安裝mcp依賴包

conda create -n mcp_demo  python=3.12


conda activate mcp_demo
pip install uv




uv pip install "mcp[cli]"

運(yùn)行服務(wù)端代碼

server_demo.py

https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/servers/streamable_config.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")

啟動(dòng)成功！

運(yùn)行客戶端獲取工具列表：

client_demo.py

https://github.com/modelcontextprotocol/python-sdk/blob/main/examples/snippets/clients/streamable_basic.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())

獲取到了服務(wù)端的工具列表：

圖片

參考文檔：https://github.com/modelcontextprotocol/python-sdk.git