A2A（Agent2Agent）协议全面学习指南

📚 系统梳理 + 易理解 + 强记忆
A2A 是 Google 于 2025 年 4 月发布的首个标准智能体交互协议，被称为「AI 智能体的通用语言」

一、核心基础

1.1 A2A 协议定义

A2A（Agent2Agent Protocol） 是 Google 于 2025 年 4 月 10 日在 Google Cloud Next 2025 大会上发布的开源智能体交互协议。

一句话定义：A2A 是让不同 AI 智能体（Agent）之间能够相互发现、通信、协作的标准化协议。

┌─────────────────────────────────────────────────────────────────┐
│                    A2A 协议核心定位                              │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│    ┌─────────┐         A2A 协议          ┌─────────┐            │
│    │ Agent A │ ◄─────────────────────► │ Agent B │            │
│    │ (框架X) │      标准化通信           │ (框架Y) │            │
│    └─────────┘                          └─────────┘            │
│         │                                    │                  │
│         │              A2A 协议              │                  │
│         └──────────────────────────────────►│                  │
│                         │                    │                  │
│                    ┌─────────┐               │                  │
│                    │ Agent C │ ◄────────────┘                  │
│                    │ (框架Z) │                                  │
│                    └─────────┘                                  │
│                                                                  │
│  **重点**：跨框架、跨供应商、跨平台的智能体间通信                   │
└─────────────────────────────────────────────────────────────────┘

1.2 设计初衷与背景

痛点	A2A 解决方案
智能体孤岛	不同框架的 Agent 无法通信 → 提供统一协议标准
重复造轮子	每次集成都需自定义接口 → 一次实现，处处可用
协作复杂	多 Agent 协同难以实现 → 原生支持任务编排
生态割裂	各厂商封闭生态 → 开源开放，50+ 企业支持

设计原则（官方定义）：

拥抱智能体能力：支持自然语言协商、任务代理、多模态交互
基于现有标准：使用 HTTP、JSON-RPC、SSE 等成熟技术
默认安全：内置企业级认证授权机制
支持长任务：原生处理耗时任务，支持实时进度反馈
模态无关：支持文本、文件、表单、流式数据等多种内容类型

1.3 核心定位：A2A vs MCP

⚠️ 易混淆点：A2A 和 MCP 不是竞争关系，而是互补关系！

┌─────────────────────────────────────────────────────────────────┐
│                    协议定位对比                                   │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│  MCP (Model Context Protocol)                                    │
│  ┌─────────┐                      ┌─────────────────────────┐   │
│  │   LLM   │ ◄─── MCP 协议 ────► │  工具/资源/数据源        │   │
│  └─────────┘                      │  (Tool, Resource, API)  │   │
│                                   └─────────────────────────┘   │
│  定位：LLM 与外部工具的连接（人→工具）                            │
│                                                                  │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│  A2A (Agent2Agent Protocol)                                      │
│  ┌─────────┐                      ┌─────────┐                   │
│  │ Agent A │ ◄─── A2A 协议 ────► │ Agent B │                   │
│  └─────────┘                      └─────────┘                   │
│  定位：智能体之间的协作（Agent→Agent）                            │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘

对比维度	MCP	A2A
发布方	Anthropic (2024.11)	Google (2025.04)
核心定位	LLM ↔ 工具/资源	Agent ↔ Agent
通信对象	模型与外部系统	智能体与智能体
主要能力	工具调用、资源获取	任务编排、能力协商
任务模型	同步为主	支持长时异步任务
发现机制	无	Agent Card 能力发现
关系	垂直连接（深度）	水平连接（广度）

⭐ 重点理解：

MCP：让 Agent 能「使用工具」（类似给机器人装上手臂）
A2A：让 Agent 能「协作对话」（类似让机器人之间交流）

1.4 与其他协议的对比

协议	定位	通信模式	适用场景
A2A	Agent 间通信	任务驱动、异步	多智能体协作
MCP	LLM-工具连接	请求-响应	工具调用
Function Calling	模型调用函数	同步调用	单次工具调用
gRPC	通用 RPC	请求-响应/流	微服务通信
MQTT	消息队列	发布-订阅	IoT、消息推送
HTTP REST	Web API	请求-响应	Web 服务

1.5 适用场景与不适用场景

✅ 适用场景

场景	说明
多智能体协作系统	多个专业 Agent 协同完成复杂任务
跨平台 Agent 集成	不同框架/供应商的 Agent 需要通信
企业级 AI 工作流	需要任务编排、状态管理的业务流程
AI 助手联邦	多个 AI 助手共享能力、分工协作
长时任务处理	需要异步执行、进度反馈的任务

❌ 不适用场景

场景	原因	替代方案
简单工具调用	杀鸡用牛刀	Function Calling
LLM 直连外部数据	不是 A2A 设计目标	MCP
实时流媒体传输	非核心优化方向	WebRTC、gRPC
IoT 设备通信	协议过重	MQTT、CoAP
单体应用内部通信	无需标准化	直接函数调用

二、技术规范

