主题
MCP Streamable HTTP 传输协议完全学习指南
Streamable HTTP 是 MCP（Model Context Protocol）协议在 2025 年 3 月 26 日推出的新一代传输机制，用于替代原有的 HTTP+SSE 方案，实现更灵活、更高效的 AI Agent 远程通信。
┌─────────────────────────────────────────────────────────────────┐
│ MCP 传输机制演进 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ 旧方案（2024） 新方案（2025.03.26） │
│ ┌─────────────┐ ┌─────────────────┐ │
│ │ HTTP + SSE │ ────────> │ Streamable HTTP │ │
│ └─────────────┘ └─────────────────┘ │
│ │
│ 问题： 改进： │
│ • 需要维护长连接 • 支持无状态服务器 │
│ • /sse 和 /messages 分离 • 统一 /mcp 端点 │
│ • 连接断开无法恢复 • 支持会话恢复和重传 │
│ • 基础设施兼容性差 • 兼容 CDN/API 网关 │
│ │
└─────────────────────────────────────────────────────────────────┘一、基础理论（Foundation）
1.1 定义与核心定位 ★重点★
什么是 Streamable HTTP？
Streamable HTTP 是 MCP 协议定义的两种标准传输机制之一（另一种是 stdio），它：
- 取代 HTTP+SSE 成为 MCP 的默认远程传输方式
- 统一所有通信到单一 HTTP 端点
- 支持按需流式传输（可选 SSE 升级）
- 兼容无状态服务器架构
一句话定义：
Streamable HTTP 是一种基于标准 HTTP 的传输协议，通过单一端点、动态 SSE 升级和会话管理机制，实现灵活、高效、可恢复的 AI Agent 远程通信。
1.2 设计初衷与解决的问题 ★重点★
HTTP+SSE 的三大痛点
| 痛点 | 问题描述 | 影响 |
|---|---|---|
| 长连接压力 | 服务器必须维护 SSE 长连接 | 高并发时资源消耗大 |
| 双端点复杂 | /sse 接收消息，/messages 发送消息 | 实现复杂，维护成本高 |
| 连接不可恢复 | 网络中断后无法恢复会话 | 用户体验差，数据丢失 |
| 基础设施兼容差 | 防火墙/负载均衡可能中断 SSE | 部署困难 |
Streamable HTTP 的解决方案
HTTP+SSE 架构（旧）：
┌─────────┐ GET /sse ┌─────────┐
│ Client │ ◄─────────────────│ Server │
│ │ POST /messages│ │
│ │ ─────────────────►│ │
└─────────┘ └─────────┘
两个端点，长连接依赖
Streamable HTTP 架构（新）：
┌─────────┐ ┌─────────┐
│ Client │ POST /mcp │ Server │
│ │ ◄────────────────►│ │
│ │ GET /mcp │ │
│ │ ◄─────────────────│ │
└─────────┘ └─────────┘
单一端点，按需流式1.3 核心特性 ★重点★
| 特性 | 说明 | 优势 |
|---|---|---|
| 统一端点 | 所有通信通过 /mcp 单一端点 | 简化部署和维护 |
| 按需流式 | 服务器可选择返回 JSON 或 SSE 流 | 灵活适配不同场景 |
| 会话管理 | Mcp-Session-Id 头部支持状态跟踪 | 支持会话恢复 |
| 可恢复性 | Last-Event-ID 支持断线重连 | 网络中断不丢数据 |
| 无状态支持 | 服务器可完全无状态运行 | 云原生友好 |
1.4 适用场景
✅ 适用场景
| 场景 | 说明 |
|---|---|
| 远程 MCP 服务 | 跨网络访问 MCP Server |
| 云原生部署 | Kubernetes、Serverless 环境 |
| 多客户端服务 | 一个 Server 服务多个 Client |
| 企业级应用 | 需要 CDN、负载均衡的场景 |
❌ 不适用场景
| 场景 | 原因 | 替代方案 |
|---|---|---|
| 本地工具集成 | 网络开销不必要 | 使用 stdio 传输 |
| 单进程通信 | 过度复杂 | 使用 stdio 传输 |
| 超低延迟要求 | HTTP 有一定开销 | 使用 stdio 或 WebSocket |
1.5 核心知识点速记
| 知识点 | 内容 |
|---|---|
| 发布时间 | 2025 年 3 月 26 日 |
| 替代方案 | HTTP+SSE |
| 核心端点 | 单一 HTTP 端点（如 /mcp） |
| 消息格式 | JSON-RPC 2.0，UTF-8 编码 |
| 会话标识 | Mcp-Session-Id 头部 |
| 流式响应 | Content-Type: text/event-stream |
📝 实操任务 1：理解协议演进
bash
# 任务：对比新旧协议的端点设计
# 思考以下问题：
1. HTTP+SSE 为什么需要两个端点？
- /sse：服务器 → 客户端（SSE 推送）
- /messages：客户端 → 服务器（POST 请求）
2. Streamable HTTP 如何统一为一个端点？
- POST /mcp：客户端发送消息
- GET /mcp：客户端监听消息（可选）
- 响应可以是 JSON 或 SSE 流
3. 画出两种方案的时序图对比二、核心语法（Core Syntax）
2.1 协议架构 ★重点★
┌─────────────────────────────────────────────────────────────────┐
│ Streamable HTTP 协议架构 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ MCP Client │ │
│ └──────────────────────────┬──────────────────────────────┘ │
│ │ │
│ ┌───────────────┼───────────────┐ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │POST /mcp │ │GET /mcp │ │DELETE/mcp│ │
│ │发送消息 │ │监听消息 │ │终止会话 │ │
│ └────┬─────┘ └────┬─────┘ └────┬─────┘ │
│ │ │ │ │
│ └───────────────┼───────────────┘ │
│ │ │
│ ┌─────────────────────────▼───────────────────────────────┐ │
│ │ MCP Server │ │
│ │ │ │
│ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │
│ │ │ 会话管理器 │ │ 消息处理器 │ │ SSE 生成器 │ │ │
│ │ └─────────────┘ └─────────────┘ └─────────────┘ │ │
│ └─────────────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘2.2 HTTP 方法规范 ★重点★ 【强制要求】
| 方法 | 用途 | 请求头 | 响应 |
|---|---|---|---|
| POST | 发送 JSON-RPC 消息 | Accept: application/json, text/event-stream | JSON 或 SSE |
| GET | 打开 SSE 监听流 | Accept: text/event-stream | SSE 流 |
| DELETE | 终止会话 | Mcp-Session-Id: <id> | 204 或 405 |
2.3 请求格式规范 ★重点★
POST 请求（发送消息）
http
POST /mcp HTTP/1.1
Host: example.com
Content-Type: application/json
Accept: application/json, text/event-stream
Mcp-Session-Id: abc123-xyz789
{
"jsonrpc": "2.0",
"id": 1,
"method": "tools/list",
"params": {}
}请求体类型
| 类型 | 说明 | 示例 |
|---|---|---|
| 单个请求 | 一个 JSON-RPC 消息 | {"jsonrpc":"2.0","id":1,...} |
| 批量请求 | 多个请求的数组 | [{...}, {...}] |
| 通知 | 无需响应的消息 | {"jsonrpc":"2.0","method":"..."} |
| 响应 | 对服务器请求的回复 | {"jsonrpc":"2.0","id":1,"result":...} |
2.4 响应格式规范 ★重点★ 【强制要求】
响应类型决策
┌─────────────────────────────────────────────────────────────────┐
│ 服务器响应决策树 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ 收到 POST 请求 │
│ │ │
│ ▼ │
│ ┌─────────────────┐ │
│ │ 请求包含什么？ │ │
│ └────────┬────────┘ │
│ │ │
│ ┌────────┼────────┬─────────────┐ │
│ │ │ │ │ │
│ ▼ ▼ ▼ ▼ │
│ 仅通知 仅响应 请求 请求+通知 │
│ │ │ │ │ │
│ ▼ ▼ ▼ ▼ │
│ 202 202 200/SSE 200/SSE │
│ Accepted Accepted Content-Type: │
│ application/json │
│ 或 text/event-stream │
│ │
└─────────────────────────────────────────────────────────────────┘JSON 响应格式
http
HTTP/1.1 200 OK
Content-Type: application/json
Mcp-Session-Id: abc123-xyz789
{
"jsonrpc": "2.0",
"id": 1,
"result": {
"tools": [
{"name": "read_file", "description": "Read a file"}
]
}
}SSE 流响应格式
http
HTTP/1.1 200 OK
Content-Type: text/event-stream
Cache-Control: no-cache
Connection: keep-alive
Mcp-Session-Id: abc123-xyz789
id: 1
data: {"jsonrpc":"2.0","id":1,"result":{"partial":"Hello"}}
id: 2
data: {"jsonrpc":"2.0","id":1,"result":{"partial":" World"}}
id: 3
data: {"jsonrpc":"2.0","id":1,"result":{"complete":true}}2.5 会话管理规范 ★重点★ 【强制要求】
会话生命周期
┌─────────────────────────────────────────────────────────────────┐
│ 会话生命周期 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ │
│ │ 1. 初始化 │ POST /mcp (InitializeRequest) │
│ │ 请求 │ 无 Mcp-Session-Id │
│ └──────┬───────┘ │
│ │ │
│ ▼ │
│ ┌──────────────┐ │
│ │ 2. 分配 │ 响应头: Mcp-Session-Id: <new-id> │
│ │ Session │ 服务器生成唯一 ID │
│ └──────┬───────┘ │
│ │ │
│ ▼ │
│ ┌──────────────┐ │
│ │ 3. 后续 │ POST /mcp │
│ │ 请求 │ 请求头: Mcp-Session-Id: <id> │
│ └──────┬───────┘ │
│ │ │
│ ┌────┴────┐ │
│ │ │ │
│ ▼ ▼ │
│ ┌──────┐ ┌──────┐ │
│ │正常 │ │终止 │ DELETE /mcp 或 服务器主动终止 │
│ │通信 │ │会话 │ 返回 404 表示会话已终止 │
│ └──────┘ └──────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘会话 ID 要求 【强制要求】
| 要求 | 说明 |
|---|---|
| 全局唯一 | 使用 UUID 或加密安全随机数 |
| 字符限制 | 仅可见 ASCII 字符 |
| 安全性 | 不可预测，防止会话劫持 |
| 格式建议 | UUID v4 或 JWT |
2.6 断线重连机制 ★重点★
Last-Event-ID 重传
http
# 首次连接
GET /mcp HTTP/1.1
Accept: text/event-stream
Mcp-Session-Id: abc123
# 服务器响应带 id
id: evt_001
data: {"jsonrpc":"2.0",...}
id: evt_002
data: {"jsonrpc":"2.0",...}
# 连接断开后重连
GET /mcp HTTP/1.1
Accept: text/event-stream
Mcp-Session-Id: abc123
Last-Event-ID: evt_001
# 服务器从 evt_002 开始重传
id: evt_002
data: {"jsonrpc":"2.0",...}📝 实操任务 2：手动构造 HTTP 请求
bash
# 任务：使用 curl 模拟 Streamable HTTP 通信
# 1. 发送初始化请求（无 Session ID）
curl -X POST http://localhost:8000/mcp \
-H "Content-Type: application/json" \
-H "Accept: application/json, text/event-stream" \
-d '{
"jsonrpc": "2.0",
"id": 1,
"method": "initialize",
"params": {
"protocolVersion": "2025-03-26",
"capabilities": {},
"clientInfo": {"name": "test", "version": "1.0"}
}
}' \
-i # 显示响应头，获取 Mcp-Session-Id
# 2. 使用 Session ID 发送后续请求
curl -X POST http://localhost:8000/mcp \
-H "Content-Type: application/json" \
-H "Accept: application/json, text/event-stream" \
-H "Mcp-Session-Id: <上一步获取的ID>" \
-d '{
"jsonrpc": "2.0",
"id": 2,
"method": "tools/list",
"params": {}
}'
# 3. 打开 SSE 监听流
curl -N http://localhost:8000/mcp \
-H "Accept: text/event-stream" \
-H "Mcp-Session-Id: <ID>"三、示例代码（Examples）
3.1 Python 服务端实现（FastAPI）★重点★
python
# streamable_http_server.py
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import StreamingResponse, JSONResponse
from fastapi.middleware.cors import CORSMiddleware
import asyncio
import uuid
import json
from typing import Dict, Any, Optional
from dataclasses import dataclass, field
app = FastAPI()
# CORS 配置
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["GET", "POST", "DELETE"],
allow_headers=["*"],
expose_headers=["Mcp-Session-Id"],
)
# 会话存储
sessions: Dict[str, "Session"] = {}
@dataclass
class Session:
"""会话管理"""
id: str
initialized: bool = False
event_counter: int = 0
pending_events: list = field(default_factory=list)
def create_session() -> Session:
"""创建新会话"""
session_id = str(uuid.uuid4())
session = Session(id=session_id)
sessions[session_id] = session
return session
def get_session(session_id: Optional[str]) -> Optional[Session]:
"""获取会话"""
if session_id:
return sessions.get(session_id)
return None
def json_rpc_response(id: Any, result: Any) -> dict:
"""构造 JSON-RPC 响应"""
return {"jsonrpc": "2.0", "id": id, "result": result}
def json_rpc_error(id: Any, code: int, message: str) -> dict:
"""构造 JSON-RPC 错误"""
return {"jsonrpc": "2.0", "id": id, "error": {"code": code, "message": message}}
def encode_sse(data: dict, event_id: Optional[str] = None) -> str:
"""编码为 SSE 格式"""
lines = []
if event_id:
lines.append(f"id: {event_id}")
lines.append(f"data: {json.dumps(data)}")
lines.append("")
lines.append("")
return "\n".join(lines)
async def handle_initialize(params: dict, session: Session) -> dict:
"""处理初始化请求"""
session.initialized = True
return {
"protocolVersion": "2025-03-26",
"capabilities": {
"tools": {"listChanged": True},
"resources": {"subscribe": True, "listChanged": True},
},
"serverInfo": {
"name": "Streamable HTTP Demo Server",
"version": "1.0.0"
}
}
async def handle_tools_list(params: dict, session: Session) -> dict:
"""处理工具列表请求"""
return {
"tools": [
{
"name": "echo",
"description": "Echo back the input message",
"inputSchema": {
"type": "object",
"properties": {
"message": {"type": "string", "description": "Message to echo"}
},
"required": ["message"]
}
},
{
"name": "add",
"description": "Add two numbers",
"inputSchema": {
"type": "object",
"properties": {
"a": {"type": "number"},
"b": {"type": "number"}
},
"required": ["a", "b"]
}
}
]
}
async def handle_tools_call(params: dict, session: Session) -> dict:
"""处理工具调用"""
tool_name = params.get("name")
arguments = params.get("arguments", {})
if tool_name == "echo":
return {"content": [{"type": "text", "text": arguments.get("message", "")}]}
elif tool_name == "add":
result = arguments.get("a", 0) + arguments.get("b", 0)
return {"content": [{"type": "text", "text": str(result)}]}
else:
raise ValueError(f"Unknown tool: {tool_name}")
# 方法处理器映射
METHOD_HANDLERS = {
"initialize": handle_initialize,
"tools/list": handle_tools_list,
"tools/call": handle_tools_call,
}
async def process_request(request_data: dict, session: Session) -> dict:
"""处理单个 JSON-RPC 请求"""
method = request_data.get("method")
params = request_data.get("params", {})
request_id = request_data.get("id")
handler = METHOD_HANDLERS.get(method)
if handler:
try:
result = await handler(params, session)
return json_rpc_response(request_id, result)
except Exception as e:
return json_rpc_error(request_id, -32603, str(e))
else:
return json_rpc_error(request_id, -32601, f"Method not found: {method}")
@app.post("/mcp")
async def handle_post(request: Request):
"""处理 POST 请求"""
# 获取或创建会话
session_id = request.headers.get("Mcp-Session-Id")
session = get_session(session_id)
# 解析请求体
body = await request.json()
# 处理初始化请求（创建新会话）
if isinstance(body, dict) and body.get("method") == "initialize":
session = create_session()
result = await process_request(body, session)
response = JSONResponse(content=result)
response.headers["Mcp-Session-Id"] = session.id
return response
# 非初始化请求需要有效会话
if not session:
raise HTTPException(status_code=400, detail="Missing or invalid Mcp-Session-Id")
# 处理单个请求
if isinstance(body, dict):
# 检查是否为通知（无 id）
if "id" not in body:
return JSONResponse(content=None, status_code=202)
result = await process_request(body, session)
response = JSONResponse(content=result)
response.headers["Mcp-Session-Id"] = session.id
return response
# 处理批量请求
if isinstance(body, list):
results = []
for item in body:
if "id" in item: # 只处理请求，忽略通知
result = await process_request(item, session)
results.append(result)
if not results:
return JSONResponse(content=None, status_code=202)
response = JSONResponse(content=results)
response.headers["Mcp-Session-Id"] = session.id
return response
@app.get("/mcp")
async def handle_get(request: Request):
"""处理 GET 请求（SSE 监听）"""
session_id = request.headers.get("Mcp-Session-Id")
session = get_session(session_id)
if not session:
raise HTTPException(status_code=400, detail="Missing or invalid Mcp-Session-Id")
async def event_generator():
"""生成 SSE 事件流"""
while True:
# 检查是否有待发送的事件
if session.pending_events:
event = session.pending_events.pop(0)
session.event_counter += 1
yield encode_sse(event, f"evt_{session.event_counter}")
else:
# 发送心跳保持连接
await asyncio.sleep(30)
yield ": keepalive\n\n"
return StreamingResponse(
event_generator(),
media_type="text/event-stream",
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"Mcp-Session-Id": session.id,
}
)
@app.delete("/mcp")
async def handle_delete(request: Request):
"""处理 DELETE 请求（终止会话）"""
session_id = request.headers.get("Mcp-Session-Id")
if session_id and session_id in sessions:
del sessions[session_id]
return JSONResponse(content=None, status_code=204)
raise HTTPException(status_code=404, detail="Session not found")
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)3.2 Python 客户端实现 ★重点★
python
# streamable_http_client.py
import httpx
import json
from typing import Optional, Dict, Any, AsyncGenerator
from dataclasses import dataclass
@dataclass
class MCPClient:
"""Streamable HTTP MCP 客户端"""
base_url: str
session_id: Optional[str] = None
request_id: int = 0
def _next_id(self) -> int:
"""生成下一个请求 ID"""
self.request_id += 1
return self.request_id
def _headers(self) -> Dict[str, str]:
"""构造请求头"""
headers = {
"Content-Type": "application/json",
"Accept": "application/json, text/event-stream",
}
if self.session_id:
headers["Mcp-Session-Id"] = self.session_id
return headers
async def initialize(self) -> Dict[str, Any]:
"""初始化连接"""
request = {
"jsonrpc": "2.0",
"id": self._next_id(),
"method": "initialize",
"params": {
"protocolVersion": "2025-03-26",
"capabilities": {},
"clientInfo": {"name": "Python Client", "version": "1.0"}
}
}
async with httpx.AsyncClient() as client:
response = await client.post(
f"{self.base_url}/mcp",
json=request,
headers=self._headers()
)
# 保存 Session ID
if "Mcp-Session-Id" in response.headers:
self.session_id = response.headers["Mcp-Session-Id"]
return response.json()
async def list_tools(self) -> Dict[str, Any]:
"""获取工具列表"""
request = {
"jsonrpc": "2.0",
"id": self._next_id(),
"method": "tools/list",
"params": {}
}
async with httpx.AsyncClient() as client:
response = await client.post(
f"{self.base_url}/mcp",
json=request,
headers=self._headers()
)
return response.json()
async def call_tool(self, name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
"""调用工具"""
request = {
"jsonrpc": "2.0",
"id": self._next_id(),
"method": "tools/call",
"params": {
"name": name,
"arguments": arguments
}
}
async with httpx.AsyncClient() as client:
response = await client.post(
f"{self.base_url}/mcp",
json=request,
headers=self._headers()
)
return response.json()
async def listen_events(self) -> AsyncGenerator[Dict[str, Any], None]:
"""监听 SSE 事件流"""
headers = self._headers()
headers["Accept"] = "text/event-stream"
async with httpx.AsyncClient() as client:
async with client.stream(
"GET",
f"{self.base_url}/mcp",
headers=headers
) as response:
async for line in response.aiter_lines():
if line.startswith("data: "):
data = json.loads(line[6:])
yield data
async def close(self):
"""关闭会话"""
if self.session_id:
async with httpx.AsyncClient() as client:
await client.delete(
f"{self.base_url}/mcp",
headers=self._headers()
)
self.session_id = None
# 使用示例
async def main():
client = MCPClient(base_url="http://localhost:8000")
# 1. 初始化
print("初始化连接...")
result = await client.initialize()
print(f"Session ID: {client.session_id}")
print(f"服务器信息: {result}")
# 2. 获取工具列表
print("\n获取工具列表...")
tools = await client.list_tools()
print(f"可用工具: {tools}")
# 3. 调用工具
print("\n调用 echo 工具...")
echo_result = await client.call_tool("echo", {"message": "Hello, MCP!"})
print(f"结果: {echo_result}")
print("\n调用 add 工具...")
add_result = await client.call_tool("add", {"a": 10, "b": 20})
print(f"结果: {add_result}")
# 4. 关闭会话
print("\n关闭会话...")
await client.close()
print("完成！")
if __name__ == "__main__":
import asyncio
asyncio.run(main())3.3 使用 FastMCP 快速实现 ★重点★
python
# fastmcp_server.py
# 最简化的 Streamable HTTP MCP Server 实现
from fastmcp import FastMCP
# 创建 MCP 实例
mcp = FastMCP("Demo Server 🚀")
@mcp.tool()
def add(a: int, b: int) -> int:
"""Add two numbers"""
return a + b
@mcp.tool()
def echo(message: str) -> str:
"""Echo back the message"""
return f"Echo: {message}"
@mcp.tool()
def greet(name: str) -> str:
"""Greet someone"""
return f"Hello, {name}! Welcome to MCP."
if __name__ == "__main__":
# 启动 Streamable HTTP 服务
mcp.run(
transport="streamable-http",
host="0.0.0.0",
port=8000,
path="/mcp"
)bash
# 安装 FastMCP
pip install fastmcp
# 运行服务
python fastmcp_server.py
# 测试
curl -X POST http://localhost:8000/mcp \
-H "Content-Type: application/json" \
-d '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-03-26","capabilities":{},"clientInfo":{"name":"test","version":"1.0"}}}'📝 实操任务 3：搭建完整的 MCP 服务
bash
# 任务：搭建一个支持 Streamable HTTP 的 MCP 服务
# 1. 创建项目
mkdir mcp-streamable-demo && cd mcp-streamable-demo
python -m venv venv
source venv/bin/activate
# 2. 安装依赖
pip install fastapi uvicorn httpx
# 3. 创建服务端（使用上面的代码）
# 保存为 server.py
# 4. 创建客户端（使用上面的代码）
# 保存为 client.py
# 5. 启动服务端
python server.py &
# 6. 运行客户端测试
python client.py
# 7. 验证会话管理
# - 观察 Mcp-Session-Id 是否正确传递
# - 测试 DELETE 终止会话四、常见问题（FAQ）
4.1 问题排查清单 ★重点★
问题 1：收不到 Session ID
症状：初始化响应中没有 Mcp-Session-Id 头
排查步骤：
┌─────────────────────────────────────────────────────────────────┐
│ Step 1: 检查服务端是否正确设置响应头 │
├─────────────────────────────────────────────────────────────────┤
│ response.headers["Mcp-Session-Id"] = session_id │
└─────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────┐
│ Step 2: 检查 CORS 配置是否暴露该头 │
├─────────────────────────────────────────────────────────────────┤
│ expose_headers=["Mcp-Session-Id"] # FastAPI CORS 配置 │
└─────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────┐
│ Step 3: 检查客户端是否正确读取 │
├─────────────────────────────────────────────────────────────────┤
│ session_id = response.headers.get("Mcp-Session-Id") │
└─────────────────────────────────────────────────────────────────┘问题 2：会话被意外终止（404）
症状：请求返回 404 Not Found
可能原因：
1. Session ID 无效或已过期
2. 服务器重启导致会话丢失
3. 会话被 DELETE 终止
解决方案：
┌─────────────────────────────────────────────────────────────────┐
│ // 客户端处理 404 │
│ if response.status_code == 404: │
│ # 重新初始化 │
│ self.session_id = None │
│ await self.initialize() │
└─────────────────────────────────────────────────────────────────┘问题 3：SSE 流中断
症状：GET /mcp 的 SSE 连接频繁断开
解决方案：
┌─────────────────────────────────────────────────────────────────┐
│ 方案 1: 服务端添加心跳 │
├─────────────────────────────────────────────────────────────────┤
│ async def event_generator(): │
│ while True: │
│ if has_events(): │
│ yield encode_sse(event) │
│ else: │
│ await asyncio.sleep(30) │
│ yield ": keepalive\n\n" # 心跳 │
└─────────────────────────────────────────────────────────────────┘
┌─────────────────────────────────────────────────────────────────┐
│ 方案 2: 客户端使用 Last-Event-ID 重连 │
├─────────────────────────────────────────────────────────────────┤
│ headers["Last-Event-ID"] = last_event_id │
│ # 服务器从该 ID 后继续发送 │
└─────────────────────────────────────────────────────────────────┘问题 4：响应类型不匹配
症状：期望 SSE 流但收到 JSON
原因：服务器根据请求内容决定响应类型
解决方案：
┌─────────────────────────────────────────────────────────────────┐
│ 客户端必须同时支持两种响应类型 │
├─────────────────────────────────────────────────────────────────┤
│ content_type = response.headers.get("Content-Type") │
│ │
│ if "text/event-stream" in content_type: │
│ # 处理 SSE 流 │
│ async for line in response.aiter_lines(): │
│ process_sse_line(line) │
│ elif "application/json" in content_type: │
│ # 处理 JSON │
│ data = response.json() │
│ process_json(data) │
└─────────────────────────────────────────────────────────────────┘4.2 安全注意事项 【强制要求】
| 安全要求 | 说明 | 实现方式 |
|---|---|---|
| 验证 Origin | 防止 DNS 重绑定攻击 | 检查请求头 Origin |
| 本地绑定 | 本地运行时只绑定 127.0.0.1 | host="127.0.0.1" |
| 身份验证 | 对所有连接实施认证 | Bearer Token / API Key |
| Session 安全 | 使用加密安全的 ID | UUID v4 / JWT |
📝 实操任务 4：问题排查练习
bash
# 任务：模拟并排查常见问题
# 1. 模拟 Session 过期
curl -X POST http://localhost:8000/mcp \
-H "Mcp-Session-Id: invalid-session-id" \
-H "Content-Type: application/json" \
-d '{"jsonrpc":"2.0","id":1,"method":"tools/list","params":{}}'
# 预期：400 Bad Request
# 2. 模拟缺少 Session ID
curl -X POST http://localhost:8000/mcp \
-H "Content-Type: application/json" \
-d '{"jsonrpc":"2.0","id":1,"method":"tools/list","params":{}}'
# 预期：400 Bad Request（非初始化请求）
# 3. 测试 DELETE 终止会话
curl -X DELETE http://localhost:8000/mcp \
-H "Mcp-Session-Id: <valid-session-id>"
# 预期：204 No Content
# 4. 验证终止后访问
curl -X POST http://localhost:8000/mcp \
-H "Mcp-Session-Id: <terminated-session-id>" \
-H "Content-Type: application/json" \
-d '{"jsonrpc":"2.0","id":1,"method":"tools/list","params":{}}'
# 预期：404 Not Found五、选型对比（Comparison）
5.1 MCP 传输方式对比 ★重点★
| 对比维度 | stdio | HTTP+SSE（旧） | Streamable HTTP（新） |
|---|---|---|---|
| 通信方向 | 双向同步 | 单向 SSE + HTTP POST | 双向灵活 |
| 部署方式 | 本地进程 | 远程服务 | 远程服务 |
| 端点数量 | N/A | 2 个（/sse, /messages） | 1 个（/mcp） |
| 状态管理 | 进程级 | 长连接依赖 | Session 机制 |
| 断线恢复 | 不支持 | 不支持 | 支持（Last-Event-ID） |
| 无状态服务 | 不适用 | 不支持 | 支持 |
| 云原生友好 | ❌ | ⚠️ | ✅ |
| 适用场景 | 本地工具 | 远程服务（已废弃） | 远程服务（推荐） |
5.2 Streamable HTTP vs HTTP+SSE 详细对比
┌─────────────────────────────────────────────────────────────────┐
│ HTTP+SSE vs Streamable HTTP │
├─────────────────────────────────────────────────────────────────┤
│ │
│ HTTP+SSE（旧） Streamable HTTP（新） │
│ ┌─────────────┐ ┌─────────────────┐ │
│ │ GET /sse │ 建立长连接 │ │ │
│ │ POST /msg │ 发送消息 │ POST/GET /mcp │ 统一端点 │
│ └─────────────┘ └─────────────────┘ │
│ │
│ 问题： 改进： │
│ ┌─────────────────────┐ ┌─────────────────────┐ │
│ │ ❌ 必须维护长连接 │ │ ✅ 可无状态运行 │ │
│ │ ❌ 两个端点维护复杂 │ │ ✅ 单一端点简化 │ │
│ │ ❌ 断线无法恢复 │ │ ✅ 支持断线重连 │ │
│ │ ❌ 防火墙可能阻断 │ │ ✅ 标准 HTTP 兼容 │ │
│ └─────────────────────┘ └─────────────────────┘ │
│ │
│ 性能对比（高并发场景）： │
│ ┌─────────────────────────────────────────────────┐ │
│ │ 指标 │ HTTP+SSE │ Streamable HTTP │ │
│ ├─────────────────────────────────────────────────┤ │
│ │ 平均响应时间 │ 较高 │ 更低（-30%） │ │
│ │ 响应稳定性 │ 波动大 │ 稳定 │ │
│ │ 连接资源占用 │ 高 │ 低 │ │
│ │ 实现复杂度 │ 中等 │ 较低 │ │
│ └─────────────────────────────────────────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────┘5.3 选型决策指南
┌─────────────────────────────────────────────────────────────────┐
│ MCP 传输方式选型 │
├─────────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────┐ │
│ │ 开始选型 │ │
│ └──────┬──────┘ │
│ │ │
│ ▼ │
│ ┌─────────────┐ │
│ │ 是否远程？ │ │
│ └──────┬──────┘ │
│ │ │
│ ┌────────────┼────────────┐ │
│ │ 否 │ │ 是 │
│ ▼ │ ▼ │
│ ┌─────────────┐ │ ┌─────────────┐ │
│ │ stdio │ │ │需要无状态？ │ │
│ │ 本地工具 │ │ └──────┬──────┘ │
│ └─────────────┘ │ │ │
│ │ ┌────────┼────────┐ │
│ │ │ 是 │ │ 否 │
│ │ ▼ │ ▼ │
│ ┌──────────────┐ │ ┌──────────────┐ │
│ │ Streamable │ │ │ Streamable │ │
│ │ HTTP (无状态)│ │ │ HTTP (有状态)│ │
│ └──────────────┘ │ └──────────────┘ │
│ │ │
│ 推荐：Streamable HTTP │
│ （无论有状态还是无状态） │
│ │
└─────────────────────────────────────────────────────────────────┘5.4 关键记忆点
| 场景 | 推荐方案 |
|---|---|
| 本地工具 | stdio |
| 远程服务（新项目） | Streamable HTTP |
| 远程服务（旧项目） | 逐步迁移到 Streamable HTTP |
| Serverless/K8s | Streamable HTTP（无状态模式） |
| 需要断线恢复 | Streamable HTTP |
📝 实操任务 5：选型实践
bash
# 任务：为以下场景选择合适的传输方式
# 场景 1: 本地文件系统操作工具
# 答案: stdio（本地进程通信，无需网络）
# 场景 2: 部署在 Kubernetes 的 MCP 服务
# 答案: Streamable HTTP（云原生友好，支持无状态）
# 场景 3: 需要跨网络访问的企业 MCP 服务
# 答案: Streamable HTTP（标准 HTTP 兼容，穿透防火墙）
# 场景 4: 已有 HTTP+SSE 实现的服务
# 答案: 逐步迁移到 Streamable HTTP（向后兼容）六、核心知识点速记清单
📋 30秒速记
| 项目 | 内容 |
|---|---|
| 发布时间 | 2025 年 3 月 26 日 |
| 替代方案 | HTTP+SSE |
| 核心端点 | 单一 /mcp 端点 |
| HTTP 方法 | POST（发送）、GET（监听）、DELETE（终止） |
| 消息格式 | JSON-RPC 2.0，UTF-8 |
| 会话标识 | Mcp-Session-Id 头部 |
| 断线恢复 | Last-Event-ID 机制 |
| 响应类型 | application/json 或 text/event-stream |
📋 公式总结
Streamable HTTP = 单一端点 + 按需流式 + 会话管理 + 可恢复
请求流程：
POST /mcp (Initialize) → 获取 Session ID → POST /mcp (带 ID) → ...
响应决策：
- 仅通知/响应 → 202 Accepted
- 包含请求 → 200 (JSON 或 SSE)
会话管理：
- 初始化 → 服务器分配 Mcp-Session-Id
- 后续请求 → 客户端携带该 ID
- 终止 → DELETE /mcp 或 404
与 HTTP+SSE 的关键区别：
- 端点：/sse + /messages → /mcp
- 状态：长连接依赖 → Session 机制
- 恢复：不支持 → Last-Event-ID七、学习资源与调试工具
📚 权威学习资源
| 资源 | 地址 | 说明 |
|---|---|---|
| MCP 官方规范 | https://modelcontextprotocol.io/specification/2025-03-26 | 协议规范文档 |
| MCP GitHub | https://github.com/modelcontextprotocol | 官方实现和示例 |
| FastMCP | https://github.com/jlowin/fastmcp | Python 快速开发框架 |
| 阿里云技术文章 | developer.aliyun.com/article/1661971 | 性能对比分析 |
🔧 常用调试工具
| 工具 | 用途 | 使用方式 |
|---|---|---|
| curl | 命令行 HTTP 测试 | curl -X POST http://localhost:8000/mcp ... |
| MCP Inspector | 官方可视化调试 | npx @modelcontextprotocol/inspector |
| Postman | API 测试工具 | 导入请求集合测试 |
| httpie | 友好的 HTTP 客户端 | http POST :8000/mcp ... |
调试示例
bash
# 使用 curl 完整测试流程
# 1. 初始化
curl -X POST http://localhost:8000/mcp \
-H "Content-Type: application/json" \
-H "Accept: application/json, text/event-stream" \
-d '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-03-26","capabilities":{},"clientInfo":{"name":"curl","version":"1.0"}}}' \
-i
# 2. 使用 httpie（更友好）
http POST :8000/mcp \
Content-Type:application/json \
Accept:"application/json, text/event-stream" \
Mcp-Session-Id:<session-id> \
jsonrpc=2.0 id:=2 method=tools/list params:='{}'总结：Streamable HTTP 是 MCP 协议的重大升级，通过单一端点、按需流式和会话管理机制，解决了 HTTP+SSE 的长连接压力、断线不可恢复等问题。掌握 Streamable HTTP 是构建现代远程 MCP 服务的必备技能。