news 2026/5/30 11:59:04

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

作者头像

张小明

前端开发工程师

1.2k 24
文章封面图
从 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_name="query_customer_profile", tool_type="crm", description="查询客户基础资料", risk_level=RiskLevel.medium, require_approval=False, 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] = None

5.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] = None

5.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 pydantic

6.2 完整示例代码

文件名：main.py

from 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(title="AI 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 f"args_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_id=audit_id, agent_id=agent_id, tool_name=tool.tool_name, risk_level=tool.risk_level, input_summary=input_summary, output_summary=output_summary, status=status, approval_required=approval_required, created_at=datetime.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_code=400, detail="tool already exists") tool = ToolDefinition( tool_name=payload.tool_name, tool_type=payload.tool_type, description=payload.description, risk_level=payload.risk_level, require_approval=payload.require_approval, enabled=True, input_schema=payload.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_model=ToolCallResult) def call_tool(payload: ToolCallRequest): tool = tool_registry.get(payload.tool_name) if not tool: raise HTTPException(status_code=404, detail="tool not found") if not tool.enabled: raise HTTPException(status_code=403, detail="tool disabled") input_summary = summarize_arguments(payload.arguments) if tool.require_approval or tool.risk_level == RiskLevel.high: audit_id = write_audit_log( agent_id=payload.agent_id, tool=tool, input_summary=input_summary, output_summary=None, status="pending_approval", approval_required=True ) return ToolCallResult( success=False, tool_name=tool.tool_name, output=None, error_message="approval required", approval_required=True, audit_id=audit_id ) # 模拟真实工具执行 output = { "message": f"tool {tool.tool_name} executed", "arguments_received": payload.arguments } audit_id = write_audit_log( agent_id=payload.agent_id, tool=tool, input_summary=input_summary, output_summary="tool executed successfully", status="success", approval_required=False ) return ToolCallResult( success=True, tool_name=tool.tool_name, output=output, error_message=None, approval_required=False, audit_id=audit_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_code=404, detail="audit log not found") return log

启动服务：

uvicorn main:app --reload

打开接口文档：

http://127.0.0.1:8000/docs

7. 注册工具示例

注册一个低风险工具：

{ "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 ↓ 工具调用网关 ↓ 真实业务系统

关键不是形式，而是要保持三个原则：

  1. 工具必须注册；
  2. 调用必须审计；
  3. 高风险操作必须审批。

如果 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 真正进入企业系统时，后端架构需要补上的一层。

版权声明: 本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若内容造成侵权/违法违规/事实不符，请联系邮箱：809451989@qq.com进行投诉反馈，一经查实，立即删除！
网站建设 2026/5/30 11:57:18

【安全与隐私】让大模型接管电脑的风险防范：沙盒隔离与敏感区域（如支付）阻断机制

当AI能帮你读邮件、写代码、甚至发起转账，谁来为它的错误决策买单？ 本文基于2025-2026年的最新安全事件、学术论文和厂商实践，深入剖析AI智能体接管电脑带来的新型安全威胁，全面对比沙盒隔离与敏感操作阻断两大防御路径，并提供一套可落地的安全架构实战指南。 一、开篇：当…

作者头像 李华
网站建设 2026/5/30 11:57:11

内网CentOS 7离线装LibreOffice 7.1，我踩过的依赖坑都给你填平了

内网CentOS 7离线部署LibreOffice 7.1全指南：从依赖解析到避坑实战 当你面对一台刚装好的CentOS 7服务器，没有外网连接，却需要部署LibreOffice时，那种"缺胳膊少腿"的依赖报错简直让人抓狂。作为经历过无数次这种折磨的运…

作者头像 李华
网站建设 2026/5/30 11:55:20

小米刚发布的全新App， 这次是必升级啊！小米米家体验版

软件获取 小米手机软件合集： 最近小米旗下的智能硬件管理平台米家 App 迎来了重大升级，版本号从 v10.x 升级到了 v11.0 大版本，最大变化在于界面 UI 设计焕新，更好看了，信息展现更直观，控制设备更方便。 现…

作者头像 李华
网站建设 2026/5/30 11:54:36

DIY遥控飞机全攻略：从零制作低成本泡沫板航模

1. 项目概述与核心思路我一直觉得，亲手做一架能飞起来的遥控飞机，是件特别酷的事。这不仅仅是把一堆电子元件和泡沫板粘在一起，更是一个理解空气动力学、电子控制和机械结构如何协同工作的绝佳过程。很多人可能觉得航模制作门槛很高&#xff…

作者头像 李华
网站建设 2026/5/30 11:53:16

MoviePilot完整指南：智能批量重命名让媒体库管理更轻松

MoviePilot完整指南：智能批量重命名让媒体库管理更轻松 【免费下载链接】MoviePilot NAS媒体库自动化管理工具 项目地址: https://gitcode.com/gh_mirrors/mo/MoviePilot 你是否厌倦了杂乱无章的媒体文件名？是否经常遇到Plex、Emby等媒体服务器无…

作者头像 李华