kaboo_workflows.hooks

Lifecycle hooks: event publishing, guards, sanitizers, and interrupts.

kaboo_workflows.hooks

Reusable HookProvider implementations for strands agents.

EventPublisher

EventPublisher(
    callback,
    agent_name,
    *,
    tool_labels=None,
    stream_group="",
    stream_title="",
    is_chat_reply=False,
    max_result_len=600,
)

Bases: HookProvider

Unified event publisher for single-agent and multi-agent orchestrations.

Initialize the EventPublisher.

Converts strands hook events into :class:StreamEvent objects and delivers them to an external callback. Emits an AGENT_COMPLETE event at the end of each invocation with usage metrics from EventLoopMetrics.

For TOKEN and REASONING events use :meth:as_callback_handler to create a strands-compatible callback_handler.

Parameters:

Name	Type	Description	Default
callback	EventCallback	Called with each :class:StreamEvent.	required
agent_name	str	Identifier for the agent or orchestrator.	required
tool_labels	dict[str, str] | None	Optional mapping of tool names to display labels.	None
stream_group	str	Dot-path stream group for hierarchical activity attribution. Empty string when not configured.	''
stream_title	str	Human-readable title for the stream group.	''
max_result_len	int	Maximum character length for tool result text in TOOL_END events. Default: 600.	600

Example::

publisher = EventPublisher(callback=on_event, agent_name="analyzer")
agent = Agent(
    hooks=[publisher],
    callback_handler=publisher.as_callback_handler(),
)

Source code in src/kaboo_workflows/hooks/event_publisher.py

def __init__(
    self,
    callback: EventCallback,
    agent_name: str,
    *,
    tool_labels: dict[str, str] | None = None,
    stream_group: str = "",
    stream_title: str = "",
    is_chat_reply: bool = False,
    max_result_len: int = 600,
) -> None:
    """Initialize the EventPublisher.

    Converts strands hook events into :class:`StreamEvent` objects and
    delivers them to an external callback.  Emits an AGENT_COMPLETE event at
    the end of each invocation with usage metrics from ``EventLoopMetrics``.

    For TOKEN and REASONING events use :meth:`as_callback_handler` to
    create a strands-compatible ``callback_handler``.

    Args:
        callback: Called with each :class:`StreamEvent`.
        agent_name: Identifier for the agent or orchestrator.
        tool_labels: Optional mapping of tool names to display labels.
        stream_group: Dot-path stream group for hierarchical activity
            attribution. Empty string when not configured.
        stream_title: Human-readable title for the stream group.
        max_result_len: Maximum character length for tool result text
            in TOOL_END events. Default: 600.

    Example::

        publisher = EventPublisher(callback=on_event, agent_name="analyzer")
        agent = Agent(
            hooks=[publisher],
            callback_handler=publisher.as_callback_handler(),
        )
    """
    self._callback = _safe_callback(callback)
    self._agent_name = agent_name
    self._tool_labels = tool_labels or {}
    self._stream_group = stream_group
    self._stream_title = stream_title
    self._is_chat_reply = is_chat_reply
    self._max_result_len = max_result_len
    # Set for the plain-agent entry node: its text and tool calls are already
    # rendered inline in the chat by the host (CopilotKit), so the activity
    # group exists only to enrich those tool rows — it must never also render
    # as a drill card. The AG-UI adapter flips this on when forwarding this
    # publisher to the ag-ui-strands clone. See create_agui_app.
    self._inline_chat_owner = False
    # Per-conversation state. A single EventPublisher instance is shared across
    # every concurrent thread that runs this agent (delegate nodes and the
    # forwarded plain-agent entry are singletons), so all per-invocation state
    # — the ``#N`` suffix, active group, and error latch — must be scoped by
    # thread_id to avoid cross-talk between concurrent conversations.
    self._invocation_count_by_thread: dict[str | None, int] = {}
    self._active_group_by_thread: dict[str | None, str] = {}
    self._errored_by_thread: dict[str | None, bool] = {}

mark_inline_chat_owner

mark_inline_chat_owner()

Flag this publisher's group as the inline chat owner (see __init__).

Source code in src/kaboo_workflows/hooks/event_publisher.py

