2025 年 4 月，Google 在 Google Cloud Next 上发布了 A2A 协议。同年 6 月，整个项目捐赠给了 Linux Foundation，由 AWS、Cisco、Google、Microsoft、Salesforce、SAP 等共同维护。截至 2026 年，已有超过 150 家组织支持该协议。这篇文章带你从头搞懂它——原理、设计思路、每一个报文字段，以及一段可以跑起来的代码演示。

---

一、为什么需要 A2A？

在 A2A 出现之前，企业构建多 Agent 系统时面临一个根本性的碎片化问题。

想象这样一个场景：你有一个部署在 Google Vertex AI 上的"招聘协调 Agent"，需要调用三件事——从 Salesforce 的 Agent 那里拿候选人列表，从 ServiceNow 的 Agent 那里安排面试，再从第三方背调服务的 Agent 那里触发背景调查。

在 A2A 之前，每一对 Agent 之间都需要写一套定制的胶水代码。Salesforce Agent 有自己的消息格式，ServiceNow Agent 有自己的 API 风格，背调服务又是另一套。每增加一个新 Agent，就要再写一套新的连接层。团队时间大量消耗在"翻译"上，而不是业务逻辑上。

A2A 用一句话解决这个问题：定义一个所有 Agent 都说的公共语言。

只要双方都支持 A2A，不管底层用的是什么框架（LangChain、AutoGen、自己写的），不管谁部署的，不管用什么语言实现，都能直接通信。

---

二、A2A 的设计思路

原则一：Agent 是不透明的黑盒（Opaque Agents）

A2A 的设计前提是：你不应该、也不需要知道对方 Agent 的内部实现。它用什么模型、怎么推理、有没有记忆、代码是什么语言写的——这些都不重要。A2A 只定义两个 Agent 之间交换什么，而不是内部怎么运行。这一点是跨厂商互操作的根本保障。

原则二：建在已有标准上，不发明新轮子

A2A 刻意没有发明新的传输协议。它直接使用：

• HTTP/HTTPS 做传输层
• JSON-RPC 2.0 做消息格式
• Server-Sent Events（SSE） 做流式推送
• OAuth 2.0 / OpenID Connect / API Key / mTLS 做认证

这意味着任何熟悉 Web 开发的工程师都不需要学新东西，直接用已有的 HTTP 客户端就能对接。

原则三：任务（Task）第一

A2A 里所有的工作都围绕"Task"展开。Agent 之间不是在"聊天"，而是在"委托任务"和"完成任务"。Task 有明确的生命周期状态，可以被查询、取消、订阅更新。这让长时间运行的复杂工作流变得可管理。

原则四：能力通过 Agent Card 公开声明

每个 Agent 在固定路径公开一份 JSON 文档，描述自己能做什么、支持哪些输入输出格式、需要什么认证方式。其他 Agent 读这份文档来决定能不能把任务交给它。这是发现机制，类似 OpenAPI 规范之于 REST API。

原则五：支持多种交互模式

A2A 同时支持：同步请求-响应（短任务立即返回）、异步轮询（发出去之后定期来查状态）、流式 SSE（长任务实时推送进度）、Push Notification（完全异步，完成了主动回调）。一个 Agent 可以根据自己的能力声明支持哪些模式，客户端按需选择。

---

三、核心数据模型：逐字段拆解

3.1 Agent Card：Agent 的"名片"

每个遵循 A2A 协议的 Agent 必须在 /.well-known/agent.json（或 /.well-known/agent-card.json）路径上托管一份 Agent Card。这是发现机制的入口。

一份完整的 Agent Card 长这样：

