Google开源的A2A协议，让两个AI智能体像发微信一样协作

摘要： 用Python从零搭一个A2A Agent，跑通智能体之间发消息、传结果的全流程。附5个真实踩坑记录。

一、你的AI助手们，互相不认识

昨天下午我同时开了三个AI——ChatGPT写脚本，Claude做报表，通义千问翻译文档。三个AI各干各的，我坐在电脑前当快递分拣员，把A的输出复制给B，B的结果粘贴给C。

折腾了两个小时。

都2026年了，AI能写代码、能做分析、能生成视频，但它们之间居然不会自己说话？我一个人类，在给三个AI当中转站。

你去问任何一个用过多个AI工具的程序员，十个人里有八个经历过这种尴尬。让销售Agent把客户需求转给技术Agent？做不到。让数据分析Agent把结果丢给报告生成Agent？得你手动搬。

Google也发现了这个问题。

2025年4月Google Cloud Next大会上，他们扔出一个叫A2A（Agent-to-Agent Protocol）的开放协议。说白了就是给AI智能体发明一套"普通话"，让不同厂商、不同框架的Agent能互相发现、对话、协作。

Salesforce、SAP、微软、PayPal，50多家厂商联名站台。2025年6月，A2A捐赠给Linux基金会，Apache 2.0开源。最新版v0.2.5，v1.0已在路上。

A2A就是智能体世界的HTTP。

这篇文章，我带你用Python从零搭一个能跑的A2A Echo Agent。不画架构图，不讲概念，写代码、跑起来、看到结果。

跑完你会搞懂三件事：Agent怎么自我介绍（Agent Card），Agent之间怎么发消息（Task + Message），Agent怎么把结果交出来（Artifact）。

---

二、没有A2A，Agent协作有多惨

每个Agent都是孤岛

你用LangGraph搭了个客服Agent，同时用CrewAI搭了个数据分析Agent。老板说让这两个配合一下，客户投诉来了自动分析原因。

打开代码一看，傻了。LangGraph用LangChain的Message格式，CrewAI用自己的Task格式，数据结构完全不兼容。想让它们对话？手写一整套适配层——解析A的输出，转成B的输入，再处理异常、重试、超时……

这活儿我干过。去年帮客户对接三个不同框架的Agent，光适配代码写了2000多行，维护成本比业务代码还高。

每加一个新Agent，就多一套适配代码。N个Agent两两对接，适配层数量是N×(N-1)/2。5个Agent是10套，10个Agent是45套。指数级爆炸。

Agent被降级成工具

现有的"伪协作"方案是什么？把Agent B封装成Tool，让Agent A通过Function Calling调用。

听起来合理，实际上有个致命问题：Agent被降级了。

一个真正的Agent应该有自主推理能力——理解上下文、多轮协商、处理长任务。封装成Tool之后，它变成无状态的函数调用。输入→输出，完事。没有中间过程，没有协商空间，没有"我需要更多信息"的反馈。

打个比方：让一个资深架构师只画UML图，不能问需求、不能提建议、不能说"这方案有问题"。这不是协作，是压榨。

A2A怎么解决

A2A的设计者在协议里做了三个关键决策：

不共享内存。 每个Agent保持完全独立，有自己的上下文、工具、推理过程。A2A只传消息，不干涉Agent内部。就像两个人聊天，你不需要知道对方脑子里在想什么，听懂话就行。

异步优先。 企业级Agent任务可能跑几分钟甚至几天（涉及人工审批的场景）。A2A原生支持长任务——Task有完整的状态生命周期，支持流式更新和推送通知。不是"发个请求等回复"那么简单。

模态无关。 Agent之间不只传文本。文件、JSON数据、甚至内嵌UI组件（iframe），A2A都支持。协作场景大大扩展。

---

三、5分钟搞懂A2A的关键术语

三个角色

用户（User）—— 发起任务的人或系统。可以是你，也可以是另一个程序。

客户端（Client）—— 代表用户向Agent发请求。理解为"发消息的手机"就行。

服务端（Server）—— 提供能力的远程Agent。理解为"接消息的手机"就行。

有个细节值得注意：A2A里没有明确定义"Host"角色。MCP协议有明确的Host概念（宿主应用），A2A更开放——Agent之间是平等的，谁都可以当Client，谁都可以当Server。

