Hook Events - PraisonAI

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.

Hooks that rewrite the payload via modified_input (before_llm, before_tool_definitions, before_tool) must be registered with sequential=True. See Sequential vs Parallel 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

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 — register with `sequential=True` when the hook needs to rewrite arguments via `modified_input`	`BeforeToolInput`	Security checks, logging
`AFTER_TOOL`	After tool execution	`AfterToolInput`	Redact/rewrite `tool_output`, block the 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.tool_output}")
    return HookResult.allow()

# Rewrite the tool output in place (propagates to the model)
@registry.on(HookEvent.AFTER_TOOL)
def redact(event_data):
    event_data.tool_output = str(event_data.tool_output).replace(
        "sk-SECRET123", "[REDACTED]"
    )
    return HookResult.allow()

# Block the tool output entirely
@registry.on(HookEvent.AFTER_TOOL)
def block_pii(event_data):
    if "ssn=" in str(event_data.tool_output):
        return HookResult.block("PII detected — result suppressed")
    return HookResult.allow()

BEFORE_TOOL and AFTER_TOOL fire on both sync (chat) and async (achat) paths. A blocking BEFORE_TOOL hook returning HookResult.block(reason=...) prevents tool execution on either path — use it for security gates without worrying which entrypoint the caller uses.

AFTER_TOOL can rewrite or block the tool result (PraisonAI PRs #3968 / #3969). Mutate event_data.tool_output in place to rewrite/redact the value the model sees, or return HookResult.block(reason) to suppress it. Applies on both sync (chat()) and async (achat()) paths. Prefer this seam over rewriting inside the tool body when the goal is a cross-cutting scrub or policy gate. See Redact / Block Tool Output.

BEFORE_TOOL fails closed (PR #3855). A BEFORE_TOOL hook that raises an exception or times out now returns HookResult(decision="deny", reason=...) — the tool does not run. A security gate can no longer silently fail open because of a bug in the hook body.

AFTER_TOOL result may be an error payload. If the tool raised, event_data.tool_output is {"error": "<message>"}; if it returned a non-JSON-serializable value (datetime, set, bytes, a custom class), it is {"result": "<str(value)>"}. The run continues either way (PR #3855), so AFTER_TOOL handlers should tolerate both shapes.

AFTER_TOOL returns can rewrite or block (PR #3969). Returning a value from an AFTER_TOOL hook replaces event_data.tool_output before the model sees it — use it to redact secrets or PII from a tool result. Raising GuardrailBlocked prevents the result from reaching the model, mirroring BEFORE_TOOL. Both apply on the sync and async tool-execution paths. Before PR #3969 the return value was silently discarded, so the hook was observe-only. See the Redact Secrets from Tool Output pattern.

Tool Definition Events

Event	Trigger	Input Type	Use Case
`BEFORE_TOOL_DEFINITIONS`	After advertised tool list is assembled, before sent to LLM — register with `sequential=True` when the hook needs to rewrite the list via `modified_input`	`BeforeToolDefinitionsInput`	Redact 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

Field	Type	Default	Description
`tool_definitions`	`List[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.
`model`	`str`	`""`	Model id the call will be sent to (e.g. `"gpt-4o"`). Use for per-model filtering.
`session_id`, `cwd`, `event_name`, `timestamp`, `agent_name`	—	—	Standard `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

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 — register with `sequential=True` when the hook needs to rewrite `messages` via `modified_input`	`BeforeLLMInput`	Request modification
`AFTER_LLM`	After LLM response	`AfterLLMInput`	Response validation
`MODEL_FALLBACK`	Primary model unavailable mid-turn — runtime switched to the next `fallback_models` entry	`ModelFallbackInput`	Alerts, metrics, per-user notice on silent quality/cost degradation

@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()

Sync/async parity — since PraisonAI PR #3908. BEFORE_LLM and AFTER_LLM fire on both sync (chat / start) and async (achat / astart) paths. A blocking BEFORE_LLM hook returning HookResult(decision="deny", reason=...) short-circuits the LLM call on either path and the agent returns "[LLM request blocked by hook: <reason>]". Earlier releases silently skipped these hooks on achat(), so features registered on BEFORE_LLM (e.g. enable_pii_redaction()) had no effect on async workflows.

MODEL_FALLBACK — silent model switch became observable

Fires the moment the runtime swaps the primary model for the next entry in fallback_models after a retryable failure, so an otherwise silent quality/cost degradation becomes an observable state transition.

Observe-only. Notification only — the turn already continued on to_model. Return HookResult.allow(); do not attempt to redirect the recovery here. Provider internals are redacted; only the failure class (reason_category) reaches your hook. Zero overhead when unsubscribed, and errors inside the hook never break the fallback path.

Register on the agent-scoped registry passed via Agent(hooks=...) (as shown below) so the hook fires on both sync and async runs — see Model Fallback → Observing the Switch. Hooks registered only on the global default registry may be skipped on the async path.

`ModelFallbackInput` Fields

Field	Type	Default	Description
`from_model`	`str`	`""`	Model id the turn was using when the failure hit (e.g. `"gpt-4o"`).
`to_model`	`str`	`""`	Next entry in the `fallback_models` chain that the turn continues on.
`reason_category`	`str`	`""`	Failure classification from the LLM error classifier (e.g. `"rate_limit"`, `"provider_down"`, `"timeout"`). Provider internals stay redacted.
`fallback_index`	`int`	`0`	0-based index into `fallback_models` of the entry now in use.
`session_id`, `cwd`, `event_name`, `timestamp`, `agent_name`	—	—	Standard `HookInput` fields.

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

registry = HookRegistry()

@registry.on(HookEvent.MODEL_FALLBACK)
def on_fallback(event_data):
    print(f"[fallback] {event_data.from_model} → {event_data.to_model}")
    print(f"           reason={event_data.reason_category} index={event_data.fallback_index}")
    return HookResult.allow()

agent = Agent(
    name="assistant",
    instructions="Be helpful.",
    llm=LLMConfig(model="gpt-4o", fallback_models=["claude-3-5-sonnet", "gpt-4o-mini"]),
    hooks=registry,
)
agent.start("Answer even during a provider outage")

The paired MODEL_FALLBACK stream event carries the same fields for live UIs. See Model Fallback → Observing the Switch for the primary user-facing page.

Session Events

Event	Trigger	Input Type	Use Case
`SESSION_START`	When session starts	`SessionStartInput`	Session initialization
`SESSION_END`	When session ends	`SessionEndInput`	Session cleanup
`SESSION_PERSIST_FAILED`	A durable session write failed (disk-full / SQLite corruption / permission / `OSError`) or a stored session file was found corrupt at load time (malformed JSON / invalid UTF-8)	`SessionPersistFailedInput`	Observe-only — alert/metric on silent data-loss; a failed write is salvaged to a spill file and re-ingested on next load, a corrupt read is quarantined to `<file>.json.corrupt-<epoch_ms>`

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()

`SESSION_PERSIST_FAILED` — a durable write failed

Fires when the session store cannot persist a turn (add_message / set_chat_history hits disk-full, SQLite/FTS corruption, or a permission / OSError). Before this hook, a failed durable write silently collapsed to False and the already-produced turn lived only in memory — lost on the next shutdown. Now the store salvages the turn to a spill file and fires this event so the failure is observable.

Observe-only. This is an observability signal, not a policy gate — it never re-runs the write and never raises on the caller’s hot path. Return HookResult.allow() (or nothing). It mirrors the outbound-side MESSAGE_UNDELIVERED posture. Skipped entirely when no such hook is registered (zero overhead). See Write-failure salvage for the spill + re-ingest flow.

On a corrupt read the error field starts with corrupt session file:, and spilled carries the quarantine path (<file>.json.corrupt-<epoch_ms>) instead of the write-failure spill path. role/content are empty because there is no in-flight turn. See Corruption-quarantine on load for the full flow.

`SessionPersistFailedInput` Fields

Field	Type	Default	Description
`role`	`str`	`""`	Role of the message that failed to persist (`user` / `assistant`)
`content`	`str`	`""`	Message content (truncated to 500 chars in `to_dict`)
`error`	`str`	`""`	Stringified error from the failing write (e.g. `"atomic write failed"` or the OSError text)
`spilled`	`bool`	`False`	Whether the last-resort fallback write itself succeeded
`spill_path`	`Optional[str]`	`None`	The file the turn was salvaged to (when `spilled=True`)
`session_id`, `cwd`, `event_name`, `timestamp`, `agent_name`	—	—	Standard `HookInput` fields

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

registry = HookRegistry()

@registry.on(HookEvent.SESSION_PERSIST_FAILED)
def on_persist_failed(data):
    # Alert / metric — do NOT retry the write here; the store has already spilled.
    print(f"[persist-fail] role={data.role} spilled={data.spilled} path={data.spill_path}")
    print(f"[persist-fail] error={data.error}")
    return HookResult.allow()

agent = Agent(name="assistant", instructions="Help the user.", hooks=registry)
agent.start("Hello")

Spill & recovery flow

A failed durable write salvages the turn to a spill file, fires this hook, then re-ingests the spill on the next session load. Spill files land under ~/.praisonai/state/session_spill/ (written 0600, atomic temp-file + os.replace).

Observe-only — never retry

The hook is a notification. Do not retry the write from it — the store has already spilled the turn and re-ingests it on the next load.

Collision-safe filenames

Spill filenames carry a random token, so consecutive same-millisecond/PID failures never overwrite each other — no lost turns.

Recovery respects retention

_reingest_spill runs _enforce_window before persisting, so recovered turns obey the retention policy like any ordinary write.

Malformed spills are skipped

A non-object root, non-list messages, or non-object message is skipped — a bad spill never blocks recovery.

Error Events

Event	Trigger	Input Type	Use Case
`ON_ERROR`	When error occurs	`OnErrorInput`	Error handling
`ON_RETRY`	Before retry attempt	`OnRetryInput`	Retry 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.

Retry is on by default, so ON_RETRY fires with no config needed on both the native OpenAI-client path and the LiteLLM path (default max_retries=3). It only stops firing if you disable retry with retry=False. See Agent Retry.

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 whenever a transient LLM error is retried, including the default RetryBackoffConfig() policy applied to every Agent — see Agent Retry). ON_RETRY now fires on default Agents too; pass retry=False to disable retries and stop these fires:

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

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):
    # event_data is the CompactionResult for this pass.
    print(f"Compacted to: {event_data.compacted_tokens} tokens")

    # 1.6.152+: summary is populated for summarising strategies.
    # Pre-1.6.152 the field existed but was always "".
    if event_data.summary:
        # e.g. mirror into your own long-term memory store here
        print(f"Summary: {event_data.summary[:120]}")
    return HookResult.allow()

Plugin persisters subscribe to AFTER_COMPACTION to read event_data.summary. See Compacted Session Resume for the built-in checkpoint that ships with PraisonAI, and Context Compaction for the full CompactionResult fields.

Message Events

Event	Trigger	Input Type	Use Case
`MESSAGE_RECEIVED`	Inbound message from a bot channel, before agent dispatch	`MessageReceivedInput`	Inbound gate — drop (deny), redact/rewrite content, authorise sender, rate-limit
`MESSAGE_SENDING`	Before message sent	`MessageSendingInput`	Outbound gate — cancel or rewrite outbound text
`MESSAGE_SENT`	After message sent	`MessageSentInput`	Confirmation
`MESSAGE_UNDELIVERED`	Reply permanently undeliverable (target confirmed dead / retries exhausted)	`MessageUndeliveredInput`	Operator routing — mirror to home channel, alert, or re-queue

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.

MESSAGE_RECEIVED and MESSAGE_SENDING payloads now include the enriched identity fields (platform, sender_id, channel_id, channel_type, message_id, session_id) when the underlying input provides them — the same fields that reach the Plugin _message_payload bridge.

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 gate — deny 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

Decision	Effect
`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.

`MESSAGE_UNDELIVERED` — close the loop on a permanent failure

Fires when the gateway’s DeliveryRouter classifies a send as permanently failed. Observability only — the hook does not gate anything.

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

registry = HookRegistry()

@registry.on(HookEvent.MESSAGE_UNDELIVERED)
def route_undelivered(event_data):
    # event_data is a MessageUndeliveredInput
    print(
        f"[UNDELIVERED] {event_data.platform}:{event_data.channel_id} "
        f"error={event_data.error} notice_delivered={event_data.notice_delivered}"
    )
    return HookResult.allow()

agent = Agent(name="broadcaster", instructions="Send updates.", hooks=registry)

See Undelivered Message Notice for the full opt-in flow and the gateway.notify_on_undelivered config.

Plugin bridge: inside a Plugin subclass, MESSAGE_SENT and MESSAGE_UNDELIVERED surface as the base methods message_sent(message) and message_undelivered(message) — see Plugins → Message-Lifecycle Plugins. The plugin bridge only wires these when your subclass overrides them.

Gateway Events

Event	Trigger	Input Type	Key Fields	Use Case
`GATEWAY_START`	`BotOS.start()`	`GatewayStartInput`	`platforms`, `bot_count`	Initialization
`GATEWAY_STOP`	`BotOS.stop()`	`GatewayStopInput`	`platforms`, `bot_count`, `reason`	Cleanup

@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).

Event	Trigger	Input Type	Key Fields	Use Case
`SCHEDULE_ADD`	Schedule job added	—	—	Audit schedule changes
`SCHEDULE_REMOVE`	Schedule job removed	—	—	Audit schedule changes
`SCHEDULE_TRIGGER`	Scheduled job runs	`ScheduleTriggerInput`	`job_name`, `job_id`, `message`	Observability, metrics
`JOB_COMPLETED`	Background job reaches terminal state (`COMPLETED` or `FAILED`)	`JobCompletedInput`	`job_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

Field	Type	Description
`job_info`	`JobInfo`	Terminal 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

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()

CLI Backend Events

Event	Trigger	Input Type	Use Case
`CLI_BACKEND_EXECUTE`	Agent delegates a turn to a CLI backend subprocess	`CliBackendExecuteInput`	Observability — confirm CLI delegation, log subprocess argv, capture errors

CLI_BACKEND_EXECUTE fires on both success and failure, so subprocess startup errors and timeouts stay traceable — the payload’s error field carries the exception message on the failure path.

This hook is observe-only — it cannot gate the CLI backend call. Return HookResult.allow() (or return nothing). For a policy gate on tool use, register on BEFORE_TOOL instead.

`CliBackendExecuteInput` Fields

Field	Type	Default	Description
`backend`	`str`	`""`	Human-readable backend id, e.g. `"claude"`, `"codex"`, `"grok"`, `"gemini"`. Comes from `config.command` when available, else the backend class name.
`command`	`Optional[List[Any]]`	`None`	The full subprocess argv the CLI backend spawned. Prompt/system values are redacted (`"<redacted>"`) at serialisation, so log sinks never see the user prompt.
`content`	`Optional[str]`	`None`	The backend’s textual response. Truncated to the first 500 characters at serialisation.
`error`	`Optional[str]`	`None`	Non-null on failure — carries the exception message from the CLI subprocess or timeout.
`transport`	`str`	`"subprocess"`	Always `"subprocess"` today. Reserved for future transports.
`praisonai_llm_http`	`bool`	`False`	Always `False` — the hook exists precisely because PraisonAI did not make an HTTP LLM call this turn. Useful for dashboards proving the LiteLLM path was skipped.
`session_id`, `cwd`, `event_name`, `timestamp`, `agent_name`	—	—	Standard `HookInput` fields.

Two safety mechanisms apply at to_dict() time (the payload that reaches log sinks): prompt/system values that follow -p, --prompt, -i, --input, -m, --message, --system are replaced with "<redacted>", and content is truncated to the first 500 characters. The live in-memory command field is left untouched — only the serialised payload is redacted.

Trace CLI delegation programmatically

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

registry = HookRegistry()

@registry.on(HookEvent.CLI_BACKEND_EXECUTE)
def trace_cli_backend(event_data):
    if event_data.error:
        print(f"[cli-backend] {event_data.backend} FAILED: {event_data.error}")
    else:
        argv = " ".join(str(a) for a in (event_data.command or []))
        print(f"[cli-backend] {event_data.backend} ok → {argv}")
    return HookResult.allow()

agent = Agent(
    name="assistant",
    instructions="Be helpful.",
    cli_backend="codex-cli",   # or claude-code, grok-cli, gemini-cli
    hooks=registry,
)

agent.start("Refactor utils.py")

The prompt is redacted at the payload boundary:

[cli-backend] codex ok → codex exec --skip-git-repo-check -C /workspace <redacted>

Enable the built-in tracer plugin

The cli_backend_tracer plugin from praisonai-plugins is the batteries-included consumer of this hook — it logs every delegation to the standard praisonai logger. Enable it with one env var and a plugin toggle:

pip install praisonai-plugins
export PRAISONAI_CLI_BACKEND_DEBUG=1
praisonai plugins enable cli_backend_tracer

Or use the standard Python knob:

export LOGLEVEL=DEBUG

With either set and the plugin enabled, every codex exec / grok -p / gemini -p / claude delegation appears in the log stream, with the prompt value already redacted.

What gets masked

redact_command masks the value that follows these flags, leaving the flag itself visible for verification:

Flag	Reason it’s masked
`-p`, `--prompt`	The user prompt
`-i`, `--input`	Alternate prompt form (Grok)
`-m`, `--message`	Chat message form
`--system`	System instruction

Non-list command values (already-serialised strings, None) pass through unchanged.

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 (includes auto-promotion: dispatcher fires `to_status='ready'` when all parents are terminal — see Dependency Auto-Promotion).
`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}")
    return HookResult.allow()

When a task with workspace_kind="worktree" fails to merge cleanly, KANBAN_TASK_BLOCKED fires with conflicted_files: list[str] in the payload so alerting agents can surface the exact merge conflict. See Kanban → Per-Task Worktree Isolation.

@registry.on(HookEvent.KANBAN_TASK_BLOCKED)
def alert_on_conflict(event_data):
    files = event_data.get("conflicted_files")
    if files:
        print(f"Merge conflict on {event_data['task_id']}: {files}")
    return HookResult.allow()

Preserved worktree comments (not a hook — a task comment). A clean-merge task may also carry a worktree_preserved at <path>: <reason> comment when the dispatcher’s lossless guard refuses to tear down a worktree with uncommitted changes or unmerged commits (or when git worktree remove itself fails). No hook fires — poll for the comment via kanban_show(task_id) if you need to alert on it. See Kanban → Lossless-only worktree teardown.

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

Event	Category	Description
`BEFORE_TOOL`	Tool	Before tool execution
`AFTER_TOOL`	Tool	After tool execution — mutate `tool_output` to rewrite/redact, or `HookResult.block(reason)` to suppress the result before it reaches the model
`BEFORE_TOOL_DEFINITIONS`	Tool	Before tool definitions are sent to LLM
`BEFORE_AGENT`	Agent	Before agent runs
`AFTER_AGENT`	Agent	After agent completes
`BEFORE_LLM`	LLM	Before LLM API call — fires sync + async
`AFTER_LLM`	LLM	After LLM response — fires sync + async
`MODEL_FALLBACK`	LLM	Primary model unavailable mid-turn — runtime switched to the next `fallback_models` entry (observe-only; provider internals redacted). See Model Fallback → Observing the Switch.
`SESSION_START`	Session	Session starts
`SESSION_END`	Session	Session ends
`SESSION_PERSIST_FAILED`	Session	A durable session write failed or a stored session file was found corrupt at load time — observe-only; a failed write is spilled and re-ingested (see Write-failure salvage), a corrupt read is quarantined (see Corruption-quarantine on load)
`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	Inbound gate — drop (deny) or redact/rewrite before agent dispatch; payload carries enriched identity (`platform`, `sender_id`, `channel_id`, `channel_type`, `message_id`, `session_id`) (see Inbound Message Gate)
`MESSAGE_SENDING`	Message	Outbound gate — cancel or rewrite before sending; payload carries the enriched identity fields
`MESSAGE_SENT`	Message	After successful delivery; enriched identity payload — also the plugin `message_sent(message)` bridge
`MESSAGE_UNDELIVERED`	Message	Reply permanently undeliverable — operator routing; enriched identity payload plus `error`/`notice_delivered`, also the plugin `message_undelivered(message)` bridge (see Undelivered Message Notice)
`GATEWAY_START`	Gateway	Gateway starts
`GATEWAY_STOP`	Gateway	Gateway stops
`SCHEDULE_ADD`	Schedule	Schedule job added
`SCHEDULE_REMOVE`	Schedule	Schedule job removed
`SCHEDULE_TRIGGER`	Schedule	Scheduled job runs
`JOB_COMPLETED`	Background	Background job reached terminal state (COMPLETED or FAILED)
`TOOL_RESULT_PERSIST`	Storage	Before result storage
`CLI_BACKEND_EXECUTE`	CLI Backend	Fires after an Agent delegates a turn to a CLI backend subprocess (success and failure). Observe-only.
`KANBAN_TASK_CREATED`	Kanban	Task added to board
`KANBAN_TASK_CLAIMED`	Kanban	Task assigned
`KANBAN_TASK_MOVED`	Kanban	Task status changed (includes auto-promotion: dispatcher fires this with `to_status='ready'` for each dependent task whose parents are all terminal)
`KANBAN_TASK_DONE`	Kanban	Task completed
`KANBAN_TASK_BLOCKED`	Kanban	Task blocked
`KANBAN_TASK_FAILED`	Kanban	Task 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. CLI_BACKEND_EXECUTE is also observe-only — it is orthogonal to the message gates and cannot stop the CLI delegation; it only reports that a subprocess turn happened.

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.

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

​How It Works

​Lifecycle Events

​Quick Start

​Core Events

​Tool Events

​Tool Definition Events

​BeforeToolDefinitionsInput Fields

​Agent Events

​LLM Events

​MODEL_FALLBACK — silent model switch became observable

ModelFallbackInput Fields

​Session Events

​SESSION_PERSIST_FAILED — a durable write failed

SessionPersistFailedInput Fields

Spill & recovery flow

​Error Events

​Async vs Sync

​OnRetryInput Fields

​Agent-Level Error Callbacks

​Hook Events vs Agent Callbacks

​Extended Events

​User Interaction Events

​Subagent Events

​System Events

​Message Events

​Drop / block an inbound message

​Redact PII before the agent sees it

​Authorise sender against an allowlist

​How the gate applies decisions

​MESSAGE_UNDELIVERED — close the loop on a permanent failure

​Gateway Events

​Schedule & background events

​Background Job Events

​JobCompletedInput Fields

​Storage Events

​CLI Backend Events

​CliBackendExecuteInput Fields

​Trace CLI delegation programmatically

​Enable the built-in tracer plugin

​What gets masked

​Kanban Events

​Dependency Auto-Promotion Events

​Complete Event Reference

​Bot Runtime Lifecycle

​Best Practices

​Related

Inbound Message Gate

Hooks

Kanban Tasks

Plugins

How It Works

Lifecycle Events

Quick Start

Core Events

Tool Events

Tool Definition Events

BeforeToolDefinitionsInput Fields

Agent Events

LLM Events

MODEL_FALLBACK — silent model switch became observable

`ModelFallbackInput` Fields

Session Events

`SESSION_PERSIST_FAILED` — a durable write failed

`SessionPersistFailedInput` Fields

Error Events

Async vs Sync

OnRetryInput Fields

Agent-Level Error Callbacks

Hook Events vs Agent Callbacks

Extended Events

User Interaction Events

Subagent Events

System Events

Message Events

Drop / block an inbound message

Redact PII before the agent sees it

Authorise sender against an allowlist

How the gate applies decisions

`MESSAGE_UNDELIVERED` — close the loop on a permanent failure

Gateway Events

Schedule & background events

Background Job Events

`JobCompletedInput` Fields

Storage Events

CLI Backend Events

`CliBackendExecuteInput` Fields

Trace CLI delegation programmatically

Enable the built-in tracer plugin

What gets masked

Kanban Events

Dependency Auto-Promotion Events

Complete Event Reference

Bot Runtime Lifecycle

Best Practices

Related