课程0基础Agent开发课 / 前沿方向 / A2A与ANP协议-多Agent时代的通信标准
— 14 min read

A2A与ANP协议-多Agent时代的通信标准

> **时效说明**:本文内容以 2026 年 3 月为基准。A2A 协议由 Google 于 2025 年 4 月发布,ANP 协议处于早期阶段。

A2A 与 ANP 协议:多 Agent 时代的通信标准

时效说明:本文内容以 2026 年 3 月为基准。A2A 协议由 Google 于 2025 年 4 月发布,ANP 协议处于早期阶段。

1.1 Agent 间通信的挑战:为什么需要专门的协议

A2A协议通信架构图
A2A 协议标准化通信架构——有协议的 Agent-A-to-Agent-B 通信(右)与无协议混乱状态(左)的对比

在理解 A2A 和 ANP 之前,先理解为什么 Agent 间通信是一个需要专门解决的问题。

这个问题的规模和复杂度与微服务通信有关但不同。 微服务架构已经有成熟的通信标准(REST、gRPC),为什么不直接复用?

区别在于 Agent 通信的语义复杂度更高

  • 微服务调用是确定性的:调用 /api/get_user?id=123,返回用户信息。输入输出格式固定,调用方完全知道另一方能做什么。
  • Agent 委托是半确定性的:调用方需要描述"任务意图"(如"分析这份销售数据"),被调用方需要理解意图并决定如何完成。输出格式不固定,任务可能需要多轮交互,执行过程可能需要实时反馈进度。

此外,Agent 通信还面临一个发现问题:微服务的端点是预先知道的(写在配置文件里),但 Agent 需要动态发现"哪个 Agent 有能力处理这类任务"。当系统中有几百个专业化 Agent 时,这个发现问题变得非常复杂。

2025 年之前的现状: 没有标准,每个框架自搞一套。LangGraph 的 Agent 不能直接和 AutoGen 的 Agent 对话,跨框架协作全靠手写 HTTP 适配层。A2A 和 ANP 正是为了解决这种碎片化问题而提出的。

MCP 解决了一个很具体的问题:Agent 怎么调用工具、怎么访问数据源。但在多 Agent 系统中,MCP 无法解决 Agent 之间的相互通信问题。

考虑这样一个场景:一个主 Agent 负责接收用户请求,一个专门的数据分析 Agent,一个负责生成报告的 Agent。主 Agent 需要把任务分配给另外两个,等它们完成后收集结果。这三个 Agent 之间怎么通信?如果分属不同框架(如 LangGraph 和 AutoGen),两个框架的 Agent 根本没法直接对话,只能通过 HTTP 调用加一堆适配层代码来勉强实现。

这就是 2025 年之前多 Agent 协作的现状:没有标准,每个框架自搞一套,跨框架协作全靠手写胶水代码。A2A 和 ANP 正在解决这个问题。

1.2 A2A 协议(Agent-to-Agent)

2025 年 4 月,Google 发布了 A2A 协议,专门解决 Agent 和 Agent 之间的点对点通信问题。

A2A 的设计思路很直接:把 Agent 之间的交互抽象成"任务(Task)"和"工件(Artifact)"。这个抽象来自软件工程中的"工单"(ticket)思想——任务有明确的生命周期状态,便于跟踪和管理。

Task 是 Agent 之间交互的基本单位。主 Agent 向子 Agent 发送一个 Task,子 Agent 处理这个 Task,最终返回结果。Task 有完整的生命周期:

code
submitted → working → completed
                    ↘ failed
                    ↘ canceled

Artifact 是 Task 的输出产物。可以是文本、结构化 JSON、文件,甚至是另一个 Agent 的引用。

Agent Card 是最有意思的设计。每个 Agent 都有一个 JSON 格式的自我描述文件,通常放在 /.well-known/agent.json,相当于 Agent 的"名片":

json
{
  "name": "数据分析 Agent",
  "description": "专门处理结构化数据分析任务,支持 CSV、JSON、数据库查询结果",
  "version": "1.0.0",
  "endpoint": "https://data-agent.example.com/a2a",
  "capabilities": {
    "streaming": true,
    "pushNotifications": false
  },
  "skills": [
    {
      "id": "data_analysis",
      "name": "数据分析",
      "description": "对结构化数据进行统计分析、趋势识别、异常检测",
      "inputModes": ["text", "data"],
      "outputModes": ["text", "data"]
    },
    {
      "id": "visualization",
      "name": "数据可视化",
      "description": "生成图表描述和可视化建议",
      "inputModes": ["data"],
      "outputModes": ["text"]
    }
  ]
}

主 Agent 先拿到子 Agent 的 Agent Card,了解它能做什么,然后才发送 Task。这样即使是完全陌生的两个 Agent,也能通过 Agent Card 协商出合适的交互方式。

