A2A Protocol
A2A Samples: Hello World Agent

项目概述

这是一个基于 A2A (Agent-to-Agent) SDK 的简单 Hello World 代理示例。该项目演示了如何创建一个基本的智能代理服务器,它可以响应用户消息并返回简单的问候语。

依赖版本说明

Python 版本要求

  • Python >= 3.10

核心依赖包

包名 版本 用途
a2a-sdk >= 0.2.5 A2A 核心 SDK,提供代理框架
uvicorn >= 0.34.2 ASGI 服务器,用于运行 Web 应用
click >= 8.1.8 命令行界面工具
httpx >= 0.28.1 异步 HTTP 客户端
pydantic >= 2.11.4 数据验证和序列化
python-dotenv >= 1.1.0 环境变量管理
langchain-google-genai >= 2.1.4 Google 生成式 AI 集成
langgraph >= 0.4.1 语言图处理框架

项目结构

helloworld/
├── __init__.py              # 包初始化文件
├── __main__.py              # 主程序入口
├── agent_executor.py        # 代理执行器实现
├── test_client.py           # 测试客户端
├── pyproject.toml           # 项目配置和依赖
├── uv.lock                  # 依赖锁定文件
└── README.md                # 项目说明

环境搭建

1. 安装 UV 包管理器

如果您还没有安装 UV,请先安装:

# macOS/Linux
curl -LsSf https://astral.sh/uv/install.sh | sh

# 或者使用 pip
pip install uv

2. 克隆项目

git clone https://github.com/google-a2a/a2a-samples.git
cd a2a-samples/samples/python/agents/helloworld

3. 安装依赖

UV 会自动根据 pyproject.tomluv.lock 文件安装所有依赖:

uv sync

代码架构说明

核心组件

1. HelloWorldAgent (agent_executor.py)

class HelloWorldAgent:
    """Hello World Agent."""
    
    async def invoke(self) -> str:
        return 'Hello World'

2. HelloWorldAgentExecutor (agent_executor.py)

class HelloWorldAgentExecutor(AgentExecutor):
    """代理执行器实现"""
    
    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
        result = await self.agent.invoke()
        event_queue.enqueue_event(new_agent_text_message(result))

3. 服务器配置 (__main__.py)

  • 定义代理技能 (AgentSkill)
  • 配置公共和扩展代理卡片 (AgentCard)
  • 设置请求处理器和任务存储
  • 启动 Uvicorn 服务器

运行步骤

1. 启动代理服务器

uv run .

服务器将在 http://localhost:9999 启动。

2. 运行测试客户端

在另一个终端窗口中:

uv run test_client.py

3. 验证服务

您可以通过以下方式验证服务是否正常运行:

访问代理卡片信息

curl http://localhost:9999/.well-known/agent.json

访问扩展代理卡片(需要认证)

curl -H "Authorization: Bearer dummy-token-for-extended-card" \
     http://localhost:9999/agent/authenticatedExtendedCard

项目流程图

A2A 客户端与服务器交互流程

sequenceDiagram
    participant Client as A2A Client
    participant Server as A2A Server
    participant Agent as HelloWorldAgent
    participant Queue as EventQueue
    
    Note over Client,Server: 1. 服务发现阶段
    Client->>Server: GET /.well-known/agent.json
    Server-->>Client: 返回公共代理卡片
    
    Note over Client,Server: 2. 扩展卡片获取(可选)
    Client->>Server: GET /agent/authenticatedExtendedCard<br/>(with Bearer token)
    Server-->>Client: 返回扩展代理卡片
    
    Note over Client,Server: 3. 消息发送流程
    Client->>Server: POST /agent/message<br/>{"message": {"role": "user", "parts": [...]}}
    
    Note over Server,Queue: 4. 服务器内部处理
    Server->>Agent: 调用 HelloWorldAgentExecutor.execute()
    Agent->>Agent: 执行 HelloWorldAgent.invoke()
    Agent-->>Queue: 生成 "Hello World" 消息
    Queue-->>Server: 返回事件队列结果
    
    Note over Client,Server: 5. 响应返回
    Server-->>Client: 返回消息响应
    
    Note over Client,Server: 6. 流式消息处理(可选)
    Client->>Server: POST /agent/message/stream
    Server-->>Client: 流式返回消息块
    Server-->>Client: 消息块 1
    Server-->>Client: 消息块 2
    Server-->>Client: 结束标记

