谷歌A2A协议实战：5分钟搭建你的第一个企业级Agent通信系统

最近和几个技术团队的朋友聊天，大家不约而同地提到了同一个痛点：公司内部那些智能助手、数据分析机器人、客服自动化工具，一个个都像信息孤岛，彼此之间没法顺畅“对话”。市场部的数据分析Agent生成的报告，没法直接推送给销售团队的跟进Agent；客服系统的意图识别结果，也无法无缝触发内部工单系统的处理流程。这种割裂，让本应智能化的流程，又退回到了人工搬运数据的原始阶段。

如果你也面临类似的困境，那么今天要聊的谷歌A2A协议，可能就是你在寻找的那把钥匙。它不是一个遥不可及的学术概念，而是一个已经开源、设计目标直指企业级应用的Agent间通信标准。简单来说，A2A让不同的智能体（Agent）能够像人类同事一样，通过一套标准化的“语言”和“工作流程”进行协作。这听起来可能有点抽象，但别担心，接下来的内容，我会带你从零开始，用大约5分钟的核心流程，快速搭建一个可运行的A2A通信系统原型。我们会绕过繁琐的理论，直接上手代码和配置，让你真切感受到如何让两个Agent“握手”并开始协作。

本文面向的是希望快速将Agent协作能力落地的企业开发者或技术决策者。我们将聚焦于环境准备、核心配置、安全认证和第一个任务流的实战步骤。你会发现，借助现有的云服务和开源代码，搭建一个符合企业基本要求（如身份认证、传输安全）的A2A系统，比想象中要简单得多。

1. 环境准备与基础概念速览

在开始敲代码之前，我们需要快速统一一下认知。A2A协议的核心思想是标准化交互。它定义了几个关键角色和交互模式，理解它们对后续的配置和开发至关重要。

首先，你需要准备一个开发环境。我个人推荐使用Python 3.9+，因为它有丰富的异步支持和成熟的HTTP客户端库。同时，确保你的网络可以访问必要的公共服务（例如容器仓库、包管理源）。我们将使用Docker来简化部署，所以请先确保Docker和Docker Compose已安装就绪。

提示：如果你在团队内部进行开发，建议预先在内部镜像仓库准备好基础镜像，以加速后续的构建和部署流程。

A2A协议中，每个参与协作的智能体都被称为一个 A2A Server。它对外暴露一个HTTP端点，接收并处理来自其他Agent（即A2A Client）的请求。它们之间交换的基本工作单元叫做 Task（任务）。一个任务从创建到结束，会经历一系列状态变迁，例如：

  • submitted： 任务已提交
  • working： 服务端正在处理
  • input-required： 需要客户端提供更多输入
  • completed： 任务成功完成
  • failed： 任务执行失败
  • canceled： 任务被取消

任务中包含的对话内容，由 Message（消息）对象承载，而消息又由多个 Part（部分）组成，比如文本、文件或结构化数据。最终的任务输出，则可能是一个或多个 Artifact（工件），例如生成的文件或最终的数据结构。

为了让Client能找到并了解Server，每个A2A Server都需要发布一个 Agent Card。你可以把它想象成一张智能体的“数字名片”或“服务说明书”，以JSON格式存放在一个约定的URL（通常是 /.well-known/agent.json）下。这张名片里写明了Agent的能力、技能、支持的交互模式以及如何进行身份认证。

概念 角色 关键作用
A2A Server 服务提供方 实现A2A协议方法，接收并执行任务。
A2A Client 服务消费方 向Server发起任务请求，并处理响应。
Agent Card 服务发现与描述 描述Server的能力、端点、认证方式等元数据。
Task 工作单元 一次具体的协作请求，拥有唯一ID和状态流。
Message 交互内容 包含用户或Agent的输入输出，由多个Part组成。

理解了这些，我们就可以进入实战环节了。整个搭建过程可以浓缩为三个核心动作：编写Agent Card、实现A2A Server、配置安全通信。下面我们逐一拆解。

2. 第一步：编写并发布你的Agent Card

Agent Card是A2A生态的“服务注册中心”。它的核心作用是让其他Agent能发现你、了解你，并知道如何与你安全地对话。我们将从一个最简单的Agent Card开始。

