A2A（Agent-to-Agent）协议详解：实际应用、落地场景与实战代码

提问

A2A 在实际应用中是怎么落地的？有哪些真实场景？请给出相关代码。

参考答案

一句话：A2A 是 Google 2025 年 4 月开源的 Agent 间通信标准协议（基于 HTTP + JSON-RPC 2.0），让不同厂商、不同框架、不同语言的 Agent 互相发现、互相对话、互相委托任务。落地核心是：Agent Card（名片）→ 发现 → 任务委托 → 流式响应 → 产物交换。

---

一、A2A 是什么、为什么需要它

1.1 问题：Agent 孤岛

2024 年：每个 Agent 都是孤岛
  - LangChain Agent 不会调用 CrewAI Agent
  - 阿里通义 Agent 不会和 Claude Agent 协作
  - 企业 A 的客服 Agent 不会和企业 B 的订单 Agent 互通
  
痛点：
  1. 不同协议：HTTP REST / gRPC / WebSocket 混乱
  2. 不同认证：OAuth2 / JWT / API Key 各自为政
  3. 不同数据格式：JSON / Protobuf / MessagePack
  4. 没有「发现机制」：不知道别的 Agent 能干啥

1.2 A2A 的目标

A2A 由 Google + 50+ 家合作伙伴（Atlassian、Salesforce、SAP、ServiceNow、MongoDB 等）于 2025-04-09 开源，目标：

让 Agent 像人一样协作：找到对方 → 自我介绍 → 委托任务 → 流式拿到结果。

三大原则：

1. 拥抱 Agent 能力：支持长时任务（小时~天）、流式、多模态
2. 基于现有 Web 标准：HTTP / SSE / JSON-RPC 2.0（不造新轮子）
3. 默认安全：企业级认证 + 审计 + 隐私

---

二、5 个核心概念

2.1 概念图谱

┌─────────────────────────────────────────────────────┐
│                   A2A Protocol                      │
├─────────────────────────────────────────────────────┤
│  Agent Card  ← 自我介绍（能力 + 技能 + 认证）       │
│  Task        ← 一次协作（一个工作单元）              │
│  Message     ← 一条对话（一问一答）                  │
│  Part        ← 消息内容（文本 / 文件 / 数据）        │
│  Artifact    ← 最终产物（代码 / 报告 / 图片）        │
└─────────────────────────────────────────────────────┘

2.2 详解

概念	类比	作用
Agent Card	简历 / 名片	JSON 文件，描述”我能干啥”
Task	工作单	一次完整协作，有 lifecycle
Message	微信消息	一条来回，多个 Part 组成
Part	消息片段	文本 / 文件 / 结构化数据
Artifact	交付物	任务的最终产出（带元数据）

---

三、A2A 协议栈详解

3.1 协议分层

┌──────────────────────────────────────────┐
│  Layer 5: 应用层                          │  Agent Card / Task / Artifact
├──────────────────────────────────────────┤
│  Layer 4: 传输层                          │  HTTP / SSE（流式）/ WebSocket
├──────────────────────────────────────────┤
│  Layer 3: 数据格式                        │  JSON-RPC 2.0
├──────────────────────────────────────────┤
│  Layer 2: 安全                            │  OAuth2 / JWT / mTLS
├──────────────────────────────────────────┤
│  Layer 1: 网络                            │  HTTPS / WSS
└──────────────────────────────────────────┘

3.2 通信模式（4 种）

模式	用途	实现
Request-Response	一次性任务（查询天气）	POST /tasks/send
Streaming	长任务实时反馈（生成报告）	POST /tasks/sendSubscribe (SSE)
Polling	不想开 SSE	GET /tasks/{id}
Push Notification	跨网络回调	Webhook

3.3 Agent Card 完整示例

