让不同 AI 智能体会「对话」：A2A 协议实战（Agent 与 Agent 通信）

A2A 协议围绕几个简单但关键的概念运转，先建立心智模型：

1) Agent Card（Agent 名片）
   每个 A2A Agent 必须暴露一个 JSON「名片」，写清楚：
   - 自己叫什么、能干什么（skills）
   - 服务地址 URL、支持的认证方式
   其他 Agent 靠这张名片「发现」它、了解它的能力。

2) Client / Remote 架构
   Client Agent = 发起任务的「甲方」
   Remote Agent  = 执行任务的「乙方」
   甲方把任务发给乙方，乙方干完把结果回传。

3) Task 生命周期（状态机）
   submitted → working → input-required → completed
                  ↓            ↓
                failed      canceled
   一次任务从「已提交」到「已完成 / 失败 / 取消」全程可追溯。

4) Message & Artifact（消息与工件）
   支持文本、图片、文件；Artifacts 是 Agent 产出的「成果物」
   （一篇报告、一段代码、一份数据表）。

Google 官方维护了一个轻量的 a2a Python SDK（开源），本教程用它搭最小系统。先建个干净环境：

# 建虚拟环境（推荐，避免污染全局）
python -m venv .venv
source .venv/bin/activate        # Windows: .venv\Scripts\activate

# 安装官方 SDK 与示例依赖
pip install a2a-sdk httpx uvicorn pydantic

# 确认装好
python -c "import a2a; print('a2a ready')"

我们先做一个最简单的「写作 Agent」：它收到一段要点，返回一篇润色后的短文。关键是写一个 Agent Card 描述它的能力，并把它跑成一个 A2A 服务。

# writer_agent/card.json  —— 它的「名片」
{
  "name": "Writer Agent",
  "description": "把要点扩写成通顺的中文短文",
  "url": "http://localhost:8001/a2a",
  "version": "2026-07-19",
  "capabilities": { "streaming": false, "pushNotifications": false },
  "skills": [
    { "id": "expand_points", "name": "要点扩写",
      "description": "输入 3-5 个要点，输出一篇 300 字短文" }
  ],
  "authentication": { "schemes": ["None"] }
}

用 SDK 把 Agent Card 挂起来，并实现「收到要点 → 调 LLM 扩写 → 返回 Artifact」的逻辑（LLM 部分用任意 OpenAI 兼容接口即可）：

# writer_agent/server.py（核心思路，伪代码贴近真实 API）
from a2a.server import A2AServer
from a2a.types import AgentCard, TaskStatus, Artifact
import json, openai

card = AgentCard.model_validate(json.load(open("card.json")))
server = A2AServer(agent_card=card)

@server.task_handler()
async def handle(task):
    points = task.message.content           # 甲方发来的要点
    # 调你自己的 LLM（此处用 OpenAI 兼容 SDK 示意）
    resp = openai.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role":"user",
          "content": f"把以下要点扩写成通顺短文：\n{points}"}]
    )
    text = resp.choices[0].message.content
    return TaskStatus(state="completed",
                      artifacts=[Artifact(type="text", content=text)])

server.run(port=8001)

现在做一个「研究 Agent」：它先产出几个要点，再通过 A2A Client 把要点派给写作 Agent，最后把拿回来的短文打印出来。

# research_agent/client.py
from a2a.client import A2AClient

async def main():
    # 1) 发现乙方：读取它的 Agent Card
    client = A2AClient("http://localhost:8001/a2a")
    card = await client.discover_agent()
    print("发现协作方：", card["name"], "技能：", card["skills"])

    # 2) 研究 Agent 自己先想出要点
    points = "1) A2A 让 Agent 跨框架协作\n2) 与 MCP 互补\n3) 2026 已成事实标准"

    # 3) 派发任务，回收 Artifact
    task = await client.send_task({"message": {"role":"user","content": points}})
    print("写作 Agent 返回：\n", task.result)

import asyncio
asyncio.run(main())

# 终端 1：启动写作 Agent（乙方）
python writer_agent/server.py
# 终端 2：运行研究 Agent（甲方）
python research_agent/client.py

# 成功会看到：打印出「发现协作方」+ 一篇扩写短文

验证通过后，可以照同样套路加更多「乙方」——比如一个搜索 Agent（内部用 MCP 接搜索工具）、一个数据分析 Agent，让研究 Agent 当编排者统一派活：

研究 Agent（编排）
   ├─ A2A → 搜索 Agent  （内部 MCP 接搜索引擎）
   ├─ A2A → 写作 Agent  （本教程已跑通）
   └─ A2A → 数据 Agent  （内部 MCP 接数据库）