Google A2A Agent協議深度解析：多Agent互操作、任务委派与MCP對比

摘要

• Google A2A（Agent-to-Agent）協議是2026年多Agent系統互操作的行業標準，解決「Agent之間怎麼協作」的核心問題
• A2A和MCP解決不同层面的問題：MCP管Agent与工具的連接，A2A管Agent与Agent的協作
• Agent Card是A2A的「名片」：聲明能力、端點、認證方式，讓其他Agent發現和呼叫
• 任务生命週期（Task Lifecycle）是A2A的核心抽象：提交→處理→完成/失敗/需要人工介入
• 本文提供从協議原理到Python實作的完整方案，含多Agent編排實戰

---

目錄

• 多Agent时代的互操作難題
• A2A協議核心概念
• Agent Card：Agent的「名片」
• 任务生命週期与狀態机
• A2A vs MCP：協議层對比
• Python A2A Server實作
• 多Agent協作實戰：客服分流系統
• 安全模型与生產部署
• 面試考點与選型指南
• 總結与延伸閱讀

---

多Agent时代的互操作難題

2026年的Agent生態现状

2025年MCP協議解決了「Agent怎麼连工具」的問題，2026年企业面临的新問題是：

我有客服Agent、訂單Agent、物流Agent、退款Agent，它们怎麼互相協作？

痛点	没有A2A时	有A2A后
Agent發現	硬編碼URL，改一个掛一片	Agent Card自動註冊發現
任务委派	自定义HTTP介面，每家不同	標準Task API
狀態同步	輪詢或自建WebSocket	標準事件流（SSE）
安全認證	各Agent各自實作	統一OAuth2/API Key
異構相容	Python Agent调不了Go Agent	協議层統一，語言無關

一个真實的協作場景

使用者问：「我上周买的耳机怎麼还没到？」

没有A2A时，客服Agent需要：

1. 自己调訂單API查訂單
2. 自己调物流API查物流
3. 自己判斷是否需要退款
4. 每個能力都要自己實作

有A2A后：

1. 客服Agent将任务委派给訂單Agent（查訂單）
2. 訂單Agent返回结果后，委派给物流Agent（查物流）
3. 如果物流異常，委派给退款Agent（发起退款）
4. 每個Agent專注自己的領域，透過A2A標準協議協作

---

A2A協議核心概念

協議架構

┌──────────────────────────────────────────────────────────┐
│                    A2A Protocol Stack                     │
│                                                            │
│  ┌─────────────┐    ┌─────────────┐    ┌──────────────┐  │
│  │ Agent Card  │    │  Task API   │    │  Event Stream│  │
│  │ (發現/註冊)  │    │ (任务管理)   │    │  (狀態推送)   │  │
│  └─────────────┘    └─────────────┘    └──────────────┘  │
│                                                            │
│  ┌─────────────────────────────────────────────────────┐  │
│  │              Transport: HTTP + JSON-RPC / SSE         │  │
│  └─────────────────────────────────────────────────────┘  │
└──────────────────────────────────────────────────────────┘

核心实体

实体	說明	類比
Agent Card	Agent的能力聲明和端點信息	微服務註冊中心的服務描述
Task	一次工作請求，有完整的生命週期	HTTP請求 + 异步Job
Message	Task中的單條互動（使用者輸入/Agent響應）	聊天訊息
Artifact	Task產出的结果（檔案、数据、报告）	API響應体
Part	Message或Artifact的内容片段（文本/檔案/数据）	MIME多部分内容

---

Agent Card：Agent的「名片」

Agent Card是一个JSON文档，描述Agent的能力和接入方式：