{
  "name": "订单查询助手",
  "description": "查询电商订单状态、物流、退款进度",
  "url": "https://orders.example.com/agent",
  "version": "1.0.0",
  "provider": {
    "organization": "某电商公司",
    "url": "https://example.com"
  },
  "capabilities": {
    "streaming": true,
    "pushNotifications": false,
    "stateTransitionHistory": true
  },
  "defaultInputModes": ["text/plain"],
  "defaultOutputModes": ["application/json", "text/plain"],
  "skills": [
    {
      "id": "query-order",
      "name": "订单查询",
      "description": "根据订单号查询状态和物流",
      "tags": ["订单", "电商", "物流"],
      "examples": [
        "查询订单 20260909001 的状态",
        "我的快递到哪了"
      ],
      "inputModes": ["text/plain"],
      "outputModes": ["application/json"]
    },
    {
      "id": "refund-request",
      "name": "退款申请",
      "description": "提交退款申请",
      "tags": ["退款", "售后"]
    }
  ],
  "securitySchemes": {
    "oauth2": {
      "type": "oauth2",
      "flows": {
        "authorizationCode": {
          "authorizationUrl": "https://auth.example.com/oauth/authorize",
          "tokenUrl": "https://auth.example.com/oauth/token",
          "scopes": {
            "orders:read": "读取订单",
            "orders:write": "操作订单"
          }
        }
      }
    }
  }
}

---

四、A2A vs MCP：最容易混淆的一对

维度	MCP	A2A
定位	Agent 调用工具	Agent 调用 Agent
发起方	Agent 是 client	Agent 既是 client 也是 server
通信	stdio / SSE	HTTP / SSE / WebSocket
协议	JSON-RPC over stdio	JSON-RPC over HTTP
类比	给员工发工具（扳手）	派员工去找另一个员工协作
场景	“查询数据库””发邮件”	“让数据 Agent 跑分析，让报告 Agent 写文档”
开源方	Anthropic（2024-11）	Google（2025-04）
关系	互补，不是竞争	互补，不是竞争

生产实战：

客服 Agent (A2A client)
  ├── MCP: 调用工具（查订单库、发邮件、查物流 API）
  └── A2A: 委托子任务（让数据分析 Agent 生成报告）
              └── 数据分析 Agent 再用 MCP 调 SQL 工具

---

五、3 大落地场景

场景 1：企业内部 — 多 Agent 编排（最常见）

客服 Agent
  ├── A2A → 订单 Agent（查订单状态）
  ├── A2A → 财务 Agent（开发票）
  ├── A2A → 物流 Agent（查快递）
  └── A2A → 推荐 Agent（个性化推荐）

业务价值：每个 Agent 单一职责，易维护 + 易测试 + 易替换。

场景 2：跨企业 — Agent 市场（未来趋势）

某旅行社 Agent
  ├── A2A → 航司 Agent（查机票）
  ├── A2A → 酒店 Agent（订房）
  ├── A2A → 景区 Agent（买门票）
  └── A2A → 保险 Agent（买保险）

业务价值：像 App Store 一样，Agent 之间形成生态。

场景 3：工作流自动化 — 长时任务委托

HR 助手 Agent
  └── A2A → 简历筛选 Agent（异步，长任务）
        └── 跑完后 push notification 回 HR 助手

---

六、完整代码：Python 实现（A2A SDK 官方版）

6.1 安装

pip install a2a-sdk[http-server]

6.2 实现一个「订单查询 Agent」（服务端）

# order_agent.py
from a2a.server import A2AServer, AgentSkill, AgentCard
from a2a.server.task_manager import InMemoryTaskManager
from a2a.types import (
    AgentCapabilities,
    Message,
    MessageSendParams,
    Task,
    TaskState,
    TextPart,
)
import asyncio
import json
from uuid import uuid4

# ============ 1. 定义业务逻辑 ============
async def query_order(order_id: str) -> dict:
    """真实业务：调数据库 / API"""
    # 模拟数据库查询
    mock_db = {
        "20260909001": {
            "status": "shipped",
            "carrier": "顺丰",
            "tracking": "SF1234567890",
            "estimated_arrival": "2026-09-10"
        }
    }
    return mock_db.get(order_id, {"error": "订单不存在"})


