A2A 协议详解：Google 的 Agent 间通信标准

之前:
  LangChain Agent ──❌── AutoGen Agent
  CrewAI Agent    ──❌── 自研 Agent

之后:
  LangChain Agent ──A2A── AutoGen Agent
  CrewAI Agent    ──A2A── 自研 Agent
  任何 Agent      ──A2A── 任何 Agent

{
  "name": "TravelPlanner",
  "description": "专业旅游规划 Agent，支持多约束行程优化",
  "url": "https://travel-agent.example.com",
  "provider": {
    "organization": "TravelAgent Inc.",
    "url": "https://travelagent.com"
  },
  "capabilities": {
    "streaming": true,
    "pushNotifications": true,
    "stateTransitionHistory": true
  },
  "defaultInputModes": ["text", "text/plain"],
  "defaultOutputModes": ["text", "text/plain"],
  "skills": [
    {
      "id": "itinerary-planning",
      "name": "行程规划",
      "description": "根据用户需求生成多天旅游行程",
      "tags": ["travel", "planning", "optimization"],
      "examples": [
        "帮我规划一个3天的西安旅游行程",
        "带老人去北京4天，少走路的行程"
      ]
    },
    {
      "id": "weather-query",
      "name": "天气查询",
      "description": "查询目的地实时天气和预报",
      "tags": ["weather", "forecast"]
    }
  ]
}

{
  "id": "task_abc123",
  "sessionId": "session_xyz789",
  "status": {
    "state": "working",
    "message": {
      "role": "agent",
      "parts": [{"type": "text", "text": "正在查询西安景点信息..."}]
    },
    "timestamp": "2025-06-19T10:30:00Z"
  },
  "history": [
    {
      "role": "user",
      "parts": [{"type": "text", "text": "帮我规划一个3天西安游"}]
    }
  ]
}

submitted → working → input-required → working → completed
                ↓                         ↓
             failed                    canceled

{
  "role": "agent",
  "parts": [
    {
      "type": "text",
      "text": "我帮你规划了一个3天西安行程：\n\n第1天：兵马俑、华清池\n第2天：古城墙、回民街\n第3天：大雁塔、陕西历史博物馆"
    },
    {
      "type": "data",
      "data": {
        "itinerary": {
          "days": 3,
          "total_budget": 2500,
          "pois": ["兵马俑", "华清池", "古城墙", "回民街", "大雁塔", "陕西历史博物馆"]
        }
      }
    }
  ]
}

{
  "artifactId": "artifact_001",
  "name": "西安3日行程.pdf",
  "parts": [
    {
      "type": "file",
      "file": {
        "url": "https://travel-agent.example.com/files/itinerary_abc123.pdf",
        "mimeType": "application/pdf",
        "name": "西安3日行程.pdf"
      }
    }
  ]
}

Assistant Agent                     Travel Agent
      │                                  │
      │  ① 发现 Agent Card              │
      │─────────────────────────────────>│
      │  GET /.well-known/agent.json     │
      │<─────────────────────────────────│
      │  {skills: [itinerary-planning]}  │
      │                                  │
      │  ② 创建 Task                    │
      │─────────────────────────────────>│
      │  POST /tasks                     │
      │  {message: "规划3天西安游"}       │
      │<─────────────────────────────────│
      │  {taskId: "task_001",            │
      │   state: "submitted"}            │
      │                                  │
      │  ③ 轮询 Task 状态               │
      │─────────────────────────────────>│
      │  GET /tasks/task_001             │
      │<─────────────────────────────────│
      │  {state: "working"}              │
      │                                  │
      │  ④ 获取结果                     │
      │─────────────────────────────────>│
      │  GET /tasks/task_001             │
      │<─────────────────────────────────│
      │  {state: "completed",            │
      │   artifact: {...}}               │

GET /tasks/task_001?stream=true
→ event: status-update, state: working
→ event: status-update, state: working, message: "正在查询天气..."
→ event: status-update, state: completed, artifact: {...}

from a2a.server import A2AServer
from a2a.server.agent import Agent

class TravelAgent(Agent):
    def get_agent_card(self):
        return {
            "name": "TravelPlanner",
            "description": "专业旅游规划",
            "skills": [
                {
                    "id": "itinerary-planning",
                    "name": "行程规划",
                    "description": "生成多天旅游行程",
                }
            ]
        }

    async def execute(self, task):
        query = task.current_message.get_text()
        itinerary = await self.plan_itinerary(query)
        return {
            "message": {
                "role": "agent",
                "parts": [
                    {"type": "text", "text": itinerary},
                    {"type": "data", "data": {"itinerary": itinerary}}
                ]
            },
            "state": "completed"
        }

# 启动服务
server = A2AServer(agent=TravelAgent())
server.start(host="0.0.0.0", port=8080)

from a2a.client import A2AClient

client = A2AClient("https://travel-agent.example.com")

# 创建任务
task = await client.create_task(
    message="帮我规划3天西安游，预算3000"
)

# 等待完成
result = await client.wait_for_completion(task.id)
print(result.artifact)

用户问题 → 意图分类 Agent → 发现需要三方面信息
  → 调用财经 Agent 查数据
  → 调用分析 Agent 做判断
  → 调用写作 Agent 写报告
→ 综合返回