当 Agent 学会「找 Agent 帮忙」：Google A2A 协议深度实战，一次把多智能体协作讲透（2026）

关键词：A2A、Agent2Agent、多智能体协作、MCP、Agent Card、a2a-sdk、智能体编排
适用版本：A2A Protocol 1.0 / a2a-sdk 1.0（2026 年正式版，已捐赠给 Linux Foundation）

如果 2025 年是「单个智能体（Agent）元年」，那么 2026 年最值得开发者盯紧的信号，是智能体之间开始说上了「通用语」。

过去一年我们见证了 MCP（Model Context Protocol）把「Agent ↔ 工具/资源」标准化了：你的 Agent 今天能连数据库、明天能调搜索、后天能操作文件系统，全部走同一套协议。但一个更棘手、也更本质的问题一直没被解决——

Agent 和 Agent 之间，还是一座座孤岛。

你有一个写代码的 Agent、一个做安全扫描的 Agent、一个查数据的 Agent，它们各自是基于不同框架（LangChain、CrewAI、自研 Runtime）、不同语言（Python、Go、TypeScript）写的，互相完全不知道对方存在，更不知道怎么安全地「拜托对方干一件事」。于是乎，多智能体系统落地时，80% 的工程成本都花在了「写一堆脆弱的、点对点的胶水代码」上。

Google 在 2025 年提出的 A2A（Agent2Agent）协议，要解决的正是这件事。到了 2026 年，A2A 已经正式捐赠给 Linux Foundation、发布 1.0 规范、并提供 Python / TypeScript / Java / Go / C# / Rust 全套官方 SDK。它和 MCP 不是竞争关系，而是互补：MCP 让 Agent 会「用工具」，A2A 让 Agent 会「找 Agent 帮忙」。

这篇文章我会从背景讲到核心概念、从协议架构讲到可运行的代码实战，最后聊性能优化与工程化落地。所有代码基于官方 a2a-sdk 1.0，clone 下来就能跑。

一、背景：为什么 Agent 之间需要一门「通用语」

1.1 三个绕不开的现实约束

异构（Heterogeneous）。 真实企业里，采购的风控 Agent 是 Java 写的，自研的客服 Agent 是 Python 写的，边缘网关上的调度 Agent 是 Rust 写的。它们永远不会统一技术栈。

不透明（Opaque）。 你调用一个远程 Agent 时，根本不应该关心它内部是 LLM 还是规则引擎、用了什么 Prompt、跑了什么工具。你只需要知道：它能做什么、怎么把任务交给它、它怎么把结果还给你。

需要企业级安全。 跨团队的 Agent 协作，必然涉及身份认证、授权、审计。不能因为「都是 AI」就放弃这些。

这三点，正是 A2A 设计时的硬约束。它刻意站在「成熟的 Web 标准」肩膀上：传输用 HTTP，消息格式用 JSON-RPC 2.0，流式用 SSE，数据用结构化 JSON。换句话说，A2A 没有发明新轮子，而是把 Agent 协作「翻译」成了企业 IT 栈本来就懂的语言。

1.2 一个有用的历史类比

• Agent Card 之于 Agent，约等于 robots.txt / OpenAPI 文档 之于网站：它是一张公开「名片」，告诉外界「我是谁、我会什么、怎么联系我」。
• A2A 协议 之于 Agent 协作，约等于 HTTP 之于 Web 服务：它定义了请求怎么发、响应怎么回、错误怎么表达。
• MCP 之于 Agent，约等于 JDBC / 驱动 之于应用：它定义 Agent 怎么连上「工具、资源、数据源」。

所以当你听到「Agent 互联网（Agentic Web）」这种说法时，脑子里应该浮现的，就是 MCP 负责「接工具」，A2A 负责「连 Agent」，两者拼起来才是完整的图景。官方规范（Appendix B）也明确写道：MCP connects agents to tools and data; A2A connects agents to agents.

1.3 它和 MCP 到底怎么分工（一张图）

