Pi Agent: Event System & Command Registration
Last updated: 2026-08-31
The event system transforms the Agent from "you ask, it answers" into true two-way communication.
1. Event System Overview
Pi Agent uses an event-driven architecture:
TEXT
📖 Display only
Event Flow
Trigger Source → Event Bus → Handler
User input Dispatch/route Tool calls
Tool return Filter/sort UI updates
Timer Priority sort Logging
External msg Error routing Notifications
2. Built-in Events
| Event | Trigger | Data |
|---|---|---|
| on_chat_start | Conversation starts | session_id |
| on_chat_end | Conversation ends | session_id, summary |
| on_user_message | User sends message | message |
| on_agent_response | Agent responds | response |
| on_tool_call | Tool called | tool_name, params |
| on_tool_result | Tool returns result | tool_name, result |
| on_error | Error occurs | error, context |
| on_model_switch | Model switched | old_model, new_model |
| on_context_overflow | Context overflow | size, limit |
3. Event Listening
(1) Decorator Style
PYTHON
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}")
(2) Class Style
PYTHON
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())
4. Custom Events
(1) Define Event
PYTHON
from pi_agent import Event
class DeployEvent(Event):
name = "deploy"
fields = ["environment", "status", "url"]
(2) Emit Event
PYTHON
agent.emit("deploy", {
"environment": "production",
"status": "success",
"url": "https://myapp.example.com"
})
(3) Listen to Custom Event
PYTHON
@agent.on("deploy")
def on_deploy(event):
if event.status == "success":
send_notification(f"Deploy succeeded: {event.url}")
5. Command Registration
(1) Register Interactive Commands
PYTHON
from pi_agent import Agent
agent = Agent(name="custom_cmd")
@agent.command("/deploy", description="Deploy project to specified environment")
def deploy_cmd(args: str):
env = args.strip() or "staging"
result = agent.run(f"Deploy current project to {env}")
print(result)
@agent.command("/review", description="Review code in specified file")
def review_cmd(args: str):
filename = args.strip()
result = agent.run(skill="code_review", file=filename)
print(result)
@agent.command("/cost", description="Show token usage stats for current session")
def cost_cmd(args: str):
usage = agent.session.get_usage()
print(f"Token usage: {usage.total_tokens}")
print(f"Estimated cost: ${usage.estimated_cost:.4f}")
6. Event Filters
(1) Conditional Filtering
PYTHON
@agent.on("tool_call", filter=lambda e: e.tool_name == "shell")
def log_shell_calls(event):
print(f"Shell command: {event.params.get('cmd')}")
(2) Priority
PYTHON
@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")
FAQ
Q Can event handlers modify event data?
A Yes, but it's recommended to only read to avoid side effects. Modified data is visible to subsequent handlers.
Q Sync or async event handling?
A Default synchronous, executed by priority. Use
async_handler=True for async handlers.Q Commands vs slash commands?
A Commands are custom extensions via
@agent.command(). Slash commands are built-in interactive mode commands (/help, /exit). Same format, different sources.Summary
- Event-driven: trigger → event bus → handler
- 9 built-in events covering full Agent lifecycle
- Two listening styles: decorator and EventHandler class
- Custom events and command registration extend interactive capabilities
- Filters and priority control event processing
Exercises
- Basic (Difficulty: ⭐): Listen to tool_call events, log all tool calls to a file.
- Intermediate (Difficulty: ⭐⭐): Create a /summarize command that generates a session summary.
- Advanced (Difficulty: ⭐⭐⭐): Implement an event-driven deployment pipeline: code review → test → deploy, each stage triggered by events.
