A2A 协议深度实战：当 AI Agent 学会「找同事」——从 Agent Card 发现、JSON-RPC 任务编排到 SSE 流式与推送通知的生产级完全指南（2026）

如果 2025 年是「Agent 元年」，那 2026 年就是「Agent 联网元年」。当企业里同时跑着 Salesforce 的 Agent、ServiceNow 的 Agent、你自己用 LangChain 写的 Agent，它们之间却像三个说不同方言、互不相识的同事——每次协作都要手写一堆胶水代码。Google 在 2025 年 4 月的 Cloud Next 大会上扔出了 A2A（Agent2Agent Protocol），并在同年 6 月把它捐赠给 Linux 基金会。到 2026 年，已有 150+ 组织承诺支持。本文带你从协议内核、传输层、生命周期，一路打到生产级 A2A Server + 跨语言客户端的完整代码。

一、背景：为什么 AI Agent 之间需要一张「名片」

1.1 问题不是「能不能调」，而是「要不要为每一对 Agent 写一套翻译」

设想一个真实的招聘场景：你有一个部署在 Vertex AI 上的「招聘协调 Agent」，它需要：

• 从 Salesforce Agent 拉候选人列表；
• 让 ServiceNow Agent 安排面试；
• 触发第三方背调服务 Agent 做背景调查。

在 A2A 出现之前，这三对 Agent 之间每一对都要写一套定制胶水代码：Salesforce 用 REST + GraphQL 混合风格，ServiceNow 是另一套事件 API，背调服务又是 SOAP/Webhook 混搭。每接一个新 Agent，就多一套连接层。团队的时间大量消耗在「翻译」，而不是业务本身。

A2A 用一句话解决这个痛点：定义一种所有 Agent 都说的公共语言。 只要双方都支持 A2A，不管底层是 LangChain、AutoGen、ADK 还是你自己手写的循环，不管谁部署的、什么语言实现的，都能直接通信。

1.2 A2A 与 MCP：不是替代，是「正交互补」

很多人把 A2A 和 MCP 对立起来，这是最大的误解。它们解决的是不同层次的问题：

一个生产级 Agent 往往是 MCP + A2A 同时使用：它对外通过 A2A 暴露能力、接受别人的委派；对内通过 MCP 调用工具和数据。这个组合被称为 MCP×A2A，是 2026 年多 Agent 系统的标配架构。

1.3 一段被低估的历程

• 2025-04-09：Google 在 Cloud Next 发布 A2A，初始即获得 Salesforce、SAP、ServiceNow、MongoDB、PayPal、Box、Cohere、LangChain 等 50+ 伙伴支持。
• 2025-06：整个项目捐赠给 Linux 基金会，由 AWS、Cisco、Google、Microsoft、Salesforce、SAP 等共同治理，避免「厂商锁进」。
• 2026：超过 150 家组织支持；官方 Python / TypeScript / Java SDK 成熟；DeepLearning.AI 联合 Google Cloud、IBM Research 推出官方课程。

这一步「捐赠给中立基金会」非常关键——它直接解决了「凭什么信你一个协议」的信任问题，也是它能从 Google 提案变成行业标准的原因。

二、核心概念：A2A 的四大对象与任务生命周期

A2A 的数据模型非常克制，只有四个核心对象：Agent Card、Task、Message、Artifact（以及承载内容的 Part）。理解这四个对象，就理解了 A2A 的 80%。

2.1 Agent Card：Agent 的「数字名片」

每个 A2A Agent 都要在固定路径暴露一张 Agent Card（默认 https://host/.well-known/agent.json）。它告诉别人：「我是谁、我会什么、我在哪、怎么认证」。

