A2A 协议实战:打通多智能体协作的开放标准

当企业里的 AI 智能体越来越多,一个绕不开的问题浮出水面:A2A 协议(Agent2Agent Protocol,谷歌于 2025 年开源的多智能体互操作开放标准)正是为解决”智能体之间如何互相发现、委派任务、交换结果”而生。本文从概念一路写到可运行代码,带你吃透这套正在成为行业事实标准的协议,并把它和 MCP、ANP 彻底区分开。

一、为什么需要 A2A:多智能体的”巴别塔”困境

设想这样一个场景:你的公司已经上线了”报销 Agent””法务审核 Agent””客服 Agent”,它们分别由不同团队、用不同框架、跑在不同服务里。当老板想要一个”差旅自动报销并同步法务留痕”的流程时,三个 Agent 却谁也不认识谁——没有统一的服务发现,没有标准的任务描述,没有约定的结果格式。这就是多智能体的”巴别塔”困境。

过去我们靠硬编码的 HTTP 接口把 Agent 串起来,一旦对方升级字段就集体崩溃。从零手写 AI Agent 时我们关注的是”单个 Agent 怎么思考”,而当规模来到十几个甚至上百个 Agent,真正的瓶颈变成了”Agent 与 Agent 之间怎么协作”。A2A 就是为这件事设计的:它定义了一套与具体框架无关、与传输协议无关、只关心”能力声明 + 任务生命周期”的轻量协议。

二、A2A 与 MCP、ANP 到底什么关系

很多人把 A2A、MCP、ANP 混为一谈,其实它们的定位完全不同。一句话记住:MCP 解决 Agent 调用工具,A2A 解决 Agent 调用 Agent,ANP 解决跨组织 Agent 的开放互联。

维度A2A(Agent2Agent)MCP(模型上下文协议)ANP(智能体网络协议)
发起方Google 主导Anthropic 主导开源社区
核心场景Agent 间协作、任务委派Agent 调用工具/数据源跨信任域的 Agent 互联
关键抽象AgentCard / Task / ArtifactTool / Resource / PromptDID / 描述协议
类比企业内部的”工单系统”函数的”标准 RPC”跨公司的”外交协议”

实践中三者并不互斥:一个超级 Agent 完全可以用 MCP 去连数据库和搜索引擎,同时用 A2A 把”写代码””做图””发邮件”三个子 Agent 编排起来。理解这一点,架构就清晰了。

三、A2A 的四大核心对象

1. AgentCard:能力的”身份证”

每个 A2A Agent 都在一个固定路径(通常是 /.well-known/agent.json)暴露一张 AgentCard,声明自己能做什么、支持哪些输入输出模式、需要什么鉴权。这是服务发现的入口。

2. Task:一次协作的基本单位

任务有状态机:submitted → working → input-required → completed / failed。需要多轮交互的任务会进入 input-required,等对方补充信息后再继续。

3. Message 与 Artifact:对话与产物

Message 是往返的对话单元(文本、文件、结构化数据),Artifact 是任务最终交付的产物。两者都用多部分(multipart)设计以支持大文件和流式。

四、五分钟跑通:用 Python 起一个最小 A2A Agent

下面用 FastAPI 起一个”翻译 Agent”,它暴露 AgentCard,并响应 tasks/send。代码刻意保持精简,突出 A2A 的数据契约而非某个 SDK 的具体版本。

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

# 1) 暴露 AgentCard 供发现
@app.get("/.well-known/agent.json")
def agent_card():
    return {
        "name": "翻译 Agent",
        "description": "把文本翻译为指定语言",
        "url": "http://localhost:8001",
        "capabilities": {"streaming": False},
        "skills": [{"id": "translate", "name": "翻译",
                    "inputModes": ["text/plain"],
                    "outputModes": ["text/plain"]}],
    }

class TaskSend(BaseModel):
    id: str
    message: dict

@app.post("/tasks/send")
def tasks_send(task: TaskSend):
    text = task.message["parts"][0]["text"]
    lang = task.message.get("metadata", {}).get("target", "en")
    # 这里用占位翻译,真实场景接 LLM
    translated = f"[{lang}] {text}"
    return {
        "id": task.id,
        "status": {"state": "completed"},
        "artifacts": [{"parts": [{"text": translated}]}],
    }

启动后访问 /tasks/send 即可拿到结构化结果,这正是 A2A 想统一的”任务进出”形态。

五、客户端如何发现并派发任务

调用方先拉 AgentCard 了解能力,再按契约发任务。下面的客户端演示了”先发现、后委派”的标准两步。

import requests

BASE = "http://localhost:8001"
# 第 1 步:发现对端能力
card = requests.get(f"{BASE}/.well-known/agent.json").json()
print("对方支持技能:", [s["id"] for s in card["skills"]])

# 第 2 步:派发一个任务
payload = {
    "id": "task-1001",
    "message": {
        "role": "user",
        "parts": [{"text": "你好,世界"}],
        "metadata": {"target": "en"},
    },
}
resp = requests.post(f"{BASE}/tasks/send", json=payload).json()
print("翻译结果:", resp["artifacts"][0]["parts"][0]["text"])

六、多智能体编排实战:任务拆解与派发

真正的价值在多 Agent 协同。下面这个”主管 Agent”把一句需求拆给”翻译 Agent”和”摘要 Agent”,再汇总。它复用了上一节的调用逻辑,只是多了一步路由。