# ============ 2. 实现 TaskManager ============
class OrderTaskManager(InMemoryTaskManager):

    async def on_send_message(self, params: MessageSendParams) -> Message:
        """处理 A2A 的 message/send 请求"""
        user_text = params.message.parts[0].text

        # 简单的意图识别
        if "查询" in user_text or "订单" in user_text:
            # 提取订单号（实际用 LLM 解析）
            order_id = "20260909001"
            order_info = await query_order(order_id)

            response_text = f"订单 {order_id} 状态：{order_info['status']}，" \
                          f"快递公司：{order_info['carrier']}，" \
                          f"运单号：{order_info['tracking']}，" \
                          f"预计到达：{order_info['estimated_arrival']}"
        else:
            response_text = "我是订单查询助手，请告诉我订单号。"

        # 构造 A2A Message 响应
        return Message(
            role="agent",
            parts=[TextPart(text=response_text)],
            messageId=str(uuid4()),
        )


# ============ 3. 定义 Agent Card ============
order_card = AgentCard(
    name="订单查询助手",
    description="查询电商订单状态、物流、退款进度",
    url="http://localhost:8001/",
    version="1.0.0",
    capabilities=AgentCapabilities(
        streaming=True,
        pushNotifications=False,
        stateTransitionHistory=True,
    ),
    defaultInputModes=["text/plain"],
    defaultOutputModes=["text/plain"],
    skills=[
        AgentSkill(
            id="query-order",
            name="订单查询",
            description="根据订单号查询状态和物流",
            tags=["订单", "电商", "物流"],
            examples=["查询订单 20260909001"],
        ),
        AgentSkill(
            id="refund-request",
            name="退款申请",
            description="提交退款申请",
            tags=["退款", "售后"],
        ),
    ],
)

# ============ 4. 启动服务端 ============
async def main():
    task_manager = OrderTaskManager()
    server = A2AServer(
        agent_card=order_card,
        task_manager=task_manager,
        host="0.0.0.0",
        port=8001,
    )
    print("订单 Agent 启动: http://localhost:8001")
    print("Agent Card: http://localhost:8001/.well-known/agent-card.json")
    await server.serve()

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

6.3 实现客服 Agent（A2A Client，调用订单 Agent）

# customer_service_agent.py
from a2a.client import A2AClient
from a2a.types import Message, MessageSendParams, TextPart, Role
import asyncio
from uuid import uuid4


async def main():
    # ============ 1. 创建客户端（指向订单 Agent）============
    client = await A2AClient.from_agent_card_url(
        "http://localhost:8001/.well-known/agent-card.json"
    )

    # ============ 2. 构造 A2A Message ============
    user_msg = Message(
        role=Role.user,
        parts=[TextPart(text="帮我查一下订单 20260909001 的物流")],
        messageId=str(uuid4()),
    )

    params = MessageSendParams(message=user_msg)

    # ============ 3. 发送请求 ============
    print(">>> 调用订单 Agent...")
    response = await client.send_message(params)

    print(f"\n✅ 订单 Agent 回复：\n{response.parts[0].text}")


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

6.4 运行

# 终端 1：启动订单 Agent
python order_agent.py

# 终端 2：运行客服 Agent
python customer_service_agent.py

# 输出：
# >>> 调用订单 Agent...
# ✅ 订单 Agent 回复：
# 订单 20260909001 状态：shipped，快递公司：顺丰，运单号：SF1234567890，预计到达：2026-09-10

---

七、完整代码：Spring AI 实现（Java 生态）

7.1 Maven 依赖

<!-- pom.xml -->
<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>com.squareup.okhttp3</groupId>
        <artifactId>okhttp</artifactId>
        <version>4.12.0</version>
    </dependency>
</dependencies>

7.2 A2A 服务端：Spring Controller

// OrderAgentController.java
package tech.javai.a2a;

import org.springframework.web.bind.annotation.*;
import java.util.*;