def mark_inline_chat_owner(self) -> None:
    """Flag this publisher's group as the inline chat owner (see ``__init__``)."""
    self._inline_chat_owner = True

register_hooks

register_hooks(registry, **kwargs)

Register hook callbacks for agent and multiagent events.

Source code in src/kaboo_workflows/hooks/event_publisher.py

@override
def register_hooks(self, registry: HookRegistry, **kwargs: Any) -> None:
    """Register hook callbacks for agent and multiagent events."""
    # Agent-level
    registry.add_callback(BeforeInvocationEvent, self._on_agent_start)
    registry.add_callback(AfterModelCallEvent, self._on_model_error)
    registry.add_callback(BeforeToolCallEvent, self._on_tool_start)
    registry.add_callback(AfterToolCallEvent, self._on_tool_end)
    registry.add_callback(AfterInvocationEvent, self._on_complete)
    # Multiagent-level
    registry.add_callback(BeforeNodeCallEvent, self._on_node_start)
    registry.add_callback(AfterNodeCallEvent, self._on_node_stop)
    registry.add_callback(BeforeMultiAgentInvocationEvent, self._on_multiagent_start)
    registry.add_callback(AfterMultiAgentInvocationEvent, self._on_multiagent_complete)

as_callback_handler

as_callback_handler()

Return a strands-compatible callback_handler for TOKEN, REASONING, and HANDOFF events.

Handles the following kwarg patterns emitted by strands: - data (str): A streamed text chunk -> TOKEN event. - reasoningText (str): A reasoning chunk -> REASONING event. - type == "multiagent_handoff": A :class:~strands.types._events.MultiAgentHandoffEvent fired during Swarm/Graph node transitions -> HANDOFF event.

Returns:

Type	Description
Callable[..., None]	A callable compatible with strands callback_handler interface.

Source code in src/kaboo_workflows/hooks/event_publisher.py

def as_callback_handler(self) -> Callable[..., None]:
    """Return a strands-compatible callback_handler for TOKEN, REASONING, and HANDOFF events.

    Handles the following kwarg patterns emitted by strands:
    - ``data`` (str): A streamed text chunk -> TOKEN event.
    - ``reasoningText`` (str): A reasoning chunk -> REASONING event.
    - ``type == "multiagent_handoff"``: A :class:`~strands.types._events.MultiAgentHandoffEvent`
      fired during Swarm/Graph node transitions -> HANDOFF event.

    Returns:
        A callable compatible with strands ``callback_handler`` interface.
    """

    def _handler(**kwargs: Any) -> None:
        text: str = kwargs.get("data", "")
        if text:
            self._callback(
                StreamEvent(
                    type=EventType.TOKEN,
                    agent_name=self._agent_name,
                    data=self._enrich({"text": text}),
                ),
            )

        reasoning: str = kwargs.get("reasoningText", "")
        if reasoning:
            self._callback(
                StreamEvent(
                    type=EventType.REASONING,
                    agent_name=self._agent_name,
                    data=self._enrich({"text": reasoning}),
                ),
            )

        if kwargs.get("type") == "multiagent_handoff":
            self._callback(
                StreamEvent(
                    type=EventType.HANDOFF,
                    agent_name=self._agent_name,
                    data=self._enrich(
                        {
                            "from_node_ids": kwargs.get("from_node_ids", []),
                            "to_node_ids": kwargs.get("to_node_ids", []),
                            "message": kwargs.get("message"),
                        }
                    ),
                )
            )

    return _handler

ForwardedPropsHook

ForwardedPropsHook(
    *, apply_agent_config=False, state_key="forwarded_props"
)

Bases: HookProvider

Copy forwarded props into agent state; optionally apply agent overrides.

Parameters:

Name	Type	Description	Default
apply_agent_config	bool	When True, forwarded_props.agent_config (system_prompt/model_id) is applied to the executing agent before each invocation. Deprecated — a run can now submit its whole config, which expresses the prompt and model along with the structure they could not. See create_agui_app(session_config_key=).	False
state_key	str	Agent-state key the props are stored under.	'forwarded_props'

Source code in src/kaboo_workflows/hooks/forwarded_props_hook.py

