作者：PySuper | 来源：zhengxingtao.com
150+ 组织采用，22K+ GitHub Stars，Linux 基金会治理——Google 的 A2A 协议让不同 Agent 真正"说上话"。如果说 MCP 是 Agent 连工具的"USB-C"，那 A2A 就是 Agent 之间互连的"互联网"。
目录
A2A 是什么：打破 Agent 孤岛
核心概念：Agent Card / Task / Part
协议流程：Request → Negotiation → Execution → Response
实战1：用 Python 实现一个 A2A Agent，发布 Agent Card
实战2：两个 Agent 通过 A2A 协作完成跨系统任务
与 MCP 的关系：垂直 vs 水平
与 Skill 的关系：封装能力 vs 发现调度能力
Google ADK 集成
生产部署：Agent 发现、身份认证、跨云联邦
踩坑记录
总结与展望
1. A2A 是什么：打破 Agent 孤岛
1.1 痛点：Agent 孤岛
2026 年，企业内部可能有几十个 AI Agent——客服 Agent、财务 Agent、HR Agent、运维 Agent。但它们彼此无法通信，每个都是"孤岛"：
plaintext
┌──────────────────────────────────────────────────────────────┐
│ Agent 孤岛问题 │
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ 客服 │ │ 财务 │ │ HR │ │ 运维 │ │
│ │ Agent │ │ Agent │ │ Agent │ │ Agent │ │
│ │ (厂商A) │ │ (厂商B) │ │ (厂商C) │ │ (厂商D) │ │
│ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │
│ ❌ 互不相通 ❌ 互不相通 ❌ 互不相通 │
│ │
│ 用户："帮我处理退货退款" │
│ → 客服 Agent 只能处理退货 │
│ → 需要手动通知财务 Agent 处理退款 │
│ → 需要手动通知 HR Agent 更新客户记录 │
│ → 流程断裂，效率低下 │
└──────────────────────────────────────────────────────────────┘
1.2 A2A 的解法
据《A2A Protocol Explained》（https://stellagent.ai/insights/a2a-protocol-google-agent-to-agent），A2A 的定位是：
A2A is an open protocol that lets AI agents discover, communicate, and collaborate as peers across frameworks and vendors.
翻译：A2A 是一个开放协议，让不同框架和厂商的 AI Agent 以对等方式发现、通信和协作。
plaintext
┌──────────────────────────────────────────────────────────────┐
│ A2A 解决方案 │
│ │
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
│ │ 客服 │ │ 财务 │ │ HR │ │ 运维 │ │
│ │ Agent │ │ Agent │ │ Agent │ │ Agent │ │
│ └────┬─────┘ └────┬─────┘ └────┬─────┘ └────┬─────┘ │
│ │ │ │ │ │
│ ═══════════════╪══════════════╪═══════════════ │
│ │ │ │
│ ┌───────▼──────────────▼────────┐ │
│ │ A2A 协议层 │ │
│ │ (Agent 发现 + 通信 + 协作) │ │
│ └───────────────────────────────┘ │
│ │
│ 用户："帮我处理退货退款" │
│ → 客服 Agent 接收请求 │
│ → 通过 A2A 发现财务 Agent 的退款能力 │
│ → 通过 A2A 委托财务 Agent 执行退款 │
│ → 通过 A2A 通知 HR Agent 更新记录 │
│ → 全流程自动化，无需人工介入 │
└──────────────────────────────────────────────────────────────┘
1.3 关键数据
表格
2. 核心概念：Agent Card / Task / Part
2.1 五大核心概念
plaintext
┌──────────────────────────────────────────────────────────────┐
│ A2A 五大核心概念 │
├──────────────┬──────────────────────────────────────────────┤
│ 概念 │ 说明 │
├──────────────┼──────────────────────────────────────────────┤
│ Agent Card │ Agent 的"能力名片" │
│ │ JSON 格式，发布在 /.well-known/agent-card.json│
│ │ 声明名称、能力、技能列表、认证方式 │
│ │ 其他 Agent 通过标准 URI 自动发现 │
├──────────────┼──────────────────────────────────────────────┤
│ Task │ 工作单元，有完整生命周期 │
│ │ submitted → working → input-required │
│ │ → completed / failed / canceled / rejected │
│ │ 原生支持长时间异步任务 │
├──────────────┼──────────────────────────────────────────────┤
│ Message │ 通信单元，角色为 "user" 或 "agent" │
│ │ 包含一个或多个 Part │
│ │ 是 Agent 间对话的载体 │
├──────────────┼──────────────────────────────────────────────┤
│ Part │ 原子内容单元 │
│ │ 可以是文本、文件、结构化数据 │
│ │ 多个 Part 组成一条 Message │
├──────────────┼──────────────────────────────────────────────┤
│ Artifact │ Agent 产出的输出物 │
│ │ PDF 报告、JSON 分析结果、图片等 │
│ │ 支持版本控制和增量更新 │
└──────────────┴──────────────────────────────────────────────┘
2.2 Agent Card 详解
Agent Card 是 A2A 的核心创新——它让 Agent 能够被自动发现：
json
{
"name": "Financial Processing Agent",
"description": "Handles payment processing, refunds, and financial reporting",
"version": "1.0.0",
"url": "https://finance.example.com/a2a",
"protocolVersions": ["1.0.0"],
"capabilities": {
"streaming": true,
"pushNotifications": true
},
"authentication": {
"schemes": ["bearer", "oauth2"]
},
"skills": [
{
"id": "process-refund",
"name": "Process Refund",
"description": "Process a customer refund request",
"tags": ["finance", "refund", "payment"],
"examples": [
"Process a refund for order #12345",
"Issue a partial refund of $50"
]
},
{
"id": "generate-report",
"name": "Generate Financial Report",
"description": "Generate financial summary reports",
"tags": ["finance", "reporting"],
"examples": [
"Generate Q1 financial summary",
"Create expense report for last month"
]
}
]
}
2.3 Task 生命周期
plaintext
┌──────────────────────────────────────────────────────────────┐
│ Task 生命周期 │
│ │
│ ┌───────────┐ │
│ │ submitted │ ← Client 创建 Task │
│ └─────┬─────┘ │
│ │ │
│ ▼ │
│ ┌───────────┐ │
│ │ working │ ← Agent 正在处理 │
│ └─────┬─────┘ │
│ │ │
│ ┌────┴────┐ │
│ │ │ │
│ ▼ ▼ │
│ ┌────────────┐ ┌────────────┐ │
│ │input-required│ │auth-required│ ← 需要更多信息或认证 │
│ └──────┬─────┘ └──────┬─────┘ │
│ │ │ │
│ └───────┬───────┘ │
│ │ │
│ ▼ │
│ ┌───────────┐ │
│ │ working │ ← 继续处理 │
│ └─────┬─────┘ │
│ │ │
│ ┌────────┬┼────────┬──────────┐ │
│ ▼ ▼▼ ▼ ▼ │
│ ┌──────────┐ ┌────────┐ ┌────────┐ ┌────────┐ │
│ │completed │ │ failed │ │canceled│ │rejected│ │
│ └──────────┘ └────────┘ └────────┘ └────────┘ │
│ │
│ 终态：任务结束，不可再变 │
└──────────────────────────────────────────────────────────────┘
3. 协议流程：Request → Negotiation → Execution → Response
3.1 完整交互流程
plaintext
┌──────────────────────────────────────────────────────────────┐
│ A2A 完整交互流程 │
│ │
│ Client Agent Remote Agent │
│ ┌────────────┐ ┌────────────┐ │
│ │ │ 1. 发现：GET │ │ │
│ │ │ /.well-known/agent-card.json│ │ │
│ │ │────────────────────────────▶│ │ │
│ │ │ ◀─── Agent Card ───────────│ │ │
│ │ │ │ │ │
│ │ │ 2. 认证：获取 OAuth Token │ │ │
│ │ │ （根据 Agent Card 中的 │ │ │
│ │ │ authentication schemes） │ │ │
│ │ │ │ │ │
│ │ │ 3. 发送任务： │ │ │
│ │ │ message/send │ │ │
│ │ │────────────────────────────▶│ │ │
│ │ │ │ │ │
│ │ │ 4. 流式更新（SSE）： │ │ │
│ │ │ ◀─── TaskStatusUpdate ─────│ │ │
│ │ │ ◀─── TaskArtifactUpdate ───│ │ │
│ │ │ │ │ │
│ │ │ 5. 需要输入： │ │ │
│ │ │ ◀─── input-required ───────│ │ │
│ │ │ ──── 补充输入 ────────────▶│ │ │
│ │ │ │ │ │
│ │ │ 6. 任务完成： │ │ │
│ │ │ ◀─── completed + artifact ─│ │ │
│ └────────────┘ └────────────┘ │
└──────────────────────────────────────────────────────────────┘
3.2 JSON-RPC 方法
A2A 定义了 11 个 JSON-RPC 方法：
表格
4. 实战1：用 Python 实现一个 A2A Agent，发布 Agent Card
4.1 完整代码
python
#!/usr/bin/env python3
"""
A2A Agent 示例：天气查询 Agent
发布 Agent Card 并处理 A2A 请求
作者：PySuper | 来源：zhengxingtao.com
"""
import json
import uuid
from datetime import datetime
from http.server import HTTPServer, BaseHTTPRequestHandler
from typing import Optional
# ============================================================
# Agent Card 定义
# ============================================================
AGENT_CARD = {
"name": "Weather Query Agent",
"description": "Provides current weather information and forecasts for cities worldwide",
"version": "1.0.0",
"url": "http://localhost:8000/a2a",
"protocolVersions": ["1.0.0"],
"capabilities": {
"streaming": False,
"pushNotifications": False
},
"authentication": {
"schemes": ["none"] # 简化示例，生产环境应使用 OAuth2
},
"skills": [
{
"id": "current-weather",
"name": "Get Current Weather",
"description": "Get current weather conditions for a specified city",
"tags": ["weather", "current", "temperature"],
"examples": [
"What's the weather in Beijing?",
"Current temperature in Tokyo"
]
},
{
"id": "weather-forecast",
"name": "Get Weather Forecast",
"description": "Get 5-day weather forecast for a specified city",
"tags": ["weather", "forecast", "prediction"],
"examples": [
"5-day forecast for Shanghai",
"Will it rain in London this week?"
]
}
]
}
# ============================================================
# 模拟天气数据
# ============================================================
def get_weather(city: str) -> dict:
"""获取天气数据（模拟）"""
# 实际项目中应调用天气 API
weather_data = {
"Beijing": {"temp": 25, "condition": "Sunny", "humidity": 45},
"Shanghai": {"temp": 28, "condition": "Cloudy", "humidity": 72},
"Tokyo": {"temp": 22, "condition": "Rainy", "humidity": 88},
"London": {"temp": 15, "condition": "Overcast", "humidity": 80},
"New York": {"temp": 20, "condition": "Partly Cloudy", "humidity": 55},
}
return weather_data.get(city, {"temp": 20, "condition": "Unknown", "humidity": 50})
# ============================================================
# Task 管理
# ============================================================
tasks = {} # task_id -> task_state
def create_task(task_id: str, message: str) -> dict:
"""创建新 Task"""
task = {
"id": task_id,
"status": {"state": "working"},
"messages": [
{
"role": "user",
"parts": [{"type": "text", "text": message}]
}
],
"artifacts": [],
"createdAt": datetime.now().isoformat(),
}
tasks[task_id] = task
return task
def process_task(task_id: str, message: str) -> dict:
"""处理 Task（简化逻辑）"""
task = tasks.get(task_id)
if not task:
task = create_task(task_id, message)
# 解析城市名（简化）
city = "Beijing"
for candidate in ["Beijing", "Shanghai", "Tokyo", "London", "New York"]:
if candidate.lower() in message.lower():
city = candidate
break
# 获取天气
weather = get_weather(city)
# 更新 Task 状态
result_text = (
f"Weather in {city}:\n"
f" Temperature: {weather['temp']}°C\n"
f" Condition: {weather['condition']}\n"
f" Humidity: {weather['humidity']}%"
)
task["status"] = {"state": "completed"}
task["messages"].append({
"role": "agent",
"parts": [{"type": "text", "text": result_text}]
})
task["artifacts"].append({
"id": str(uuid.uuid4()),
"parts": [{"type": "text", "text": result_text}]
})
return task
# ============================================================
# A2A HTTP Handler
# ============================================================
class A2AHandler(BaseHTTPRequestHandler):
"""处理 A2A 协议的 HTTP 请求"""
def do_GET(self):
"""处理 GET 请求（Agent Card 发现）"""
if self.path == "/.well-known/agent-card.json":
self.send_json_response(200, AGENT_CARD)
else:
self.send_json_response(404, {"error": "Not found"})
def do_POST(self):
"""处理 POST 请求（A2A JSON-RPC）"""
content_length = int(self.headers.get("Content-Length", 0))
body = self.rfile.read(content_length).decode("utf-8")
try:
request = json.loads(body)
except json.JSONDecodeError:
self.send_json_response(400, {
"jsonrpc": "2.0",
"error": {"code": -32700, "message": "Parse error"},
"id": None
})
return
method = request.get("method", "")
params = request.get("params", {})
request_id = request.get("id")
if method == "message/send":
result = self.handle_message_send(params)
elif method == "tasks/get":
result = self.handle_tasks_get(params)
elif method == "agent/card":
result = AGENT_CARD
else:
self.send_json_response(200, {
"jsonrpc": "2.0",
"error": {"code": -32601, "message": f"Method not found: {method}"},
"id": request_id
})
return
self.send_json_response(200, {
"jsonrpc": "2.0",
"result": result,
"id": request_id
})
def handle_message_send(self, params: dict) -> dict:
"""处理 message/send 请求"""
task_id = params.get("taskId", str(uuid.uuid4()))
message = params.get("message", {})
# 提取消息文本
text_parts = []
for part in message.get("parts", []):
if part.get("type") == "text":
text_parts.append(part.get("text", ""))
user_text = " ".join(text_parts)
task = process_task(task_id, user_text)
return task
def handle_tasks_get(self, params: dict) -> dict:
"""处理 tasks/get 请求"""
task_id = params.get("taskId")
task = tasks.get(task_id)
if not task:
return {"error": {"code": -32001, "message": f"Task not found: {task_id}"}}
return task
def send_json_response(self, status_code: int, data: dict):
"""发送 JSON 响应"""
self.send_response(status_code)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(json.dumps(data, ensure_ascii=False).encode("utf-8"))
# ============================================================
# 启动 Server
# ============================================================
if __name__ == "__main__":
host = "0.0.0.0"
port = 8000
print(f"[Weather A2A Agent] Starting on {host}:{port}")
print(f"[Weather A2A Agent] Agent Card: http://{host}:{port}/.well-known/agent-card.json")
print(f"[Weather A2A Agent] A2A Endpoint: http://{host}:{port}/a2a")
server = HTTPServer((host, port), A2AHandler)
server.serve_forever()
4.2 测试
bash
# 启动 Agent
python weather_agent.py
# 获取 Agent Card
curl http://localhost:8000/.well-known/agent-card.json | python -m json.tool
# 发送 A2A 请求
curl -X POST http://localhost:8000/ -H "Content-Type: application/json" -d '{
"jsonrpc": "2.0",
"id": 1,
"method": "message/send",
"params": {
"message": {
"role": "user",
"parts": [{"type": "text", "text": "What is the weather in Tokyo?"}]
}
}
}' | python -m json.tool
5. 实战2：两个 Agent 通过 A2A 协作完成跨系统任务
5.1 场景：客服 Agent + 财务 Agent 协作
plaintext
┌──────────────────────────────────────────────────────────────┐
│ 跨 Agent 协作示例 │
│ │
│ 用户："我要退货订单 #12345，请帮我处理退款" │
│ │
│ Step 1: 客服 Agent 接收请求 │
│ Step 2: 客服 Agent 发现财务 Agent（读取 Agent Card） │
│ Step 3: 客服 Agent 委托财务 Agent 处理退款 │
│ Step 4: 财务 Agent 执行退款，返回结果 │
│ Step 5: 客服 Agent 综合结果，回复用户 │
└──────────────────────────────────────────────────────────────┘
5.2 客服 Agent 代码（A2A Client）
python
#!/usr/bin/env python3
"""
客服 Agent —— A2A Client 示例
发现并委托财务 Agent 处理退款
作者：PySuper | 来源：zhengxingtao.com
"""
import json
import uuid
import httpx
class A2AClient:
"""A2A 协议客户端"""
def __init__(self, base_url: str):
self.base_url = base_url
self.client = httpx.Client(timeout=30)
async def discover(self) -> dict:
"""发现远程 Agent 的能力"""
response = self.client.get(
f"{self.base_url}/.well-known/agent-card.json"
)
return response.json()
def send_message(self, message: str, task_id: str = None) -> dict:
"""发送消息给远程 Agent"""
if not task_id:
task_id = str(uuid.uuid4())
payload = {
"jsonrpc": "2.0",
"id": 1,
"method": "message/send",
"params": {
"taskId": task_id,
"message": {
"role": "user",
"parts": [{"type": "text", "text": message}]
}
}
}
response = self.client.post(
self.base_url,
json=payload,
headers={"Content-Type": "application/json"}
)
return response.json()
def handle_refund_request(order_id: str, amount: float):
"""处理退货退款请求——客服 Agent 逻辑"""
print(f"[客服 Agent] 收到退货退款请求：订单 {order_id}，金额 ${amount}")
# Step 1: 发现财务 Agent
print("[客服 Agent] 正在发现财务 Agent...")
finance_client = A2AClient("http://localhost:8001")
try:
agent_card = finance_client.discover()
print(f"[客服 Agent] 发现财务 Agent: {agent_card['name']}")
print(f"[客服 Agent] 可用技能: {[s['name'] for s in agent_card.get('skills', [])]}")
except Exception as e:
print(f"[客服 Agent] 无法发现财务 Agent: {e}")
return "抱歉，暂时无法连接财务系统，请稍后重试"
# Step 2: 检查财务 Agent 是否支持退款
refund_skill = None
for skill in agent_card.get("skills", []):
if "refund" in skill.get("id", "").lower():
refund_skill = skill
break
if not refund_skill:
return "抱歉，财务系统暂不支持自动退款"
# Step 3: 委托财务 Agent 处理退款
print("[客服 Agent] 正在委托财务 Agent 处理退款...")
refund_message = (
f"Process refund for order #{order_id}. "
f"Amount: ${amount}. Reason: Customer return."
)
result = finance_client.send_message(refund_message)
# Step 4: 处理结果
if "result" in result:
task = result["result"]
status = task.get("status", {}).get("state", "unknown")
if status == "completed":
# 提取结果文本
artifacts = task.get("artifacts", [])
if artifacts:
result_text = artifacts[0].get("parts", [{}])[0].get("text", "")
print(f"[客服 Agent] 退款处理成功: {result_text}")
return f"✅ 退款已处理完成。\n{result_text}"
elif status == "failed":
return "❌ 退款处理失败，请联系人工客服"
elif status == "input-required":
return "⚠️ 退款处理需要更多信息，请提供额外信息"
return "退款请求已提交，等待处理中"
if __name__ == "__main__":
# 模拟退款请求
result = handle_refund_request("12345", 299.99)
print(f"\n最终回复: {result}")
6. 与 MCP 的关系：垂直 vs 水平
plaintext
┌──────────────────────────────────────────────────────────────┐
│ MCP vs A2A：互补而非竞争 │
│ │
│ ┌────────────────────────┐ │
│ │ A2A（水平层） │ │
│ │ Agent ←→ Agent │ │
│ │ 发现、通信、协作 │ │
│ └───────────┬────────────┘ │
│ │ │
│ ┌───────────────┼───────────────┐ │
│ │ │ │ │
│ ┌─────▼─────┐ ┌─────▼─────┐ ┌─────▼─────┐ │
│ │ Agent A │ │ Agent B │ │ Agent C │ │
│ │ │ │ │ │ │ │
│ │ MCP ↓ │ │ MCP ↓ │ │ MCP ↓ │ │
│ └─────┬─────┘ └─────┬─────┘ └─────┬─────┘ │
│ │ │ │ │
│ ┌───────▼───────┐┌──────▼──────┐┌───────▼───────┐ │
│ │ MCP Server ││ MCP Server ││ MCP Server │ │
│ │ (GitHub) ││ (数据库) ││ (浏览器) │ │
│ └───────────────┘└─────────────┘└───────────────┘ │
│ │
│ MCP（垂直层）：Agent → 工具/数据 │
│ A2A（水平层）：Agent → Agent │
│ │
│ 类比： │
│ MCP = USB-C（设备连外设） │
│ A2A = WiFi（设备连设备） │
└──────────────────────────────────────────────────────────────┘
详细对比
表格
7. 与 Skill 的关系：封装能力 vs 发现调度能力
plaintext
┌──────────────────────────────────────────────────────────────┐
│ Skill + MCP + A2A 的三位一体关系 │
│ │
│ Skill（封装能力） │
│ "知道做什么" —— 流程、规则、最佳实践 │
│ → Agent 内部的能力模块 │
│ │
│ MCP（连接工具） │
│ "知道怎么连" —— 标准化协议、发现、调用 │
│ → Agent 向下连接外部工具的通道 │
│ │
│ A2A（发现调度） │
│ "知道找谁做" —— Agent 发现、通信、协作 │
│ → Agent 之间水平互联的协议 │
│ │
│ 工作流示例： │
│ 1. 用户请求到达 Agent A │
│ 2. Agent A 的 Skill 判断需要财务处理 │
│ 3. Agent A 通过 A2A 发现财务 Agent B │
│ 4. Agent A 通过 A2A 委托 Agent B │
│ 5. Agent B 通过 MCP 调用财务数据库 │
│ 6. Agent B 通过 A2A 返回结果给 Agent A │
│ 7. Agent A 综合结果回复用户 │
└──────────────────────────────────────────────────────────────┘
8. Google ADK 集成
Google Agent Development Kit (ADK) 1.0 原生支持 A2A：
python
# 使用 Google ADK 创建 A2A 兼容的 Agent
from google.adk import Agent, A2AServer
# 定义 Agent
agent = Agent(
name="customer-support",
description="Customer support agent for order management",
model="gemini-2.5-pro"
)
# 添加 A2A 能力
agent.add_a2a_skill(
skill_id="process-return",
name="Process Return",
description="Handle customer return requests"
)
# 启动 A2A Server
a2a_server = A2AServer(agent, host="0.0.0.0", port=8000)
a2a_server.start()
9. 生产部署：Agent 发现、身份认证、跨云联邦
9.1 Agent 发现
plaintext
┌──────────────────────────────────────────────────────────────┐
│ Agent 发现机制 │
│ │
│ 方式1：Well-Known URI（去中心化） │
│ ┌─────────────────────────────────────────────┐ │
│ │ GET https://agent.example.com/ │ │
│ │ .well-known/agent-card.json │ │
│ └─────────────────────────────────────────────┘ │
│ → 任何 Agent 只要知道域名就能发现 │
│ → 适合公开 Agent │
│ │
│ 方式2：Agent Registry（集中式） │
│ ┌─────────────────────────────────────────────┐ │
│ │ 企业内部 Agent 注册中心 │ │
│ │ - 财务 Agent → https://finance.internal/a2a │ │
│ │ - HR Agent → https://hr.internal/a2a │ │
│ │ - 运维 Agent → https://ops.internal/a2a │ │
│ └─────────────────────────────────────────────┘ │
│ → 适合企业内部 Agent │
│ │
│ 方式3：Signed Agent Card（v1.0 新增） │
│ ┌─────────────────────────────────────────────┐ │
│ │ Agent Card + 数字签名 │ │
│ │ → 接收方可验证 Card 是否由域名所有者签发 │ │
│ │ → 防止 Agent Card 伪造攻击 │ │
│ └─────────────────────────────────────────────┘ │
│ → 适合跨组织联邦 │
└──────────────────────────────────────────────────────────────┘
9.2 身份认证
A2A v1.0 支持多种认证方式：
python
# OAuth2 认证配置示例
AUTH_CONFIG = {
"schemes": [
{
"type": "oauth2",
"flows": {
"clientCredentials": {
"tokenUrl": "https://auth.example.com/oauth/token",
"scopes": {
"a2a:read": "Read agent capabilities",
"a2a:write": "Submit tasks to agent"
}
}
}
},
{
"type": "mutualTLS",
"description": "mTLS for internal service-to-service communication"
}
]
}
9.3 跨云联邦
plaintext
┌──────────────────────────────────────────────────────────────┐
│ 跨云 Agent 联邦 │
│ │
│ ┌──────────────────────┐ ┌──────────────────────┐ │
│ │ AWS 环境 │ │ Azure 环境 │ │
│ │ │ │ │ │
│ │ ┌──────────────┐ │ │ ┌──────────────┐ │ │
│ │ │ Bedrock │ │ │ │ AI Foundry │ │ │
│ │ │ AgentCore │ │ │ │ Agent │ │ │
│ │ │ │ │ │ │ │ │ │
│ │ │ Agent A ─────┼────┼────┼──│ Agent C │ │ │
│ │ └──────────────┘ │ │ └──────────────┘ │ │
│ │ │A2A│ │ │
│ │ ┌──────────────┐ │ │ ┌──────────────┐ │ │
│ │ │ Agent B │ │ │ │ Agent D │ │ │
│ │ └──────────────┘ │ │ └──────────────┘ │ │
│ └──────────────────────┘ └──────────────────────┘ │
│ │
│ A2A 的跨云能力： │
│ - Agent Card 通过 HTTPS 跨云发现 │
│ - OAuth2 支持跨组织认证 │
│ - Signed Agent Card 防止联邦中的身份伪造 │
└──────────────────────────────────────────────────────────────┘
10. 踩坑记录
坑1：Agent Card 发现被防火墙阻断
问题：企业内网 Agent 无法访问外部 .well-known 端点。
解决：部署内部 Agent Registry，配置防火墙规则允许内网 A2A 通信。
坑2：长任务超时
问题：A2A Task 处理时间超过 HTTP 超时限制。
解决：使用 message/stream（SSE）模式，避免 HTTP 超时：
python
# 使用流式模式
payload = {
"jsonrpc": "2.0",
"id": 1,
"method": "message/stream", # 而非 message/send
"params": {
"message": {
"role": "user",
"parts": [{"type": "text", "text": "复杂任务..."}]
}
}
}
坑3：Agent Card 缺少 skills 字段
问题：Agent Card 中没有声明 skills，导致其他 Agent 无法判断你的能力。
解决：确保 Agent Card 包含完整的 skills 列表和 examples。
坑4：OAuth2 Token 过期导致长任务失败
问题：长时间运行的 Task 在执行过程中 Token 过期。
解决：实现 Token 自动刷新，或在 Task 生命周期内使用 refresh_token。
坑5：跨云 A2A 通信延迟过高
问题：AWS 和 Azure 之间的 A2A 通信延迟达到数百毫秒。
解决：
使用 gRPC 传输替代 JSON-RPC over HTTP
部署消息队列（如 Kafka）作为缓冲
对于高频交互，考虑将相关 Agent 部署在同一云区域
11. 总结与展望
A2A 的核心价值
A2A 的核心价值不是又一个协议，而是让 Agent 生态从"孤岛"变成"网络" 。当每个 Agent 都能被发现、被委托、被组合时，Agent 的价值将指数级增长。
2026 AI Agent 通信协议全景
plaintext
┌──────────────────────────────────────────────────────────────┐
│ 2026 AI Agent 通信协议全景 │
│ │
│ ┌────────────────────────────────────────────────────┐ │
│ │ A2A（Agent ↔ Agent，水平） │ │
│ │ Google 发起，Linux 基金会治理 │ │
│ │ 150+ 组织，22K+ Stars │ │
│ └──────────────────────┬─────────────────────────────┘ │
│ │ │
│ ┌──────────────────────┼─────────────────────────────┐ │
│ │ MCP（Agent ↔ Tool，垂直） │ │
│ │ Anthropic 发起，Linux 基金会治理 │ │
│ │ 97M 月下载，10K+ Servers │ │
│ └──────────────────────┼─────────────────────────────┘ │
│ │ │
│ ┌──────────────────────┼─────────────────────────────┐ │
│ │ Skill（Agent 内部能力，知识封装） │ │
│ │ Anthropic 发起，agentskills.io 开放标准 │ │
│ │ 30+ Agent 兼容，800K+ 生态 │ │
│ └──────────────────────┴─────────────────────────────┘ │
│ │
│ 三者互补，构成完整的 AI Agent 基础设施 │
└──────────────────────────────────────────────────────────────┘
未来趋势
AP2 扩展：A2A 上的支付协议，让 Agent 间可以自动结算
Agent 联邦：跨企业、跨云的 Agent 联邦将成为主流
与 MCP 深度集成：Agent 通过 MCP 连工具，通过 A2A 连其他 Agent——两者合一是必然方向
A2A 不是一个独立的协议，它是 AI Agent 网络化的基础设施。掌握 A2A，你就掌握了构建多 Agent 系统的标准语言。
本文作者：PySuper | 来源：zhengxingtao.com
参考资源：
A2A 官方规范：https://a2a-protocol.org
A2A GitHub：https://github.com/a2aproject/a2a
A2A Protocol 一周年报告：https://www.prnewswire.com/news-releases/a2a-protocol-surpasses-150-organizations
评论区