┌──────────────────────────────────────────────────────────────┐
│                      用户 / 上层业务                           │
└───────────────┬──────────────────────────────────────────────┘
                │  A2A（Agent ↔ Agent）
                ▼
┌─────────────────────────┐         A2A          ┌─────────────────────────┐
│   编排 Agent（Orchestrator）│ ─────────────────► │   专家 Agent（Specialist） │
│   技术负责人 / 路由器        │ ◄───────────────── │   代码评审 / 安全 / 数据    │
└───────────────┬───────────┘                    └───────────┬─────────────┘
                │  MCP（Agent ↔ 工具/资源）                     │  MCP
                ▼                                               ▼
        ┌──────────────┐                            ┌──────────────────┐
        │ 搜索 / 日历 / │                            │ 数据库 / 内部 API │
        │ 代码仓库 / CLI │                            │ / 文件系统        │
        └──────────────┘                            └──────────────────┘

记住这张图，后面所有代码都是在这张图的某个格子里落地的。

二、核心概念：A2A 的四个「一等公民」

A2A 的数据模型非常克制，真正需要你记住的只有四个对象：AgentCard、Task、Message/Part、Artifact。

2.1 AgentCard —— 智能体的「名片」

AgentCard 是一个 JSON 对象，通常托管在 https://<base-url>/.well-known/agent.json。它回答了三个问题：

1. 你是谁？ name、description、version、provider。
2. 你会什么？ skills 数组，每一项是一个 AgentSkill（id、name、description、tags、examples、input_modes、output_modes）。
3. 怎么联系你？ capabilities（是否支持流式、推送通知、扩展卡）、supported_interfaces（协议绑定 + 地址 + 协议版本）、auth（安全方案）。

下面是一张真实的 AgentCard（线上 wire 格式，节选）：

{
  "name": "Code Review Agent",
  "description": "接收 Python 源码，返回结构化静态评审意见",
  "version": "1.0.0",
  "default_input_modes": ["application/json"],
  "default_output_modes": ["application/json"],
  "capabilities": {
    "streaming": true,
    "pushNotifications": false,
    "extended_agent_card": false
  },
  "supported_interfaces": [
    {
      "protocol_binding": "JSONRPC",
      "url": "http://127.0.0.1:9999",
      "protocol_version": "1.0"
    }
  ],
  "skills": [
    {
      "id": "python_code_review",
      "name": "Python 代码评审",
      "description": "对给定 Python 源码做静态评审，返回 findings",
      "tags": ["code-review", "python", "static-analysis"],
      "examples": ["{\"code\": \"def f():\\n    print(1)\"}"],
      "input_modes": ["application/json"],
      "output_modes": ["application/json"]
    }
  ]
}

这里有个很巧妙的设计：Public Card 与 Extended Card。extended_agent_card: true 的 Agent 会额外暴露一张「认证后才可见」的扩展卡——上面可以有更高权限的技能、更敏感的能力。未登录用户只看到公开卡，登录后才看到全量。这一招直接把「能力最小化暴露」做进了协议层。

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

A2A 里每一次「委托—执行」都是一个 Task。它的生命周期是协议里最值得琢磨的部分：

 submitted ──► working ──► input-required ──► working ──► ... ──► completed
                │                                             │
                └─────────────────────────────────────────────┘
                │                                  └──► failed
                └──► canceled

• submitted：任务刚建好。
• working：远程 Agent 正在干。
• input-required：信息不够，反过来问你（多轮对话就靠它）。
• completed / failed / canceled：终态。

两个关键 ID 一定要分清：

• context_id：一次「会话 / 多轮对话」的标识。多轮澄清时，context_id 不变。
• task_id：单次任务的标识。

A2A 原生支持长时任务——几天跑完的数据 pipeline 和几毫秒的查询，走的是同一套状态机，区别只在于你用「流式 + 推送通知」还是「轮询 tasks/get」。

2.3 Message / Part / Artifact —— 内容的载体

