A2A——Agent 到 Agent 协议

A2A——Agent 到 Agent 协议

本节摘要:Google 2025/04 发布 A2A,到 2026/04 规范在 a2a-protocol.org、150+ 组织背书。A2A 是 MCP 的横向补充:MCP 是纵向(Agent↔工具),A2A 是对等(Agent↔Agent)。它定义了 Agent Card(发现)、带产物的任务(文本、结构化数据、视频)、不透明的任务生命周期、认证。生产系统越来越多地 MCP + A2A 配对使用,Google Cloud 在 2025-2026 把 A2A 支持-roll 进 Vertex AI Agent Builder。本节聚焦 A2A 作为跨系统 Agent 调用的通用线协议,用 http.server + JSON 从零搭一个最小 A2A 服务器与客户端,走通发现-提交-轮询-取产物的完整流程,并讲清它何时胜过又何时败给直接 RPC、ACP、ANP、NLIP。

学习目标

阅读完本节,你应当能够:

1. 说清 A2A 的四个要素(Agent Card、Task、Artifact、不透明生命周期)及其与 MCP 的纵向/横向分工。
2. 用 http.server + JSON 从零实现一个最小 A2A 服务器(暴露 Agent Card、收任务、管状态、返产物)与客户端。
3. 选择认证模式(bearer/mTLS/签名)与同步模型(轮询 vs SSE 流式)。
4. 判断 A2A 何时适合(跨组织、异构框架、类型化产物、长时任务)与何时败给直接 RPC(亚毫秒、同进程、小团队)。
5. 在 MCP/A2A 双协议栈中定位 A2A,并区别于 ACP/ANP/NLIP。

一、问题与直觉

你的 Agent 需要调用另一个系统上的 Agent。怎么做?你可以暴露一个 HTTP 端点、定义一套自定义 JSON schema、祈祷对面讲这套。每对 Agent 都成了一次定制集成。

A2A 就是那个调用的通用线协议。标准发现、标准任务模型、标准传输、标准产物。像 HTTP+REST,但把 Agent 当一等公民。

四个要素

• Agent Card:JSON 文档,在 /.well-known/agent.json,描述名称、技能、端点、支持模态、认证要求。发现即读这张卡。
• Task:工作单元,异步有状态对象,生命周期 submitted → working → completed / failed / canceled。客户端发任务,轮询或订阅更新。
• Artifact:任务产出的结果类型。文本、结构化 JSON、图像、视频、音频。产物是类型化的,不同模态都是一等公民。
• 不透明生命周期:A2A 不规定远端 Agent 如何解题。客户端看到状态迁移与产物;实现方自由用任何框架。

MCP/A2A 分工

• MCP(第 13 章):Agent↔工具。Agent 经 JSON-RPC 读写工具服务器,默认无状态。
• A2A:Agent↔Agent。对等协议,两边都是带自己推理的 Agent。

生产多智能体系统两者都用:一个 A2A 对等端在自己一侧调用 MCP 工具。这种分工让两个关注点保持干净。

发现流程

Client                     Agent server
  ├──GET /.well-known/agent.json──>
  <──Agent Card JSON─────────────
  ├──POST /tasks {skill, input}──>
  <──201 task_id, state=submitted
  ├──GET /tasks/{id}──────────────>
  <──state=working, 42% done──────
  ├──GET /tasks/{id}──────────────>
  <──state=completed, artifacts──
​

或用流式:SSE 订阅 /tasks/{id}/events 做推送更新。

认证

A2A 支持三种常见模式:

• Bearer token:OAuth2 或不透明 token。
• mTLS:双向 TLS,组织间互证身份。
• 签名请求:对载荷做 HMAC。

认证在 Agent Card 里声明;客户端发现后遵守。

150+ 组织(到 2026/04)

企业采用驱动了 A2A 规模。要点:A2A 成了企业 Agent 系统跨越信任边界的方式。Google Cloud 的 Vertex AI Agent Builder 支持 A2A;Microsoft Agent Framework 支持;主流框架(LangGraph、CrewAI、AutoGen)都提供 A2A 适配器。

何时胜出

• 跨组织调用:A 公司 Agent 调 B 公司 Agent。无 A2A 则每对都是定制契约。
• 异构框架:LangGraph Agent 调 CrewAI Agent 调自定义 Python Agent,A2A 归一化。
• 类型化产物:视频结果、结构化 JSON、音频都是一等公民。
• 长时任务:不透明生命周期 + 轮询让数小时任务变得直接。

何时挣扎

• 延迟敏感的微调用:A2A 生命周期是异步的,亚毫秒 Agent 间调用不合适,用直接 RPC。
• 紧耦合同进程 Agent:若两个 Agent 跑在同一 Python 进程,A2A 的 HTTP 往返是杀鸡用牛刀。
• 小团队:规范开销是真实的,纯内部 Agent 可能不需要这种正式度。

