Agent 系列(11):A2A 协议——Agent 与 Agent 如何协作
MCP 解决了 Agent ↔ 工具,谁来解决 Agent ↔ Agent?
上一篇讲了 MCP:一个 Agent 通过标准协议连接工具服务。工具是被动的——它等待被调用,执行,返回结果。
但有些场景里,你需要委托的不是一个工具,而是另一个有自主决策能力的 Agent:
- 你有一个研究型 Agent,擅长收集和分析技术资料
- 你有一个写作型 Agent,擅长把分析结论转化成可读文档
- 你有一个代码审查 Agent,擅长发现安全问题
当这三个 Agent 需要协作时,它们之间怎么沟通?谁知道谁的存在?怎么传递工作?
这就是A2A(Agent-to-Agent)协议要解决的问题。
MCP: Agent ←→ 工具/数据源(垂直集成,Agent 主动调工具) A2A: Agent ←→ Agent (水平协作,Agent 委托给 Agent)A2A 的三个核心概念
AgentCard:Agent 的"名片"
每个 Agent 发布一张AgentCard,描述它能做什么:
froma2a.typesimportAgentCard,AgentSkilldefmake_skill(skill_id,name,description,tags):s=AgentSkill()s.id=skill_id s.name=name s.description=description s.tags.extend(tags)returns research_card=AgentCard()research_card.name="research-agent"research_card.description="Gathers factual background on technical topics"research_card.skills.append(make_skill("research","Research","Collect key facts on a topic",["research","facts"]))AgentCard 的关键字段:name、description、skills(每个 skill 有tags用于发现)。
这就像是 OpenAPI Spec 的 Agent 版本——机器可读的能力声明。
Task:工作单元
Agent 间传递的不是函数调用,而是Task:
froma2a.typesimportTask,TaskState,TaskStatus,Message,Part,Role task=Task()task.id=str(uuid.uuid4())task.status.state=TaskState.TASK_STATE_SUBMITTED# 输入:User 角色的 Messagemsg=Message()msg.role=Role.ROLE_USER part=Part();part.text="Should I use Python or Go?"msg.parts.append(part)task.history.append(msg)Task有生命周期状态:SUBMITTED → WORKING → COMPLETED / FAILED。完成时,Agent 把结果追加为ROLE_AGENT的 Message。
Registry:Agent 的黄页
AgentRegistry存储所有注册的 AgentCard,支持按 tag 发现:
classAgentRegistry:defregister(self,card:AgentCard,handler:Callable[[Task],Task])->None:self._agents[card.name]=AgentEntry(card=card,handler=handler)defdiscover(self,tag:str)->list[AgentCard]:"""返回所有 skill 包含指定 tag 的 Agent"""...defdelegate(self,agent_name:str,input_text:str)->Task:"""创建 Task 并通过注册的 handler 执行"""...Demo 1:直接调用——简单但耦合
没有协议时,orchestrator 直接调用三个 Python 函数:
defdirect_orchestrator(question:str)->str:research=research_agent_fn(question)# 硬依赖analysis=analysis_agent_fn(research)# 硬依赖answer=writing_agent_fn(analysis)# 硬依赖returnanswer实测结果:
→ calling research_agent (direct) → calling analysis_agent (direct) → calling writing_agent (direct) Answer: Choose Python if you need rapid development with broad libraries. Select Go if performance and concurrency are critical...功能没问题。问题是:orchestrator 里写死了三个函数名。替换任何一个 Agent,需要修改 orchestrator 代码。如果 orchestrator 在另一个服务里,就是跨服务改代码。
Demo 2:A2A AgentCard 注册表
注册三个 Agent,每个带不同的 skill tag:
[registry] registered: research-agent [registry] registered: analysis-agent [registry] registered: writing-agent发现测试:
researchers=registry.discover("research")# → Found: research-agent — Gathers factual background on technical topicswriters=registry.discover("writing")# → Found: writing-agent — Composes clear technical prose from analysis outputorchestrator 完全不写 Agent 名字,只按 tag 发现:
defa2a_orchestrator(question:str)->str:researchers=registry.discover("research")t1=registry.delegate(researchers[0].name,question)analysts=registry.discover("analysis")t2=registry.delegate(analysts[0].name,task_output(t1))writers=registry.discover("writing")t3=registry.delegate(writers[0].name,task_output(t2))returntask_output(t3)实测执行:
→ delegating to research-agent (discovered via tag) → delegating to analysis-agent (discovered via tag) → delegating to writing-agent (discovered via tag) Answer: Choose Python for rapid development; Go for high-throughput performance...关键差别:orchestrator 代码里没有出现research-agent、writing-agent这些名字。新注册一个writing-agent-v2(带同样的writingtag),orchestrator 立刻可以发现并使用它——零代码改动。
Demo 3:LLM 驱动的 Agent 路由
A2A 最强大的用法:LLM 读取 AgentCard catalog,自己决定调用哪些 Agent、按什么顺序。
向 LLM 展示 Agent 目录:
Available agents: research-agent: Gathers factual background [skills: Research(research, facts)] analysis-agent: Analyzes research notes [skills: Analysis(analysis, tradeoffs)] writing-agent: Composes technical prose [skills: Writing(writing, prose)]LLM 输出执行计划:
["research-agent","analysis-agent","writing-agent"]按计划执行:
Executing 3 agents: → delegating to research-agent → delegating to analysis-agent → delegating to writing-agent Final answer: Choose Python for rapid development; Go for high-throughput...这是 A2A 的终态:不需要预先配置 orchestrator,LLM 根据任务需求和 AgentCard 描述,在运行时自主规划协作链路。
MCP vs A2A vs ANP:协议选型矩阵
维度 MCP A2A ────────────────────────────────────────────────────────────────────── 解决什么问题 Agent ↔ 工具/数据源 Agent ↔ Agent 发现机制 list_tools()(工具目录) discover()(Agent 注册表) 工作单元 Tool call(同步) Task(异步就绪) 耦合方式 Agent 直接使用工具 Orchestrator 委托给 Agent 另一端的性质 被动的工具服务 有自主逻辑的 Agent 跨服务 工具是独立进程 Agent 是独立服务四种协作方式的完整选型:
场景 推荐方案 ────────────────────────────────────────────────────── 同一代码库,调用确定 直接函数调用 Agent 需要调用外部工具 MCP 协议(tools as service) Agent 委托给专业 Agent A2A 协议(agents as service) 跨组织大规模 Agent 网络 ANP(去中心化发现,Web3 风格)设计 Checklist
AgentCard 设计
description用一句话描述 Agent 擅长什么,LLM 会读它来决策skill.tags用语义明确的标签(research、analysis、writing),不用版本号或 IDAgentCard应该是机器可读、人类也能理解的(参考 OpenAPI 风格)
Task 设计
Task.id用 UUID,便于追踪和幂等重试- 用
history(Message 链)传递上下文,而不是在 part.text 里拼接所有历史 - 区分
ROLE_USER(输入)和ROLE_AGENT(输出)的 Message
Registry 与发现
- 一个 tag 可以对应多个 Agent(负载均衡、A/B 测试)
- 支持按 tag 筛选,也支持按 name 精确查找
- 生产环境用持久化 Registry(数据库或服务注册中心),而非内存字典
LLM 驱动路由
- Agent catalog 要简洁:name + description + skill tags,不要把整个 AgentCard JSON 塞给 LLM
- 解析 LLM 输出时做校验(
agent_name not in registry),给出兜底路径 - 记录 LLM 生成的执行计划,便于事后分析和调试
总结
五个核心结论:
- A2A 和 MCP 不竞争,互补:MCP 是 Agent 调工具,A2A 是 Agent 委托 Agent;一个系统里可以同时用两种协议
- AgentCard 是 A2A 的核心:它让 Agent 变成可被发现、可被组合的服务单元,而不是硬编码的依赖
- Task 是比函数调用更丰富的工作单元:有生命周期状态,支持异步,可以携带结构化历史
- 直接调用 vs A2A 的分界线:同一团队、同一代码库 → 直接调用;跨服务、需要解耦 → A2A
- LLM 驱动路由是 A2A 的最高形态:Agent 根据任务描述自主规划协作链路,无需预先配置 orchestrator
下一篇:Agent 评估框架—— 怎么系统地测试 Agent?用什么指标衡量好坏?DeepEval 怎么用?
参考资料
- A2A Protocol 官方规范
- A2A Python SDK
- ANP(Agent Network Protocol)
- 本系列完整 Demo 代码:agent-10-a2a
更多实用知识和有趣产品,欢迎访问我的个人主页
