实现一个支持 Streamable HTTP 的 MCP 服务器及客户端

在现代的 Web 开发中，实时数据传输是一个常见的需求。本文将介绍如何使用 FastAPI 实现一个支持 Streamable HTTP 的 MCP（Model Context Protocol）服务器，并提供 Python 客户端和前端客户端的实现。

1. 什么是 Streamable HTTP 和 MCP？

Streamable HTTP 是一种允许服务器以流的形式向客户端发送数据的技术。这在处理长时间运行的操作或实时数据更新时非常有用。

MCP（Model Context Protocol） 是一种协议，用于在客户端和服务器之间传输模型上下文信息。它支持初始化、消息传输和进度跟踪等功能。

2. 服务器端实现

我们将使用 FastAPI 来实现一个支持 Streamable HTTP 的 MCP 服务器。FastAPI 是一个现代、快速的 Web 框架，基于 Python 类型提示，支持异步操作。

安装 FastAPI 和 Uvicorn

首先，确保你已经安装了 FastAPI 和 Uvicorn。运行以下命令进行安装：

pip install fastapi uvicorn

服务器端代码

以下是服务器端的完整代码：

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import StreamingResponse, JSONResponse
from fastapi.middleware.cors import CORSMiddleware
import asyncio
import uuid

app = FastAPI()

# 配置 CORS
app.add_middleware(
CORSMiddleware,
allow_origins=[\”*\”], # 允许所有来源
allow_credentials=True,
allow_methods=[\”*\”], # 允许所有 HTTP 方法
allow_headers=[\”*\”], # 允许所有头部
expose_headers=[\”Mcp-Session-Id\”], # 允许客户端访问的自定义头
)

# 模拟存储会话信息
sessions = {

}

@app.post(\”/message\”)
async def handle_message(request: Request):
session_id = request.headers.get(\”Mcp-Session-Id\”)
if not session_id:
session_id = str(uuid.uuid4())

# 获取请求数据
data = await request.json()
print(f\”Received message: {

data}\”)

# 模拟处理请求并发送响应
response_data = {

\”result\”: \”Request processed\”}

# 返回 JSON 响应，并在头中包含 Mcp-Session-Id
return JSONResponse(content={

\”jsonrpc\”: \”2.0\”, \”id\”: data.get(\”id\”), \”result\”: response_data},
headers={

\”Mcp-Session-Id\”: session_id})

@app.get(\”/message\”)
async def handle_sse(request: Request):
session_id = request.headers.get(\”Mcp-Session-Id\”) or request.query_params.get(\”Mcp-Session-Id\”)
if not session_id:
raise HTTPException(status_code=400, detail=\”Session ID is required\”)

async def event_generator():
for i in range(5):
await asyncio.sleep(1) # 模拟延迟
yield f\”data: Message {

i + 1}\\n\\n\”

return StreamingResponse(event_generator(), media_type=\”text/event-stream\”)

if __name__ == \”__main__\”:
import uvicorn

uvicorn.run(app, host=\”0.0.0.0\”, port=8000)

服务器端说明

3. Python 客户端实现

接下来，我们实现一个 Python 客户端，用于与服务器进行交互。

安装依赖

确保你已经安装了 requests 库。如果尚未安装，可以运行以下命令进行安装：

pip install requests

客户端代码

以下是 Python 客户端的完整代码：