STARWARCLOUD STARWAR.AI
返回市场
MCP数据库

MCP数据库

作者：bh-rat8 星标更新：2025-08-11
安装使用来源：spider==github.com

项目介绍

mcp-db - MCP服务器会话和事件存储实现

License Python Version Project Status: Experimental

⚠️ 实验性库：此项目处于实验阶段，尚未发布到PyPI。要使用它，您需要从源代码安装。

分布式会话协调用于模型上下文协议（MCP）服务器。它拦截MCP流式HTTP流量（JSON + SSE），持久化会话和事件，并启用跨节点准入/预热，以便任何节点都可以在负载均衡器后面提供任何现有会话。

功能

  • 运输层拦截（ASGI）——无需更改处理器
  • 持久会话（使用服务器提供的Mcp-Session-Id）+事件溯源
  • Redis存储（会话作为JSON；事件在Redis流中）
  • 跨节点准入——在任何节点上重建SDK传输
  • 活跃会话的可选节点预热（内部notifications/initialized）
  • 示例：单节点，轮询负载均衡器，Redis监控

使用场景

  • 持久会话存储：在Redis或其他后端存储MCP会话和事件流。会话在服务器重启后仍然存在，并且可以查询以恢复状态。

  • 高可用性：通过自动会话故障转移消除单点故障。当服务器崩溃或重启时，其他节点无缝接管活跃会话。

  • 零停机部署：在执行滚动更新和蓝绿部署的同时保持会话连续性。新实例立即从Redis提供现有会话。

  • 水平扩展：在负载均衡器后面运行MCP服务器，而无需粘性会话。任何服务器实例都可以处理现有会话的任何请求。

  • 审计与合规性：通过事件溯源捕获完整的会话历史。每个协议消息都被持久化，用于调试、监控和监管合规性。

要求

  • Python >= 3.10
  • MCP SDK 1.12.x
  • Redis >= 7（用于流）

安装

由于此库尚未发布到PyPI，请从源代码安装：

# 克隆仓库
git clone https://github.com/yourusername/mcp-db.git
cd mcp-db

# 使用uv安装（推荐）
uv pip install -e .

# 或者使用pip安装
pip install -e .

# 为Redis支持
pip install -e ".[redis]"

快速开始（流式HTTP）

  1. 使用MCP SDK创建您的MCP服务器（StreamableHTTPSessionManager）。

  2. 将包装器与基于存储的SessionManager + EventStore连接：

from mcp_db.core.session_manager import SessionManager
from mcp_db.core.interceptor import ProtocolInterceptor
from mcp_db.core.asgi_wrapper import ASGITransportWrapper
from mcp_db.core.admission import StreamableHTTPAdmissionController
from mcp_db.session import RedisStorage

# 假设您已经从应用中获取了MCP SDK的session_manager

# 设置mcp-db组件
storage = RedisStorage(url="redis://localhost:6379/0", prefix="mcp")
db_session_manager = SessionManager(storage=_storage, event_store=None)
interceptor = ProtocolInterceptor(db_session_manager)
admission = StreamableHTTPAdmissionController(manager=session_manager, app=app)

async def handle_streamable_http(scope, receive, send):
    await session_manager.handle_request(scope, receive, send)

wrapped_asgi = ASGITransportWrapper(
    interceptor,
    admission_controller=admission,
    session_lookup=db_session_manager.get,  # 可选：允许包装器咨询存储
).wrap(handle_streamable_http)
  1. 在您的MCP端点路径（例如，/mcp）处将wrapped_asgi挂载到您的ASGI应用程序。

分布式扩展快速入门

运行两个服务器和一个简单的轮询负载均衡器，交替请求：

# 终端1和2：运行服务器
uv run python -m examples.streamable_http_server --port 3001
uv run python -m examples.streamable_http_server --port 3002

# 终端3：运行负载均衡器
uv run python -m examples.round_robin_lb \
  --listen-port 3000 \
  --backend http://127.0.0.1:3001 \
  --backend http://127.0.0.1:3002

# 终端4：通过负载均衡器运行客户端
# 流式HTTP（POST+SSE）
uv run python -m examples.streamable_http_client --base http://127.0.0.1:3000/mcp/ --shttp

# 或JSON模式
uv run python -m examples.streamable_http_client --base http://127.0.0.1:3000/mcp/ --no-shttp

# 对现有会话进行后续调用（跨节点工作）
uv run python -m examples.continue_session <SESSION_ID> --base http://127.0.0.1:3000/mcp/ --shttp

工作原理：

  • 初始化：SDK创建传输并返回Mcp-Session-Id；包装器持久化会话记录（初始化）。
  • notifications/initialized：包装器在存储中标记会话为活动。
  • 在任何节点上的后续请求：包装器在SDK检查其内存映射之前为该会话ID重建SDK传输。对于活动会话，包装器可以在节点上预热一次。

可恢复性演示（Last-Event-ID）

运行MCP示例（使用SDK的EventStore）或最小的SSE演示：

# MCP示例服务器（包括请求绑定的SSE和独立GET流）
# 内存中的EventStore：
uv run python -m examples.streamable_http_server --port 3000 --event-store memory
# 或Redis支持的EventStore：
uv run python -m examples.streamable_http_server --port 3000 --event-store redis --redis-url redis://localhost:6379/0 --redis-prefix mcp
uv run python -m examples.streamable_http_client --base http://127.0.0.1:3000/mcp/ --shttp

# 最小的SSE仅演示
uv run python -m examples.resumability_demo

# 终端A：流式SSE（初始无Last-Event-ID）
curl -N http://127.0.0.1:5050/sse

# 终端B：发布一些JSON-RPC消息（每个都成为具有ID的SSE事件）
curl -s -X POST http://127.0.0.1:5050/publish \
  -H 'Content-Type: application/json' \
  -d '{"jsonrpc":"2.0","id":1,"method":"tools/list"}'

# 模拟断开连接，然后使用Last-Event-ID=<最后看到的ID>恢复，只接收后来的事件
curl -N -H 'Last-Event-ID: <粘贴最后的ID>' http://127.0.0.1:5050/sse

行为：

  • 每流ID：ID按流分配，并仅在该流内充当游标。
  • 恢复重放：当提供Last-Event-ID时，仅重放同一流中该ID之后的事件；其他流不会被重放。
  • 实时延续：重放后，新事件实时交付。

短（快速尝试）：

uv run python -m examples.resumability_demo
curl -N http://127.0.0.1:5050/sse
curl -N -H 'Last-Event-ID: <id>' http://127.0.0.1:5050/sse

限制（简洁）

  • GET/SSE（客户端功能如采样/引诱）是针对每个节点的：实时流/队列仅存在于接受GET的节点上；
  • 持久化≠实时流：存储会话/事件不会重新创建anyio流；使用所有权路由或重新连接+Last-Event-ID重放。
  • SDK支持的event_store待定：SDK支持的事件存储仍待定。

计划

  • 所有权路由示例；SDK EventStore连接；探索mcp_db.transport以增强跨节点连续性。

注意事项及技巧

  • 客户端必须发送Accept: application/json, text/event-stream和Content-Type: application/json。
  • 对于流式HTTP，解析SSE data:行以获取响应/事件。
  • 包装器从不生成会话ID；它存储由服务器传输提供的ID。
  • 优先使用uv run来运行示例。

许可证

Apache-2.0