Hooks API

---

Hook events

All events live in elsai.hooks:

from elsai.hooks import (
    AgentInitializedEvent,
    UserInputReceived,
    BeforeInvocationEvent,
    AfterInvocationEvent,
    BeforeModelCallEvent,
    AfterModelCallEvent,
    BeforeToolCallEvent,
    AfterToolCallEvent,
    BeforeMemoryWrite,
    MessageAddedEvent,
    BeforeResponseDispatch,
    AgentErrorEvent,
)

Streaming types

Streaming token deltas are not hook events. Use Agent.stream_async at the agent layer, or Model.stream at the model layer. Type-hint model chunks with StreamEvent from elsai.types.streaming — not internal types from elsai.types._events.

---

Agent lifecycle events

AgentInitializedEvent

Fires once when Agent.__init__ completes.

Attribute	Type
agent	Agent

UserInputReceived

Fires when new input arrives (before processing starts).

Attribute	Type
agent	Agent
invocation_state	dict
messages	list[Message]

BeforeInvocationEvent

Fires before the event loop starts for this turn.

Attribute	Writable	Type
agent	No	Agent
messages	✅	list[Message] | None
invocation_state	No	dict

AfterInvocationEvent

Fires after the event loop ends (even on exception).

Attribute	Writable	Type
agent	No	Agent
result	No	AgentResult | None
resume	✅	AgentInput | None

Set resume to trigger another agent turn.

BeforeResponseDispatch

Fires just before the final AgentResult is returned.

Attribute	Writable	Type
agent	No	Agent
result	✅	AgentResult
invocation_state	No	dict

AgentErrorEvent

Fires when an unhandled exception occurs.

Attribute	Type
agent	Agent
exception	Exception
invocation_state	dict

---

Model call events

BeforeModelCallEvent

Attribute	Type
agent	Agent
invocation_state	dict
projected_input_tokens	int | None

AfterModelCallEvent

Attribute	Writable	Type
agent	No	Agent
invocation_state	No	dict
stop_response	No	ModelStopResponse | None
exception	No	Exception | None
retry	✅	bool

When streaming is enabled, token deltas flow between these two hooks internally. To consume them in application code, use Agent.stream_async or Model.stream and type-hint chunks with StreamEvent.

---

Tool call events

BeforeToolCallEvent

Attribute	Writable	Type
agent	No	Agent
selected_tool	✅	AgentTool
tool_use	No	dict

AfterToolCallEvent

Attribute	Writable	Type
agent	No	Agent
selected_tool	No	AgentTool | None
tool_use	No	ToolUse
result	✅	ToolResult
exception	No	Exception | None
retry	✅	bool

---

Memory events

BeforeMemoryWrite

Attribute	Writable	Type
agent	No	Agent
messages	✅	list[Message]

MessageAddedEvent

Attribute	Type
agent	Agent
message	Message

---

HookProvider

from elsai.hooks import HookProvider, HookRegistry

class MyHooks(HookProvider):
    def register_hooks(self, registry: HookRegistry) -> None:
        registry.add_callback(BeforeModelCallEvent, self.on_before_model)
        registry.add_callback(AfterToolCallEvent, self.on_after_tool)

    def on_before_model(self, event: BeforeModelCallEvent) -> None:
        ...

    def on_after_tool(self, event: AfterToolCallEvent) -> None:
        ...

from elsai.agent import AgentConfig

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

---

HookRegistry

registry = HookRegistry()
registry.add_callback(event_type, callback)
registry.add_hook(hook_provider)

Accessible via agent.hooks.

---

Registration patterns

# From type hint
agent.add_hook(lambda event: print(event), BeforeModelCallEvent)

# Inferred from type hint
def my_hook(event: AfterToolCallEvent) -> None:
    ...
agent.add_hook(my_hook)

# Multiple types
def multi(event: BeforeToolCallEvent | AfterToolCallEvent) -> None:
    ...
agent.add_hook(multi)

# List of types
agent.add_hook(my_hook, [BeforeModelCallEvent, AfterModelCallEvent])