单个大模型应用通常只需要处理一条请求链路：接收输入、调用模型、返回结果。当系统进一步拆分为规划代理、检索代理、执行代理和审核代理时，问题就从“模型能不能回答”变成了“多个代理能不能可靠地协作”。
常见故障包括：工具参数没有统一格式，代理之间传递了无法解析的自然语言；执行代理重复提交订单或重复发送通知；一个代理失败后，上游无法判断任务是否已经执行；模型供应商更换后，调用代码、重试策略和审计字段全部散落在业务逻辑中。
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 os
import time
import uuid
import requests
AGENT_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_wait
while 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 data
time.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 os
from openai import OpenAI
client = 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，再实现只读工具，随后加入写操作和人工审批，最后根据真实失败类型设计重试与补偿。模型和接入服务只是链路中的一个可替换组件，协议、权限和业务状态才是系统长期稳定性的基础。
