A2A 实践——用 Python 从零搭建多 Agent 协作系统

导读：上一篇我们用"企业微信"的类比搞懂了 A2A 协议的四个核心概念。这一篇，我们用 Python 写两个 Agent，让它们在本地跑起来，完成一次真正的多 Agent 协作。两个文件，约 90 行代码，每一行都能看懂。

上一篇我们搞懂了 A2A 是什么，这一篇我们来动手写代码。

在上一篇文章中，我们把 A2A 彻底讲清楚了：A2A 是 Agent 之间的"企业微信"——一套让不同 Agent 互相发现、沟通、协作的标准协议。 Agent Card 是名片，Task 是任务单，Message 是工作沟通，Artifact 是工作成果。

概念懂了，你可能会想：说了这么多，代码到底长什么样？Agent Card 在代码里是什么？Task 的状态怎么流转？两个 Agent 之间到底是怎么"对话"的？

今天我们就来回答这个问题。我们会用 Python 从零搭建两个 Agent——一个"简历筛选 Agent"（干活的），一个"招聘主管"（派活的）——让它们通过 A2A 协议在本地完成一次完整的协作。

两个文件加起来约 90 行代码，每一行都能看懂。

一、准备工作

开始之前，确认两件事。

Python 版本

A2A 的官方 SDK 要求 Python 3.10 或以上。运行以下命令检查：

python3 --version

如果版本低于 3.10，先升级 Python。

安装 SDK 和创建项目

# 创建项目目录
mkdir a2a-demo && cd a2a-demo

# 创建虚拟环境
python3 -m venv .venv
source .venv/bin/activate   # Windows 用: .venv\Scripts\activate

# 安装 A2A SDK（含 HTTP 服务端支持）和 HTTP 服务器
pip install "a2a-sdk[http-server]" uvicorn

a2a-sdk 是 A2A 协议的官方 Python 实现，一个库搞定 Server 端和 Client 端。uvicorn 是 HTTP 服务器，用来启动 Agent 服务。

安装完成后，我们的项目结构很简单——只需要写两个文件：

a2a-demo/
├── resume_screener.py   # 简历筛选 Agent（Remote Agent，干活的）
└── hiring_manager.py    # 招聘主管（Client，派活的）

二、写 Remote Agent：简历筛选 Agent

还记得上一篇的类比吗？Remote Agent 就是接任务的一方。 它需要做三件事：

1. 亮出名片（Agent Card）—— 告诉别人自己能干什么
2. 实现业务逻辑（AgentExecutor）—— 收到任务后怎么处理
3. 启动服务—— 在网络上"上线"，等待别人来派活

我们一步步来。

第一步：定义名片（Agent Card）

上一篇说过，Agent Card 就是 Agent 的"员工名片"。在代码里，它是一个 AgentCard 对象：

from a2a.types import AgentCard, AgentCapabilities, AgentSkill

# 定义技能
skill = AgentSkill(
    id='resume_screening',
    name='简历筛选',
    description='根据职位要求筛选匹配的候选人',
    tags=['recruiting', 'screening'],
    examples=['筛选高级后端工程师候选人'],
)

# 定义名片
agent_card = AgentCard(
    name='简历筛选 Agent',
    description='根据职位要求，从候选人库中筛选匹配的人选',
    url='http://127.0.0.1:9999',
    version='1.0.0',
    default_input_modes=['text/plain'],
    default_output_modes=['text/plain'],
    capabilities=AgentCapabilities(streaming=False),
    skills=[skill],
)

对照上一篇的概念：

• AgentSkill = 名片上写的"擅长什么"
• AgentCard = 完整的名片，包含姓名、简介、技能、联系地址
• url = "怎么联系我"——服务地址

这就是科普里说的"Agent Card"在代码里的样子。

第二步：实现业务逻辑（AgentExecutor）

Agent Card 只是"自我介绍"，真正干活的是 AgentExecutor。它是一个抽象类，我们需要实现两个方法——execute（收到任务后怎么处理）和 cancel（如何取消任务）。

from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.client.helpers import create_text_message_object
from a2a.types import Role

classResumeScreenerExecutor(AgentExecutor):
"""简历筛选 Agent 的业务逻辑"""

asyncdefexecute(self, context: RequestContext, event_queue: EventQueue):
# 1. 读取用户发来的消息
        user_message = ''
if context.message and context.message.parts:
for part in context.message.parts:
if hasattr(part.root, 'text'):
                    user_message = part.root.text
break

# 2. 执行"筛选"逻辑（这里用模拟数据演示）
        result = self._screen_resumes(user_message)

# 3. 把结果发回去
        msg = create_text_message_object(role=Role.agent, content=result)
await event_queue.enqueue_event(msg)

asyncdefcancel(self, context: RequestContext, event_queue: EventQueue):
raise Exception('cancel not supported')

