MCP协议的Streamable HTTP：革新数据传输的未来

引言

在数字化时代，数据传输的效率和稳定性是推动技术进步的关键。MCP（Model Context Protocol）作为AI生态系统中的重要一环，通过引入Streamable HTTP传输机制，为数据交互带来了革命性的变化。本文将深入解读MCP协议的Streamable HTTP，从会话协商到正式通信传输数据的全过程，探讨其技术架构、协议内容、实现方式以及对AI应用的影响。

技术架构

MCP协议采用客户端-服务器架构，其核心组件包括：

MCP主机（MCP Host）：包含MCP客户端的应用程序，如Claude Desktop、Cursor IDE等AI工具。
MCP客户端（MCP Client）：在主机内部与服务器保持1:1连接的协议实现。
MCP服务器（MCP Server）：轻量级程序，通过标准化协议暴露特定功能，可以是本地Node.js/Python程序或远程服务。

MCP服务器提供三类标准能力：

资源（Resources）：如文件读取、API数据获取。
工具（Tools）：第三方服务或功能函数，如Git操作、浏览器控制。
提示词（Prompts）：预定义的任务模板，增强模型特定场景表现。

Streamable HTTP的协议内容

Streamable HTTP作为MCP协议的一项重大更新，旨在解决传统HTTP+SSE方案的局限性，同时保留其优势。其核心内容包括以下几个方面：

统一消息入口：所有客户端到服务器的消息都通过/message端点发送，不再需要专门的SSE端点。
动态升级SSE流：服务器可以根据需要将客户端发往/message的请求升级为SSE连接，用于推送通知或请求。
会话管理机制：客户端通过请求头中的Mcp-Session-Id与服务器建立会话，服务器可选择是否维护会话状态。
支持无状态服务器：服务器可以选择完全无状态运行，不再需要维持长期连接。

实现方式

从会话协商到正式通信传输数据

1. 会话协商

会话协商是Streamable HTTP通信的初始阶段，客户端与服务器通过以下步骤建立会话：

客户端发送初始化请求：客户端通过HTTP POST向MCP服务器的/message端点发送一个InitializeRequest消息，携带协议版本和客户端能力信息。
服务器响应初始化：服务器收到请求后，返回一个InitializeResult消息，包含服务器支持的协议版本、服务器能力以及会话ID（Mcp-Session-Id）。
客户端发送已初始化通知：客户端收到服务器的响应后，发送一个Initialized通知，告知服务器初始化已完成。

示例：客户端发送初始化请求

POST /message HTTP/1.1
Host: mcp.example.com
Content-Type: application/json
Accept: text/event-stream, application/json{"jsonrpc": "2.0", "method": "initialize", "params": {"clientInfo": {"name": "MCP Client", "version": "1.0"}, "capabilities": {}}}

示例：服务器响应初始化

HTTP/1.1 200 OK
Content-Type: application/json
Mcp-Session-Id: 12345{"jsonrpc": "2.0", "id": 1, "result": {"serverInfo": {"name": "MCP Server", "version": "1.0"}, "capabilities": {}}}

示例：客户端发送已初始化通知

POST /message HTTP/1.1
Host: mcp.example.com
Content-Type: application/json
Accept: text/event-stream, application/json
Mcp-Session-Id: 12345{"jsonrpc": "2.0", "method": "initialized", "params": {}}

2. 正式通信传输数据

会话建立后，客户端和服务器可以通过以下步骤进行正式通信：

客户端发送消息：客户端通过HTTP POST向MCP服务器的/message端点发送JSON-RPC消息，携带会话标识Mcp-Session-Id。
服务器处理请求并响应：服务器根据请求内容处理消息，并通过SSE流或JSON对象返回响应。
动态升级为SSE流：如果需要实时推送消息，服务器可以将连接升级为SSE流。
断线重连与数据恢复：如果网络波动导致连接中断，客户端可以携带Last-Event-ID重新连接，服务器根据该ID重放未发送的消息。

示例：客户端发送消息

POST /message HTTP/1.1
Host: mcp.example.com
Content-Type: application/json
Accept: text/event-stream, application/json
Mcp-Session-Id: 12345{"jsonrpc": "2.0", "id": 1, "method": "get_file", "params": {"path": "/example.txt"}}

