A2A 多 Agent 互相通信调用实现

1. 引言

随着 AI Agent 应用场景从单点任务向复杂工作流演进，多 Agent 跨平台协作已成为工程实践中的刚需。不同框架（如 LangChain、CrewAI、Vertex AI Agent Engine）和不同语言实现的 Agent 之间缺乏标准通信协议，导致集成成本高、重复造轮子。Google 推出的 Agent2Agent（A2A）协议通过标准化通信机制解决了这一问题。

本文首先介绍 A2A 协议的核心原理（基于 JSON-RPC 2.0 over HTTP(S)），然后结合 Cloud Run 与 Agent Engine 给出两个 Agent 互相调用的完整实战步骤与代码示例，最后梳理常见踩坑与排查方法。读完本文你将掌握：A2A 多 Agent 通信原理、JSON-RPC 2.0 Agent 互调实战、Cloud Run 部署 A2A Agent 服务的方法、Agent Engine A2A 协议配置要点，以及多 Agent 角色分工与 A2A 调用设计思路。

2. A2A 协议核心概念

A2A 协议是一项开放标准，旨在让不同平台和框架之间的 AI Agent 能够通信和协作，而无需考虑其底层技术实现。其核心设计理念是“发现-通信-协作”三阶段模型。

2.1 基于 JSON-RPC 2.0 的通信基础

A2A 协议选用 JSON-RPC 2.0 over HTTP(S) 作为核心通信方法。JSON-RPC 2.0 是一种轻量级、无状态的远程过程调用协议，请求和响应均以 JSON 格式编码。一个典型的 A2A 请求包含：

jsonrpc: 固定为 “2.0”
method: 要调用的 RPC 方法名，如 “send_task”
params: 方法参数，通常包含 task_id、message 等
id: 请求唯一标识符，用于匹配请求与响应

{
  "jsonrpc": "2.0",
  "method": "send_task",
  "params": {
    "task_id": "task-001",
    "message": {
      "role": "user",
      "content": "请分析上季度销售数据"
    }
  },
  "id": "req-abc-123"
}

为什么选择 JSON-RPC 2.0？ 它足够简单、跨语言支持广泛、无状态特性天然适合 HTTP，且社区积累了大量现成库与工具。相比 gRPC，JSON-RPC 2.0 的调试门槛更低，适合异构系统集成。

2.2 AgentCard：能力声明与发现

每个 A2A Agent 必须暴露一个 /.well-known/agent-card 端点（或通过服务元数据注册），返回一个描述自身能力的 JSON 文档——即 AgentCard。AgentCard 包含以下关键字段：

name: Agent 名称
description: 功能描述，供其他 Agent 根据文本匹配选择调用
url: 服务端点 URL
capabilities: 能力数组，声明该 Agent 支持哪些 RPC 方法（如 send_task、get_task、cancel_task）
authentication: 可选，说明支持的认证方式

实践中建议为每个 Agent 定义明确的 AgentCard，并保持其内容与实际的 API 端点一致。AgentCard 的存在使得 A2A 协议具备“服务发现”能力——调用方可以拉取目标 Agent 的 AgentCard，判断其是否具备所需能力后再发起任务。

2.3 任务（Task）生命周期管理

A2A 协议将一次 Agent 调用抽象为一个 Task，其完整生命周期包括：

创建（createTask/send_task）：发起方发送任务请求，接收方返回 task_id，任务进入“pending”或“working”状态
查询（get_task）：发起方轮询任务状态与中间结果/最终结果
更新（task_update）：接收方主动推送任务状态变更（需发起方支持回调或流式）
取消（cancel_task）：发起方请求终止任务，接收方清理资源后返回确认

A2A 支持同步（发起方阻塞等待结果）和异步（发起方提交任务后轮询或等待推送）两种模式。流式场景下，接收方可以通过分块 HTTP 响应（Server-Sent Events）或返回多段 JSON 来逐步返回结果。

2.4 消息与多模态支持

A2A 中的消息（Message）结构支持文本、结构化数据（如 JSON）、以及指向外部资源的引用（如图片 URL、二进制 Base64 编码）。这保证了 Agent 之间可以交换非文本信息——例如一个视觉 Agent 接收图片 URL，分析后返回结构化结果。

