最近在做 AI Agent 网关层时一直被一个问题困扰模型越来越多、工具越来越多但每个应用接入新的模型和新的 MCP Server 时都要重复做适配。后来认真梳理了一遍发现模型路由层Model Router如果只负责“把请求转发给哪个大模型”其实没有解决最核心的问题——应用与外部工具之间的协议不统一。真正顺手的方式是在 Model Router 内部直接集成一个 MCP Gateway让模型路由和工具路由在同一个网关层完成。这篇文章会从 Model Router 和 MCP Gateway 的概念讲起分析为什么这两者应该合体并带大家实现一个最小可运行的统一网关示例。1. 背景与核心概念1.1 从一次 Agent 调用说起假设你正在开发一个 AI 助手用户问“北京今天适合穿什么衣服”这个请求到了后端一般会经过几个步骤将用户问题和系统提示词组装成消息列表。调用大模型让模型判断是否需要查询天气。模型返回一个工具调用tool call例如调用get_weather(北京)。系统执行真实天气 API。将天气结果再次发给模型模型生成最终回答。整个过程内部涉及两类关键组件一个是“选择哪个模型”另一个是“如何连接工具”。在项目比较早期这两件事都写在业务代码里耦合严重。团队一旦引入多个模型、多个工具服务就会看到类似这样的混乱# 伪代码混乱的调用方式 if provider openai: resp openai_client.chat.completions.create(...) elif provider anthropic: resp anthropic_client.messages.create(...) if tool_provider mcp_server_a: result await call_mcp_server_a(tool_name, args) elif tool_provider mcp_server_b: result await call_mcp_server_b(tool_name, args)这种写法的问题是业务代码同时感知了“模型供应商差异”和“工具服务差异”每接入一个新后端都要改一遍。Model Router 解决前者MCP Gateway 解决后者而把两者放在一起能一次性把这两个差异都封在网关层。1.2 什么是 Model RouterModel Router 也叫“模型路由器”或“LLM Gateway”。它的核心职责是根据请求特征、模型能力、成本、延迟、可用性等维度把一次模型请求路由到合适的模型后端。常见的 Model Router 能力包括多供应商接入OpenAI、Anthropic、Azure OpenAI、通义千问、智谱、Ollama、vLLM 等。模型切换与容灾主模型不可用时自动切换备用模型。成本控制根据预算、模型单价将不同请求分发到不同模型。负载均衡多个 API Key 或多个实例之间均衡流量。统一鉴权与计量输出统一的 API Key并对每个用户/应用做配额管理。日志与审计记录每个请求使用的模型、上下文长度、耗时、费用。1.3 什么是 MCP 与 MCP GatewayMCPModel Context Protocol是一套开放协议旨在标准化“大模型应用与外部工具、数据源之间”的连接方式。它由 Anthropic 在 2024 年底推出目前已经被众多 AI 工具和 Agent 框架支持。可以把它理解为“AI 应用的 USB-C 接口”只要工具服务实现了 MCP任何支持 MCP 的客户端都可以直接使用该工具不用再针对每个工具单独写适配代码。MCP 架构中有三个角色MCP Host发起连接的应用例如 Claude Desktop、自研 Agent。MCP Client在 Host 内部负责与 Server 建立连接。MCP Server暴露工具、资源或提示词的服务。MCP Gateway 则是在 MCP Client 与多个 MCP Server 之间加一层代理/网关。它不是标准协议中的强制组件而是实际工程中的演进产物。当团队内部有几十个 MCP Server 时如果每个应用都直连所有 Server会带来几个问题连接管理复杂每个应用都要维护到每个 Server 的认证、重连、超时逻辑。工具发现困难应用不知道该调用哪个 Server 的哪个工具。安全边界缺失不是所有应用都应该能调用所有工具。重复开发协议转换、鉴权、限流、日志在每个应用中各写一遍。MCP Gateway 归纳起来核心职责包括统一入口应用只连接一个网关就可以发现所有已授权的工具。工具注册与发现动态获取各 MCP Server 的工具列表形成统一目录。协议转换将不同传输方式stdio、SSE、Streamable HTTP统一代理或将 MCP 协议转换为内部协议。鉴权与授权在网关层校验调用方身份并判断是否有权调用某个工具。流控与配额对工具调用频率、并发数做控制。审计与日志记录每一次工具调用的入参、出参、耗时、调用方。1.4 为什么 Model Router 应该内置 MCP Gateway在实际架构中Model Router 和 MCP Gateway 解决的是“AI 应用接入后端”的两个关键问题模型选择和工具连接。两者分离在理论上是清晰的但在工程落地时存在明显痛点。痛点一模型选择和工具调用在同一个请求链路中。一次 Agent 对话往往交替发生“调模型”和“调工具”两个动作如果分属两套网关请求需要在两个网关之间来回跳转调试、追踪、鉴权都会变得困难。痛点二不同模型对工具调用的格式不同。OpenAI 使用tools参数和tool_calls返回结构Anthropic 使用tools和tool_useGoogle Gemini 使用functionDeclarations。上层应用如果直接对接多个模型工具定义格式差异会频繁侵入业务代码。而 Router 如果内置 MCP Gateway就可以在网关层完成“统一 MCP Schema → 目标模型 Schema”的转换。痛点三模型供应商和工具服务往往是同样的后端资源。模型路由需要根据上下文长度、成本、延迟做决策工具路由也需要根据工具名、参数、权限做决策。两者共享相同的基础设施比如 Redis、注册中心、监控面板、配置中心。放在一起可以复用这些能力。痛点四工具调用的结果回传也需要模型能力。MCP Server 返回工具结果后还需要再调用一次模型去生成最终回答。如果 Model Router 不在场这个“二次调用模型”的过程要么由上层应用实现要么由编排层实现。当 Router 集成 MCP Gateway 后网关可以自动完成“模型 → 工具 → 模型”的完整回路上层应用只需发送一次请求。2. 从 API 网关类比理解 MCP Gateway2.1 传统 API 网关解决了什么问题在微服务架构中API 网关是系统的统一入口。它负责路由请求到后端服务、做负载均衡、限流、鉴权、转发、协议转换、日志采集。如果没有 API 网关客户端需要知道所有微服务的地址、协议和鉴权方式系统内部细节直接暴露给调用方。MCP Gateway 在 AI Agent 架构中扮演的角色和 API 网关在微服务架构中的角色非常相似客户端不再直连每一个工具服务。服务端可以在网关层做统一控制和审计。协议细节被屏蔽在网关内部。2.2 MCP Gateway 的职责拆解为了实现上述目标MCP Gateway 通常需要具备以下几类能力工具目录Tool Catalog网关维护一份可用工具清单。每个工具至少包含工具名称全局唯一或带命名空间。描述。输入参数的 JSON Schema。所属 MCP Server。授权范围哪些调用方可用。连接管理Connection Manager网关负责与下游 MCP Server 建立和维护连接。需要支持启动时批量连接。连接断开后的重连机制。连接数控制避免每个请求都建立新连接。支持多种 MCP Transport。转发与协议适配当请求调用工具时网关需要将请求转发给对应的 MCP Server并把 MCP 返回的结果转换成统一格式。策略控制包括鉴权、限流、超时、重试、熔断。尤其要防止某个异常 MCP Server 拖垮整个网关。2.3 Model Router 与 MCP Gateway 的边界两个组件虽然可以合并部署但职责边界仍然要清晰。Model Router 的决策对象是“模型”输入是 messages、模型参数、预算、上下文长度MCP Gateway 的决策对象是“工具”输入是工具名、参数、调用方身份、工具权限。在统一网关中这两者形成一条清晰的流水线客户端请求 ↓ 认证鉴权 ↓ 模型路由选择目标模型 ↓ 工具发现加载 MCP 工具 Schema ↓ 调用模型兼容目标模型格式 ↓ 是否触发工具调用 ├── 否 → 返回响应 └── 是 → 通过 MCP Gateway 调用工具 ↓ 携带工具结果再次调用模型 ↓ 返回最终响应这条流水线既减少了请求跳数也把模型格式差异和工具格式差异都封装在网关层。3. 统一网关架构设计3.1 目标架构假设我们要建设一个面向内部多个业务线的“AI 统一网关”它的核心组件如下---------------- ---------------------------------------- | | | AI Gateway | | Web / App / | ---- | Auth Nginx / 统一入口 | | Agent SDK | | ----------------------------- | | | | | Router Engine | | ---------------- | | - Model Router | | | | - Tool Router | | | ----------------------------- | | | MCP Gateway Engine | | | | - Tool Catalog | | | | - Connection Manager | | | | - Schema Converter | | | ----------------------------- | | | Observation / Audit | | | ----------------------------- | ------------------------------------- | ------------------------------------------ | | ------v----- ------v----- | LLM | | MCP | | Provider | | Server A/B | ------------ ------------3.2 核心模块设计Router EngineRouter Engine 是网关的决策核心。它包含两个子模块Model Router根据可用模型、上下文长度、费用、请求优先级决定调用哪个模型。Tool Router根据工具名和调用方身份决定请求转发到哪个 MCP Server。MCP Gateway Engine这一层负责工具生命周期管理启动时从配置中心加载 MCP Server 地址列表。通过 MCP Client 连接各个 Server获取工具定义。将工具定义放入 Tool Catalog并缓存到内存或 Redis。收到工具调用请求时通过对应的 MCP Client 转发。Schema Converter负责把 MCP 工具定义转换为目标模型需要的格式OpenAItools数组 function对象。Anthropictools数组 input_schema。GeminifunctionDeclarations数组。Ollama/vLLM 等本地模型可能只支持 OpenAI 兼容格式。Observation Audit记录两类日志请求日志模型名、tokens、耗时、费用和工具调用日志工具名、入参、出参、耗时、调用方。建议全部带上 trace_id方便端到端排查问题。3.3 数据流一次带工具调用的请求以“查询北京天气”为例完整数据流如下客户端 POST/v1/chat/completions携带 messages。网关认证通过解析调用方身份。Model Router 根据请求内容和模型策略选择模型gpt-4o-mini。网关从 Tool Catalog 加载get_weather工具定义转换为 OpenAI 格式。调用 OpenAI模型返回tool_calls。网关收到tool_calls将参数解析后通过 MCP Gateway 转发给对应 MCP Server。MCP Server 返回天气数据。网关把工具结果追加到 messages再次调用同一个模型。模型生成最终回复网关以 OpenAI 兼容格式返回给客户端。在这个过程中客户端感知不到“后端切换了模型”也感知不到“工具实际来自哪个 MCP Server”这是统一网关的核心价值。4. 实战用 FastAPI 实现一个最小 Model Router MCP Gateway接下来我们实现一个极简版本。这个版本不依赖完整的 MCP SDK而是用“类似 MCP 工具注册表”的方式演示核心链路。因为完整接入真实 MCP Server 涉及不同传输方式和版本差异我们先用本地函数模拟工具服务再用真实 MCP Client 展示连接方式。4.1 环境准备与项目结构建议环境Python 3.11 FastAPI 0.110 uvicorn 0.29 openai 1.30mkdir ai-router-gateway-demo cd ai-router-gateway-demo python -m venv .venv source .venv/bin/activate pip install fastapi uvicorn[standard] openai项目结构ai-router-gateway-demo/ ├── main.py # 入口与接口 ├── model_router.py # 模型路由 ├── tool_registry.py # MCP 工具注册表 ├── llm_client.py # 统一 LLM 调用客户端 └── config.py # 配置4.2 定义模型注册表在config.py中维护一个简单的模型注册表。这里不写死具体供应商配置而是用环境变量隔离敏感信息。# config.py import os from dataclasses import dataclass dataclass class ModelRecord: name: str provider: str base_url: str api_key_env: str context_window: int priority: int property def api_key(self) - str: return os.getenv(self.api_key_env, ) # 模型注册表实际项目中可以改为从配置文件或配置中心读取 MODEL_REGISTRY [ ModelRecord( namegpt-4o-mini, provideropenai, base_urlhttps://api.openai.com/v1, api_key_envOPENAI_API_KEY, context_window128000, priority1, ), ModelRecord( namedeepseek-chat, provideropenai, base_urlhttps://api.deepseek.com/v1, api_key_envDEEPSEEK_API_KEY, context_window64000, priority2, ), ]说明priority数字越小优先级越高默认走gpt-4o-mini。provider这里统一用openai兼容格式实际项目中可能是anthropic、gemini等。base_url是为了支持 OpenAI 兼容接口的供应商。4.3 实现模型路由逻辑在model_router.py中实现一个简单的路由函数。考虑到真实场景需要考虑 tokens 估算、成本、负载这里做一个可扩展的版本# model_router.py from typing import Optional from config import MODEL_REGISTRY, ModelRecord def estimate_tokens(messages: list[dict]) - int: 粗略估算 token 数实际项目建议使用 tiktoken 等库。 total_chars sum(len(m.get(content) or ) for m in messages) return int(total_chars / 3) # 粗略估算约 3 个字符 1 token def route_model( messages: list[dict], preferred_model: Optional[str] None, ) - ModelRecord: 根据请求和模型注册表选择模型。 if preferred_model: for record in MODEL_REGISTRY: if record.name preferred_model: return record raise ValueError(f模型 {preferred_model} 不在注册表中) # 按优先级排序 candidates sorted(MODEL_REGISTRY, keylambda r: r.priority) # 如果上下文窗口不够自动选择更大的模型 tokens estimate_tokens(messages) for record in candidates: if tokens record.context_window: return record # 兜底返回上下文窗口最大的模型 return max(candidates, keylambda r: r.context_window)这个路由函数虽然简单但演示了核心思路先按优先级选择默认模型再根据上下文长度做兜底。真实项目中还可以加入成本预估、响应时间、故障熔断等策略。4.4 实现 MCP 工具注册表在tool_registry.py中我们定义两个工具get_weather查询天气模拟真实工具。calculate四则运算演示不带外部服务的工具。每个工具记录包含三部分schemaOpenAI 格式的工具定义、handler实际执行函数、mcp_server模拟所属 MCP Server在真实场景中可能是某个 mcp server 标识。# tool_registry.py import json from typing import Any, Awaitable, Callable # 工具处理器注册表 ToolHandler Callable[..., Awaitable[Any]] async def get_weather(city: str) - str: 模拟查询天气。 mock_data { 北京: {weather: 晴, temperature: 25, advice: 适合穿短袖}, 上海: {weather: 小雨, temperature: 22, advice: 建议带伞}, } return json.dumps(mock_data.get(city, {weather: 未知, temperature: 0}), ensure_asciiFalse) async def calculate(expression: str) - str: 模拟计算工具。生产环境绝对不能这样执行表达式这里仅为演示。 return str(eval(expression)) # 仅用于演示实际项目务必使用安全解析库 TOOL_REGISTRY: dict[str, dict[str, Any]] { get_weather: { mcp_server: weather-mcp-server, schema: { type: function, function: { name: get_weather, description: 查询指定城市的实时天气, parameters: { type: object, properties: { city: {type: string, description: 城市名称例如 北京} }, required: [city], }, }, }, handler: get_weather, }, calculate: { mcp_server: calc-mcp-server, schema: { type: function, function: { name: calculate, description: 计算数学表达式, parameters: { type: object, properties: { expression: {type: string, description: 待计算的数学表达式} }, required: [expression], }, }, }, handler: calculate, }, } def get_all_tool_schemas() - list[dict]: return [item[schema] for item in TOOL_REGISTRY.values()] def get_tool_handler(name: str) - ToolHandler: return TOOL_REGISTRY[name][handler]这里需要注意eval在真实生产环境有严重安全风险这里仅为演示工具转发链路。真实项目中如果要实现计算工具应该使用ast模块解析或专门的数学表达式库。4.5 实现统一 LLM 调用客户端在llm_client.py中封装一次模型调用。这里的重点是“把工具定义传给模型”和“解析工具调用返回”。# llm_client.py import json from typing import Optional from openai import AsyncOpenAI from config import ModelRecord async def chat_completion_with_tools( model_record: ModelRecord, messages: list[dict], tools: Optional[list[dict]] None, ) - dict: 调用 OpenAI 兼容接口。 返回 dict包含 content 和 tool_calls 两个字段。 client AsyncOpenAI( api_keymodel_record.api_key, base_urlmodel_record.base_url, ) request_kwargs { model: model_record.name, messages: messages, } if tools: request_kwargs[tools] tools request_kwargs[tool_choice] auto response await client.chat.completions.create(**request_kwargs) choice response.choices[0] message choice.message result {content: message.content or } if message.tool_calls: result[tool_calls] [ { id: tc.id, name: tc.function.name, arguments: json.loads(tc.function.arguments or {}), } for tc in message.tool_calls ] return result这里有一个关键点tool_choice设置为auto表示让模型自行决定是否调用工具。如果业务明确要求“本次必须调用某个工具”也可以传入具体定义。4.6 实现完整调用链路在main.py中组装整个流程。FastAPI 提供一个/v1/chat/completions接口。这个接口做了以下事情解析请求中的 preferred_model。调用 model_router 选择模型。从 tool_registry 加载全部工具。第一次调用模型。如果模型返回 tool_calls则依次执行工具。将工具结果追加到 messages第二次调用模型。返回最终结果。# main.py from typing import Optional from fastapi import FastAPI from pydantic import BaseModel, Field from llm_client import chat_completion_with_tools from model_router import route_model from tool_registry import get_all_tool_schemas, get_tool_handler app FastAPI(titleAI Router MCP Gateway Demo) class ChatMessage(BaseModel): role: str content: str class ChatRequest(BaseModel): messages: list[ChatMessage] model: Optional[str] Field(defaultNone, description优先使用的模型名) class ChatResponse(BaseModel): content: str model: str app.post(/v1/chat/completions) async def chat_completions(req: ChatRequest): messages [m.model_dump() for m in req.messages] # 1. 路由模型 model_record route_model(messages, preferred_modelreq.model) # 2. 获取工具定义 tools get_all_tool_schemas() # 3. 第一次调用模型 first_response await chat_completion_with_tools( model_recordmodel_record, messagesmessages, toolstools, ) # 4. 如果模型需要调用工具 if first_response.get(tool_calls): tool_messages [] for tool_call in first_response[tool_calls]: handler get_tool_handler(tool_call[name]) result await handler(**tool_call[arguments]) tool_messages.append({ role: tool, tool_call_id: tool_call[id], name: tool_call[name], content: result, }) # 将模型返回的 tool_calls 消息和工具结果一起追加 messages.append({ role: assistant, content: first_response[content], tool_calls: [ { id: tc[id], type: function, function: { name: tc[name], arguments: json.dumps(tc[arguments], ensure_asciiFalse), }, } for tc in first_response[tool_calls] ], }) messages.extend(tool_messages) # 5. 第二次调用模型生成最终回答 final_response await chat_completion_with_tools( model_recordmodel_record, messagesmessages, toolstools, ) return ChatResponse(contentfinal_response[content], modelmodel_record.name) return ChatResponse(contentfirst_response[content], modelmodel_record.name)注意第二次调用时仍然传入tools是为了让模型在生成最终回答时仍能看到工具定义。如果确认工具调用已经完成也可以传toolsNone。4.7 运行验证启动服务uvicorn main:app --reload --port 8000用 curl 测试curl -X POST http://localhost:8000/v1/chat/completions \ -H Content-Type: application/json \ -d { messages: [ {role: user, content: 北京今天天气怎么样适合穿什么衣服} ] }预期流程是网关路由到gpt-4o-mini。模型返回tool_calls调用get_weather。网关执行本地 handler返回{weather: 晴, temperature: 25, advice: 适合穿短袖}。模型生成最终回答。客户端只看到最终回答。如果没有配置真实 API Key可以先用calculate工具演示链路或者在代码中临时写一个 Mock 模型响应。5. 常见问题与排查思路5.1 MCP Server 连接失败问题现象常见原因解决思路启动时获取工具失败MCP Server 地址错误或网络不可达检查 Server 地址、Token 和网络连通性连接后立即断开Server 端版本与 Client 不兼容确认 MCP 协议版本升级 SDK工具列表为空Server 没有声明任何工具在 MCP Server 端确认list_tools()返回内容调用超时Server 处理慢或被防火墙拦截在网关配置超时时间并排查网络策略5.2 工具 Schema 在不同模型间不兼容不同模型对工具定义的约束不同。例如 OpenAI 的parameters要求是 JSON SchemaAnthropic 的input_schema也是 JSON Schema但某些模型对additionalProperties、anyOf等关键字支持不一致。建议在网关中做一层 Schema Converter统一转换为目标模型格式。转换时注意删除目标模型不支持的 JSON Schema 关键字。对type为array或object的工具参数要给足示例说明否则模型容易生成错误参数。工具描述尽量精简避免超长描述消耗上下文。5.3 模型陷入工具调用循环有时模型会反复调用同一个工具或者调用参数越来越奇怪。可能原因工具返回结果不够清晰模型无法判断已完成。系统提示词没有明确“完成任务后停止调用工具”。上下文过长导致模型忽略关键信息。排查思路在网关中增加工具调用次数上限。在系统提示词中加入“仅在需要信息时调用工具”。检查工具返回结果是否包含足够、明确的信息。在审计日志中对比同一请求的多次工具调用参数。5.4 认证与安全边界网关层统一鉴权之后安全风险集中在两个方向调用方是否越权调用敏感工具。模型的 prompt 注入漏洞可能导致工具被诱导调用。建议实现网关对工具做白名单控制 调用方身份 → 角色 → 可用工具列表 工具参数做类型校验和敏感信息过滤 工具调用记录完整的输入输出6. 最佳实践与工程建议6.1 统一工具命名空间当接入了多个 MCP Server 后很可能出现两个 Server 都定义了get_order这种同名工具。建议在网关层为每个 MCP Server 设置命名空间例如order_service.get_order payment_service.get_order命名空间在工具调用转发时自动去掉后端 Server 仍然收到原始工具名。6.2 缓存工具定义MCP Server 的工具定义在大多数场景下是静态的。每次请求都去 Server 拉取工具列表会造成不必要的延迟。建议在网关中增加缓存启动时批量拉取。定期刷新或通过事件通知更新。缓存过期后异步刷新而不是同步阻塞请求。6.3 全链路可观测性统一网关最大的优势之一就是可以在一个地方采集全链路数据。建议为每个请求生成 trace_id并记录trace_id 调用方 app_id 目标模型 模型 tokens 和费用 工具调用列表工具名、耗时、是否成功 总耗时 错误码这些数据不仅能帮助排查问题还能用来做成本分摊和容量规划。6.4 谨慎处理工具返回结果MCP Server 返回的内容可能是任意文本直接拼接到 messages 中可能引入大量无关信息。建议限制工具返回内容长度。对工具结果做截断或摘要。对包含敏感信息的工具结果做脱敏。不要把工具内部错误堆栈直接暴露给模型。6.5 生产环境部署建议统一网关本身是高可用组件生产环境部署时需要关注多实例无状态部署Session 状态存放在 Redis。为不同模型供应商配置超时和重试策略。对 MCP Server 增加熔断机制避免单点故障拖垮网关。将模型路由规则和工具权限配置放入配置中心支持动态下发。灰度发布先让少量调用方使用新模型或新工具再逐步放量。7. 总结与下一步学习建议这篇文章的核心观点是Model Router 的价值不只是“切换模型”把 MCP Gateway 集成到路由层之后才能真正实现“一个入口屏蔽模型差异和工具差异”。我们分析了 Model Router 与 MCP Gateway 的分工画出了统一网关的核心流程并用 FastAPI 实现了一个最小可运行的模型路由 工具调用链路。接下来你可以继续深入的方向包括阅读 MCP 官方规范理解 stdio、SSE、Streamable HTTP 三种传输方式的差异。将示例中的本地工具注册表替换为真实 MCP Client连接一个实际的 MCP Server。引入 JSON Schema 校验在网关层拦截非法工具参数。将路由规则抽象为可配置策略支持按用户、按应用、按成本中心分流。增加模型调用失败时的自动降级和重试机制。如果你正在建设内部 AI 平台或 Agent 基础设施建议优先把“模型路由 工具网关”作为一个整体来设计。分开做虽然短期看起来清晰长期会因为请求链路割裂、调试困难、权限分散而付出更高的维护成本。动手搭一个最小版本你会很快理解这层网关在实际业务中的杠杆作用。