Pi Agent: Custom Tool Development
Last updated: 2026-08-31
Built-in tools not enough? Write your own — Pi Agent's tool protocol is so simple you can get started in 10 minutes.
1. Tool Protocol
All Pi Agent tools follow a unified protocol:
PYTHON
from pi_agent import tool
@tool(
name="tool_name",
description="Tool description",
parameters={...}
)
def tool_function(param1: str, param2: int = 0) -> dict:
return {"result": "value"}
Key rules:
- Functions must have type annotations
- Return value must be dict or str
- Parameters declared via the decorator's parameters field
- Exceptions are caught by Pi Agent and converted to friendly messages
2. Basic Tool Development
Example 1: Markdown to HTML Tool (Difficulty: ⭐)
PYTHON
from pi_agent import tool
import markdown
@tool(
name="md_to_html",
description="Convert Markdown text to HTML",
parameters={
"md_text": {"type": "string", "description": "Markdown text"},
"title": {"type": "string", "description": "Page title", "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)}
Example 2: JSON Query Tool (Difficulty: ⭐⭐)
PYTHON
from pi_agent import tool
import json
@tool(
name="json_query",
description="Query values from JSON data by path",
parameters={
"data": {"type": "string", "description": "JSON string"},
"path": {"type": "string", "description": "Query path, e.g. 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 parse failed: {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 '{path}' not found: {e}"}
return {"value": current, "type": type(current).__name__}
3. Async Tools
Tools needing network or file I/O should be async:
PYTHON
from pi_agent import tool
import aiohttp
@tool(
name="http_get",
description="Send HTTP GET request",
parameters={
"url": {"type": "string", "description": "Request URL"},
"headers": {"type": "object", "description": "Request headers", "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. Tool Chains
Multiple tools can be chained into a processing pipeline:
PYTHON
from pi_agent import Agent, tool
@tool(name="fetch_url", description="Fetch URL content")
def fetch_url(url: str) -> dict:
import requests
resp = requests.get(url)
return {"content": resp.text}
@tool(name="extract_links", description="Extract links from HTML")
def extract_links(html: str) -> dict:
import re
links = re.findall(r'href=["\']([^"\']+)["\']', html)
return {"links": links}
@tool(name="check_status", description="Check if URL is accessible")
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("Check all links on https://example.com for availability")
5. Error Handling
PYTHON
from pi_agent import tool, ToolError
@tool(name="safe_divide", description="Safe division")
def safe_divide(a: float, b: float) -> dict:
if b == 0:
return {"error": "Division by zero", "suggestion": "Provide a non-zero divisor"}
return {"result": a / b}
@tool(name="send_email", description="Send email")
def send_email(to: str, subject: str, body: str) -> dict:
if "@" not in to:
raise ToolError("Invalid recipient email format")
if not subject.strip():
raise ToolError("Email subject cannot be empty")
return {"status": "sent", "to": to}
6. Tool Testing
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
FAQ
Q Can tools access Agent state?
A Not directly. Pass Agent state via context parameter, or use hooks to read/modify state around tool calls.
Q Tool return data size limit?
A No hard limit, but recommend keeping under 10KB. Long returns increase context consumption and latency.
Q Can tools call other tools?
A Not directly. The Agent chains tools automatically in multi-step reasoning. For combined operations, write a composite tool.
Summary
- Tool protocol: @tool decorator + type annotations + dict return value
- Network/IO tools use async mode
- Tool chains let multiple tools chain automatically
- Error handling: return error dict or raise ToolError
- Test with pytest for input/output validation
Exercises
- Basic (Difficulty: ⭐): Create a simple string processing tool (word count, dedup).
- Intermediate (Difficulty: ⭐⭐): Create an async API call tool supporting GET and POST.
- Advanced (Difficulty: ⭐⭐⭐): Create a "database query" tool chain with connect, query, and format steps, complete with error handling and tests.
