-
快速上手：开启streamable HTTP -
深入两个核心参数 -
认识 session-id -
样例实测与验证 -
Low-Level API开发服务端 -
解读多应用实例模式
01
快速上手：开启streamable HTTP
# 创建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')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
02
深入两个核心参数
-
Post通道：用作客户端到服务端的所有消息传输；同时服务端对客户端的响应消息也通过Post通道直接返回（基本类似Restful HTTP Post交互）。响应形式可能是SSE流，也可能是JSON数据。 -
SSE通道：变为可选的通道，仅用作服务端发往客户端的通知（Notification，比如进度更新）、请求（Request，比如Sampling请求）。很显然这里的形式只能是SSE流。（实际还有一种“会话恢复”功能会用到SSE通道）
-
控制是否会建立长连接的SSE通道。为False时则会开启SSE通道。 -
控制是否会管理客户端的会话。为False时会对客户端会话进行管理。
-
控制Post请求响应形式是否用JSON。默认为False，会使用SSE流式响应（注意并不是走SSE通道）。 -
客户端必须在Post请求头中声明可同时接收两种形式的响应。即： "Accept": "application/json, text/event-stream"
03
认识session-id
-
什么时候有：只有当stateless_http=False时才会有session-id
-
什么时候生成：在发起初始化请求时服务端生成并在headers中返回 -
客户端如何使用：后续所有的客户端发起的请求会自动携带该session-id；你也可开发其他用途，比如用来关联一组与MCP服务的多次交互 -
什么时候失效：客户端退出streamablehttp_client的上下文区域时会自动触发一个HTTP Delete请求，服务端会删除会话，session-id失效 -
在服务端的作用：后续的请求无需每次建立新的连接与会话，而是从一个“池”里查询出已经创建的会话模块用来处理请求
04
样例实测与验证
@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}'
05
完整的通信流程
-
红色代表SSE通道的交互；其他都是HTTP Post/Get/Delete。 -
虚线代表只有在有状态模式下才会有出现的交互或信息。
06
Low-level API开发服务端
...
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")
......07
解读多应用实例模式
......
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应用实例内的session_manager都必须在启动时初始化（借助lifespan生命周期管理器）； -
这里的mount映射是将应用实例挂载到指定路径；与streamable HTTP默认的服务端点/mcp不矛盾（类似FastAPI中的@app.post("/mcp"）。因此这种模式下客户端连接的URL要变为（以上面的代码中的server1为例）：
end