系统架构流程图

graph TB
    subgraph "客户端环境"
        C1[测试客户端启动]
        C2[A2ACardResolver<br/>解析代理卡片]
        C3[A2AClient<br/>初始化]
        C4[发送消息请求]
        C5[处理响应]
    end
    
    subgraph "网络通信"
        N1[HTTP/HTTPS 请求]
        N2[JSON 数据传输]
    end
    
    subgraph "服务器环境"
        S1[A2AStarletteApplication<br/>Web 服务器]
        S2[DefaultRequestHandler<br/>请求处理器]
        S3[HelloWorldAgentExecutor<br/>代理执行器]
        S4[HelloWorldAgent<br/>核心代理逻辑]
        S5[InMemoryTaskStore<br/>任务存储]
        S6[EventQueue<br/>事件队列]
    end
    
    subgraph "代理配置"
        A1[公共代理卡片<br/>基础技能]
        A2[扩展代理卡片<br/>增强技能]
    end
    
    %% 客户端流程
    C1 --> C2
    C2 --> C3
    C3 --> C4
    C4 --> C5
    
    %% 网络通信
    C4 --> N1
    N1 --> N2
    N2 --> S1
    S1 --> N2
    N2 --> C5
    
    %% 服务器流程
    S1 --> S2
    S2 --> S3
    S3 --> S4
    S4 --> S6
    S6 --> S3
    S3 --> S2
    S2 --> S1
    
    %% 配置关联
    A1 --> S1
    A2 --> S1
    S2 --> S5
    
    %% 样式
    style C3 fill:#e3f2fd
    style S1 fill:#f3e5f5
    style S4 fill:#e8f5e8
    style N2 fill:#fff3e0

API 端点

公共端点

端点 方法 描述
/.well-known/agent.json GET 获取公共代理卡片信息
/agent/message POST 发送消息给代理
/agent/message/stream POST 流式发送消息

认证端点

端点 方法 描述 认证
/agent/authenticatedExtendedCard GET 获取扩展代理卡片 Bearer Token

技能配置

基础技能

  • ID: hello_world
  • 名称: Returns hello world
  • 描述: just returns hello world
  • 示例: ['hi', 'hello world']

扩展技能(需要认证)

  • ID: super_hello_world
  • 名称: Returns a SUPER Hello World
  • 描述: A more enthusiastic greeting, only for authenticated users
  • 示例: ['super hi', 'give me a super hello']

故障排除

常见问题

  1. 端口被占用

    # 检查端口使用情况
    lsof -i :9999
    # 杀死占用进程
    kill -9 <PID>
    
  2. 依赖安装失败

    # 清理缓存重新安装
    uv cache clean
    uv sync --reinstall
    
  3. Python 版本不兼容

    # 检查 Python 版本
    python --version
    # 确保版本 >= 3.10
    

扩展开发

添加新技能

  1. __main__.py 中定义新的 AgentSkill
  2. 修改 agent_executor.py 中的逻辑处理
  3. 更新代理卡片配置

集成外部 API

  1. pyproject.toml 中添加新依赖
  2. agent_executor.py 中实现 API 调用
  3. 处理异步响应和错误

总结

这个 Hello World 示例展示了 A2A SDK 的基本用法,包括:

  • 代理服务器的创建和配置
  • 技能定义和管理
  • 客户端-服务器通信
  • 认证和扩展功能

通过这个示例,您可以快速了解如何构建自己的智能代理应用。