AI导哈

多租户MCP代理聚合网关

位于MCP客户端与多个下游MCP服务器之间的网关，提供身份验证、租户权限管理、速率限制、使用日志记录和工具路由功能。

  • by Alexanderk30
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 Alexanderk30/ClaudeMCP

使用说明

MCP网关——多租户MCP代理和聚合服务器

位于MCP客户端（Claude、Cowork、定制代理）和 多个下游MCP服务器，添加 认证, 每位租户 权限, 速率限制, 使用日志记录，以及 刀具路径 在一个 单一过程。

┌──────────────┐
│  Tenant A    │──┐
│ (Claude)     │  │
└──────────────┘  │    ┌─────────────────────┐    ┌──────────────────┐
                  ├───▶│                     │───▶│ Filesystem MCP   │
┌──────────────┐  │    │    MCP  Gateway     │    └──────────────────┘
│  Tenant B    │──┤    │                     │    ┌──────────────────┐
│ (Cowork)     │  │    │  • Auth / Perms     │───▶│ GitHub MCP       │
└──────────────┘  │    │  • Rate Limiting    │    └──────────────────┘
                  ├───▶│  • Usage Logging    │    ┌──────────────────┐
┌──────────────┐  │    │  • Tool Routing     │───▶│ Google Drive MCP │
│  Tenant C    │──┘    │  • Proxy / Agg      │    └──────────────────┘
│ (Custom)     │       │                     │    ┌──────────────────┐
└──────────────┘       └─────────────────────┘───▶│ Custom MCP       │
                                                  └──────────────────┘

快速开始

# Install in dev mode
pip install -e ".[dev]"

# Copy and edit the example config
cp examples/tenants.yaml tenants.yaml
# → set real API keys and downstream server commands/URLs

# Start the gateway
mcp-gateway --config tenants.yaml --port 8000

# Run the tests (80 tests, ~2 seconds)
pytest

运行后，租户使用其API密钥连接到SSE端点：

GET http://localhost:8000/sse?api_key=sk-acme-XXXXXXXXXXXXXXXXXXXX

网关还为两个HTTP端点提供服务： GET /health 返回刀具计数 以及正常运行时间信息，以及 GET /stats （仅限管理员，需要 x-api-key 头球 返回每个租户的使用统计数据。

建筑

四个层，每个层都可以独立测试，组成一个请求 管道 gateway.routing.pipeline.RequestPipeline.

第1层——身份验证和权限(gateway.middleware.auth)

每个入站请求都携带一个API密钥，作为 api_key 怎么翻译 SSE端点上的参数或作为 x-api-key 头球身份验证中间件 使用定时安全验证密钥 hmac.compare_digest，解决租户问题 身份和角色，然后运行两次授权检查：第一，是否 请求的工具与租户的匹配 allowed_tools 球状图案（例如。 filesystem:*)；第二，工具的服务器前缀是否在租户的 downstream 列表。这两项检查都必须通过。

支持通过以下方式进行热重新加载 reload() 因此，租户配置可以在没有 重新启动网关。

第2层——速率限制和使用日志记录

速率限制器 (gateway.middleware.rate_limiter)--每个租户一个 滑动窗口计数器，限制每分钟的请求数。返回a RateStatus 随着 remaining 和 retry_after 字段，以便服务器层可以浮出水面 速率限制标头。 peek() 在不消耗插槽的情况下检查容量。 reload() 在速率限制未更改时保留存储桶历史记录，以及 当它们有时，会干净地重置。

使用情况记录器 (gateway.middleware.usage_logger)--异步上下文管理器 它封装了每个工具调用，记录租户ID、工具名称、延迟和 成功/失败进入有界环形缓冲区（默认50k条记录）。 query(tenant_id=..., tool_name=..., since=..., limit=...) 检索已筛选 记录，以及 stats_for() 计算包括总呼叫的聚合， 成功/失败计数、平均延迟和p99延迟。

第3层——工具路由和请求管道

路由器 (gateway.routing.router)--保存合并的工具清单。工具 名称被限定为 server:tool （例如。 github:create_issue).路由器 将合格名称解析回发起下游服务器，并 通过聚合器发送。 list_tools() 接受glob模式 每个租户过滤。

管道 (gateway.routing.pipeline)--单一入口点 组成所有层：验证→ 授权→ 速率限制→ 使用日志→ 路线→ 代理。通配符租户(allowed_tools: ["*"])自动 仅限于其允许的下游服务器。暴露 reload_config() 和 refresh_tools() 用于实时重新配置。

第4层——代理和聚合(gateway.proxy.aggregator)

