万字深度解析 A2A 协议：当 AI Agent 遇见「巴别塔解决方案」——从 Agent Card 发现机制到 Task 生命周期管理、从 JSON-RPC 2.0 到 SSE 流式推送的完整技术指南（2026）

本文基于 A2A 协议最新规范（2026 年 Linux Foundation 托管版本）撰写，涵盖完整的协议架构分析、Python/TypeScript 双语言实战代码、以及与 MCP/ACP/ANP 协议的深度对比。全文约 12000 字，阅读时间约 30 分钟。

前言：AI Agent 的「巴别塔困境」

2025 年 4 月，Google 在 Cloud Next 大会上做了一件看起来不那么「性感」的事——发布了一个名为 Agent2Agent（A2A） 的开放协议。没有炫酷的 Demo，没有基准测试的刷榜，只是一份规范文档和几份参考实现。

但恰恰是这件事，可能被历史证明是 2025-2026 年 AI 领域最重要的基础设施动作之一。

问题的本质：为什么 Agent 之间无法「说话」？

让我们先理解问题到底是什么。今天一个典型的企业 AI 部署场景是这样的：

• HR 部门用的是 Salesforce Agentforce，管理招聘流程
• 财务系统跑的是 SAP 的 AI Agent，处理报销审批
• 技术团队用 LangChain 搭了个内部知识库 Agent
• 产品经理在用 Cursor（背后是 Claude）写需求文档

这四个 Agent，每一个都是某个垂直领域的强者。但如果我想让它们协作——比如：HR Agent 收到简历 → 调用技术团队的知识库 Agent 判断技能匹配度 → 结果传给财务 Agent 核算薪资预算——在今天，你需要给每一对 Agent 写一套定制胶水代码。

这不是夸张。每一个 Agent 有自己的：

• 消息格式（有的用 JSON，有的用 Protobuf，有的干脆是自由文本）
• 认证方式（有的用 OAuth，有的用 API Key，有的根本没鉴权）
• 调用语义（有的用 RPC，有的用 REST，有的用消息队列）
• 能力描述方式（有的有 OpenAPI 文档，有的啥也没有，只能读源码）

这就是 AI 世界的「巴别塔困境」：每个 Agent 都说不同的语言，协作成本随 Agent 数量呈 O(N²) 增长。

A2A 协议要做的，就是给所有 Agent 一本通用的「普通话教材」。

一、A2A 协议全景：它到底是什么？

1.1 官方定义与核心定位

A2A（Agent-to-Agent Protocol） 是一个开放标准的通信协议，专门用于 AI Agent 之间的互操作。

用一句话说清楚：A2A 定义了两个 Agent 如何发现彼此、协商能力、委托任务、跟踪进度、交付结果——而且这些 Agent 可以来自不同厂商、用不同框架、部署在不同环境。

关键定位（非常重要，理解了这段就理解了 A2A 的整个设计哲学）：

1.2 A2A 与 MCP 的关系：为什么需要两个协议？

这是被问最多的问题，值得用一整节讲清楚。

MCP（Model Context Protocol） 是 Anthropic 在 2024 年 11 月发布的协议，解决的是 Agent 与工具/数据源之间的连接问题。用个比喻：MCP 是 Agent 的「手和眼睛」——让 Agent 能调用外部 API、查数据库、读文件。

A2A 解决的是 Agent 与 Agent 之间的协作问题。继续比喻：A2A 是 Agent 的「嘴和耳朵」——让 Agent 能与其他 Agent 对话、委托任务、协同完成复杂工作。

两者是互补关系，不是竞争关系。一个典型的复杂 AI 系统会同时使用两个协议：

用户提问：「帮我分析竞品 Q3 的财务表现，并生成 PPT」
  ↓
主 Agent（协调者）
  ├── 通过 A2A 调用「财务分析 Agent」（远程，SAP 系统）
  │     └── 财务 Agent 通过 MCP 连接 SAP API（工具调用）
  ├── 通过 A2A 调用「PPT 生成 Agent」（远程，另一团队维护）
  │     └── PPT Agent 通过 MCP 连接 Google Slides API
  └── 通过 MCP 调用本地「邮件发送工具」（完成任务后通知用户）

一句话总结：MCP 让 Agent 能力更强（能用的工具更多），A2A 让 Agent 能找人帮忙（能协作的同伴更多）。

二、核心数据模型：A2A 的「词汇表」

A2A 协议定义了四组核心数据结构。理解这四组结构，就等于理解了 A2A 的 80%。

2.1 Agent Card：Agent 的「数字名片」

每个支持 A2A 的 Agent 必须在固定路径发布一份 Agent Card JSON 文件：

https://your-agent.example.com/.well-known/agent.json

