Skip to main content
Hook events are triggered at specific points in the agent lifecycle, allowing you to intercept, modify, or block operations.
This page covers in-process lifecycle hooks (SETUP, SESSION_START, BEFORE_AGENT, etc.) that fire inside a running agent. For HTTP inbound triggers that start agent runs from external services via POST /hooks/<path>, see Gateway Inbound Hooks.
from praisonaiagents import Agent
from praisonaiagents.hooks import add_hook

@add_hook("before_tool")
def log_tool(event_data):
    print(f"Tool: {event_data.tool_name}")

agent = Agent(name="MyAgent", instructions="Be helpful.")
agent.start("Run a tool and show lifecycle hooks")
The user sends a prompt; hooks fire at each lifecycle event as the agent runs.

How It Works

Lifecycle Events

Quick Start

1

Import Hook Components

from praisonaiagents.hooks import HookRegistry, HookEvent, HookResult
2

Register a Hook

registry = HookRegistry()

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

Use with Agent

from praisonaiagents import Agent

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

Core Events

Tool Events

EventTriggerInput TypeUse Case
BEFORE_TOOLBefore tool executionBeforeToolInputSecurity checks, logging
AFTER_TOOLAfter tool executionAfterToolInputResult 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()

Tool Definition Events

EventTriggerInput TypeUse Case
BEFORE_TOOL_DEFINITIONSAfter advertised tool list is assembled, before sent to LLMBeforeToolDefinitionsInputRedact tools per request, append usage notes, constrain schema per model
BEFORE_TOOL_DEFINITIONS lets you shape the tool list the LLM actually sees — without changing the agent’s permanent tool registration. Mutate tool_definitions in place; the runtime only adopts in-place mutations.
Mutate in place using event_data.tool_definitions[:] = [...]. Reassigning the local name (event_data.tool_definitions = [...]) is silently ignored by the runtime.

BeforeToolDefinitionsInput Fields

FieldTypeDefaultDescription
tool_definitionsList[Dict[str, Any]][]The fully-assembled OpenAI-style tool definition list about to be sent to the LLM. Mutate in place to filter or rewrite.
modelstr""Model id the call will be sent to (e.g. "gpt-4o"). Use for per-model filtering.
session_id, cwd, event_name, timestamp, agent_nameStandard HookInput fields.
from praisonaiagents import Agent
from praisonaiagents.hooks import HookRegistry, HookEvent, HookResult

registry = HookRegistry()

@registry.on(HookEvent.BEFORE_TOOL_DEFINITIONS)
def sandbox_tools(event_data):
    # Drop dangerous tools for this request
    event_data.tool_definitions[:] = [
        t for t in event_data.tool_definitions
        if t["function"]["name"] != "delete_file"
    ]
    # Annotate remaining tools
    for t in event_data.tool_definitions:
        if t["function"]["name"] == "read_file":
            t["function"]["description"] += " (sandboxed: /workspace only)"
    return HookResult.allow()

agent = Agent(
    name="SafeAgent",
    instructions="Help the user with file operations.",
    hooks=registry,
)

agent.start("List my files")
Per-model filtering using the model field:
@registry.on(HookEvent.BEFORE_TOOL_DEFINITIONS)
def small_model_subset(event_data):
    if event_data.model.startswith("gpt-3.5"):
        event_data.tool_definitions[:] = event_data.tool_definitions[:5]
    return HookResult.allow()

Agent Events

EventTriggerInput TypeUse Case
BEFORE_AGENTBefore agent runsBeforeAgentInputSetup, initialization
AFTER_AGENTAfter agent completesAfterAgentInputCleanup, 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

EventTriggerInput TypeUse Case
BEFORE_LLMBefore LLM API callBeforeLLMInputRequest modification
AFTER_LLMAfter LLM responseAfterLLMInputResponse 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

EventTriggerInput TypeUse Case
SESSION_STARTWhen session startsSessionStartInputSession initialization
SESSION_ENDWhen session endsSessionEndInputSession cleanup
Plugin authors: subclass Plugin and override session_start / session_end — see Plugins → How the Bridge Works.
Bot runtime semantics: In the bot runtime (BotOS), SESSION_START fires exactly once per user session lifetime — on the first message, not on every message. SESSION_END fires when the user sends /new, when a policy auto-reset triggers, when stale sessions are reaped, or on reset_all. The reason field on SessionEndInput is one of clear, policy, stale, or clear_all.
@registry.on(HookEvent.SESSION_START)
def session_start(event_data):
    print(f"Session: {event_data.session_id}")
    print(f"Source: {event_data.source}")
    print(f"Platform: {event_data.session_name}")
    return HookResult.allow()

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