• Message：一次发言，有 role（user 或 agent）、parts[]、context_id、task_id。
• Part：内容最小单元，三种类型：
  • TextPart：纯文本（text）。
  • FilePart：文件，用 url 或内联 file_with_bytes（raw）。
  • DataPart：结构化数据（data，任意 JSON-serializable 对象）。这是 Agent 之间交换「机器可读结果」的主力，比如把「评审结果」直接以 JSON 传过去，而不是塞进一段自然语言里让对方再解析。
• Artifact：任务的最终「成果物」，可包含多块 Part，且支持流式增量（append）。

一句话总结：人话用 TextPart，文件用 FilePart，机器之间传结构化结果用 DataPart，最终交付物叫 Artifact。

2.4 三个角色

• User：最终发起者（人或上游服务）。
• Client Agent：发起委托的一方（「我需要有人帮我评审代码」）。
• Remote Agent：真正执行的一方（「代码评审 Agent」）。

三、架构分析：A2A 到底是怎么跑起来的

3.1 协议绑定（Protocol Bindings）：同一套语义，三种 transport

A2A 1.0 定义了**功能等价（Functionally Equivalent）**的三种绑定，你可以按场景任选：

核心方法就这几个，记住它们就够用了：

• message/send：发一条消息，触发（或继续）一个 Task，同步拿结果。
• message/stream：同上，但用 SSE 流式返回状态与产物增量。
• tasks/get：按 task_id 拉任务当前状态（长任务轮询用）。
• tasks/list：按 context_id 列出该会话下的所有任务。
• tasks/cancel：取消任务。
• tasks/pushNotificationConfig/set：注册一个 webhook，任务完成/更新时由服务端主动回调（避免客户端傻轮询）。
• agentCard/getExtended：拿认证后的扩展卡。

3.2 服务端骨架：Executor + EventQueue + Handler

服务端运行模型非常优雅，核心就四块：

你的业务逻辑(AgentExecutor)
        │ 把 TaskStatusUpdateEvent / TaskArtifactUpdateEvent 入队
        ▼
   EventQueue（事件总线）
        │
        ▼
DefaultRequestHandler（编排：建 Task、管状态、调 Executor）
        │
        ▼
   TaskStore（状态持久化：InMemory / Postgres / MySQL / SQLite）
        │
        ▼
  ASGI App（Starlette / FastAPI）+ 协议路由

你作为开发者，只需要实现 AgentExecutor.execute(context, event_queue)：在里面把任务状态变化和产物「吐」到 EventQueue，剩下的序列化、传输、流式、状态机维护，框架全包了。这就是为什么 A2A 的「上手成本」比想象中低得多。

3.3 发现机制（Discovery）

Agent 之间怎么「找到彼此」？A2A 给了三种标准路径：

1. 开放发现（Open Discovery）：直接读 /.well-known/agent.json。适合公开 Agent。
2. 筛选式发现（Curated / Registry）：从一个 Agent 注册中心拉清单。适合企业内部。
3. 私有发现（Private Discovery）：先认证，再拿 Extended Agent Card。适合高敏感能力。

这其实就是微服务时代「服务发现」在 Agent 世界的翻版，没有新概念，落地心智成本极低。

四、代码实战：从零搭一个多智能体协作系统

环境准备（Python 3.10+）：

pip install "a2a-sdk[http-server]"
# 或 uv add "a2a-sdk[http-server]"

下面我们做三件事，复杂度逐级递增：

1. 一个可运行的代码评审 Agent（纯 Python，零外部依赖，能直接跑）。
2. 流式进度 + 多轮澄清。
3. 一个多智能体编排器（TechLead），把多个专家 Agent 编排起来——这才是 A2A 的「杀手级」用法。

4.1 实战一：代码评审 Agent（服务端）

先写评审逻辑。为了让 demo 不依赖任何 LLM API Key，我们用标准库 ast 做真实的静态分析：

# code_review_executor.py
import ast

from a2a.helpers import (
    get_data_parts,
    get_message_text,
    new_data_part,
    new_task_from_user_message,
    new_text_message,
)
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.server.tasks import TaskUpdater
from a2a.types import TaskState