这份文件是其他 Agent 自动发现这个 Agent 的唯一依据。它的设计灵感来自 OpenAPI Spec，但专门针对 Agent 能力描述做了扩展。

完整字段解析

{
  "name": "财务分析专用 Agent",
  "description": "接收企业财务数据，输出多维度分析报告，支持趋势预测",
  "version": "2.1.0",
  "url": "https://finance-agent.example.com/a2a",
  "provider": {
    "organization": "FinTech AI Corp",
    "url": "https://fintechai.example.com"
  },
  "capabilities": {
    "streaming": true,
    "pushNotifications": true,
    "stateTransitionHistory": false
  },
  "defaultInputModes": ["text", "data", "file"],
  "defaultOutputModes": ["text", "data", "file"],
  "securitySchemes": {
    "bearerAuth": {
      "type": "http",
      "scheme": "bearer",
      "bearerFormat": "JWT",
      "description": "使用企业级 JWT Token 认证，Token 需包含 finance:read 权限"
    },
    "apiKeyAuth": {
      "type": "apiKey",
      "in": "header",
      "name": "X-API-Key"
    }
  },
  "security": [
    { "bearerAuth": ["finance:read", "finance:analyze"] }
  ],
  "skills": [
    {
      "id": "financial-analysis-v2",
      "name": "多维度财务分析",
      "description": "对输入的财务数据进行毛利率、净利率、现金流等多维度分析，输出结构化报告",
      "tags": ["finance", "analysis", "reporting", "china-GAAP"],
      "examples": [
        "分析 Q3 营收趋势，重点看毛利率变化",
        "对比各事业部 Q2 vs Q3 的 EBITDA 表现"
      ],
      "inputModes": ["text", "data", "file"],
      "outputModes": ["text", "data", "file"]
    },
    {
      "id": "forecast-v1",
      "name": "财务预测",
      "description": "基于历史财务数据，使用时间序列模型预测未来 4 季度的核心财务指标",
      "tags": ["finance", "forecast", "time-series"],
      "inputModes": ["data"],
      "outputModes": ["data", "file"]
    }
  ],
  "rateLimits": [
    {
      "requestsPerMinute": 60,
      "description": "免费层速率限制"
    }
  ]
}

逐字段深度解读：

• url：A2A JSON-RPC 端点的完整 URL。所有 message/send、tasks/get 等方法都 POST 到这个地址。注意：这不是 Agent 的 Web UI 地址，是纯 API 端点。

• capabilities.streaming：是否支持 SSE 流式响应。如果设为 false，客户端就不能用 message/stream 方法，只能用同步的 message/send。

• capabilities.pushNotifications：是否支持任务完成后主动回调（Webhook）。开启后，客户端可以在发起任务时附带一个回调 URL，Agent 完成后会 POST 结果到这个 URL。

• securitySchemes：认证方案的完整描述，格式完全兼容 OpenAPI 3.0 的 Security Scheme。这意味着如果你已经用过 OpenAPI，对 A2A 的认证配置不会感到任何陌生。

• skills：这是 Agent Card 最重要的部分——能力目录。每个 Skill 有唯一 id、人类可读的 name 和 description、用于语义匹配的 tags、帮助其他 Agent 理解用法的 examples。

  设计精妙之处：examples 字段。这不是给人类读的文档，而是给 LLM 读的上下文！当一个协调 Agent 用 LLM 做任务路由决策时，它把各个 Agent 的 skills.examples 送给 LLM，LLM 就能准确判断「这个任务应该派给谁」。

2.2 Task：工作的基本单元

Task 是 A2A 中最核心的对象。所有跨 Agent 的交互都围绕 Task 进行。

Task 的完整生命周期

submitted（已提交）
    ↓
working（处理中）
    ↓
completed（完成）
    ↓
[ artifacts 可获取 ]

也可分支到：
working → input-required（需要补充信息）→ working → completed
working → failed（失败）
working → canceled（被取消）
submitted → rejected（被拒绝，未开始处理）

Task 数据结构详解

{
  "kind": "task",
  "id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
  "contextId": "cafe-beef-1234-5678-90abcdef1234",
  "status": {
    "state": "working",
    "timestamp": "2026-07-02T10:30:00.000Z",
    "message": {
      "role": "agent",
      "parts": [
        {
          "kind": "text",
          "text": "正在分析中，已处理 60% 的数据..."
        }
      ]
    }
  },
  "artifacts": [],
  "history": [
    {
      "role": "user",
      "parts": [
        {
          "kind": "text",
          "text": "请分析附件中的 Q3 财务数据"
        },
        {
          "kind": "file",
          "file": {
            "name": "q3_data.xlsx",
            "mimeType": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
            "data": "UEsFBgAAAAAGAA...（Base64 编码）"
          }
        }
      ],
      "messageId": "msg-001",
      "taskId": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
      "contextId": "cafe-beef-1234-5678-90abcdef1234"
    }
  ],
  "metadata": {
    "createdBy": "orchestrator-agent-v3",
    "priority": "high",
    "estimatedCompletionMs": 45000
  }
}

