🌐 A2A：让AI Agent"说上话"的通用语言

🔍 原理揭秘

世界上有一种协议叫A2A。它就像AI世界的"联合国同声传译"——"OpenClaw Agent说日语，Claude Agent说法语，GPT Agent说中文，没关系，我A2A统一翻译，大家都能懂。"

┌─────────────────────────────────────────────────────────────────┐
│                    A2A 生态系统                                │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│  ┌──────────────┐                         ┌──────────────┐       │
│  │  OpenClaw   │                         │   Claude     │       │
│  │   Agent     │◄─────────────────────►│    Agent     │       │
│  └──────┬───────┘                         └──────┬───────┘       │
│         │                                          │              │
│         │  A2A Client                             │              │
│         │  ┌─────────┐                          │              │
│         │  │ Agent   │                          │              │
│         │  │ Card    │◄─────────────────────┐  │              │
│         │  └─────────┘                       │  │              │
│         │                                     │  │              │
│         └─────────────────────────────────────┘  │              │
│                             A2A JSON-RPC        │              │
│                                                  │              │
│  ┌──────────────┐                         ┌──────┴───────┐       │
│  │    GPT       │◄─────────────────────►│   任意Agent   │       │
│  │   Agent      │                         │   (A2A兼容)   │       │
│  └──────────────┘                         └──────────────┘       │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘

Agent Card (代理卡):
{
  "name": "客服Agent",
  "description": "处理用户咨询",
  "url": "https://agent.example.com/a2a",
  "version": "1.0",
  "capabilities": {
    "streaming": true,
    "pushNotifications": true
  },
  "skills": [
    {"id": "customer_support", "name": "客户服务"}
  ]
}

A2A 核心特性

• Agent Discovery（代理发现） - 通过Agent Card发布自己的能力
• Task Management（任务管理） - 支持同步/异步任务提交和状态追踪
• Streaming（流式传输） - 实时获取Agent响应
• Push Notifications（推送通知） - 任务完成后主动通知
• Security（安全机制） - 内置认证和授权

⚡ A2A vs MCP：各司其职

简单理解：MCP是AI的"手"（操作工具），A2A是AI的"嘴"（和其他AI对话）。两者互补，配合使用效果更佳！

💻 代码示例

场景一：发现并调用其他Agent

# A2A 客户端示例
from a2a import AgentClient, AgentCard

# 1. 发现Agent
async def discover_agents():
    registry_url = "https://agent-registry.example.com"
    client = AgentClient(registry_url)
    
    # 搜索能写代码的Agent
    agents = await client.search_agents(
        capability="coding",
        language="python"
    )
    return agents

# 2. 获取Agent Card
async def get_agent_card(agent_url: str):
    """获取Agent的能力描述"""
    card = await AgentClient.fetch_card(
        f"{agent_url}/.well-known/agent.json"
    )
    return card

# 3. 发送任务
async def send_task(agent_url: str, task: str):
    client = AgentClient(agent_url)
    
    # 发送任务并获取结果
    result = await client.send_task({
        "id": "task-123",
        "message": {
            "role": "user",
            "parts": [{"type": "text", "text": task}]
        }
    })
    
    return result

场景二：实现A2A服务器

# 实现一个A2A兼容的Agent服务
from a2a import AgentServer, A2ATask
from fastapi import FastAPI

app = FastAPI()
server = AgentServer()

# 1. 定义Agent Card
@server.agent_card
def get_card():
    return {
        "name": "写作助手",
        "description": "帮你写各类文章",
        "url": "https://myagent.com/a2a",
        "version": "1.0",
        "capabilities": {
            "streaming": True,
            "pushNotifications": False
        },
        "skills": [
            {"id": "article", "name": "写文章"},
            {"id": "summary", "name": "写摘要"}
        ]
    }

# 2. 处理任务
@server.on_task
async def handle_task(task: A2ATask):
    """接收并处理任务"""
    user_message = task.message.parts[0].text
    
    # 调用实际处理逻辑
    result = await my_llm.generate(user_message)
    
    # 返回结果
    await task.send_result({
        "id": task.id,
        "status": "completed",
        "message": {
            "role": "agent",
            "parts": [{"type": "text", "text": result}]
        }
    })

# 3. 启动服务
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8080)

场景三：流式响应

# 使用流式响应
async def stream_task(agent_url: str, prompt: str):
    client = AgentClient(agent_url)
    
    # 发送任务并获取流
    async for chunk in client.send_task_streaming({
        "id": "stream-123",
        "message": {
            "role": "user", 
            "parts": [{"type": "text", "text": prompt}]
        }
    }):
        if chunk.type == "content":
            print(chunk.text, end="")
        elif chunk.type == "status":
            print(f"\n[状态: {chunk.status}]")

⚡ OpenClaw 实战应用

场景一：跨平台Agent协作

# OpenClaw 中使用 A2A 调用外部Agent
agent:
  name: "智能协调员"
  tools:
    # OpenClaw 原生工具
    - web_search
    - write
    
    # A2A 外部Agent
    - a2a_agent:
        name: "Claude Writer"
        url: "https://claude-agent.example.com/a2a"
        skill: "writing"
        
    - a2a_agent:
        name: "GPT Coder"
        url: "https://gpt-code.example.com/a2a"
        skill: "coding"

  workflow:
    - search_info      # 搜索资料
    - call Claude      # A2A调用Claude写初稿
    - call GPT         # A2A调用GPT优化代码示例
    - compile_final    # 整合输出

🎯 A2A 最佳实践

• Agent Card要完整 - 清晰描述能力，方便被发现
• 错误处理要做好 - 网络超时、Agent不可用都要考虑
• 安全不能忘 - 生产环境要加认证
• 结合MCP使用 - A2A负责Agent间通信，MCP负责工具调用