{
  "protocolVersion": "0.3.0",
  "name": "DataAnalyst Agent",
  "description": "接收 CSV 文件，返回描述性统计与异常行检测的可运行 Agent",
  "url": "https://data-analyst.example.com/",
  "provider": {
    "organization": "程序员茄子实验室",
    "url": "https://chenxutan.com"
  },
  "version": "1.4.0",
  "capabilities": {
    "streaming": true,
    "pushNotifications": true,
    "stateTransitionHistory": false
  },
  "authentication": {
    "schemes": ["bearer"]
  },
  "defaultInputModes": ["text/plain", "text/csv"],
  "defaultOutputModes": ["application/json", "text/markdown"],
  "skills": [
    {
      "id": "csv_summary",
      "name": "CSV 描述性统计",
      "description": "对数值列计算均值/中位数/分位数/缺失率，并标记异常行",
      "tags": ["data", "etl", "statistics"],
      "examples": ["分析这份销售数据，找出异常订单"],
      "inputModes": ["text/csv"],
      "outputModes": ["application/json"]
    }
  ]
}

几个关键点：

• protocolVersion：客户端会据此协商兼容性，旧客户端遇到新版本可优雅降级。
• capabilities.streaming：是否支持 tasks/sendSubscribe 的 SSE 流式。
• capabilities.pushNotifications：是否支持异步任务的推送通知（Webhook 回调）。
• skills：Agent 真正能干的「活儿」，客户端根据 tags / examples 选最合适的 Agent。
• authentication.schemes：声明支持的认证方式，避免无谓握手。

2.2 Task：一次协作的「工单」

A2A 里所有协作都围绕一个 Task 展开。Task 有严格的生命周期状态机：

submitted → working → (input-required) → working → ... → completed
                              ↓
                          canceled / failed

状态枚举（TaskState）：

注意 input-required 这个状态——它让 Agent 之间能像人一样「先问一句再继续」，而不是一次性要么成功要么失败。这是 A2A 区别于普通 RPC 的关键设计。

2.3 Message 与 Artifact：对话与交付物

• Message：Agent 之间围绕任务的对话内容，包含 role（user / agent）和若干 Part。
• Artifact：任务的最终交付物。一个任务可以有多个 Artifact（比如一份统计 JSON + 一份 Markdown 报告）。
• Part：真正装内容的最小单元，三种类型：
  • TextPart：{ "kind": "text", "text": "..." }
  • FilePart：{ "kind": "file", "file": { "mimeType": "text/csv", "bytes": "<base64>", "name": "x.csv" } }
  • DataPart：{ "kind": "data", "data": { "mean": 3.2, ... } }

为什么用 Part 而不是裸字符串？因为 Agent 之间要协商「用不用文件、用不用结构化数据」。模态协商（modality negotiation） 是 A2A 的精髓：两个 Agent 先交换自己支持的 inputModes / outputModes，再决定用什么格式说话。

三、架构分析：A2A 为什么「不发明新轮子」

A2A 最聪明的地方，是它刻意没有发明任何新传输协议。它全部建在 Web 工程师早已熟悉的栈上：

这个设计决策意味着：你不需要学任何新东西就能上手 A2A。 一个会写 REST 的工程师，十分钟就能看懂 A2A 的报文。

3.1 发现（Discovery）

客户端要找 Agent，标准做法：

GET https://data-analyst.example.com/.well-known/agent.json
→ 返回 Agent Card（上面那份 JSON）

客户端解析 Card，确认对方支持的能力、输入/输出模态、认证方式，再决定是否委托。

3.2 两个核心 RPC 方法

A2A 的 JSON-RPC 方法命名极其直白：

• tasks/send：提交一个消息、同步等结果（适合秒级短任务）。
• tasks/sendSubscribe：提交消息并建立一个 SSE 流，持续收到 TaskStatusUpdateEvent / TaskArtifactUpdateEvent（适合长任务）。
• 辅助方法：tasks/get（拉状态）、tasks/cancel（取消）、tasks/pushNotification/set（注册 Webhook）。

tasks/send 请求示例：

