A2A（Agent-to-Agent）协议

YanDDDeat2026/5/27...大约 9 分钟

一、什么是 A2A？

1.1 定义

A2A = Agent-to-Agent，是一套标准化的 Agent 间通信协议。它定义了 Agent 如何：

声明能力（我能做什么）
发现彼此（谁可以帮我）
委托任务（请你做这件事）
异步协作（做完通知我）

1.2 解决的核心痛点

传统模式	A2A 模式
Agent 相互孤立，各自为战	Agent 动态发现、自由协作
点对点定制集成，每对接一次写一套代码	标准化协议，一次对接、处处复用
紧耦合系统，改一处牵全身	松耦合生态，独立演进
受限于单一平台 / 框架	跨平台、跨语言协同

核心转变：从"写胶水代码连接 Agent"变成"Agent 自己会握手"。

二、A2A 的协作机制

2.1 四步工作流

能力发现 ──→ 任务委托 ──→ 动态沟通 ──→ 体验协商

① 能力发现（Discovery）

Agent A 访问 Agent B 的 /.well-known/agent.json，拿到 B 的"能力名片"（Agent Card），了解 B 能处理哪些类型的任务。

② 任务委托（Delegation）

Agent A 根据能力匹配，把子任务派发给最合适的 Agent B。这是一个异步过程——A 发完请求就可以干别的，不用傻等。

③ 动态沟通（Communication）

任务执行过程中，双方可以多轮交互：B 可以问 A"这个参数不够，再给点信息"，A 可以查询任务进度。

④ 体验协商（UX Negotiation）

双方协商输出格式——是纯文本、结构化 JSON、还是富文本？确保最终结果对用户友好。

2.2 一个典型场景

假设你有一个"企业超级助手"，用户问："我下周去上海出差三天，预算够吗？"

用户提问
  │
  ▼
主控 Agent（Orchestrator）
  ├──→ 财务 Agent：查询部门剩余差旅预算
  ├──→ 成本 Agent：获取上海当地酒店 / 餐饮消费水平
  └──→ 整合结果，返回用户

整个过程用户只感知到一个助手，背后是三个 Agent 协同工作。这就是 A2A 的价值——对用户透明，对开发者解耦。

三、技术实现

3.1 技术栈

层面	选型
传输协议	HTTP/JSON（RESTful）或 gRPC
核心模型	Agent Card + Task + Message
通信模式	异步优先，支持回调
服务发现	`/.well-known/agent.json` 标准端点

3.2 核心组件

Agent Card（能力名片）

每个 Agent 通过 GET /.well-known/agent.json 暴露自己的能力：

{
  "name": "FlightAgent",
  "url": "https://api.flight-demo.com/a2a",
  "capabilities": [
    {
      "id": "search_flights",
      "description": "搜索机票",
      "input_schema": {
        "type": "object",
        "properties": {
          "origin": { "type": "string" },
          "dest": { "type": "string" },
          "date": { "type": "string" }
        },
        "required": ["origin", "dest"]
      },
      "output_schema": { "type": "array" }
    }
  ]
}

Agent Card 是 A2A 的"握手协议"——其他 Agent 通过它了解你能提供什么服务、需要什么输入、返回什么输出。

Task 模型（任务生命周期）

submitted ──→ working ──→ completed
                │
                ├──→ input_required（需要人工介入）
                └──→ failed

submitted：任务已提交，排队等待
working：Agent 正在处理
input_required：需要用户补充信息（支持 Human-in-the-loop）
completed：任务完成，附带结果
failed：执行失败，附带错误信息

3.3 完整交互时序

[主控Agent]                         [机票Agent]
    │                                     │
    │── GET /.well-known/agent.json ────→│  ① 能力发现
    │←── Agent Card ──────────────────── │
    │                                     │
    │── POST /tasks ───────────────────→│  ② 任务委托
    │   { task_id, capability_id,        │
    │     input, callback_url }          │
    │←── 202 Accepted ────────────────── │
    │                                     │
    │         [机票Agent 内部处理]         │  ③ 异步执行
    │                                     │
    │←── POST /callback ────────────────│  ④ 结果回调
    │   { task_id, status: "completed",   │
    │     output: [...] }                │