假设我们正在构建一个“销售数据分析Agent”，它能够接收自然语言查询，并返回指定时间段的销售汇总。那么，它的Agent Card可能看起来像这样：

{
  "name": "Sales Data Analyst Agent",
  "description": "An agent that queries and analyzes enterprise sales data from multiple sources.",
  "url": "https://sales-agent.your-company.com/a2a",
  "version": "1.0.0",
  "provider": {
    "organization": "YourTechCorp",
    "url": "https://www.your-company.com"
  },
  "capabilities": {
    "streaming": true,
    "pushNotifications": false
  },
  "authentication": {
    "schemes": ["Bearer"]
  },
  "defaultInputModes": ["text/plain"],
  "defaultOutputModes": ["application/json"],
  "skills": [
    {
      "id": "query-sales-summary",
      "name": "Query Sales Summary",
      "description": "Retrieves and summarizes sales data for a given time period.",
      "tags": ["sales", "data-analysis", "reporting"],
      "examples": ["Show me sales for the last quarter", "What were the top products in January?"]
    }
  ]
}

这个JSON文件定义了Agent的基本信息。其中几个关键字段需要你根据实际情况调整：

  • url: 你的A2A Server实际监听的HTTP(S)端点地址。
  • authentication.schemes: 声明了客户端必须使用Bearer Token进行认证。这是企业级应用的基础安全要求。
  • skills: 定义了该Agent具体能做什么。这里我们只定义了一个技能，但一个Agent完全可以拥有多个。

编写好这个JSON文件后，你需要将它发布到一个可通过HTTP访问的URL。标准做法是将其放置在Server根域名的 /.well-known/ 路径下，即 https://your-domain.com/.well-known/agent.json。这类似于网站提供 robots.txt 或 security.txt 的惯例，是一种开放的发现机制。

对于企业内部或需要控制访问的场景，A2A也支持通过注册表目录或私有API进行“受控发现”和“私有发现”。例如，你可以建立一个内部门户，只有经过审批的Agent才能将其Card注册到目录中，供其他内部Agent查询。

注意：Agent Card中可能包含认证信息（如所需的API密钥类型）。强烈建议不要在未经验证身份的情况下公开访问包含敏感信息的Agent Card。 即使是放在 /.well-known/ 路径下，也应考虑通过网络策略（如mTLS）限制仅允许特定的客户端IP或服务账户访问。

3. 第二步：快速实现一个A2A Server

有了“名片”，接下来就要打造提供服务的“本体”——A2A Server。得益于谷歌开源的A2A Python SDK，实现一个基础Server变得非常直接。我们继续以销售数据分析Agent为例。

首先，安装必要的依赖。创建一个新的Python虚拟环境，然后安装：

pip install aiohttp google-a2a

google-a2a 这个包提供了A2A协议的核心类型定义和基础工具类，能极大简化开发。接下来，我们创建一个最简单的Server，它接收一个文本查询，模拟处理，并返回一个固定的JSON结果。

# sales_agent_server.py
import asyncio
import logging
from typing import AsyncIterable, Dict, Any
from aiohttp import web
from google.a2a import (
    AgentCard,
    SendTaskRequest,
    SendTaskResponse,
    Task,
    TaskStatus,
    TaskState,
    Message,
    TextPart,
    Artifact
)

# 1. 定义我们之前创建的Agent Card
AGENT_CARD = AgentCard(
    name="Sales Data Analyst Agent",
    description="An agent that queries and analyzes enterprise sales data.",
    url="https://sales-agent.internal.com/a2a",
    version="1.0.0",
    authentication={"schemes": ["Bearer"]},
    defaultInputModes=["text/plain"],
    defaultOutputModes=["application/json"],
    skills=[{
        "id": "query-sales-summary",
        "name": "Query Sales Summary",
        "description": "Retrieves sales summary.",
        "tags": ["sales", "analysis"]
    }]
)

class SalesDataAgent:
    """一个简单的销售数据分析Agent实现"""
    async def execute_query(self, user_query: str) -> Dict[str, Any]:
        """模拟执行数据查询。在实际应用中，这里会连接数据库或数据仓库。"""
        # 这里只是一个模拟响应
        await asyncio.sleep(0.5)  # 模拟处理耗时
        return {
            "period": "Q1 2024",
            "total_sales": 1250000,
            "top_product": "Product X",
            "growth_rate": "15%"
        }