{
  "name": "order-agent",
  "description": "處理訂單查詢、建立和狀態變更的Agent",
  "url": "https://agents.example.com/order",
  "version": "1.0.0",
  "capabilities": {
    "streaming": true,
    "pushNotifications": false
  },
  "skills": [
    {
      "id": "query-order",
      "name": "查詢訂單",
      "description": "根据訂單号或使用者ID查詢訂單详情",
      "inputModes": ["text"],
      "outputModes": ["text", "data"]
    },
    {
      "id": "create-order",
      "name": "建立訂單",
      "description": "根据商品列表建立新訂單",
      "inputModes": ["text", "data"],
      "outputModes": ["text", "data"]
    }
  ],
  "authentication": {
    "schemes": ["bearer"]
  }
}

Agent發現流程

協調者Agent                    Agent Registry              目标Agent
     │                              │                          │
     │  1. 查找能"查詢訂單"的Agent    │                          │
     ├─────────────────────────────→│                          │
     │                              │  2. 返回匹配的Agent Card  │
     │←─────────────────────────────┤                          │
     │                              │                          │
     │  3. 向目标Agent發送Task                                   │
     ├────────────────────────────────────────────────────────→│
     │                              │                          │
     │  4. 接收Task狀態更新（SSE）                                │
     │←────────────────────────────────────────────────────────┤

Agent Card通常託管在 /.well-known/agent.json 路徑，類似OAuth的 .well-known/openid-configuration。

---

任务生命週期与狀態机

Task狀態机

         ┌──────────┐
         │ submitted │  任务已提交
         └────┬─────┘
              │
              ▼
         ┌──────────┐
    ┌────│ working  │────┐
    │    └──────────┘    │
    │         │          │
    ▼         ▼          ▼
┌────────┐ ┌────────┐ ┌──────────────────┐
│ input- │ │completed│ │ failed           │
│required│ └────────┘ └──────────────────┘
└────────┘
    │
    ▼
 需要更多輸入
（如缺少訂單号）

Task API核心方法

方法	說明	使用場景
tasks/send	發送新任务	委派工作给其他Agent
tasks/get	查詢任务狀態	輪詢等待结果
tasks/cancel	取消任务	使用者放棄或逾時
tasks/subscribe	訂閱任务事件流	即時接收進度更新

發送任务範例

{
  "jsonrpc": "2.0",
  "method": "tasks/send",
  "id": "req-001",
  "params": {
    "id": "task-uuid-001",
    "message": {
      "role": "user",
      "parts": [
        {
          "type": "text",
          "text": "查詢訂單号 ORD-20260703-8842 的物流狀態"
        }
      ]
    }
  }
}

任务结果

{
  "jsonrpc": "2.0",
  "id": "req-001",
  "result": {
    "id": "task-uuid-001",
    "status": "completed",
    "artifacts": [
      {
        "name": "物流信息",
        "parts": [
          {
            "type": "data",
            "data": {
              "order_id": "ORD-20260703-8842",
              "carrier": "顺丰速运",
              "tracking_no": "SF1234567890",
              "status": "运输中",
              "estimated_delivery": "2026-07-05"
            }
          }
        ]
      }
    ]
  }
}

---

A2A vs MCP：協議层對比

这是面試和架構選型的核心問題：

維度	MCP (Model Context Protocol)	A2A (Agent-to-Agent)
解決問題	Agent如何連接和使用工具/数据	Agent之間如何協作和委派任务
通訊模式	Client-Server（Agent是Client）	Peer-to-Peer（Agent互為Client/Server）
核心抽象	Tool、Resource、Prompt	Task、Message、Artifact
發現機制	MCP Server聲明工具列表	Agent Card聲明技能
狀態管理	无狀態（單次工具呼叫）	有狀態（Task生命週期）
傳輸協議	stdio / HTTP SSE	HTTP JSON-RPC / SSE
典型場景	讀檔案、查資料庫、调API	任务委派、多Agent編排
關係	互補，不是競爭	互補，不是競爭

一句话總結：MCP让Agent能用手（工具），A2A让Agent能組隊（協作）。

組合使用架構

使用者
 │
 ▼
