Pi Agent: 自定义工具开发
最后更新：2026-08-31
内置工具不够用？自己写一个——Pi Agent 的工具协议简单到你 10 分钟就能上手。
1. 工具协议
所有 Pi Agent 工具都遵循统一协议：
PYTHON
from pi_agent import tool
@tool(
name="tool_name",
description="工具描述",
parameters={...}
)
def tool_function(param1: str, param2: int = 0) -> dict:
# 工具逻辑
return {"result": "value"}
关键规则：
- 函数必须有类型标注
- 返回值必须是 dict 或 str
- 参数通过装饰器的 parameters 字段声明
- 异常会被 Pi Agent 捕获并转为友好提示
2. 基础工具开发
▶ 示例 1：Markdown 转 HTML 工具（难度⭐）
PYTHON
from pi_agent import tool
import markdown
@tool(
name="md_to_html",
description="将 Markdown 文本转换为 HTML",
parameters={
"md_text": {"type": "string", "description": "Markdown 文本"},
"title": {"type": "string", "description": "页面标题", "required": False}
}
)
def md_to_html(md_text: str, title: str = "") -> dict:
html_body = markdown.markdown(md_text, extensions=["tables", "fenced_code"])
if title:
html = f"<html><head><title>{title}</title></head><body>{html_body}</body></html>"
else:
html = html_body
return {"html": html, "length": len(html)}
▶ 示例 2：JSON 数据查询工具（难度⭐⭐）
PYTHON
from pi_agent import tool
import json
@tool(
name="json_query",
description="从 JSON 数据中查询指定路径的值",
parameters={
"data": {"type": "string", "description": "JSON 字符串"},
"path": {"type": "string", "description": "查询路径，如 users.0.name"}
}
)
def json_query(data: str, path: str) -> dict:
try:
obj = json.loads(data)
except json.JSONDecodeError as e:
return {"error": f"JSON 解析失败: {e}"}
keys = path.split(".")
current = obj
for key in keys:
if key.isdigit():
key = int(key)
try:
current = current[key]
except (KeyError, IndexError, TypeError) as e:
return {"error": f"路径 '{path}' 不存在: {e}"}
return {"value": current, "type": type(current).__name__}
3. 异步工具
需要网络请求或文件 I/O 的工具应使用异步：
PYTHON
from pi_agent import tool
import aiohttp
@tool(
name="http_get",
description="发送 HTTP GET 请求",
parameters={
"url": {"type": "string", "description": "请求 URL"},
"headers": {"type": "object", "description": "请求头", "required": False}
},
async_tool=True
)
async def http_get(url: str, headers: dict = None) -> dict:
async with aiohttp.ClientSession() as session:
async with session.get(url, headers=headers) as resp:
body = await resp.text()
return {
"status": resp.status,
"body": body[:5000],
"headers": dict(resp.headers)
}
4. 工具链
多个工具可以串联成处理流水线：
PYTHON
from pi_agent import Agent, tool
@tool(name="fetch_url", description="获取 URL 内容")
def fetch_url(url: str) -> dict:
import requests
resp = requests.get(url)
return {"content": resp.text}
@tool(name="extract_links", description="提取 HTML 中的链接")
def extract_links(html: str) -> dict:
import re
links = re.findall(r'href=["\']([^"\']+)["\']', html)
return {"links": links}
@tool(name="check_status", description="检查 URL 是否可访问")
def check_status(url: str) -> dict:
import requests
try:
resp = requests.head(url, timeout=5)
return {"url": url, "status": resp.status_code, "ok": resp.status_code < 400}
except Exception as e:
return {"url": url, "status": 0, "ok": False, "error": str(e)}
agent = Agent(name="link_checker", tools=["fetch_url", "extract_links", "check_status"])
result = agent.run("检查 https://example.com 上所有链接的可用性")
5. 错误处理
(1) 工具内部错误
PYTHON
@tool(name="safe_divide", description="安全除法")
def safe_divide(a: float, b: float) -> dict:
if b == 0:
return {"error": "除数不能为零", "suggestion": "请提供非零的除数"}
return {"result": a / b}
(2) 参数验证
PYTHON
from pi_agent import tool, ToolError
@tool(name="send_email", description="发送邮件")
def send_email(to: str, subject: str, body: str) -> dict:
if "@" not in to:
raise ToolError("收件人邮箱格式不正确")
if not subject.strip():
raise ToolError("邮件主题不能为空")
# 发送逻辑...
return {"status": "sent", "to": to}
6. 工具测试
PYTHON
import pytest
from my_tools import json_query
def test_json_query_basic():
result = json_query('{"name": "Alice"}', "name")
assert result["value"] == "Alice"
def test_json_query_nested():
data = '{"users": [{"name": "Bob"}]}'
result = json_query(data, "users.0.name")
assert result["value"] == "Bob"
def test_json_query_invalid_path():
result = json_query('{"a": 1}', "b")
assert "error" in result
❓ 常见问题
Q 工具函数能访问 Agent 的状态吗？
A 不能直接访问。但可以通过 context 参数传入 Agent 状态，或使用钩子函数在工具调用前后读取/修改状态。
Q 工具返回的数据量有限制吗？
A 没有硬性限制，但建议控制在 10KB 以内。超长返回值会增加上下文消耗和延迟。
Q 工具能调用其他工具吗？
A 不能直接调用。但 Agent 会在多步推理中自动串联工具。如果你需要组合操作，写一个组合工具更高效。
📖 小节
- 工具遵循统一协议：@tool 装饰器 + 类型标注 + dict 返回值
- 网络和 I/O 工具用 async 模式
- 工具链让多个工具自动串联
- 错误处理用返回 error 字典或抛出 ToolError
- 用 pytest 测试工具的输入输出
📝 作业
- 基础题（难度⭐）：创建一个简单的字符串处理工具（如统计字数、去重）。
- 进阶题（难度⭐⭐）：创建一个异步的 API 调用工具，支持 GET 和 POST 方法。
- 挑战题（难度⭐⭐⭐）：创建一个"数据库查询"工具链，包含连接、查询、格式化三个步骤，并编写完整的错误处理和测试。