3.4 关键 API

创建任务：

POST /tasks
Content-Type: application/json

{
  "task_id": "req-001",
  "capability_id": "search_flights",
  "input": {
    "origin": "PEK",
    "dest": "PVG",
    "date": "2026-06-01"
  },
  "callback_url": "https://my-agent.com/a2a/callback"
}

回调通知：

POST /callback
Content-Type: application/json

{
  "task_id": "req-001",
  "status": "completed",
  "output": {
    "flights": [
      { "airline": "CA1234", "price": 1280 },
      { "airline": "MU5678", "price": 1450 }
    ]
  }
}

四、A2A vs MCP vs LangGraph

这三个概念经常被混淆，但定位完全不同：

	A2A	MCP	LangGraph
解决什么问题	Agent ↔ Agent 协作	Agent ↔ 外部工具/数据	Agent 内部流程编排
类比	人与人之间的语言	人使用工具（锤子、电脑）	团队内部管理流程
通信方向	水平（Agent 之间）	垂直（Agent 调用工具）	内部（Agent 内部状态图）
典型场景	旅行 Agent 调用机票 Agent	Agent 查询数据库、调 API	多步骤 RAG 管线

三者协作关系

┌─────────────────────────────────────────┐
│  LangGraph 编排层（内部流程）              │
│  ┌──────────┐  ┌──────────┐              │
│  │ Agent A  │  │ Agent B  │              │
│  └────┬─────┘  └────┬─────┘              │
│       │ MCP          │ MCP               │
│       ▼              ▼                   │
│  ┌─────────┐    ┌─────────┐              │
│  │ 数据库   │    │ API     │   ← 工具调用  │
│  └─────────┘    └─────────┘              │
└──────────────┬───────────────────────────┘
               │ A2A                        ← Agent 间通信
               ▼
┌─────────────────────────────────────────┐
│  外部 Agent（天气 / 股票 / 翻译...）      │
└─────────────────────────────────────────┘

一句话总结：LangGraph 编排内部流程，MCP 连接工具和数据，A2A 连接外部 Agent。

五、框架与工具生态

5.1 官方 SDK

语言	仓库	特点
Python	`a2a-sdk-python`	官方首选，生态最完整
TypeScript	`a2a-sdk-ts`	Node.js 原生支持
Java	`a2a-sdk-java`	Spring Boot 集成

5.2 社区集成

框架	集成方式	适用场景
LangGraph	A2A 适配器	将 LangGraph Agent 暴露为 A2A 服务
CrewAI	`crewai-a2a` 插件	在 CrewAI 中调用外部 Agent
FastA2A	轻量封装（FastAPI）	快速原型验证
Dify	内置支持	低代码平台一键导出 A2A 端点

5.3 选型建议

场景	推荐
从零实现 A2A 服务	Google 官方 Python SDK
已有 LangGraph 系统	LangGraph A2A 适配器
快速原型 / Demo	FastA2A
生产环境高并发	`a2a-rs`（Rust）/ `a2a-go`（Go）

六、实践：旅行助手

场景

旅行规划 Agent 需要查询机票价格，但自己没有机票数据——它把任务委托给专门的机票 Agent。

下面用 LangGraph 写机票 Agent 的业务逻辑，用 Google 官方 a2a-sdk（v1.x） 把它暴露为符合 A2A 协议的服务，再用 a2a-sdk 的客户端 API 让主控 Agent 调用。

依赖

pip install "a2a-sdk[http-server]" langgraph langchain-openai httpx uvicorn fastapi

步骤 1：用 LangGraph 实现机票查询业务

LangGraph 负责"业务编排"——抽取参数、调数据源、整合结果。这里 mock 数据，生产环境替换为真实 LLM 节点和机票 API：

# flight_agent/graph.py
from typing import TypedDict
from langgraph.graph import StateGraph, END

