Pi Agent: Custom Tool Development
آخر تحديث: 2026-08-31
--- title: "تطوير الأدوات المخصصة" description: "تعمّق في تطوير أدوات Pi Agent المخصصة، بما في ذلك بروتوكول الأدوات والتحقق من المعلمات ومعالجة الأخطاء والأنماط المتقدمة." order: 15 lang: ar
الأدوات المدمجة غير كافية؟ اكتب أداتك — بروتوكول أدوات 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
❓ أسئلة شائعة
س هل يمكن للأدوات الوصول إلى حالة الوكيل؟
ج ليس مباشرة. مرر حالة الوكيل عبر معلمة context، أو استخدم خطافات لقراءة/تعديل الحالة حول استدعاءات الأدوات.
س حد حجم بيانات إرجاع الأداة؟
ج لا حد صارم، لكن يُنصح بالبقاء تحت 10KB. الإرجاعات الطويلة تزيد استهلاك السياق وزمن الاستجابة.
س هل يمكن للأدوات استدعاء أدوات أخرى؟
ج ليس مباشرة. الوكيل يربط الأدوات تلقائياً في الاستدلال متعدد الخطوات. للعمليات المجمعة، اكتب أداة مركبة.
📖 ملخص
- بروتوكول الأدوات: مُزخرف @tool + تعليقات توضيحية للنوع + قيمة إرجاع dict
- أدوات الشبكة/I/O تستخدم الوضع غير المتزامن
- سلاسل الأدوات تربط أدوات متعددة تلقائياً
- معالجة الأخطاء: أرجع error dict أو ارفع ToolError
- اختبر بـ pytest للتحقق من المدخلات/المخرجات
📝 تمارين
- أساسي (الصعوبة: ⭐): أنشئ أداة معالجة نصوص بسيطة (عد الكلمات، إزالة التكرار).
- متوسط (الصعوبة: ⭐⭐): أنشئ أداة استدعاء API غير متزامنة تدعم GET وPOST.
- متقدم (الصعوبة: ⭐⭐⭐): أنشئ سلسلة أدوات "استعلام قاعدة بيانات" بخطوات الاتصال والاستعلام والتنسيق، مع معالجة أخطاء واختبارات كاملة.