五个核心概念

我用快递类比帮你快速理解：

Agent Card（名片）—— Agent的自我介绍文件，JSON格式。写着"我叫什么、能做什么、怎么认证、地址在哪"。托管在 /.well-known/agent.json 路径下，其他Agent访问这个路径就能了解你。

Task（任务）—— 一次协作的完整过程。状态流转：submitted（已提交）→ working（执行中）→ completed（已完成），中间还可能经过 input-required（需要补充信息）或 auth-required（需要额外认证）。

Message（消息）—— 客户端和Agent之间的一轮对话。有role字段区分是"user"发的还是"agent"回的。

Part（片段）—— 消息里的最小内容单元。TextPart（文本）、FilePart（文件）、DataPart（结构化数据），一条Message可以包含多个Part。

Artifact（工件）—— Agent最终产出的不可变结果。一份报告、一段代码、一张图表，都算Artifact。

通信方式

A2A的协议栈很简洁：HTTP + JSON。

传输层： 所有通信走HTTP(S)，生产环境必须HTTPS。

数据格式： JSON-RPC 2.0。2009年就定下来的远程调用协议，A2A直接复用，没发明新格式。

流式传输： Server-Sent Events（SSE）。Agent需要实时推送进度时，用SSE流式返回状态更新和中间结果。

推送通知： Webhook。超长任务（比如需要人工审批），客户端注册回调URL，Agent完成后主动POST通知。

我的体感：A2A的技术选型非常务实。全是业界用了十几年的成熟方案。你现有的HTTP基础设施、负载均衡、监控工具，全部直接复用。

---

四、环境搭建（10分钟搞定）

你需要什么

一台能联网的电脑，三个工具：

• Python 3.12+

  ——A2A的Python SDK硬性要求3.12，低版本跑不起来（后面细说这个坑）

• VS Code

  （或你顺手的编辑器）

• uv

  ——超快的Python包管理器，类似Rust界的cargo  没装uv？一条命令：

curl -LsSf https:0

Windows用户用PowerShell：

powershell -ExecutionPolicy ByPass -c "irm https:0

创建项目

打开终端，三步走：

uv init --package my-a2a-agent

cd my-a2a-agent

uv venv .venv

source .venv/bin/activate

Windows用户最后一步换成 .venv\Scripts\activate。

终端提示符前面出现 (.venv) 就说明环境OK。

验证Python版本：

uv run python -c 0

看到 3.12.x 就行。3.11或更低？运行 uv python install 3.12。

安装A2A依赖

uv add git+https:0

uv add click

uv会自动创建 pyproject.toml 并锁定依赖版本。装完后项目结构长这样：

my-a2a-agent/

├── pyproject.toml

├── README.md

├── src/

│   └── my_a2a_agent/

│       └── __init__.py

再创建两个文件：

touch src/my_a2a_agent/task_manager.py

touch src/my_a2a_agent/agent.py

环境搭建完毕。GitHub访问慢的话设个代理或用国内镜像。

---

五、定义Agent技能和名片

告诉别人你会什么

打开 src/my_a2a_agent/__init__.py，先写最基础的东西：

import logging

import click

from dotenv import load_dotenv

from google_a2a.common.types import (

    AgentSkill, AgentCapabilities, AgentCard

)


logging.basicConfig(level=logging.INFO)

logger = logging.getLogger(__name__)



defmain(host: str, port: int):

    skill = AgentSkill(

id=0,

        name=1,

        description=2,

        tags=[3, 4, 5],

        examples=[6],

        inputModes=[7],

        outputModes=[8],

    )

    logging.info(skill)



if __name__ == 9:

main()

跑一下：

uv run my-a2a-agent

看到类似输出就对了：

INFO:root:id=0 name=1 description=2 ...

Skill的每个字段都有明确用途：

• id

  ：唯一标识，其他Agent通过它调用你的技能

• name

  ：人类可读的名称

• description

  ：详细描述，帮助其他Agent理解这个技能做什么

• tags

  ：标签，用于分类和搜索

• examples

  ：示例输入，让调用方知道怎么用

• inputModes / outputModes

  ：支持的输入输出格式

Agent的身份证

有了Skill，组装AgentCard：