{
  "name": "财务分析 Agent",
  "description": "专业处理财务数据分析、报表生成和趋势预测任务",
  "version": "1.2.0",
  "url": "https://finance-agent.example.com/a2a",
  "provider": {
    "organization": "FinTech Corp",
    "url": "https://fintechcorp.example.com"
  },
  "capabilities": {
    "streaming": true,
    "pushNotifications": true,
    "stateTransitionHistory": false
  },
  "defaultInputModes": ["text", "data"],
  "defaultOutputModes": ["text", "data", "file"],
  "securitySchemes": {
    "bearerAuth": {
      "type": "http",
      "scheme": "bearer",
      "bearerFormat": "JWT"
    }
  },
  "security": [
    { "bearerAuth": [] }
  ],
  "skills": [
    {
      "id": "financial-analysis",
      "name": "财务数据分析",
      "description": "对财务数据进行多维度分析，输出结构化报告",
      "tags": ["finance", "analysis", "reporting"],
      "examples": [
        "分析 Q3 的营收趋势",
        "对比各部门的预算执行情况"
      ],
      "inputModes": ["text", "data"],
      "outputModes": ["text", "data", "file"]
    },
    {
      "id": "forecast",
      "name": "财务预测",
      "description": "基于历史数据进行未来财务预测",
      "tags": ["finance", "forecast", "ml"],
      "inputModes": ["data"],
      "outputModes": ["data", "file"]
    }
  ]
}

• url：这个 Agent 的 A2A 服务端点，客户端把所有 JSON-RPC 请求发到这里
• capabilities.streaming：是否支持 SSE 流式响应。true 才能用 message/stream 方法
• capabilities.pushNotifications：是否支持完成后主动回调客户端
• defaultInputModes：支持接收哪些类型的输入——text（纯文本）、data（JSON 结构化数据）、file（文件）、audio、video
• securitySchemes：认证方案，格式与 OpenAPI 3.0 的 Security Scheme 完全一致
• skills：Agent 能做什么的列表，每个 skill 有唯一 id、名称、描述和示例。客户端可以根据 skill 的 tags 和 examples 来判断该把什么任务交给这个 Agent

3.2 Task：工作的基本单元

Task 是 A2A 里最核心的对象。每个 Task 有：

{
  "id": "363422be-b0f9-4692-a24d-278670e7c7f1",
  "contextId": "c295ea44-7543-4f78-b524-7a38915ad6e4",
  "status": {
    "state": "completed",
    "timestamp": "2025-04-17T17:47:09Z",
    "message": {
      "role": "agent",
      "parts": [
        { "kind": "text", "text": "分析完成，详见 artifact。" }
      ]
    }
  },
  "artifacts": [...],
  "history": [...]
}

• id：Task 的唯一标识，UUID 格式。用于后续查询（tasks/get）和取消（tasks/cancel）
• contextId：会话上下文 ID。同一个多轮对话里的所有 Task 共享同一个 contextId。你可以把 taskId 理解成"这次具体做的事"，contextId 理解成"整个对话脉络"
• status.state：当前状态，取值有 submitted、working、input-required、completed、failed、canceled、rejected
• artifacts：任务完成后的输出物。一个 Task 可以有多个 Artifact，每个 Artifact 包含多个 Part
• history：这个 Task 经历过的消息历史

Task 的状态机：

submitted → working → completed
                    → failed
                    → input-required → working → completed
         → canceled（任意时刻客户端主动取消）
         → rejected（服务端拒绝执行）

input-required 状态是多轮交互的关键：Agent 处理到一半发现需要更多信息，就把 Task 切到这个状态，告诉客户端"我还需要你告诉我 X"。客户端补充信息后，继续用同一个 taskId 发送，Task 重新进入 working。

3.3 Message：通信的载体

Message 是客户端和 Agent 之间实际传递的内容结构：

{
  "role": "user",
  "parts": [
    {
      "kind": "text",
      "text": "请分析附件中的 Q3 财务数据，重点看毛利率变化趋势"
    },
    {
      "kind": "file",
      "file": {
        "name": "q3_finance.xlsx",
        "mimeType": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
        "data": "<base64编码的文件内容>"
      }
    }
  ],
  "messageId": "9229e770-767c-417b-a0b0-f0741243c589",
  "taskId": "363422be-b0f9-4692-a24d-278670e7c7f1",
  "contextId": "c295ea44-7543-4f78-b524-7a38915ad6e4"
}

