从 MCP 到 A2A：AI Agent 工具调用网关的后端架构设计

1. 背景AI 热点正在从模型转向 Agent 工程化过去讨论 AI 应用重点更多是模型本身。例如模型理解能力、上下文长度、推理能力、多模态能力、生成质量等。但最近一段时间开发者更关注另一个方向AI 如何进入真实工作流。OpenAI 的 Agents SDK 文档中将 Agent 描述为可以规划、调用工具、在不同专家之间协作并保持足够状态来完成多步骤任务的应用。OpenAI 文档也提到Responses API 是构建 Agent 的未来方向Assistants API 已进入迁移和退场周期。Google 也在推动 Agent 工程化。Google Cloud 在 2025 年推出 Agent Development Kit官方介绍称 ADK 是一个开源框架用于简化 Agent 和多 Agent 系统的端到端开发Google Cloud 文档也将 ADK 描述为可以构建、调试和部署企业级可靠 AI Agent 的开源开发框架。与此同时MCP 和 A2A 也成为热门方向。Anthropic 将 MCP 定义为连接 AI 助手与外部数据源、业务工具、开发环境的开放标准Google 发布的 A2A 协议则强调让不同企业平台或应用上的 AI Agent 能安全交换信息并协同行动。这些变化说明AI 应用的技术重点正在发生转移从“模型回答得好不好”转向“Agent 能不能安全、稳定、可控地使用工具完成任务”。对后端开发者来说这里会出现一个非常具体的问题当 AI Agent 能调用工具、读取数据、操作系统、访问接口时企业应该如何管理这些调用这就是本文要讨论的主题AI Agent 工具调用网关。2. 为什么需要 Agent 工具调用网关AI Agent 一旦可以调用工具就不再只是聊天机器人。它可能会做这些事情查询数据库读取文档调用内部 API创建工单修改代码运行测试发送通知创建 Pull Request查询订单调用 CRM操作业务系统。这些能力很有价值但也带来风险。例如Agent 调用了不该调用的接口Agent 读取了敏感数据Agent 在没有审批的情况下执行了高风险操作Agent 调用失败后没有记录Agent 输出结果无法追溯多个 Agent 同时调用工具造成状态混乱某个工具被误注册导致权限越界。所以企业不能简单让 Agent 直接连所有工具。更合理的方式是增加一层网关Agent ↓ 工具调用网关 ↓ 权限校验 / 风险分级 / 审批 / 审计 ↓ 真实工具与业务系统这个网关的作用不是限制 AI而是让 AI 工具调用变得可管理。3. 系统目标一个基础版 Agent 工具调用网关需要解决 6 个问题。3.1 工具能不能统一注册每个工具都应该有清晰定义工具名称工具类型工具用途输入参数输出结构风险等级是否启用是否需要人工审批。3.2 Agent 能不能按权限调用工具不同 Agent 不应该拥有相同权限。例如客服 Agent 可以查询 FAQ运营 Agent 可以生成内容数据 Agent 可以查询报表研发 Agent 可以读取代码管理 Agent 可以创建审批任务。权限要按 Agent、工具、资源范围进行控制。3.3 高风险调用能不能审批低风险调用可以直接执行。例如查询 FAQ读取公开文档生成摘要查询无敏感数据的统计报表。高风险调用必须审批。例如修改数据库调用付款接口批量发送通知修改生产配置创建外部发布任务删除数据。3.4 每次调用能不能审计Agent 调用工具时必须记录谁调用调用哪个工具传入了什么摘要返回了什么摘要是否成功是否触发审批是否失败失败原因是什么调用时间是什么。3.5 能不能兼容 MCP 和 A2A 思路MCP 解决的是 Agent 与工具、数据源之间的连接问题。A2A 解决的是 Agent 与 Agent 之间协作通信的问题。企业内部可以不一开始完整实现协议但架构设计要预留接口。3.6 能不能先做 MVP第一版不要追求大而全。先做工具注册风险分级调用日志简单审批FastAPI 接口后续再接 MCP Server、A2A Agent、企业内部系统。4. 架构设计可以设计成下面几层┌─────────────────────────────┐ │ Agent 应用层 │ │ 客服 Agent / 研发 Agent / 数据 Agent │ └──────────────┬──────────────┘ │ ┌──────────────▼──────────────┐ │ 工具调用网关层 │ │ 统一入口 / 鉴权 / 风险判断 / 审计 │ └──────────────┬──────────────┘ │ ┌──────────────▼──────────────┐ │ 工具注册中心 │ │ 工具定义 / 参数 Schema / 风险等级 │ └──────────────┬──────────────┘ │ ┌──────────────▼──────────────┐ │ 审批与策略层 │ │ 人工审批 / 白名单 / 黑名单 / 限流 │ └──────────────┬──────────────┘ │ ┌──────────────▼──────────────┐ │ 真实工具层 │ │ 数据库 / CRM / Git / CI / 文档 / API │ └─────────────────────────────┘这个架构的重点是Agent 不直接接业务系统而是通过工具调用网关统一访问。这样做有几个好处工具权限可控调用过程可追溯高风险操作可审批多 Agent 可以共享工具后期可以接入 MCP、A2A企业内部系统不会被 Agent 直接暴露。5. 数据模型设计下面用 Pydantic 模型描述核心数据结构。5.1 工具风险等级from enum import Enum class RiskLevel(str, Enum): low low medium medium high high风险等级可以这样理解low只读、无敏感信息、无业务副作用medium可能影响业务状态但可回滚high可能影响资金、权限、生产环境、客户数据。5.2 工具定义模型from typing import Optional, Dict, Any from pydantic import BaseModel, Field class ToolDefinition(BaseModel): tool_name: str Field(..., description工具名称) tool_type: str Field(..., description工具类型如 crm、database、git、search) description: str Field(..., description工具说明) risk_level: RiskLevel RiskLevel.low require_approval: bool False enabled: bool True input_schema: Optional[Dict[str, Any]] None例如tool ToolDefinition( tool_namequery_customer_profile, tool_typecrm, description查询客户基础资料, risk_levelRiskLevel.medium, require_approvalFalse, input_schema{ type: object, properties: { customer_id: {type: string} }, required: [customer_id] } )5.3 工具调用请求class ToolCallRequest(BaseModel): agent_id: str tool_name: str arguments: Dict[str, Any] request_id: Optional[str] None5.4 工具调用结果class ToolCallResult(BaseModel): success: bool tool_name: str output: Optional[Dict[str, Any]] None error_message: Optional[str] None approval_required: bool False audit_id: Optional[str] None5.5 审计日志from datetime import datetime class AuditLog(BaseModel): audit_id: str agent_id: str tool_name: str risk_level: RiskLevel input_summary: str output_summary: Optional[str] status: str approval_required: bool created_at: datetime审计日志不要直接保存敏感原文。更合理的方式是保存摘要。6. FastAPI 最小实现6.1 安装依赖pip install fastapi uvicorn pydantic6.2 完整示例代码文件名main.pyfrom enum import Enum from typing import Dict, Any, Optional from datetime import datetime from uuid import uuid4 from fastapi import FastAPI, HTTPException from pydantic import BaseModel, Field app FastAPI(titleAI Agent Tool Gateway) class RiskLevel(str, Enum): low low medium medium high high class ToolDefinition(BaseModel): tool_name: str Field(..., description工具名称) tool_type: str Field(..., description工具类型) description: str Field(..., description工具说明) risk_level: RiskLevel RiskLevel.low require_approval: bool False enabled: bool True input_schema: Optional[Dict[str, Any]] None class ToolRegisterRequest(BaseModel): tool_name: str tool_type: str description: str risk_level: RiskLevel RiskLevel.low require_approval: bool False input_schema: Optional[Dict[str, Any]] None class ToolCallRequest(BaseModel): agent_id: str tool_name: str arguments: Dict[str, Any] request_id: Optional[str] None class ToolCallResult(BaseModel): success: bool tool_name: str output: Optional[Dict[str, Any]] None error_message: Optional[str] None approval_required: bool False audit_id: Optional[str] None class AuditLog(BaseModel): audit_id: str agent_id: str tool_name: str risk_level: RiskLevel input_summary: str output_summary: Optional[str] status: str approval_required: bool created_at: datetime tool_registry: Dict[str, ToolDefinition] {} audit_logs: Dict[str, AuditLog] {} def summarize_arguments(arguments: Dict[str, Any]) - str: keys list(arguments.keys()) return fargs_keys{keys} def write_audit_log( agent_id: str, tool: ToolDefinition, input_summary: str, output_summary: Optional[str], status: str, approval_required: bool ) - str: audit_id str(uuid4()) log AuditLog( audit_idaudit_id, agent_idagent_id, tool_nametool.tool_name, risk_leveltool.risk_level, input_summaryinput_summary, output_summaryoutput_summary, statusstatus, approval_requiredapproval_required, created_atdatetime.now() ) audit_logs[audit_id] log return audit_id app.post(/api/tools/register) def register_tool(payload: ToolRegisterRequest): if payload.tool_name in tool_registry: raise HTTPException(status_code400, detailtool already exists) tool ToolDefinition( tool_namepayload.tool_name, tool_typepayload.tool_type, descriptionpayload.description, risk_levelpayload.risk_level, require_approvalpayload.require_approval, enabledTrue, input_schemapayload.input_schema ) tool_registry[payload.tool_name] tool return { message: tool registered, tool: tool } app.get(/api/tools) def list_tools(): return list(tool_registry.values()) app.post(/api/tools/call, response_modelToolCallResult) def call_tool(payload: ToolCallRequest): tool tool_registry.get(payload.tool_name) if not tool: raise HTTPException(status_code404, detailtool not found) if not tool.enabled: raise HTTPException(status_code403, detailtool disabled) input_summary summarize_arguments(payload.arguments) if tool.require_approval or tool.risk_level RiskLevel.high: audit_id write_audit_log( agent_idpayload.agent_id, tooltool, input_summaryinput_summary, output_summaryNone, statuspending_approval, approval_requiredTrue ) return ToolCallResult( successFalse, tool_nametool.tool_name, outputNone, error_messageapproval required, approval_requiredTrue, audit_idaudit_id ) # 模拟真实工具执行 output { message: ftool {tool.tool_name} executed, arguments_received: payload.arguments } audit_id write_audit_log( agent_idpayload.agent_id, tooltool, input_summaryinput_summary, output_summarytool executed successfully, statussuccess, approval_requiredFalse ) return ToolCallResult( successTrue, tool_nametool.tool_name, outputoutput, error_messageNone, approval_requiredFalse, audit_idaudit_id ) app.get(/api/audit/{audit_id}) def get_audit_log(audit_id: str): log audit_logs.get(audit_id) if not log: raise HTTPException(status_code404, detailaudit log not found) return log启动服务uvicorn main:app --reload打开接口文档http://127.0.0.1:8000/docs7. 注册工具示例注册一个低风险工具{ tool_name: search_faq, tool_type: knowledge_base, description: 查询企业 FAQ 知识库, risk_level: low, require_approval: false }注册一个高风险工具{ tool_name: refund_order, tool_type: payment, description: 执行订单退款操作, risk_level: high, require_approval: true }低风险工具可以直接调用。高风险工具会进入审批状态。8. 调用工具示例调用 FAQ 工具{ agent_id: customer-service-agent, tool_name: search_faq, arguments: { query: 如何申请售后 } }返回结果类似{ success: true, tool_name: search_faq, output: { message: tool search_faq executed, arguments_received: { query: 如何申请售后 } }, error_message: null, approval_required: false, audit_id: xxx }调用退款工具{ agent_id: customer-service-agent, tool_name: refund_order, arguments: { order_id: O20260529001, amount: 99.00 } }返回结果会进入审批{ success: false, tool_name: refund_order, output: null, error_message: approval required, approval_required: true, audit_id: xxx }这就是工具调用网关的核心价值不是让 Agent 不能做事而是让 Agent 在合适权限内做事。9. 如何和 MCP 结合MCP 可以理解为 Agent 和外部工具之间的连接标准。企业内部的工具调用网关可以和 MCP 这样结合Agent ↓ 工具调用网关 ↓ MCP Client ↓ MCP Server ↓ 数据库 / 文档 / Git / 内部系统也可以反过来Agent ↓ MCP Client ↓ 企业自定义 MCP Server ↓ 工具调用网关 ↓ 真实业务系统关键不是形式而是要保持三个原则工具必须注册调用必须审计高风险操作必须审批。如果 MCP Server 直接暴露敏感系统而没有网关和权限控制风险会比较高。10. 如何和 A2A 结合A2A 更关注 Agent 与 Agent 之间的通信。例如客服 Agent ↓ 调用订单 Agent ↓ 订单 Agent 调用库存 Agent ↓ 库存 Agent 返回缺货信息 ↓ 客服 Agent 生成用户回复在这种场景下工具调用网关仍然有价值。因为多个 Agent 协作时更容易出现调用链复杂、权限边界不清、责任难以追溯的问题。建议每个 Agent 调用外部工具时都统一经过网关。这样可以记录哪个 Agent 发起请求调用了哪个工具是否经过其他 Agent 中转是否触发高风险操作最终结果是什么。11. 安全设计要点11.1 不要让 Agent 直接访问生产数据库Agent 可以生成查询建议但不应该直接执行生产数据库写操作。11.2 不要把密钥放进提示词不要把 API Key、Token、数据库密码、私钥等内容放进 Agent 上下文。11.3 高风险工具必须人工审批例如退款删除数据修改权限执行部署批量发消息修改生产配置。11.4 审计日志要脱敏不要把完整用户隐私、敏感字段、业务机密直接写进日志。11.5 工具要有启停开关一旦工具异常可以快速禁用。12. MVP 落地路线第一阶段只做工具注册和审计先统一管理工具清单。做到工具可注册工具可启用和禁用调用可记录风险可标记。第二阶段增加审批流针对中高风险工具增加审批。做到高风险调用不直接执行生成审批记录人工确认后再调用真实系统。第三阶段接入 MCP Server把内部文档、知识库、CRM、Git、数据系统逐步封装成 MCP Server 或工具接口。第四阶段支持多 Agent 协作当多个 Agent 协作时通过统一网关记录调用链避免权限混乱。13. 总结近期 AI 行业的热点已经不只是模型本身。OpenAI Agents SDK、Google ADK、MCP、A2A 等方向都说明AI 应用正在进入工具调用、多步骤任务执行和多 Agent 协作阶段。对后端开发者来说真正要关注的不是“Agent 能不能调用工具”而是工具如何注册权限如何控制风险如何分级审批如何设计调用如何审计多 Agent 协作如何追溯生产系统如何避免被误操作。因此企业在引入 AI Agent 时不建议让 Agent 直接访问所有系统。更合理的方式是先建设一个工具调用网关。让 Agent 通过网关调用工具让网关负责鉴权、审计、审批和风险控制。一句话总结Agent 越强工具调用越要有边界。这才是 AI Agent 真正进入企业系统时后端架构需要补上的一层。