2.1 核心架构

核心组件说明：

组件	职责	类比
A2A Client	发起请求的一方（可以是另一个 Agent）	客户
A2A Server	接收请求、执行任务的 Agent	服务商
Agent Card	描述 Agent 能力的元数据文件	名片/简历
Task	工作的基本单位，有完整生命周期	工单
Message	Client 与 Server 之间的通信内容	对话消息
Artifact	任务执行产生的输出物	工作成果

2.2 核心概念详解

2.2.1 Agent Card（智能体名片）

⭐ 重点：Agent Card 是 A2A 的发现机制核心，类似「AI 的数字简历」

硬性要求：必须部署在 /.well-known/agent.json 路径

json

{
  "name": "智能文档助手",
  "description": "专业的文档处理和分析智能体",
  "url": "https://doc-agent.example.com/a2a",
  "version": "1.0.0",
  
  "capabilities": {
    "streaming": true,
    "pushNotifications": true,
    "stateTransitionHistory": true
  },
  
  "authentication": {
    "schemes": ["Bearer"],
    "credentials": "oauth2"
  },
  
  "defaultInputModes": ["text/plain", "application/json"],
  "defaultOutputModes": ["text/plain", "application/json", "application/pdf"],
  
  "skills": [
    {
      "id": "document_analysis",
      "name": "文档分析",
      "description": "分析文档内容、提取关键信息",
      "inputModes": ["application/pdf", "text/plain"],
      "outputModes": ["application/json"]
    },
    {
      "id": "document_translation",
      "name": "文档翻译",
      "description": "将文档翻译成目标语言",
      "inputModes": ["text/plain"],
      "outputModes": ["text/plain"]
    }
  ]
}

字段说明：

字段	必选	说明
`name`	✅	Agent 名称
`description`	✅	Agent 功能描述
`url`	✅	A2A 服务端点 URL
`version`	✅	版本号
`capabilities`	❌	支持的能力标志
`authentication`	❌	认证方式
`skills`	✅	技能列表（核心！）
`defaultInputModes`	❌	默认输入格式
`defaultOutputModes`	❌	默认输出格式

2.2.2 Task（任务）

⭐ 重点：Task 是 A2A 的工作基本单位，有完整的生命周期

任务状态流转：

任务状态说明：

状态	说明	是否终态
`submitted`	已提交，等待处理	❌
`working`	正在处理中	❌
`input_required`	需要客户端提供更多输入	❌
`completed`	任务成功完成	✅
`failed`	任务执行失败	✅
`canceled`	任务被取消	✅

Task 数据结构：

json

{
  "id": "task-uuid-12345",
  "sessionId": "session-uuid-67890",
  "status": {
    "state": "working",
    "message": {
      "role": "agent",
      "parts": [
        {
          "type": "text",
          "text": "正在分析文档..."
        }
      ]
    },
    "timestamp": "2025-04-10T10:30:00Z"
  },
  "artifacts": [],
  "history": []
}

2.2.3 Message（消息）

Message 是 Client 和 Agent 之间的通信单元：

json

{
  "role": "user",  // "user" 或 "agent"
  "parts": [
    {
      "type": "text",
      "text": "请帮我分析这份文档"
    },
    {
      "type": "file",
      "file": {
        "name": "report.pdf",
        "mimeType": "application/pdf",
        "bytes": "base64编码内容..."
      }
    }
  ]
}

2.2.4 Part（内容单元）

Part 类型	说明	示例
TextPart	文本内容	普通文字、Markdown
FilePart	文件内容（内联或 URI）	PDF、图片、音视频
DataPart	结构化数据	JSON 表单、配置

json

// TextPart
{ "type": "text", "text": "这是文本内容" }

// FilePart (内联)
{ 
  "type": "file", 
  "file": { 
    "name": "image.png", 
    "mimeType": "image/png", 
    "bytes": "base64..." 
  } 
}

// FilePart (URI)
{ 
  "type": "file", 
  "file": { 
    "name": "video.mp4", 
    "mimeType": "video/mp4", 
    "uri": "https://example.com/video.mp4" 
  } 
}

// DataPart
{ 
  "type": "data", 
  "data": { 
    "formField1": "value1", 
    "formField2": 123 
  } 
}

2.2.5 Artifact（工件/产出物）

Agent 执行任务后产生的输出：

json

{
  "name": "分析报告",
  "description": "文档分析结果",
  "parts": [
    {
      "type": "text",
      "text": "## 分析摘要\n..."
    },
    {
      "type": "file",
      "file": {
        "name": "analysis.json",
        "mimeType": "application/json",
        "bytes": "eyJyZXN1bHQiOi..."
      }
    }
  ],
  "index": 0,
  "append": false,
  "lastChunk": true
}

2.3 通信模型

2.3.1 通信流程

2.3.2 请求/响应格式（JSON-RPC 2.0）

请求格式：

json

{
  "jsonrpc": "2.0",
  "id": "req-001",
  "method": "tasks/send",
  "params": {
    "id": "task-uuid-12345",
    "message": {
      "role": "user",
      "parts": [
        { "type": "text", "text": "分析这个文档" }
      ]
    }
  }
}

