Fastmcp Gateway

fastmcp网关

   

MCP的渐进式工具发现网关。 聚合来自多个上游的工具 主控程序 服务器，并通过4个元工具公开它们，使LLM能够发现和使用数百个工具，而无需预先加载所有模式。

LLM
 │
 └── fastmcp-gateway (4 meta-tools)
       ├── discover_tools    → browse domains and tools
       ├── get_tool_schema   → get parameter schema for a tool
       ├── execute_tool      → run any discovered tool
       │     ├── apollo      (upstream MCP server)
       │     ├── hubspot     (upstream MCP server)
       │     ├── slack       (upstream MCP server)
       │     └── ...
       └── refresh_registry  → re-query upstreams for changes

为什么？

当LLM连接到许多MCP服务器时，它一次接收所有工具模式。使用100多种工具时，上下文窗口会填满，工具选择精度会下降。 fastmcp网关 通过渐进式发现解决了这个问题：LLM从4个元工具开始，按需加载单个模式。

安装

pip install fastmcp-gateway

快速开始

Python API

import asyncio
from fastmcp_gateway import GatewayServer

gateway = GatewayServer(
    {
        "apollo": "http://apollo-mcp:8080/mcp",
        "hubspot": "http://hubspot-mcp:8080/mcp",
    },
    refresh_interval=300,  # Re-query upstreams every 5 minutes (optional)
)

async def main():
    await gateway.populate()     # Discover tools from upstreams
    gateway.run(transport="streamable-http", port=8080)

asyncio.run(main())

命令行界面

export GATEWAY_UPSTREAMS='{"apollo": "http://apollo-mcp:8080/mcp", "hubspot": "http://hubspot-mcp:8080/mcp"}'
python -m fastmcp_gateway

网关启动 http://0.0.0.0:8080/mcp 并向任何MCP客户端公开4个工具。

运作原理

1. discover_tools() --无参数调用以查看所有域和工具计数。致电 domain="apollo" 查看该域的工具及其描述。通过 format="signatures" 接收Python风格的函数签名(apollo_search(query: str, limit: int = None) -> any)而不是默认的JSON摘要——当LLM随后将针对列出的工具编写代码时非常有用。

1. get_tool_schema("apollo_people_search") --返回工具参数的完整JSON模式。支持模糊匹配。

1. execute_tool("apollo_people_search", {"query": "Anthropic"}) --将呼叫路由到正确的上游服务器并返回结果。

1. refresh_registry() --重新查询所有上游服务器，并返回每个域添加/删除的工具的摘要。在网关运行时更新上游时很有用。

LLM从网关的内置系统指令中学习工作流程，只加载他们实际需要的工具的模式。

配置

所有配置都是通过环境变量进行的：

根据上游授权

如果您的上游服务器需要不同的身份验证，请使用 GATEWAY_UPSTREAM_HEADERS 要设置每个域的标头，请执行以下操作：

export GATEWAY_UPSTREAM_HEADERS='{"ahrefs": {"Authorization": "Bearer sk-xxx"}}'

没有覆盖的域使用请求传递（来自传入MCP请求的标头被转发到上游）。

动态上游注册

当 GATEWAY_REGISTRATION_TOKEN 设置后，网关将公开REST端点以进行运行时上游管理——在不重新启动的情况下添加、删除和列出上游MCP服务器。

端点

所有端点都需要 Authorization: Bearer 匹配配置的令牌。

注册上游：

curl -X POST http://gateway:8080/registry/servers \
  -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"domain": "apollo", "url": "http://apollo-mcp:8080/mcp", "description": "Apollo.io CRM"}'

答复： {"registered": "apollo", "url": "...", "tools_discovered": 12, "tools_added": ["search", ...]}

注销上游：

curl -X DELETE http://gateway:8080/registry/servers/apollo \
  -H "Authorization: Bearer $TOKEN"

列出已注册的上游：

curl http://gateway:8080/registry/servers \
  -H "Authorization: Bearer $TOKEN"

Python API

gateway = GatewayServer(upstreams, registration_token="secret-token")

当未设置令牌（默认）时，注册端点为 不 已安装——现有部署不受影响。

线程安全

所有注册表突变（填充、添加、删除、刷新）都使用 asyncio.Lock 以防止并发腐败。

访问控制