关键设计决策解析：

1. id 与 contextId 的分离：

   • id 标识「这一次具体的任务」
   • contextId 标识「这一整段对话/会话」
   • 举例：用户问「分析 Q3 财务数据」→ 创建 Task-1（contextId: C1）；用户接着问「那 Q2 呢？」→ 创建 Task-2（contextId: C1，同一个上下文）。协调 Agent 可以通过相同的 contextId 把所有相关 Task 关联起来。

2. history 字段：

   • 存储 Task 的完整消息历史（包括多轮 input-required 的交互）
   • 这不是审计日志（审计有单独的机制），而是让 Agent 在处理过程中能「记住之前说了什么」
   • 注意：history 的大小可能很大，生产环境需要限制历史长度或做摘要

3. artifacts 与 message 的语义区别：

   • message 是「说话」——通信过程中的对话内容
   • artifacts 是「交付物」——任务完成后产生的正式输出
   • 一个 Task 可以有多个 Artifact（比如：分析报告 Text + 原始数据 Data + 图表图片 File）

2.3 Message 与 Part：通信的载体

A2A 的消息结构非常灵活，一条 Message 可以包含多种类型的内容块（Part）。

{
  "role": "user",
  "parts": [
    {
      "kind": "text",
      "text": "请对比分析苹果和谷歌 Q3 的财报，重点看：\n1. 营收增长率\n2. 研发支出占比\n3. 现金流变化"
    },
    {
      "kind": "data",
      "data": {
        "companies": ["AAPL", "GOOGL"],
        "metrics": ["revenue_growth", "rd_expense_ratio", "operating_cash_flow"],
        "quarter": "2026-Q3",
        "currency": "USD",
        "includeCharts": true
      }
    },
    {
      "kind": "file",
      "file": {
        "name": "去年同期分析模板.docx",
        "mimeType": "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
        "data": "UEsFBgAAAAA...（Base64）"
      }
    }
  ],
  "messageId": "msg-orchestrator-001",
  "taskId": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
  "contextId": "cafe-beef-1234-5678-90abcdef1234"
}

Part 类型完整说明：

一个重要的实际考虑：Base64 编码大文件的性能问题。A2A 协议允许 FilePart 通过 uri 字段引用外部 URL，而不必内嵌 Base64：

{
  "kind": "file",
  "file": {
    "name": "large-dataset.zip",
    "mimeType": "application/zip",
    "uri": "https://storage.example.com/datasets/q3-financials.zip",
    "expiresAt": "2026-07-03T00:00:00.000Z"
  }
}

三、协议交互详解：从发现到完成的完整流程

3.1 第一步：Agent 发现

发现有两种模式：

模式一：开放发现（Open Discovery）

最适合公网可访问的 Agent。客户端直接 HTTP GET 固定路径：

curl -s https://finance-agent.example.com/.well-known/agent.json \
  -H "Accept: application/json"

模式二：注册表发现（Registry-Based Discovery）

适合企业内网场景。所有 Agent 向一个中央注册表注册自己的 Agent Card，客户端查询注册表来发现 Agent。

# 伪代码：通过注册表发现擅长「财务分析」的 Agent
registry_client = A2ARegistryClient("https://registry.mycorp.com")
agents = registry_client.search(
    tags=["finance", "analysis"],
    capabilities={"streaming": True}
)
# agents 是 AgentCard 对象的列表

安全扩展：Agent Card 可以部分公开、部分需要认证才能查看。公开部分只包含基本描述，完整的能力列表（包括 skills 的 examples）需要携带有效 Token 通过 agent/authenticatedExtendedCard 方法获取。

3.2 第二步：发起任务

A2A 提供了两种发起任务的方式，适应不同场景。

同步模式：message/send

适合短任务（秒级完成）：

import json
import requests

def send_a2a_task(agent_url, message, api_token):
    """发送同步 A2A 任务请求"""
    payload = {
        "jsonrpc": "2.0",
        "id": "req-001",
        "method": "message/send",
        "params": {
            "message": message,
            "configuration": {
                "blocking": True,  # 等待任务完成
                "acceptedOutputModes": ["text", "data"]
            }
        }
    }
    
    response = requests.post(
        agent_url,
        json=payload,
        headers={
            "Authorization": f"Bearer {api_token}",
            "Content-Type": "application/json"
        },
        timeout=30
    )
    
    result = response.json()
    if "error" in result:
        raise Exception(f"A2A Error: {result['error']['message']}")
    
    task = result["result"]
    return task