defmain(host: str, port: int):

    skill = AgentSkill(

id=1,

        name=2,

        description=3,

        tags=[4, 5, 6],

        examples=[7],

        inputModes=[8],

        outputModes=[9],

    )


    capabilities = AgentCapabilities(

        streaming=True,

        pushNotifications=False,

    )


    agent_card = AgentCard(

        name=10,

        description=11,

        url=f120.1.013text14text"],

        capabilities=capabilities,

        skills=[skill],

    )

    logging.info(agent_card)

再跑一次 uv run my-a2a-agent，你会看到完整的AgentCard信息。

AgentCard就是你的Agent在A2A世界里的身份证。其他Agent想跟你协作，第一步就是拿到这张Card，看你叫什么、能做什么、怎么通信。

真实部署中，这张Card自动托管在 http://你的域名/.well-known/agent.json。别的Agent只需知道你的域名，就能自动发现你。

---

六、实现Task Manager（Agent的大脑）

Task Manager是什么

Task Manager是A2A Server的核心。有请求进来，Server把请求丢给Task Manager处理。你需要实现两个方法：

• on_send_task

  ——同步请求，一次性返回结果

• on_send_task_subscribe

  ——流式请求，分多次推送更新  打开 src/my_a2a_agent/task_manager.py：

from typing import AsyncIterable


from google_a2a.common.server.task_manager import InMemoryTaskManager

from google_a2a.common.types import (

    Artifact,

    JSONRPCResponse,

    Message,

    SendTaskRequest,

    SendTaskResponse,

    SendTaskStreamingRequest,

    SendTaskStreamingResponse,

    Task,

    TaskState,

    TaskStatus,

    TaskStatusUpdateEvent,

)



classEchoTaskManager(InMemoryTaskManager):

012


asyncdefon_send_task(

self, request: SendTaskRequest

    ) -> SendTaskResponse:

345

awaitself.upsert_task(request.params)


        task_id = request.params.id

        received_text = request.params.message.parts[0].text


        task = awaitself._update_task(

            task_id=task_id,

            task_state=TaskState.COMPLETED,

            response_text=f6,

        )


returnSendTaskResponse(id=request.id, result=task)


asyncdefon_send_task_subscribe(

self, request: SendTaskStreamingRequest

    ) -> AsyncIterable[SendTaskStreamingResponse] | JSONRPCResponse:

789

awaitself.upsert_task(request.params)


        task_id = request.params.id

        received_text = request.params.message.parts[0].text


        sse_event_queue = awaitself.setup_sse_consumer(

            task_id=task_id

        )


import asyncio

        asyncio.create_task(

self._stream_echo(request, received_text)

        )


returnawait sse_event_queue.get()


asyncdef_stream_echo(

self, request: SendTaskStreamingRequest, text: str

    ):

101112

        steps = [13, 14, 15]

for i, step_text inenumerate(steps):

            is_last = (i == len(steps) - 1)

            task_state = (

                TaskState.COMPLETED if is_last

else TaskState.WORKING

            )


            parts = [

                {16: 17, 18: f19}

            ]

            message = Message(role=20, parts=parts)

            task_status = TaskStatus(

                state=task_state, message=message

            )


            event = TaskStatusUpdateEvent(

id=request.params.id,

                status=task_status,

final=is_last,

            )

awaitself.enqueue_events_for_sse(

                request.params.id, event

            )


asyncdef_update_task(

self, task_id: str, task_state: TaskState,

        response_text: str

    ) -> Task:

212223

        task = self.tasks[task_id]

        agent_response_parts = [

            {24: 25, 26: response_text}

        ]

        task.status = TaskStatus(

            state=task_state,

            message=Message(

                role=27, parts=agent_response_parts

            ),

        )

        task.artifacts = [

Artifact(parts=agent_response_parts),

        ]

return task

代码有点长，逻辑很清晰。

on_send_task 做四件事：存任务→拿文本→标记完成→返回结果。

on_send_task_subscribe 做五件事：存任务→创建SSE队列→启动异步工作→分3次推送进度→最后一次标记完成。

一个关键细节

enqueue_events_for_sse 这个方法，从 InMemoryTaskManager 继承来的，负责把事件放入SSE队列。

我第一次写的时候漏掉了这个调用。客户端连上了但一直收不到数据，排查了半小时才发现。事件造好了没放进队列，客户端当然收不到。

---

七、启动A2A Server

把所有组件串起来