async def handle_send_task(request: SendTaskRequest) -> SendTaskResponse:
    """处理A2A的 tasks/send 请求"""
    agent = SalesDataAgent()
    user_message = request.params.message

    # 提取用户查询文本（假设是TextPart）
    user_text = ""
    for part in user_message.parts:
        if isinstance(part, TextPart):
            user_text = part.text
            break

    if not user_text:
        # 如果输入不支持，返回错误状态
        status = TaskStatus(state=TaskState.FAILED)
        task = Task(id=request.params.id, status=status)
        return SendTaskResponse(id=request.id, result=task)

    # 执行核心业务逻辑
    try:
        result_data = await agent.execute_query(user_text)
        # 将结果构建为Message Part
        result_part = TextPart(text=f"Query Result: {result_data}")
        response_message = Message(role="agent", parts=[result_part])

        # 创建任务完成状态和工件
        status = TaskStatus(state=TaskState.COMPLETED, message=response_message)
        artifact = Artifact(parts=[result_part], index=0, append=False)

        task = Task(
            id=request.params.id,
            status=status,
            artifacts=[artifact]
        )
        return SendTaskResponse(id=request.id, result=task)

    except Exception as e:
        logging.error(f"Task execution failed: {e}")
        status = TaskStatus(state=TaskState.FAILED)
        task = Task(id=request.params.id, status=status)
        return SendTaskResponse(id=request.id, result=task)

async def agent_card_handler(request):
    """处理 /.well-known/agent.json 请求，返回Agent Card"""
    return web.json_response(AGENT_CARD.dict())

async def a2a_endpoint_handler(request):
    """处理 /a2a 路径下的A2A协议请求"""
    # 这里需要根据请求体路由到不同的处理方法，例如 handle_send_task
    # 为简化示例，我们直接返回一个占位符响应
    return web.json_response({"status": "A2A endpoint is running"})

async def init_app():
    app = web.Application()
    # 注册路由
    app.router.add_get('/.well-known/agent.json', agent_card_handler)
    app.router.add_post('/a2a', a2a_endpoint_handler) # 实际应解析JSON-RPC请求
    return app

if __name__ == '__main__':
    logging.basicConfig(level=logging.INFO)
    web.run_app(init_app(), port=8080)

这个示例虽然简化，但勾勒出了一个A2A Server的骨架：提供Agent Card发现端点，并实现任务处理逻辑。在实际企业中，execute_query 方法内部会集成真正的数据查询引擎，比如连接公司的CRM、数据仓库，或者调用像MindsDB这样的AI原生数据库。

为了让这个Server能处理真实的A2A JSON-RPC请求，你需要集成一个完整的A2A服务器框架，它负责解析协议规定的 tasks/send 或 tasks/sendSubscribe 请求，并调用你写的业务逻辑。谷歌的A2A GitHub仓库提供了更完整的示例。

4. 第三步：配置企业级安全与认证

对于企业级应用，安全不是可选项，而是生命线。A2A协议基于HTTP/HTTPS，天然继承了Web安全的最佳实践。我们需要在三个层面构建安全防线：传输层、服务端身份、客户端身份。

传输层安全 (TLS) 这是最基本的要求。在生产环境中，必须使用HTTPS。这意味着你需要为你的A2A Server域名配置有效的TLS证书。你可以使用Let‘s Encrypt获取免费证书，或者使用企业内部的私有CA签发证书。在Kubernetes或云负载均衡器上，通常可以很方便地配置自动化的证书管理。

服务端身份验证 客户端如何确认它连接的是真正的、可信的Sales Data Agent，而不是一个冒名顶替者？这需要通过TLS证书来实现。服务端应使用由受信任的证书颁发机构（CA）签发的证书。客户端在建立TLS连接时，会验证证书的有效性和域名匹配性。

