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

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='')

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.

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

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

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)

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)

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