回到 src/my_a2a_agent/__init__.py：

import logging

import click

from dotenv import load_dotenv

from google_a2a.common.types import (

    AgentSkill, AgentCapabilities, AgentCard

)

from google_a2a.common.server import A2AServer

from my_a2a_agent.task_manager import EchoTaskManager


logging.basicConfig(level=logging.INFO)

logger = logging.getLogger(__name__)



@click.command()

@click.option(2, default=3)

@click.option(4, default=10002, type=int)

defmain(host: str, port: int):

    skill = AgentSkill(

id=5,

        name=6,

        description=7,

        tags=[8, 9, 10],

        examples=[11],

        inputModes=[12],

        outputModes=[13],

    )


    capabilities = AgentCapabilities(streaming=True)


    agent_card = AgentCard(

        name=14,

        description=15,

        url=f160.1.017text18text19Starting A2A Server at http:1

    )

    server.start()



if __name__ == 20:

main()

启动

uv run my-a2a-agent

看到这堆输出就说明成功了：

INFO:root:Starting A2A Server at http:0

INFO: Started server process [xxxxx]

INFO: Waiting for application startup.

INFO: Application startup complete.

INFO: Uvicorn running on http:1

你的A2A Agent已经在本地跑起来了，监听10002端口，等待对话。

打开浏览器访问 http://localhost:10002/.well-known/agent.json，能看到Agent Card的JSON内容。这就是其他Agent发现你的方式——访问这个路径，拿到你的名片。

---

八、跟Agent对话

写个测试客户端

Server跑着了，需要一个客户端来跟它对话。打开 src/my_a2a_agent/agent.py：

import asyncio

import httpx

from a2a.client import (

    A2ACardResolver, ClientConfig, create_client

)

from a2a.helpers import new_text_message

from a2a.types.a2a_pb2 import Role, SendMessageRequest



asyncdeftest_non_streaming():

678

print(9)


asyncwith httpx.AsyncClient() as httpx_client:

0

        resolver = A2ACardResolver(

            httpx_client=httpx_client,

            base_url=10发现Agent: {agent_card.name}11描述: {agent_card.description}12你好，我是第一个跟你对话的Agent！13\n发送: {text_query}14响应:1516流式：发一条消息，实时接收多次更新。1718\n=== 流式测试 ===19http:5

        )

        agent_card = await resolver.get_agent_card()


        config = ClientConfig(streaming=True)

        client = awaitcreate_client(

            agent=agent_card, client_config=config

        )


        text_query = 20

        message = new_text_message(

            text_query, role=Role.ROLE_USER

        )

        request = SendMessageRequest(message=message)


print(f21)

print(22)

asyncfor chunk in client.send_message(request):

print(chunk)


await client.close()



if __name__ == 23:

    asyncio.run(test_non_streaming())

    asyncio.run(test_streaming())

运行测试

确保A2A Server还在运行（关了就重新 uv run my-a2a-agent）。

开一个新终端：

source .venv/bin/activate

uv run python -m my_a2a_agent.agent

非流式输出：

=== 非流式测试 ===

发现Agent: My First A2A Agent

描述: A simple echo agent built with A2A protocol.


发送: 你好，我是第一个跟你对话的Agent！

响应:

task {

id: 0

  status { state: TASK_STATE_COMPLETED }

  artifacts {

    parts { text: 1 }

  }

}

流式输出：

=== 流式测试 ===


发送: 你好，请用流式模式回复我！

流式响应:

task {

id: 0

  status { state: TASK_STATE_SUBMITTED }

}


status_update {

  status { state: TASK_STATE_WORKING,

    message { parts { text: 1 } } }

}


status_update {

  status { state: TASK_STATE_WORKING,

    message { parts { text: 2 } } }

}


status_update {

  status { state: TASK_STATE_COMPLETED,

    message { parts { text: 3 } } }

}

两种模式有什么区别

非流式：你发一条消息，等一会儿，收到一个完整的Task。里面有最终状态（COMPLETED）和完整Artifact。中间过程不可见。

流式：你发一条消息，实时收到4次推送——初始Task（SUBMITTED），3次status_update（WORKING→WORKING→COMPLETED）。每一步进展都看得到。

我的体感：流式模式体验好太多。Agent处理复杂任务时，你能看到"正在工作"的反馈，不会以为卡死了。

---

