Lifecycle Callbacks

A lifecycle hook is a class that runs code at structural points of a graph run: start, end, error, interrupt, resume, checkpoint and each state update. Subclass GraphLifecycleHook, override the methods you need, and register it on a CallbackManager. For per-LLM, tool and MCP call hooks, see Callback Manager. For patterns, see Use callbacks.

Import

Python

from tenxgraph.utils.callbacks import GraphLifecycleHook, GraphLifecycleContext
from tenxgraph.core.state import AgentState, Message

GraphLifecycleContext

Passed to every hook with metadata about the current execution.

Fields:

Field	Type	Description
`config`	`dict[str, Any]`	Full config dict passed to `invoke()` or `stream()`.

Properties:

Property	Type	Returns
`thread_id`	`str`	`config.get("thread_id", "")`, conversation ID or empty string.
`run_id`	`str`	`config.get("run_id", "")`, execution ID or empty string.

GraphLifecycleHook

Base class whose methods are all no-ops that return None. Override only the methods you need.

on_graph_start

Fires after state is loaded but before the first node executes.

Signature:

Python

async def on_graph_start(
    self,
    context: GraphLifecycleContext,
    state: AgentState,
) -> AgentState | None:

Parameters:

Param	Type	Description
`context`	`GraphLifecycleContext`	Execution context with thread_id, run_id, config.
`state`	`AgentState`	Loaded or freshly created initial state.

Returns: AgentState | None

Return None to use original state unchanged.
Return modified AgentState to replace initial state before graph execution.

Example:

Python

class TraceStartHook(GraphLifecycleHook):
    async def on_graph_start(self, context, state):
        print(f"Starting: thread={context.thread_id} run={context.run_id}")
        return state  # or None to keep unchanged

on_graph_end

Fires after execution loop completes successfully, before final persistence.

Signature:

Python

async def on_graph_end(
    self,
    context: GraphLifecycleContext,
    final_state: AgentState,
    messages: list[Message],
    total_steps: int,
) -> AgentState | None:

Parameters:

Param	Type	Description
`context`	`GraphLifecycleContext`	Execution context.
`final_state`	`AgentState`	State after execution completes.
`messages`	`list[Message]`	All messages produced during the run.
`total_steps`	`int`	Final step count.

Returns: AgentState | None

Return None to persist unchanged.
Return modified AgentState to persist and return modified state.

Example:

Python

class MetricsEndHook(GraphLifecycleHook):
    async def on_graph_end(self, context, final_state, messages, total_steps):
        print(f"Completed: {total_steps} steps, {len(messages)} messages")
        return None

on_graph_error

Fires when an unhandled exception escapes the execution loop. Cannot suppress the error. The exception is always re-raised after the hook completes.

Signature:

Python

async def on_graph_error(
    self,
    context: GraphLifecycleContext,
    error: Exception,
    partial_state: AgentState,
    messages: list[Message],
    step: int,
    node_name: str,
) -> tuple[AgentState, str] | None:

Parameters:

Param	Type	Description
`context`	`GraphLifecycleContext`	Execution context.
`error`	`Exception`	The unhandled exception.
`partial_state`	`AgentState`	State at error point (after `state.error()`).
`messages`	`list[Message]`	Messages collected before error.
`step`	`int`	Step number where error occurred.
`node_name`	`str`	The node that failed.

Returns: tuple[AgentState, str] | None

Return None to persist unchanged; re-raise original error message.
Return (AgentState, str) to persist modified state and error message, then re-raise.

Example:

Python

class ErrorAlertHook(GraphLifecycleHook):
    async def on_graph_error(self, context, error, partial_state, messages, step, node_name):
        print(f"Error at {node_name}: {error}")
        return None  # or modify state before re-raising

on_interrupt

Fires when graph execution pauses at an interrupt point.

Signature:

Python

async def on_interrupt(
    self,
    context: GraphLifecycleContext,
    interrupted_node: str,
    interrupt_type: str,
    state: AgentState,
) -> AgentState | None:

Parameters:

Param	Type	Description
`context`	`GraphLifecycleContext`	Execution context.
`interrupted_node`	`str`	The node where execution paused.
`interrupt_type`	`str`	`"before"`, `"after"`, `"stop"`, `"remote_tool"` or `"interrupt"`.
`state`	`AgentState`	State at interrupt point.

Returns: AgentState | None

Return None to persist unchanged.
Return modified AgentState to persist modified state.

Example:

Python

class InterruptNotificationHook(GraphLifecycleHook):
    async def on_interrupt(self, context, interrupted_node, interrupt_type, state):
        print(f"Paused at {interrupted_node} ({interrupt_type})")
        return state

on_resume

Fires when a previously interrupted graph is resumed.

Signature:

Python

async def on_resume(
    self,
    context: GraphLifecycleContext,
    resumed_node: str,
    state: AgentState,
    resume_data: dict[str, Any],
) -> AgentState | None:

Parameters:

Param	Type	Description
`context`	`GraphLifecycleContext`	Execution context.
`resumed_node`	`str`	The node being resumed.
`state`	`AgentState`	The loaded interrupted state (before clearing interrupt).
`resume_data`	`dict[str, Any]`	Mutable dict with data passed to `ainvoke()` on resume.

Returns: AgentState | None

Return None to continue with loaded state.
Return modified AgentState to continue with modified state.

Example:

Python

class ResumeValidationHook(GraphLifecycleHook):
    async def on_resume(self, context, resumed_node, state, resume_data):
        print(f"Resumed at {resumed_node}")
        return state

on_checkpoint

Fires immediately before state/messages are persisted to the checkpointer. Fires every time state is persisted, including during interrupts, stops, and errors.

Signature:

Python

async def on_checkpoint(
    self,
    context: GraphLifecycleContext,
    state: AgentState,
    messages: list[Message],
    is_context_trimmed: bool,
) -> tuple[AgentState, list[Message]] | AgentState | None:

Parameters:

Param	Type	Description
`context`	`GraphLifecycleContext`	Execution context.
`state`	`AgentState`	State about to be persisted.
`messages`	`list[Message]`	Messages about to be persisted.
`is_context_trimmed`	`bool`	Whether context compression was applied.

Returns: tuple[AgentState, list[Message]] | AgentState | None

Return None to persist unchanged.
Return AgentState to persist returned state with current messages.
Return (AgentState, list[Message]) to persist both returned values.

Example:

Python

class CheckpointHook(GraphLifecycleHook):
    async def on_checkpoint(self, context, state, messages, is_context_trimmed):
        print(f"Checkpointing {len(messages)} messages")
        return None

on_state_update

Fires after each node transition, after a node executes and state is merged. Most granular graph-level hook.

Signature:

Python

async def on_state_update(
    self,
    context: GraphLifecycleContext,
    node_name: str,
    old_state: AgentState,
    new_state: AgentState,
    step: int,
) -> AgentState | None:

Parameters:

Param	Type	Description
`context`	`GraphLifecycleContext`	Execution context.
`node_name`	`str`	The node that just executed.
`old_state`	`AgentState`	Deep copy of state before node ran.
`new_state`	`AgentState`	State after node result merged.
`step`	`int`	Current step number.

Returns: AgentState | None

Return None to use new_state unchanged.
Return modified AgentState to replace new_state for rest of graph.

Example:

Python

class ObserveTransitionHook(GraphLifecycleHook):
    async def on_state_update(self, context, node_name, old_state, new_state, step):
        msg_delta = len(new_state.messages) - len(old_state.messages)
        print(f"Step {step}: {node_name} (+{msg_delta} messages)")
        return None

on_turn_start and on_turn_end

Realtime sessions only. Fire when a conversation turn begins or completes. A turn spans one model generation; turn_index is 1-based. Never fire for turn-based invoke or stream runs.

Signatures:

Python

async def on_turn_start(
    self,
    context: GraphLifecycleContext,
    state: AgentState,
    turn_index: int,
) -> AgentState | None:

async def on_turn_end(
    self,
    context: GraphLifecycleContext,
    state: AgentState,
    turn_index: int,
) -> AgentState | None:

Parameters:

Param	Type	Description
`context`	`GraphLifecycleContext`	Execution context.
`state`	`AgentState`	Current state.
`turn_index`	`int`	1-based turn number.

Returns: AgentState | None

Return modified AgentState to replace current state, or None to keep it.

Example:

Python

class TurnLogHook(GraphLifecycleHook):
    async def on_turn_end(self, context, state, turn_index):
        print(f"Turn {turn_index} ended")
        return None

Register a hook

Python

from tenxgraph.core.graph import StateGraph
from tenxgraph.core.state import AgentState
from tenxgraph.utils.callbacks import CallbackManager, GraphLifecycleHook

class MyLifecycleHook(GraphLifecycleHook):
    async def on_graph_start(self, context, state):
        print(f"Run {context.run_id} started")
        return None


cbm = CallbackManager()
cbm.register_lifecycle_hook(MyLifecycleHook())

graph = StateGraph(AgentState)
# Add your nodes and edges here, then compile with the manager.
app = graph.compile(callback_manager=cbm)

# Hooks fire automatically during invoke/stream
# result = await app.ainvoke({"messages": [Message.text_message("Hi")]})

Common errors

Error	Cause	Fix
Lifecycle hook never fires	Hook not registered on `CallbackManager` passed to `compile()`.	Call `cbm.register_lifecycle_hook(MyHook())` before `compile()`.
`on_graph_error` doesn’t suppress error	By design; the exception always re-raises.	Use node-level `on_error` callbacks in Callback Manager for error recovery.
`on_interrupt` not called	Interrupt not triggered; node completed successfully.	Set `interrupt_before` or `interrupt_after` on `compile()`, or call `stop()` through the API.