class FlightState(TypedDict):
    query: str            # 用户原始问题
    origin: str
    dest: str
    date: str
    flights: list         # 查询结果

def parse_query(state: FlightState) -> FlightState:
    """从自然语言抽参（生产环境接 LLM；这里 mock 演示）"""
    return {**state, "origin": "PEK", "dest": "PVG", "date": "2026-06-01"}

def search_flights(state: FlightState) -> FlightState:
    """调用机票数据源（这里 mock）"""
    return {
        **state,
        "flights": [
            {"airline": "CA1234", "price": 1280},
            {"airline": "MU5678", "price": 1450},
        ],
    }

def build_graph():
    g = StateGraph(FlightState)
    g.add_node("parse", parse_query)
    g.add_node("search", search_flights)
    g.set_entry_point("parse")
    g.add_edge("parse", "search")
    g.add_edge("search", END)
    return g.compile()

flight_graph = build_graph()

步骤 2：用 a2a-sdk 把 LangGraph 包装成 A2A 服务

AgentExecutor 是 a2a-sdk 的抽象类，作用是把 A2A 协议层和你的业务逻辑（LangGraph）粘起来：

# flight_agent/executor.py
import json
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.utils import new_agent_text_message
from flight_agent.graph import flight_graph

class FlightAgentExecutor(AgentExecutor):
    async def execute(
        self,
        context: RequestContext,
        event_queue: EventQueue,
    ) -> None:
        # 1. 从 A2A 请求上下文里拿用户输入
        query = context.get_user_input()

        # 2. 跑 LangGraph
        result = flight_graph.invoke({"query": query})

        # 3. 把结果通过 EventQueue 回写——框架会按 A2A 协议序列化下发
        payload = json.dumps({"flights": result["flights"]}, ensure_ascii=False)
        event_queue.enqueue_event(new_agent_text_message(payload))

    async def cancel(
        self,
        context: RequestContext,
        event_queue: EventQueue,
    ) -> None:
        raise NotImplementedError("flight agent does not support cancel")

步骤 3：声明 Agent Card 并启动服务

# flight_agent/server.py
import uvicorn
from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore
from a2a.types import AgentCard, AgentCapabilities, AgentSkill
from flight_agent.executor import FlightAgentExecutor

# Agent Card：访问 /.well-known/agent-card.json 会返回它
agent_card = AgentCard(
    name="FlightAgent",
    description="根据出发地、目的地、日期查询可选航班与价格",
    url="http://localhost:9999/",
    version="1.0.0",
    capabilities=AgentCapabilities(streaming=True, pushNotifications=True),
    defaultInputModes=["text"],
    defaultOutputModes=["text"],
    skills=[
        AgentSkill(
            id="search_flights",
            name="搜索机票",
            description="给定 origin/dest/date，返回航班列表",
            tags=["flight", "travel"],
            examples=["北京到上海，下周一最便宜的机票"],
        )
    ],
)

request_handler = DefaultRequestHandler(
    agent_executor=FlightAgentExecutor(),
    task_store=InMemoryTaskStore(),   # 生产环境换持久化实现
)

app = A2AStarletteApplication(
    agent_card=agent_card,
    http_handler=request_handler,
)

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

启动后这个服务自动具备：

GET /.well-known/agent-card.json —— 返回 Agent Card
POST / —— 接收 JSON-RPC 2.0 的 message/send / message/stream / tasks/get 等标准方法

步骤 4：主控 Agent 用 a2a-sdk 客户端调用

# travel_agent/client.py
import asyncio
from uuid import uuid4

import httpx
from a2a.client import A2ACardResolver, A2AClient
from a2a.types import MessageSendParams, SendMessageRequest

