快速开始
快速开始
本教程将帮助您在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智能体。接下来您可以:
常见问题
问:服务启动失败怎么办?
检查以下项目:
- 环境变量设置是否正确
- 端口是否被占用
- Python版本是否满足要求
- 依赖是否完全安装
问:如何处理API错误?
- 验证API密钥是否正确
- 确认API端点是否可访问
- 查看详细错误日志
- 参考故障排除指南
💡 提示:遇到问题时:
- 使用
a2a doctor
检查环境- 通过
a2a logs
查看日志- 在GitHub提交问题