10c - A2A 协议详解

┌─────────────────────────────────────────────────────┐
│                   A2A 协议架构                        │
│                                                       │
│  ┌──────────┐    HTTP/JSON-RPC    ┌──────────┐      │
│  │ Client   │◄──────────────────►│ Remote   │      │
│  │ Agent    │                     │ Agent    │      │
│  │          │    SSE (流式)        │          │      │
│  │ ┌──────┐ │◄────────────────── │ ┌──────┐ │      │
│  │ │ MCP  │ │                     │ │ MCP  │ │      │
│  │ │Tools │ │  Push Notification  │ │Tools │ │      │
│  │ └──────┘ │◄────────────────── │ └──────┘ │      │
│  └──────────┘                     └──────────┘      │
│       │                                │             │
│       ▼                                ▼             │
│  ┌──────────┐                    ┌──────────┐       │
│  │Agent Card│                    │Agent Card│       │
│  │(发现)    │                    │(发现)    │       │
│  └──────────┘                    └──────────┘       │
└─────────────────────────────────────────────────────┘

{
  "name": "Travel Booking Agent",
  "description": "搜索并预订航班和酒店",
  "url": "https://travel-agent.example.com",
  "version": "1.0.0",
  "defaultInputModes": ["text/plain"],
  "defaultOutputModes": ["text/plain", "application/json"],
  "capabilities": {
    "streaming": true,
    "pushNotifications": true,
    "stateTransitionHistory": true
  },
  "skills": [
    {
      "id": "book-flight",
      "name": "Book Flight",
      "description": "搜索并预订航班",
      "tags": ["travel", "flight", "booking"],
      "examples": ["帮我订一张从北京到上海的机票", "查询明天的航班"]
    },
    {
      "id": "book-hotel",
      "name": "Book Hotel",
      "description": "搜索并预订酒店",
      "tags": ["travel", "hotel", "booking"],
      "examples": ["帮我在上海找一家酒店"]
    }
  ],
  "securitySchemes": {
    "oauth2": {
      "type": "oauth2",
      "flows": {
        "clientCredentials": {
          "tokenUrl": "https://auth.example.com/token",
          "scopes": {
            "agent:invoke": "调用 Agent 服务"
          }
        }
      }
    }
  },
  "supportsAuthenticatedExtendedCard": true
}

                    ┌──────────┐
                    │SUBMITTED │ ← 客户端创建任务
                    └────┬─────┘
                         │
                    ┌────▼─────┐
              ┌────►│ WORKING  │◄────┐
              │     └────┬─────┘     │
              │          │           │
              │    ┌─────▼──────┐    │
              │    │  INPUT     │    │
              └────┤  REQUIRED  ├────┘
                   └─────┬──────┘
                         │
           ┌─────────────┼─────────────┐
           │             │             │
     ┌─────▼────┐  ┌────▼─────┐ ┌────▼──────┐
     │COMPLETED │  │  FAILED  │ │ CANCELLED │
     └──────────┘  └──────────┘ └───────────┘
                         │
                   ┌─────▼──────┐
                   │  REJECTED  │
                   └────────────┘

Client                          Remote Agent
  │                                  │
  │── POST /tasks/send ────────────►│
  │                                  │ (处理)
  │◄── 200 OK + Task(completed) ────│
  │                                  │

Client                          Remote Agent
  │                                  │
  │── POST /tasks/sendSubscribe ───►│
  │                                  │
  │◄── SSE: status=working ─────────│
  │◄── SSE: artifact(部分结果) ─────│
  │◄── SSE: artifact(更多结果) ─────│
  │◄── SSE: status=completed ───────│
  │                                  │

Client                          Remote Agent
  │                                  │
  │── POST /tasks/send ────────────►│
  │◄── 200 OK + Task(submitted) ────│
  │                                  │ (长时间处理...)
  │                                  │
  │◄── Push: status=working ────────│
  │◄── Push: status=completed ──────│
  │                                  │
  │── GET /tasks/{id} ─────────────►│
  │◄── 200 OK + Task + Artifacts ───│
  │                                  │

用户 ──登录──► Agent 1 ──OBO Token Exchange──► Auth Server
                                                    │
                                              签发新 Token
                                              (sub=alice,
                                               act=agent1)
                                                    │
                Agent 2 ◄──── 携带 OBO Token ───────┘
                  │
                  ├── 知道是 Alice 发起的请求
                  ├── 知道是通过 Agent 1 转发的
                  └── 可以执行用户级别的权限控制