• role：发送方角色，只有 user（客户端 Agent）和 agent（服务端 Agent）两种
• parts：内容块数组。一条消息可以同时包含多个不同类型的块
• Part 的 kind 取值：text（文本）、file（文件，可以内嵌 base64 或提供 URL）、data（任意 JSON 结构数据）

3.4 Artifact：任务输出物

Artifact 是 Agent 完成任务后交付的成果，结构和 Message 里的 Parts 类似：

{
  "artifactId": "9b6934dd-37e3-4eb1-8766-962efaab63a1",
  "name": "财务分析报告",
  "description": "Q3毛利率趋势分析结果",
  "parts": [
    {
      "kind": "text",
      "text": "## Q3 毛利率分析\n\n根据数据，Q3 毛利率为 42.3%，环比下降 1.2 个百分点..."
    },
    {
      "kind": "data",
      "data": {
        "grossMargin": 0.423,
        "qoqChange": -0.012,
        "trend": "declining",
        "monthlyBreakdown": [0.435, 0.428, 0.406]
      }
    }
  ]
}

Artifact 和 Message 的区别在于语义：Message 是通信过程中的"说话"，Artifact 是任务完成后交付的"产品"。

---

四、协议交互方式详解

4.1 方法列表

A2A 通过 JSON-RPC 2.0 定义了一套方法，全部 POST 到 Agent 的 url 端点：

方法名	用途
message/send	发送消息，同步等待完整结果
message/stream	发送消息，以 SSE 流式接收进度和结果
tasks/get	查询某个 Task 的当前状态和结果
tasks/list	列出所有（或过滤条件下的）Task
tasks/cancel	取消一个正在进行的 Task
tasks/resubscribe	重新订阅某个 Task 的 SSE 流（断线重连用）
tasks/pushNotificationConfig/set	配置 Task 完成后的推送回调地址
agent/authenticatedExtendedCard	获取需要认证才能看到的完整 Agent Card

所有 JSON-RPC 请求的基本结构都是：

{
  "jsonrpc": "2.0",
  "id": "请求ID（客户端自己生成，用于对应响应）",
  "method": "方法名",
  "params": { ... }
}

响应结构：

{
  "jsonrpc": "2.0",
  "id": "与请求相同的ID",
  "result": { ... }
}

出错时：

{
  "jsonrpc": "2.0",
  "id": "与请求相同的ID",
  "error": {
    "code": -32602,
    "message": "Invalid params",
    "data": { "detail": "taskId 格式不合法" }
  }
}

4.2 同步请求（message/send）

最简单的场景：客户端发一条消息，等着 Agent 处理完后返回完整结果。

请求：

{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "message/send",
  "params": {
    "message": {
      "role": "user",
      "parts": [
        {
          "kind": "text",
          "text": "帮我把这段 Python 代码转换成 Go 语言：\ndef add(a, b):\n    return a + b"
        }
      ],
      "messageId": "msg-001"
    },
    "configuration": {
      "blocking": true,
      "acceptedOutputModes": ["text"]
    }
  }
}

configuration 里的关键字段：

• blocking: true：客户端愿意一直等到 Task 完成，如果服务端支持就直接返回最终结果；如果不支持就先返回 Task 的初始状态，客户端再轮询
• acceptedOutputModes：客户端能接受哪些格式的输出

响应（Task 已完成）：

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "kind": "task",
    "id": "task-abc-123",
    "contextId": "ctx-xyz-789",
    "status": {
      "state": "completed",
      "timestamp": "2025-04-17T10:30:00Z"
    },
    "artifacts": [
      {
        "artifactId": "artifact-001",
        "name": "Go 代码",
        "parts": [
          {
            "kind": "text",
            "text": "func add(a, b int) int {\n    return a + b\n}"
          }
        ]
      }
    ]
  }
}

4.3 流式响应（message/stream）

适合长时间运行的任务，比如生成长文档、复杂分析。HTTP 响应头是 Content-Type: text/event-stream，内容是一连串 SSE 事件。

请求：（格式与 message/send 相同，方法名换成 message/stream）