使用带有glob匹配的每个上游允许/拒绝列表限制通过网关公开的下游工具。在注册表填充过程中应用策略——被阻止的工具永远不会进入注册表，因此每个元工具(discover_tools, get_tool_schema, execute_tool，搜索）会自动查看过滤后的视图。

配置（环境变量）

扩展 GATEWAY_UPSTREAMS 值（可选） allowed_tools / denied_tools 列表。简单的字符串值仍然像以前一样工作。

export GATEWAY_UPSTREAMS='{
  "apollo": {
    "url": "http://apollo:8080/mcp",
    "allowed_tools": ["apollo_search_*", "apollo_contact_*"]
  },
  "hubspot": {
    "url": "http://hubspot:8080/mcp",
    "denied_tools": ["*_delete"]
  },
  "linear": "http://linear:8080/mcp"
}'

配置（Python API）

在线通过每个上游过滤器，或构建 AccessPolicy 并明确地传递它。

from fastmcp_gateway import AccessPolicy, GatewayServer

policy = AccessPolicy(
    allow={
        "apollo":  ["apollo_search_*", "apollo_contact_*"],
        "hubspot": ["*"],
    },
    deny={"apollo": ["*_delete"]},
)
gateway = GatewayServer(
    {"apollo": "http://apollo:8080/mcp", "hubspot": "http://hubspot:8080/mcp"},
    access_policy=policy,
)

语义学

模式使用 fnmatch.fnmatchcase （区分大小写 * / ? 球体）。与注册的工具名称及其 original_name 因此，冲突前缀重命名不能绕过策略。

• allow：当非空时，仅显示此处列出的域，并且仅显示与至少一个模式匹配的工具。非空域名缺失 allow 地图被完全否认。保留为空，默认情况下允许每个域。
• deny：始终在之后应用 allow一种与…相匹配的工具 deny 即使模式也与 allow 图案。

当两个物体都成形时 upstreams 以及一个明确的 access_policy= 如果提供，显式论证获胜。

执行挂钩

钩子提供围绕工具执行和发现的中间件式生命周期回调。将它们用于身份验证、授权、令牌交换、审计日志记录或结果转换。

Python API

from fastmcp_gateway import GatewayServer, ExecutionContext, ExecutionDenied

class AuthHook:
    async def on_authenticate(self, headers: dict[str, str]):
        token = headers.get("authorization", "").removeprefix("Bearer ")
        return validate_jwt(token)  # Return user identity or None

    async def before_execute(self, context: ExecutionContext):
        if not has_permission(context.user, context.tool.domain):
            raise ExecutionDenied("Insufficient permissions", code="forbidden")
        # Inject headers for the upstream server
        context.extra_headers["X-User-Token"] = exchange_token(context.user)

gateway = GatewayServer(upstreams, hooks=[AuthHook()])

CLI（环境变量为）

点 GATEWAY_HOOK_MODULE 在返回钩子实例列表的工厂函数中：

export GATEWAY_HOOK_MODULE='my_package.hooks:create_hooks'

钩子生命周期

对于每一个 execute_tool 呼叫：

1. on_authenticate(headers) --从请求标头中提取用户标识。最后一个非无结果在多个钩子中获胜。
2. before_execute(context) --验证权限、修改参数、设置 extra_headers.提高 ExecutionDenied 阻止。
3. 上游呼叫 — extra_headers 以高于静态的最高优先级合并 upstream_headers.
4. after_execute(context, result, is_error) --转换或记录结果。每个钩子接收前一个钩子的输出。
5. on_error(context, error) --仅可观察性（钩子中的异常被记录，而不是引发）。

所有方法都是可选的——只实现你需要的方法。

工具可见性挂钩

这 after_list_tools 钩子阶段允许您在将工具列表返回给客户端之前对其进行过滤，这对于每个用户的访问控制非常有用：

from fastmcp_gateway import ListToolsContext

class AccessControlHook:
    async def after_list_tools(self, context: ListToolsContext, tools: list) -> list:
        # Filter tools based on user permissions
        return [t for t in tools if has_access(context.user, t.domain)]

隐藏的工具也会返回 tool_not_found 从 get_tool_schema 以防止信息泄露。

代码模式（实验）

实验，默认情况下关闭。 代码模式公开了第五元工具， execute_code，运行LLM编写的Python 蒙蒂 沙箱。每个注册的工具都被预先绑定为沙箱中的一个命名异步可调用工具，因此该模型可以链接调用并使用 asyncio.gather 扇出——在一次往返中，没有中间有效载荷通过代理的上下文窗口。

不适用于分析工作负载。 Monty沙箱的大小适合小负载跨工具链接（数十行，千字节的JSON）。大负载数据分析属于具有完整Python沙箱的专用分析服务器。