# 构造消息
message = {
    "role": "user",
    "parts": [
        {
            "kind": "text",
            "text": "分析苹果公司 2026 Q3 财报，重点看营收增长和毛利率变化"
        }
    ],
    "messageId": "msg-001"
}

# 调用
task_result = send_a2a_task(
    agent_url="https://finance-agent.example.com/a2a",
    message=message,
    api_token="eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9..."
)

print(f"Task 状态: {task_result['status']['state']}")
print(f"结果 Artifact: {task_result['artifacts'][0]['parts'][0]['text']}")

流式模式：message/stream

适合长任务（分钟级），通过 SSE 实时获取进度：

import requests
import json

def stream_a2a_task(agent_url, message, api_token):
    """发送流式 A2A 任务请求，实时接收进度"""
    payload = {
        "jsonrpc": "2.0",
        "id": "req-stream-001",
        "method": "message/stream",
        "params": {
            "message": message
        }
    }
    
    response = requests.post(
        agent_url,
        json=payload,
        headers={
            "Authorization": f"Bearer {api_token}",
            "Content-Type": "application/json",
            "Accept": "text/event-stream"
        },
        stream=True,
        timeout=300
    )
    
    artifacts_buffer = {}  # artifactId -> 累积的 parts
    
    for line in response.iter_lines():
        if not line:
            continue
        line = line.decode("utf-8")
        if not line.startswith("data: "):
            continue
        
        data_str = line[6:]  # 去掉 "data: " 前缀
        event = json.loads(data_str)
        
        result = event.get("result", {})
        kind = result.get("kind")
        
        if kind == "status-update":
            state = result["status"]["state"]
            print(f"[状态更新] {state}")
            if result.get("final"):
                print("[任务完成]")
                break
        
        elif kind == "artifact-update":
            artifact = result["artifact"]
            artifact_id = artifact["artifactId"]
            is_append = result.get("append", False)
            is_last = result.get("lastChunk", False)
            
            if not is_append:
                # 新的 Artifact 开始
                artifacts_buffer[artifact_id] = artifact
            else:
                # 追加到已有 Artifact
                existing = artifacts_buffer[artifact_id]
                for part in artifact["parts"]:
                    existing["parts"].append(part)
            
            if is_last:
                print(f"[Artifact 完成] {artifact_id}")
    
    return artifacts_buffer

# 使用
message = {
    "role": "user",
    "parts": [{"kind": "text", "text": "生成长篇财务分析报告，包括图表"}],
    "messageId": "msg-stream-001"
}

artifacts = stream_a2a_task(
    "https://finance-agent.example.com/a2a",
    message,
    "eyJhbG..."
)

3.3 多轮交互：input-required 状态

这是 A2A 协议最精妙的设计之一。Agent 在处理过程中如果发现需要更多信息，可以把 Task 状态设为 input-required，并在 status.message 中说明需要什么。

场景：用户让 Agent「分析这份数据」，但 Agent 发现数据里缺少「币种」字段，需要用户补充。

// Agent 的响应（Task 状态变为 input-required）
{
  "jsonrpc": "2.0",
  "id": "req-001",
  "result": {
    "kind": "task",
    "id": "task-001",
    "status": {
      "state": "input-required",
      "message": {
        "role": "agent",
        "parts": [
          {
            "kind": "text",
            "text": "数据分析需要币种信息。请问附件中的金额是人民币（CNY）还是美元（USD）？"
          },
          {
            "kind": "data",
            "data": {
              "missingFields": ["currency"],
              "allowedValues": ["CNY", "USD", "EUR", "GBP"],
              "reason": "汇率转换需要明确的币种信息"
            }
          }
        ]
      }
    }
  }
}

客户端补充信息后，使用同一个 taskId 再次调用 message/send：

# 客户端补充信息，继续同一个 Task
continue_message = {
    "role": "user",
    "parts": [{"kind": "text", "text": "金额单位是人民币 CNY"}],
    "messageId": "msg-002",
    "taskId": "task-001",  # 关键：同一个 Task ID
    "contextId": "ctx-001"
}

response = requests.post(
    agent_url,
    json={
        "jsonrpc": "2.0",
        "id": "req-002",
        "method": "message/send",
        "params": {"message": continue_message}
    },
    headers={"Authorization": f"Bearer {api_token}"}
)
# Task 重新进入 working 状态

四、代码实战：用 Python 构建一个完整的 A2A Agent

理论讲完了，现在动手。我们从零实现一个支持 A2A 协议的财务分析 Agent，包括服务端和客户端。