示例：服务器响应并升级为SSE流

HTTP/1.1 200 OK
Content-Type: text/event-stream
Cache-Control: no-cache
Connection: keep-alive
Mcp-Session-Id: 12345data: {"jsonrpc": "2.0", "id": 1, "result": "File content here"}

示例：客户端断线重连

GET /message HTTP/1.1
Host: mcp.example.com
Accept: text/event-stream
Last-Event-ID: 12345
Mcp-Session-Id: 12345

示例：服务器重放未发送的消息

data: {"jsonrpc": "2.0", "id": 2, "result": "Continued content here"}

服务端代码实现

from datetime import datetime
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.routing import Route
from starlette.responses import JSONResponse, StreamingResponse
import json
import uuid
from starlette.middleware.cors import CORSMiddleware
import asyncio
from typing import Dict, Any
import aiofiles
import random# 存储会话ID和对应的任务队列
sessions: Dict[str, Dict[str, Any]] = {}# 添加CORS支持
app = Starlette()
app.add_middleware(CORSMiddleware,allow_origins=["*"],allow_credentials=True,allow_methods=["*"],allow_headers=["*"],expose_headers=["Mcp-Session-Id"],
)@app.route('/message', methods=["POST", "GET"])
async def handle_message(request: Request):"""处理POST和GET请求。"""session_id = request.headers.get("Mcp-Session-Id") or request.query_params.get("Mcp-Session-Id")if request.method == "POST":try:data = await request.json()if data.get("method") == "initialize":# 初始化会话session_id = str(uuid.uuid4())sessions[session_id] = {"initialized": True,"task_queue": asyncio.Queue()}response = JSONResponse(content={"jsonrpc": "2.0","id": data.get("id"),"result": {"serverInfo": {"name": "MCP Server", "version": "1.0"},"capabilities": {},},})response.headers["Mcp-Session-Id"] = session_idreturn responseelif session_id and sessions.get(session_id, {}).get("initialized"):# 处理已初始化的请求if data.get("method") == "get_file":try:# 异步读取文件内容content = await async_read_file(data.get("params", {}).get("path", ""))return JSONResponse(content={"jsonrpc": "2.0","id": data.get("id"),"result": content,})except Exception as e:return JSONResponse(content={"jsonrpc": "2.0","id": data.get("id"),"error": f"Error reading file: {str(e)}",})else:return JSONResponse(content={"error": "Unknown method"})else:return JSONResponse(content={"error": "Session not initialized"}, status_code=400)except Exception as e:return JSONResponse(content={"error": f"Internal server error: {str(e)}"}, status_code=500)elif request.method == "GET":# 处理SSE流请求if not session_id or session_id not in sessions:return JSONResponse(content={"error": "Session not found"}, status_code=404)async def event_generator(session_id):while True:try:message = await asyncio.wait_for(sessions[session_id]["task_queue"].get(), timeout=10)  # 超时时间10秒yield f"data: {json.dumps(message)}\n\n"except asyncio.TimeoutError as e:yield f"data: {e}\n\n"  # 发送空数据作为心跳包，防止超时断开return StreamingResponse(event_generator(session_id), media_type="text/event-stream")async def async_read_file(path: str) -> str:"""异步读取文件内容。"""try:async with aiofiles.open(path, "r") as file:content = await file.read()return contentexcept Exception as e:raise Exception(f"Error reading file: {str(e)}")async def background_task(session_id: str, task: Dict[str, Any]):"""后台任务处理。"""# 模拟耗时操作await asyncio.sleep(1)# 将结果放入任务队列sessions[session_id]["task_queue"].put_nowait(task)@app.on_event("startup")
async def startup_event():async def push_test_messages():while True:sp = random.randint(1, 3)await asyncio.sleep(sp)  # 每5秒推送一个消息for session_id in sessions.keys():if sessions[session_id]["initialized"]:sessions[session_id]["task_queue"].put_nowait({"message": f"Hello from server!", "sleep": sp,"datetime": datetime.now().strftime("%Y-%m-%d %H:%M:%S")})asyncio.create_task(push_test_messages())  # 创建后台任务if __name__ == "__main__":import uvicornuvicorn.run(app, host="0.0.0.0", port=8000)

