universal-mcp-gateway

<p align="center"> <img src="assets/banner.svg" alt="通用MCP网关 — 在您运行的每个MCP服务器前面的一个受控门户" width="100%"> </p> # 通用MCP网关 [](https://github.com/MANOJ21K/universal-mcp-gateway/actions/workflows/ci.yml) [](https://www.python.org/) [](https://github.com/jlowin/fastmcp) [](https://modelcontextprotocol.io) [](LICENSE) 运行一个MCP服务器，您可以合理地在脑海中保持授权。运行三个，每个服务器都悄悄成为自己的独立领地，任何客户端都可以调用任何工具，没有人检查谁在请求，也没有记录后续发生了什么 — 这个空白点在代理执行了不应该执行的操作之前一直是不可见的。 通用MCP网关在不触及服务器本身的情况下填补了这一空白。它是一个MCP服务器，适用于连接到它的任何东西，以及一个MCP客户端，适用于它所连接的任何东西，因此代理与它通信就像与真正的工具服务器通信一样 — 只不过现在每个调用都经过身份验证，根据策略检查，记录和速率限制。它的所有强制执行内容 — 哪些服务器存在，谁可以调用什么，如何调用，什么被删减 — 都存在于一个YAML文件中，因此“通用”不是一个口号：同一个网关支持计算器，键值存储或内部MCP服务而无需代码更改，只需配置即可。它建立在FastMCP 3.4.4的基础上，用于实际的代理；这个存储库是建立在它之上的治理层。 ## 为什么是网关，而不是每个服务器的补丁 显而易见的替代方案是为每个MCP服务器单独添加身份验证，策略检查和审计日志记录。实际上，这种做法往往会逐渐演变：第二个服务器实现方式略有不同，第三个可能是供应商的服务器，您无法修改，到了添加更多服务器时，很难自信地说出给定的调用者是否被允许执行特定操作，而不阅读涉及的每个代码库。通过在一个地方维护治理，所有操作都以相同的方式工作，无论您是否拥有上游服务器的代码，还是有一个单一的，可审查的答案来回答“谁可以调用什么” — 而不是每个服务器一个答案。 --- ## 目录 - [五分钟内看到它强制执行策略](#五分钟内看到它强制执行策略) - [它的工作原理](#它的工作原理) - [您获得的功能](#您获得的功能) - [配置网关](#配置网关) - [在生产环境中运行](#在生产环境中运行) - [测试](#测试) - [相关内容的位置](#相关内容的位置) - [接下来](#接下来) - [贡献](#贡献) - [许可证](#许可证) --- ## 五分钟内看到它强制执行策略 需要Python 3.10+（持续集成运行3.10-3.12）。Redis和Docker是可选的 — 仅在整个速率限制或容器化堆栈时需要。 ```powershell python -m venv .venv .\.venv\Scripts\Activate.ps1 # macOS/Linux: source .venv/bin/activate pip install -e ./agentctx # 依赖网关的本地 sibling 包 pip install -e . # 网关本身 + `mcp-gateway` CLI ``` 启动两个示例上游服务 — 计算器和内存键值存储 — 每个终端都在自己的终端中： ```powershell python examples/server.py # calc → http://127.0.0.1:8100/mcp (8100, 不是 8000 — 8000 在本地被占用) python examples/mock_kv_server.py # kv → http://127.0.0.1:8001/mcp ``` 然后是网关，默认读取 [`gateway.config.yaml`](gateway.config.yaml)： ```powershell mcp-gateway run # → http://127.0.0.1:9000/mcp ``` 现在值得关注的部分。已配置的配置文件定义了三个具有不同规则集的主体 — 更改下面的API密钥而无需其他更改，网关强制执行不同的现实： ```python import asyncio from fastmcp import Client from fastmcp.client.auth import BearerAuth async def main(): async with Client("http://127.0.0.1:9000/mcp", auth=BearerAuth("sk-analyst-abc")) as client: print([t.name for t in await client.list_tools()]) result = await client.call_tool("calc_query_tool", {"user_input": "add 2 and 3"}) print(result.structured_content) asyncio.run(main()) ``` | 主体 | 密钥 | 角色 | 允许 | 阻止 | 限制 | |---|---|---|---|---|---| | `analyst-1` | `sk-analyst-abc` | `analyst` | `calc_*` | 所有 `kv_*` | 30/分钟 | | `editor-1` | `sk-editor-def` | `editor` | `calc_*`, `kv_*` | `kv_delete` （显式拒绝） | 60/分钟 | | `admin-1` | `sk-admin-xyz` | `admin` | 所有 | 无 | 120/分钟 | 以分析师的身份运行，`list_tools()` 永远不会提及 `kv_*` — 不是过滤后，而是从菜单中完全缺失。无论如何调用 `kv_set`，都会得到 `Access denied`，因为网关在调用时重新检查，而不管它宣传了什么。换成编辑器的密钥，`kv_get`/`kv_set` 生效，但 `kv_delete` 不生效，即使允许 `kv_*` — 角色的显式拒绝覆盖了自己的允许。换成管理员的密钥，所有操作都生效，`kv_delete` 在每次使用时在审计日志中标记为 `sensitive`。整个过程中，客户端代码保持不变；只有密钥发生变化。 [`examples/client.py`](examples/client.py) 将相同的调用包装在一个小的FastAPI服务中 — 一个代表实际面向代理的应用程序的占位符 — 其身份由 `GATEWAY_API_KEY` 环境变量设置。并且在任何涉及部署之前，无需服务器即可验证配置： ```powershell mcp-gateway check --config gateway.config.yaml # 遇到任何错误时退出状态非零 — 一个CI门控 ``` --- ## 它的工作原理 ```mermaid flowchart LR Agent["代理或应用程序<br/>examples/client.py"] -->|Bearer API 密钥| GW subgraph GW["通用MCP网关 — 端口9000"] direction TB Auth["1. 身份验证<br/>密钥或JWT转换为主体"] Audit["2. 审计<br/>外部，馈送/ metrics"] Policy["3. 策略<br/>RBAC + 通配符 + 租户"] Resilience["4. 弹性<br/>限制，断路器，重试"] Auth --> Audit --> Policy --> Resilience end Resilience -->|联合| Calc["计算服务器<br/>端口8100"] Resilience -->|联合| KV["键值存储服务器<br/>端口8001"] ``` 首先解析身份，因为下游的所有内容都需要知道谁在请求。之后，嵌套顺序是一个刻意的选择，而不是一个意外： ``` 审计 → 策略 → 弹性 → 上游 ``` 审计位于外部，因此它会看到调用的实际结果 — 包括内部层抛出的拒绝和限流 — 而不仅仅是记录成功。策略位于弹性之前，因此一个永远不会被允许的调用不会在被拒绝之前花费任何速率限制预算或重试插槽。 工具通过上游密钥注册命名空间：`calc` 的 `query_tool` 成为 `calc_query_tool`，`kv` 的 `get`/`set`/`delete` 成为 `kv_get`/`kv_set`/`kv_delete`。每个策略通配符都与命名空间名称匹配 — 即使只配置了一个上游，因此策略永远不必随着后续添加服务器而更改。 | 层 | 责任 | 源代码 | |---|---|---| | 身份验证 | `Authorization: Bearer …` → 主体，或在其他任何内容运行之前返回401 | [`gateway/auth.py`](gateway/auth.py)，[`gateway/identity.py`](gateway/identity.py) | | 策略 | 允许 ∧ ¬ 拒绝工具/资源/提示；过滤菜单 * 和 重新检查每个调用 | [`gateway/policy_engine.py`](gateway/policy_engine.py)，[`gateway/middleware/policy.py`](gateway/middleware/policy.py) | | 审计 | 每次尝试的JSONL + Prometheus计数器 + 可选的OpenTelemetry跨度 | [`gateway/middleware/audit.py`](gateway/middleware/audit.py)，[`gateway/observability.py`](gateway/observability.py) | | 弹性 | 速率限制器，断路器，超时，重试 + 回退 | [`gateway/middleware/resilience.py`](gateway/middleware/resilience.py)，[`gateway/circuit.py`](gateway/circuit.py)，[`gateway/ratelimit.py`](gateway/ratelimit.py) | --- ## 您获得的功能 | | | |---|---| | **身份验证** | 两种模式：JWT来自您的IdP（HS/RS/ES/PS，JWKS，声明映射 — 生产姿态，无用户列表在配置中；参见 [docs/production-auth.md](docs/production-auth.md)）或静态API密钥（存储哈希，可旋转，过期 — 零设置演示模式）。可选发布RFC 9728 OAuth受保护资源元数据，以便客户端可以自动发现您的授权服务器。 | | **策略** | RBAC与整个MCP表面的shell-glob模式 — 工具，资源和提示，而不仅仅是工具。拒绝始终优先于允许。 | | **审计** | 每个尝试，无论是否允许，都成为一个结构化的JSON行，敏感参数被删减 — 提供Prometheus指标和可选的OpenTelemetry跨度。 | | **弹性** | 每调用者速率限制（内存或Redis），每个上游的断路器，每个调用的超时，重试和回退。 | | **联合** | 通过编辑配置添加上游服务 — HTTP或stdio；其工具自动出现，命名空间化，因此策略永远不必猜测哪个服务器拥有哪个工具。 | | **多租户** | 用所有者租户标记上游服务，来自任何其他租户的调用者都无法访问它，即使他们的角色可能允许调用。 | | **运维就绪** | `/healthz`，`/readyz`，Prometheus `/metrics`，Dockerfile + Compose堆栈和 `mcp-gateway check` 作为CI配置门控。 | 网关旁边是 [`agentctx`](#the-agentctx-package) — 一个身份和跟踪上下文库，没有依赖于FastMCP或此网关，可用于任何代理代码库，而不仅仅是这个。 --- ## 配置网关 [`gateway.config.yaml`](gateway.config.yaml) 是行为的唯一配置文件 — 这是整个契约，已注释： ```yml gateway: host: 127.0.0.1 port: 9000 request_timeout_s: 15 # 每个调用的上游超时 audit_log_path: audit.log # JSONL接收器；按大小旋转 audit_max_bytes: 10485760 # 10 MiB，0 = 永远不旋转 audit_backup_count: 5 # 联邦注册表 — 此网关支持的每个上游服务。工具按此处键命名空间： # calc -> calc_<tool>，kv -> kv_<tool> — 即使您只配置一个上游服务，也使用相同的前缀规则， # 因此策略通配符永远不必更改。 upstreams: calc: url: "http://127.0.0.1:8100/mcp" transport: http kv: url: "http://127.0.0.1:8001/mcp" transport: http tenant: acme # 只有tenant=acme的调用者才能访问此上游服务的 # 工具/资源/提示 — 关闭后，即使RBAC允许调用 files: # ... 或支持stdio的服务， transport: stdio # 作为子进程启动 — 不需要HTTP端点， command: "npx" # 本地安装的MCP服务器的常见情况 args: ["-y", "@modelcontextprotocol/server-filesystem", "/data"] env: { LOG_LEVEL: "info" } # 可选 # cwd: /srv # 可选的工作目录用于子进程 # 身份验证 — 允许调用哪些主体，及其成为哪些主体。密钥 # 支持 ${ENV_VAR} / ${ENV_VAR:-default} 插值（以及此文件旁的可选.env） # 并存储哈希值（sha256）；明文在加载后被丢弃。 # 单个 `api_key: "${VAR}"` 字段如果不需要轮换则有效； # 以下 `api_keys` 是多密钥表单 — 给主体两个密钥， # 在旧密钥过期之前轮换，无需停机，两个密钥在此期间都有效。 principals: - id: analyst-1 roles: [analyst] tenant: acme api_keys: - { key: "${ANALYST_KEY_NEW}" } - { key: "${ANALYST_KEY_OLD}", expires_at: "2026-08-01T00:00:00Z" } # 策略 — 工具在满足某些角色的允许且不满足任何角色的拒绝时可调用 # （拒绝始终优先）。具有多个角色的调用者获得角色的并集。 roles: analyst: allow: ["calc_*"] deny: [] rate_limit_per_min: 30 editor: allow: ["calc_*", "kv_*"] deny: ["kv_delete"] # kv_* 已允许，但此项仍优先 rate_limit_per_min: 60 sensitive_tools: ["kv_delete"] # 提升审计可见性；后续人为批准门控 redact_arg_keys: ["password", "token", "secret", "api_key", "authorization"] resilience: rate_limit_backend: memory # 或：redis（需要下面的redis_url） redis_url: "redis://localhost:6379/0" retries: 1 # 首次尝试失败后额外尝试次数，带有回退 + 抖动 backoff_base_s: 0.1 backoff_max_s: 2.0 breaker_failure_threshold: 5 # 连续上游失败次数，断路器打开 breaker_cooldown_s: 30 # 打开 -> 半开冷却 ``` 已配置的 [`gateway.config.yaml`](gateway.config.yaml) 具有默认的后备值，因此演示可以直接运行； [`deploy/gateway.config.example.yaml`](deploy/gateway.config.example.yaml) 是生产模板 — 没有后备值，因此缺少的机密会在启动时大声失败，而不是静默默认为可猜测的值。上面的每个键都经过模式验证：未知键，错误类型和悬空角色引用都快速失败，并带有清晰的错误，这是 `mcp-gateway check` 的用途。 两种重塑配置的模式，而不是仅仅添加一个键： **JWT代替静态密钥 — 生产模式。** 删除 `principals`，身份来自令牌的声明，因此用户，代理及其角色分配存在于您的IdP中，而不涉及此文件。这是网关在具有真实用户群时运行的方式 — [docs/production-auth.md](docs/production-auth.md) 涵盖了架构，每个IdP的声明映射配方（Keycloak / Auth0 / Entra ID），服务代理和委托流程，以及为什么策略保留在此，而身份不保留。 [`deploy/gateway.config.jwt.example.yaml`](deploy/gateway.config.jwt.example.yaml) 是可以直接复制的模板， [`examples/jwt_demo.py`](examples/jwt_demo.py) 在没有安装IdP的情况下本地运行整个流程。添加 `resource` + `authorization_servers`，网关还会在 `/.well-known/oauth-protected-resource` 发布RFC 9728发现文档，供客户端自动查找，而不是异地告知发行者。 ```yml auth: type: jwt algorithm: RS256 # HS*/RS*/ES*/PS* jwks_uri: "https://idp.example/.well-known/jwks.json" # 或 public_key: "${JWT_SECRET}" issuer: "https://idp.example" audience: "mcp-gateway" principal_claim: sub roles_claim: roles # 列表，或以空格分隔的作用域字符串 tenant_claim: tenant resource: "https://gateway.example.com" # 可选：启用下面的发现 authorization_servers: ["https://idp.example"] scopes_supported: ["analyst", "editor", "admin"] # 可选，在发现文档中宣传 ``` **响应缓存或您通常会为 sidecar 达到的任何其他功能** — 它还不存在；参见 [What’s next](#what%E2%80%99s-next). --- ## 生产环境运行 三个未经身份验证的端点存在于监视进程中，以及在选择加入后出现的第四个端点： | | | |---|---| | `GET /healthz` | 存活 — `{"status":"ok"}` | | `GET /readyz` | 就绪 — 回显已配置的上游服务 | | `GET /metrics` | Prometheus 文本 — `mcpgw_calls_total{kind,decision}` 和延迟摘要 | | `GET /.well-known/oauth-protected-resource` | RFC 9728发现文档 — 仅当设置了 `auth.resource` 时才发布 | `docker compose up --build` 将网关、两个示例上游服务和Redis一起启动；如果您不希望使用演示默认值，请在环境中设置 `ANALYST_KEY`/`EDITOR_KEY`/`ADMIN_KEY`。OpenTelemetry 是可选的：安装SDK并在主机进程中配置跟踪器/导出器，审计记录也将开始发出跨度 — 退出时不会发生任何变化。 审计日志的实际内容 — 每行一个尝试，无论是否允许，因为审计是外部层，看到所有内部层拒绝的内容： ```json {"ts": "2026-07-14T09:38:13.611773+00:00", "correlation_id": "904cfb7c8191", "principal": "analyst-1", "tenant": "acme", "roles": ["analyst"], "kind": "tool", "tool": "calc_query_tool", "sensitive": false, "args": {"user_input": "add 2 and 3"}, "decision": "allow", "error": null, "duration_ms": 798.89, "result_size": 55} ``` ```json {"ts": "2026-07-14T09:38:16.116129+00:00", "correlation_id": "1f640dee60f9", "principal": "analyst-1", "tenant": "acme", "roles": ["analyst"], "kind": "tool", "tool": "kv_set", "sensitive": false, "args": {"key": "x", "value": "1"}, "decision": "deny", "error": "Access denied: analyst-1 (roles=['analyst']) may not call 'kv_set' (no allow rule matched).", "duration_ms": 0.04, "result_size": 0} ``` `kind` 涵盖 `tool` / `resource` / `prompt`；`decision` 是 `allow` / `deny` / `throttled` / `circuit_open` / `error` — 可直接查询以获取安全性和可靠性信号，以及每个行还增加了一个 `/metrics` 计数器。任何在 `redact_arg_keys` 之下的内容都会被掩盖成 `***` 然后写入。 ## 测试 ```powershell python -m pytest tests -q # 网关 — 63 个测试通过 python -m pytest agentctx/tests -q # agentctx — 49 个测试通过，独立于网关 ``` 对安全性最重要的部分——策略引擎、熔断器、限流器——故意保持纯粹和同步，以便直接测试它们，而无需启动 HTTP 或 FastMCP。 有几件特定的事情是经过测试的，而不仅仅是在文档字符串中被断言：JWT 验证器通过实际的代币往返过程（有效、签名错误、过期、范围→角色）进行测试，熔断器的状态机由注入的时钟驱动，而不是实际的睡眠，并且 `agentctx` 声称对该网关没有任何依赖关系——一个测试静态检查 [`agentctx/examples/standalone_worker.py`](agentctx/examples/standalone_worker.py) 的导入语句中是否有 `gateway`/`fastmcp`，然后实际运行它。 [.github/workflows/ci.yml](.github/workflows/ci.yml) 在每次推送和拉取请求时运行这两个测试套件以及 `mcp-gateway check`，使用的 Python 版本为 3.10、3.11 和 3.12。 完整的按文件分解在 [`tests/`](tests/) 和 [`agentctx/tests/`](agentctx/tests/) 中，如果你需要的话。 --- ## 代码结构 ``` gateway/ 控制平面 ├── app.py 组合中间件，操作路由，运行代理 ├── cli.py `mcp-gateway run|check` ├── config.py 模式验证，${env} 插值加载 ├── auth.py / identity.py 代币验证 → 主体解析 ├── policy_engine.py 纯粹的允许/拒绝 + 租户门控 ├── circuit.py / ratelimit.py 熔断器和可插拔的限流器 ├── observability.py 指标 + 可选的 OTel span ├── errors.py 类型异常，审计层读取 └── middleware/ ├── audit.py 最外层 — 结构化 JSONL ├── policy.py 强制执行工具/资源/提示 └── resilience.py 限流 · 熔断器 · 超时 · 重试 agentctx/ 身份 + 跟踪上下文，无网关依赖 ├── src/agentctx/ │ ├── identity.py 身份，身份解析器，静态密钥/JWT/代币交换 │ ├── context.py 运行上下文，W3C 跟踪父节点传播 │ ├── tracing.py OTel GenAI span 发射 │ └── oauth_metadata.py RFC 9728 受保护资源元数据文档塑形 ├── examples/standalone_worker.py 通过实际运行证明解耦，无 FastMCP └── tests/ examples/ ├── server.py, mock_kv_server.py 示例上游，网关前置 ├── client.py 示例代理面向的应用，通过网关调用 └── jwt_demo.py + gateway.config.jwt-demo.yaml JWT 身份模式，无 IdP 可运行 deploy/ ├── gateway.config.example.yaml 生产模板（静态密钥），无秘密回退 ├── gateway.config.jwt.example.yaml 生产模板（JWT/IdP），无主体块 └── gateway.docker.config.yaml Docker 组合栈挂载的配置 docs/ └── production-auth.md 身份验证规模：IdP 接线，委托，多租户，操作 gateway.config.yaml 直播演示配置 tests/ 网关测试套件（见上） ``` ### `agentctx` 包 [`agentctx/`](agentctx/) 不是一个网关模块——它是一个独立的库，网关恰好构建在它之上。 它回答“这是谁”（`Identity`，`IdentityResolver`，带有静态密钥，JWT 和 RFC 8693 代币交换解析器）和“这属于哪个运行”（`RunContext`，W3C 跟踪父节点传播，OTel GenAI span 发射），并且完全没有导入 FastMCP 或任何网关特定的内容。 [`gateway/auth.py`](gateway/auth.py) 只是一个消费者，将 FastMCP 的 `TokenVerifier` 适配到 `agentctx` 的解析器上，而不是重新实现验证。 它故意停止在生成符合标准的跟踪；评分和评估留给为此构建的工具（Langfuse，Arize Phoenix，DeepEval，Ragas）。 --- ## 下一步计划 | 支柱 | 下一步 | |---|---| | 身份验证 | 完整的 OIDC 发现/动态客户端注册，或 mTLS — 中间件不需要更改，因为它只读取 `identity.current_caller()` | | 策略 | 每参数 ABAC 条件，`sensitive_tools` 的人工批准门控，或外部化到 OPA | | 审计 | OTLP 导出到真正的后端（Datadog，Grafana，Honeycomb），一旦您准备好抓取 `/metrics` | | 弹性 | 响应缓存用于幂等读取，每个上游的并发批量上限 | 目前尚未完成的最大事情是配置热重载——在不重启进程的情况下编辑主体或策略。 列表上的其他所有内容都是在今天已经可以运行的支柱上附加的。 --- ## 贡献 欢迎 PR 和问题——请参阅 [CONTRIBUTING.md](.github/CONTRIBUTING.md) 以了解开发设置，测试命令和一对设计不变式（一方向的 `agentctx` 依赖，纯粹/同步安全关键代码）在您触摸 `gateway/` 之前需要了解。 本项目遵循 [Contributor Covenant](.github/CODE_OF_CONDUCT.md)，安全报告通过 [SECURITY.md](.github/SECURITY.md) 进行。 --- ## 许可证 [MIT](LICENSE)