在学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减少了浪费的资源。SSEHTTP的例子# 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 fdata: 这是服务器主动推送的消息\n\n await asyncio.sleep(2) return StreamingResponse(generate(), media_typetext/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 fdata: 流式输出第 {i1} 个Token\n\n await asyncio.sleep(1) return StreamingResponse(generate(), media_typetext/event-stream) # 场景 C客户端主动发 GET 请求建立 SSE 流用于接收服务器主动推送 elif request.method GET: async def generate(): while True: yield fdata: 服务器主动推送通知\n\n await asyncio.sleep(3) return StreamingResponse(generate(), media_typetext/event-stream)