┌──────────────┐
│ 協調者Agent   │
│ (Orchestrator)│
└──┬───┬───┬──┘
   │   │   │        A2A協議
   ▼   ▼   ▼
┌─────┐┌─────┐┌─────┐
│客服  ││訂單  ││物流  │    Agent层
│Agent││Agent││Agent│
└──┬──┘└──┬──┘└──┬──┘
   │      │      │         MCP協議
   ▼      ▼      ▼
┌─────┐┌─────┐┌─────┐
│知识库││訂單DB││物流API│   工具/資料層
│MCP  ││MCP  ││MCP  │
└─────┘└─────┘└─────┘

---

Python A2A Server實作

from dataclasses import dataclass, field
from typing import Optional
from enum import Enum
import uuid

class TaskStatus(Enum):
    SUBMITTED = "submitted"
    WORKING = "working"
    INPUT_REQUIRED = "input-required"
    COMPLETED = "completed"
    FAILED = "failed"

@dataclass
class Task:
    id: str
    status: TaskStatus
    messages: list[dict] = field(default_factory=list)
    artifacts: list[dict] = field(default_factory=list)

class OrderAgent:
  agent_card = {
      "name": "order-agent",
      "url": "https://agents.example.com/order",
      "skills": [
          {"id": "query-order", "name": "查詢訂單"},
          {"id": "create-order", "name": "建立訂單"},
      ],
  }

  def __init__(self, order_service):
      self.order_service = order_service
      self.tasks: dict[str, Task] = {}

  async def handle_task_send(self, params: dict) -> dict:
      task_id = params.get("id", str(uuid.uuid4()))
      message = params["message"]
      user_text = message["parts"][0]["text"]

      task = Task(id=task_id, status=TaskStatus.WORKING)
      task.messages.append(message)
      self.tasks[task_id] = task

      try:
          if "查詢" in user_text or "query" in user_text.lower():
              order_id = self._extract_order_id(user_text)
              if not order_id:
                  task.status = TaskStatus.INPUT_REQUIRED
                  task.messages.append({
                      "role": "agent",
                      "parts": [{"type": "text", "text": "请提供訂單号"}],
                  })
                  return self._task_to_response(task)

              order = await self.order_service.get_order(order_id)
              task.status = TaskStatus.COMPLETED
              task.artifacts.append({
                  "name": "訂單详情",
                  "parts": [{"type": "data", "data": order}],
              })
          else:
              task.status = TaskStatus.FAILED
              task.messages.append({
                  "role": "agent",
                  "parts": [{"type": "text", "text": "不支援的操作"}],
              })
      except Exception as e:
          task.status = TaskStatus.FAILED
          task.messages.append({
              "role": "agent",
              "parts": [{"type": "text", "text": f"處理失敗: {e}"}],
          })

      return self._task_to_response(task)

  def _extract_order_id(self, text: str) -> Optional[str]:
      import re
      match = re.search(r'ORD-\d{8}-\d+', text)
      return match.group(0) if match else None

  def _task_to_response(self, task: Task) -> dict:
      return {
          "id": task.id,
          "status": task.status.value,
          "messages": task.messages,
          "artifacts": task.artifacts,
      }

---

多Agent協作實戰：客服分流系統

架構設計