通信方式是 HTTP + SSE(Server-Sent Events)。普通的请求-响应适合短任务,SSE 流式推送适合长时间运行的任务——子 Agent 可以持续推送进度更新,主 Agent 不用一直轮询。

1.2.1 A2A 通信流程

报告生成 Agent数据分析 Agent主 Agent用户报告生成 Agent数据分析 Agent主 Agent用户帮我分析这份销售数据并生成报告GET /.well-known/agent.jsonAgent Card(能力描述)GET /.well-known/agent.jsonAgent Card(能力描述)POST /tasks(发送数据分析 Task)task_id: task_001, status: submittedSSE: status=working, progress=30%SSE: status=working, progress=80%SSE: status=completed, artifact=分析结果JSONPOST /tasks(附带分析结果,请求生成报告)task_id: task_002, status=submittedSSE: status=workingSSE: status=completed, artifact=报告文本分析完成,报告已生成

1.2.2 用 Python 实现一个简单的 A2A Server

A2A 协议本质上是一套 HTTP API 规范,用 FastAPI(Python 的高性能 Web 框架,专门用于快速构建 API 服务)实现起来不复杂。下面的代码验证 A2A 协议的三个核心接口:Agent Card 暴露能力、任务创建接口、SSE 流式状态推送。

python
# 安装依赖:pip install fastapi uvicorn sse-starlette

from fastapi import FastAPI
from fastapi.responses import JSONResponse
from sse_starlette.sse import EventSourceResponse
import asyncio
import uuid
import json

app = FastAPI()

# 任务存储(生产环境用数据库)
tasks = {}


@app.get("/.well-known/agent.json")
def get_agent_card():
    """Agent Card:对外暴露能力描述"""
    return {
        "name": "数据分析 Agent",
        "description": "处理结构化数据分析",
        "version": "1.0.0",
        "endpoint": "http://localhost:8001/a2a",
        "capabilities": {"streaming": True},
        "skills": [
            {
                "id": "analyze",
                "name": "数据分析",
                "description": "统计分析、趋势识别",
                "inputModes": ["text", "data"],
                "outputModes": ["text", "data"]
            }
        ]
    }


@app.post("/tasks")
async def create_task(task_request: dict):
    """接收主 Agent 发来的任务"""
    task_id = str(uuid.uuid4())
    tasks[task_id] = {
        "id": task_id,
        "status": "submitted",
        "input": task_request.get("message"),
        "artifacts": []
    }
    # 异步处理任务
    asyncio.create_task(process_task(task_id))
    return {"task_id": task_id, "status": "submitted"}


@app.get("/tasks/{task_id}/stream")
async def stream_task_status(task_id: str):
    """SSE 流式推送任务状态"""
    async def event_generator():
        while True:
            task = tasks.get(task_id)
            if not task:
                yield {"data": json.dumps({"error": "task not found"})}
                break

            yield {"data": json.dumps({
                "task_id": task_id,
                "status": task["status"],
                "artifacts": task.get("artifacts", [])
            })}

            if task["status"] in ("completed", "failed"):
                break

            await asyncio.sleep(0.5)

    return EventSourceResponse(event_generator())


async def process_task(task_id: str):
    """模拟任务处理过程"""
    tasks[task_id]["status"] = "working"
    await asyncio.sleep(2)  # 模拟分析耗时

    tasks[task_id]["status"] = "completed"
    tasks[task_id]["artifacts"] = [
        {
            "type": "data",
            "content": {"summary": "数据分析完成", "trend": "上升", "anomalies": 0}
        }
    ]

1.3 ANP 协议(Agent Network Protocol)

A2A 解决了两个 Agent 之间点对点通信的问题。但如果系统里有几百个 Agent,每个 Agent 都要和其他 Agent 直接通信,这就是 O(n²) 的连接问题,管理起来会很混乱。

O(n²) 问题的直觉: 如果有 10 个 Agent,两两之间的通信关系有 45 条;如果有 100 个 Agent,就有 4950 条。每条通信关系都需要配置、维护和监控。这种"网状"结构在系统规模增长时会迅速变得不可管理。

微服务架构早就解决了类似的问题:引入"服务注册中心",每个服务不需要知道其他服务的地址,只需要向注册中心查询"哪个服务能处理 X 类请求"。ANP 把这个思路引入了 Agent 网络。

ANP(Agent Network Protocol,Agent网络协议)是 2025 年出现的另一个协议,定位是更大规模的 Agent 网络的服务发现(自动找到哪个Agent能处理某类任务)和路由。

ANP 的核心思路借鉴了微服务架构:Agent 不需要知道其他每个 Agent 的地址,只需要声明自己的能力,注册到服务注册中心。需要某种能力时,去注册中心查找,路由层负责把请求转发到合适的 Agent。