from a2a.types import AgentCard, AgentSkill, AgentCapabilities
 
skill = AgentSkill(
    id="data-analysis",
    name="数据分析",
    description="分析数据集并生成可视化报告",
    tags=["data", "analysis", "visualization"],
    examples=["分析这份销售数据", "生成月度报告"],
)
 
agent_card = AgentCard(
    name="Data Analysis Agent",
    description="专业数据分析 Agent",
    url="https://data-agent.example.com",
    version="1.0.0",
    default_input_modes=["text", "application/json"],
    default_output_modes=["text", "application/json", "image/png"],
    capabilities=AgentCapabilities(
        streaming=True,
        push_notifications=True,
    ),
    skills=[skill],
)

from a2a.server import A2AServer, TaskHandler
from a2a.types import Task, TaskState, Message, Part
 
class DataAnalysisHandler(TaskHandler):
    async def handle_task(self, task: Task) -> Task:
        # 更新状态为 working
        task.state = TaskState.WORKING
        await self.send_update(task)
 
        # 解析用户请求
        user_message = task.messages[-1]
        query = user_message.parts[0].text
 
        # 执行分析（通过 MCP 访问数据源）
        result = await self.analyze_data(query)
 
        # 返回结果
        task.artifacts = [result]
        task.state = TaskState.COMPLETED
        return task
 
# 启动 A2A 服务
server = A2AServer(
    agent_card=agent_card,
    handler=DataAnalysisHandler(),
    port=9999,
)
server.start()

from a2a.client import A2AClient
 
# 发现远程 Agent
client = A2AClient()
card = await client.discover("https://travel-agent.example.com")
 
print(f"Agent: {card.name}")
print(f"Skills: {[s.name for s in card.skills]}")
 
# 发送任务
task = await client.send_task(
    url=card.url,
    message="帮我查询明天从北京到上海的航班",
)
 
# 流式接收结果
async for update in client.subscribe_task(task.id):
    if update.state == "working":
        print(f"处理中... {update.progress}")
    elif update.state == "completed":
        for artifact in update.artifacts:
            print(f"结果: {artifact}")

你是一个 A2A 协议集成专家。请帮我设计一个多 Agent 系统的 A2A 集成方案。

系统需求：
- 参与的 Agent 类型：[列出 Agent 及其职责]
- 通信模式：[同步/异步/流式]
- 安全要求：[认证方式、数据敏感度]
- 部署环境：[云平台、内部网络]

请提供：
1. 每个 Agent 的 Agent Card 设计
2. 任务流转的状态机设计
3. 认证方案选择和配置
4. 错误处理和重试策略
5. 监控和可观测性方案

┌─────────────────────────────────────────────┐
│              应用层 / 用户界面                 │
├─────────────────────────────────────────────┤
│                                              │
│    A2A 协议层（水平协作）                      │
│    Agent ◄──────────────► Agent              │
│    "Agent 之间如何协作"                        │
│                                              │
├─────────────────────────────────────────────┤
│                                              │
│    MCP 协议层（垂直集成）                      │
│    Agent ◄──────────────► Tools/Data         │
│    "Agent 如何访问工具和数据"                   │
│                                              │
├─────────────────────────────────────────────┤
│              基础设施层                        │
│         HTTP / SSE / JSON-RPC                │
└─────────────────────────────────────────────┘

客户提交复杂工单
       │
       ▼
┌──────────────┐
│ 客服 Agent    │──── MCP ────► CRM 数据库
│              │──── MCP ────► 订单系统
│              │──── MCP ────► 知识库
└──────┬───────┘
       │ 需要技术诊断（A2A 委托）
       ▼
┌──────────────┐
│ 诊断 Agent    │──── MCP ────► 系统日志
│              │──── MCP ────► 监控面板
└──────┬───────┘
       │ 发现计费问题（A2A 委托）
       ▼
┌──────────────┐
│ 计费 Agent    │──── MCP ────► 支付系统
│              │
└──────┬───────┘
       │ 结果通过 A2A 回传
       ▼
    客户收到解决方案

from google.adk import Agent
from google.adk.tools import MCPTool
from google.adk.a2a import RemoteAgent
 
agent = Agent(
    name="project-manager",
    description="项目管理协调 Agent",
 
    # MCP 工具：直接访问数据
    tools=[
        MCPTool(server="database-server"),
        MCPTool(server="calendar-server"),
    ],
 
    # A2A 远程 Agent：委托专业任务
    remote_agents=[
        RemoteAgent(url="https://research.example.com"),
        RemoteAgent(url="https://analytics.example.com"),
    ],
)
 
