模型训练

深度剖析 MCP SDK 最新版： Streamable HTTP 模式正式发布，为你实测揭秘

2025 年 5 月 21 日

揭开MCP SDK最新版streamable HTTP模式的神秘面纱，深入解析与实操指南。核心内容： 1. MCP SDK v1.9.0版本更新亮点：streamable HTTP模式 2. 快速开启streamable HTTP模式的步骤与核心参数解析 3. 客户端和服务端代码示例，以及session-id的理解和应用

最近，MCP SDK 新版本更新发布（最新为v1.9.0），其中最大的更新莫过于终于提供了新版协议中的传输模式 — streamable HTTP。不过由于MCP SDK的文档一直以来”语焉不详“的风格，很多开发者知其然却不知其所以然，很容易在应用中踩坑。

本文将对这种模式进行全面剖析与实测，帮助大家深入认识这种新的模式。

快速上手：开启streamable HTTP
深入两个核心参数
认识 session-id
样例实测与验证
Low-Level API开发服务端
解读多应用实例模式

快速上手：开启streamable HTTP

从版本SDK 1.8.0开始，MCP支持三种传输模式：stdio、sse与streamable-http。这三种模式的服务端与客户端必须匹配使用，即streamable HTTP的客户端无法与sse模式服务端交互，所以在一些客户端工具中配置MCP时，不要盲目采用新模式。

【服务端】

服务端开启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）。

【客户端】

对应的客户端代码修改如下，注意这里的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）。

让我们来逐步深入。

深入两个核心参数

首先看到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时，可具备一项额外的能力：会话恢复。即：服务端返回的事件流具备“断点”续传的功能。该功能需要自行实现服务端事件的持久化接口，且完善性有待验证，本文暂不对此深入。

认识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失效
在服务端的作用：后续的请求无需每次建立新的连接与会话，而是从一个“池”里查询出已经创建的会话模块用来处理请求

样例实测与验证

我们用一个例子来体验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通道（哪怕是流形式的响应）。

完整的通信流程

用下图来整理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通道：服务端发起的通知与请求、以及会话恢复的事件发送。

Low-level API开发服务端

Low-level API开发服务端要比SSE模式下更为简洁。这是由于服务端现在引入了SessionManager模块来管理客户端会话。因此只需要：创建SessionManager模块，并把所有/mcp端点的请求都路由到该模块的handle_request方法即可。此处注意SessionManager模块需要首先通过run方法初始化。

...
    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的应用实例，不同实例采用不同的参数，这有助灵活的根据不同的场景需求来设计MCP服务端，比如可以把企业API访问的工具全部用无状态模式提供；其他需要长连接的工具使用有状态模式提供。参考如下实现：

......
app = FastMCP(
    name="SimpleServer",...
    stateless_http=True,
    json_response=False
)

app2 = FastMCP(
    name="SimpleServer2",...
    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，还有一些功能尚未完善（比如会话恢复），期待后面的更新吧。

end

立即咨询 JOTO

JOTO 提供覆盖企业智能体规划与搭建、AI 平台私有化部署、RAG 知识工程、AI 安全治理、FDE 驻场共创及持续运营优化的全周期 AI 落地服务，帮助企业把验证中的 AI 能力转化为安全、可控、可持续迭代的生产力。联系 JOTO 获取 AI 落地咨询

想把这些做法用到你的业务里?

留下你的场景和痛点,我们帮你判断从哪一步开始。

联系我们

深度剖析 MCP SDK 最新版： Streamable HTTP 模式正式发布，为你实测揭秘

立即咨询 JOTO

想把这些做法用到你的业务里?

开启企业级 AI 落地

填写需求单