响应格式（成功）：

json

{
  "jsonrpc": "2.0",
  "id": "req-001",
  "result": {
    "id": "task-uuid-12345",
    "sessionId": "session-uuid-67890",
    "status": {
      "state": "working",
      "timestamp": "2025-04-10T10:30:00Z"
    }
  }
}

响应格式（错误）：

json

{
  "jsonrpc": "2.0",
  "id": "req-001",
  "error": {
    "code": -32600,
    "message": "Invalid Request",
    "data": { "details": "Missing required field: message" }
  }
}

2.4 核心方法（Methods）

方法	说明	请求方向
`tasks/send`	发送任务消息（同步）	Client → Server
`tasks/sendSubscribe`	发送任务并订阅更新（流式）	Client → Server
`tasks/get`	获取任务状态	Client → Server
`tasks/cancel`	取消任务	Client → Server
`tasks/pushNotification/set`	设置推送通知	Client → Server
`tasks/pushNotification/get`	获取推送配置	Client → Server
`tasks/resubscribe`	重新订阅任务更新	Client → Server

方法详解：

python

# tasks/send - 发送任务（最常用）
{
    "method": "tasks/send",
    "params": {
        "id": "task-123",           # 任务ID（可选，不填则自动生成）
        "sessionId": "session-456", # 会话ID（可选，用于关联多轮对话）
        "message": {                # 消息内容
            "role": "user",
            "parts": [...]
        },
        "acceptedOutputModes": ["text/plain", "application/json"],  # 可接受的输出格式
        "pushNotification": {       # 推送配置（可选）
            "url": "https://my-app.com/webhook",
            "authentication": {...}
        }
    }
}

# tasks/sendSubscribe - 流式任务（SSE）
# 请求格式同 tasks/send，响应为 SSE 流

2.5 传输方式

传输方式	说明	使用场景
HTTP POST	标准请求-响应	短任务、同步调用
HTTP + SSE	服务器推送事件	长任务、流式输出
Webhook	推送通知	异步回调、状态更新

HTTP 端点要求：

POST {agent_url}           # JSON-RPC 请求端点
GET  {agent_url}/sse       # SSE 订阅端点（用于 sendSubscribe）
GET  /.well-known/agent.json  # Agent Card 发现端点

2.6 认证与安全机制

硬性要求：企业级部署必须配置认证！

支持的认证方式：

认证方式	说明	配置示例
API Key	简单密钥认证	`Authorization: Bearer xxx`
OAuth 2.0	标准 OAuth 流程	支持多种 Grant Type
mTLS	双向 TLS 证书	企业内网推荐
自定义	自定义认证头	根据业务需求

Agent Card 中的认证配置：

json

{
  "authentication": {
    "schemes": ["Bearer", "OAuth2"],
    "credentials": "oauth2",
    "oauth2": {
      "authorizationUrl": "https://auth.example.com/authorize",
      "tokenUrl": "https://auth.example.com/token",
      "scopes": {
        "read": "读取权限",
        "write": "写入权限"
      }
    }
  }
}

2.7 错误处理规范

标准错误码（基于 JSON-RPC 2.0）：

错误码	名称	说明
`-32700`	Parse Error	JSON 解析错误
`-32600`	Invalid Request	无效请求
`-32601`	Method Not Found	方法不存在
`-32602`	Invalid Params	参数错误
`-32603`	Internal Error	内部错误

A2A 扩展错误码：

错误码	名称	说明
`-32001`	Task Not Found	任务不存在
`-32002`	Task Canceled	任务已取消
`-32003`	Authentication Failed	认证失败
`-32004`	Permission Denied	权限不足
`-32005`	Rate Limited	请求限流

三、核心能力

3.1 智能体发现（Agent Discovery）

⭐ 重点：通过 Agent Card 实现即插即用的能力发现

发现流程代码示例：

python

import httpx

async def discover_agent(base_url: str):
    """发现并解析 Agent 能力"""
    agent_card_url = f"{base_url}/.well-known/agent.json"
    
    async with httpx.AsyncClient() as client:
        response = await client.get(agent_card_url)
        agent_card = response.json()
    
    print(f"Agent: {agent_card['name']}")
    print(f"描述: {agent_card['description']}")
    print(f"技能列表:")
    for skill in agent_card.get('skills', []):
        print(f"  - {skill['name']}: {skill['description']}")
    
    return agent_card

3.2 任务编排（Task Orchestration）

A2A 原生支持复杂任务的编排和协调：