class CodeReviewAgent:
    """纯 Python 静态代码评审，无需任何外部 LLM 或 API。"""

    def review(self, source: str) -> dict:
        findings = []
        try:
            tree = ast.parse(source)
        except SyntaxError as e:
            return {"ok": False, "error": f"语法错误: {e}", "findings": []}

        for node in ast.walk(tree):
            if isinstance(node, ast.FunctionDef):
                if ast.get_docstring(node) is None:
                    findings.append({
                        "severity": "warning",
                        "line": node.lineno,
                        "rule": "missing-docstring",
                        "message": f"函数 `{node.name}` 缺少文档字符串",
                    })
                for stmt in node.body:
                    if (isinstance(stmt, ast.Expr)
                            and isinstance(stmt.value, ast.Call)
                            and isinstance(stmt.value.func, ast.Name)
                            and stmt.value.func.id == "print"):
                        findings.append({
                            "severity": "info",
                            "line": stmt.lineno,
                            "rule": "no-print-in-prod",
                            "message": f"第 {stmt.lineno} 行用了 print()，生产代码建议用日志",
                        })

        for i, line in enumerate(source.splitlines(), start=1):
            if len(line) > 100:
                findings.append({
                    "severity": "info",
                    "line": i,
                    "rule": "line-too-long",
                    "message": f"第 {i} 行超过 100 字符（{len(line)}）",
                })

        return {
            "ok": True,
            "findings": findings,
            "summary": {
                "total": len(findings),
                "warning": sum(1 for f in findings if f["severity"] == "warning"),
                "info": sum(1 for f in findings if f["severity"] == "info"),
            },
        }


class CodeReviewAgentExecutor(AgentExecutor):
    def __init__(self) -> None:
        self.agent = CodeReviewAgent()

    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
        # 1. 拿到或新建 Task（多轮续作时 context.current_task 已存在）
        if context.current_task:
            task = context.current_task
        else:
            task = new_task_from_user_message(context.message)
            await event_queue.enqueue_event(task)

        updater = TaskUpdater(
            event_queue=event_queue, task_id=task.id, context_id=task.context_id
        )

        # 2. 进入 working，先给调用方一个「已受理」信号
        await updater.update_status(
            state=TaskState.TASK_STATE_WORKING,
            message=new_text_message("正在解析 AST 并评审代码..."),
        )

        # 3. 从 DataPart 取代码（get_data_parts 直接返回 Python dict）
        data_parts = get_data_parts(context.message)
        if not data_parts:
            await updater.update_status(
                state=TaskState.TASK_STATE_FAILED,
                message=new_text_message("未收到代码数据（期望 DataPart）"),
            )
            return
        code = data_parts[0].get("code", "")
        result = self.agent.review(code)

        # 4. 以 DataPart 形式回传结构化结果（机器可读，无需对方再解析自然语言）
        await updater.add_artifact(
            parts=[new_data_part(data=result)],
            name="review-report",
        )

        # 5. 完成
        total = result.get("summary", {}).get("total", 0)
        await updater.update_status(
            state=TaskState.TASK_STATE_COMPLETED,
            message=new_text_message(f"评审完成，共发现 {total} 条建议"),
        )

    async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
        raise NotImplementedError("该 Agent 暂不支持取消")

注意第 3 步：get_data_parts(context.message) 返回的是已经转回 Python dict 的列表，所以 data_parts[0].get("code", "") 直接可用——这是 a2a-sdk 1.0 的贴心设计，底层是用 google.protobuf.json_format.MessageToDict 做的转换。

然后把它挂成 HTTP 服务：

# server.py
import uvicorn

from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.routes import create_agent_card_routes, create_jsonrpc_routes
from a2a.server.tasks import InMemoryTaskStore
from a2a.types import AgentCapabilities, AgentCard, AgentInterface, AgentSkill
from code_review_executor import CodeReviewAgentExecutor
from starlette.applications import Starlette

