Hook Events - PraisonAI

Hook events are triggered at specific points in the agent lifecycle, allowing you to intercept, modify, or block operations.

Quick Start

Import Hook Components

from praisonaiagents.hooks import HookRegistry, HookEvent, HookResult

registry = HookRegistry()

@registry.on(HookEvent.BEFORE_TOOL)
def log_tool(event_data):
    print(f"Tool: {event_data.tool_name}")
    return HookResult.allow()

Use with Agent

from praisonaiagents import Agent

agent = Agent(
    name="MyAgent",
    instructions="You are helpful",
    hooks=registry
)

Core Events

Tool Events

Event	Trigger	Input Type	Use Case
`BEFORE_TOOL`	Before tool execution	`BeforeToolInput`	Security checks, logging
`AFTER_TOOL`	After tool execution	`AfterToolInput`	Result validation, logging

@registry.on(HookEvent.BEFORE_TOOL)
def before_tool(event_data):
    print(f"Calling: {event_data.tool_name}")
    print(f"Args: {event_data.arguments}")
    return HookResult.allow()

@registry.on(HookEvent.AFTER_TOOL)
def after_tool(event_data):
    print(f"Result: {event_data.result}")
    return HookResult.allow()

Agent Events

Event	Trigger	Input Type	Use Case
`BEFORE_AGENT`	Before agent runs	`BeforeAgentInput`	Setup, initialization
`AFTER_AGENT`	After agent completes	`AfterAgentInput`	Cleanup, reporting

@registry.on(HookEvent.BEFORE_AGENT)
def before_agent(event_data):
    print(f"Agent starting: {event_data.agent_name}")
    return HookResult.allow()

@registry.on(HookEvent.AFTER_AGENT)
def after_agent(event_data):
    print(f"Agent completed: {event_data.result}")
    return HookResult.allow()

LLM Events

Event	Trigger	Input Type	Use Case
`BEFORE_LLM`	Before LLM API call	`BeforeLLMInput`	Request modification
`AFTER_LLM`	After LLM response	`AfterLLMInput`	Response validation

@registry.on(HookEvent.BEFORE_LLM)
def before_llm(event_data):
    print(f"Model: {event_data.model}")
    print(f"Messages: {len(event_data.messages)}")
    return HookResult.allow()

@registry.on(HookEvent.AFTER_LLM)
def after_llm(event_data):
    print(f"Response length: {len(event_data.response)}")
    print(f"Tokens: {event_data.usage}")
    return HookResult.allow()

Session Events

Event	Trigger	Input Type	Use Case
`SESSION_START`	When session starts	`SessionStartInput`	Session initialization
`SESSION_END`	When session ends	`SessionEndInput`	Session cleanup

@registry.on(HookEvent.SESSION_START)
def session_start(event_data):
    print(f"Session: {event_data.session_id}")
    return HookResult.allow()

@registry.on(HookEvent.SESSION_END)
def session_end(event_data):
    print(f"Session ended: {event_data.session_id}")
    return HookResult.allow()

Error Events

Event	Trigger	Input Type	Use Case
`ON_ERROR`	When error occurs	`OnErrorInput`	Error handling
`ON_RETRY`	Before retry attempt	`OnRetryInput`	Retry logic

@registry.on(HookEvent.ON_ERROR)
def on_error(event_data):
    print(f"Error: {event_data.error}")
    # Log to external service
    return HookResult.allow()

@registry.on(HookEvent.ON_RETRY)
def on_retry(event_data):
    # New tool-level fields
    print(f"[retry] {event_data.tool_name} attempt {event_data.attempt}/{event_data.max_attempts}")
    print(f"Error type: {event_data.error_type}, delay: {event_data.delay_ms}ms")
    
    # Legacy fields still available for backward compatibility
    # event_data.retry_count, event_data.max_retries, event_data.error_message
    
    if event_data.attempt > 2:
        return HookResult.deny("Too many retries")
    return HookResult.allow()

OnRetryInput Fields

The OnRetryInput event includes both new tool-specific fields and legacy fields for backward compatibility: New fields (recommended):

tool_name: Name of the failing tool
attempt: Current attempt number (1-based)
max_attempts: Maximum attempts configured
delay_ms: Delay before this retry in milliseconds
error_type: Classified error type (timeout, rate_limit, connection_error, unknown)
error: Original exception object