┌─────────────────────────────────────────────────────────────────┐
│                    多 Agent 任务编排示例                          │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│  ┌───────────────┐                                              │
│  │   用户请求     │ "帮我写一篇技术博客并翻译成英文"              │
│  └───────┬───────┘                                              │
│          │                                                       │
│          ▼                                                       │
│  ┌───────────────┐      A2A       ┌───────────────┐             │
│  │ 编排 Agent    │ ─────────────► │ 写作 Agent    │             │
│  │ (Orchestrator)│                │               │             │
│  └───────┬───────┘                └───────┬───────┘             │
│          │                                │                      │
│          │                                │ 返回中文文章         │
│          │                                ▼                      │
│          │         A2A            ┌───────────────┐             │
│          └──────────────────────► │ 翻译 Agent    │             │
│                                   │               │             │
│                                   └───────┬───────┘             │
│                                           │                      │
│                                           │ 返回英文翻译         │
│                                           ▼                      │
│                                   ┌───────────────┐             │
│                                   │   最终结果     │             │
│                                   └───────────────┘             │
└─────────────────────────────────────────────────────────────────┘

3.3 多智能体协同

协同模式：

模式	说明	适用场景
串行协作	Agent A → Agent B → Agent C	流水线任务
并行协作	Agent A ⇉ Agent B, C, D	分解子任务
竞争协作	多 Agent 竞标同一任务	最优选择
投票协作	多 Agent 共同决策	质量保证

3.4 消息路由

A2A 支持灵活的消息路由机制：

json

{
  "method": "tasks/send",
  "params": {
    "id": "task-123",
    "message": {
      "role": "user",
      "parts": [{ "type": "text", "text": "处理请求" }]
    },
    "routing": {
      "preferredAgent": "agent-uuid-456",  // 优先路由
      "fallbackAgents": ["agent-uuid-789"], // 备选路由
      "loadBalancing": "round-robin"        // 负载均衡
    }
  }
}

3.5 关键特性总结

特性	说明	对比优势
跨平台	支持任何框架、任何语言	vs 私有协议
低耦合	Agent 之间松散耦合	vs 紧耦合集成
可扩展	自定义 Skill、Part 类型	vs 固定能力
多模态	支持文本、文件、流媒体	vs 纯文本
长任务	原生异步、进度反馈	vs 同步阻塞
安全性	内置认证、授权机制	vs 无安全考虑

四、应用实践

4.1 典型落地场景

场景一：多智能体协作系统

┌─────────────────────────────────────────────────────────────────┐
│                    企业智能助手联邦                              │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│  用户: "帮我处理本月的报销单，并生成财务报表"                      │
│                                                                  │
│  ┌─────────────┐    A2A    ┌─────────────┐                      │
│  │ 入口 Agent  │ ────────► │ 文档 Agent  │  提取报销单信息       │
│  └─────────────┘           └──────┬──────┘                      │
│         │                         │                              │
│         │                         ▼                              │
│         │          A2A     ┌─────────────┐                      │
│         └────────────────► │ 财务 Agent  │  生成财务报表        │
│                            └──────┬──────┘                      │
│                                   │                              │
│                                   ▼                              │
│                            ┌─────────────┐                      │
│                            │ 审批 Agent  │  自动提交审批        │
│                            └─────────────┘                      │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘

场景二：AI 工作流

python

# 示例：代码审查工作流
async def code_review_workflow(code_diff: str):
    # 1. 静态分析 Agent
    static_result = await call_agent(
        agent_url="https://static-analyzer.example.com",
        skill="code_analysis",
        input=code_diff
    )
    
    # 2. 安全扫描 Agent（并行）
    security_result = await call_agent(
        agent_url="https://security-scanner.example.com",
        skill="vulnerability_scan",
        input=code_diff
    )
    
    # 3. 代码审查 Agent
    review_result = await call_agent(
        agent_url="https://code-reviewer.example.com",
        skill="review",
        input={
            "code": code_diff,
            "static_analysis": static_result,
            "security_scan": security_result
        }
    )
    
    return review_result

场景三：分布式智能体平台

┌─────────────────────────────────────────────────────────────────┐
│                    分布式 Agent 平台架构                          │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│  ┌─────────────────────────────────────────────────────────┐    │
│  │                    Agent Registry                        │    │
│  │                  (Agent Card 注册中心)                    │    │
│  └────────────────────────┬────────────────────────────────┘    │
│                           │                                      │
│         ┌─────────────────┼─────────────────┐                   │
│         │                 │                 │                    │
│         ▼                 ▼                 ▼                    │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐             │
│  │ Agent Pod 1 │  │ Agent Pod 2 │  │ Agent Pod N │             │
│  │ (K8s 部署)  │  │ (K8s 部署)  │  │ (K8s 部署)  │             │
│  └─────────────┘  └─────────────┘  └─────────────┘             │
│         │                 │                 │                    │
│         └─────────────────┴─────────────────┘                   │
│                           │                                      │
│                           ▼                                      │
│  ┌─────────────────────────────────────────────────────────┐    │
│  │                    A2A Gateway                           │    │
│  │            (负载均衡、认证、路由、监控)                    │    │
│  └─────────────────────────────────────────────────────────┘    │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘

4.2 主流适配框架/工具