{
  "jsonrpc": "2.0",
  "id": "req-001",
  "method": "tasks/send",
  "params": {
    "id": "task-9f3a",
    "message": {
      "role": "user",
      "parts": [
        {
          "kind": "file",
          "file": {
            "mimeType": "text/csv",
            "bytes": "aWQsdmFsdWUKMSwxMAoyLDkKMywyMDAwCjQsLTEyCg==",
            "name": "orders.csv"
          }
        },
        {
          "kind": "text",
          "text": "分析这份订单数据，标记异常行"
        }
      ]
    }
  }
}

响应（同步完成）：

{
  "jsonrpc": "2.0",
  "id": "req-001",
  "result": {
    "id": "task-9f3a",
    "status": { "state": "completed" },
    "artifacts": [
      {
        "artifactId": "art-1",
        "name": "summary",
        "parts": [
          {
            "kind": "data",
            "data": {
              "rows": 4,
              "columns": ["id", "value"],
              "mean": 499.75,
              "median": 9.5,
              "anomaly_rows": [3, 4]
            }
          }
        ]
      }
    ]
  }
}

3.3 设计原则一：Agent 是「不透明的黑盒」

A2A 的核心前提：你不应该、也不需要知道对方 Agent 的内部实现。 它用什么模型、怎么推理、有没有记忆、用什么语言写的——都不重要。A2A 只定义「两个 Agent 之间交换什么」，不定义「内部怎么运行」。

这一点是跨厂商互操作的基石。Salesforce 的 Agent 和你的 Agent 能协作，恰恰是因为双方都同意「只聊 Task 和 Artifact」。

3.4 流式与推送：长任务不阻塞

对于可能跑几分钟甚至更久的任务：

• 流式（tasks/sendSubscribe）：客户端保持一个 SSE 连接，Agent 边算边推 TaskStatusUpdateEvent（如 working）和增量 TaskArtifactUpdateEvent。适合「用户在界面上看着进度」的场景。
• 推送（pushNotifications）：客户端提交任务后断开，Agent 完成后通过注册好的 Webhook（https://client.example.com/a2a/callback）主动回调。适合「提交后去干别的、完了通知我」的异步场景。

四、代码实战：用官方 Python SDK 搭一个生产级 A2A Agent

下面我们亲手写一个可离线运行的 A2A Agent：DataAnalyst Agent。它接收一个 CSV 文件（FilePart），返回描述性统计与异常行检测（DataPart + TextPart）。代码不依赖任何外部 API，你 clone 下来就能跑。

4.1 安装依赖

pip install a2a-sdk "uvicorn[standard]" fastapi
# a2a-sdk 是官方 Python SDK（仓库 google-a2a/a2a-python）

4.2 定义 Agent Card

新建 agent_card.json：

{
  "protocolVersion": "0.3.0",
  "name": "DataAnalyst Agent",
  "description": "接收 CSV，返回描述性统计与异常行检测",
  "url": "http://localhost:9999/",
  "version": "1.0.0",
  "capabilities": { "streaming": true, "pushNotifications": true },
  "authentication": { "schemes": ["bearer"] },
  "defaultInputModes": ["text/csv"],
  "defaultOutputModes": ["application/json", "text/markdown"],
  "skills": [
    {
      "id": "csv_summary",
      "name": "CSV 描述性统计",
      "description": "对数值列计算均值/中位数/分位数/缺失率，并标记异常行",
      "tags": ["data", "etl", "statistics"],
      "examples": ["分析这份销售数据，找出异常订单"]
    }
  ]
}

4.3 实现 AgentExecutor（核心业务逻辑）

A2A SDK 要求我们实现一个 AgentExecutor，把「收到消息 → 产出 Task/Artifact」的逻辑填进去：

# executor.py
import base64
import csv
import io
import json
import statistics
from typing import AsyncIterable

from a2a.server.agent_execution import AgentExecutor
from a2a.server.agent_execution.context import RequestContext
from a2a.server.events import EventQueue
from a2a.types import (
    Part,
    TextPart,
    DataPart,
    FilePart,
    TaskArtifactUpdateEvent,
    TaskStatusUpdateEvent,
    TaskState,
    Message,
)