def_screen_resumes(self, requirement: str) -> str:
"""模拟简历筛选逻辑"""
# 实际项目中，这里会连接数据库或调用大模型
        candidates = [
            {'name': '张三', 'experience': '6年 Go，分布式系统', 'match': '95%'},
            {'name': '李四', 'experience': '8年 Go，微服务架构', 'match': '90%'},
            {'name': '王五', 'experience': '5年 Go，云原生', 'match': '85%'},
        ]
        lines = [f'根据需求「{requirement}」，筛选出以下候选人：\n']
for c in candidates:
            lines.append(f"- {c['name']}：{c['experience']}（匹配度 {c['match']}）")
return'\n'.join(lines)

逐段解析：

• **context.message**：用户发来的消息，对应科普里的 Message
• **event_queue.enqueue_event()**：把结果包装成 Message 发回去，SDK 会自动将消息推送给 Client
• **_screen_resumes()**：业务逻辑。这里用模拟数据演示，实际项目中可以连数据库或调用大模型

AgentExecutor 就是 Agent 的"大脑"——收到任务，处理，返回结果。

第三步：启动服务

名片有了，业务逻辑有了，最后把它们组装起来，启动 HTTP 服务：

import uvicorn
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore
from a2a.server.apps.jsonrpc.starlette_app import A2AStarletteApplication

# 组装：执行器 + 任务存储 → 请求处理器
request_handler = DefaultRequestHandler(
    agent_executor=ResumeScreenerExecutor(),
    task_store=InMemoryTaskStore(),
)

# 创建 A2A 应用：名片 + 请求处理器 → Starlette 应用
a2a_app = A2AStarletteApplication(
    agent_card=agent_card,
    http_handler=request_handler,
)
app = a2a_app.build()

# 启动
uvicorn.run(app, host='127.0.0.1', port=9999)

这段代码做了三件事：

1. DefaultRequestHandler：把 AgentExecutor 和 InMemoryTaskStore（内存中的任务存储）组合在一起，负责处理所有传入的 A2A 请求
2. A2AStarletteApplication：把 AgentCard 和请求处理器组装成一个完整的 Starlette 应用，自动发布 Agent Card 到 /.well-known/agent-card.json 并处理 JSON-RPC 消息
3. **uvicorn.run()**：启动应用，监听 9999 端口

启动服务 = 在"企业微信"上注册了一个账号，别的 Agent 可以找到你、给你派活了。

完整代码：resume_screener.py

把以上三步合在一起，就是 resume_screener.py 的完整代码：

import uvicorn
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore
from a2a.server.apps.jsonrpc.starlette_app import A2AStarletteApplication
from a2a.client.helpers import create_text_message_object
from a2a.types import AgentCapabilities, AgentCard, AgentSkill, Role


classResumeScreenerExecutor(AgentExecutor):
"""简历筛选 Agent 的业务逻辑"""

asyncdefexecute(self, context: RequestContext, event_queue: EventQueue):
        user_message = ''
if context.message and context.message.parts:
for part in context.message.parts:
if hasattr(part.root, 'text'):
                    user_message = part.root.text
break

        result = self._screen_resumes(user_message)
        msg = create_text_message_object(role=Role.agent, content=result)
await event_queue.enqueue_event(msg)

asyncdefcancel(self, context: RequestContext, event_queue: EventQueue):
raise Exception('cancel not supported')

def_screen_resumes(self, requirement: str) -> str:
        candidates = [
            {'name': '张三', 'experience': '6年 Go，分布式系统', 'match': '95%'},
            {'name': '李四', 'experience': '8年 Go，微服务架构', 'match': '90%'},
            {'name': '王五', 'experience': '5年 Go，云原生', 'match': '85%'},
        ]
        lines = [f'根据需求「{requirement}」，筛选出以下候选人：\n']
for c in candidates:
            lines.append(f"- {c['name']}：{c['experience']}（匹配度 {c['match']}）")
return'\n'.join(lines)


if __name__ == '__main__':
    skill = AgentSkill(
        id='resume_screening',
        name='简历筛选',
        description='根据职位要求筛选匹配的候选人',
        tags=['recruiting', 'screening'],
        examples=['筛选高级后端工程师候选人'],
    )

    agent_card = AgentCard(
        name='简历筛选 Agent',
        description='根据职位要求，从候选人库中筛选匹配的人选',
        url='http://127.0.0.1:9999',
        version='1.0.0',
        default_input_modes=['text/plain'],
        default_output_modes=['text/plain'],
        capabilities=AgentCapabilities(streaming=False),
        skills=[skill],
    )

    request_handler = DefaultRequestHandler(
        agent_executor=ResumeScreenerExecutor(),
        task_store=InMemoryTaskStore(),
    )

    a2a_app = A2AStarletteApplication(
        agent_card=agent_card,
        http_handler=request_handler,
    )
    app = a2a_app.build()
    uvicorn.run(app, host='127.0.0.1', port=9999)

总共 60 行左右。Server 端就完成了。

三、写 Client：招聘主管发起协作

Server 端的"简历筛选 Agent"已经就绪了。现在我们来写 Client 端——招聘主管。它要做三件事：