async def query_flight(question: str) -> dict:
    base_url = "http://localhost:9999"
    async with httpx.AsyncClient() as http:
        # 1. 能力发现：自动拉 /.well-known/agent-card.json
        resolver = A2ACardResolver(httpx_client=http, base_url=base_url)
        card = await resolver.get_agent_card()
        print(f"发现 {card.name}，技能：{[s.id for s in card.skills]}")

        # 2. 建客户端
        client = A2AClient(httpx_client=http, agent_card=card)

        # 3. 发 message/send
        request = SendMessageRequest(
            id=str(uuid4()),
            params=MessageSendParams(message={
                "role": "user",
                "parts": [{"kind": "text", "text": question}],
                "messageId": uuid4().hex,
            }),
        )
        response = await client.send_message(request)
        return response.model_dump(mode="json", exclude_none=True)

if __name__ == "__main__":
    result = asyncio.run(query_flight("帮我查 6 月 1 日北京到上海的机票"))
    print(result)

步骤 5：长任务场景——Push Notification

机票查询是秒级返回，同步等就行。但签证申请、合同审核这类分钟乃至小时级的长任务，A2A 提供 Push Notification 让服务端在状态变化时回调客户端 webhook。

① 客户端注册推送配置：

from a2a.types import PushNotificationConfig, TaskPushNotificationConfig

await client.create_task_push_notification_config(
    TaskPushNotificationConfig(
        taskId=task_id,
        pushNotificationConfig=PushNotificationConfig(
            url="https://my-agent.com/webhook/a2a",
            token="my-shared-secret",   # 服务端推送时带回，用于身份校验
            id="webhook-1",
        ),
    )
)

② Webhook 接收端（用 FastAPI 演示）：

# travel_agent/webhook.py
from fastapi import FastAPI, Request, HTTPException
from a2a.types import Task

app = FastAPI()
SHARED_TOKEN = "my-shared-secret"

@app.post("/webhook/a2a")
async def on_a2a_push(request: Request):
    # 1. spec 强制：校验来源
    #    PushNotificationConfig.token 是最低门槛；
    #    生产环境推荐 JWT 签名 + 时间戳防重放
    if request.headers.get("X-A2A-Notification-Token") != SHARED_TOKEN:
        raise HTTPException(401, "invalid token")

    # 2. 反序列化为 Task 对象（不是扁平 dict）
    body = await request.json()
    task = Task.model_validate(body)

    # 3. 按 state 分发——A2A 状态枚举是中划线
    state = task.status.state
    if state == "completed":
        for artifact in task.artifacts or []:
            print(f"task {task.id} 完成，产物：{artifact.model_dump()}")
    elif state == "input-required":
        # task.status.message 是追问内容，转给用户拿到回答后
        # 再用 A2AClient.send_message 继续这一轮对话
        prompt = task.status.message
        ...
    elif state == "failed":
        print(f"task {task.id} 失败：{task.status.message}")

    return {"ok": True}

A2A 与上一版示意的差异

本节代码使用 a2a-sdk v1.0+ 的真实 API，与早期博客中常见的伪代码相比，注意几点：

字段命名是驼峰（taskId、messageId），不是 task_id
状态值是 input-required、auth-required（中划线），不是下划线
Webhook payload 是完整的 Task 对象，承载结果的是 artifacts 而非 output
传输协议是 JSON-RPC 2.0 over HTTPS，不是裸 REST
Push 接收端必须校验身份（token 或 JWT）

七、总结与展望

A2A 的核心价值

解耦：Agent 之间不再需要一对一硬编码对接，Agent Card 提供标准化的能力描述
异步：任务提交和结果回调分离，不阻塞主流程，天然支持长时间任务
开放：协议层与实现层分离，Python / TS / Java / Rust / Go 任意语言都可以实现
可组合：A2A + MCP + LangGraph 三者协作，形成完整的 Agent 能力矩阵

未来方向

Agent 市场：类似于 API Marketplace，Agent 可以被发布、发现、付费调用
安全与权限：Agent 间的信任模型、访问控制、审计日志
多模态 A2A：不只是文本，图片 / 音频 / 视频 Agent 之间的协作
去中心化：Agent 之间 P2P 发现，不依赖中心注册中心

一句话记住 A2A：让 Agent 像微服务一样互相调用。