Error Events

EventTriggerInput TypeUse Case
ON_ERRORWhen error occursOnErrorInputError handling
ON_RETRYBefore retry attemptOnRetryInputRetry logic
Plugin authors: subclass Plugin and override on_error — see Plugins → How the Bridge Works.
@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()

Async vs Sync

ON_RETRY fires on both sync and async retry paths. If you register the handler as a regular def, it is dispatched in a thread executor on the async path; if you register an async def, it is awaited directly. Either form is safe on both paths.
Released in PraisonAI #2386 — earlier versions skipped this event on the async path.

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
LLM retry fields (populated when fired by Agent(retry=...) — see Agent Retry):
  • delay_seconds: Seconds the agent will sleep before the next attempt
  • attempt: Current attempt number (0-based)
  • operation: "llm_request" (sync) or "async_llm_request" (async)
  • error_message: String representation of the failing LLMError
  • max_retries: Configured RetryBackoffConfig.max_retries
  • retry_count: Same as attempt + 1 (1-based legacy alias)
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 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

ApproachScopeReturn ValueWhen Called
HookEvent.ON_ERRORGlobal, all agentsHookResultAny hook system error
agent.on_errorSingle agentNoneLLM 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

EventTriggerUse Case
USER_PROMPT_SUBMITUser submits promptInput validation, logging
NOTIFICATIONNotification sentAlert 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

EventTriggerUse Case
SUBAGENT_STOPSubagent completesResult 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

EventTriggerUse Case
SETUPInitialization/maintenanceConfig loading
BEFORE_COMPACTIONBefore context compactionPre-compaction hooks
AFTER_COMPACTIONAfter context compactionPost-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

EventTriggerInput TypeUse Case
MESSAGE_RECEIVEDInbound message from a bot channel, before agent dispatchMessageReceivedInputInbound gate — drop (deny), redact/rewrite content, authorise sender, rate-limit
MESSAGE_SENDINGBefore message sentMessageSendingInputOutbound gate — cancel or rewrite outbound text
MESSAGE_SENTAfter message sentMessageSentInputConfirmation
MESSAGE_RECEIVED is a first-class control point: hooks can drop an inbound message so the agent never runs, or rewrite the message content before the agent (or memory) sees it. This is symmetric with MESSAGE_SENDING on the outbound side. The hook now returns a decision that every platform adapter (Telegram, Slack, Discord, WhatsApp, Email, AgentMail) honours:
  • HookResult.deny(reason=...) → the message is dropped; agent dispatch is skipped entirely.
  • HookResult(modified_input={"content": "..."}) → the inbound message content is rewritten before dispatch.
  • None or HookResult.allow() → message passes through unchanged (default).
  • Hook errors are non-fatal — a raising hook logs the error and lets the message through.
MESSAGE_RECEIVED and MESSAGE_SENDING are the two message-lifecycle events that do gatedeny drops the message entirely, modified_input["content"] rewrites it. The gate is safe from both sync and async adapters (Telegram, Slack, Discord, WhatsApp, Email, AgentMail) — no async def required in your hook. See PraisonAI #2589.

Drop / block an inbound message

from praisonaiagents import Agent
from praisonaiagents.hooks import HookRegistry, HookEvent, HookResult

registry = HookRegistry()

@registry.on(HookEvent.MESSAGE_RECEIVED)
def block_bots(event_data):
    if event_data.sender_id.endswith("bot"):
        return HookResult.deny("no bots")
    return HookResult.allow()

agent = Agent(name="Concierge", instructions="Be helpful.", hooks=registry)

Redact PII before the agent sees it

import re
from praisonaiagents import Agent
from praisonaiagents.hooks import HookRegistry, HookEvent, HookResult

registry = HookRegistry()

@registry.on(HookEvent.MESSAGE_RECEIVED)
def redact_pii(event_data):
    cleaned = re.sub(r"\b\d{3}-\d{2}-\d{4}\b", "[SSN]", event_data.content)
    return HookResult(decision="allow", modified_input={"content": cleaned})

agent = Agent(name="SafeAgent", instructions="Be helpful.", hooks=registry)

Authorise sender against an allowlist

from praisonaiagents import Agent
from praisonaiagents.hooks import HookRegistry, HookEvent, HookResult

registry = HookRegistry()
ALLOWED = {"U123", "U456"}