客户端与用户身份验证 这是控制“谁可以调用我的Agent”的关键。A2A协议本身不定义具体的认证方式，而是通过Agent Card中的 authentication 字段来声明支持哪些方案。常见的企业级方案包括：

  • API密钥 (Bearer Token): 最简单直接的方式，适合服务到服务的通信。客户端在HTTP请求头中携带 Authorization: Bearer <token>。
  • OAuth 2.0 / OIDC: 更适合需要用户上下文或更细粒度权限控制的场景。例如，一个任务可能需要同时访问需要用户A权限的系统A和需要用户B权限的系统B，这时就需要联合身份认证。

在你的Agent Card中，需要明确声明支持的认证方案。同时，在你的A2A Server实现中，必须对每个请求进行验证。以下是一个使用API Key进行验证的中间件示例（基于aiohttp）：

# middleware/auth_middleware.py
from aiohttp import web
import os

API_KEYS = set(os.getenv('A2A_API_KEYS', '').split(','))

@web.middleware
async def auth_middleware(request, handler):
    # 放过Agent Card发现端点
    if request.path == '/.well-known/agent.json':
        return await handler(request)

    # 检查Authorization头
    auth_header = request.headers.get('Authorization')
    if not auth_header or not auth_header.startswith('Bearer '):
        raise web.HTTPUnauthorized(reason='Missing or invalid Authorization header')

    token = auth_header[7:]  # 去掉 'Bearer ' 前缀
    if token not in API_KEYS:
        raise web.HTTPForbidden(reason='Invalid API key')

    # 认证通过，继续处理请求
    return await handler(request)

然后在初始化app时加载这个中间件：

app = web.Application(middlewares=[auth_middleware])

授权与数据隐私 认证解决了“你是谁”，授权则要解决“你能做什么”。A2A建议从两个维度管理授权：

  1. 技能级授权: 控制某个客户端可以调用Agent的哪些技能（Skill）。例如，只有财务团队的客户端才能调用“生成财务报表”技能。
  2. 工具级授权: 在技能内部，控制对特定数据或操作的访问。例如，“查询销售数据”技能内部，需要根据客户端身份过滤其可访问的区域数据。

这通常需要与公司现有的IAM（身份和访问管理）系统集成，在Agent的业务逻辑中进行权限判断。

5. 实战：构建一个任务编排中心

现在，我们已经有了一个安全的、可被发现和调用的A2A Server。但在真实的企业场景中，往往不是简单的点对点调用，而是需要复杂的任务编排。例如，用户向一个“总控Agent”提出需求：“分析上一季度销售情况，并生成一份给管理层的简报”。这个总控Agent可能需要：

  1. 调用我们刚建的“销售数据分析Agent”获取数据。
  2. 调用另一个“报告生成Agent”将数据转化为PPT。
  3. 调用“邮件发送Agent”将报告发送给管理层。

这个“总控Agent”就是一个A2A Client，同时也是一个A2A Server（因为它也可能被其他Agent调用）。它的核心是一个任务编排引擎。我们来快速实现一个这样的编排中心雏形。

首先，这个编排中心需要能发现并调用其他Agent。它通过读取目标Agent的Card来了解如何调用。

# orchestrator_agent.py
import aiohttp
import json
from typing import Dict, Any