class CustomerServiceOrchestrator:
    def __init__(self, agents: dict[str, str]):
        self.agents = agents  # skill_id -> agent_url

    async def handle_user_query(self, query: str) -> str:
        intent = await self._classify_intent(query)

        agent_url = self.agents.get(intent)
        if not agent_url:
            return "抱歉，暫時無法處理您的問題"

        task_result = await self._send_a2a_task(agent_url, query)

        if task_result["status"] == "input-required":
            return task_result["messages"][-1]["parts"][0]["text"]

        if task_result["status"] == "completed":
            if task_result.get("artifacts"):
                return self._format_artifacts(task_result["artifacts"])
            return task_result["messages"][-1]["parts"][0]["text"]

        return "處理出现問題，请稍後重試"

    async def _classify_intent(self, query: str) -> str:
        if any(kw in query for kw in ["訂單", "下单", "購買"]):
            return "order"
        if any(kw in query for kw in ["物流", "快递", "發貨", "到哪"]):
            return "logistics"
        if any(kw in query for kw in ["退款", "退貨", "取消"]):
            return "refund"
        return "general"

    async def _send_a2a_task(self, agent_url: str, query: str) -> dict:
        import httpx
        async with httpx.AsyncClient() as client:
            resp = await client.post(f"{agent_url}/a2a", json={
                "jsonrpc": "2.0",
                "method": "tasks/send",
                "id": str(uuid.uuid4()),
                "params": {
                    "message": {
                        "role": "user",
                        "parts": [{"type": "text", "text": query}],
                    }
                },
            })
            return resp.json()["result"]

---

安全模型与生產部署

A2A安全要求

层级	措施	說明
傳輸层	HTTPS/TLS 1.3	所有Agent间通訊必須加密
認證层	OAuth2 Bearer Token	Agent Card聲明支援的認證方式
授權层	Skill級別權限	協調者Agent只能呼叫被授權的Skill
審計层	Task ID全鏈路追蹤	每個Task有唯一ID，可審計

生產部署清單

• Agent Card託管在 /.well-known/agent.json
• 每個Agent獨立部署，透過K8s Service暴露
• 使用API Gateway統一入口，做認證和限流
• Task逾時设置（預設30秒，長任務用subscribe模式）
• 監控：Task成功率、平均處理时间、Agent可用性

---

面試考點与選型指南

Q1：A2A和MCP是什么關係？

互補關係。MCP解決Agent与工具/数据的連接（垂直集成），A2A解決Agent与Agent的協作（水平集成）。生產系統通常两者同時使用。

Q2：A2A和直接调HTTP API有什么區別？

A2A提供了標準化的Task生命週期、Agent發現、狀態推送和Artifact傳遞。直接调HTTP需要自己實作这些，且不同团队的Agent無法互操作。

Q3：什么时候不需要A2A？

单Agent + MCP工具就够用时不需A2A。A2A的价值在多個專業Agent需要協作的場景，如客服分流、多步骤审批、跨域任务編排。

Q4：A2A的任务逾時怎麼處理？

發送方设置逾時，逾時后呼叫 tasks/cancel。对于長任務，使用 tasks/subscribe 訂閱SSE事件流，而不是輪詢 tasks/get。

---

A2A 協議演進與生態現狀（2026）

Linux Foundation 治理，與 MCP 形成「工具層 + 協作層」完整協議棧。

流式響應與 SSE 實戰

tasks/subscribe 訂閱 SSE 事件流，長任務即時推送進度。

多租戶與 Agent Registry

租戶隔離、Skill 級限流、Task 全鏈路審計。

2026 年趨勢

Agent Marketplace、低代碼平台 A2A 匯出、OWASP Agent 安全框架、邊緣 Agent 協作。

---

總結与延伸閱讀

Google A2A協議填補了多Agent互操作的標準化空白，与MCP形成「工具层 + 協作层」的完整協議栈。

設計要點回顧：

1. A2A解決Agent间協作，MCP解決Agent与工具的連接
2. Agent Card是發現機制，聲明在 /.well-known/agent.json
3. Task是核心抽象，有完整的生命週期狀態机
4. 多Agent編排用協調者模式，按意圖路由到專業Agent
5. 生產必須HTTPS + OAuth2 + Task全鏈路追蹤

相關閱讀：

• MCP協議AI Agent工具開發 — MCP協議實戰
• AI Agent協議對比 — A2A/MCP/ACP全面對比
• Python LangGraph多Agent編排 — 框架级多Agent方案

權威參考：

• Google A2A Protocol Specification
• Model Context Protocol