def _parse_csv_from_message(message: Message) -> tuple[list[str], list[dict]]:
    """从 user message 的 FilePart 里取出 CSV 文本并解析。"""
    csv_text = None
    for part in message.parts:
        if isinstance(part.root, FilePart):
            fp = part.root.file
            raw = base64.b64decode(fp.bytes) if fp.bytes else b""
            csv_text = raw.decode("utf-8")
    if not csv_text:
        raise ValueError("未找到 CSV 文件 Part")
    reader = csv.DictReader(io.StringIO(csv_text))
    rows = list(reader)
    columns = reader.fieldnames or []
    return columns, rows


def _analyze(columns: list[str], rows: list[dict]) -> dict:
    """纯 Python 的描述性统计 + IQR 异常检测（不依赖 pandas）。"""
    report: dict = {"rows": len(rows), "columns": columns, "numeric": {}, "anomaly_rows": []}
    for col in columns:
        try:
            values = [float(r[col]) for r in rows if r[col] not in (None, "",)]
        except (ValueError, TypeError):
            continue  # 非数值列跳过
        if not values:
            continue
        q1, q3 = (
            statistics.quantiles(values, n=4)[0],
            statistics.quantiles(values, n=4)[2],
        )
        iqr = q3 - q1
        lower, upper = q1 - 1.5 * iqr, q3 + 1.5 * iqr
        report["numeric"][col] = {
            "mean": round(statistics.mean(values), 4),
            "median": statistics.median(values),
            "min": min(values),
            "max": max(values),
            "missing_rate": round(1 - len(values) / max(len(rows), 1), 4),
        }
        for idx, r in enumerate(rows):
            try:
                v = float(r[col])
                if v < lower or v > upper:
                    if idx not in report["anomaly_rows"]:
                        report["anomaly_rows"].append(idx)
            except (ValueError, TypeError):
                pass
    return report


class DataAnalystExecutor(AgentExecutor):
    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
        # 1. 取消息
        message = context.message
        if message is None:
            raise ValueError("缺少 message")
        # 2. 解析 + 分析
        columns, rows = _parse_csv_from_message(message)
        report = _analyze(columns, rows)

        # 3. 先推一个「进行中」状态（流式体验）
        await event_queue.enqueue_event(
            TaskStatusUpdateEvent(status={"state": TaskState.working}, final=False)
        )

        # 4. 产出结构化 Artifact（DataPart）
        data_part = Part(root=DataPart(data=report))
        await event_queue.enqueue_event(
            TaskArtifactUpdateEvent(
                artifact={"artifactId": "summary", "name": "summary", "parts": [data_part]},
                final=False,
            )
        )

        # 5. 再产出一份人读 Markdown 报告（TextPart）
        md = self._to_markdown(report)
        text_part = Part(root=TextPart(text=md))
        await event_queue.enqueue_event(
            TaskArtifactUpdateEvent(
                artifact={"artifactId": "report", "name": "report", "parts": [text_part]},
                final=True,
            )
        )

        # 6. 标记任务完成
        await event_queue.enqueue_event(
            TaskStatusUpdateEvent(status={"state": TaskState.completed}, final=True)
        )

    async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
        await event_queue.enqueue_event(
            TaskStatusUpdateEvent(status={"state": TaskState.canceled}, final=True)
        )

    @staticmethod
    def _to_markdown(report: dict) -> str:
        lines = [f"# 数据分析报告", f"- 行数：{report['rows']}", f"- 列：{', '.join(report['columns'])}", ""]
        for col, stat in report["numeric"].items():
            lines.append(f"## {col}")
            lines.append(f"- 均值 {stat['mean']} / 中位数 {stat['median']}")
            lines.append(f"- 范围 [{stat['min']}, {stat['max']}] / 缺失率 {stat['missing_rate']}")
        lines.append("")
        lines.append(f"**异常行（IQR 法）**：{report['anomaly_rows']}")
        return "\n".join(lines)