客户端代码实现

import httpx
import json
import asyncio
import aiofilesclass MCPClient:def __init__(self, server_url: str):self.server_url = server_urlself.session_id = Noneself.headers = {"Content-Type": "application/json","Accept": "text/event-stream, application/json"}async def initialize(self):"""初始化会话。"""async with httpx.AsyncClient() as client:try:response = await client.post(f"{self.server_url}/message",headers=self.headers,json={"jsonrpc": "2.0","method": "initialize","params": {"clientInfo": {"name": "MCP Client", "version": "1.0"},"capabilities": {},},},)response.raise_for_status()self.session_id = response.headers.get("Mcp-Session-Id")print(f"Session ID: {self.session_id}")return self.session_idexcept Exception as e:print(f"Failed to initialize session: {e}")return Noneasync def send_message(self, method: str, params: dict = None):"""发送消息。"""if not self.session_id:await self.initialize()async with httpx.AsyncClient() as client:try:response = await client.post(f"{self.server_url}/message",headers={"Mcp-Session-Id": self.session_id, **self.headers},json={"jsonrpc": "2.0","id": 1,"method": method,"params": params or {},},)response.raise_for_status()return response.json()except Exception as e:print(f"Failed to send message: {e}")return Noneasync def listen_sse(self):if not self.session_id:await self.initialize()async with httpx.AsyncClient(timeout=None) as client:  # 取消超时限制try:async with client.stream("GET",f"{self.server_url}/message",headers={"Mcp-Session-Id": self.session_id, **self.headers},) as response:async for line in response.aiter_lines():if line.strip():  # 避免空行print(f"SSE Message: {line}")except Exception as e:print(f"Failed to listen SSE: {e}")await self.reconnect()async def reconnect(self):"""断线重连。"""print("Attempting to reconnect...")await asyncio.sleep(5)  # 等待5秒后重试await self.initialize()await self.listen_sse()async def main():client = MCPClient("http://localhost:8000")await client.initialize()response = await client.send_message("get_file", {"path": "/Users/houjie/PycharmProjects/python-sdk/examples/mcp-server/example.txt"})print(f"Response: {response}")await client.listen_sse()if __name__ == "__main__":asyncio.run(main())

前端页面代码实现

