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.

Registering a hook

Inline callback (type-inferred)

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

from elsai import Agent
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:
    print(f"[←] Model responded in {event.metrics.latency_ms:.0f}ms")

agent = Agent(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, AfterModelCallEvent

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

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

    def on_after(self, event: AfterModelCallEvent) -> None:
        print(f"Used {event.metrics.total_tokens} tokens")

agent = Agent(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
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
ModelStreamChunkEvent	Each token/chunk streamed from the model
AfterModelCallEvent	After the model finishes responding

Tool calls

Event	When
BeforeToolCallEvent	Before a tool is executed
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
    print(f"Tool {event.tool.name} returned: {event.response}")

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