4.1 环境准备

pip install a2a-sdk  # Google 官方 Python SDK
# 或者从源码安装最新版
pip install git+https://github.com/google-a2a/a2a-python.git

4.2 服务端：实现 A2A Agent

"""
finance_a2a_agent.py
一个完整的 A2A 协议服务端实现
支持：Agent Card 发布、tasks/send、message/stream、tasks/get
"""

import asyncio
import json
import uuid
from datetime import datetime
from typing import Dict, List, Optional
from a2a import (
    A2AServer,
    AgentCard,
    Skill,
    Task,
    TaskStatus,
    TaskState,
    Message,
    TextPart,
    DataPart,
    Artifact
)
from a2a.server import InMemoryTaskStore

# ============ Agent Card 定义 ============
AGENT_CARD = AgentCard(
    name="财务分析 A2A Agent",
    description="基于 Python 和 Pandas 的财务数据分析 Agent，支持多维度分析和趋势预测",
    version="1.0.0",
    url="http://localhost:8080/a2a",
    capabilities={
        "streaming": True,
        "pushNotifications": False,
        "stateTransitionHistory": True
    },
    defaultInputModes=["text", "data", "file"],
    defaultOutputModes=["text", "data", "file"],
    skills=[
        Skill(
            id="financial-analysis",
            name="财务数据分析",
            description="对财务数据进行多维度分析，支持毛利率、净利率、现金流等指标",
            tags=["finance", "analysis", "pandas"],
            examples=[
                "分析 Q3 营收趋势",
                "计算各事业部净利率并排序"
            ],
            inputModes=["text", "data", "file"],
            outputModes=["text", "data"]
        )
    ],
    securitySchemes={
        "bearerAuth": {
            "type": "http",
            "scheme": "bearer",
            "bearerFormat": "JWT"
        }
    },
    security=[{"bearerAuth": []}]
)

# ============ Task 处理器 ============
class FinanceTaskProcessor:
    """处理财务分析任务的核心逻辑"""
    
    async def process_task(self, task: Task, message: Message) -> Task:
        """
        核心处理方法：
        1. 解析输入（文本指令 + 数据）
        2. 执行分析
        3. 更新 Task 状态
        4. 返回 Artifact
        """
        
        # 更新状态：开始处理
        task.status = TaskStatus(
            state=TaskState.WORKING,
            timestamp=datetime.utcnow().isoformat(),
            message=Message(
                role="agent",
                parts=[TextPart(text="开始分析财务数据...")]
            )
        )
        
        # 解析输入
        instruction = ""
        data = None
        for part in message.parts:
            if part.kind == "text":
                instruction = part.text
            elif part.kind == "data":
                data = part.data
            elif part.kind == "file":
                # 实际项目中这里应该解析上传的文件
                pass
        
        # 模拟分析过程（实际项目中这里会调用 Pandas/NumPy）
        await asyncio.sleep(1)  # 模拟处理时间
        
        # 生成分析结果
        analysis_result = self._analyze_financial_data(instruction, data)
        
        # 更新状态：完成
        task.status = TaskStatus(
            state=TaskState.COMPLETED,
            timestamp=datetime.utcnow().isoformat(),
            message=Message(
                role="agent",
                parts=[TextPart(text="分析完成")]
            )
        )
        
        # 添加 Artifact
        task.artifacts = [
            Artifact(
                artifactId=str(uuid.uuid4()),
                name="财务分析报告",
                parts=[
                    TextPart(text=analysis_result["summary"]),
                    DataPart(data=analysis_result["structured_data"])
                ]
            )
        ]
        
        return task
    
    def _analyze_financial_data(self, instruction: str, data: Optional[dict]) -> dict:
        """实际的分析逻辑（简化版）"""
        # 这里应该接入真实的 Pandas 分析
        # 为演示目的，返回模拟结果
        
        return {
            "summary": (
                "# 财务分析报告\n\n"
                "## 核心发现\n\n"
                "- Q3 营收：42.3 亿元，同比增长 12.5%\n"
                "- 毛利率：41.8%，环比提升 0.6 个百分点\n"
                "- 净利率：18.2%，维持稳定\n\n"
                "## 风险提示\n\n"
                "原材料成本上升压力仍在，建议关注 Q4 供应链情况。"
            ),
            "structured_data": {
                "revenue": {
                    "q3": 4.23e9,
                    "yoy_growth": 0.125,
                    "qoq_growth": 0.031
                },
                "margins": {
                    "gross_margin": 0.418,
                    "net_margin": 0.182
                },
                "recommendation": "hold"
            }
        }

