网站导航
> 技术文档 > 深度剖析 MCP SDK 最新版：Streamable HTTP 模式_fastmcp lifespan

深度剖析 MCP SDK 最新版：Streamable HTTP 模式_fastmcp lifespan

网友: 技术文档 2025-07-30 12:12:13

好记忆不如烂笔头，能记下点东西，就记下点，有时间拿出来看看，也会发觉不一样的感受.

目录

一、概述

二、快速上手：开启 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(\"会话初始化完成\") # 获取会话 ID session_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 = 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}\'

客户端测试

  • 有状态模式（stateless_http=False，json_response=False） ：连接成功后获取 session-id，可接收服务端发送的进度通知，通过 SSE 通道以流形式发送。

  • 无状态模式（stateless_http=True，json_response=False） ：无法接收到进度通知，服务端不会建立 SSE 通道。

六、完整的通信流程

  1. 连接 ：客户端无需事先创建 SSE 连接，直接发起初始化请求。

  2. 初始化请求 ：客户端发起 Initialize 请求，有状态模式下服务端返回带 session-id 的 HTTP 头。

  3. 初始化确认 ：客户端发起 Initialized 确认，有状态模式下客户端发起 HTTP Get 请求建立独立 SSE 通道。

  4. 正常交互 ：普通交互通过 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__\':...... @asynccontextmanager async 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()) 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 应用实例内的 session_manager 必须在启动时初始化。

  • mount 映射将应用实例挂载到指定路径，客户端连接的 URL 要相应变化。

九、思考与总结

Streamable HTTP 模式相比之前的传输模式，提供了更高的灵活性和更多的配置选项。开发者可以根据实际需求，选择有状态或无状态模式，决定是否使用长连接的 SSE 通道，以及控制响应的形式。这种灵活性使得 MCP SDK 能够更好地适应各种复杂的应用场景。

然而，在实际使用过程中，我们也发现了一些问题，如会话恢复功能的完善性有待验证。希望在后续的版本更新中，这些问题能够得到解决，进一步提升 Streamable HTTP 模式的稳定性和可靠性。

对于开发者来说，深入理解 Streamable HTTP 模式的原理和配置，熟练掌握其开发方法，将有助于更好地利用 MCP SDK 构建高效、可靠的通信系统，满足不断增长的业务需求。

免费在线游戏

网络标签：

相关问题