@registry.on(HookEvent.MESSAGE_RECEIVED)
def only_allowed_users(event_data):
    if event_data.sender_id not in ALLOWED:
        return HookResult.deny("unauthorised")
    return HookResult.allow()

agent = Agent(name="PrivateAgent", instructions="Be helpful.", hooks=registry)

How the gate applies decisions

DecisionEffect
HookResult.deny("reason")Message is dropped; the agent is never invoked; the adapter returns immediately
HookResult(decision="allow", modified_input={"content": "..."})Content is rewritten in place; agent, memory, and command parser all see the new text
HookResult.allow()Message passes through unchanged
When multiple MESSAGE_RECEIVED hooks run, the last matching modification wins. Hook errors are non-fatal — the message passes through unchanged.
@registry.on(HookEvent.MESSAGE_SENDING)
def on_message_sending(event_data):
    print(f"Sending: {event_data.content}")
    return HookResult.allow()
Since PraisonAI PR #2589, every messaging adapter (Telegram, Slack, Discord, WhatsApp, email, agentmail) honours these decisions. Hook exceptions are non-fatal — the message passes through. See Inbound Message Gate for patterns and failure semantics.

Gateway Events

EventTriggerInput TypeKey FieldsUse Case
GATEWAY_STARTBotOS.start()GatewayStartInputplatforms, bot_countInitialization
GATEWAY_STOPBotOS.stop()GatewayStopInputplatforms, bot_count, reasonCleanup
@registry.on(HookEvent.GATEWAY_START)
def on_gateway_start(event_data):
    print(f"Gateway starting on: {event_data.platforms}")
    print(f"Bot count: {event_data.bot_count}")
    return HookResult.allow()

@registry.on(HookEvent.GATEWAY_STOP)
def on_gateway_stop(event_data):
    print(f"Gateway stopping: {event_data.platforms}")
    print(f"Reason: {event_data.reason}")
    return HookResult.allow()

Schedule & background events

Schedule hooks fire when scheduled jobs are managed and triggered by BotOS. JOB_COMPLETED fires when a background subagent job launched via spawn_subagent(background=True) reaches a terminal state — after the internal on_complete callback runs, best-effort (a raising handler cannot crash the worker).
EventTriggerInput TypeKey FieldsUse Case
SCHEDULE_ADDSchedule job addedAudit schedule changes
SCHEDULE_REMOVESchedule job removedAudit schedule changes
SCHEDULE_TRIGGERScheduled job runsScheduleTriggerInputjob_name, job_id, messageObservability, metrics
JOB_COMPLETEDBackground job reaches terminal state (COMPLETED or FAILED)JobCompletedInputjob_info (job_id, status, result/error, duration, origin, deliver, platform, chat_id, thread_id)Observability, custom delivery routing, chat-back delivery (see Background Subagents)
@registry.on(HookEvent.SCHEDULE_TRIGGER)
def on_schedule_trigger(event_data):
    print(f"Job fired: {event_data.job_name}")
    print(f"Job ID: {event_data.job_id}")
    print(f"Message: {event_data.message}")
    return HookResult.allow()

@registry.on(HookEvent.JOB_COMPLETED)
def on_job_completed(event_data):
    print(f"Job finished: {event_data.job_id}{event_data.status}")
    if event_data.error:
        print(f"Error: {event_data.error}")
    return HookResult.allow()

@registry.on(HookEvent.SCHEDULE_ADD)
def on_schedule_add(event_data):
    print(f"Schedule added for agent: {event_data.agent_name}")
    return HookResult.allow()

Background Job Events

JOB_COMPLETED fires when a background job reaches a terminal state (COMPLETED or FAILED). It fires after the internal on_complete callback runs — a raising handler cannot crash the worker.
from praisonaiagents.hooks import HookEvent, register_hook

@register_hook(HookEvent.JOB_COMPLETED)
def log_job_result(event_data):
    job = event_data.job_info
    print(f"Job {job.job_id}{job.status.value}")
    if job.status.value == "completed":
        print(f"Result: {job.result}")
    else:
        print(f"Error: {job.error}")
    return HookResult.allow()

JobCompletedInput Fields

FieldTypeDescription
job_infoJobInfoTerminal job state. Contains job_id, status, result, error, duration, origin.
Typical uses:
  • Observability and metrics on background job durations and failure rates
  • Custom delivery routing when the built-in deliver= token is insufficient
  • External side effects on completion (webhooks, database writes)

Storage Events

EventTriggerUse Case
TOOL_RESULT_PERSISTBefore result storageResult 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

