快速开始

本教程将帮助您在5分钟内搭建环境并运行第一个A2A智能体应用。

环境搭建

系统要求

  • Python 3.8+ 或 Node.js 14+
  • Git
  • 网络连接

安装CLI工具

选择以下安装方式之一:

# 使用pip安装
pip install a2a-cli

# 或使用npm安装
npm install -g @a2a/cli

创建项目

1. 初始化项目

# 创建项目目录
a2a init my-first-agent
cd my-first-agent

目录结构:

my-first-agent/
├── src/
│   ├── agent.py      # 智能体实现
│   └── config.py     # 配置文件
├── tests/            # 测试文件
├── requirements.txt  # 依赖清单
└── README.md        # 项目文档

2. 配置环境

创建配置文件 .env

A2A_API_KEY=your-api-key
A2A_API_URL=https://api.a2acn.com

安装依赖:

pip install -r requirements.txt

开发智能体

1. 实现智能体逻辑

编辑 src/agent.py

from a2a import Agent, Task

class AssistantAgent(Agent):
    def __init__(self):
        super().__init__(name="assistant")
        
    async def handle_chat(self, task: Task):
        message = task.input.get("message", "")
        return {
            "reply": f"您说:{message}"
        }

    async def handle_task(self, task: Task):
        if task.type == "chat":
            return await self.handle_chat(task)
        return {
            "error": "不支持的任务类型"
        }

2. 启动服务

运行以下命令启动智能体服务:

a2a start

服务默认运行在 http://localhost:8000

测试功能

1. 使用CLI

# 发送测试消息
a2a chat "你好,我是小明"

2. 使用SDK

创建测试脚本 test.py

from a2a import Agent, Task

async def main():
    # 创建客户端
    client = Agent()
    
    # 发送消息
    task = await client.create_task(
        type="chat",
        input={"message": "你好,我是小明"}
    )
    
    # 等待结果
    result = await task.wait()
    print(f"收到回复:{result.output['reply']}")

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

运行测试:

python test.py

3. 使用API

curl -X POST http://localhost:8000/v1/tasks \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer your-api-key" \
  -d '{
    "type": "chat",
    "input": {
      "message": "你好,我是小明"
    }
  }'

添加功能

1. 支持更多任务类型

修改 agent.py 添加新功能:

class AssistantAgent(Agent):
    # ...已有代码...

    async def handle_translate(self, task: Task):
        text = task.input.get("text", "")
        target = task.input.get("target", "en")
        # 实现翻译逻辑
        return {
            "translated": f"翻译结果:{text}"
        }

    async def handle_task(self, task: Task):
        handlers = {
            "chat": self.handle_chat,
            "translate": self.handle_translate
        }
        handler = handlers.get(task.type)
        if handler:
            return await handler(task)
        return {
            "error": "不支持的任务类型"
        }

2. 添加工具集成

from a2a.tools import WebSearch, Calculator

class AssistantAgent(Agent):
    def __init__(self):
        super().__init__(name="assistant")
        self.search = WebSearch()
        self.calc = Calculator()
    
    async def handle_chat(self, task: Task):
        message = task.input.get("message", "")
        
        # 处理搜索请求
        if "搜索" in message:
            results = await self.search.search(message)
            return {"reply": f"搜索结果:{results}"}
            
        # 处理计算请求
        if "计算" in message:
            expr = message.replace("计算", "").strip()
            result = await self.calc.evaluate(expr)
            return {"reply": f"计算结果:{result}"}
            
        return {"reply": f"您说:{message}"}

部署服务

1. 构建Docker镜像

创建 Dockerfile

FROM python:3.8-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
CMD ["a2a", "start"]

构建镜像:

docker build -t my-first-agent .

2. 运行容器

docker run -d \
  -p 8000:8000 \
  -e A2A_API_KEY=your-api-key \
  my-first-agent

下一步

恭喜!您已经成功创建并运行了第一个A2A智能体。接下来您可以:

  1. 深入了解核心概念
  2. 探索API文档
  3. 查看示例项目
  4. 加入开发者社区

常见问题

问:服务启动失败怎么办?

检查以下项目:

  1. 环境变量设置是否正确
  2. 端口是否被占用
  3. Python版本是否满足要求
  4. 依赖是否完全安装

问:如何处理API错误?

  1. 验证API密钥是否正确
  2. 确认API端点是否可访问
  3. 查看详细错误日志
  4. 参考故障排除指南

💡 提示:遇到问题时:

  • 使用 a2a doctor 检查环境
  • 通过 a2a logs 查看日志
  • GitHub提交问题