@RestController
@RequestMapping("/a2a")
public class OrderAgentController {

    /**
     * A2A 标准端点：暴露 Agent Card
     * 客户端通过 GET /.well-known/agent-card.json 发现 Agent
     */
    @GetMapping("/.well-known/agent-card.json")
    public Map<String, Object> agentCard() {
        Map<String, Object> card = new LinkedHashMap<>();
        card.put("name", "订单查询助手");
        card.put("description", "查询电商订单状态、物流");
        card.put("url", "http://localhost:8080/a2a");
        card.put("version", "1.0.0");
        card.put("defaultInputModes", List.of("text/plain"));
        card.put("defaultOutputModes", List.of("text/plain"));

        card.put("capabilities", Map.of(
                "streaming", true,
                "pushNotifications", false,
                "stateTransitionHistory", true
        ));

        card.put("skills", List.of(
                Map.of(
                        "id", "query-order",
                        "name", "订单查询",
                        "description", "根据订单号查询订单状态和物流",
                        "tags", List.of("订单", "电商", "物流"),
                        "examples", List.of("查询订单 20260909001")
                ),
                Map.of(
                        "id", "refund-request",
                        "name", "退款申请",
                        "description", "提交退款申请",
                        "tags", List.of("退款", "售后")
                )
        ));

        return card;
    }

    /**
     * A2A 标准端点：message/send（JSON-RPC 2.0）
     */
    @PostMapping(value = "/", consumes = "application/json", produces = "application/json")
    public Map<String, Object> sendMessage(@RequestBody JsonRpcRequest request) {
        // 解析 JSON-RPC 请求
        Map<String, Object> params = (Map<String, Object>) request.getParams();
        Map<String, Object> message = (Map<String, Object>) params.get("message");
        List<Map<String, Object>> parts = (List<Map<String, Object>>) message.get("parts");
        String userText = (String) parts.get(0).get("text");

        // 业务处理
        String responseText = processOrderQuery(userText);

        // 构造 JSON-RPC 2.0 响应
        Map<String, Object> result = new LinkedHashMap<>();
        result.put("kind", "message");
        result.put("role", "agent");
        result.put("messageId", UUID.randomUUID().toString());
        result.put("parts", List.of(Map.of("kind", "text", "text", responseText)));

        Map<String, Object> response = new LinkedHashMap<>();
        response.put("jsonrpc", "2.0");
        response.put("id", request.getId());
        response.put("result", result);
        return response;
    }

    private String processOrderQuery(String userText) {
        // 简单业务逻辑：实际用 Spring AI + LLM 解析意图
        if (userText.contains("查询") || userText.contains("订单")) {
            return "订单 20260909001 状态：shipped，快递公司：顺丰，运单号：SF1234567890";
        }
        return "我是订单查询助手，请告诉我订单号。";
    }
}

// JSON-RPC 请求 DTO
class JsonRpcRequest {
    private String jsonrpc;
    private String method;
    private Object id;
    private Map<String, Object> params;

    public String getJsonrpc() { return jsonrpc; }
    public void setJsonrpc(String jsonrpc) { this.jsonrpc = jsonrpc; }
    public String getMethod() { return method; }
    public void setMethod(String method) { this.method = method; }
    public Object getId() { return id; }
    public void setId(Object id) { this.id = id; }
    public Map<String, Object> getParams() { return params; }
    public void setParams(Map<String, Object> params) { this.params = params; }
}

7.3 A2A 客户端：Java 调用其他 Agent

// A2AClientService.java
package tech.javai.a2a;

import okhttp3.*;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.stereotype.Service;
import java.util.*;

@Service
public class A2AClientService {

    private final OkHttpClient httpClient = new OkHttpClient();
    private final ObjectMapper json = new ObjectMapper();

    /**
     * 1. 发现远端 Agent（拿 Agent Card）
     */
    public Map<String, Object> discoverAgent(String agentCardUrl) throws Exception {
        Request req = new Request.Builder().url(agentCardUrl).get().build();
        try (Response resp = httpClient.newCall(req).execute()) {
            return json.readValue(resp.body().string(), Map.class);
        }
    }

