A2A 协议讲透:Google 开源、Linux 基金会托管的 Agent 互通标准,跑一个最小 Demo

一、A2A 到底解决了什么问题

先说背景。一个团队用 LangGraph 做了"行程规划 Agent"，另一个团队用 CrewAI 做了"机票预订 Agent"，第三个团队用 AutoGen 做了"酒店比价 Agent"。现在产品想串起来：用户说一句"五一去成都玩三天"，规划 Agent 要调用机票、酒店两个 Agent 协同完成。

过去你只有两条路：要么把三个 Agent 全部重写到同一个框架里（成本爆炸）；要么把对方 Agent 当成 HTTP API 强行包一层（丢失了"Agent 是状态机+流式输出+多轮澄清"这些核心特征）。

A2A 的目标就是补上"Agent 之间相互发现、对话、协作"这一层。它把三件事标准化了：

注意 A2A 把对方 Agent 视为不透明的黑盒——你不需要知道对方用什么模型、什么框架、什么 Prompt。只要它实现 A2A 的 HTTP 端点和 JSON 结构，就可以被任何 A2A 客户端调用。这是它和"把 Agent 包成普通 REST API"最本质的区别。

二、A2A、MCP、AG-UI 三协议的边界

这是面试和落地最常被问的问题，今天用一张表把边界钉死。

记一句话就够了：MCP 给 Agent 装手脚，A2A 让 Agent 组队，AG-UI 把 Agent 接到屏幕上。它们互相不冲突，可以同时用——一个生产级 Agent 系统大概率三个全要。

反过来要警惕的是：见到"Agent 互通"就上 A2A 是过度设计。如果你的对方 Agent 是自家团队、同一个进程里、不需要跨网络发现，那 LangGraph 的 subgraph 或者直接函数调用就够了，根本不用引一个 HTTP 协议进来。

三、最小可跑 Demo：用 a2a-sdk 起一个 HelloAgent

Google 官方维护了 Python SDK，包名 a2a-sdk，仓库在 a2aproject/A2A（之前叫 google/A2A，已迁移）。环境要求 Python 3.10+。

3.1 安装

pip install a2a-sdk uvicorn

3.2 服务端：定义 AgentCard 和 AgentExecutor

from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.server.tasks import InMemoryTaskStore
from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.types import AgentCard, AgentSkill, AgentCapabilities
from a2a.utils import new_agent_text_message
import uvicorn


class HelloExecutor(AgentExecutor):
    """最小执行器：把用户输入回显，并加一句问候"""

    async def execute(
        self, context: RequestContext, event_queue: EventQueue
    ) -> None:
        user_text = context.get_user_input()
        reply = f"Hello from A2A! 你刚才说：{user_text}"
        # 把回复封装成一条 agent message 推送给客户端
        await event_queue.enqueue_event(new_agent_text_message(reply))

    async def cancel(
        self, context: RequestContext, event_queue: EventQueue
    ) -> None:
        # 演示版本不支持取消，直接抛
        raise Exception("cancel not supported")


def build_app():
    skill = AgentSkill(
        id="hello",
        name="Hello Skill",
        description="回显用户输入并加问候",
        tags=["demo", "hello"],
        examples=["你好", "ping"],
    )
    card = AgentCard(
        name="HelloAgent",
        description="A2A 最小演示 Agent",
        url="http://localhost:9999/",
        version="0.1.0",
        default_input_modes=["text"],
        default_output_modes=["text"],
        capabilities=AgentCapabilities(streaming=True),
        skills=[skill],
    )
    handler = DefaultRequestHandler(
        agent_executor=HelloExecutor(),
        task_store=InMemoryTaskStore(),
    )
    return A2AStarletteApplication(
        agent_card=card, http_handler=handler
    ).build()


if __name__ == "__main__":
    uvicorn.run(build_app(), host="0.0.0.0", port=9999)

跑起来后访问 http://localhost:9999/.well-known/agent.json，浏览器会直接返回这张 Agent Card——这就是"被发现"的入口。任何兼容 A2A 的客户端拿到这个 URL 就能开始对话。

3.3 客户端：发起一次同步消息

import asyncio
from uuid import uuid4
import httpx
from a2a.client import A2ACardResolver, ClientFactory, ClientConfig
from a2a.types import Message, Part, TextPart, Role


async def main():
    base_url = "http://localhost:9999"
    async with httpx.AsyncClient(timeout=30) as http:
        # 1. 先抓 Agent Card
        card = await A2ACardResolver(
            httpx_client=http, base_url=base_url
        ).get_agent_card()
        print("发现 Agent：", card.name, "| skills:", [s.id for s in card.skills])

        # 2. 构造 client，发一条消息
        client = ClientFactory(ClientConfig(httpx_client=http)).create(card)
        msg = Message(
            role=Role.user,
            message_id=uuid4().hex,
            parts=[Part(root=TextPart(text="A2A 这玩意能跑吗"))],
        )
        async for event in client.send_message(msg):
            # event 里会带回 agent 的回复
            print("收到事件：", event)


