DataPulseSSE——基于SSE的轻量级MCP服务器（FastAPI）
一个使用FastAPI构建的小型、专注的MCP（模型上下文协议）服务器，它：
- 维护每个会话的SSE（服务器发送事件）流
- 接受MCP样式的消息，并通过SSE异步发出结果
- 支持简单的承载身份验证（本地）和可选的JWT验证（RS256）
- 附带示例工具（get_time、fetch_url）
它在本地运行（Uvicorn），可以托管在Azure Functions上（通过一个小型适配器）。
目录
- 需求
- 配置
- 快速启动（本地）
- 授权
- API端点
- 示例（卷曲）
- 浏览器客户端（EventSource）
- 数据模型
- 会话和队列
- 内置工具
- 在Azure函数上运行
- 安全和生产说明
- 限制/TODO
需求
- Python 3.10+
- 点
配置
环境变量：
- LOCAL_BARER_TOKEN--开发人员的本地BEARER令牌（默认值：
secret_token) - JWT_ISSUER——可选，JWT RS256发卡机构（OIDC发卡机构URL）
- JWT_AUDIENCE——可选的、预期的JWT观众
- JWT_KEY——可选的PEM公钥，用于验证RS256令牌（如果您的身份验证代码支持静态密钥）
示例（bash）：
export LOCAL_BEARER_TOKEN="secret_token"
# Optional JWT (RS256) verification:
export JWT_ISSUER="https://login.microsoftonline.com//v2.0"
export JWT_AUDIENCE="api://your-app-id"
export JWT_KEY="$(cat public_key.pem)"注意：对于针对旋转密钥的生产级JWT验证，请使用JWKS发现（首选）而不是静态JWT_KEY。
快速启动（本地）
- 安装依赖项：
pip install -r requirements.txt
# If needed (not present in your requirements.txt):
pip install fastapi uvicorn- 运行服务器：
uvicorn app.main:app --reload- 打开SSE流（在新终端中）：
# Use the same session_id for /events and /message
curl -N \
-H "Authorization: Bearer ${LOCAL_BEARER_TOKEN:-secret_token}" \
"http://localhost:8000/events?session_id=abc123"提示： -N 禁用curl中的缓冲，这对SSE很重要。
授权
- 本地承运人：发送
Authorization: Bearer secret_token（或您的LOCAL_BARER_TOKEN）。 - JWT（RS256）：如果通过环境变量配置，服务器将进行验证
Authorization: Bearer并拒绝无效令牌。
如果没有提供有效的身份验证，服务器将返回401 Unauthorized。
API端点
GET/事件
- 查询：
session_id（必填） - 标题：
Authorization: Bearer ... - 返回SSE流。消息以行的形式发送：
data: {"type":"...","request_id":"...","data":{...}}
- Keep‑alive comments
: ping在空闲时定期发送。
状态代码：
- 200--溪流打开
- 401--未经授权
POST/消息
- 查询：
session_id（必填） - 标题：
Authorization: Bearer ...,Content-Type: application/json - 正文（MCP请求）：
{
"type": "list_tools" | "call_tool",
"request_id": "req-123",
"data": {
"name": "get_time" | "fetch_url",
"url": "https://example.com"
}
}行为：
- 验证会话（TTL~1h）
- 处理请求并将该会话的结果排队
- 返回确认（实际结果到达SSE流）
示例（卷曲）
列出工具：
curl -X POST "http://localhost:8000/message?session_id=abc123" \
-H "Authorization: Bearer ${LOCAL_BEARER_TOKEN:-secret_token}" \
-H "Content-Type: application/json" \
-d '{"type":"list_tools","request_id":"req-1","data":{}}'调用get_time：
curl -X POST "http://localhost:8000/message?session_id=abc123" \
-H "Authorization: Bearer ${LOCAL_BEARER_TOKEN:-secret_token}" \
-H "Content-Type: application/json" \
-d '{"type":"call_tool","request_id":"req-2","data":{"name":"get_time"}}'调用fetch_url：
curl -X POST "http://localhost:8000/message?session_id=abc123" \
-H "Authorization: Bearer ${LOCAL_BEARER_TOKEN:-secret_token}" \
-H "Content-Type: application/json" \
-d '{"type":"call_tool","request_id":"req-3","data":{"name":"fetch_url","url":"https://example.com"}}'SSE事件示例：
data: {"type":"tool_result","request_id":"req-2","data":{"time":"2025-08-31T12:00:00Z"}}浏览器客户端（EventSource）
const sessionId = "abc123";
const url = `http://localhost:8000/events?session_id=${encodeURIComponent(sessionId)}`;
const es = new EventSource(url);
es.onmessage = (e) => {
console.log("SSE message:", e.data);
};
es.onerror = (e) => {
console.warn("SSE error:", e);
};
重要提示：浏览器不允许在EventSource中使用自定义标头，因此您无法直接附加授权。对于浏览器使用，放置一个网关/代理（例如APIM、反向代理），终止身份验证并转发到后端。
数据模型
MCPRequest（简称）：
{"type":"list_tools|call_tool","request_id":"string","data":{...}}MCP响应：
{"type":"list_tools|tool_result|error","request_id":"string","data":{...}}错误：
{"type":"error","request_id":"string","error":"..."}会话和队列
- 会话在进程内存储（TTL~1h）。
- 每个会话都有自己的消息队列。
- 始终使用相同的
session_id用于SSE流和POST消息。
内置工具
- list_tools--返回可用的工具
- get_time--返回当前UTC时间（ISO 8601，Z后缀）
- fetch_url（url）--从url获取内容（短超时），返回HTTP状态和内容片段
在Azure函数上运行
- 该仓库包括适配器（例如。，
app/azure_adapter.py)使用Mangum在Azure Functions HTTP触发器上公开FastAPI应用程序。 - 在功能应用程序中：
- 配置适配器的入口点 - 像在本地一样设置环境变量
安全和生产说明
- 首选网关（例如Azure API管理）来处理OAuth/PKCE、速率限制和注入后端密钥。
- 对于JWT，支持JWKS发现（旋转密钥）而不是静态公钥。
- SSE可靠性：保持生命释放；考虑连接限制和适当的清理。
- 可观察性：添加结构化日志记录和基本指标（例如请求计数、队列深度）。
- 横向扩展：内存会话不会跨实例扩展；如果需要，使用共享存储（例如Redis）。
限制/TODO
- 没有完整的MCP授权流（具有加密会话密钥的OAuth/PKCE）
- 仅限内存会话/队列（无跨实例共享）
- 默认情况下为最小身份验证；JWT是可选的，需要配置
- 默认情况下不包括CI/CD测试自动化