# ============ 启动服务器 ============
def main():
    task_store = InMemoryTaskStore()
    processor = FinanceTaskProcessor()
    
    server = A2AServer(
        agent_card=AGENT_CARD,
        task_store=task_store,
        task_processor=processor,
        host="0.0.0.0",
        port=8080
    )
    
    print(f"A2A Agent 启动成功！")
    print(f"Agent Card: http://localhost:8080/.well-known/agent.json")
    print(f"A2A Endpoint: http://localhost:8080/a2a")
    
    server.run()

if __name__ == "__main__":
    main()

4.3 客户端：调用远程 A2A Agent

"""
a2a_client_example.py
调用远程 A2A Agent 的完整客户端示例
支持：Agent Card 获取、同步调用、流式调用、多轮交互
"""

import requests
import json
from typing import Optional

class A2AClient:
    """A2A 协议客户端"""
    
    def __init__(self, base_url: str, token: Optional[str] = None):
        self.base_url = base_url.rstrip("/")
        self.token = token
        self.session = requests.Session()
        if token:
            self.session.headers.update({
                "Authorization": f"Bearer {token}"
            })
    
    def get_agent_card(self) -> dict:
        """获取目标 Agent 的 Agent Card"""
        url = f"{self.base_url}/.well-known/agent.json"
        resp = self.session.get(url)
        resp.raise_for_status()
        return resp.json()
    
    def send_task_sync(self, message: dict, timeout: int = 60) -> dict:
        """
        同步发送任务（等待完成）
        注意：如果任务执行时间超过 timeout，会抛异常
        """
        payload = {
            "jsonrpc": "2.0",
            "id": "req-sync-001",
            "method": "message/send",
            "params": {
                "message": message,
                "configuration": {
                    "blocking": True,
                    "acceptedOutputModes": ["text", "data"]
                }
            }
        }
        
        resp = self.session.post(
            f"{self.base_url}/a2a",
            json=payload,
            timeout=timeout
        )
        resp.raise_for_status()
        result = resp.json()
        
        if "error" in result:
            raise Exception(f"A2A Error {result['error']['code']}: {result['error']['message']}")
        
        return result["result"]
    
    def send_task_stream(self, message: dict, progress_callback=None) -> dict:
        """
        流式发送任务，实时获取进度
        progress_callback: 可选，接收进度更新的回调函数
        """
        payload = {
            "jsonrpc": "2.0",
            "id": "req-stream-001",
            "method": "message/stream",
            "params": {
                "message": message
            }
        }
        
        resp = self.session.post(
            f"{self.base_url}/a2a",
            json=payload,
            headers={"Accept": "text/event-stream"},
            stream=True,
            timeout=300
        )
        resp.raise_for_status()
        
        artifacts = {}
        current_task_status = None
        
        for line in resp.iter_lines():
            if not line:
                continue
            line = line.decode("utf-8")
            if not line.startswith("data: "):
                continue
            
            event = json.loads(line[6:])
            result = event.get("result", {})
            
            if result.get("kind") == "status-update":
                current_task_status = result["status"]["state"]
                if progress_callback:
                    progress_callback(result["status"])
                
                if result.get("final"):
                    break
            
            elif result.get("kind") == "artifact-update":
                artifact_id = result["artifact"]["artifactId"]
                if not result.get("append"):
                    artifacts[artifact_id] = result["artifact"]
                else:
                    artifacts[artifact_id]["parts"].extend(result["artifact"]["parts"])
        
        return {
            "status": current_task_status,
            "artifacts": list(artifacts.values())
        }
    
    def get_task(self, task_id: str) -> dict:
        """查询任务状态（用于异步轮询场景）"""
        payload = {
            "jsonrpc": "2.0",
            "id": "req-get-001",
            "method": "tasks/get",
            "params": {"taskId": task_id}
        }
        
        resp = self.session.post(f"{self.base_url}/a2a", json=payload)
        resp.raise_for_status()
        return resp.json()["result"]

# ============ 使用示例 ============
def main():
    client = A2AClient(
        base_url="http://localhost:8080",
        token="your-jwt-token-here"
    )
    
    # 1. 先获取 Agent Card，了解对方能力
    card = client.get_agent_card()
    print(f"目标 Agent: {card['name']}")
    print(f"支持的能力: {[s['name'] for s in card['skills']]}")
    
    # 2. 构造消息
    message = {
        "role": "user",
        "parts": [
            {
                "kind": "text",
                "text": "请分析 Q3 财务数据，重点看营收增长和毛利率变化，输出中文报告"
            }
        ],
        "messageId": "msg-client-001"
    }
    
    # 3. 同步调用
    print("\n[同步调用] 发送任务...")
    try:
        task_result = client.send_task_sync(message, timeout=30)
        print(f"Task 状态: {task_result['status']['state']}")
        for artifact in task_result.get("artifacts", []):
            for part in artifact["parts"]:
                if part["kind"] == "text":
                    print(f"\n分析结果：\n{part['text']}")
    except Exception as e:
        print(f"调用失败: {e}")

