Streamable HTTP 是 MCP（一种通信协议）在 2025-03-26 规范中引入的新型传输方式，旨在改进现有 HTTP + SSE 模型的不足。它通过统一端点、灵活响应模式、会话管理及断线恢复等特性，提供了更高效、可靠和兼容的通信机制。

1 Streamable Http是什么？

Streamable HTTP 是 MCP 在较新版本（2025-03-26 规范中的一个传输方式）里推出的一个改进版本。

它的主要内容 / 特性如下：

特性	说明
统一端点（single endpoint）	所有 MCP 的通信（初始化、普通请求／响应、流式响应等）都通过同一个 HTTP 路径／端点（通常是`/mcp`）来完成，而不再区分一个专用的 SSE 端点。
灵活响应模式	对于服务器：可以选择直接返回普通 HTTP 响应（适合短小交互、瞬时请求／响应的情形），也可以升级连接为 SSE 流式响应，以推送一系列事件（例如用于长任务的进度等）。客户端可能发送 GET 来打开 SSE 流，也可能通过 POST 请求得到即时响应。
会话管理（Session ID）	引入`Mcp-Session-Id`的机制。在初始化阶段，服务器可以生成一个会话 ID 返回给客户端；客户端在后续请求中带上这个 session ID，以标识该会话。这样做可以支持断线重连、消息恢复等功能。
断线恢复 / 可重续流	如果 HTTP／SSE 的连接中断了，通过 session ID +流里的事件 ID (event ID) 等机制，客户端可以重新连接，并从上次中断的地方继续接收服务器未收到／未处理（或未推送）的事件。
向后兼容 / 与旧传输的共存	由于 MCP 的早期版本采用的是 HTTP + SSE 机制，新传输方式（Streamable HTTP）在设计中考虑了让客户端／服务器能够兼容旧方式，比如在服务器支持的情况下同时提供旧的 endpoint，也让客户端在初始化尝试新方式失败时可以降级使用旧方式。

2 为什么需要Streamable Http？

相比旧的 HTTP + SSE 模型，Streamable HTTP 解决／缓解了一些问题。这里是几个核心原因：

减少连接管理复杂性与资源开销

SSE 要求服务器维持一个长期开启的连接去推事件；在大量并发客户端／不稳定网络环境下，这种长连接会带来资源压力和连接中断问题。Streamable HTTP 在非流模式下可以直接用常规 HTTP 响应，不必总是开启持续连接。
更好的可靠性和断线恢复

在旧的 HTTP + SSE 模式，一旦 SSE断掉，中间状态可能丢失；客户端可能得重建整个会话。Streamable HTTP 的 session ID + 可重续流 (resumability) 能在很多场景中恢复断点，减少因网络问题导致的体验中断。
部署与基础设施兼容性更好

不同的网络基础设施（如负载均衡器、代理、CDN、防火墙等）有时不太友好对长时间的 SSE 通道，会断开或超时。Streamable HTTP 的灵活性让它更易于在这些环境中稳定运行。
简化客户端／服务器的复杂性

统一一个 endpoint、统一协议的流程，比分两个 endpoint（一个用于请求响应，一个用于 SSE 流）要清晰、易于管理。客户端的逻辑也可以更简单地判断什么情况下要流式，有时只要普通响应即可。

3 如何使用Streamable Http（Quick Start）

3.1 Server.py

from fastmcp import FastMCP

mcp = FastMCP("Demo 🚀")

@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two numbers"""
    return a + b

@mcp.tool()
def check_mysql(a: int, b: int) -> bool:
    """Check MySQL Connection"""
    return True

if __name__ == "__main__":
    mcp.run(transport="streamable-http", host="0.0.0.0", port=8000, path="/mcp")

3.2 Client.py

import asyncio
from typing import Optional
from mcp.client.session import ClientSession
from mcp.client.streamable_http import streamablehttp_client

URL = "http://localhost:8000/mcp"
HEADERS: dict[str, str] = {
    # "Authorization": "Bearer <TOKEN>",
}

async def _list_tools_once(base_url: str, headers: Optional[dict]):
    async with streamablehttp_client(base_url, headers=headers) as (read_stream, write_stream, _close):
        async with ClientSession(read_stream, write_stream) as session:
            await session.initialize()
            return await session.list_tools()

def list_tools_sync():
    async def _run():
        print(f"Connecting to {URL} ...")
        tools = await _list_tools_once(URL, HEADERS)
        print(f"Connected to {URL}")
        return tools
    return asyncio.run(_run())

if __name__ == "__main__":
    tools = list_tools_sync()
    print("== MCP Tools ==")
    for t in tools.tools:
        print(f"- {t.name}: {getattr(t, 'description', '') or ''}")

4 注意事项

虽然 Streamable HTTP 很多方面是改进，但在实际使用里也有一些要注意／约束的地方：

不是所有实现都完全支持所有特性

比如一些 Java SDK 在实现中可能还不完全支持session ID 管理或者 GET 请求建立 SSE 流等特性。某些实现可能只是以无状态（stateless）模式为主。
客户端／服务器端都需要支持并遵守规范

如果客户端只支持旧的 HTTP+SSE 或者服务器只实现了部分 Streamable HTTP 功能，就可能无法利用全部好处。要看版本／SDK 支持情况。
延迟／超时／资源清理

保持流／SSE 长连接需要考虑代理／负载均衡器的连接超时、心跳或者 Ping 机制、安全／认证、流断开时的清理等问题。Streamable HTTP 带来这些灵活性，也带来了在这些边缘情况要处理的额外复杂性。
安全性／身份验证

使用 Session ID 时，要确保那个 ID 是安全生成、不会被伪造，同时对于每个请求／连接，认证和权限控制是必要的。还要注意 CORS（浏览器环境）等问题。