AI导哈

会话存储服务

为MCP服务器提供分布式会话协调和事件存储功能，支持高可用性和水平扩展。

  • by bh-rat
  • Python
  • 本地部署
  • 8 星标
  • 更新于
在 GitHub 查看 bh-rat/mcp-db

使用说明

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

⚠️ 实验图书馆：该项目处于实验阶段，尚未发布到PyPI。要使用它，您需要直接从源代码安装。

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

特性

  • 传输级拦截（ASGI）——无处理程序更改
  • 持久会话（使用服务器提供的 Mcp-Session-Id)+活动采购
  • Redis存储（JSON格式的会话；Redis流中的事件）
  • 跨节点准入——在任何节点上重建SDK传输
  • 主动会话的可选每个节点预热（内部 notifications/initialized)
  • 示例：单节点、轮询负载均衡器、Redis监视器

用例

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

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

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

  • 水平缩放：在负载平衡器后面运行MCP服务器，而不会出现粘滞会话。任何服务器实例都可以处理对现有会话的任何请求。

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

需求

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

安装

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

# Clone the repository
git clone https://github.com/yourusername/mcp-db.git
cd mcp-db

# Install with uv (recommended)
uv pip install -e .

# Or install with pip
pip install -e .

# For Redis support
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

# Assume you have MCP SDK session_manager from your app

# Setup mcp-db components
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,  # optional: lets the wrapper consult storage
).wrap(handle_streamable_http)
  1. 安装 wrapped_asgi 在MCP端点路径上连接到您的ASGI应用程序（例如。， /mcp).

分布式扩展快速入门

运行两台服务器和一个简单的循环LB，交替请求：

# Terminals 1 & 2: run servers
uv run python -m examples.streamable_http_server --port 3001
uv run python -m examples.streamable_http_server --port 3002

# Terminal 3: run the LB
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

# Terminal 4: run the client via the LB
# Streamable HTTP (POST+SSE)
uv run python -m examples.streamable_http_client --base http://127.0.0.1:3000/mcp/ --shttp

# Or JSON mode
uv run python -m examples.streamable_http_client --base http://127.0.0.1:3000/mcp/ --no-shttp

# Exercise follow-up calls on an existing session (works across nodes)
uv run python -m examples.continue_session <SESSION_ID> --base http://127.0.0.1:3000/mcp/ --shttp

它是如何工作的：

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

可恢复性演示（上次事件ID）

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

# MCP example server (includes request-tied SSE and standalone GET stream)
# In-memory EventStore:
uv run python -m examples.streamable_http_server --port 3000 --event-store memory
# Or Redis-backed 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

# Minimal SSE-only demo
uv run python -m examples.resumability_demo

# Terminal A: stream SSE (no Last-Event-ID initially)
curl -N http://127.0.0.1:5050/sse

# Terminal B: publish some JSON-RPC messages (each becomes an SSE event with an id)
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"}'

# Simulate disconnect, then resume with Last-Event-ID=<last seen id> to receive only later events
curl -N -H 'Last-Event-ID: <paste-last-id-here>' 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的节点上；
  • 持久性≠直播流：存储会话/事件不会重新创建任何io流；使用所有权路由或重新连接+Last-Event-ID 重播。
  • 支持SDK event_store 待定：sdk支持的事件存储仍处于待定状态。

计划的

  • 所有权路由示例；SDK事件存储布线；探索 mcp_db.transport 以增强跨节点连续性。

注意事项和提示

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

许可证

阿帕奇-2.0

相关 MCP