if __name__ == "__main__":
    main()

五、企业级实践：安全性、可靠性与性能优化

5.1 安全架构设计

A2A 协议的内置安全机制：

认证（Authentication）

# A2A 支持多种认证方案，以下是 JWT Bearer Token 的验证示例

from functools import wraps
import jwt

def verify_a2a_token(token: str, expected_audience: str) -> dict:
    """
    验证 A2A 请求中的 JWT Token
    返回解码后的 claims，失败则抛异常
    """
    try:
        claims = jwt.decode(
            token,
            key=YOUR_PUBLIC_KEY,  # 从 IdP 获取的公钥
            algorithms=["RS256"],
            audience=expected_audience,
            issuer="https://auth.mycorp.com"
        )
        
        # 验证 A2A 特定的 scope
        scopes = claims.get("scope", "").split()
        if "a2a:read" not in scopes:
            raise PermissionError("Token 缺少 a2a:read 权限")
        
        return claims
    except jwt.ExpiredSignatureError:
        raise PermissionError("Token 已过期")
    except jwt.InvalidTokenError as e:
        raise PermissionError(f"Token 无效: {e}")

多租户隔离

企业场景中，不同租户的 Agent 不能互相访问。解决方案：每个租户使用独立的 Agent Card 路径 + Token scope 验证。

https://api.mycorp.com/tenant-a/.well-known/agent.json
https://api.mycorp.com/tenant-b/.well-known/agent.json

5.2 可靠性：错误处理与重试

A2A 定义了完整的错误码体系。客户端需要针对不同类型的错误采取不同策略：

class A2ARetryStrategy:
    """A2A 请求的重试策略"""
    
    # JSON-RPC 标准错误码
    PARSE_ERROR = -32700
    INVALID_REQUEST = -32600
    METHOD_NOT_FOUND = -32601
    INVALID_PARAMS = -32602
    INTERNAL_ERROR = -32603
    
    # A2A 扩展错误码
    TASK_NOT_FOUND = -33001
    TASK_ALREADY_CANCELED = -33002
    UNSUPPORTED_MODE = -33003  # 比如请求了 streaming，但 Agent 不支持
    
    @classmethod
    def should_retry(cls, error_code: int) -> bool:
        """判断是否应该重试"""
        # 这些错误应该重试
        retryable = [
            cls.INTERNAL_ERROR,
            cls.TASK_NOT_FOUND  # 可能是最终一致性延迟
        ]
        return error_code in retryable
    
    @classmethod
    def should_fail_fast(cls, error_code: int) -> bool:
        """判断是否应该快速失败，不重试"""
        non_retryable = [
            cls.INVALID_PARAMS,      # 客户端参数错误，重试也没用
            cls.METHOD_NOT_FOUND,    # Agent 不支持这个方法
            cls.UNSUPPORTED_MODE     # Agent 能力不支持
        ]
        return error_code in non_retryable

5.3 性能优化

优化点一：流式响应减少感知延迟

对于生成长文档的任务，一定要用 message/stream，而不是等整个任务完成再获取结果。

# 不好的做法：同步等待（用户看到的是「转圈 30 秒」）
task = client.send_task_sync(message, timeout=60)
# 30 秒后才拿到完整结果

# 好的做法：流式响应（用户能看到实时进度）
def on_progress(status):
    state = status["state"]
    if state == "working":
        print(".", end="", flush=True)  # 实时进度指示

result = client.send_task_stream(message, progress_callback=on_progress)
# 用户立即看到任务已开始，进度实时更新

优化点二：Artifact 分块传输

生成长文档时，Agent 应该把输出分成多个 chunk 通过 SSE 逐步推送，而不是等全部生成完再返回。

# Agent 服务端：分块推送 Artifact
async def generate_report_stream(task_id: str, sse_sender):
    """逐步生成报告，实时推送"""
    sections = [
        "# 财务分析报告\n\n",
        "## 摘要\n\n本报告分析了...",
        "## 营收分析\n\n详细分析内容...",
        "## 毛利率分析\n\n...",
        "## 结论与建议\n\n..."
    ]
    
    artifact_id = str(uuid.uuid4())
    
    for i, section in enumerate(sections):
        is_first = (i == 0)
        is_last = (i == len(sections) - 1)
        
        event = {
            "jsonrpc": "2.0",
            "id": "stream-001",
            "result": {
                "kind": "artifact-update",
                "taskId": task_id,
                "artifact": {
                    "artifactId": artifact_id,
                    "parts": [{"kind": "text", "text": section}]
                },
                "append": not is_first,
                "lastChunk": is_last
            }
        }
        
        await sse_sender.send(event)
        await asyncio.sleep(0.5)  # 模拟生成延迟