Legacy fields (for backward compatibility):

retry_count: Same as attempt
max_retries: Same as max_attempts
error_message: String representation of error

Agent-Level Error Callbacks

In addition to hook events, agents support a direct on_error callback for LLM failures:

from praisonaiagents import Agent
from praisonaiagents.errors import LLMError

def agent_error_handler(error):
    """Called when LLM chat completion fails"""
    print(f"Agent {error.agent_id} LLM error: {error.message}")
    print(f"Model: {error.model_name}")
    print(f"Retryable: {error.is_retryable}")
    
    # Custom error recovery logic
    if error.is_retryable:
        print("Will be retried by orchestration")
    else:
        print("Fatal error - check configuration")

# Agent-specific error handling
agent = Agent(
    name="Error Aware Agent",
    instructions="Process user requests",
    on_error=agent_error_handler  # Direct callback, not a hook
)

Hook Events vs Agent Callbacks

Approach	Scope	Return Value	When Called
HookEvent.ON_ERROR	Global, all agents	`HookResult`	Any hook system error
agent.on_error	Single agent	None	LLM chat completion errors

# Using both together for comprehensive error handling
registry = HookRegistry()

@registry.on(HookEvent.ON_ERROR)
def global_hook_error(event_data):
    """Catches all hook system errors"""
    return HookResult.allow()

def llm_specific_error(error):
    """Catches LLM errors for this agent only"""
    pass

agent = Agent(
    name="Comprehensive Error Handling",
    hooks=registry,           # Global error hooks
    on_error=llm_specific_error  # Agent-specific LLM errors
)

Extended Events

User Interaction Events

Event	Trigger	Use Case
`USER_PROMPT_SUBMIT`	User submits prompt	Input validation, logging
`NOTIFICATION`	Notification sent	Alert routing

@registry.on(HookEvent.USER_PROMPT_SUBMIT)
def on_prompt(event_data):
    print(f"User prompt: {event_data.prompt}")
    # Validate or modify input
    return HookResult.allow()

@registry.on(HookEvent.NOTIFICATION)
def on_notification(event_data):
    print(f"Notification: {event_data.message}")
    # Route to external service
    return HookResult.allow()

Subagent Events

Event	Trigger	Use Case
`SUBAGENT_STOP`	Subagent completes	Result handling

@registry.on(HookEvent.SUBAGENT_STOP)
def on_subagent_stop(event_data):
    print(f"Subagent completed: {event_data.agent_name}")
    print(f"Result: {event_data.result}")
    return HookResult.allow()

System Events

Event	Trigger	Use Case
`SETUP`	Initialization/maintenance	Config loading
`BEFORE_COMPACTION`	Before context compaction	Pre-compaction hooks
`AFTER_COMPACTION`	After context compaction	Post-compaction validation

@registry.on(HookEvent.SETUP)
def on_setup(event_data):
    print("System initializing...")
    # Load configuration
    return HookResult.allow()

@registry.on(HookEvent.BEFORE_COMPACTION)
def before_compaction(event_data):
    print(f"Compacting context: {event_data.token_count} tokens")
    return HookResult.allow()

@registry.on(HookEvent.AFTER_COMPACTION)
def after_compaction(event_data):
    print(f"Compacted to: {event_data.new_token_count} tokens")
    return HookResult.allow()

Message Events

Event	Trigger	Use Case
`MESSAGE_RECEIVED`	Message received	Preprocessing
`MESSAGE_SENDING`	Before message sent	Modification
`MESSAGE_SENT`	After message sent	Confirmation

@registry.on(HookEvent.MESSAGE_RECEIVED)
def on_message_received(event_data):
    print(f"Received: {event_data.message}")
    return HookResult.allow()

@registry.on(HookEvent.MESSAGE_SENDING)
def on_message_sending(event_data):
    print(f"Sending: {event_data.message}")
    return HookResult.allow()

Gateway Events

Event	Trigger	Use Case
`GATEWAY_START`	Gateway starts	Initialization
`GATEWAY_STOP`	Gateway stops	Cleanup