3. 开发环境与前期准备

开始构建 A2A Agent 前，需要准备以下环境与资源：

Google Cloud 项目：A2A 协议的 Codelab 及官方示例通常基于 Google Cloud Run 和 Agent Engine。你需要拥有一个激活状态的项目，并已启用 Cloud Run API 和 Vertex AI API（如果使用 Agent Engine）。
安装并配置 gcloud CLI：执行 gcloud init 认证并设置默认项目。确认已安装 Python 3.10+ 及 pip。
获取示例代码仓库：运行 git clone https://github.com/google/agent2agent-codelab（若此仓库不可用，可基于下文提供的简化代码搭建）。
安装依赖项：在项目目录下创建 requirements.txt，包含以下核心库：
1
2
3
flask>=2.3 requests>=2.31 gunicorn>=20.1
如需在 Agent Engine 上使用 A2A，还需安装 google-cloud-aiplatform SDK。

执行 pip install -r requirements.txt。

（可选）准备 Docker 环境：Cloud Run 部署会构建容器镜像，建议本地安装 Docker Desktop 用于调试。

4. 实战：部署第一个 A2A Agent 服务（Cloud Run）

本节将部署一个简单的 Python Agent 服务，它暴露 A2A 协议端点，并能被其他 Agent 发现和调用。

4.1 服务代码实现

创建文件 agent_service.py，实现 HTTP 服务器与 A2A 端点：

from flask import Flask, request, jsonify
import uuid
import json

app = Flask(__name__)

# 模拟的任务存储（生产环境应使用持久化存储）
tasks = {}

@app.route('/.well-known/agent-card', methods=['GET'])
def agent_card():
    """返回本 Agent 的能力声明"""
    card = {
        "name": "销售数据分析助手",
        "description": "分析销售数据，生成月度/季度报告",
        "url": request.host_url.rstrip('/'),
        "capabilities": ["send_task", "get_task", "cancel_task"],
        "authentication": {}
    }
    return jsonify(card)

@app.route('/', methods=['POST'])
def handle_generic():
    """
    通用 A2A 请求处理器
    建议根据 method 分发到不同处理函数
    """
    try:
        data = request.get_json(force=True)
    except Exception:
        return jsonify({"jsonrpc": "2.0", "error": {"code": -32700, "message": "Parse error"}, "id": None}), 400

    method = data.get('method')
    params = data.get('params', {})
    req_id = data.get('id', str(uuid.uuid4()))

if method == 'send_task':
        # 创建任务
        task_id = params.get('task_id', str(uuid.uuid4()))
        tasks[task_id] = {
            "status": "working",
            "messages": [params.get('message')]
        }
        # 执行代理逻辑（此处省略实际 LLM 调用，仅返回确认）
        tasks[task_id]["status"] = "completed"
        tasks[task_id]["messages"].append({
            "role": "agent",
            "content": f"已分析任务 {task_id}，结果：销售趋势平稳，建议增加客户关怀投入。

"
        })
        return jsonify({
            "jsonrpc": "2.0",
            "result": {
                "task_id": task_id,
                "status": "completed"
            },
            "id": req_id
        })

elif method == 'get_task':
        task_id = params.get('task_id')
        task = tasks.get(task_id)
        if not task:
            return jsonify({
                "jsonrpc": "2.0",
                "error": {"code": -32000, "message": "Task not found"},
                "id": req_id
            }), 404
        return jsonify({
            "jsonrpc": "2.0",
            "result": task,
            "id": req_id
        })

elif method == 'cancel_task':
        task_id = params.get('task_id')
        tasks.pop(task_id, None)
        return jsonify({
            "jsonrpc": "2.0",
            "result": {"task_id": task_id, "status": "cancelled"},
            "id": req_id
        })

    else:
        return jsonify({
            "jsonrpc": "2.0",
            "error": {"code": -32601, "message": "Method not found"},
            "id": req_id
        }), 405

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=8080, debug=True)

4.2 本地验证

在终端启动服务：python agent_service.py。然后使用 curl 验证 AgentCard 和任务创建：

# 获取 AgentCard
curl http://localhost:8080/.well-known/agent-card | jq .