这段代码展示了 A2A 的核心范式：Executor 不直接返回结果，而是往 EventQueue 里「投喂」事件。框架会负责把这些事件序列化成 SSE 流（或同步聚合为 tasks/send 的 response）。这正是「流式优先」的设计——同一份代码，既能同步用，也能流式用。

4.4 启动 Server（FastAPI + Uvicorn）

# server.py
import json
from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.agent_execution.inmemory import InMemoryTaskStore
from a2a.types import AgentCard
from executor import DataAnalystExecutor

with open("agent_card.json", "r", encoding="utf-8") as f:
    card = AgentCard(**json.load(f))

# RequestHandler 把「任务如何存储」和「任务如何执行」解耦
handler = DefaultRequestHandler(
    agent_execution=DataAnalystExecutor(),
    task_store=InMemoryTaskStore(),  # 生产可换 RedisTaskStore 做多副本共享
)

app = A2AStarletteApplication(agent_card=card, http_handler=handler).build()

# uvicorn server:app --host 0.0.0.0 --port 9999

启动后：

• GET /.well-known/agent.json → 返回 Agent Card
• POST / + tasks/send → 同步执行
• POST / + tasks/sendSubscribe → SSE 流式

4.5 客户端：用官方 SDK 委派任务

# client.py
import asyncio
import base64
from a2a.client import A2AClient, A2ACardResolver
from a2a.types import (
    Message,
    Part,
    FilePart,
    TextPart,
    Role,
    SendMessageRequest,
)

CSV_BYTES = base64.b64encode(b"id,value\n1,10\n2,9\n3,2000\n4,-12\n").decode()

async def main():
    # 1. 发现：拉 Agent Card
    resolver = A2ACardResolver(url="http://localhost:9999")
    card = await resolver.get_agent_card()

    # 2. 建客户端
    client = A2AClient(card)

    # 3. 组装消息：一个 FilePart + 一个 TextPart
    message = Message(
        role=Role.user,
        parts=[
            Part(root=FilePart(file={
                "mimeType": "text/csv",
                "bytes": CSV_BYTES,
                "name": "orders.csv",
            })),
            Part(root=TextPart(text="分析这份订单数据，标记异常行")),
        ],
    )

    # 4. 同步发送（短任务用 tasks/send）
    req = SendMessageRequest(id="req-1", params={"message": message})
    resp = await client.send_message(req)
    print(json.dumps(resp.model_dump(), ensure_ascii=False, indent=2))

asyncio.run(main())

跑起来你会看到 artifacts 里既有 summary（DataPart 结构化统计），也有 report（TextPart Markdown）。同一个 Agent，Python 客户端、TypeScript 客户端都能调——这就是标准化的价值。

4.6 跨语言互操作：TypeScript 客户端直连

A2A 不依赖任何 Python 特性。下面这段 TypeScript（用官方 a2a-js）做同样的事，证明「跨框架、跨语言」不是口号：

// client.ts  —— 使用官方 a2a-js SDK
import { A2AClient } from "@a2a-js/sdk";

const CSV_B64 = Buffer.from("id,value\n1,10\n2,9\n3,2000\n4,-12\n").toString("base64");

async function main() {
  const client = new A2AClient("http://localhost:9999");

  const res = await client.sendMessage({
    id: "req-ts-1",
    params: {
      message: {
        role: "user",
        parts: [
          { kind: "file", file: { mimeType: "text/csv", bytes: CSV_B64, name: "orders.csv" } },
          { kind: "text", text: "分析这份订单数据，标记异常行" },
        ],
      },
    },
  });

  console.log(JSON.stringify(res, null, 2));
}

main();

你会发现：TypeScript 客户端无需知道 Python Agent 的存在，只凭一张 Agent Card 就完成了完整委托。这就是 A2A 想达成的「Agent 互联网」。

4.7 流式调用：看得到进度

把 4.5 的 client.send_message 换成 send_message_streaming，就能拿到 AsyncIterable 的事件流：

async for event in client.send_message_streaming(req):
    if event.root.kind == "status-update":
        print("状态:", event.root.status.state)
    elif event.root.kind == "artifact-update":
        print("收到 artifact:", event.root.artifact.get("name"))

