我要投稿

A2A实战：用 Python 实现 A2A交互的简单示例

发布日期：2025-07-20 08:53:04 浏览次数： 1529

作者：蛋哥学测开

微信搜一搜，关注“蛋哥学测开”

人工智能技术日益发展的今天，Agent 之间的自动通信与协作正变得越来越重要。今天我们来实战一下，使用 Python 实现一个简单的 A2A（Agent-to-Agent）交互示例。

✅ 前提条件

在开始之前，请确保你的开发环境中已安装以下工具：

Python 3.10 或更高版本
Git（用于克隆项目仓库）

🧰 环境配置步骤

第一步：克隆项目仓库

访问 GitHub 并克隆官方提供的 A2A 示例仓库：

bash深色版本
git clone https://github.com/google-a2a/a2a-samples.git -b main --depth 1
cd a2a-samples

注意：该项目是 Google A2A 的官方示例代码库，我们只拉取最新主分支的简化版本。

第二步：创建并激活虚拟环境

Windows 用户：

bash深色版本
python -m venv .venv
.venv\Scripts\activate

macOS / Linux 用户：

bash深色版本
python -m venv .venv
source .venv/bin/activate

第三步：安装依赖

进入项目目录后，安装所需的 Python 包以及 A2A SDK：

bash深色版本
pip install -r samples/python/requirements.txt

第四步：验证安装是否成功

你可以运行以下命令再次确认依赖是否安装完整：

bash深色版本
pip install -r samples/python/requirements.txt

如果看到类似输出：

深色版本
A2A SDK imported successfully

恭喜！你的开发环境已经准备就绪！

🚀 启动服务端

从项目根目录启动服务端：

bash深色版本
python samples/python/agents/helloworld/__main__.py

服务启动后，你会看到如下日志信息：

深色版本
INFO:     Started server process [xxxxx]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:9999 (Press CTRL+C to quit)

此时，服务已成功运行在 http://localhost:9999 上。

💬 客户端交互测试

打开一个新的终端窗口，保持虚拟环境激活状态，运行客户端脚本：

bash深色版本
python samples/python/agents/helloworld/test_client.py

🔍 客户端代码解析

让我们看看客户端是如何实现与服务端通信的。

1️⃣ 初始化 Agent Card 解析器

python深色版本
base_url = 'http://localhost:9999'

asyncwith httpx.AsyncClient() as httpx_client:
    resolver = A2ACardResolver(
        httpx_client=httpx_client,
        base_url=base_url,
# agent_card_path 使用默认值，extended_agent_card_path 也使用默认值
    )

这段代码初始化了一个异步 HTTP 客户端，并通过 A2ACardResolver 获取目标 Agent 的能力描述（Card），为后续交互做准备。

2️⃣ 发送非流式消息请求

python深色版本
client = A2AClient(httpx_client=httpx_client, agent_card=final_agent_card_to_use)

logger.info('A2AClient initialized.')

send_message_payload: dict[str, Any] = {
'message': {
'role': 'user',
'parts': [
            {'kind': 'text', 'text': 'how much is 10 USD in INR?'}
        ],
'messageId': uuid4().hex,
    },
}

request = SendMessageRequest(id=str(uuid4()), params=MessageSendParams(**send_message_payload))

response = await client.send_message(request)
print(response.model_dump(mode='json', exclude_none=True))

上述代码构建了一个包含用户问题的 JSON 请求，并通过 A2AClient 发送给服务端 Agent。

3️⃣ 预期响应结果

运行完成后，你应该会看到类似如下的响应：

json深色版本
{
"jsonrpc": "2.0",
"id": "xxxxxxxx",
"result": {
"type": "message",
"role": "agent",
"parts": [
      { "type": "text", "text": "Hello World" }
    ],
"messageId": "yyyyyyyy"
  }
}