服务端推送的 SSE 事件流：

// 事件1：Task 被接受，开始处理
data: {"jsonrpc":"2.0","id":1,"result":{"kind":"task","id":"task-001","status":{"state":"submitted","timestamp":"2025-04-17T10:30:00Z"}}}

// 事件2：进入处理中状态
data: {"jsonrpc":"2.0","id":1,"result":{"kind":"status-update","taskId":"task-001","status":{"state":"working"}}}

// 事件3：第一个内容块到达（append: false 表示这是新 Artifact 的第一块）
data: {"jsonrpc":"2.0","id":1,"result":{"kind":"artifact-update","taskId":"task-001","artifact":{"artifactId":"art-001","parts":[{"kind":"text","text":"# Q3 财务分析报告\n\n## 摘要\n"}]},"append":false,"lastChunk":false}}

// 事件4：追加更多内容（append: true 表示追加到同一个 Artifact）
data: {"jsonrpc":"2.0","id":1,"result":{"kind":"artifact-update","taskId":"task-001","artifact":{"artifactId":"art-001","parts":[{"kind":"text","text":"Q3 毛利率为 42.3%，环比下降 1.2 个百分点，主要受原材料成本上升影响..."}]},"append":true,"lastChunk":false}}

// 事件5：最后一块内容（lastChunk: true）
data: {"jsonrpc":"2.0","id":1,"result":{"kind":"artifact-update","taskId":"task-001","artifact":{"artifactId":"art-001","parts":[{"kind":"text","text":"\n\n## 结论\n建议关注供应链成本优化机会。"}]},"append":true,"lastChunk":true}}

// 事件6：Task 完成，final: true，服务端关闭 SSE 连接
data: {"jsonrpc":"2.0","id":1,"result":{"kind":"status-update","taskId":"task-001","status":{"state":"completed","timestamp":"2025-04-17T10:30:45Z"},"final":true}}

流式响应里，客户端需要把 append: false 的块作为 Artifact 内容的开始，把后续 append: true 的块拼接到前面，直到 lastChunk: true 为止。

4.4 多轮交互（input-required 状态）

当 Agent 需要更多信息时：

第一轮：客户端发起，Agent 表示需要补充信息

请求（同上，省略）

响应：

{
  "jsonrpc": "2.0",
  "id": "req-001",
  "result": {
    "kind": "task",
    "id": "task-flight-001",
    "contextId": "ctx-flight-session",
    "status": {
      "state": "input-required",
      "message": {
        "role": "agent",
        "parts": [
          {
            "kind": "text",
            "text": "我可以帮你订机票！请告诉我出发城市、目的地和出行日期？"
          }
        ],
        "messageId": "agent-msg-001"
      }
    }
  }
}

第二轮：客户端提供补充信息，使用同一个 taskId 和 contextId

{
  "jsonrpc": "2.0",
  "id": "req-002",
  "method": "message/send",
  "params": {
    "message": {
      "role": "user",
      "parts": [
        {
          "kind": "text",
          "text": "从上海出发飞东京，10月15日去，10月22日回"
        }
      ],
      "messageId": "user-msg-002",
      "taskId": "task-flight-001",
      "contextId": "ctx-flight-session"
    },
    "configuration": { "blocking": true }
  }
}

关键：续接任务时，message 里必须带上 taskId 和 contextId，Agent 才能把这条消息关联到正在等待 input-required 的 Task 上继续处理。

4.5 错误码

A2A 在 JSON-RPC 标准错误码之外定义了协议专属错误码：

错误码	含义
-32001	TaskNotFound：任务 ID 不存在
-32002	TaskNotCancelable：任务已处于终止状态，无法取消
-32003	PushNotificationNotSupported：Agent 不支持推送通知
-32004	UnsupportedOperation：不支持的方法
-32005	ContentTypeNotSupported：不支持该输入/输出格式
-32006	InvalidAgentResponse：Agent 内部返回了非法数据
-32700	ParseError：JSON 解析失败
-32600	InvalidRequest：请求格式不合法
-32601	MethodNotFound：方法不存在
-32602	InvalidParams：参数不合法

