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.

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

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.

Tool Definition Events

Event	Trigger	Input Type	Use Case
`BEFORE_TOOL_DEFINITIONS`	After advertised tool list is assembled, before sent to LLM	`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	`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

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

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

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
`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
`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	Inbound gate — drop (deny) or redact/rewrite before agent dispatch (see Inbound Message Gate)
`MESSAGE_SENDING`	Message	Outbound gate — cancel or rewrite before sending
`MESSAGE_SENT`	Message	After message sent
`MESSAGE_UNDELIVERED`	Message	Reply permanently undeliverable — operator routing (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

​Session Events

​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

Session Events

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