class Orchestrator:
    def __init__(self):
        self.http_client = aiohttp.ClientSession()

    async def discover_agent(self, agent_card_url: str) -> Dict[str, Any]:
        """发现并获取Agent Card"""
        async with self.http_client.get(agent_card_url) as resp:
            if resp.status == 200:
                card = await resp.json()
                return card
            else:
                raise Exception(f"Failed to fetch Agent Card from {agent_card_url}")

    async def send_task_to_agent(self, agent_card: Dict, task_input: str, api_key: str):
        """向一个已知的Agent发送任务"""
        task_endpoint = agent_card['url']  # 从Card中获取服务端点
        task_payload = {
            "jsonrpc": "2.0",
            "id": "unique-task-id-123",
            "method": "tasks/send",
            "params": {
                "id": "unique-task-id-123",
                "message": {
                    "role": "user",
                    "parts": [{"type": "text", "text": task_input}]
                },
                "acceptedOutputModes": ["application/json"]
            }
        }

        headers = {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {api_key}"
        }

        async with self.http_client.post(task_endpoint, json=task_payload, headers=headers) as resp:
            result = await resp.json()
            # 处理结果，可能包含任务状态、消息、工件等
            return self._parse_task_result(result)

    def _parse_task_result(self, rpc_response: Dict):
        """解析A2A JSON-RPC响应"""
        # 这里需要根据A2A协议规范解析响应，提取任务状态和结果
        # 简化处理，直接返回
        return rpc_response.get('result', {})

    async def orchestrate_sales_report(self, user_request: str):
        """编排一个生成销售报告的任务流"""
        # 1. 发现销售数据分析Agent
        sales_agent_card = await self.discover_agent("https://sales-agent.internal.com/.well-known/agent.json")
        # 2. 发送查询任务
        sales_data = await self.send_task_to_agent(sales_agent_card, user_request, "your-sales-agent-api-key")

        # 3. 发现报告生成Agent
        report_agent_card = await self.discover_agent("https://report-agent.internal.com/.well-known/agent.json")
        # 4. 将销售数据发送给报告Agent，请求生成简报
        report_input = f"基于以下数据生成管理层简报：{sales_data}"
        report_artifact = await self.send_task_to_agent(report_agent_card, report_input, "your-report-agent-api-key")

        # 5. （可选）调用邮件发送Agent
        # ...
        return report_artifact

这个编排器展示了A2A的核心价值：标准化接口下的灵活组合。每个Agent专注于自己的领域（数据、报告、通知），通过A2A协议暴露清晰的能力边界和交互契约，使得上层编排逻辑变得清晰且可维护。

在实际部署时，你可以将这个编排器本身也包装成一个A2A Server，并发布自己的Agent Card。这样，其他系统或最终用户就可以通过一个统一的入口，触发复杂的跨Agent工作流。

6. 与MCP协议的关系及选型思考

在构建Agent生态时，你可能会听到另一个协议：MCP（Model Context Protocol）。它和A2A是什么关系？应该如何选择？

简单来说，MCP专注于连接大模型与工具/数据源，它统一了“函数调用”这个接口，让AI应用能更容易地接入各种资源。而A2A专注于连接智能体与智能体，它定义的是应用层的工作流和对话协议。

用一个比喻来理解：MCP像是给AI大脑（大模型）配备了标准化的“手”和“眼睛”（工具和数据连接器），让它能操作外部系统。而A2A则是定义了一套“商务沟通礼仪”和“合作流程”，让多个拥有大脑和手脚的AI员工（Agent）能够在一起开会、分工协作。

协议 核心关注点 类比 典型使用场景
MCP 模型与工具的连接 为AI配备标准化的“工具套件” 让一个AI助手能查询数据库、操作日历、发送邮件
A2A 智能体与智能体的协作 定义AI员工间的“协作流程与合同” 让数据分析AI、报告生成AI、审批AI自动串联完成一个复杂业务流程

在实际的企业架构中，它们不是二选一，而是互补共存的。一个设计良好的Agent很可能同时使用两者：

  • 对内（获取能力）：通过MCP协议，连接公司内部的数据库、API、知识库，获取执行任务所需的数据和工具。
  • 对外（提供服务与协作）：通过A2A协议，将自己封装成一个具有明确技能和接口的服务，供其他Agent或系统调用。

例如，前面提到的销售数据分析Agent，其内部可能通过MCP连接了公司的Snowflake数据仓库和内部指标系统。而对外，它通过A2A协议暴露一个干净的“查询销售摘要”技能，供编排中心调用。这种分层设计使得系统更清晰、更易维护。

在技术选型时，如果你的主要需求是让一个大模型应用能安全、统一地调用各种后端工具，那么优先考虑MCP。如果你已经在公司内部有多个功能各异的自动化工具或AI服务，现在想让它们能相互对话、串联成更智能的流程，那么A2A就是你该深入研究的协议。

Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

cover

我测试了一个0.9B模型，它竟然能自动区分多人讲话）

avatar AI Agent技术社区
cover

从一张商品图到正式上架：实测店匠 Shoplazza Athena，AI Agent 不只是聊天框

avatar AI Agent技术社区
cover

【AI Agent】《2026年9月 AI Agent全栈开发技术选型专项面试宝典》

avatar AI Agent技术社区