Pi Agent: Event System & Command Registration

تدفق الأحداث
مصدر المشغل → ناقل الأحداث → المعالج
  إدخال المستخدم    توزيع/توجيه      استدعاءات الأدوات
  عودة الأداة      تصفية/فرز        تحديثات الواجهة
  مؤقت             فرز بالأولوية     تسجيل الدخول
  رسالة خارجية     توجيه الأخطاء    إشعارات

from pi_agent import Agent

agent = Agent(name="monitored")

@agent.on("tool_call")
def log_tool_call(event):
    print(f"Tool called: {event.tool_name}({event.params})")

@agent.on("error")
def handle_error(event):
    print(f"Error: {event.error}")
    with open("error_log.txt", "a") as f:
        f.write(f"{event.error}\n")

@agent.on("agent_response")
def log_response(event):
    print(f"Token usage: {event.response.usage}")

from pi_agent import Agent, EventHandler

class MyHandler(EventHandler):
    def on_tool_call(self, event):
        print(f"Tool: {event.tool_name}")

    def on_tool_result(self, event):
        print(f"Result: {event.result}")

    def on_error(self, event):
        print(f"Error: {event.error}")

agent = Agent(name="monitored", event_handler=MyHandler())

from pi_agent import Event

class DeployEvent(Event):
    name = "deploy"
    fields = ["environment", "status", "url"]

agent.emit("deploy", {
    "environment": "production",
    "status": "success",
    "url": "https://myapp.example.com"
})

@agent.on("deploy")
def on_deploy(event):
    if event.status == "success":
        send_notification(f"Deploy succeeded: {event.url}")

from pi_agent import Agent

agent = Agent(name="custom_cmd")

@agent.command("/deploy", description="نشر المشروع إلى البيئة المحددة")
def deploy_cmd(args: str):
    env = args.strip() or "staging"
    result = agent.run(f"انشر المشروع الحالي إلى {env}")
    print(result)

@agent.command("/review", description="مراجعة الكود في الملف المحدد")
def review_cmd(args: str):
    filename = args.strip()
    result = agent.run(skill="code_review", file=filename)
    print(result)

@agent.command("/cost", description="عرض إحصائيات استخدام الرموز للجلسة الحالية")
def cost_cmd(args: str):
    usage = agent.session.get_usage()
    print(f"استخدام الرموز: {usage.total_tokens}")
    print(f"التكلفة المقدرة: ${usage.estimated_cost:.4f}")

@agent.on("tool_call", filter=lambda e: e.tool_name == "shell")
def log_shell_calls(event):
    print(f"Shell command: {event.params.get('cmd')}")

@agent.on("error", priority=10)
def critical_error(event):
    send_alert(f"Critical error: {event.error}")

@agent.on("error", priority=1)
def log_error(event):
    with open("errors.log", "a") as f:
        f.write(f"{event.error}\n")