skill = AgentSkill(
    id="python_code_review",
    name="Python 代码评审",
    description="对给定 Python 源码做静态评审，返回结构化 findings",
    input_modes=["application/json"],
    output_modes=["application/json"],
    tags=["code-review", "python", "static-analysis"],
    examples=['{"code": "def f():\\n    print(1)"}'],
)

agent_card = AgentCard(
    name="Code Review Agent",
    description="一个用纯 Python 实现的多智能体协作者：接收源码，返回评审意见",
    version="1.0.0",
    default_input_modes=["application/json"],
    default_output_modes=["application/json"],
    capabilities=AgentCapabilities(streaming=True, extended_agent_card=False),
    supported_interfaces=[
        AgentInterface(
            protocol_binding="JSONRPC",
            url="http://127.0.0.1:9999",
            protocol_version="1.0",
        )
    ],
    skills=[skill],
)

request_handler = DefaultRequestHandler(
    agent_executor=CodeReviewAgentExecutor(),
    task_store=InMemoryTaskStore(),
    agent_card=agent_card,
)

app = Starlette(routes=[
    *create_agent_card_routes(agent_card),       # 暴露 /.well-known/agent.json
    *create_jsonrpc_routes(request_handler, "/"),  # 暴露 JSON-RPC 端点
])

if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=9999)

4.2 客户端：发现 + 调用

客户端不需要知道服务端用什么框架写的，只要 A2ACardResolver 能拿到它的 AgentCard：

# client.py
import asyncio
import httpx

from a2a.client import A2ACardResolver, ClientConfig, create_client
from a2a.helpers import new_data_message
from a2a.types import Role, SendMessageRequest
from google.protobuf.json_format import MessageToDict

SOURCE = '''
def add(a, b):
    return a + b

def main():
    print(add(1, 2))
'''


async def main():
    async with httpx.AsyncClient() as httpx_client:
        # 1. 发现：读对方的 AgentCard
        resolver = A2ACardResolver(
            httpx_client=httpx_client, base_url="http://127.0.0.1:9999"
        )
        card = await resolver.get_agent_card()
        print(f"发现 Agent: {card.name} - {card.description}")

        # 2. 建客户端（非流式）
        client = await create_client(
            agent=card, client_config=ClientConfig(streaming=False)
        )

        # 3. 用 DataPart 把代码结构化地发过去
        message = new_data_message({"code": SOURCE}, role=Role.ROLE_USER)
        request = SendMessageRequest(message=message)

        report = None
        async for chunk in client.send_message(request):
            if chunk.HasField("task") and chunk.task.artifacts:
                # 取第一个 Artifact 的第一个 DataPart，转回 dict
                report = MessageToDict(chunk.task.artifacts[0].parts[0].data)
        await client.close()

    print("评审结果:", report)


if __name__ == "__main__":
    asyncio.run(main())

跑起来：先 python server.py，再 python client.py，你会看到一段结构化 JSON 评审报告。这就是两个进程、用一套开放协议完成的「Agent 协作」——客户端完全不知道服务端是 ast 还是 LLM，服务端也完全不知道客户端是谁。

4.3 实战二：流式进度 + 多轮澄清

把 ClientConfig(streaming=True)，客户端就能一边等一边收进度。服务端只要多发几次状态事件即可：

# 在 execute 内，把「一次性返回」改成「边做边报」
await updater.update_status(
    state=TaskState.TASK_STATE_WORKING,
    message=new_text_message("步骤 1/3：解析 AST..."),
)
# ... 做第一步 ...
await updater.update_status(
    state=TaskState.TASK_STATE_WORKING,
    message=new_text_message("步骤 2/3：检查文档字符串与 print..."),
)
# ... 做第二步 ...

流式客户端读起来像这样（每个 chunk 是 StreamResponse）：

from a2a.helpers import get_stream_response_text