asyncio.run(main())

跑通之后你会看到：客户端先 GET 了 /.well-known/agent.json 完成发现，然后 POST 到根路径 / 发送 JSON-RPC 请求 message/send，服务端走 HelloExecutor 把回复推回事件队列，最后客户端拿到 agent message。整个回路就是 A2A 最小闭环。

3.4 流式：把 message/send 换成 message/stream

把上面 capabilities.streaming 设为 True 后，客户端可以走 SSE 流式订阅。底层是一个长连接 HTTP 响应，每一条 data: 都是一条 JSON-RPC 响应，包含 task 状态变更或新产物 chunk。这一点和 AG-UI 的事件流非常像，但 A2A 的事件 Schema 更聚焦于"任务状态机"，而不是"前端 UI 渲染"。

四、四个生产踩坑（来自真实项目）

4.1 AgentCard 不是写完就完事，要做版本灰度

Card 里的 version 字段不是装饰品。生产中 Skill 增删一定会发生，但客户端缓存 Card 是常态。一旦你把 skill id 改名又不升版本，下游 Agent 会拿着旧描述去调用，要么参数不匹配，要么走到错误分支还以为是自己 Bug。

4.2 Task 状态机要老老实实走，不要"一把梭"

新手最容易写出来的代码是：执行器里跑完所有逻辑，最后一次性 push 一条 message 完事。这样客户端体验上就是"提交后无任何反馈，几十秒后突然出结果"，体验糟糕，长任务还容易撞上代理超时。

4.3 Nginx / 网关默认会"吃掉"SSE 流

这是上次讲 AG-UI 时也踩过的坑，A2A 流式同样适用。Nginx 默认开启响应缓冲，SSE 数据会被攒在缓冲区里，客户端看上去就是几秒、几十秒一阵抖动地"喷"出来，破坏流式体验。

# nginx.conf 关键配置
location / {
    proxy_pass http://a2a_upstream;
    proxy_http_version 1.1;
    proxy_set_header Connection "";
    proxy_buffering off;          # 关掉响应缓冲
    proxy_cache off;
    proxy_read_timeout 3600s;     # SSE 需要长连接
    chunked_transfer_encoding on;
}

另外别忘了在响应里设置 X-Accel-Buffering: no，对部分网关是兜底开关。云厂商的 ALB/CLB 也要单独检查空闲超时，常见的 60 秒默认值能把任何稍长的 Task 全部砍断。

4.4 鉴权：别用 Bearer 裸跑公网

A2A 把"双方都可能是外部 Agent"作为默认假设，所以鉴权是协议一等公民。AgentCard 的 securitySchemes 字段允许你声明 OAuth2、API Key、mTLS 等多种方案。落地时记两条原则：

① 跨组织协作必须 mTLS 或 OAuth2 client_credentials，不要用静态 Bearer Token。Token 一旦泄露，对方拿着你的 Card URL 可以无限制调起任意 skill，账单会非常刺激。

② 同组织内部可以走短期 Token + 调用配额。配额尤其重要——A2A 让 Agent 互调变得太顺滑了，一不小心就会写出 A → B → C → A 的循环依赖，几分钟把 LLM 配额烧穿。生产里务必给每个 Agent 加最大递归深度和单 Task token 预算。

五、什么时候不该用 A2A

讲完正面，必须再泼一盆冷水。下面三种场景，A2A 是负优化：

A2A 的甜蜜点很清晰：跨团队、跨框架、跨网络的多 Agent 协作。如果你正在做的是企业内部"Agent 中台"——一边接各业务部门的 LangGraph/CrewAI/AutoGen 实现，一边对外提供统一编排——A2A 几乎是目前唯一标准答案。

六、写在最后

从 MCP（2024/11）到 A2A（2025/04）再到 AG-UI（2025/05），整个 Agent 工程栈在一年内补完了三个最关键的接口层。现在站在 2026 年回头看，这三个协议像 HTTP/SQL/REST 一样，正在成为 Agent 时代的"水电煤"——你不需要每个都精通，但作为后端工程师/AI 工程师，至少要清楚它们各自的边界、典型用法和常见坑。

下一篇我们会继续往深里挖，挑一个具体的 A2A 生产案例：一个 LangGraph 主 Agent 通过 A2A 调用三个不同框架（CrewAI、AutoGen、纯 OpenAI SDK）的子 Agent，串起一条端到端的复杂任务。明晚见。