对于「跑 5 分钟的大数据分析」，前端可以据此渲染进度条，而不是转圈圈干等。

五、生产级实战：推送通知、鉴权、可观测性、错误处理

能跑通 demo 和能上生产，中间隔着十条鸿沟。这一节讲 A2A 落地的四道关。

5.1 推送通知：提交即走，完成回调

同步/流式都要求客户端「挂着」连接。对长任务（比如「跑一夜的报表」），更好的模式是 推送通知：

# 客户端先注册一个 Webhook 回调地址
from a2a.types import PushNotificationConfig, TaskPushNotificationConfig

await client.set_task_push_notification(
    TaskPushNotificationConfig(
        id="task-9f3a",
        pushNotificationConfig=PushNotificationConfig(
            url="https://client.example.com/a2a/callback",
            authentication={"schemes": ["bearer"]},
        ),
    )
)

Agent 完成后会主动 POST 一个通知到这个 URL。生产要点：

• 双向认证：Webhook 必须带 token，否则任何人都能伪造「任务完成」。
• 幂等：客户端回调处理要做幂等（同一个 task 可能重推），用 task.id 做去重键。
• 超时兜底：若 30 分钟没收到推送，客户端用 tasks/get 主动轮询一次。

5.2 鉴权：企业级直接复用 OAuth2 / API Key

A2A 的认证写在 Agent Card 的 authentication.schemes 里。常见组合：

• API Key：最简单，适合内部服务。Authorization: Bearer <apikey>。
• OAuth2 / OIDC：跨组织协作的标准做法，Agent 之间走 client_credentials 换 token。
• mTLS：金融、政务等高安全场景，双向证书。

服务端用 FastAPI 中间件统一校验，不要把鉴权散落到 Executor 里：

# auth_middleware.py
from starlette.middleware.base import BaseHTTPMiddleware

class BearerAuthMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        if request.url.path == "/.well-known/agent.json":
            return await call_next(request)  # 发现接口可公开
        auth = request.headers.get("Authorization", "")
        if not auth.startswith("Bearer ") or not _valid(auth[7:]):
            return JSONResponse(status_code=401, content={"error": "unauthorized"})
        return await call_next(request)

5.3 错误处理：JSON-RPC 错误码 + TaskState

A2A 复用 JSON-RPC 2.0 的错误码体系（-32700 解析错误、-32600 非法请求、-32601 方法不存在、-32602 无效参数、-32603 内部错误），并叠加业务层失败：

# 业务失败：把 Task 标记为 failed，并附上可读原因
await event_queue.enqueue_event(
    TaskStatusUpdateEvent(
        status={"state": TaskState.failed, "message": "CSV 解析失败：缺少数值列"},
        final=True,
    )
)

生产级 Agent 必须区分三类失败：

1. 输入错误（400 类）：客户端传错格式 → 返回 input-required 或 failed，并告诉对方怎么改。
2. 瞬时错误（5xx 类）：下游超时 → 重试 + 指数退避，上游客户端用 tasks/get 续查。
3. 终态错误：逻辑不可恢复 → failed，写日志 + 告警。

5.4 MCP × A2A：一个 Agent 同时「对外委派、对内调工具」

真实生产 Agent 很少只做透传。典型的「对外的 A2A 壳 + 对内的 MCP 心」：

# 这个 A2A Agent 内部通过 MCP 调用一个 PostgreSQL 工具
class SalesAgentExecutor(AgentExecutor):
    async def execute(self, context, event_queue):
        # 1. 作为 A2A Server 接收委派
        query = _extract_text(context.message)
        # 2. 作为 MCP Client 调用内部工具（数据库/检索）
        async with MCPClient(server="postgres-mcp:8000") as mcp:
            rows = await mcp.call_tool("query_sales", {"sql": query})
        # 3. 把结果包成 A2A Artifact 回给上游 Agent
        await event_queue.enqueue_event(
            TaskArtifactUpdateEvent(artifact={"parts": [Part(root=DataPart(data=rows))]}, final=True)
        )
        await event_queue.enqueue_event(
            TaskStatusUpdateEvent(status={"state": TaskState.completed}, final=True)
        )