框架/工具	支持程度	说明
LangChain	✅ 官方支持	首批支持企业之一
AutoGPT	✅ 社区支持	多智能体协作
CrewAI	✅ 原生支持	团队协作框架
Google ADK	✅ 官方 SDK	Agent Development Kit
Vertex AI	✅ 深度集成	Google Cloud 产品
Salesforce	✅ 企业支持	Einstein Agent
SAP	✅ 企业支持	Joule Agent

4.3 快速上手实践

Step 1：创建 Agent Card

json

// /.well-known/agent.json
{
  "name": "我的第一个 A2A Agent",
  "description": "一个简单的 Echo Agent，用于学习 A2A",
  "url": "http://localhost:8000/a2a",
  "version": "1.0.0",
  "capabilities": {
    "streaming": false,
    "pushNotifications": false
  },
  "skills": [
    {
      "id": "echo",
      "name": "Echo 回显",
      "description": "将输入内容原样返回",
      "inputModes": ["text/plain"],
      "outputModes": ["text/plain"]
    }
  ]
}

Step 2：实现 A2A Server（Python）

python

# a2a_server.py
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
import uuid

app = FastAPI()

# 存储任务
tasks = {}

@app.get("/.well-known/agent.json")
async def get_agent_card():
    """返回 Agent Card"""
    return {
        "name": "Echo Agent",
        "description": "简单的回显 Agent",
        "url": "http://localhost:8000/a2a",
        "version": "1.0.0",
        "capabilities": {"streaming": False},
        "skills": [{
            "id": "echo",
            "name": "Echo",
            "description": "回显输入内容",
            "inputModes": ["text/plain"],
            "outputModes": ["text/plain"]
        }]
    }

@app.post("/a2a")
async def handle_a2a(request: Request):
    """处理 A2A JSON-RPC 请求"""
    body = await request.json()
    
    method = body.get("method")
    params = body.get("params", {})
    request_id = body.get("id")
    
    if method == "tasks/send":
        return handle_task_send(params, request_id)
    elif method == "tasks/get":
        return handle_task_get(params, request_id)
    elif method == "tasks/cancel":
        return handle_task_cancel(params, request_id)
    else:
        return JSONResponse({
            "jsonrpc": "2.0",
            "id": request_id,
            "error": {
                "code": -32601,
                "message": f"Method not found: {method}"
            }
        })

def handle_task_send(params: dict, request_id: str):
    """处理 tasks/send 请求"""
    task_id = params.get("id") or str(uuid.uuid4())
    message = params.get("message", {})
    
    # 提取输入文本
    input_text = ""
    for part in message.get("parts", []):
        if part.get("type") == "text":
            input_text += part.get("text", "")
    
    # 创建任务并立即完成（Echo 逻辑）
    task = {
        "id": task_id,
        "sessionId": params.get("sessionId") or str(uuid.uuid4()),
        "status": {
            "state": "completed",
            "message": {
                "role": "agent",
                "parts": [{"type": "text", "text": f"Echo: {input_text}"}]
            }
        },
        "artifacts": [{
            "name": "echo_result",
            "parts": [{"type": "text", "text": input_text}]
        }]
    }
    
    tasks[task_id] = task
    
    return JSONResponse({
        "jsonrpc": "2.0",
        "id": request_id,
        "result": task
    })

def handle_task_get(params: dict, request_id: str):
    """处理 tasks/get 请求"""
    task_id = params.get("id")
    task = tasks.get(task_id)
    
    if not task:
        return JSONResponse({
            "jsonrpc": "2.0",
            "id": request_id,
            "error": {
                "code": -32001,
                "message": "Task not found"
            }
        })
    
    return JSONResponse({
        "jsonrpc": "2.0",
        "id": request_id,
        "result": task
    })

def handle_task_cancel(params: dict, request_id: str):
    """处理 tasks/cancel 请求"""
    task_id = params.get("id")
    task = tasks.get(task_id)
    
    if task:
        task["status"]["state"] = "canceled"
    
    return JSONResponse({
        "jsonrpc": "2.0",
        "id": request_id,
        "result": task or {"id": task_id, "status": {"state": "canceled"}}
    })

# 运行：uvicorn a2a_server:app --host 0.0.0.0 --port 8000

Step 3：实现 A2A Client

python

# a2a_client.py
import httpx
import asyncio

async def discover_agent(base_url: str) -> dict:
    """发现 Agent 能力"""
    async with httpx.AsyncClient() as client:
        response = await client.get(f"{base_url}/.well-known/agent.json")
        return response.json()

async def send_task(agent_url: str, message: str) -> dict:
    """发送任务"""
    payload = {
        "jsonrpc": "2.0",
        "id": "req-001",
        "method": "tasks/send",
        "params": {
            "message": {
                "role": "user",
                "parts": [{"type": "text", "text": message}]
            }
        }
    }
    
    async with httpx.AsyncClient() as client:
        response = await client.post(agent_url, json=payload)
        return response.json()

async def main():
    base_url = "http://localhost:8000"
    
    # 1. 发现 Agent
    print("=== 发现 Agent ===")
    agent_card = await discover_agent(base_url)
    print(f"Agent: {agent_card['name']}")
    print(f"Skills: {[s['name'] for s in agent_card['skills']]}")
    
    # 2. 发送任务
    print("\n=== 发送任务 ===")
    result = await send_task(
        f"{base_url}/a2a",
        "Hello, A2A!"
    )
    print(f"Result: {result}")