EventInput TypeUse Case
KANBAN_TASK_CREATEDKanbanHookInputTask added to board
KANBAN_TASK_CLAIMEDKanbanHookInputTask assigned
KANBAN_TASK_MOVEDKanbanHookInputTask status changed (includes auto-promotion: dispatcher fires to_status='ready' when all parents are terminal — see Dependency Auto-Promotion).
KANBAN_TASK_DONEKanbanHookInputTask completed
KANBAN_TASK_BLOCKEDKanbanHookInputTask blocked
KANBAN_TASK_FAILEDKanbanHookInputTask 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}")
    return HookResult.allow()

Dependency Auto-Promotion Events

When the dispatcher auto-promotes a child task, KANBAN_TASK_MOVED fires with this payload:
{
    "task_id": "t_child",
    "to_status": "ready",
    "task": {...},
}
Auto-promotion events do not include from_status. Check for its absence if you need to distinguish auto-promotions from manual moves.
@registry.on(HookEvent.KANBAN_TASK_MOVED)
def on_auto_promotion(event_data):
    if event_data.get("to_status") == "ready" and not event_data.get("from_status"):
        print(f"Task {event_data['task_id']} auto-promoted to ready")
    return HookResult.allow()

Complete Event Reference

EventCategoryDescription
BEFORE_TOOLToolBefore tool execution
AFTER_TOOLToolAfter tool execution
BEFORE_TOOL_DEFINITIONSToolBefore tool definitions are sent to LLM
BEFORE_AGENTAgentBefore agent runs
AFTER_AGENTAgentAfter agent completes
BEFORE_LLMLLMBefore LLM API call
AFTER_LLMLLMAfter LLM response
SESSION_STARTSessionSession starts
SESSION_ENDSessionSession ends
ON_ERRORErrorError occurs
ON_RETRYErrorBefore retry
USER_PROMPT_SUBMITUserUser submits prompt
NOTIFICATIONUserNotification sent
SUBAGENT_STOPSubagentSubagent completes
SETUPSystemInitialization
BEFORE_COMPACTIONContextBefore compaction
AFTER_COMPACTIONContextAfter compaction
MESSAGE_RECEIVEDMessageInbound gate — drop (deny) or redact/rewrite before agent dispatch (see Inbound Message Gate)
MESSAGE_SENDINGMessageOutbound gate — cancel or rewrite before sending
MESSAGE_SENTMessageAfter message sent
GATEWAY_STARTGatewayGateway starts
GATEWAY_STOPGatewayGateway stops
SCHEDULE_ADDScheduleSchedule job added
SCHEDULE_REMOVEScheduleSchedule job removed
SCHEDULE_TRIGGERScheduleScheduled job runs
JOB_COMPLETEDBackgroundBackground job reached terminal state (COMPLETED or FAILED)
TOOL_RESULT_PERSISTStorageBefore result storage
KANBAN_TASK_CREATEDKanbanTask added to board
KANBAN_TASK_CLAIMEDKanbanTask assigned
KANBAN_TASK_MOVEDKanbanTask status changed (includes auto-promotion: dispatcher fires this with to_status='ready' for each dependent task whose parents are all terminal)
KANBAN_TASK_DONEKanbanTask completed
KANBAN_TASK_BLOCKEDKanbanTask blocked
KANBAN_TASK_FAILEDKanbanTask failed

Bot Runtime Lifecycle

Gateway and session hooks are emitted by BotOS and BotSessionManager — no extra wiring needed when your agent is passed to a bot.
  • All emission is best-effort and a no-op when no hooks are registered (zero overhead)
  • BEFORE_AGENT / AFTER_AGENT are fired by agent.chat() itself — they are not re-fired at the gateway boundary to avoid double-dispatch
  • In async contexts (e.g. inside BotSessionManager.chat), emission is fire-and-forget; in sync contexts it is blocking
MESSAGE_RECEIVED and MESSAGE_SENDING are policy gates, not passive observers. deny drops the message; modified_input["content"] rewrites it. Gateway/session lifecycle events (GATEWAY_START, GATEWAY_STOP, SESSION_START, SESSION_END) remain best-effort observability points and do not gate startup or shutdown.

Best Practices

Hooks run synchronously. Avoid heavy operations that could slow down agent execution.
Use pattern matchers to only run hooks for specific tools or operations.
Return HookResult.allow() quickly for non-matching cases to minimize overhead.
Wrap hook logic in try/except to prevent breaking agent execution.

Inbound Message Gate

Drop or redact incoming messages before the agent sees them

Hooks

Hook system overview

Kanban Tasks

Kanban hook events and lifecycle

Plugins

Plugin system with hooks