九、给Agent接上大模型

Echo Agent的局限

到目前为止，你的Agent只会复读。发什么回什么。Hello World级别的demo。

真正的Agent应该能理解意图、做推理、调工具、给出有价值的回答。要做到这些，得接上大模型。

用Ollama跑本地模型

最简单的方式。不需要API Key，数据不出你的电脑。

先装Ollama：

curl -fsSL https:0

ollama pull qwen2.5:7b

然后修改 task_manager.py，把Echo逻辑替换成大模型调用：

import httpx

from google_a2a.common.server.task_manager import (

    InMemoryTaskManager

)

from google_a2a.common.types import (

    Artifact, Message, SendTaskRequest, SendTaskResponse,

    Task, TaskState, TaskStatus,

)


OLLAMA_URL = 4qwen2.5:7b56接入了Ollama大模型的Task Manager。78正在思考中...9model10prompt11stream12response13（模型无响应）14调用模型失败: {str(e)}15type16text17text18agent", parts=parts),

        )

if task_state == TaskState.COMPLETED:

            task.artifacts = [Artifact(parts=parts)]

return task

在 __init__.py 里把 EchoTaskManager 换成 LLMTaskManager：

from my_a2a_agent.task_manager import LLMTaskManager


0

task_manager = LLMTaskManager()

重启Server，再用客户端发一条消息。这次Agent不会复读你了，它会调本地大模型，给你一个真正的AI回答。

想用云端模型？

用Google的Agent Development Kit（ADK）就行。把Ollama的API调用替换成云端模型调用，加上API Key认证。官方文档有详细接入指南。

核心思路不变：在Task Manager里调用大模型，把结果包装成A2A格式的Message和Artifact返回。

---

十、我踩过的5个坑

坑1：Python版本不对。

A2A的Python SDK硬性要求3.12+。我用3.11，pip install直接报错，报错信息还看不懂。翻GitHub Issue才发现是版本问题。uv python install 3.12，一行搞定。

坑2：Card的url和Server地址不一致。

Card里写的 http://localhost:10002/，Server实际监听 http://127.0.0.1:10002/。客户端拿到Card去访问，连接失败。确保Card里的url和Server的host/port完全一致。

坑3：流式响应漏调enqueueeventsfor_sse。

最惨的一个坑。流式模式写好了，客户端连上了，就是收不到数据。排查半小时，发现 _stream_echo 里忘记调用 enqueue_events_for_sse。事件造好了没放进队列，客户端当然收不到。

坑4：虚拟环境没激活。

低级但真容易犯。一个终端激活了venv，开新终端忘了激活，直接跑脚本就报 ModuleNotFoundError。养成习惯：每次开新终端先 source .venv/bin/activate。

坑5：端口被占用。

10002端口看起来冷门，我测试时恰好跟另一个服务冲突。Address already in use。换端口：--port 10003。

---

A2A vs MCP：一句话说清楚

很多人搞不清这两个协议的关系。

MCP（Anthropic出品）= Agent调工具的"USB-C接口"。解决Agent怎么连数据库、API、文件系统。

A2A（Google出品）= Agent之间对话的"微信协议"。解决Agent怎么跟另一个Agent协作。

不是竞争，是互补。一个Agent通过MCP调工具拿数据，再通过A2A把结果发给另一个Agent。2026年5月微软宣布Copilot平台同时支持MCP和A2A，行业在往"双协议融合"走。

---

下一步往哪走

跑完这个demo，你已经掌握了A2A的核心机制。三个方向可以深入：

1. 搭第二个Agent，实现真正的双Agent协作。

   一个Echo Agent太单薄。试试搭一个"翻译Agent"和一个"摘要Agent"，让它们通过A2A互相委托任务。

2. 学A2A的认证鉴权。

   生产环境必须做认证。A2A支持OAuth 2.0、API Key、mTLS，Agent Card里有 securitySchemes 字段专门声明。

3. 探索A2A + MCP联合使用。

   让你的Agent既能通过MCP调外部工具，又能通过A2A跟其他Agent协作。这才是2026年Agent开发的完整形态。  A2A协议还在快速迭代（v0.2.5→v1.0），但核心设计已经稳定。现在入场，正好赶上Agent互联网的早期红利期。  就像2010年学HTTP开发Web应用——协议很简单，但机会很大。