Pi Agent: Custom Tool Development
最終更新：2026-08-31
--- title: "カスタムツール開発" description: "Pi Agentのカスタムツール開発を深く学び、ツールプロトコル、パラメータ検証、エラー処理を習得します。" order: 15 lang: ja
組み込みツールじゃ足りない？自分で書けばいい——Pi Agentのツールプロトコルはシンプルで、10分で始められる。
1. ツールプロトコル
すべてのPi Agentツールは統一されたプロトコルに従います：
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"}
主要ルール：
- 関数には型アノテーションが必須
- 戻り値はdictまたはstrでなければならない
- パラメータはデコレータのparametersフィールドで宣言
- 例外はPi Agentがキャッチしてフレンドリーなメッセージに変換
2. 基本的なツール開発
例1：Markdown→HTML変換ツール（難易度：⭐）
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)}
例2：JSONクエリツール（難易度：⭐⭐）
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. 非同期ツール
ネットワークやファイルI/Oを必要とするツールは非同期にすべきです：
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. ツールチェーン
複数のツールを処理パイプラインにチェーンできます：
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. エラー処理
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. ツールのテスト
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ツールは非同期モードを使用
- ツールチェーンで複数ツールを自動的にチェーン
- エラー処理：error dictを返すかToolErrorをraise
- pytestで入出力の検証テスト
📝 練習問題
- 基礎（難易度：⭐）: シンプルな文字列処理ツール（単語数カウント、重複排除）を作成してください。
- 中級（難易度：⭐⭐）: GETとPOSTをサポートする非同期API呼び出しツールを作成してください。
- 上級（難易度：⭐⭐⭐）: connect、query、formatステップを持つ「データベースクエリ」ツールチェーンを作成し、エラー処理とテストを完備させてください。