def dispatch(agents: dict, task_text: str) -> dict:
    results = {}
    for name, base in agents.items():
        card = requests.get(f"{base}/.well-known/agent.json").json()
        # 按 skill 名称做简单路由
        skill = card["skills"][0]["id"]
        payload = {"id": f"{name}-1",
                   "message": {"role": "user",
                               "parts": [{"text": task_text}]}}
        r = requests.post(f"{base}/tasks/send", json=payload).json()
        results[name] = r["artifacts"][0]["parts"][0]["text"]
    return results

agents = {"translate": "http://localhost:8001",
          "summary":   "http://localhost:8002"}
print(dispatch(agents, "把这段产品说明翻译并总结成三句话"))

这段不到二十行的代码,已经具备”服务发现 + 技能路由 + 结果聚合”的雏形,正是 LangChain Agent 这类框架在底层悄悄做的事,而 A2A 把它标准化、跨框架化了。

七、与 MCP 协同:边界要划清

落地时最容易踩的坑是”什么都往 A2A 里塞”。请记住分界:凡是要访问一个具体工具或数据源(查库、读文件、调 API),走 MCP;凡是需要另一个会思考的 Agent 来接力,走 A2A。例如”翻译 Agent”内部用 MCP 连术语库,对外则通过 A2A 被主管 Agent 调用。这样既保证了工具调用的标准化,又让 Agent 协作不被框架绑架。

在评估这类多 Agent 系统时,也别忘了用 AI Agent 评测 的方法量化协作质量,而不是只看单 Agent 的聪明程度。

八、生产落地的三条底线

鉴权:AgentCard 之外必须验身份

公开暴露 AgentCard 没问题,但 /tasks/send 必须加 OAuth2 / mTLS 之类的鉴权,否则等于把内部能力裸奔到公网。

可观测:给每个 Task 一个 traceId

多跳调用一旦出错极难定位,建议把 A2A 的任务 id 透传到日志与链路追踪系统,配合 Function Calling 的调用日志一起看。

超时与退避:跨 Agent 调用默认要短

Agent 间网络跳数多,务必设置整体超时与指数退避,避免一个慢 Agent 拖垮整条编排链路。

九、两种通信模式:同步请求与流式推送

A2A 并不强制传输方式。最简单的情况,一次 tasks/send 走同步 HTTP 请求/响应,适合几秒内就能返回的任务。但对于”生成一份五十页行业报告”这类长任务,同步调用必然超时。此时应使用 tasks/sendSubscribe 走 SSE(Server-Sent Events)流式返回中间状态,或者采用”任务创建 + 主动推送”的异步模式:调用方先拿到 task id,Agent 完成后通过预先登记的 webhook 把 Artifact 推回来。三种模式的核心取舍是延迟敏感度与任务时长——选错就会要么白白占着连接,要么永远拿不到最终结果。

在流式场景下,建议给每个增量 Artifact 都带上递增序号,客户端据此做断点续收;一旦网络抖动断开,用 task id 重新订阅即可,无需从头再来。这和 Function Calling 流式处理里强调的”可恢复”思路一脉相承。落地时把传输模式写进 AgentCard 的 capabilities 字段,调用方就能在一开始就知道对方支持哪种协作方式,避免运行时才发现不支持流式而整体降级。

十、选型清单:什么时候该上 A2A

最后给一张务实的判定表:如果你的智能体数量 ≤ 2,硬编码调用更省事;如果智能体之间需要”思考接力”且来自不同框架/团队,A2A 的收益立刻显现;如果还要跨越公司边界,则叠加 ANP 这类开放互联协议。不要为了用而用——A2A 的价值在于”解耦”,当你开始为 Agent 之间的耦合而痛苦时,它就是对症的药。

结语

A2A 协议把”多智能体怎么协作”这件过去靠私拉硬编码的事,变成了可发现、可编排、可替换的标准能力。它不取代 MCP,而是和 MCP、ANP 一起,构成智能体时代的”网络协议栈”。如果你的 Agent 数量已经超过三个,现在就是引入 A2A 理清协作边界的最佳时机。

上一篇 《Pod Evicted 排查实战:K8s 资源驱逐与磁盘压力定位》
下一篇 Qwen 本地部署实战:通义千问私有化与 API 调用
最新评论
Docker 网络模式详解:bridge/host 生产选型 - FSData
Docker 网络模式详解:bridge/host 生产选型 - FSData
8月20日
[…] 注意最后那条:把端口绑定到 127.0.0.1,再由宿主机上的 Nginx 反代进来,比直接 -p 80:80 暴露安全得多,也方便统一做
Vue3 + TypeScript 工程化:类型安全的前端架构 - FSData
Vue3 + TypeScript 工程化:类型安全的前端架构 - FSData
8月20日
[…] Pinia 的工程化用法,可以参考我们之前的实战文章 《Vue3 组合式 API + Pinia:状态管理从入门到实战》,那里覆盖了 Sto
2026 年最值得装的 10 个 VS Code 插件(AI/前端/后端/效率全覆盖) - FSData
2026 年最值得装的 10 个 VS Code 插件(AI/前端/后端/效率全覆盖) - FSData
8月19日
[…] 状态,点一下就能起停容器,配合我们讲过的 Docker Compose 编排 […]
Redis 缓存设计:穿透、击穿、雪崩与最佳实践 - FSData
Redis 缓存设计:穿透、击穿、雪崩与最佳实践 - FSData
8月19日
[…] 缓存三大坑的本质都是”缓存失效导致请求穿透到数据库”。记住口诀:穿透用布隆、击穿用锁、雪崩用错峰。配合合理的过期策略
GitHub Actions 实战:从零搭建 CI/CD 流水线 - FSData
GitHub Actions 实战:从零搭建 CI/CD 流水线 - FSData
8月19日
[…] Docker 镜像部署。结合我们之前写的 Docker Compose 一键编排实战,可以在 CI […]