Hooks

Hooks provide a strongly-typed lifecycle event system. They let you observe, log, modify, or intercept any stage of the agent loop without touching the core logic.

Default model is Amazon Bedrock

Agent() with no model= uses Amazon Bedrock (Claude Sonnet 4.6 in us-west-2). Configure AWS credentials (aws configure or AWS_* env vars), or pass an explicit model — see Installation and Model Providers.

Registering a hook

Inline callback (type-inferred)

The event type is inferred from the function's type hint:

from elsai import Agent
from elsai.agent import AgentConfig
from elsai.hooks import BeforeModelCallEvent, AfterModelCallEvent

def log_before(event: BeforeModelCallEvent) -> None:
    print(f"[→] Calling model for agent: {event.agent.name}")

def log_after(event: AfterModelCallEvent) -> None:
    if event.exception:
        print(f"[←] Model error: {event.exception}")
    elif event.stop_response:
        print(f"[←] Model responded ({event.stop_response.stop_reason})")

agent = Agent(config=AgentConfig(hooks=[log_before, log_after]))

Add hook after construction

agent = Agent()
agent.add_hook(log_before)
agent.add_hook(log_after)

Handle multiple event types

from elsai.hooks import BeforeToolCallEvent, AfterToolCallEvent

def log_tools(event: BeforeToolCallEvent | AfterToolCallEvent) -> None:
    print(f"Tool event: {type(event).__name__}")

agent.add_hook(log_tools)

Hook provider class

Implement HookProvider to bundle related hooks:

from elsai.hooks import HookProvider, HookRegistry, BeforeModelCallEvent, AfterInvocationEvent

class AuditLogger(HookProvider):
    def register_hooks(self, registry: HookRegistry) -> None:
        registry.add_callback(BeforeModelCallEvent, self.on_before)
        registry.add_callback(AfterInvocationEvent, self.on_after)

    def on_before(self, event: BeforeModelCallEvent) -> None:
        print(f"Calling {event.agent.model.__class__.__name__}")

    def on_after(self, event: AfterInvocationEvent) -> None:
        if event.result:
            tokens = event.result.metrics.accumulated_usage.get("totalTokens", 0)
            print(f"Used {tokens} tokens")

agent = Agent(config=AgentConfig(hooks=[AuditLogger()]))

Available events

Agent lifecycle

Event	When
AgentInitializedEvent	Agent constructor completes
UserInputReceived	Just before processing a new user message
BeforeInvocationEvent	Before the event loop starts for this turn. TypeSafe Classifier can route here.
AfterInvocationEvent	After the event loop ends for this turn
BeforeResponseDispatch	Before the final result is returned
AgentErrorEvent	When an unhandled exception occurs

Model calls

Event	When
BeforeModelCallEvent	Before sending messages to the LLM
AfterModelCallEvent	After the model finishes responding

When streaming is enabled, token deltas are emitted between these hooks. Consume them via Agent.stream_async — see Streaming.

Tool calls

Event	When
BeforeToolCallEvent	Before a tool is executed. Permissions cancel or interrupt here. TypeSafe Classifier can gate side-effect tools here.
AfterToolCallEvent	After a tool returns

Memory

Event	When
BeforeMemoryWrite	Before a message is appended to history
MessageAddedEvent	After a message is appended to history

Multi-agent

Event	When
BeforeNodeCallEvent	Before a node (agent) starts in a graph/swarm
AfterNodeCallEvent	After a node finishes
MultiAgentInitializedEvent	When a Graph or Swarm is constructed

Writable event properties

Some events let you modify the agent's behaviour mid-flight:

from elsai.hooks import BeforeToolCallEvent

def tool_guard(event: BeforeToolCallEvent) -> None:
    # Swap the selected tool for a safer alternative
    if event.selected_tool.name == "dangerous_tool":
        event.selected_tool = safe_alternative_tool

agent.add_hook(tool_guard)

BeforeInvocationEvent

Property	Writable	Description
messages	✅	Override the messages for this turn

AfterInvocationEvent

Property	Writable	Description
resume	✅	Set to trigger another agent turn

BeforeToolCallEvent

Property	Writable	Description
selected_tool	✅	Swap the tool that will be called

BeforeResponseDispatch

Property	Writable	Description
result	✅	Override the final AgentResult

Async hooks

Hooks can be async:

import asyncio
from elsai.hooks import AfterToolCallEvent

async def async_logger(event: AfterToolCallEvent) -> None:
    await asyncio.sleep(0)  # yield to event loop
    name = event.selected_tool.name if event.selected_tool else "unknown"
    print(f"Tool {name} returned: {event.result}")

agent.add_hook(async_logger)

Paired events

Every Before event has a corresponding After event, even when an exception occurs — this guarantees proper cleanup:

from elsai.hooks import BeforeModelCallEvent, AfterModelCallEvent

class Timer(HookProvider):
    def __init__(self):
        self._start = {}

    def register_hooks(self, registry):
        registry.add_callback(BeforeModelCallEvent, self.start)
        registry.add_callback(AfterModelCallEvent, self.end)

    def start(self, event: BeforeModelCallEvent):
        import time
        self._start[id(event.agent)] = time.time()

    def end(self, event: AfterModelCallEvent):
        import time
        elapsed = time.time() - self._start.pop(id(event.agent), time.time())
        print(f"Model call took {elapsed:.2f}s")