---

五、安全机制

A2A 对安全的要求很明确：

传输层：必须使用 HTTPS，不得使用明文 HTTP（生产环境）。

客户端认证：客户端读取 Agent Card 里的 securitySchemes 和 security，按声明的方式提供凭证。支持的方案：API Key（放在 Header 里）、HTTP Basic/Bearer、OAuth 2.0（Authorization Code、Client Credentials、Device Code 流程）、OpenID Connect、mTLS（双向 TLS 证书认证）。

Agent Card 签名（v1.0 新特性）：Agent Card 可以被签名，防止中间人篡改。客户端在使用前可以验证签名，确认 Agent Card 是真实的、没被修改过的。签名格式基于 JWS（JSON Web Signature）。

Push Notification 安全：当 Agent 使用 Push Notification 回调客户端时，客户端可以在配置回调地址时同时提供 JWKS URL，要求 Agent 对每次回调请求做签名。客户端验证签名来确认通知来自合法的 Agent，而不是伪造的。

---

好的，从头重写，结构更清晰，文件职责更直观。

---

六、实现一个 A2A 交互演示

用最少的代码演示完整的 A2A 通信流程。文件结构如下：

a2a_demo/
├── agent_card.json   ← Agent 名片
├── server.py         ← A2A 服务端，接收并处理任务
├── client.py         ← A2A 客户端，发起调用
└── main.py          

安装依赖：

pip install fastapi uvicorn httpx

---

6.1 agent_card.json

{
  "name": "代码翻译 Agent",
  "description": "将代码从一种编程语言翻译成另一种语言的专业 Agent",
  "version": "1.0.0",
  "url": "http://localhost:8000/a2a",
  "provider": {
    "organization": "Demo Corp",
    "url": "https://demo.example.com"
  },
  "capabilities": {
    "streaming": false,
    "pushNotifications": false
  },
  "defaultInputModes": ["text"],
  "defaultOutputModes": ["text"],
  "skills": [
    {
      "id": "code-translate",
      "name": "代码翻译",
      "description": "将源代码从一种语言翻译成另一种语言",
      "tags": ["code", "translation", "programming"],
      "examples": [
        "把这段 Python 代码翻译成 Go",
        "将以下 JavaScript 代码转换为 TypeScript"
      ],
      "inputModes": ["text"],
      "outputModes": ["text"]
    }
  ]
}

---

6.2 server.py

服务端只做三件事：发布 Agent Card、处理 message/send、处理 tasks/get。

simulate_translation() 是业务逻辑占位符，真实场景换成 LLM 调用即可。

# server.py

import json
import uuid
from datetime import datetime, timezone
from pathlib import Path

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

# 任务存储，生产环境换成数据库
_tasks: dict = {}


#  业务逻辑（替换成真实 LLM 调用）
def simulate_translation(text: str) -> str:
    if "python" in text.lower() and "go" in text.lower():
        return (
            "// Go 翻译结果\n\n"
            "package main\n\n"
            "func add(a, b int) int {\n"
            "    return a + b\n"
            "}\n"
        )
    return f"// 翻译完成（输入摘要：{text[:40]}...）"


# 端点一：发布 Agent Card
@app.get("/.well-known/agent.json")
async def get_agent_card():
    """直接把 agent_card.json 的内容原样返回，不做任何修改"""
    card = json.loads(Path("agent_card.json").read_text(encoding="utf-8"))
    return JSONResponse(content=card)


# 端点二：A2A 主入口，所有 JSON-RPC 请求都发到这里
@app.post("/a2a")
async def a2a_endpoint(request: Request):
    body = await request.json()
    method = body.get("method")

    if method == "message/send":
        return _handle_send(body)
    elif method == "tasks/get":
        return _handle_get(body)
    else:
        return _error(body["id"], -32601, "Method not found")