# 创建任务
curl -X POST http://localhost:8080/ \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "method": "send_task",
    "params": {
        "task_id": "test-001",
        "message": {
            "role": "user",
            "content": "分析7月销售数据"
        }
    },
    "id": "req-1"
  }' | jq .

4.3 部署到 Cloud Run

编写 Dockerfile（省略，基于 python:3.11-slim，暴露 8080 端口，CMD 使用 gunicorn）
构建并推送镜像：gcloud builds submit --tag gcr.io/[PROJECT_ID]/a2a-agent-v1
部署服务：gcloud run deploy a2a-agent-v1 --image gcr.io/[PROJECT_ID]/a2a-agent-v1 --platform managed --region us-central1 --allow-unauthenticated
记下返回的 HTTPS 服务 URL，后续其他 Agent 将通过此 URL 访问。

注意：生产环境建议关闭 --allow-unauthenticated，并配置服务账号或 API Key 鉴权。

5. 实战：部署第二个 Agent 并实现互相调用

现在部署第二个 Agent（如“策略推荐助手”），它依赖第一个 Agent 的数据分析结果来生成策略建议。

5.1 第二个 Agent 服务代码

创建 agent_planner.py，在内部的 send_task 处理逻辑中，主动调用第一个 Agent（数据分析助手）的任务端点：

import requests
import json

# 第一个 Agent 的 Cloud Run 服务 URL
DATA_AGENT_URL = "https://a2a-agent-v1-xxxxxx-uc.a.run.app"

def call_data_agent(query):
    """向数据分析助手发送任务并等待结果"""
    payload = {
        "jsonrpc": "2.0",
        "method": "send_task",
        "params": {
            "task_id": f"plan-{uuid.uuid4()[:8]}",
            "message": {
                "role": "user",
                "content": query
            }
        },
        "id": str(uuid.uuid4())
    }
    resp = requests.post(DATA_AGENT_URL, json=payload, timeout=30)
    resp.raise_for_status()
    return resp.json()

在 send_task 处理函数中，若接收到策略生成请求，则先调用数据分析助手，再基于其返回的结果生成策略建议。完整代码模式同 4.1 节，仅额外加入内部 RPC 调用。

5.2 部署与联调

部署第二个 Agent 至 Cloud Run（步骤同上）。然后通过 curl 向策略推荐助手发送任务：

curl -X POST https://agent-planner-xxxxxx-uc.a.run.app/ \
  -H "Content-Type: application/json" \
  -d '{
    "method": "send_task",
    "params": {
        "message": {"role": "user", "content": "基于上月销售数据，给出客户维护策略"}
    },
    "id": "req-2"
  }'

预期返回的结果中应包含由数据分析助手生成的中间结果字段（示例中简化为一条字符串），以及策略推荐助手自己的输出。这种链式调用模式在多 Agent 协作中非常常见。

6. 关键代码解读：JSON-RPC 2.0 Agent 互调实现

A2A 协议在编码层面需要重点关注以下几点：

6.1 核心 RPC 方法

方法名	用途	关键参数	典型响应
`send_task`	创建并开始执行任务	`task_id`, `message`	`{task_id, status}`
`get_task`	查询任务状态与结果	`task_id`	`{status, messages, artifacts}`
`cancel_task`	取消正在执行的任务	`task_id`	`{task_id, status:"cancelled"}`

A2A 协议标准化这些方法名，使得不同 Agent 无需协商 API 命名即可互调。id 必须唯一：同一 Agent 发出的不同请求不应使用相同的 id 值，否则接收方可能无法区分响应归属。实践中建议使用 UUID。

6.2 AgentCard JSON 结构

{
  "name": "数据分析助手",
  "description": "支持销售数据统计、趋势分析、异常检测",
  "url": "https://my-agent.example.com",
  "capabilities": ["send_task", "get_task"],
  "authentication": {
    "schemes": [
      { "type": "http", "in": "header", "scheme": "bearer" }
    ]
  }
}

调用方在发起任务前应检查目标 Agent 的 AgentCard，确认其 capabilities 包含自己需要的 RPC 方法。如果能力不匹配，应拒绝调用而非盲目重试。

6.3 流式与非流式响应

A2A 支持两种模式：

非流式：服务端在一次 HTTP 响应中返回完整的结果（包括中间结果与最终结果）。适用于无需实时显示过程的任务。
流式：服务端使用 Transfer-Encoding: chunked 或 SSE，先返回中间状态（如 working），再逐步返回增量结果或最终结果。客户端需要支持流解析。