def __init__(
    self,
    *,
    apply_agent_config: bool = False,
    state_key: str = "forwarded_props",
) -> None:
    if apply_agent_config:
        warnings.warn(
            "runtime.allow_invocation_overrides / forwardedProps.agent_config is "
            "deprecated and will be removed in 0.15.0. Submit the run's config "
            "instead: create_agui_app(config, session_config_key='workflow_config').",
            DeprecationWarning,
            stacklevel=2,
        )
    self._apply_agent_config = apply_agent_config
    self._state_key = state_key

HistoryHook

HistoryHook(key, *, enabled)

Bases: HookProvider

Seeds an agent's history from, and captures it back into, the request's :class:~kaboo_workflows._context.HistoryExchange.

One instance is attached per agent, carrying that agent's history key (its shared-transcript bucket or stable dot-path) and whether history is enabled for it.

Initialize the HistoryHook.

Parameters:

Name	Type	Description	Default
key	str	History bucket key for this agent. Agents that share a history.group share a key (and therefore a transcript).	required
enabled	bool	Whether this agent remembers across turns of the same conversation. When False the agent is reset to an empty transcript before every invocation and never persists.	required

Source code in src/kaboo_workflows/hooks/history_hook.py

def __init__(self, key: str, *, enabled: bool) -> None:
    """Initialize the HistoryHook.

    Args:
        key: History bucket key for this agent. Agents that share a
            ``history.group`` share a key (and therefore a transcript).
        enabled: Whether this agent remembers across turns of the same
            conversation. When ``False`` the agent is reset to an empty
            transcript before every invocation and never persists.
    """
    self._key = key
    self._enabled = enabled

register_hooks

register_hooks(registry, **kwargs)

Register seed (before) and capture (after) invocation callbacks.

Source code in src/kaboo_workflows/hooks/history_hook.py

@override
def register_hooks(self, registry: HookRegistry, **kwargs: Any) -> None:
    """Register seed (before) and capture (after) invocation callbacks."""
    registry.add_callback(BeforeInvocationEvent, self._on_before)
    registry.add_callback(AfterInvocationEvent, self._on_after)

InterruptHook

InterruptHook(tools, *, agent_name='', ttl_seconds=None)

Bases: HookProvider

Gate specific tool calls with user approval via strands interrupts.

When the agent attempts to call a tool whose name is in the configured tools list, this hook fires event.interrupt() with a structured reason payload. The agent pauses until the user approves or rejects.

When ttl_seconds is set, the reason carries an expiresAt ISO-8601 timestamp which the AG-UI adapter surfaces on the interrupt descriptor (Interrupt.expiresAt), letting clients render a countdown and servers expire unanswered approvals.

An approval response may carry an edited tool_input (AG-UI approveWithEdits): the gated call then executes with the user's arguments instead of the agent's — e.g. approving a subset of a bulk proposal. The edit applies to this one call only.

Source code in src/kaboo_workflows/hooks/interrupt_hook.py

def __init__(
    self,
    tools: list[str],
    *,
    agent_name: str = "",
    ttl_seconds: int | None = None,
) -> None:
    self._tools = set(tools)
    self._agent_name = agent_name
    self._ttl_seconds = ttl_seconds

MaxToolCallsGuard

MaxToolCallsGuard(max_calls=25)

Bases: HookProvider

Stops the agent after a maximum number of tool calls per invocation.

Initialize the MaxToolCallsGuard.

On first violation the LLM is instructed to stop using tools and write a final answer (graceful shutdown). If the LLM ignores that and requests another tool call, the event loop is terminated immediately (hard stop).

Uses strands' invocation_state dict for per-invocation state — the counter and flags reset automatically on each new agent() call.

Parameters:

Name	Type	Description	Default
max_calls	int	Maximum tool calls allowed per invocation. Default: 25.	25

Source code in src/kaboo_workflows/hooks/max_calls_guard.py

def __init__(self, max_calls: int = 25) -> None:
    """Initialize the MaxToolCallsGuard.

    On first violation the LLM is instructed to stop using tools and write a
    final answer (graceful shutdown).  If the LLM ignores that and requests
    another tool call, the event loop is terminated immediately (hard stop).

    Uses strands' ``invocation_state`` dict for per-invocation state — the
    counter and flags reset automatically on each new ``agent()`` call.

    Args:
        max_calls: Maximum tool calls allowed per invocation. Default: 25.
    """
    self.max_calls = max_calls