# 处理 message/send
def _handle_send(body: dict) -> JSONResponse:
    params  = body["params"]
    message = params["message"]
    req_id  = body["id"]

    # 从 parts 里拼出用户输入的文本
    user_text = "".join(
        p.get("text", "")
        for p in message.get("parts", [])
        if p.get("kind") == "text"
    )

    task_id    = str(uuid.uuid4())
    context_id = message.get("contextId") or str(uuid.uuid4())
    now        = datetime.now(timezone.utc).isoformat()

    task = {
        "kind":      "task",
        "id":        task_id,
        "contextId": context_id,
        "status": {
            "state":     "completed",
            "timestamp": now
        },
        "artifacts": [
            {
                "artifactId": str(uuid.uuid4()),
                "name":  "翻译结果",
                "parts": [
                    {"kind": "text", "text": simulate_translation(user_text)}
                ]
            }
        ],
        "history": [
            {
                "role":      "user",
                "parts":     message.get("parts", []),
                "messageId": message.get("messageId", str(uuid.uuid4())),
                "taskId":    task_id,
                "contextId": context_id
            }
        ]
    }

    _tasks[task_id] = task

    return JSONResponse(content={"jsonrpc": "2.0", "id": req_id, "result": task})


# 处理 tasks/get
def _handle_get(body: dict) -> JSONResponse:
    req_id  = body["id"]
    task_id = body["params"].get("id")

    if task_id not in _tasks:
        return _error(req_id, -32001, "TaskNotFound", {"taskId": task_id})

    return JSONResponse(content={
        "jsonrpc": "2.0",
        "id":      req_id,
        "result":  _tasks[task_id]
    })


# 工具函数

def _error(req_id, code: int, message: str, data: dict = None) -> JSONResponse:
    err = {"code": code, "message": message}
    if data:
        err["data"] = data
    return JSONResponse(content={"jsonrpc": "2.0", "id": req_id, "error": err})

---

6.3 client.py

客户端封装成一个类，屏蔽 JSON-RPC 组包细节，对外只暴露语义清晰的方法。

# client.py

import uuid
import json
import httpx


class A2AClient:

    def __init__(self, agent_card_url: str):
        self.agent_card_url = agent_card_url
        self.agent_card: dict = {}
        self.a2a_url: str     = ""

    #  发现：读取 Agent Card 
    async def discover(self) -> None:
        """GET /.well-known/agent.json，获取对方的能力描述和通信地址"""
        async with httpx.AsyncClient() as http:
            resp = await http.get(self.agent_card_url)
            self.agent_card = resp.json()

        self.a2a_url = self.agent_card["url"]

        print("[ Agent Card ]")
        print(json.dumps(self.agent_card, ensure_ascii=False, indent=2))
        print()

    #  发送消息
    async def send_message(self, text: str,
                           task_id: str    = None,
                           context_id: str = None) -> dict:
        """
        POST /a2a，method = message/send
        多轮对话时传入 task_id + context_id，让服务端识别这是同一个任务的续接
        """
        message = {
            "role":      "user",
            "messageId": str(uuid.uuid4()),
            "parts":     [{"kind": "text", "text": text}]
        }
        if task_id:
            message["taskId"] = task_id
        if context_id:
            message["contextId"] = context_id

        payload = {
            "jsonrpc": "2.0",
            "id":      str(uuid.uuid4()),
            "method":  "message/send",
            "params": {
                "message":       message,
                "configuration": {"blocking": True}
            }
        }

        print("[ 发送请求 ] message/send")
        print(json.dumps(payload, ensure_ascii=False, indent=2))
        print()

        async with httpx.AsyncClient() as http:
            resp = await http.post(
                self.a2a_url,
                json=payload,
                headers={"Content-Type": "application/json"}
            )
            result = resp.json()

        print("[ 收到响应 ]")
        print(json.dumps(result, ensure_ascii=False, indent=2))
        print()

        return result

    # 查询任务状态

    async def get_task(self, task_id: str) -> dict:
        """POST /a2a，method = tasks/get"""
        payload = {
            "jsonrpc": "2.0",
            "id":      str(uuid.uuid4()),
            "method":  "tasks/get",
            "params":  {"id": task_id}
        }

        print("[ 发送请求 ] tasks/get")
        print(json.dumps(payload, ensure_ascii=False, indent=2))
        print()

        async with httpx.AsyncClient() as http:
            resp = await http.post(self.a2a_url, json=payload)
            result = resp.json()

        print("[ 收到响应 ]")
        print(json.dumps(result, ensure_ascii=False, indent=2))
        print()

        return result

    # 工具方法

    @staticmethod
    def extract_text(response: dict) -> str:
        """从响应的 artifacts 里提取文本内容"""
        for artifact in response.get("result", {}).get("artifacts", []):
            texts = [
                p["text"] for p in artifact.get("parts", [])
                if p.get("kind") == "text"
            ]
            if texts:
                return "\n".join(texts)
        return "（无输出）"