if __name__ == "__main__":
    asyncio.run(main())

Step 4：运行测试

bash

# 终端 1：启动 Server
pip install fastapi uvicorn httpx
uvicorn a2a_server:app --host 0.0.0.0 --port 8000

# 终端 2：运行 Client
python a2a_client.py

预期输出：

=== 发现 Agent ===
Agent: Echo Agent
Skills: ['Echo']

=== 发送任务 ===
Result: {
  "jsonrpc": "2.0",
  "id": "req-001",
  "result": {
    "id": "xxx-xxx-xxx",
    "status": {"state": "completed", ...},
    "artifacts": [{"name": "echo_result", ...}]
  }
}

五、进阶拓展

5.1 版本迭代与更新

版本	发布时间	主要更新
1.0.0	2025.04	首次发布，核心协议定义
1.x	规划中	增强流式支持、多模态优化

关注官方更新：

GitHub: google/a2a-protocol（假设仓库名）
文档: https://a2a-protocol.dev（假设官网）

5.2 性能优化技巧

优化点	方法	效果
连接复用	HTTP/2 多路复用	减少连接开销
批量请求	合并多个小任务	降低网络延迟
流式响应	使用 SSE 订阅	提升响应体验
缓存 Agent Card	本地缓存 + TTL	减少发现开销
异步处理	使用 Webhook 回调	释放客户端资源

示例：连接池配置：

python

import httpx

# 创建带连接池的客户端
client = httpx.AsyncClient(
    http2=True,  # 启用 HTTP/2
    limits=httpx.Limits(
        max_keepalive_connections=20,
        max_connections=100
    ),
    timeout=httpx.Timeout(30.0, connect=10.0)
)

5.3 私有化部署要点

5.3.1 部署架构

┌─────────────────────────────────────────────────────────────────┐
│                    私有化部署架构                                │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│  ┌─────────────────────────────────────────────────────────┐    │
│  │                    负载均衡器 (Nginx/Kong)               │    │
│  └────────────────────────────┬────────────────────────────┘    │
│                               │                                  │
│         ┌─────────────────────┼─────────────────────┐           │
│         │                     │                     │            │
│         ▼                     ▼                     ▼            │
│  ┌─────────────┐      ┌─────────────┐      ┌─────────────┐      │
│  │ Agent Pod 1 │      │ Agent Pod 2 │      │ Agent Pod 3 │      │
│  └─────────────┘      └─────────────┘      └─────────────┘      │
│                                                                  │
│  ┌─────────────────────────────────────────────────────────┐    │
│  │                    共享存储层                            │    │
│  │         (Redis 缓存 + PostgreSQL 持久化)                 │    │
│  └─────────────────────────────────────────────────────────┘    │
│                                                                  │
│  ┌─────────────────────────────────────────────────────────┐    │
│  │                    监控告警 (Prometheus + Grafana)       │    │
│  └─────────────────────────────────────────────────────────┘    │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘

5.3.2 关键配置

Nginx 代理配置：

nginx

upstream a2a_agents {
    least_conn;
    server agent1:8000;
    server agent2:8000;
    server agent3:8000;
}

server {
    listen 443 ssl http2;
    server_name a2a.example.com;
    
    # Agent Card 发现
    location /.well-known/agent.json {
        proxy_pass http://a2a_agents;
        proxy_cache agent_card_cache;
        proxy_cache_valid 200 5m;
    }
    
    # A2A 端点
    location /a2a {
        proxy_pass http://a2a_agents;
        proxy_http_version 1.1;
        proxy_set_header Connection '';
        proxy_buffering off;  # SSE 支持
    }
}

5.3.3 安全加固

措施	说明
mTLS	双向 TLS 认证
网络隔离	VPC 内网部署
访问控制	RBAC 权限管理
审计日志	记录所有 A2A 调用
速率限制	防止滥用

5.4 生态与周边工具链

┌─────────────────────────────────────────────────────────────────┐
│                    A2A 生态工具链                                │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│  开发工具                                                        │
│  ├── Google Agent Development Kit (ADK)                         │
│  ├── A2A SDK (Python/TypeScript/Go)                             │
│  └── A2A CLI (命令行工具)                                        │
│                                                                  │
│  测试工具                                                        │
│  ├── A2A Inspector (可视化调试)                                  │
│  ├── Mock Agent Server                                          │
│  └── 集成测试框架                                                │
│                                                                  │
│  运维工具                                                        │
│  ├── Agent Registry (注册中心)                                   │
│  ├── A2A Gateway (网关)                                         │
│  └── 监控面板 (Grafana Dashboard)                                │
│                                                                  │
│  集成框架                                                        │
│  ├── LangChain A2A Adapter                                      │
│  ├── CrewAI A2A Plugin                                          │
│  └── Vertex AI Agent Builder                                    │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘

核心知识点速记清单

📋 30 秒速记

A2A = Agent 之间的标准化通信协议（Google 出品）

核心组件：
  - Agent Card = AI 的数字名片 (/.well-known/agent.json)
  - Task = 工作基本单位（6 种状态）
  - Message = 通信内容（包含多个 Part）
  - Artifact = 任务产出物

关键方法：
  - tasks/send = 发送任务
  - tasks/sendSubscribe = 流式任务
  - tasks/get = 查询状态
  - tasks/cancel = 取消任务

vs MCP：
  - MCP = LLM ↔ 工具（垂直连接）
  - A2A = Agent ↔ Agent（水平连接）
  - 两者互补，不是竞争！

📋 一图速记

┌─────────────────────────────────────────────────────────────────┐
│                    A2A 协议知识地图                              │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│  [发现] Agent Card ──► [通信] JSON-RPC ──► [执行] Task          │
│     │                      │                   │                │
│     ▼                      ▼                   ▼                │
│  /.well-known/       tasks/send           状态流转              │
│  agent.json          tasks/subscribe      submitted→working     │
│                      tasks/get            →completed/failed     │
│                      tasks/cancel                               │
│                                                                  │
│  [安全] 认证机制     [传输] HTTP/SSE      [产出] Artifact       │
│  Bearer/OAuth2       同步/流式            文本/文件/数据         │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘

📋 公式速记

A2A 完整流程 = 发现(Agent Card) + 通信(JSON-RPC) + 执行(Task) + 产出(Artifact)

协议选择公式：
  - 需要调用工具？ → MCP
  - 需要 Agent 协作？ → A2A
  - 简单函数调用？ → Function Calling
  - 两者都需要？ → MCP + A2A

常见误区避坑指南

❌ 误区一：A2A 和 MCP 是竞争关系

正解：两者是互补关系！

┌─────────────────────────────────────────────────────────────────┐
│                    正确理解                                      │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│  ┌─────────────────────────────────────────────────────────┐    │
│  │                      完整 AI 系统                        │    │
│  │                                                          │    │
│  │  ┌───────┐         ┌───────┐         ┌───────┐          │    │
│  │  │Agent A│◄──A2A──►│Agent B│◄──A2A──►│Agent C│          │    │
│  │  └───┬───┘         └───┬───┘         └───┬───┘          │    │
│  │      │                 │                 │               │    │
│  │      MCP               MCP               MCP             │    │
│  │      │                 │                 │               │    │
│  │  ┌───▼───┐         ┌───▼───┐         ┌───▼───┐          │    │
│  │  │ 工具1 │         │ 工具2 │         │ 工具3 │          │    │
│  │  └───────┘         └───────┘         └───────┘          │    │
│  │                                                          │    │
│  └─────────────────────────────────────────────────────────┘    │
│                                                                  │
│  A2A 负责 Agent 间协作，MCP 负责 Agent 使用工具                  │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘

❌ 误区二：A2A 只能用于 Google 产品

正解：A2A 是开源开放协议，任何框架都可以实现！

已支持的企业/框架（首批 50+）：

Salesforce、SAP、ServiceNow
LangChain、MongoDB、PayPal
Atlassian、Box、Cohere
...

❌ 误区三：简单任务也要用 A2A

正解：A2A 适合跨 Agent 协作场景，简单任务使用 Function Calling 即可。

场景	推荐方案
单 Agent 调用计算器	Function Calling
单 Agent 查数据库	MCP
多 Agent 协作完成报告	A2A
Agent 联邦共同决策	A2A

❌ 误区四：Agent Card 可以随便放

正解：Agent Card 必须放在 /.well-known/agent.json 路径！

✅ https://my-agent.com/.well-known/agent.json
❌ https://my-agent.com/agent.json
❌ https://my-agent.com/api/agent-card

❌ 误区五：忽略任务状态管理

正解：A2A 的 Task 有完整生命周期，必须正确处理状态转换！

python

# ❌ 错误：忽略 input_required 状态
if task.status.state == "completed":
    return task.artifacts

# ✅ 正确：处理所有状态
match task.status.state:
    case "completed":
        return task.artifacts
    case "failed":
        raise TaskFailedError(task.status.message)
    case "input_required":
        # 处理需要更多输入的情况
        additional_input = await get_user_input()
        return await send_additional_input(task.id, additional_input)
    case "working":
        # 等待或轮询
        await asyncio.sleep(1)
        return await poll_task_status(task.id)

❌ 误区六：不配置认证就上生产

正解：生产环境必须配置认证机制！

json

// ❌ 危险：无认证
{
  "name": "My Agent",
  "url": "https://my-agent.com/a2a"
}

// ✅ 安全：配置认证
{
  "name": "My Agent",
  "url": "https://my-agent.com/a2a",
  "authentication": {
    "schemes": ["Bearer"],
    "credentials": "oauth2"
  }
}

学习检查清单

完成本文档学习后，你应该能够：