async for chunk in client.send_message(request):
    if chunk.HasField("status_update"):
        print("状态:", chunk.status_update.status.state,
              get_stream_response_text(chunk))
    elif chunk.HasField("artifact_update"):
        print("增量产物到达")
    elif chunk.HasField("task"):
        print("最终任务:", chunk.task)

多轮澄清则用 input-required 状态：当输入信息不足，服务端直接反问，客户端在同一个 context_id 下补发一条消息继续。因为 execute 开头有 if context.current_task: task = context.current_task，续作时框架会自动把历史 Task 塞进 context，你接着干就行：

# 信息不足时反问（演示多轮）
lang = (get_data_parts(context.message) or [{}])[0].get("language")
if not lang:
    await updater.update_status(
        state=TaskState.TASK_STATE_INPUT_REQUIRED,
        message=new_text_message("请补充 language 字段（python / go / ...）"),
    )
    return
# 拿到 language 后，继续真正的评审...

这就是 A2A 把「长对话 / 澄清」做成协议一等公民的价值：你不需要自己设计一套多轮状态机。

4.4 实战三：多智能体编排器（核心）

前面都是「一对一」。A2A 真正的威力在**「一对多编排」**——一个 TechLead（编排）Agent，发现并委托多个专家 Agent。关键洞察：编排器完全不关心专家 Agent 用什么框架/语言实现，只要它暴露合规的 AgentCard。

# orchestrator.py（示意：把多个专家 Agent 扇出编排）
import asyncio
import httpx

from a2a.client import A2ACardResolver, ClientConfig, create_client
from a2a.helpers import new_data_message
from a2a.types import Role, SendMessageRequest
from google.protobuf.json_format import MessageToDict

SUBAGENTS = {
    "code-review": "http://127.0.0.1:9999",
    "doc-gen":     "http://127.0.0.1:9998",
    "test-gen":    "http://127.0.0.1:9997",
}


class TechLead:
    def __init__(self):
        self.registry = {}  # tag -> (base_url, card)

    async def discover(self):
        """启动时拉取所有子 Agent 的卡片，建成「技能标签 -> 端点」索引。"""
        async with httpx.AsyncClient() as c:
            for name, url in SUBAGENTS.items():
                card = await A2ACardResolver(
                    httpx_client=c, base_url=url
                ).get_agent_card()
                for skill in card.skills:
                    for tag in skill.tags:
                        self.registry[tag] = (url, card)
        return self.registry

    async def delegate(self, tag, payload):
        url, card = self.registry[tag]
        client = await create_client(
            agent=card, client_config=ClientConfig(streaming=False)
        )
        msg = new_data_message(payload, role=Role.ROLE_USER)
        result = None
        async for chunk in client.send_message(SendMessageRequest(message=msg)):
            if chunk.HasField("task") and chunk.task.artifacts:
                result = MessageToDict(chunk.task.artifacts[0].parts[0].data)
        await client.close()
        return result

    async def run(self, request: dict):
        # 简单意图路由：按关键词匹配技能标签（生产可用 embedding 相似度）
        tags = [t for t in self.registry if t in request.get("intent", "")]
        # 扇出（fan-out）：并行委托多个专家 Agent
        results = await asyncio.gather(
            *(self.delegate(t, request) for t in tags)
        )
        return dict(zip(tags, results))

几个值得拍大腿的设计点：

• asyncio.gather 扇出：多个专家 Agent 并行干活，总耗时 ≈ 最慢的那个，而不是串行求和。A2A 的 Task 是异步原语，天然契合。
• 递归组合：TechLead 自己也可以再被包装成一个 A2A Agent 暴露出去，于是「编排器之上还能有编排器」。这才是「Agent 互联网」的 scalable 形态。
• 零耦合：把 doc-gen 从 Python 换成 Go 重写，只要它的 AgentCard 不变，TechLead 一行代码都不用改。

4.5 实战四：企业级能力（收尾不收工）

真实生产里，下面两块绕不开：

认证与授权。 AgentCard 里声明 securitySchemes（支持 apiKey、http（Basic/Bearer）、oauth2、openIdConnect、mutualTLS）。高敏感技能放进 Extended Agent Card，只有走完 OAuth2 拿到 token 后才能看到。协议层把「最小权限暴露」做进了标准。

