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"}

القواعد الرئيسية:


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. الإرجاعات الطويلة تزيد استهلاك السياق وزمن الاستجابة.
س هل يمكن للأدوات استدعاء أدوات أخرى؟
ج ليس مباشرة. الوكيل يربط الأدوات تلقائياً في الاستدلال متعدد الخطوات. للعمليات المجمعة، اكتب أداة مركبة.

📖 ملخص


📝 تمارين

  1. أساسي (الصعوبة: ⭐): أنشئ أداة معالجة نصوص بسيطة (عد الكلمات، إزالة التكرار).
  2. متوسط (الصعوبة: ⭐⭐): أنشئ أداة استدعاء API غير متزامنة تدعم GET وPOST.
  3. متقدم (الصعوبة: ⭐⭐⭐): أنشئ سلسلة أدوات "استعلام قاعدة بيانات" بخطوات الاتصال والاستعلام والتنسيق، مع معالجة أخطاء واختبارات كاملة.
Web-Tutorial.com

فريق Web-Tutorial التقني

منصة دروس برمجية يديرها عدة مطورين. كل درس يتم كتابته ومراجعته بواسطة مطورين متخصصين في المجال. نعمل على ضمان دقة وموثوقية المحتوى — إذا لاحظت أي مشكلة، فيرجى إخبارنا.

100%