六、A2A 生态全景：与周边协议的关系

6.1 A2A vs MCP vs ACP vs ANP

2026 年，Agent 通信协议领域出现了多个标准。它们之间的关系经常让人困惑。

它们的关系：

用户请求
  ↓
主 Agent（协调者）
  ├── 通过 A2A ───→ 远程专业 Agent（其他团队/公司维护）
  ├── 通过 MCP ───→ 本地工具（数据库、API）
  └── 通过 ANP ───→ 公共 Agent 服务市场（发现第三方 Agent）

关键洞察：这四个协议不是零和博弈。未来的 AI 系统会同时使用多个协议，各自解决各自的问题。

6.2 A2A 在企业中的典型部署架构

┌─────────────────────────────────────────────────────┐
│                  企业 AI 平台                        │
│                                                     │
│  ┌─────────────┐     ┌─────────────┐            │
│  │  HR Agent   │     │ 财务 Agent  │            │
│  │ (Salesforce)│     │  (SAP)      │            │
│  └──────┬──────┘     └──────┬──────┘            │
│         │ A2A                 │ A2A               │
│         ▼                     ▼                    │
│  ┌─────────────────────────────────────┐          │
│  │     A2A 注册表 + 认证网关            │          │
│  │  (企业内部，管理所有 Agent 的        │          │
│  │   Agent Card 和访问权限)             │          │
│  └─────────────────────────────────────┘          │
│                     │                             │
│         ┌───────────┼───────────┐               │
│         ▼           ▼           ▼                │
│  ┌──────────┐ ┌──────────┐ ┌──────────┐        │
│  │ 技术团队  │ │ 产品团队  │ │ 外部供应商│        │
│  │ Agent    │ │ Agent    │ │ Agent    │        │
│  │(LangChain)│ │(自研)   │ │(第三方)  │        │
│  └──────────┘ └──────────┘ └──────────┘        │
└─────────────────────────────────────────────────────┘

七、未来展望：A2A 的下一步

7.1 即将到来的特性（基于社区路线图）

1. A2A 1.1 规范：增加批处理支持（一次提交多个相关 Task）
2. Agent Card 版本化：支持 Agent 能力的版本管理和兼容性声明
3. 跨域信任框架：企业之间 Agent 互操作的信任建立和撤销机制
4. 更丰富的 Part 类型：支持音频流、视频帧等实时媒体类型

7.2 对开发者的影响

如果你正在构建 AI Agent 系统，现在就应该考虑 A2A 兼容性：

• 如果你在做单体 Agent：提前把 Agent 的能力通过 Agent Card 描述出来，未来接入多 Agent 系统会容易得多
• 如果你在做多 Agent 编排框架：直接把 A2A 作为 Agent 间通信的默认协议，不要自己发明私有协议
• 如果你在做企业 AI 平台：把 A2A 注册表作为平台核心组件，统一管理所有 Agent 的能力目录和访问权限

八、总结

A2A 协议看起来「不够性感」——它没有大模型基准测试的刷榜快感，没有新模型发布的轰动效应。但正是这样看似平淡的基础设施工作，往往决定了整个生态的长期竞争力。

A2A 的核心价值可以归纳为一句话：把 Agent 之间的协作从 N×M 的定制集成问题，变成了 N+M 的标准适配问题。

当 2026 年结束的时候，我们很可能会看到这样的图景：

• 你公司内部的 HR Agent、财务 Agent、技术 Agent 全都在 A2A 协议下互联互通
• 你通过 A2A 注册表就能发现合作厂商暴露的 Agent 能力，像调用本地函数一样发起跨企业协作
• 一个新的开源 Agent 只要实现 A2A 协议，就能无缝接入你的多 Agent 系统

这就是开放标准的力量。

参考资源

• A2A 官方文档：https://google.github.io/A2A/
• A2A Python SDK：https://github.com/google-a2a/a2a-python
• A2A 规范仓库：https://github.com/google/A2A
• Linux Foundation A2A 项目：https://lfaidata.foundation/projects/
• MCP 协议对比分析：Anthropic MCP Documentation
• A2A 社区讨论：https://github.com/google/A2A/discussions

本文撰写于 2026 年 7 月，基于 A2A 协议 Linux Foundation 托管版本。如有技术细节更新，请以官方文档为准。

如果你在实现 A2A Agent 过程中遇到问题，欢迎在评论区留言讨论。