ANP 的几个核心组件:

ServiceRegistry(注册中心):Agent 启动时向这里注册自己的能力描述。注册中心维护全局的 Agent 能力目录。

ServiceDiscovery(服务发现):按能力查询:哪个 Agent 能做"数据分析"?哪个 Agent 能做"翻译"?支持能力标签、版本、优先级等多维度筛选。

Router(路由层):根据任务类型和当前各 Agent 的负载,决定把任务路由给哪个 Agent。支持负载均衡、故障转移。

DID(去中心化身份,Decentralized Identifier,不依赖微信/Google等平台验证身份,而是用密码学技术自证身份):这是 ANP 比 A2A 多出来的一个关键设计。每个 Agent 有一个基于密码学的去中心化身份标识,不依赖任何中心化的身份提供商。Agent 之间通信时可以互相验证身份,防止恶意 Agent 冒充。

1.3.1 ANP 服务发现流程

请求方

Agent 集群

服务注册中心

注册能力: data_analysis

注册能力: report_generation

注册能力: translation

注册能力: code_generation

查询: 我需要 data_analysis 能力

返回: A1 可以处理,端点是...

携带 DID 身份发送任务

验证 DID,处理任务,返回结果

Agent 能力目录

数据分析 Agent
DID: did:web:agent1

报告生成 Agent
DID: did:web:agent2

翻译 Agent
DID: did:web:agent3

代码生成 Agent
DID: did:web:agent4

主 Agent

ANP 更适合平台型的 AI 系统:几十上百个专业化 Agent,动态上下线,需要统一管理。对于只有三五个 Agent 的系统,用 A2A 直连就够了,不需要引入注册中心的复杂性。

1.4 三协议对比:MCP、A2A、ANP

标准化的意义: 为什么需要协议标准,而不是让每个团队自己定义接口?

标准化解决的是互操作性问题:当所有 Agent 遵循同一套协议,任何 Agent 都可以与任何其他 Agent 协作,不需要为每对 Agent 组合写适配代码。这就像 HTTP 标准让任何浏览器都能访问任何网站一样。

对于 Agent 生态而言,标准化还有更深层的意义:第三方服务商可以把自己的服务封装为标准 A2A 接口,任何支持 A2A 的主 Agent 都可以直接调用,形成类似"App Store"的市场效应——专业化 Agent 的多样性会促进整个生态的发展。

这三个协议解决的是不同层次的问题,不是竞争关系。

维度 MCP A2A ANP
定位 Agent ↔ 工具/数据源 Agent ↔ Agent 点对点 Agent 网络服务发现和路由
方向 垂直集成(Agent 调用工具) 水平协作(Agent 之间委派任务) 大规模网络组织(服务发现+路由)
适用场景 接入外部工具、数据库、API 任务分解、专业化子 Agent 协作 数百个 Agent 的企业级平台
身份认证 基本的 OAuth/API Key 基本的 HTTP 认证 DID 去中心化身份
成熟度 相对成熟,生态活跃 2025 年发布,早期阶段 2025 年提出,实验性
发起方 Anthropic Google 社区主导

用一个比喻来理解:

MCP 是插座和电器之间的标准接口(220V、三角插头),解决的是"工具能不能插上"的问题。

A2A 是公司内部部门之间的任务协作流程,一个部门把任务正式委派给另一个部门,有标准的工单格式和状态跟踪。

ANP 是企业内部的黄页目录加上快递路由系统,不需要知道每个部门的内线电话,只需要说"我要找能做翻译的人",系统帮你找到并转接。

三者可以同时使用,互不冲突:Agent 内部用 MCP 调工具,Agent 之间用 A2A 协作,整个 Agent 平台用 ANP 做服务发现。

1.5 现状和局限

A2A 和 ANP 在 2025 年都还处于早期阶段。

A2A 是 Google 推动的,已经有了相对完整的规范文档和参考实现,LangGraph、CrewAI 等框架开始接入支持。但生产环境的案例还不多,很多边缘场景的处理还在讨论和完善中。

ANP 更早期,主要停留在协议设计层面,实现库和工具链还不完善。DID 的理念来自 Web3 领域,落地需要解决密钥管理、身份恢复等复杂问题。


未来的 AI 系统一定是多 Agent 协作的架构。单个 Agent 解决复杂问题的能力有限,把任务分解给专业化的 Agent 协作处理,才是可扩展的方向。

了解 A2A 和 ANP 的意义在于理解多 Agent 通信的设计思路和演进方向。等这两个协议成熟、生态完善的时候,熟悉底层逻辑能大幅降低上手成本。对于早期技术,正确的态度是:先理解原理,等待合适的时机再投入生产。

本页目录