1. 发现：找到简历筛选 Agent（读取 Agent Card）
2. 派任务：发送一条消息，描述招聘需求
3. 收成果：接收筛选结果

这正是上一篇科普里讲的 A2A 工作流程：发现 → 派任务 → 收成果。

完整代码：hiring_manager.py

import asyncio
import uuid
import httpx
from a2a.client import A2ACardResolver, A2AClient
from a2a.client.helpers import create_text_message_object
from a2a.types import MessageSendParams, SendMessageRequest, Role


asyncdefmain():
asyncwith httpx.AsyncClient() as httpx_client:
# 第一步：发现 —— 读取远程 Agent 的名片
        resolver = A2ACardResolver(
            httpx_client=httpx_client,
            base_url='http://127.0.0.1:9999',
        )
        agent_card = await resolver.get_agent_card()
        print(f'发现 Agent：{agent_card.name}')
        print(f'技能：{[s.name for s in agent_card.skills]}\n')

# 第二步：派任务 —— 发送招聘需求
        client = A2AClient(httpx_client=httpx_client, agent_card=agent_card)
        message = create_text_message_object(
            role=Role.user,
            content='筛选高级后端工程师候选人，要求 5 年以上 Go 语言经验，有分布式系统背景',
        )
        request = SendMessageRequest(
            id=str(uuid.uuid4()),
            params=MessageSendParams(message=message),
        )

        print('派发任务...\n')

# 第三步：收成果 —— 接收响应
        response = await client.send_message(request)
        result = response.root.result
        print('收到 Agent 回复：')
for part in result.parts:
if hasattr(part.root, 'text'):
                print(part.root.text)


asyncio.run(main())

逐段对照科普里的概念：

不到 30 行代码，Client 端也完成了。

四、运行 & 验证

现在是最激动人心的时刻——让两个 Agent 跑起来。

启动 Server

打开第一个终端，启动简历筛选 Agent：

cd a2a-demo
source .venv/bin/activate
python resume_screener.py

你会看到类似这样的输出：

INFO:     Started server process [PID]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://127.0.0.1:9999 (Press CTRL+C to quit)

这表示 Agent 已经"上线"了，它的 Agent Card 发布在 http://127.0.0.1:9999/.well-known/agent-card.json。

运行 Client

打开第二个终端，运行招聘主管：

cd a2a-demo
source .venv/bin/activate
python hiring_manager.py

你会看到这样的输出：

发现 Agent：简历筛选 Agent
技能：['简历筛选']

派发任务...

收到 Agent 回复：
根据需求「筛选高级后端工程师候选人，要求 5 年以上 Go 语言经验，有分布式系统背景」，筛选出以下候选人：

- 张三：6年 Go，分布式系统（匹配度 95%）
- 李四：8年 Go，微服务架构（匹配度 90%）
- 王五：5年 Go，云原生（匹配度 85%）

整个过程发生了什么？

1. Client 发现了 Server：通过 A2ACardResolver 读取了 Agent Card，知道了对方叫"简历筛选 Agent"，技能是"简历筛选"
2. Client 派了任务：把招聘需求包装成 Message，通过 SendMessageRequest 发送
3. Server 执行了任务：ResumeScreenerExecutor.execute() 被调用，处理需求，返回结果
4. Client 收到了回复：Agent 返回的 Message 里包含了筛选出的候选人名单

这就是 A2A 协议的完整工作流程——发现、派活、收货，用代码走了一遍。

五、代码 vs 概念：一一对照

让我们用一张表格，把代码中的关键类名映射回上一篇科普里的四个核心概念：

还有几个科普里没详细展开、但代码中出现的重要角色：

概念到代码，只差一层"翻译"。上一篇记住了四个概念，这一篇就能读懂所有代码。

六、总结

回顾一下我们今天做了什么：

1. 安装 a2a-sdk：一个库搞定 Server 和 Client
2. 写了一个 Remote Agent（简历筛选）：定义 AgentCard + 实现 AgentExecutor + 启动服务
3. 写了一个 Client（招聘主管）：发现 Agent Card + 发送任务 + 接收结果
4. 本地运行：两个终端，完整走了一遍 A2A 工作流程

总共约 90 行代码。

当然，今天的示例用的是模拟数据——_screen_resumes 方法里的候选人是写死的。在真实项目中，这里可以替换成：调用大模型做智能匹配、查询数据库获取真实简历等。

回到我们系列的概念链：

LLM → Token → Context → Prompt → Tool → MCP → Agent → Skill → A2A

这条链上的每一环，我们都从"概念"走到了"代码"。A2A 是这条链的最后一环——从单个 Agent 的能力建设，到多个 Agent 的团队协作。

下一篇，我们将使用 Google ADK 来简化这个过程——用一行 to_a2a() 替代 40 行样板代码，实现同样的多 Agent 协作。

如果这篇文章对你有帮助，欢迎点赞、收藏、转发。有任何问题，欢迎在评论区交流。