30秒快速回答: A2A(Agent2Agent Protocol)是 Google 主导的开放标准,解决「AI Agent 之间怎么通信」的问题。它让一个 Agent 能发现其他 Agent 的能力、把子任务派给它们、并接收结构化结果。A2A 与 MCP 是互补关系——MCP 管「Agent 连工具」,A2A 管「Agent 连 Agent」,两者共同构成多智能体系统的通信底座。
为什么需要 A2A?
2025-2026 年,AI 应用从「单 Agent 解决一切」转向「多 Agent 协作」。但不同团队开发的 Agent 各自为政:
没有 A2A 的世界:
Agent A(写代码)← 不知道 → Agent B(查数据库)
Agent C(画图) ← 不知道 → Agent D(发邮件)
每个 Agent 都是信息孤岛,无法协作。
A2A 让所有 Agent 说同一种「语言」:
有 A2A 的世界:
Agent A 发现 Agent B 能查数据库 → 派任务 → 收结果
Agent C 发现 Agent D 能发邮件 → 派任务 → 收确认
多 Agent 协作网络自动形成。
A2A 的核心概念
Agent Card(Agent 名片)
每个 A2A Agent 都暴露一个 agent.json 名片,描述自己的能力:
{
"name": "DatabaseQueryAgent",
"description": "查询公司内部 PostgreSQL 数据库",
"url": "https://db-agent.internal/a2a",
"capabilities": {
"streaming": true,
"pushNotifications": false
},
"skills": [
{
"id": "query_users",
"name": "查询用户表",
"description": "按条件查询 users 表",
"tags": ["database", "users"],
"examples": ["查最近7天注册的用户"]
},
{
"id": "query_orders",
"name": "查询订单表",
"description": "按条件查询 orders 表",
"tags": ["database", "orders"]
}
]
}
任务生命周期
Client Agent Server Agent
| |
|── TaskSend (派任务) ──────→|
| | 执行中...
|←── TaskStatusUpdate ───────| (streaming)
| | 执行完成
|←── TaskArtifact (结果) ────|
| |
A2A 定义了完整的任务状态机:submitted → working → input-required → completed → failed。
A2A vs MCP:互补而非竞争
| 维度 | MCP | A2A |
|---|---|---|
| 连接对象 | Agent ↔ 工具/数据源 | Agent ↔ Agent |
| 提出者 | Anthropic | |
| 核心概念 | Resources / Tools / Prompts | Agent Card / Task / Message |
| 通信方式 | JSON-RPC(stdio/SSE) | HTTP + JSON-RPC + SSE |
| 类比 | AI 的 USB 接口 | AI 的 TCP/IP 协议 |
| 关系 | 底层工具调用 | 上层 Agent 协作 |
实际架构:
┌─────────────────────────────────────────┐
│ 多 Agent 协作层 │
│ Agent A ──A2A── Agent B ──A2A── Agent C
└──────────┬──────────┬──────────┬────────┘
│ MCP │ MCP │ MCP
↓ ↓ ↓
[数据库] [文件系统] [API服务]
A2A 负责 Agent 之间的「对话」,MCP 负责 Agent 与工具的「操作」。
实战:用 Python 搭建 A2A Agent
# 安装: pip install a2a-sdk
from a2a_sdk import AgentCard, AgentSkill, AgentExecutor, Task
from a2a_sdk.server import A2AServer
# 1. 定义 Agent 能力
skill = AgentSkill(
id="weather_query",
name="天气查询",
description="查询指定城市的实时天气",
tags=["weather", "query"],
examples=["北京今天天气怎么样?"]
)
agent_card = AgentCard(
name="WeatherAgent",
description="提供全球城市天气查询服务",
url="http://localhost:8080/a2a",
version="1.0.0",
skills=[skill]
)
# 2. 实现任务处理逻辑
class WeatherExecutor(AgentExecutor):
async def execute(self, task: Task):
# 解析任务输入
query = task.input["message"]
city = self._extract_city(query)
# 调用天气 API(实际项目)
result = f"{city} 今天晴,25°C,湿度 60%"
# 返回结构化结果
return Task(
id=task.id,
status="completed",
artifacts=[{"type": "text", "content": result}]
)
# 3. 启动 A2A 服务
server = A2AServer(
agent_card=agent_card,
executor=WeatherExecutor()
)
server.run(host="0.0.0.0", port=8080)
Client 端调用
from a2a_sdk import A2AClient
# 发现 Agent
client = A2AClient(url="http://localhost:8080/a2a")
# 派发任务
task = await client.send_task(
message="北京今天天气怎么样?"
)
print(task.artifacts[0]["content"])
# 输出: 北京 今天晴,25°C,湿度 60%
2026 年 A2A 生态
| 参与方 | 角色 | 说明 |
|---|---|---|
| 发起者 | 提出协议规范,开源 SDK | |
| Salesforce | 早期采用 | Agentforce 支持 A2A |
| LangChain | 框架集成 | LangGraph 可发布为 A2A Agent |
| Microsoft | 生态支持 | Copilot Studio 接入 A2A |
| CrewAI | 框架集成 | 多 Agent 协作支持 A2A 协议 |
常见问题
Q: A2A 和 MCP 会合并吗?
目前不会。两者解决不同层次的问题,Google 和 Anthropic 已明确表示将保持协作关系。实际项目中通常同时使用两者。
Q: A2A 安全吗?
A2A 支持 OAuth 2.0 认证、Agent 间 TLS 加密通信。企业部署时建议配合 API Gateway 做访问控制。
Q: 小项目需要 A2A 吗?
如果只有一个 Agent 或几个硬编码协作的 Agent,不需要 A2A。A2A 的价值在「动态发现 + 跨组织协作」场景——比如你的 Agent 需要调用第三方提供的 Agent 服务。
下一步建议: 用 a2a-sdk 把你的一个 LangGraph 工作流包装成 A2A Agent,体验「让其他 Agent 发现并调用你的能力」的感觉。