聚合器通过以下方式管理MCP客户端与每个下游服务器的连接 真实的 mcp.ClientSession 通过stdio或SSE传输的实例。启动时 按顺序连接到每个服务器（anyio的任务组范围要求）， 获取分页的工具清单，并将其合并为一个 server:tool 具有重复检测的命名空间。失败的连接会被记录，但不会 阻止网关——其余服务器仍在服务流量。

refresh_manifest() 从所有下游重新获取工具以进行热重新加载。 call_downstream() 代理工具调用具有可配置的每次调用超时。

服务器层(gateway.server)

通过SSE传输将所有内容连接到真实的MCP服务器 Starlette和uvicorn。租户在 GET /sse 终点；一 contextvars-基于的方法将API密钥从HTTP层线程化到 MCP协议处理程序，因此每个SSE会话只能看到其授权的工具 为。

项目结构

mcp-gateway/
├── pyproject.toml
├── README.md
├── examples/
│   └── tenants.yaml              # Sample config (3 tenants, 4 downstreams)
├── src/gateway/
│   ├── __init__.py
│   ├── cli.py                    # Click CLI: mcp-gateway --config ... --port ...
│   ├── server.py                 # Starlette app, SSE transport, /health, /stats
│   ├── config/
│   │   └── loader.py             # YAML → Pydantic models, ${VAR} interpolation
│   ├── middleware/
│   │   ├── auth.py               # Timing-safe auth, glob ACLs, downstream scoping
│   │   ├── rate_limiter.py       # Sliding-window rate limiter with status/peek
│   │   └── usage_logger.py       # Ring-buffer logger with query/stats
│   ├── routing/
│   │   ├── router.py             # Qualified-name resolution + glob filtering
│   │   └── pipeline.py           # Full middleware chain composition
│   ├── proxy/
│   │   └── aggregator.py         # MCP ClientSession management + manifest merge
│   └── utils/
│       ├── env.py                # ${VAR} and ${VAR:-default} interpolation
│       └── errors.py             # GatewayError hierarchy + MCP error formatting
└── tests/
    ├── mock_downstream.py        # Minimal MCP server (echo + add) for integration tests
    ├── test_boot_smoke.py        # End-to-end: boot → connect → proxy → HTTP
    ├── test_aggregator.py        # Layer 4 unit tests (12 tests)
    ├── test_pipeline.py          # Layer 3 pipeline tests (10 tests)
    ├── test_auth.py              # Layer 1 auth tests (14 tests)
    ├── test_rate_limiter.py      # Layer 2a rate limiter tests (8 tests)
    ├── test_usage_logger.py      # Layer 2b usage logger tests (9 tests)
    ├── test_server.py            # HTTP endpoint tests (6 tests)
    ├── test_config.py            # Config loading + env interpolation (7 tests)
    └── test_utils.py             # Env/error utility tests (11 tests)

配置

所有租户和下游服务器定义都存在于一个YAML文件中。看 examples/tenants.yaml 对于带有注释的完整模式。

租户 --每个条目定义一个 api_key, role (admin/editor/viewer), allowed_tools （球状图案如 filesystem:* 或 github:create_issue), rate_limit （请求/分钟），以及 downstream 租户可以访问的服务器。

下游服务器 --每个条目指定一个 transport (stdio 或 sse) 以及连接详细信息（stdio的command+args，SSE的URL）。

环境变量插值 — ${VAR} 和 ${VAR:-default} 语法 支持 api_key上海证券交易所 url，以及stdio env 领域。这保持 配置文件中的秘密：

tenants:
  acme:
    api_key: "${ACME_API_KEY}"
    # ...

downstream_servers:
  github:
    transport: stdio
    command: "npx"
    args: ["-y", "@modelcontextprotocol/server-github"]
    env:
      GITHUB_PERSONAL_ACCESS_TOKEN: "${GITHUB_TOKEN}"

  gdrive:
    transport: sse
    url: "${GDRIVE_URL:-http://localhost:3002/sse}"

延伸

添加新的下游 --向添加条目 downstream_servers 在你的YAML中， 然后在每个租户的 downstream 列表。网关发现 工具在下次启动时自动启动。

自定义角色 --the role 字段验证为 admin|editor|viewer 但是 网关仅将其用于 /stats 端点（仅限管理员）。工具级别 访问完全由以下人员控制 allowed_tools globs，所以角色是 您自己的策略层的约定。

交换存储后端 — UsageLogger 在内存中保存记录 deque.将其子类化并重写记录以写入SQLite、Postgres、， 或指标后端。这 query() 和 stats_for() 界面保持不变。

热重载 --呼叫 pipeline.reload_config(new_config) 更新租户 无需重新启动即可进行身份验证和速率限制，以及 pipeline.refresh_tools() 到 重新获取下游工具清单。

相关 MCP