# 该 Agent 现在可以：
# - 通过 MCP 查询数据库
# - 通过 MCP 检查日历
# - 通过 A2A 委托研究任务给专业 Agent
# - 通过 A2A 请求分析 Agent 生成报告

┌─────────────────────────────────────────────────┐
│                                                   │
│   AAIF（Agentic AI Foundation）                   │
│   ┌─────────────────────────────────────────┐    │
│   │  治理层：开放标准、社区协作、中立治理       │    │
│   │  项目：MCP、goose、AGENTS.md 等           │    │
│   └─────────────────────────────────────────┘    │
│                                                   │
│   ┌──────────────────┐  ┌──────────────────┐    │
│   │   A2A 协议        │  │   ACP/ANP 等     │    │
│   │   Agent 间通信     │  │   补充协议        │    │
│   │   (水平协作层)     │  │                  │    │
│   └──────────────────┘  └──────────────────┘    │
│                                                   │
│   ┌─────────────────────────────────────────┐    │
│   │   MCP 协议                                │    │
│   │   Agent 到工具/数据连接（垂直集成层）       │    │
│   └─────────────────────────────────────────┘    │
│                                                   │
│   ┌─────────────────────────────────────────┐    │
│   │   基础设施：HTTP、SSE、JSON-RPC、OAuth    │    │
│   └─────────────────────────────────────────┘    │
│                                                   │
└─────────────────────────────────────────────────┘

# 1. 客服 Agent（协调者）
from google.adk import Agent
from google.adk.tools import MCPTool
from google.adk.a2a import RemoteAgent
 
customer_service_agent = Agent(
    name="customer-service",
    description="处理客户咨询、投诉和退款请求",
 
    # 内部工具（MCP）
    tools=[
        MCPTool(server="crm-server"),        # 客户关系管理
        MCPTool(server="order-server"),       # 订单查询
        MCPTool(server="knowledge-base"),     # 知识库
    ],
 
    # 外部 Agent（A2A）
    remote_agents=[
        RemoteAgent(
            url="https://logistics-partner.com",
            auth={"type": "oauth2", "client_id": "..."}
        ),
        RemoteAgent(
            url="https://payment-partner.com",
            auth={"type": "oauth2", "client_id": "..."}
        ),
    ],
)

# 2. 物流 Agent 的 Agent Card（第三方发布）
logistics_card = {
    "name": "Logistics Tracking Agent",
    "description": "实时物流追踪和配送状态查询",
    "url": "https://logistics-partner.com/a2a",
    "version": "2.1.0",
    "skills": [
        {
            "id": "track-package",
            "name": "包裹追踪",
            "description": "根据订单号查询物流状态",
            "tags": ["logistics", "tracking"],
        },
        {
            "id": "estimate-delivery",
            "name": "预计送达",
            "description": "预估包裹送达时间",
            "tags": ["logistics", "delivery", "estimate"],
        }
    ],
    "capabilities": {"streaming": True},
    "securitySchemes": {
        "oauth2": {
            "type": "oauth2",
            "flows": {
                "clientCredentials": {
                    "tokenUrl": "https://auth.logistics-partner.com/token",
                    "scopes": {"tracking:read": "查询物流信息"}
                }
            }
        }
    }
}

# 3. 完整的任务流转
async def handle_refund_request(customer_query: str):
    """处理退款请求的完整 A2A 流程"""
 
    # Step 1: 客服 Agent 通过 MCP 查询订单
    order = await crm_tool.query(customer_query)
 
    # Step 2: 通过 A2A 查询物流状态
    logistics_task = await a2a_client.send_task(
        url="https://logistics-partner.com/a2a",
        message=f"查询订单 {order.id} 的物流状态",
    )
 
    # Step 3: 根据物流状态决定是否退款
    if logistics_task.artifacts[0].data["status"] == "delivered":
        # Step 4: 通过 A2A 发起退款
        refund_task = await a2a_client.send_task(
            url="https://payment-partner.com/a2a",
            message=f"为订单 {order.id} 发起退款，金额 {order.amount}",
        )
 
        # Step 5: 流式跟踪退款进度
        async for update in a2a_client.subscribe_task(refund_task.id):
            if update.state == "completed":
                return f"退款成功：{update.artifacts[0].data}"
            elif update.state == "input-required":
                # 需要额外验证
                return "需要客户提供额外验证信息"
 
    return "物流显示未签收，暂不支持退款"