---

6.4 main.py

把四个演示步骤串起来，服务端和客户端在同一个进程里启动，方便直接运行。

# main.py

import asyncio
import time
import threading
import uvicorn

from server import app
from client import A2AClient

AGENT_CARD_URL = "http://localhost:8000/.well-known/agent.json"


async def demo():

    client = A2AClient(AGENT_CARD_URL)

    # Step 1：发现远端 Agent
    print("=" * 60)
    print("  Step 1  读取 Agent Card（服务发现）")
    print("=" * 60)
    await client.discover()

    # Step 2：发送翻译任务
    print("=" * 60)
    print("  Step 2  发送翻译任务（message/send）")
    print("=" * 60)

    response = await client.send_message(
        "把下面 Python 代码翻译成 Go：\n\ndef add(a, b):\n    return a + b"
    )

    task_id    = response["result"]["id"]
    context_id = response["result"]["contextId"]
    state      = response["result"]["status"]["state"]

    print(f"任务状态：{state}")
    print(f"翻译结果：\n{client.extract_text(response)}\n")

    # Step 3：主动查询任务状态
    print("=" * 60)
    print("  Step 3  查询任务详情（tasks/get）")
    print("=" * 60)

    detail = await client.get_task(task_id)
    print(f"任务 ID：  {task_id}")
    print(f"当前状态：{detail['result']['status']['state']}\n")

    # Step 4：查询不存在的任务，观察错误码
    print("=" * 60)
    print("  Step 4  查询不存在的 Task（预期 -32001 错误）")
    print("=" * 60)

    await client.get_task("00000000-0000-0000-0000-000000000000")


if __name__ == "__main__":
    # 后台线程启动服务端
    thread = threading.Thread(
        target=lambda: uvicorn.run(
            app, host="0.0.0.0", port=8000, log_level="warning"
        ),
        daemon=True
    )
    thread.start()
    time.sleep(1)  # 等待服务端就绪

    asyncio.run(demo())

---

6.5 运行

cd a2a_demo
python main.py

四个步骤会依次执行，每一步都打印完整的 JSON-RPC 报文。

---

七、A2A 生态

A2A 的生态已经相当成熟：

协议版本走到了 v1.0，带来了 Signed Agent Cards（防篡改签名）和更稳定的接口。Google Cloud、Azure AI Foundry、Amazon Bedrock AgentCore 都原生集成了 A2A。LangChain 的 Agent Server 也支持了 A2A 端点，LangGraph 部署的 Agent 可以直接暴露 A2A 接口供其他 Agent 调用。

值得一提的是，IBM 的 ACP（Agent Communication Protocol）在 2025 年 8 月并入了 A2A 项目。原本 A2A 最大的潜在竞争者，选择了合并而非对抗。这基本确立了 A2A 作为跨厂商 Agent 通信标准的地位。

还有一个扩展值得关注：2025 年 9 月，Google Cloud 和 Coinbase 联合推出了 AP2（Agent Payments Protocol），作为 A2A 的正式扩展（在 AgentCard 里注册为 extension）。这让 Agent 之间不只能交换信息，还能发起支付——这是 Agentic Commerce 的基础设施层。

---

官方规范：a2a-protocol.org
GitHub：github.com/a2aproject/A2A
Python SDK：pip install a2a-sdk