在 Cloud Run 上使用 Flask 实现流式响应，建议使用 flask.Response 配合生成器函数。注意：Cloud Run 对超时时间有限制（最长 30 分钟），长时间流式任务需考虑分片或多轮轮询。

7. 进阶技巧：Agent Engine A2A 协议配置优化

如果使用 Vertex AI Agent Engine（即 Gen AI Agent Builder 的 Agent Engine 功能），可以通过以下配置项精细控制 A2A 行为：

7.1 配置项说明

配置项	作用	推荐值
`agent_name`	Agent 在 Agent Engine 中的唯一标识	与 AgentCard 的 name 保持一致
`authentication`	设置 AgentCard 中的 `authentication` 字段	内部调用可用 NONE，跨项目必选 API Key
`allowed_hosts`	白名单，限制只接受来自特定域名的请求	`["*.mycompany.com"]`
`logging.enabled`	是否记录 A2A 请求日志	true
`timeout`	单次 RPC 调用最大等待时间（秒）	300（根据实际任务复杂度调整）

7.2 通过环境变量开启详细日志

在 Agent Engine 部署配置中添加环境变量：

env:
  - name: A2A_LOGGING_LEVEL
    value: DEBUG
  - name: A2A_TRACE_ENABLED
    value: true

日志会输出每次 RPC 的请求体、响应体及处理耗时，对排查跨项目调用失败非常有帮助。

7.3 性能建议

超时时间：对内调用设为 30–60 秒，对外部依赖（如第三方 API）设为 10 秒，避免 Agent 被慢调用阻塞。
重试策略：对于 send_task 返回 5xx 或网络错误，使用指数退避重试（最多 3 次）。对于 get_task 轮询，建议间隔 1–5 秒，避免对服务端造成压力。
连接池复用：使用 requests.Session 或 urllib3.PoolManager 管理连接，减少 TLS 握手开销。

8. 踩坑记录与通信失败排查指南

以下汇总了 A2A Agent 通信失败的典型原因及修复方法：

8.1 网络防火墙阻塞端口

现象：发起方请求超时，curl 返回 Connection timed out。
分析：Cloud Run 服务默认只开放 443（HTTPS）和 80（HTTP）端口。如果 Agent 部署在 VPC 内且使用非标准端口，需要配置防火墙规则或负载均衡器。
解决：确保双方都使用默认 HTTPS 端口（443），并确认服务端点可达。

临时可运行 nc -vz [host] [port] 检测端口连通性。

8.2 HTTPS 证书不匹配或无效

现象：requests.exceptions.SSLError。
分析：调用了自签证书的服务，或服务域名未被信任。Cloud Run 使用 Google 管理的 SSL 证书自动为自定义域名配置证书，但如果没有绑定域而直接使用 IP 地址调用，证书将不匹配。
解决：始终使用 Cloud Run 生成的 HTTPS URL 调用，不要直接解析 IP。

若必须在内部网络使用，可设置 verify=False（仅限开发环境）。

8.3 JSON-RPC 请求中的 id 重复或缺失

现象：接收方返回 "error": {"code": -32600, "message": "Invalid Request"}。
分析：JSON-RPC 2.0 规范要求每个请求必须有非 null 的 id，且建议唯一。如果调用方误传了空字符串或重复 id，接收方可能无法正确处理。

解决：在构建请求体时，统一使用 str(uuid.uuid4()) 生成 id，并在代码中校验字段非空。

8.4 Agent 能力不兼容导致任务被拒绝

现象：返回 "error": {"code": -32601, "message": "Method not found"}。
分析：调用方尝试调用目标 Agent 未声明支持的 RPC 方法。例如，目标 Agent AgentCard 只列出了 send_task 和 get_task，但调用方调用了 cancel_task。

解决：在调用前先获取目标 AgentCard，断言所需方法在 capabilities 数组中，否则直接返回错误并打日志。

8.5 Agent Engine 默认鉴权未关闭导致跨项目调用失败

现象：跨项目调用时返回 401 或 403。
分析：Agent Engine 默认开启身份验证，且未在 AgentCard 中声明 authentication 字段。调用方未携带有效 token。
解决：

