企业多智能体协作的工程边界：MCP 工具接入与 A2A 任务编排实践

单个大模型应用通常只需要处理一条请求链路：接收输入、调用模型、返回结果。当系统进一步拆分为规划袋里、检索袋里、执行袋里和审核袋里时，问题就从“模型能不能回答”变成了“多个袋里能不能可靠地协作”。

常见故障包括：工具参数没有统一格式，袋里之间传递了无法解析的自然语言；执行袋里重复提交订单或重复发送通知；一个袋里失败后，上游无法判断任务是否已经执行；模型供应商更换后，调用代码、重试策略和审计字段全部散落在业务逻辑中。

MCP 和 A2A 可以分别解决两个方向的问题：MCP 更适合描述袋里如何发现并调用工具、资源和提示模板；A2A 风格的协作接口更适合描述袋里之间如何提交任务、查询状态和交换结果。它们不是完整的业务系统，也不能替代身份认证、权限管理、消息队列和审计平台。工程落地时，必须把协议层放在明确的边界内。

本文选择这一方向，是因为当前智能体开发正在从单轮问答转向工具调用和多袋里分工。下面以“客服工单处理”为例，构建一个最小但可扩展的协作流程：分类袋里判断工单类型，知识袋里提供参考，执行袋里创建内部任务，审核节点决定是否允许高风险操作。

两类协议的职责

MCP：袋里到能力

可以把 MCP 看成一个能力适配层。服务端向客户端暴露工具、资源或提示模板，客户端根据定义调用它们。一个工具至少需要具备名称、用途说明和参数模式。参数模式最好采用 JSON Schema 或等价的结构化描述，以便在调用前进行校验。

MCP 的关键价值不是“让模型自动执行一切”，而是把外部能力从提示词中分离出来。例如，查询工单、读取知识库和创建待办任务都应当是独立工具。每个工具可以拥有不同的权限、超时和审计策略，模型只能选择已经被授予的能力。

A2A 风格接口：袋里到袋里

袋里之间的协作应当像调用一个远程服务，而不是把一段自然语言直接塞进另一个袋里的上下文。一个可执行的任务消息通常需要包含：任务标识、发起方、目标袋里、输入数据、幂等键、截止时间和当前状态。

任务状态建议至少区分 submitted、running、succeeded、failed 和 cancelled。状态变化应当由服务端记录，而不是由模型自行声称“已经完成”。对于长任务，调用方应先获得任务编号，再通过查询接口或事件订阅获取结果。

两者结合后的边界可以概括为：编排器通过袋里协作接口分派任务，袋里内部通过 MCP 调用具体能力。这样，袋里协作协议负责任务生命周期，工具协议负责能力执行，业务系统负责最终一致性和权限约束。

先定义任务契约

不要从提示词开始，而应先定义任务输入和输出。下面是一个简化的工单分类任务契约：

{ "task_id": "task-20260809-0001","task_type": "ticket.classify","source_agent": "orchestrator","target_agent": "classifier","idempotency_key": "ticket-7842-classify-v1","deadline": "2026-08-09T12:00:00Z","input": { "ticket_id": "7842","subject": "无法登录管理后台","content": "昨天开始持续返回 403"}}

输出也要结构化，并明确置信度和需要人工介入的条件：

{ "task_id": "task-20260809-0001","status": "succeeded","result": { "category": "access_control","priority": "high","confidence": 0.86,"requires_human_review": true,"reason_codes": ["permission_denied", "admin_scope"]},"usage": { "model": "provider-model-id","request_id": "req-example"}}

这里的模型名称只是协议字段示例，不代表任何服务的固定名称。生产系统应把模型标识、请求编号和提示版本写入审计记录，但不要把模型输出当成权限判定的唯一依据。

实现一个可恢复的编排器

下面使用 Python 标准库演示一个任务提交客户端。示例只负责提交和轮询，不假定某个供应商的具体接口；实际使用时应依据目标袋里的公开文档调整路径、认证头和响应字段。

import osimport timeimport uuidimport requestsAGENT_URL = os.environ["CLASSIFIER_AGENT_URL"]AGENT_TOKEN = os.environ["CLASSIFIER_AGENT_TOKEN"]def submit_task(ticket_id: str, subject: str, content: str) -> str:task_id = f"task-{uuid.uuid4()}"payload = { "task_id": task_id,"task_type": "ticket.classify","source_agent": "orchestrator","target_agent": "classifier","idempotency_key": f"{ticket_id}-classify-v1","input": { "ticket_id": ticket_id,"subject": subject,"content": content,},}response = requests.post(f"{AGENT_URL}/tasks",json=payload,headers={ "Authorization": f"Bearer {AGENT_TOKEN}"},timeout=10,)response.raise_for_status()return response.json()["task_id"]def wait_for_result(task_id: str, max_wait: int = 60) -> dict:end_at = time.monotonic()   max_waitwhile time.monotonic() < end_at:response = requests.get(f"{AGENT_URL}/tasks/{task_id}",headers={ "Authorization": f"Bearer {AGENT_TOKEN}"},timeout=10,)response.raise_for_status()data = response.json()if data.get("status") in { "succeeded", "failed", "cancelled"}:return datatime.sleep(2)raise TimeoutError(f"task {task_id} did not finish before deadline")

