当企业里的 AI 智能体越来越多,一个绕不开的问题浮出水面:A2A 协议(Agent2Agent Protocol,谷歌于 2025 年开源的多智能体互操作开放标准)正是为解决”智能体之间如何互相发现、委派任务、交换结果”而生。本文从概念一路写到可运行代码,带你吃透这套正在成为行业事实标准的协议,并把它和 MCP、ANP 彻底区分开。
一、为什么需要 A2A:多智能体的”巴别塔”困境
设想这样一个场景:你的公司已经上线了”报销 Agent””法务审核 Agent””客服 Agent”,它们分别由不同团队、用不同框架、跑在不同服务里。当老板想要一个”差旅自动报销并同步法务留痕”的流程时,三个 Agent 却谁也不认识谁——没有统一的服务发现,没有标准的任务描述,没有约定的结果格式。这就是多智能体的”巴别塔”困境。
过去我们靠硬编码的 HTTP 接口把 Agent 串起来,一旦对方升级字段就集体崩溃。从零手写 AI Agent 时我们关注的是”单个 Agent 怎么思考”,而当规模来到十几个甚至上百个 Agent,真正的瓶颈变成了”Agent 与 Agent 之间怎么协作”。A2A 就是为这件事设计的:它定义了一套与具体框架无关、与传输协议无关、只关心”能力声明 + 任务生命周期”的轻量协议。
二、A2A 与 MCP、ANP 到底什么关系
很多人把 A2A、MCP、ANP 混为一谈,其实它们的定位完全不同。一句话记住:MCP 解决 Agent 调用工具,A2A 解决 Agent 调用 Agent,ANP 解决跨组织 Agent 的开放互联。
| 维度 | A2A(Agent2Agent) | MCP(模型上下文协议) | ANP(智能体网络协议) |
|---|---|---|---|
| 发起方 | Google 主导 | Anthropic 主导 | 开源社区 |
| 核心场景 | Agent 间协作、任务委派 | Agent 调用工具/数据源 | 跨信任域的 Agent 互联 |
| 关键抽象 | AgentCard / Task / Artifact | Tool / Resource / Prompt | DID / 描述协议 |
| 类比 | 企业内部的”工单系统” | 函数的”标准 RPC” | 跨公司的”外交协议” |
实践中三者并不互斥:一个超级 Agent 完全可以用 MCP 去连数据库和搜索引擎,同时用 A2A 把”写代码””做图””发邮件”三个子 Agent 编排起来。理解这一点,架构就清晰了。
三、A2A 的四大核心对象
1. AgentCard:能力的”身份证”
每个 A2A Agent 都在一个固定路径(通常是 /.well-known/agent.json)暴露一张 AgentCard,声明自己能做什么、支持哪些输入输出模式、需要什么鉴权。这是服务发现的入口。
2. Task:一次协作的基本单位
任务有状态机:submitted → working → input-required → completed / failed。需要多轮交互的任务会进入 input-required,等对方补充信息后再继续。
3. Message 与 Artifact:对话与产物
Message 是往返的对话单元(文本、文件、结构化数据),Artifact 是任务最终交付的产物。两者都用多部分(multipart)设计以支持大文件和流式。
四、五分钟跑通:用 Python 起一个最小 A2A Agent
下面用 FastAPI 起一个”翻译 Agent”,它暴露 AgentCard,并响应 tasks/send。代码刻意保持精简,突出 A2A 的数据契约而非某个 SDK 的具体版本。
from fastapi import FastAPI
from pydantic import BaseModel
app = FastAPI()
# 1) 暴露 AgentCard 供发现
@app.get("/.well-known/agent.json")
def agent_card():
return {
"name": "翻译 Agent",
"description": "把文本翻译为指定语言",
"url": "http://localhost:8001",
"capabilities": {"streaming": False},
"skills": [{"id": "translate", "name": "翻译",
"inputModes": ["text/plain"],
"outputModes": ["text/plain"]}],
}
class TaskSend(BaseModel):
id: str
message: dict
@app.post("/tasks/send")
def tasks_send(task: TaskSend):
text = task.message["parts"][0]["text"]
lang = task.message.get("metadata", {}).get("target", "en")
# 这里用占位翻译,真实场景接 LLM
translated = f"[{lang}] {text}"
return {
"id": task.id,
"status": {"state": "completed"},
"artifacts": [{"parts": [{"text": translated}]}],
}
启动后访问 /tasks/send 即可拿到结构化结果,这正是 A2A 想统一的”任务进出”形态。
五、客户端如何发现并派发任务
调用方先拉 AgentCard 了解能力,再按契约发任务。下面的客户端演示了”先发现、后委派”的标准两步。
import requests
BASE = "http://localhost:8001"
# 第 1 步:发现对端能力
card = requests.get(f"{BASE}/.well-known/agent.json").json()
print("对方支持技能:", [s["id"] for s in card["skills"]])
# 第 2 步:派发一个任务
payload = {
"id": "task-1001",
"message": {
"role": "user",
"parts": [{"text": "你好,世界"}],
"metadata": {"target": "en"},
},
}
resp = requests.post(f"{BASE}/tasks/send", json=payload).json()
print("翻译结果:", resp["artifacts"][0]["parts"][0]["text"])
六、多智能体编排实战:任务拆解与派发
真正的价值在多 Agent 协同。下面这个”主管 Agent”把一句需求拆给”翻译 Agent”和”摘要 Agent”,再汇总。它复用了上一节的调用逻辑,只是多了一步路由。
def dispatch(agents: dict, task_text: str) -> dict:
results = {}
for name, base in agents.items():
card = requests.get(f"{base}/.well-known/agent.json").json()
# 按 skill 名称做简单路由
skill = card["skills"][0]["id"]
payload = {"id": f"{name}-1",
"message": {"role": "user",
"parts": [{"text": task_text}]}}
r = requests.post(f"{base}/tasks/send", json=payload).json()
results[name] = r["artifacts"][0]["parts"][0]["text"]
return results
agents = {"translate": "http://localhost:8001",
"summary": "http://localhost:8002"}
print(dispatch(agents, "把这段产品说明翻译并总结成三句话"))
这段不到二十行的代码,已经具备”服务发现 + 技能路由 + 结果聚合”的雏形,正是 LangChain Agent 这类框架在底层悄悄做的事,而 A2A 把它标准化、跨框架化了。
七、与 MCP 协同:边界要划清
落地时最容易踩的坑是”什么都往 A2A 里塞”。请记住分界:凡是要访问一个具体工具或数据源(查库、读文件、调 API),走 MCP;凡是需要另一个会思考的 Agent 来接力,走 A2A。例如”翻译 Agent”内部用 MCP 连术语库,对外则通过 A2A 被主管 Agent 调用。这样既保证了工具调用的标准化,又让 Agent 协作不被框架绑架。
在评估这类多 Agent 系统时,也别忘了用 AI Agent 评测 的方法量化协作质量,而不是只看单 Agent 的聪明程度。
八、生产落地的三条底线
鉴权:AgentCard 之外必须验身份
公开暴露 AgentCard 没问题,但 /tasks/send 必须加 OAuth2 / mTLS 之类的鉴权,否则等于把内部能力裸奔到公网。
可观测:给每个 Task 一个 traceId
多跳调用一旦出错极难定位,建议把 A2A 的任务 id 透传到日志与链路追踪系统,配合 Function Calling 的调用日志一起看。
超时与退避:跨 Agent 调用默认要短
Agent 间网络跳数多,务必设置整体超时与指数退避,避免一个慢 Agent 拖垮整条编排链路。
九、两种通信模式:同步请求与流式推送
A2A 并不强制传输方式。最简单的情况,一次 tasks/send 走同步 HTTP 请求/响应,适合几秒内就能返回的任务。但对于”生成一份五十页行业报告”这类长任务,同步调用必然超时。此时应使用 tasks/sendSubscribe 走 SSE(Server-Sent Events)流式返回中间状态,或者采用”任务创建 + 主动推送”的异步模式:调用方先拿到 task id,Agent 完成后通过预先登记的 webhook 把 Artifact 推回来。三种模式的核心取舍是延迟敏感度与任务时长——选错就会要么白白占着连接,要么永远拿不到最终结果。
在流式场景下,建议给每个增量 Artifact 都带上递增序号,客户端据此做断点续收;一旦网络抖动断开,用 task id 重新订阅即可,无需从头再来。这和 Function Calling 流式处理里强调的”可恢复”思路一脉相承。落地时把传输模式写进 AgentCard 的 capabilities 字段,调用方就能在一开始就知道对方支持哪种协作方式,避免运行时才发现不支持流式而整体降级。
十、选型清单:什么时候该上 A2A
最后给一张务实的判定表:如果你的智能体数量 ≤ 2,硬编码调用更省事;如果智能体之间需要”思考接力”且来自不同框架/团队,A2A 的收益立刻显现;如果还要跨越公司边界,则叠加 ANP 这类开放互联协议。不要为了用而用——A2A 的价值在于”解耦”,当你开始为 Agent 之间的耦合而痛苦时,它就是对症的药。
结语
A2A 协议把”多智能体怎么协作”这件过去靠私拉硬编码的事,变成了可发现、可编排、可替换的标准能力。它不取代 MCP,而是和 MCP、ANP 一起,构成智能体时代的”网络协议栈”。如果你的 Agent 数量已经超过三个,现在就是引入 A2A 理清协作边界的最佳时机。