在 Agent Engine 配置中设置 authentication: NONE（仅限内部非敏感场景）
或者在调用方生成 token 并放入 Authorization header。

最简单的方法是使用 Google Cloud ID 令牌（需要调用方有相应权限）。

8.6 CORS 策略拒绝浏览器端调用

现象：前端网页直调 Agent 服务时，浏览器控制台报 CORS 错误。
分析：Cloud Run 默认不开放跨域请求。
解决：在 Flask 应用中添加 CORS 中间件：

1 2	`from flask_cors import CORS CORS(app, resources={r"/": {"origins": ""}})`

生产环境应将 origins 限定为具体的信任域名。

9. 多 Agent 协作模式与设计建议

9.1 常见协作模式

顺序调用（任务链）：Agent A 调用 Agent B，Agent B 再调用 Agent C，形成管线。适用于数据处理流程（如清洗 -> 分析 -> 报告）。
并行调用（结果聚合）：主管 Agent 同时向多个子 Agent 发送任务，待所有结果返回后汇总输出。适用于多维度分析（如同时查询销售、库存、客户反馈）。
动态发现（注册中心）：维护一个 AgentCard 注册中心（可通过服务网格或简单的配置中心实现）。新 Agent 注册后，其他 Agent 通过查询注册中心获知其能力与端点，实现热插拔。

9.2 角色分工设计原则

单一职责：每个 Agent 只负责一个域的业务逻辑，不混合不同功能。例如，数据分析 Agent 不负责发送邮件，邮件 Agent 不负责生成报告。
无状态设计：Agent 服务最好保持无状态，任务状态存储在后端数据库或内存中（但需考虑重启丢失）。这有利于水平扩展。
入口 Agent 做编排：通常一个“编排 Agent”作为外部入口，负责接收用户请求、拆分子任务、发起 A2A 调用、收集结果并组装最终输出。

编排 Agent 内部不执行具体业务逻辑，只做调度。

9.3 A2A 协议 vs. 简单 HTTP 调用

对比项	简单 HTTP 调用	A2A 协议
标准化	无，需双方协商	有，基于 JSON-RPC 2.0
自发现	无	通过 AgentCard 支持
任务管理	无	完整生命周期
跨平台	需适配	天然支持
调试工具	少	curl + jq 可调试

所以，多 Agent 协作时优先选择 A2A 协议，除非仅有一个双向调用关系且不关心扩展。

10. 总结与拓展

本文从核心原理到完整实战，逐步说明了如何基于 A2A 协议实现跨平台多 Agent 互相通信调用。关键要点回顾：

A2A 协议是开放标准，基于 JSON-RPC 2.0 over HTTP(S)，通过 AgentCard 实现能力发现，通过 Task 生命周期管理任务状态。
实战部署：使用 Python + Flask 或 Cloud Run 可快速搭建 A2A Agent 服务，通过 HTTP POST 请求实现 Agent 间互相调用。
JSON-RPC 2.0 Agent 互调的核心在于 send_task、get_task、cancel_task，以及遵守 id 唯一性、能力检查等编码规范。
Agent Engine 的配置优化包括认证方式、日志级别、白名单和超时策略。
踩坑排查应优先检查网络连通性、证书、JSON-RPC 格式、Agent 能力兼容性及鉴权配置。

随着 AI Agent 生态的发展，A2A 协议将与 MCP（Model Context Protocol）互补——MCP 定位为模型与工具之间的通用接口，A2A 定位为 Agent 与 Agent 之间的协作协议。未来，大型组织内部可能形成 Agent 网格，通过注册中心统一管理数百个 Agent 的 A2A 端点。

社区也在讨论支持更多传输层（如 WebSocket、gRPC）以及更高的安全性要求（如双向 TLS）。

建议团队在内部项目中逐步积累 Agent 能力库：先定义 AgentCard 规范，确保新开发的 Agent 都暴露标准 A2A 端点；然后在小范围业务中验证顺序调用和并行调用模式；最终实现可编排的多 Agent 协作系统，提升业务响应效率。

总结

通过本文的学习，相信你已经对「A2A协议多Agent通信原理」有了更深入的理解。建议结合实际项目多加练习。如有疑问，欢迎交流！