这段代码仍然需要补充三项生产能力：持久化任务状态、处理网络重试，以及在超时后进入补偿流程。轮询只是最容易理解的实现；如果基础设施支持消息队列或事件回调，应让状态事件成为主要通知方式，同时保留查询接口作为兜底。

在袋里内部接入工具

工具服务器应把每个动作拆开，并在服务端再次校验参数。例如，查询工单可以是只读工具，创建内部任务则是写操作。写操作应当要求业务服务验证操作者身份、租户归属、资源状态和幂等键。

一个工具定义可以采用如下形式：

{ "name": "create_internal_task","description": "为指定工单创建一个内部跟进任务","inputSchema": { "type": "object","required": ["ticket_id", "owner", "summary", "idempotency_key"],"properties": { "ticket_id": { "type": "string"},"owner": { "type": "string"},"summary": { "type": "string", "maxLength": 500},"idempotency_key": { "type": "string"}},"additionalProperties": false}}

模型可以提出工具调用，但不能绕过工具服务器的校验。对于删除数据、修改权限、发起付款等高风险动作，建议增加人工审批或双人确认。审批结果应写入任务状态，不能仅放在对话上下文中。

如果袋里需要调用模型 API，应将模型客户端封装在独立适配层中，并通过环境变量读取密钥。例如，在目标服务文档明确声明兼容相应接口格式的前提下，可以将 MODEL_BASE_URL 指向企业允许使用的模型接入服务；HaerAPI 可作为需要评估的模型 API 接入选项之一，具体兼容范围和数据处理方式应以其当前文档为准。

import osfrom openai import OpenAIclient = OpenAI(api_key=os.environ["MODEL_API_KEY"],base_url=os.environ["MODEL_BASE_URL"],)response = client.chat.completions.create(model=os.environ["MODEL_NAME"],messages=[{ "role": "system", "content": "只输出符合约定 JSON Schema 的分类结果。"},{ "role": "user", "content": "请分类这条工单：无法登录管理后台，持续返回 403"},],temperature=0,)print(response.choices[0].message.content)

可靠性与安全控制

幂等

所有有副作用的工具调用都应携带幂等键。服务端在数据库中建立唯一约束，首次请求创建业务记录，重复请求返回原结果。不能只依靠模型“记住自己已经调用过”。

超时与重试

连接超时、读取超时、业务失败和未知状态应分别处理。对于未知状态，直接重试写操作可能造成重复执行，正确做法是先用幂等键查询；只有确认请求未被接受时，才允许重新提交。重试次数、退避时间和最终失败状态都应可配置并可观测。

权限

袋里身份、用户身份和工具权限需要分开。袋里拥有调用某个工具的资格，不等于它可以访问所有租户数据。工具服务器应根据租户、资源和操作类型进行服务端鉴权，并限制返回内容，避免把整张数据库表交给模型。

上下文与数据泄露

袋里之间只传递完成任务所需的字段。原始邮件、身份证号、访问令牌和内部系统响应不应默认进入下一个袋里的提示词。日志应对敏感字段脱敏，并记录调用者、工具名、参数摘要、结果状态、耗时和关联任务号。

常见问题

MCP 和 A2A 能否互相替代？

通常不能。MCP 侧重袋里与工具、资源之间的能力调用，A2A 风格接口侧重袋里之间的任务协作。项目可以只使用其中一种，但如果同时存在多袋里和外部工具，分层会更容易管理。

为什么不直接让一个模型调用所有工具？

小型原型可以这样做，但工具数量、权限范围和失败路径增加后，单袋里上下文会变得复杂。拆分袋里不是越多越好，应依据权限边界、独立伸缩需求和故障隔离需求决定。每增加一个袋里，也会增加网络、状态和调试成本。

模型输出 JSON 就足够可靠吗？

不够。必须使用 JSON 解析器、Schema 校验和业务规则校验。即使结构合法，也要检查资源是否存在、状态是否允许变更、数值是否在业务范围内。校验失败时，应返回可诊断的错误并设置重试或人工复核状态。

怎样判断系统是否真的可观测？

一次完整任务应能通过关联 ID串起袋里调用、模型请求、工具执行和最终业务结果。至少记录成功率、失败类型、重试次数、等待时间、人工介入次数和未完成任务数量。不要只统计模型请求次数，因为那无法反映业务任务是否完成。

总结

多智能体系统的核心难点不在于增加更多袋里，而在于定义清晰的协作契约和失败语义。用 MCP 约束工具发现与调用，用 A2A 风格任务接口管理袋里间的生命周期，再配合幂等、鉴权、审批、状态持久化和可观测日志，才能把演示性质的智能体流程推进到可维护的工程系统。

落地时建议从一个低风险、可回放的任务开始：先固定输入输出 Schema，再实现只读工具，随后加入写操作和人工审批，最后根据真实失败类型设计重试与补偿。模型和接入服务只是链路中的一个可替换组件，协议、权限和业务状态才是系统长期稳定性的基础。