    /**
     * 2. 发送 A2A message/send 请求
     */
    public String sendMessage(String agentUrl, String text) throws Exception {
        // 构造 JSON-RPC 2.0 请求
        Map<String, Object> request = new LinkedHashMap<>();
        request.put("jsonrpc", "2.0");
        request.put("method", "message/send");
        request.put("id", UUID.randomUUID().toString());
        request.put("params", Map.of(
                "message", Map.of(
                        "role", "user",
                        "parts", List.of(Map.of("kind", "text", "text", text)),
                        "messageId", UUID.randomUUID().toString()
                )
        ));

        String body = json.writeValueAsString(request);

        Request httpReq = new Request.Builder()
                .url(agentUrl)
                .post(RequestBody.create(body, MediaType.parse("application/json")))
                .build();

        try (Response resp = httpClient.newCall(httpReq).execute()) {
            Map<String, Object> response = json.readValue(resp.body().string(), Map.class);
            Map<String, Object> result = (Map<String, Object>) response.get("result");
            List<Map<String, Object>> parts = (List<Map<String, Object>>) result.get("parts");
            return (String) parts.get(0).get("text");
        }
    }

    /**
     * 3. 实战：客服 Agent 调用订单 Agent
     */
    public String queryOrder(String orderId) throws Exception {
        // 先发现（实际可缓存）
        Map<String, Object> card = discoverAgent(
                "http://localhost:8080/a2a/.well-known/agent-card.json"
        );
        System.out.println(">>> 发现 Agent: " + card.get("name"));

        // 再调用
        return sendMessage(
                "http://localhost:8080/a2a/",
                "查询订单 " + orderId
        );
    }
}

7.4 测试

// A2ATest.java
@SpringBootTest
class A2ATest {

    @Autowired
    private A2AClientService client;

    @Test
    void testOrderAgent() throws Exception {
        String result = client.queryOrder("20260909001");
        assertNotNull(result);
        assertTrue(result.contains("shipped"));
        System.out.println("订单 Agent 返回: " + result);
    }
}

---

八、进阶：流式响应（SSE）

8.1 服务端：Streaming 模式

# streaming_agent.py
from a2a.server import A2AServer
from a2a.types import (
    MessageSendParams, Message, TextPart, TaskStatusUpdateEvent, TaskArtifactUpdateEvent
)
import asyncio

class StreamingReportAgent(InMemoryTaskManager):

    async def on_send_message_subscribe(self, params: MessageSendParams):
        """流式返回报告生成进度"""
        yield TaskStatusUpdateEvent(
            status="working",
            message="开始生成报告..."
        )
        await asyncio.sleep(1)

        # 分段返回 Artifact
        sections = [
            "## 一、数据概览\n本月订单增长 15%",
            "## 二、问题诊断\n退货率上升 3%",
            "## 三、建议\n加强物流合作"
        ]
        for sec in sections:
            yield TaskArtifactUpdateEvent(
                artifact=SectionArtifact(parts=[TextPart(text=sec)])
            )
            await asyncio.sleep(1)

        yield TaskStatusUpdateEvent(status="completed", message="报告完成")

8.2 客户端：消费 SSE 流

async with client.send_message_stream(params) as stream:
    async for event in stream:
        if isinstance(event, TaskStatusUpdateEvent):
            print(f"[状态] {event.status}")
        elif isinstance(event, TaskArtifactUpdateEvent):
            print(f"[产物] {event.artifact.parts[0].text}")

---

九、生产级实战：3 个真实案例

案例 1：电商客服系统

架构：
  用户 → 前端 → 客服 Agent (A2A client)
              ├─ A2A → 订单 Agent（查订单）
              ├─ A2A → 物流 Agent（查快递）
              ├─ A2A → 售后 Agent（退款/换货）
              └─ A2A → 推荐 Agent（个性化推荐）
  