[ ] 清晰定义 A2A 协议及其核心价值
[ ] 区分 A2A 和 MCP 的定位与适用场景
[ ] 编写标准的 Agent Card
[ ] 理解 Task 的 6 种状态及其转换
[ ] 使用 JSON-RPC 格式发送 A2A 请求
[ ] 实现简单的 A2A Server 和 Client
[ ] 配置认证和安全机制
[ ] 处理常见错误和异常情况
[ ] 设计多 Agent 协作架构

📝 文档版本：v1.0
最后更新：2026-02-09
协议版本：A2A 1.0.0 (2025.04)
适用人群：AI 应用开发者、架构师

A2A（Agent2Agent）协议全面学习指南 ​

目录 ​

一、核心基础 ​

1.1 A2A 协议定义 ​

1.2 设计初衷与背景 ​

1.3 核心定位：A2A vs MCP ​

1.4 与其他协议的对比 ​

1.5 适用场景与不适用场景 ​

✅ 适用场景 ​

❌ 不适用场景 ​

二、技术规范 ​

2.1 核心架构 ​

2.2 核心概念详解 ​

2.2.1 Agent Card（智能体名片） ​

2.2.2 Task（任务） ​

2.2.3 Message（消息） ​

2.2.4 Part（内容单元） ​

2.2.5 Artifact（工件/产出物） ​

2.3 通信模型 ​

2.3.1 通信流程 ​

2.3.2 请求/响应格式（JSON-RPC 2.0） ​

2.4 核心方法（Methods） ​

2.5 传输方式 ​

2.6 认证与安全机制 ​

2.7 错误处理规范 ​

三、核心能力 ​

3.1 智能体发现（Agent Discovery） ​

3.2 任务编排（Task Orchestration） ​

3.3 多智能体协同 ​

3.4 消息路由 ​

3.5 关键特性总结 ​

四、应用实践 ​

4.1 典型落地场景 ​

场景一：多智能体协作系统 ​

场景二：AI 工作流 ​

场景三：分布式智能体平台 ​

4.2 主流适配框架/工具 ​

4.3 快速上手实践 ​

Step 1：创建 Agent Card ​

Step 2：实现 A2A Server（Python） ​

Step 3：实现 A2A Client ​

Step 4：运行测试 ​

五、进阶拓展 ​

5.1 版本迭代与更新 ​

5.2 性能优化技巧 ​

5.3 私有化部署要点 ​

5.3.1 部署架构 ​

5.3.2 关键配置 ​

5.3.3 安全加固 ​

5.4 生态与周边工具链 ​

核心知识点速记清单 ​

📋 30 秒速记 ​

📋 一图速记 ​

📋 公式速记 ​

常见误区避坑指南 ​

❌ 误区一：A2A 和 MCP 是竞争关系 ​

❌ 误区二：A2A 只能用于 Google 产品 ​

❌ 误区三：简单任务也要用 A2A ​

❌ 误区四：Agent Card 可以随便放 ​

❌ 误区五：忽略任务状态管理 ​

❌ 误区六：不配置认证就上生产 ​

学习检查清单 ​

A2A（Agent2Agent）协议全面学习指南

目录

一、核心基础

1.1 A2A 协议定义

1.2 设计初衷与背景

1.3 核心定位：A2A vs MCP

1.4 与其他协议的对比

1.5 适用场景与不适用场景

✅ 适用场景

❌ 不适用场景

二、技术规范

2.1 核心架构

2.2 核心概念详解

2.2.1 Agent Card（智能体名片）

2.2.2 Task（任务）

2.2.3 Message（消息）

2.2.4 Part（内容单元）

2.2.5 Artifact（工件/产出物）

2.3 通信模型

2.3.1 通信流程

2.3.2 请求/响应格式（JSON-RPC 2.0）

2.4 核心方法（Methods）

2.5 传输方式

2.6 认证与安全机制

2.7 错误处理规范

三、核心能力

3.1 智能体发现（Agent Discovery）

3.2 任务编排（Task Orchestration）

3.3 多智能体协同

3.4 消息路由

3.5 关键特性总结

四、应用实践

4.1 典型落地场景

场景一：多智能体协作系统

场景二：AI 工作流

场景三：分布式智能体平台

4.2 主流适配框架/工具

4.3 快速上手实践

Step 1：创建 Agent Card

Step 2：实现 A2A Server（Python）

Step 3：实现 A2A Client

Step 4：运行测试

五、进阶拓展

5.1 版本迭代与更新

5.2 性能优化技巧

5.3 私有化部署要点

5.3.1 部署架构

5.3.2 关键配置

5.3.3 安全加固

5.4 生态与周边工具链

核心知识点速记清单

📋 30 秒速记

📋 一图速记

📋 公式速记

常见误区避坑指南

❌ 误区一：A2A 和 MCP 是竞争关系

❌ 误区二：A2A 只能用于 Google 产品

❌ 误区三：简单任务也要用 A2A

❌ 误区四：Agent Card 可以随便放

❌ 误区五：忽略任务状态管理

❌ 误区六：不配置认证就上生产

学习检查清单