这就是 2026 年「MCP×A2A」标准架构：A2A 管「 Agent 之间」，MCP 管「Agent 与工具/数据之间」，两者正交、互不替代。

六、性能优化与选型建议

6.1 同步 vs 流式 vs 推送：怎么选

经验法则：默认流式，长任务上推送。

6.2 Task Store 必须外置

官方 demo 用 InMemoryTaskStore，但多副本部署时每个实例内存独立，任务状态会「找不着」。生产必须换成共享存储：

from a2a.server.agent_execution.redis import RedisTaskStore
handler = DefaultRequestHandler(
    agent_execution=DataAnalystExecutor(),
    task_store=RedisTaskStore(redis_url="redis://cache:6379/0"),
)

这样任意副本都能用 tasks/get 取到同一份 Task 状态，横向扩容无忧。

6.3 Executor 保持无状态

Executor 里不要存任务相关状态到实例变量。每个请求从 context.task_id 重新拉取。配合无状态 Executor + 外置 TaskStore，你可以像无状态 Web 服务一样随便扩缩容。

6.4 大文件走引用，别走 base64

FilePart 的 bytes 字段是 base64 内联，适合小文件。大 CSV/PDF 请用 file.uri 传一个可下载 URL，Agent 自己去拉，避免把几十 MB 塞进 JSON-RPC 报文把连接拖死。

6.5 何时不该用 A2A

• 两个 Agent 在同一个进程、同一套代码里 → 直接函数调用，别上协议。
• 只是「模型调个工具」→ 用 MCP，别用 A2A。
• 一次性、无复用价值的临时对接 → 直接 REST 可能更省事。

A2A 的甜区是：跨组织 / 跨框架 / 需要被「发现」和「复用」的 Agent 协作。

七、总结与展望

A2A 的出现，标志着 AI Agent 从「单人工具」走向「协作网络」。它的几个设计选择值得所有做基础设施的人学习：

1. 不发明轮子：全部建在 HTTP + JSON-RPC + SSE 上，学习成本为 0。
2. 黑盒契约：只约定交换什么，不侵入对方实现，这是跨厂商互操作的必要条件。
3. 模态协商：用 Part + inputModes/outputModes 让 Agent 自己商量格式，而不是硬编码。
4. 中立治理：捐给 Linux 基金会，解决了「凭什么信你」的信任问题。

到 2026 年，A2A 已与 MCP、AG-UI（Agent-UI 双向交互协议）一起，构成「Agent 互联网」的三块地基：MCP 连工具、A2A 连 Agent、AG-UI 连人。当一个招聘协调 Agent 能在 3 秒内拉起 Salesforce、ServiceNow 和背调三方的 Agent 并聚合结果时，我们才真正摸到了「多 Agent 系统」的门槛。

对于工程师，我的建议很直接：现在就把 A2A 写进你的 Agent 架构 checklist。 哪怕今天只有一个 Agent，明天它可能就要和别人的 Agent 握手。提前按 Agent Card 的方式暴露能力，是你未来进入「Agent 互联网」的入场券。

最后一句掏心窝的：协议永远比框架长寿。框架会换（LangChain → LlamaIndex → 你自己的循环），但 A2A 这种「Agent 之间说什么」的标准，一旦被 150+ 组织认下，就会像 HTTP 一样长期存在。早点吃透它，值。

参考资料（可溯源）

• Google 官方公告《Announcing the Agent2Agent Protocol (A2A)》，2025-04-09
• A2A Project（Linux Foundation）：github.com/a2aproject —— 协议规范、Python/JS/Java SDK
• 清华大学《2026 年 Agent-to-Agent (A2A) 研究报告》，2026-06
• 官方 Python SDK：google-a2a/a2a-python；官方 JS SDK：a2aproject/a2a-js