register_hooks

register_hooks(registry, **kwargs)

Register the BeforeToolCallEvent callback.

Source code in src/kaboo_workflows/hooks/max_calls_guard.py

@override
def register_hooks(self, registry: HookRegistry, **kwargs: Any) -> None:
    """Register the BeforeToolCallEvent callback."""
    registry.add_callback(BeforeToolCallEvent, self._on_before_tool)

MCPCallMetaHook

MCPCallMetaHook(provider=None, *, stamp_tool_call_id=True)

Bases: HookProvider

Attach per-call _meta to every MCP tool call an agent makes.

On :class:~strands.hooks.events.BeforeToolCallEvent — which fires in the caller's request context — the hook computes the metadata for this call and swaps event.selected_tool for a delegate that forwards it via MCPClient.call_tool_async(meta=...) (the documented tool-replacement seam of BeforeToolCallEvent). Non-MCP tools are untouched.

Parameters:

Name	Type	Description	Default
provider	Any | None	Callable (tool_use) -> dict | None returning metadata for one call, or an import spec string (module.path:name / ./file.py:name) resolved to such a callable. None stamps only toolCallId.	None
stamp_tool_call_id	bool	When True (default), toolCallId is set to the strands toolUseId unless the provider already supplied one.	True

Source code in src/kaboo_workflows/hooks/mcp_meta_hook.py

def __init__(
    self,
    provider: Any | None = None,
    *,
    stamp_tool_call_id: bool = True,
) -> None:
    if isinstance(provider, str):
        provider = load_object(provider, target="MCP call-meta provider")
    if provider is not None and not callable(provider):
        raise TypeError(
            f"MCPCallMetaHook provider must be callable or an import spec, "
            f"got {type(provider).__name__}."
        )
    self._provider = provider
    self._stamp_tool_call_id = stamp_tool_call_id

ReferenceHook

ReferenceHook(*, enabled, inline, tool_enabled)

Bases: HookProvider

Injects the reference manifest into an in-scope agent before each invocation.

One instance is attached per agent, carrying that agent's policy: whether it is in scope at all, whether it also gets inline media, and whether the resolver tool is available (for the manifest wording).

Initialize the ReferenceHook.

Parameters:

Name	Type	Description	Default
enabled	bool	Whether this agent is in scope for references. When False the hook is a no-op (attachments: none).	required
inline	bool	Whether to additionally prepend resolved media ContentBlocks for a vision/doc-capable model.	required
tool_enabled	bool	Whether the built-in reference tools are registered (affects the manifest's guidance wording only).	required

Source code in src/kaboo_workflows/hooks/reference_hook.py

def __init__(self, *, enabled: bool, inline: bool, tool_enabled: bool) -> None:
    """Initialize the ReferenceHook.

    Args:
        enabled: Whether this agent is in scope for references. When
            ``False`` the hook is a no-op (``attachments: none``).
        inline: Whether to additionally prepend resolved media
            ``ContentBlock``s for a vision/doc-capable model.
        tool_enabled: Whether the built-in reference tools are registered
            (affects the manifest's guidance wording only).
    """
    self._enabled = enabled
    self._inline = inline
    self._tool_enabled = tool_enabled

register_hooks

register_hooks(registry, **kwargs)

Register manifest (before-invocation) + inline media (before-model) callbacks.

The before-model callback also serves fetch_attachment requests (media the model asked to read), so it is registered whenever the agent is inline or has the reference tools available.

Source code in src/kaboo_workflows/hooks/reference_hook.py

@override
def register_hooks(self, registry: HookRegistry, **kwargs: Any) -> None:
    """Register manifest (before-invocation) + inline media (before-model) callbacks.

    The before-model callback also serves ``fetch_attachment`` requests
    (media the model asked to read), so it is registered whenever the agent
    is inline **or** has the reference tools available.
    """
    registry.add_callback(BeforeInvocationEvent, self._on_before_invocation)
    if self._inline or self._tool_enabled:
        registry.add_callback(BeforeModelCallEvent, self._on_before_model)

SessionStateHook

Bases: HookProvider

Restore client-supplied interrupt state onto the executing agent.

Also records that agent on the request's :class:~kaboo_workflows._context.SessionExchange, so the adapter can serialize its state once the run has paused or finished.

MultiAgentStopGuard

MultiAgentStopGuard(stop_check)

Bases: HookProvider

Cancels node execution when an external stop condition is met.

Initialize the MultiAgentStopGuard.

Counterpart to :class:StopGuard for multi-agent orchestrations. Registers a BeforeNodeCallEvent callback on a Swarm or Graph's hook registry. When the stop_check callable returns True, the hook sets cancel_node to prevent the next node from starting.

Parameters:

Name	Type	Description	Default
stop_check	Callable[[], bool]	Callable returning True when stop is requested. Must be thread-safe.	required

Source code in src/kaboo_workflows/hooks/stop_guard.py

def __init__(self, stop_check: Callable[[], bool]) -> None:
    """Initialize the MultiAgentStopGuard.

    Counterpart to :class:`StopGuard` for multi-agent orchestrations.
    Registers a ``BeforeNodeCallEvent`` callback on a Swarm or Graph's
    hook registry.  When the *stop_check* callable returns ``True``, the
    hook sets ``cancel_node`` to prevent the next node from starting.

    Args:
        stop_check: Callable returning ``True`` when stop is requested.
            Must be thread-safe.
    """
    self._stop_check = stop_check

register_hooks

register_hooks(registry, **kwargs)

Register the before-node-call guard.

Source code in src/kaboo_workflows/hooks/stop_guard.py

@override
def register_hooks(self, registry: HookRegistry, **kwargs: Any) -> None:
    """Register the before-node-call guard."""
    registry.add_callback(BeforeNodeCallEvent, self._on_before_node)

StopGuard

StopGuard(stop_check)

Bases: HookProvider

Cancels the agent's event loop when an external stop condition is met.

Initialize the StopGuard.

The stop condition is checked before every tool call. When it returns True, the current tool is cancelled and the event loop is stopped.

Parameters:

Name	Type	Description	Default
stop_check	Callable[[], bool]	Callable that returns True when the agent should stop. Must be thread-safe. Common patterns: threading.Event().is_set lambda: some_shared_flag lambda: not process_is_alive()	required

Source code in src/kaboo_workflows/hooks/stop_guard.py

def __init__(self, stop_check: Callable[[], bool]) -> None:
    """Initialize the StopGuard.

    The stop condition is checked before every tool call.  When it returns
    ``True``, the current tool is cancelled and the event loop is stopped.

    Args:
        stop_check: Callable that returns ``True`` when the agent should stop.
            Must be thread-safe.  Common patterns:

            - ``threading.Event().is_set``
            - ``lambda: some_shared_flag``
            - ``lambda: not process_is_alive()``
    """
    self._should_stop = stop_check

register_hooks

register_hooks(registry, **kwargs)

Register the before-tool-call guard.

Source code in src/kaboo_workflows/hooks/stop_guard.py

@override
def register_hooks(self, registry: HookRegistry, **kwargs: Any) -> None:
    """Register the before-tool-call guard."""
    registry.add_callback(BeforeToolCallEvent, self._on_before_tool)

ToolNameSanitizer

Bases: HookProvider

Strips model-injected artifacts from tool names so Strands can look them up.

Registers on: - AfterModelCallEvent: rewrites tool names in the model response message. - BeforeToolCallEvent: safety net — fixes or cancels still-garbled names.

register_hooks

register_hooks(registry, **kwargs)

Register AfterModelCall and BeforeToolCall sanitization callbacks.

Source code in src/kaboo_workflows/hooks/tool_name_sanitizer.py

@override
def register_hooks(self, registry: HookRegistry, **kwargs: Any) -> None:
    """Register AfterModelCall and BeforeToolCall sanitization callbacks."""
    registry.add_callback(AfterModelCallEvent, self._on_after_model)
    registry.add_callback(BeforeToolCallEvent, self._on_before_tool)

restore_session_state

restore_session_state(agent)

Seed agent with the interrupt state the client sent, if any.

Returns whether anything was restored. Idempotent, so the adapter can call it on the resume path — where the state is needed before the run starts, to build the interrupt responses — and the hook can call it again for a fresh run without either clobbering the other.

Source code in src/kaboo_workflows/hooks/session_state_hook.py

def restore_session_state(agent: Any) -> bool:
    """Seed *agent* with the interrupt state the client sent, if any.

    Returns whether anything was restored. Idempotent, so the adapter can call it
    on the resume path — where the state is needed *before* the run starts, to
    build the interrupt responses — and the hook can call it again for a fresh
    run without either clobbering the other.
    """
    exchange = get_session_exchange()
    if exchange is None:
        return False
    exchange.agent = agent
    state = exchange.inbound.get("interrupt_state")
    if not _is_interrupt_state(state):
        return False
    # A warm agent's own state is at least as fresh as the client's, since the
    # client got it from us. Restoring over it would let a stale snapshot
    # resurrect an answered gate, so only seed a clean agent.
    current = getattr(agent, "_interrupt_state", None)
    if current is not None and (current.activated or current.interrupts):
        return False
    try:
        agent._interrupt_state = _InterruptState.from_dict(state)
    except Exception:
        logger.exception("kaboo_session restore failed; starting with clean state")
        return False
    logger.debug(
        "kaboo_session restored | activated=%s interrupts=%s",
        state.get("activated"),
        list(state.get("interrupts") or {}),
    )
    return True

session_state_snapshot

session_state_snapshot()

Serialize the current request's agent state for the outgoing snapshot.

Returns None when there is nothing worth sending: no exchange, no agent reached invocation, or the agent is not holding an interrupt. Writing an empty state would still be correct, but it would add a key to every snapshot of every run that never pauses.

Source code in src/kaboo_workflows/hooks/session_state_hook.py

def session_state_snapshot() -> dict[str, Any] | None:
    """Serialize the current request's agent state for the outgoing snapshot.

    Returns ``None`` when there is nothing worth sending: no exchange, no agent
    reached invocation, or the agent is not holding an interrupt. Writing an
    empty state would still be correct, but it would add a key to every
    snapshot of every run that never pauses.
    """
    exchange = get_session_exchange()
    if exchange is None or exchange.agent is None:
        return None
    state = getattr(exchange.agent, "_interrupt_state", None)
    if state is None:
        return None
    try:
        serialized = state.to_dict()
    except Exception:
        logger.exception("kaboo_session capture failed; snapshot left unchanged")
        return None
    if not serialized.get("activated") and not serialized.get("interrupts"):
        # Nothing pending. Send it anyway if the client sent state, so a
        # resolved gate is cleared rather than replayed on the next turn.
        if not _is_interrupt_state(exchange.inbound.get("interrupt_state")):
            return None
    return {"interrupt_state": serialized}

stop_guard_from_event

stop_guard_from_event(event=None)

Create a StopGuard backed by a threading.Event.

Convenience factory for the common pattern of using a threading.Event as the external stop signal. Can be used to wire stop-on-disconnect or for programmatic stop control.

Example::

guard, stop = stop_guard_from_event()
agent.hooks.add_hook(guard)

# later, from any thread:
stop.set()  # agent stops at next tool-call boundary

Parameters:

Name	Type	Description	Default
event	Event | None	Optional pre-existing threading.Event. When omitted a new event is created internally.	None

Returns:

Type	Description
tuple[StopGuard, Event]	Tuple of (guard, event). Call event.set() to trigger the stop.

Source code in src/kaboo_workflows/hooks/stop_guard.py

def stop_guard_from_event(
    event: threading.Event | None = None,
) -> tuple[StopGuard, threading.Event]:
    """Create a StopGuard backed by a ``threading.Event``.

    Convenience factory for the common pattern of using a
    ``threading.Event`` as the external stop signal.  Can be used to
    wire stop-on-disconnect or for programmatic stop control.

    Example::

        guard, stop = stop_guard_from_event()
        agent.hooks.add_hook(guard)

        # later, from any thread:
        stop.set()  # agent stops at next tool-call boundary

    Args:
        event: Optional pre-existing ``threading.Event``.  When omitted a
            new event is created internally.

    Returns:
        Tuple of ``(guard, event)``.  Call ``event.set()`` to trigger the stop.
    """
    if event is None:
        event = threading.Event()
    return StopGuard(stop_check=event.is_set), event