⚠️ A2A 解决的是「跨边界协调」——跨组织、跨框架、跨进程。在这些场景下它把每对 Agent 的定制契约归一成一份标准;但在边界内(同进程、亚毫秒、小团队),它的开销反而成累赘。协调工具的价值随边界数量增长,这是选 A2A 与否的第一判据。

A2A vs ACP/ANP/NLIP

2024-2026 涌现的相邻规范(第 03 节有详尽对比):

• ACP(IBM/Linux Foundation):A2A 的前身,作用域更窄。
• ANP(Agent Network Protocol):重对等发现,去中心化优先。
• NLIP(Ecma 自然语言交互协议,2025/12 标准化):自然语言内容类型。

截至 2026/04,A2A 是采用最广的对等协议。综述见 arXiv:2505.02279。

二、从零实现

code/main.py 用 http.server + JSON 实现一个最小 A2A 服务器与客户端。

服务器

from http.server import BaseHTTPRequestHandler, HTTPServer
import json, uuid, threading

TASKS = {}

class A2AServer(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.path == "/.well-known/agent.json":
            self._respond(200, {
                "name": "code-review-agent",
                "skills": ["review-python", "review-typescript"],
                "endpoints": {"tasks": "http://localhost:8080/tasks"},
                "auth": {"type": "bearer"}, "modalities": ["text","structured"]})
        elif self.path.startswith("/tasks/"):
            tid = self.path.split("/")[-1]
            self._respond(200, TASKS[tid])
    def do_POST(self):
        if self.path == "/tasks":
            body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
            tid = str(uuid.uuid4())
            TASKS[tid] = {"id": tid, "state": "submitted", "input": body}
            threading.Thread(target=self._process, args=(tid,)).start()
            self._respond(201, {"task_id": tid, "state": "submitted"})
    def _process(self, tid):
        TASKS[tid]["state"] = "working"
        result = do_review(TASKS[tid]["input"])     # 真实工作
        TASKS[tid].update({"state": "completed",
            "artifacts": [{"type":"text","content": result}]})
    def _respond(self, code, body):
        self.send_response(code); self.send_header("Content-Type","application/json")
        self.end_headers(); self.wfile.write(json.dumps(body).encode())
​

客户端

def a2a_call(server_url, skill, payload):
    card = http_get(f"{server_url}/.well-known/agent.json")   # 1. 发现
    task = http_post(card["endpoints"]["tasks"],
                    {"skill": skill, "input": payload})        # 2. 提交
    while True:                                                # 3. 轮询
        state = http_get(f"{server_url}/tasks/{task['task_id']}")
        if state["state"] in ("completed","failed","canceled"): break
        time.sleep(0.5)
    return state["artifacts"]                                  # 4. 取产物
​

设计要点:客户端完全不知道服务器用什么框架解题——它只见状态迁移与产物。这正是「不透明生命周期」的价值:实现方可自由替换框架,客户端契约不变。

三、框架对比

💡 心法:MCP 接工具(纵向),A2A 接 Agent(横向)。 生产栈两者都要——一个 Agent 既用 MCP 调本地工具,又用 A2A 调其他系统的 Agent。把它们当互补而非互斥。

四、可复用产物

outputs/skill-a2a-integrator.md:设计 A2A 集成——Agent Card 内容、任务 schema、认证选择、流式 vs 轮询。

五、练习

1. 跑通:跑 code/main.py,确认客户端发现服务器并收到正确产物。
2. 加技能:给服务器加第二个技能(如「summarize」),更新 Agent Card,写一个按任务类型选技能的客户端。
3. SSE 流式:实现 /tasks/{id}/events 的 SSE 端点发状态变化。客户端要怎么改?
4. 读规范:读 A2A 规范,识别本演示未实现的三个规范要求。
5. 对比发现:对比 A2A 的 Agent Card 发现与 MCP 的 listTools 能力列出。自描述 Agent 与能力探测各自的权衡?

本节要点回顾

1. A2A 是 Agent↔Agent 的横向对等协议:MCP 纵向接工具,A2A 横向接 Agent,生产栈两者都用。
2. 四个要素:Agent Card(发现)、Task(异步有状态工作)、Artifact(类型化产物)、不透明生命周期(实现方框架自由)。
3. 发现 = 读 Agent Card:GET /.well-known/agent.json;同步模型是轮询或 SSE 流式。
4. 三种认证:bearer、mTLS、签名请求,在 Agent Card 里声明。
5. 150+ 组织背书:Google Cloud Vertex AI、Microsoft Agent Framework、主流框架都有适配器。
6. 何时胜出:跨组织、异构框架、类型化产物、长时任务。
7. 何时挣扎:亚毫秒微调用、同进程紧耦合、小团队(规范开销不划算)。
8. 选型第一判据:协调工具的价值随边界数量增长——边界多就上 A2A,边界内用直接 RPC。

下一节,我们回到多智能体内部——共享记忆与黑板模式,看多个 Agent 如何通过一块共享的「黑板」读写来协调,而不必点对点通信。