推送通知（Push Notifications）。 对跑几小时的长任务，别让客户端傻轮询。服务端在 AgentCard 声明 pushNotifications: true，客户端通过 tasks/pushNotificationConfig/set 注册一个 webhook URL；任务有进展时，服务端主动 POST 回调。这把「长时任务」的工程成本直接打下来。

可观测性。 a2a-sdk[telemetry] 接入 OpenTelemetry，每次 message/send、每个 Task 状态变迁都有 trace，排障时不再两眼一抹黑。

五、性能与工程化优化

写能跑不难，写能扛量才见功力。A2A 落地时有几条很实在的优化路线：

1. TaskStore 后端要选对。
Demo 用 InMemoryTaskStore 没问题；但生产多副本部署时，内存态会在实例间不一致。a2a-sdk 提供 postgresql / mysql / sqlite 后端，把任务状态落到共享存储，水平扩容无压力。

2. 传输绑定看场景。
对外/浏览器/网关走 JSON-RPC 或 REST；内网高吞吐、Agent 之间疯狂互调就上 gRPC 绑定（强类型 + 二进制，延迟和吞吐都更优）。同一套语义，换绑定零改业务代码。

3. 流式降低「感知延迟」。
先回 working + 中间态，再回 artifact。用户/上游不会卡在「转圈」，体验质变。大产物用 append=True 分块推，避免一次性序列化巨型 JSON。

4. 幂等与续传。
协议层 context_id + task_id 是幂等键（spec §3.3.1）。网络抖动重试时，同 task_id 不会建出第二个任务；长任务中断后，用 tasks/get 按 task_id 恢复，不用从头跑。

5. 连接复用。
客户端用 httpx.AsyncClient 复用连接；create_client 建出来的 client 在多次调用间复用，别每次请求都新建。扇出时尤其明显。

6. 鉴权开销。
OAuth2 client_credentials 拿到的 token 要缓存，别每次委托都走一遍 token endpoint。

六、总结与展望

把今天的内容收个尾：

• MCP 解决「Agent 用工具」，A2A 解决「Agent 连 Agent」，两者拼成完整的智能体协作底座。A2A 站在 HTTP / JSON-RPC / SSE 这些成熟标准上，没有发明新轮子，企业落地心智成本很低。
• 四个核心对象记牢：AgentCard（名片）、Task（工单，带 submitted→working→input-required→completed 状态机）、Message/Part（内容，Text/File/Data 三型）、Artifact（成果物）。
• 开发者只写 AgentExecutor.execute，把状态与产物吐到 EventQueue，剩下的序列化、流式、状态机、传输全由框架接管。
• 编排器（Orchestrator）才是杀手级用法：发现专家 Agent → 按意图路由 → asyncio.gather 扇出 → 聚合结果，且完全不耦合对方实现。
• 工程化三件套：共享 TaskStore、按场景选协议绑定、OTel 可观测。

2026 年 A2A 进入 1.0 + Linux Foundation 治理，意味着它不再是「Google 的一家之言」，而是行业共建的开放标准。对普通开发者的建议很朴素：新做的 Agent，默认出门就带一张 AgentCard；内部的系统，先把「编排器 + 专家 Agent」的边界想清楚。当你的每个 Agent 都「即插即用、可被其他 Agent 发现与委托」时，你就已经提前站在了「Agent 互联网」的入口。

最后留一个问题给读者：当 Agent 能自由地找 Agent 帮忙，信任边界、权限闸门、卡片签名（防伪造的 AgentCardSignature）就成了新的安全主战场——这恰恰是下一个值得深挖的方向。

本文代码示例基于 a2a-sdk 1.0（对应 A2A Protocol 1.0）。官方仓库：github.com/a2aproject，文档：a2a-protocol.org。建议配合 DeepLearning.AI 的《A2A: The Agent2Agent Protocol》短课程一起食用。