@registry.on(HookEvent.GATEWAY_START)
def on_gateway_start(event_data):
    print("Gateway starting...")
    return HookResult.allow()

@registry.on(HookEvent.GATEWAY_STOP)
def on_gateway_stop(event_data):
    print("Gateway stopping...")
    return HookResult.allow()

Storage Events

Event	Trigger	Use Case
`TOOL_RESULT_PERSIST`	Before result storage	Result modification

@registry.on(HookEvent.TOOL_RESULT_PERSIST)
def on_persist(event_data):
    print(f"Persisting: {event_data.tool_name}")
    # Modify or filter result before storage
    return HookResult.allow()

Kanban Events

Event	Input Type	Use Case
`KANBAN_TASK_CREATED`	`KanbanHookInput`	Task added to board
`KANBAN_TASK_CLAIMED`	`KanbanHookInput`	Task assigned
`KANBAN_TASK_MOVED`	`KanbanHookInput`	Task status changed
`KANBAN_TASK_DONE`	`KanbanHookInput`	Task completed
`KANBAN_TASK_BLOCKED`	`KanbanHookInput`	Task blocked
`KANBAN_TASK_FAILED`	`KanbanHookInput`	Task failed

@registry.on(HookEvent.KANBAN_TASK_MOVED)
def track_progress(event_data):
    print(f"Task {event_data.task_id}: {event_data.from_status} → {event_data.to_status}")
    return HookResult.allow()

@registry.on(HookEvent.KANBAN_TASK_BLOCKED)
def handle_blocked(event_data):
    print(f"🔒 Task blocked: {event_data.task_id}")
    # Send alert to Slack, email, etc.
    return HookResult.allow()

Complete Event Reference

Event	Category	Description
`BEFORE_TOOL`	Tool	Before tool execution
`AFTER_TOOL`	Tool	After tool execution
`BEFORE_AGENT`	Agent	Before agent runs
`AFTER_AGENT`	Agent	After agent completes
`BEFORE_LLM`	LLM	Before LLM API call
`AFTER_LLM`	LLM	After LLM response
`SESSION_START`	Session	Session starts
`SESSION_END`	Session	Session ends
`ON_ERROR`	Error	Error occurs
`ON_RETRY`	Error	Before retry
`USER_PROMPT_SUBMIT`	User	User submits prompt
`NOTIFICATION`	User	Notification sent
`SUBAGENT_STOP`	Subagent	Subagent completes
`SETUP`	System	Initialization
`BEFORE_COMPACTION`	Context	Before compaction
`AFTER_COMPACTION`	Context	After compaction
`MESSAGE_RECEIVED`	Message	Message received
`MESSAGE_SENDING`	Message	Before message sent
`MESSAGE_SENT`	Message	After message sent
`GATEWAY_START`	Gateway	Gateway starts
`GATEWAY_STOP`	Gateway	Gateway stops
`TOOL_RESULT_PERSIST`	Storage	Before result storage
`KANBAN_TASK_CREATED`	Kanban	Task added to board
`KANBAN_TASK_CLAIMED`	Kanban	Task assigned
`KANBAN_TASK_MOVED`	Kanban	Task status changed
`KANBAN_TASK_DONE`	Kanban	Task completed
`KANBAN_TASK_BLOCKED`	Kanban	Task blocked
`KANBAN_TASK_FAILED`	Kanban	Task failed

Best Practices

Keep hooks lightweight

Hooks run synchronously. Avoid heavy operations that could slow down agent execution.

Use matchers for filtering

Use pattern matchers to only run hooks for specific tools or operations.

Return early

Return HookResult.allow() quickly for non-matching cases to minimize overhead.

Handle errors gracefully

Wrap hook logic in try/except to prevent breaking agent execution.

Hooks

Hook system overview

Kanban Tasks

Kanban hook events and lifecycle

Plugins

Plugin system with hooks

​Quick Start

​Core Events

​Tool Events

​Agent Events

​LLM Events

​Session Events

​Error Events

​OnRetryInput Fields

​Agent-Level Error Callbacks

​Hook Events vs Agent Callbacks

​Extended Events

​User Interaction Events

​Subagent Events

​System Events

​Message Events

​Gateway Events

​Storage Events

​Kanban Events

​Complete Event Reference

​Best Practices

​Related