在学MCP协议时，MCP在本地通信用stdio协议，而在远程通信采用SSE流+HTTP。后来由SSE流+HTTP升级为Streamable HTTP。我也是临时学习了一下这两者的异同优劣，在这里和大家分享浅谈一下。

【注】所以无论是SSE流还是Streamable讨论的范围都是传输层（transport）的事。

什么是SSE流（Server-Sent Events）？

"流"可以理解成“数据通道”，应该是来自于Bash那边的专业术语。SSE流翻译过来是服务器发送事件，可以理解一种端到端的传输协议，专门服务于服务端发送数据（数据在协议中称为事件events），利用HTTP协议进行单向传输，传输到客户端。当然服务端是受到客户端的申请请求才会发送数据，不会无缘无故的发送数据。

异同点

SSE流和Streamable 都是用于远程通信，都会建立一个长连接，都能达到双向通信的效果，都可做到流式输出。

• Streamable HTTP 可以记住数据的会话ID，做到断点续传；
• 而且Streamable HTTP可以按需分配，所以总体上会比SSE流更省资源；
• Streamable是单通道，也就是双向通道，需要一个端口号，单通道有POST和GET两种请求方式。SSE是双通道，也就是两个单向通道，需要二个端口号，一个通道用POST请求方式，一个用GET请求方式。
• Streamable即支持像JSON格式响应的短连接，也支持SSE长连接。而SSE就只有长连接。

Streamable升级里哪里？

我前面讲Streamable是SSE流的升级版，那升级了哪里？这要从SSE的缺点说起。

SSE的工作模式是收发分离。假如Server端用POST占据3030端口去仅发送数据，我用GET占据3031端口去仅接受请求。那Client端就要也用2个端口去接、发数据。Server端收到Client端的请求，采用流式输出的方式，发送数据到Client端，发送一半突然网络故障。此时，传统的SSE流协议是无法恢复中断，继续传输。只能重新建立对话，重新连接。

而Streamable 也是用POST占据3030端口去发送数据，它接受数据也是靠3030端口,建立双向通道，收发一体。当遇到网络故障是，POST也是无法工作了，但Streamable HTTP还有个替补：GET方式。

替补也是用3030端口，用GET方式去尝试建立短连接，相当于搜救兵，来搞清楚状况，看看到底是网络故障，还是Server端自己不发送数据了。如果是网络故障，他就会临时升级为长连接，也就是化身24小时工作超长待机的战士，直到危机解除为止，替代POST发送数据到Client端。这个功能叫断点续传。

Streamable可以根据需求自由切换长/短连接方式，而SSE它只能长连接，即使是短连接的业务，它也必须24小时侯着，直到整个会话结束。所以说Streamable比SSE更省资源，或者说Streamable减少了浪费的资源。

SSE+HTTP的例子：

# old_http_sse.py from fastapi import FastAPI from fastapi.responses import StreamingResponse import asyncio app = FastAPI() # 通道 1：客户端通过 POST 发送消息 @app.post("/message") async def send_message(): # 收到消息后，服务器无法直接返回结果，只能返回一个 202 (Accepted) # 意思是：“我收到了，结果我会通过 SSE 通道推给你” return {"status": "Message received, check SSE stream"} # 通道 2：客户端必须通过 GET 保持一个长连接，用来接收服务器的推送 @app.get("/sse") async def sse_stream(): async def generate(): while True: # 模拟服务器持续向客户端推送数据 yield f"data: 这是服务器主动推送的消息\n\n" await asyncio.sleep(2) return StreamingResponse(generate(), media_type="text/event-stream")

Streamable HTTP的例子：

# streamable_http.py from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse, JSONResponse import asyncio app = FastAPI() # 唯一通道：所有通信都在这一个端点完成 @app.api_route("/mcp", methods=["POST", "GET"]) async def mcp_endpoint(request: Request): # 场景 A：客户端发来普通请求，服务器直接返回普通 JSON（一问一答） if request.method == "POST": data = await request.json() # 服务器判断：如果不需要流式输出，直接返回普通 HTTP 响应 if data.get("type") == "simple_query": return JSONResponse({"result": "这是普通的同步响应"}) # 场景 B：客户端发来复杂请求，服务器“按需升级”为 SSE 流 elif data.get("type") == "stream_query": async def generate(): for i in range(3): yield f"data: 流式输出第 {i+1} 个Token\n\n" await asyncio.sleep(1) return StreamingResponse(generate(), media_type="text/event-stream") # 场景 C：客户端主动发 GET 请求，建立 SSE 流（用于接收服务器主动推送） elif request.method == "GET": async def generate(): while True: yield f"data: 服务器主动推送通知\n\n" await asyncio.sleep(3) return StreamingResponse(generate(), media_type="text/event-stream")