每个 Agent 单独部署，单一职责。

收益：

• 客服 Agent 不需要懂订单 API
• 业务方可以独立升级订单 Agent
• 新增 Agent 不影响现有系统

案例 2：跨企业数据分析

企业 A 的 BI Agent
  └── A2A → 企业 B 的财务 Agent（提供财务数据）
  └── A2A → 企业 C 的销售 Agent（提供销售数据）
  
组成跨企业的数据分析 pipeline。

收益：无需数据中台，通过 A2A 直接对接。

案例 3：智能 Agent 市场（未来）

用户：「帮我订明天的杭州-上海机票 + 酒店 + 景区门票」
  旅行 Agent
    ├── A2A → 航司 Agent（订票）
    ├── A2A → 酒店 Agent（订房）
    ├── A2A → 景区 Agent（订门票）
    └── 统一结算
    
类似 App Store，Agent 之间形成生态。

---

十、踩坑清单

坑	怎么避
Agent Card 不更新	客户端缓存设 TTL（如 1 小时）
长任务超时	用 streaming（message/stream）而非 send
跨域 CORS	Agent 服务端配 Access-Control-Allow-Origin: *
认证不一致	用 OAuth2 标准方案，避免自创 token
流式断连	客户端实现重连 + 服务端断点续传
任务状态丢失	用 Redis 持久化 Task 状态
Agent 死循环	设置 max_turns 和 timeout
信任问题	私用 mTLS、公用 OAuth2 + Scope

---

十一、面试怎么答

面试官问「A2A 怎么落地」时：

A2A 是 Google 2025-04 推出的 Agent 间通信协议，让不同厂商/框架的 Agent 互相发现、对话、委托任务。

核心概念：Agent Card（名片）+ Task（任务）+ Message（消息）+ Part（片段）+ Artifact（产物）。

通信模式：Request-Response / Streaming（SSE） / Polling / Push Notification。

落地场景：

1. 企业内部多 Agent 编排：客服 Agent → 订单/物流/财务 Agent
2. 跨企业 Agent 协作：旅行 Agent → 航司/酒店 Agent
3. 长时任务委托：HR 助手 → 简历筛选 Agent（异步）

A2A vs MCP：A2A 是 Agent ↔ Agent，MCP 是 Agent ↔ 工具，互补不冲突。生产中 Agent 同时用两者（A2A 派活 + MCP 干活）。

实战要点：

• Agent Card 一定要详细（skill + example）
• 长任务用 streaming
• 认证用 OAuth2 标准
• 客户端缓存 Agent Card（TTL 1h）

---

十二、A2A 生态现状（2026-09）

项目	语言	状态
a2a-sdk（官方）	Python	✅ GA
a2a-java-sdk	Java	✅ Beta（社区）
a2a-js	TypeScript	✅ Beta
a2a-go	Go	⚠️ 实验
MCP-A2A Bridge	Python	⚠️ 实验（社区）

合作伙伴：Google、Atlassian、Salesforce、SAP、ServiceNow、MongoDB、Datadog、Couchbase、LangChain、Llamaindex 等 50+。

---

十三、总结

维度	A2A
目标	Agent ↔ Agent 通信
协议	HTTP + JSON-RPC 2.0 + SSE
核心概念	Agent Card / Task / Message / Part / Artifact
发现机制	Agent Card（JSON）
认证	OAuth2 / JWT / mTLS
场景	多 Agent 编排、跨企业协作、长任务委托
生态	Google + 50+ 合作伙伴
vs MCP	互补（A2A 找人，MCP 找工具）
生产建议	单 Agent 职责 + Agent Card 详细 + 流式 + OAuth2

一句话：A2A 让 Agent 从「单兵作战」走向「团队协作」——这是 2025 年 Agent 领域最重要的协议之一。

---

相关阅读

• MCP 和 Skills 有什么区别？
• RAG 完整详细流程
• Agent 上下文工程：四种框架对比
• Agent 框架选型对比

---

最后更新：2026-09-09