Hooks¶

Hooks are insertion points opened onto the execution loop. The argument object is mutable in place and execution continues afterwards, so a hook can both observe and intervene.

from xun import Agent, Hooks

hooks = Hooks()
hooks.after_initialize.add(lambda a: print("initialized:", a.agent.name))
hooks.before_tool_call.add(lambda a: print([t.function.name for t in a.tool_calls]))

agent = Agent(name="HookAgent", hooks=hooks).initialize()

Containers and registries¶

Every hook is a HookRegistry:

Method	Behavior
`add(fn)`	Persistent — called on every trigger
`add_once(fn)`	Fires once and is removed automatically after triggering
`invoke(args)`	Calls each callback in registration order, then drops the non-persistent ones

flowchart LR
    A["add(fn)"] --> L["Callback list<br/>persistent=True"]
    B["add_once(fn)"] --> L2["persistent=False"]
    L --> I["invoke(args)"]
    L2 --> I
    I --> K["Calls all of them, then keeps only the persistent ones"]

Hook exceptions are swallowed

Every callback is wrapped by except_safe: a raised exception is turned into a Result and discarded, and execution continues. To make errors visible from inside a hook, call agent.error(...) yourself or raise CancelledError / KeyboardInterrupt (both propagate normally). Also, there is no removal API — persistent hooks cannot be undone; use add_once for one-off needs, or build your own Hooks container.

The seventeen hooks¶

Hook	Argument class	Fields	Timing
`run_start`	`RunArgs`	`agent`	On entering the outermost `run_scope()` (nested runs do not trigger it)
`run_end`	`RunArgs`	`agent`	When the matching scope exits — on success, error or cancellation alike
`before_execution`	`BeforeExecutionArgs`	`agent`, `schema`, `max_iterations`, `context_value`, `takeover_result`	Before a round of execution starts
`before_command`	`CommandArgs`	`agent`, `command`, `arguments`	After a command parses successfully (unknown commands do not trigger it)
`after_command`	`CommandArgs`	Same as above	After the command runs — success or handled error both count
`before_execution_step`	`BeforeExecutionStepArgs`	`agent`	Before each model call; automatic compaction hangs here
`after_execution_step`	`AfterExecutionStepArgs`	`agent`	After tool results are written to the conversation, before the next model call
`before_tool_call`	`BeforeToolCallArgs`	`agent`, `tool_calls`	Before a batch of tool calls executes; the list is mutable
`after_tool_call`	`AfterToolCallArgs`	`agent`, `tool_results`	Afterward; the `(tool_id, Result)` list is mutable
`after_initialize`	`AfterInitializeArgs`	`agent`	At the end of `initialize()`, when the configuration is complete
`before_finalize`	`before_finalize`	`agent`	At the start of `finalize()`, a good place to release resources
`model_text_delta`	`TextDelta`	`agent`, `model_call_id`, `content`	Before a content delta is committed; `content` is mutable
`model_reasoning_delta`	`TextDelta`	Same as above	Reasoning delta
`completion_token_update`	`TokenMetrics`	`model_call_id`, `completion_tokens`, `prompt_tokens`, `total_tokens`, `prompt_tokens_cached`	At the end of each model call; fields are read-only, mainly for metrics
`before_display_info`	`AgentMessageArgs`	`agent`, `message`	Before `agent.info()` emits its event; `message` is mutable
`before_display_warning`	`AgentMessageArgs`	Same as above	`agent.warning()`
`before_display_error`	`AgentMessageArgs`	Same as above	`agent.error()`

Intervening in execution¶

Changing arguments¶

before_execution receives the execution arguments themselves, so mutate them in place:

from xun import HookArgs

def limit(args: HookArgs.BeforeExecutionArgs) -> None:
    args.max_iterations = 20
    args.context_value = {"budget": 100}

agent.hooks.before_execution.add(limit)

Taking over a round of execution¶

def stub(args: HookArgs.BeforeExecutionArgs) -> None:
    args.takeover_result = "Cache hit; no model call needed for this round"

agent.hooks.before_execution.add(stub)

Once takeover_result is set: the model is not called, no other hooks or events inside the execution fire, and nothing is written to the conversation history; if the round carries a schema, the takeover content is validated against that schema all the same.

Vetoing a round of execution¶

Exceptions raised in a hook are swallowed, so to genuinely stop execution raise a cancellation:

from xun import CancelledError, HookArgs

def budget_guard(args: HookArgs.BeforeExecutionArgs) -> None:
    if over_budget():
        raise CancelledError("Budget exhausted")

agent.hooks.before_execution.add(budget_guard)

Filtering output¶

agent.hooks.before_display_info.add(
    lambda a: setattr(a, "message", a.message.replace("TODO", "to-do"))
)

Fire once and leave¶

def once_after_write(args: HookArgs.AfterToolCallArgs) -> None:
    agent.display_event(HTMLInfoEvent(html="<em>Written, remember to review it</em>"))

agent.hooks.after_tool_call.add_once(once_after_write)

Components that ship their own hooks¶

AutoCompactor attaches itself to before_execution_step during initialize() — replace agent.compactor to change the strategy.
The sub-agent task pool registers cleanup on before_finalize, so background tasks are reclaimed when the parent agent finalizes.
Display-layer extensions (for example a logging extension that only applies to the web display) usually hang on before_display_*, see the Plugin Development Guide.

A custom Hooks container¶

Hooks is a dataclass: you can build your own container holding only the hooks you need and pass it as Agent(hooks=…); if you pass none, the default Hooks() is used. Agent.inherit does not inherit hooks — a derived agent always starts from an empty container, which keeps a parent's hooks from being implicitly amplified across the whole subtree.