<!DOCTYPE html>
<html lang="en">
<head><meta charset="UTF-8"><meta name="viewport" content="width=device-width, initial-scale=1.0"><title>MCP Streamable HTTP Demo</title><style>body {font-family: Arial, sans-serif;margin: 0;padding: 20px;background-color: #f5f5f5;}.container {max-width: 800px;margin: 0 auto;background-color: white;padding: 20px;border-radius: 8px;box-shadow: 0 2px 4px rgba(0, 0, 0, 0.1);}h1 {text-align: center;color: #333;}.message-area {margin-top: 20px;}.message {padding: 10px;margin-bottom: 10px;border-radius: 4px;background-color: #e9f7fe;border-left: 4px solid #0099cc;}.sse-message {padding: 10px;margin-bottom: 10px;border-radius: 4px;background-color: #f0f9ff;border-left: 4px solid #0077cc;}button {background-color: #0099cc;color: white;border: none;padding: 10px 15px;border-radius: 4px;cursor: pointer;font-size: 14px;}button:hover {background-color: #0077cc;}input[type="text"] {width: 100%;padding: 10px;margin-bottom: 10px;border: 1px solid #ddd;border-radius: 4px;font-size: 14px;}</style>
</head>
<body><div class="container"><h1>MCP Streamable HTTP Demo</h1><div><input type="text" id="serverUrl" placeholder="Enter server URL" value="http://localhost:8000"><button id="initBtn">Initialize Session</button></div><div id="sessionId"></div><div><input type="text" id="filePath" placeholder="Enter file path"><button id="sendBtn">Send Message</button></div><div class="message-area" id="messages"></div></div><script>let client = null;let sessionInitialized = false;document.getElementById('initBtn').addEventListener('click', async () => {const serverUrl = document.getElementById('serverUrl').value;client = new MCPClient(serverUrl);await client.initialize();sessionInitialized = true;document.getElementById('sessionId').textContent = `Session ID: ${client.session_id}`;});document.getElementById('sendBtn').addEventListener('click', async () => {if (!sessionInitialized) {alert('Please initialize the session first.');return;}const filePath = document.getElementById('filePath').value;const response = await client.send_message('get_file', { path: filePath });addMessage(`Response: ${JSON.stringify(response)}`);});class MCPClient {constructor(serverUrl) {this.serverUrl = serverUrl;this.session_id = null;this.headers = {'Content-Type': 'application/json','Accept': 'text/event-stream, application/json'};}async initialize() {try {const response = await fetch(`${this.serverUrl}/message`, {method: 'POST',headers: this.headers,body: JSON.stringify({jsonrpc: '2.0',method: 'initialize',params: {clientInfo: { name: 'MCP Client', version: '1.0' },capabilities: {}}})});if (!response.ok) {throw new Error(`HTTP error! status: ${response.status}`);}this.session_id = response.headers.get('Mcp-Session-Id');addMessage(`Session ID: ${this.session_id}`);this.listen_sse();} catch (error) {addMessage(`Failed to initialize session: ${error}`);}}async send_message(method, params) {if (!this.session_id) {await this.initialize();}try {const response = await fetch(`${this.serverUrl}/message`, {method: 'POST',headers: { 'Mcp-Session-Id': this.session_id, ...this.headers },body: JSON.stringify({jsonrpc: '2.0',id: 1,method: method,params: params || {}})});if (!response.ok) {throw new Error(`HTTP error! status: ${response.status}`);}return await response.json();} catch (error) {addMessage(`Failed to send message: ${error}`);return null;}}listen_sse() {if (!this.session_id) {return;}const eventSource = new EventSource(`${this.serverUrl}/message?Mcp-Session-Id=${this.session_id}`, {headers: { 'Mcp-Session-Id': this.session_id }});eventSource.onmessage = (event) => {addSSEMessage(event.data);};eventSource.onerror = (error) => {addMessage(`Failed to listen SSE: ${error}`);this.reconnect();};}async reconnect() {addMessage('Attempting to reconnect...');await new Promise(resolve => setTimeout(resolve, 5000));await this.initialize();this.listen_sse();}}function addMessage(message) {const messagesDiv = document.getElementById('messages');const messageDiv = document.createElement('div');messageDiv.className = 'message';messageDiv.textContent = message;messagesDiv.appendChild(messageDiv);messagesDiv.scrollTop = messagesDiv.scrollHeight;}function addSSEMessage(message) {const messagesDiv = document.getElementById('messages');const messageDiv = document.createElement('div');messageDiv.className = 'sse-message';messageDiv.textContent = `SSE Message: ${message}`;messagesDiv.appendChild(messageDiv);messagesDiv.scrollTop = messagesDiv.scrollHeight;}</script>
</body>
</html>

运行步骤

安装依赖：
确保安装了所需的库：

pip install starlette uvicorn httpx aiofiles

启动服务器：
将服务端代码保存为 server.py，然后运行以下命令启动服务器：
```
uvicorn server:app --reload
```
运行客户端：
将客户端代码保存为 client.py，然后运行以下命令启动客户端：
```
python client.py
```
打开前端页面：
将前端页面代码保存为 index.html，然后在浏览器中打开该文件。

示例运行效果

客户端输出

Session ID: 587bb6ad-08f5-4102-8b27-4c276e9d7815
Response: {'jsonrpc': '2.0', 'id': 1, 'result': 'File content here'}
Listening for SSE messages...
SSE Message: data: {"message": "Hello from server!", "sleep": 1, "datetime": "2024-01-01 12:00:00"}
SSE Message: data: {"message": "Hello from server!", "sleep": 2, "datetime": "2024-01-01 12:00:02"}
...

服务器输出

INFO:     Started server process [12345]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
INFO:     127.0.0.1:51487 - "POST /message HTTP/1.1" 200 OK