安装额外的

pip install "fastmcp-gateway[code-mode]"

启用

from fastmcp_gateway import GatewayServer

async def may_use_code_mode(user, context) -> bool:
    return user.id in {"alice", "bob"}  # bind this to your policy engine

gateway = GatewayServer(
    {"crm": "http://crm:8080/mcp", "analytics": "http://analytics:8080/mcp"},
    code_mode=True,
    code_mode_authorizer=may_use_code_mode,  # optional; any authenticated caller allowed when None
)

或者通过env变量：

export GATEWAY_CODE_MODE=true
export GATEWAY_CODE_MODE_MAX_DURATION_SECS=30
export GATEWAY_CODE_MODE_MAX_NESTED_CALLS=50

LLM如何使用它

呼叫 discover_tools(format="signatures") 首先获取可读的Python签名，然后编写调用这些函数的代码：

# What the LLM emits as the `code` argument to execute_code:
people = await crm_search(query="Anthropic", limit=5)
emails = [p["email"] for p in people["people"]]
{"count": len(emails), "emails": emails}

安全保证

• 每个嵌套的工具调用都经过相同的过程 before_execute / after_execute 直接钩住管道 execute_tool，因此访问策略和审计挂钩不变。
• 只有工具幸存 after_list_tools 过滤被绑定到沙盒命名空间中——未经授权的工具名称永远不会显示为可调用的，因此攻击者无法通过读取沙盒作用域来枚举它们。
• 外部请求头和用户身份在边界处捕获一次，并在每个包装器中关闭；沙盒的工作线程从不直接读取auth ContextVar。
• 资源限制（持续时间、内存、分配、递归深度、嵌套调用计数）适用于每次运行。
• 审核：默认值 code_mode.invoked INFO记录携带 code_sha256, tool_names_invoked, step_count，以及 duration_ms原始代码仅在调试时发出 code_mode_audit_verbatim=True --仅用于调试；原始LLM代码对PII敏感。

建造商参考

可观测性

网关为所有操作发出OpenTetry跨度。带上自己的导出器（Logfire、Jaeger、OTLP等）——网关使用 opentelemetry-api 并将拾取任何已配置的 TracerProvider.

关键跨度： gateway.discover_tools, gateway.get_tool_schema, gateway.execute_tool, gateway.refresh_registry, gateway.populate_all, gateway.background_refresh.

每个跨度包括以下属性 gateway.domain, gateway.tool_name, gateway.result_count，以及 gateway.error_code 用于过滤和警报。

错误处理

所有元工具返回结构化JSON错误，并带有 code 用于程序化处理和人类可读的字段 error 消息：

{"error": "Unknown tool 'crm_contacts'.", "code": "tool_not_found", "details": {"suggestions": ["crm_contacts_search"]}}

错误代码： tool_not_found, domain_not_found, group_not_found, execution_error, upstream_error, refresh_error.

工具名称冲突

当两个上游域以相同的名称注册工具时，网关会自动在这两个域前加上它们的域名，以防止冲突：

apollo registers "search"  →  apollo_search
hubspot registers "search" →  hubspot_search

原始名称仍可通过以下方式搜索 discover_tools(query="search").

MCP握手说明

之后 populate()，网关会自动构建MCP中包含的域感知指令 InitializeResult 握手。MCP客户端无需调用即可立即知道哪些工具域可用 discover_tools() 第一：

You have access to a tool discovery gateway with tools across these domains:

- **apollo** (12 tools) — Apollo.io CRM and sales intelligence
- **hubspot** (8 tools) — HubSpot CRM for contacts, companies, and deals

Workflow: discover_tools() → get_tool_schema() → execute_tool()

当注册表在后台刷新或动态注册过程中发生更改时，指令会自动重建。自定义 instructions= 在构建时传递的内容永远不会被覆盖。

健康终点

网关公开了与Kubernetes兼容的健康检查：

• GET /healthz --生命探测。总是返回200。
• GET /readyz --准备就绪探测。如果工具已填充，则返回200，否则返回503。

Docker和Kubernetes

看 examples/kubernetes/ 用于现成的Dockerfile和Kubernetes清单。

# Build
docker build -f examples/kubernetes/Dockerfile -t fastmcp-gateway .

# Run
docker run -e GATEWAY_UPSTREAMS='{"svc": "http://host.docker.internal:8080/mcp"}' \
  -p 8080:8080 fastmcp-gateway

贡献

看 贡献.md 用于开发设置、架构概述和指南。

许可证

Apache许可证2.0。看 许可证.