kaboo_workflows.hooks

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

kaboo_workflows.hooks

Reusable HookProvider implementations for strands agents.

APPROVAL_SCOPES module-attribute

APPROVAL_SCOPES = ('once', 'thread', 'always')

Decision scopes, narrowest first: this call, this conversation, from now on.

GRANTS_PROP module-attribute

GRANTS_PROP = 'kaboo_grants'

forwardedProps key carrying the decisions the host remembers.

TOOL_APPROVAL_KEY module-attribute

TOOL_APPROVAL_KEY = 'kaboo_approval'

Key of the reply a tool sends instead of a result when a call needs approval.

ContextHook

ContextHook(
    *,
    clear_trigger=None,
    keep_last=3,
    min_chars=DEFAULT_MIN_CHARS,
    keep_tools=(),
    client_tools=(),
    workspace=(),
    views=(),
)

Bases: HookProvider

Rewrites the copy of an agent's conversation its model is sent.

Initialize the ContextHook.

Pass it to the agent at construction (Agent(hooks=[...])): it installs its middleware when the agent finishes initializing, so a hook added to an existing agent never runs.

Parameters:

Name	Type	Description	Default
clear_trigger	int | None	Estimated input tokens (system prompt plus messages, four characters a token) past which old tool results are cleared before a model call. None never clears.	None
keep_last	int	How many of the most recent turns of tool results always stay; a turn is every result of one model response.	3
min_chars	int	Tool results shorter than this always stay.	DEFAULT_MIN_CHARS
keep_tools	Collection[str]	Tools whose results always stay. A delegate's always do.	()
client_tools	Collection[str]	The browser tools the agent may call; context items that belong to another browser tool stay out of its calls.	()
workspace	Collection[str]	JSON Pointers into the shared state the agent sees at every model call (see :func:~kaboo_workflows.hooks.state_effects.workspace_view).	()
views	Collection[str | Callable[[ViewRequest], View | None]]	Working views: callables taking a :class:ViewRequest and returning a :class:View or None, or import specs of them. A view that raises is skipped for that call and logged.	()

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

def __init__(
    self,
    *,
    clear_trigger: int | None = None,
    keep_last: int = 3,
    min_chars: int = DEFAULT_MIN_CHARS,
    keep_tools: Collection[str] = (),
    client_tools: Collection[str] = (),
    workspace: Collection[str] = (),
    views: Collection[str | Callable[[ViewRequest], View | None]] = (),
) -> None:
    """Initialize the ContextHook.

    Pass it to the agent at construction (``Agent(hooks=[...])``): it
    installs its middleware when the agent finishes initializing, so a hook
    added to an existing agent never runs.

    Args:
        clear_trigger: Estimated input tokens (system prompt plus messages,
            four characters a token) past which old tool results are
            cleared before a model call. ``None`` never clears.
        keep_last: How many of the most recent turns of tool results always
            stay; a turn is every result of one model response.
        min_chars: Tool results shorter than this always stay.
        keep_tools: Tools whose results always stay. A delegate's always do.
        client_tools: The browser tools the agent may call; context items
            that belong to another browser tool stay out of its calls.
        workspace: JSON Pointers into the shared state the agent sees at
            every model call (see
            :func:`~kaboo_workflows.hooks.state_effects.workspace_view`).
        views: Working views: callables taking a :class:`ViewRequest` and
            returning a :class:`View` or ``None``, or import specs of them.
            A view that raises is skipped for that call and logged.
    """
    self._clear_trigger = clear_trigger
    self._keep_last = keep_last
    self._min_chars = min_chars
    self._keep_tools = frozenset(keep_tools)
    self._client_tools = frozenset(client_tools)
    self._workspace = tuple(workspace)
    self._views = tuple(
        load_object(view, target="working view") if isinstance(view, str) else view
        for view in views
    )

register_hooks

register_hooks(registry, **kwargs)

Install the model-input middleware once the agent is initialized.

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

@override
def register_hooks(self, registry: HookRegistry, **kwargs: Any) -> None:
    """Install the model-input middleware once the agent is initialized."""
    registry.add_callback(AgentInitializedEvent, self._on_initialized)

View dataclass

View(base=None, changes=None)

A working view: a base for the whole invocation and the changes since it.

ViewRequest dataclass

ViewRequest(
    agent,
    thread_id,
    run_id,
    first_call,
    context,
    state=dict(),
)

What a working view is asked to render at one model call.

first_call instance-attribute

first_call

Whether this is the invocation's first model call, when the base is kept.

context instance-attribute

context

The run's AG-UI context items, such as what the canvas has open.

state class-attribute instance-attribute

state = field(default_factory=dict)

The shared state as of this call, state effects included. Read it; never modify it.

EventPublisher

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

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: 16,000.	_MAX_RESULT_LEN

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 = _MAX_RESULT_LEN,
) -> 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: 16,000.

    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}),
                ),
            )

        current = kwargs.get("current_tool_use")
        if (
            kwargs.get("type") == "tool_use_stream"
            and isinstance(current, dict)
            and current.get("toolUseId")
        ):
            self._callback(
                StreamEvent(
                    type=EventType.TOOL_INPUT,
                    agent_name=self._agent_name,
                    data=self._enrich(
                        {
                            "tool_use_id": current["toolUseId"],
                            "tool_name": current.get("name", ""),
                            "tool_label": _resolve_tool_label(
                                current.get("name", ""), self._tool_labels
                            ),
                            "input": current.get("input", ""),
                        }
                    ),
                ),
            )

        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, policy=None)

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
policy	HistoryPolicy | None	How the written-back transcript is shaped. Defaults to writing it back in full.	None

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

def __init__(self, key: str, *, enabled: bool, policy: HistoryPolicy | None = None) -> 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.
        policy: How the written-back transcript is shaped. Defaults to
            writing it back in full.
    """
    self._key = key
    self._enabled = enabled
    self._policy = policy or HistoryPolicy()

register_hooks

register_hooks(registry, **kwargs)

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

The capture is asynchronous only for a policy that summarises, since only then does it call a model.

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.

    The capture is asynchronous only for a policy that summarises, since
    only then does it call a model.
    """
    registry.add_callback(BeforeInvocationEvent, self._on_before)
    if self._policy.keep == "masked" and self._policy.summarize_over is not None:
        registry.add_callback(AfterInvocationEvent, self._on_after_summarizing)
    else:
        registry.add_callback(AfterInvocationEvent, self._on_after)

HistoryPolicy dataclass

HistoryPolicy(
    keep="full",
    min_chars=DEFAULT_MIN_CHARS,
    keep_tools=frozenset(),
    summarize_over=None,
    summarizer=None,
)

How a :class:HistoryHook shapes the transcript it writes back.

Attributes:

Name	Type	Description
keep	Literal['full', 'masked']	"full" writes the transcript back as captured. "masked" masks every finished call first, and masks a transcript stored in full when the agent loads it (history.keep).
min_chars	int	Tool results shorter than this survive masking.
keep_tools	frozenset[str]	Tools whose results survive masking. A delegate's always do.
summarize_over	int | None	Estimated tokens a masked copy may reach before its oldest calls are summarised. None never summarises.
summarizer	Model | None	The model that writes the summary.

GateRule dataclass

GateRule(pattern, remember=None, subject=None)

Which tools a gate covers, and how far a decision about them may reach.

Attributes:

Name	Type	Description
pattern	str	Tool name or fnmatch glob (merchant___*).
remember	str | None	Widest scope for these tools, or None for the hook's.
subject	SubjectFn | None	Callable (tool_name, tool_input) -> {key, label} | str | None deciding what a remembered decision covers, or None to cover the whole tool. Returning None means no approval needed.

InterruptHook

InterruptHook(
    tools,
    *,
    agent_name="",
    ttl_seconds=None,
    skip_read_only=False,
    remember="once",
)

Bases: HookProvider

Gate tool calls on the person's approval via strands interrupts.

When the agent calls a tool matching one of tools, this hook raises an approval request (event.interrupt()) describing the call, the subject a remembered decision would cover and the scopes the person may choose. The call runs only when the answer is approved; anything else — a denial, a skip, an expired request — cancels it with a message the model can read.

A permission the host remembers (forwardedProps.kaboo_grants) for the same tool and subject lets the call run without asking. Scopes wider than once are offered only when the host remembers decisions and the run publishes its open requests to the host.

An approval may carry an edited tool_input (AG-UI approveWithEdits): the call then runs with the person's arguments instead of the agent's.

Configure which tool calls wait for approval.

Parameters:

Name	Type	Description	Default
tools	list[str | GateRule]	Tool names, globs or :class:GateRule objects. The first matching entry applies.	required
agent_name	str	The agent's name.	''
ttl_seconds	int | None	Optional deadline for each request. An answer arriving after it does not run the call.	None
skip_read_only	bool	Let tools annotated readOnlyHint: true run without asking.	False
remember	str	Default widest scope: once, thread or always.	'once'

Raises:

Type	Description
ValueError	If a scope is not one of :data:APPROVAL_SCOPES.

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

def __init__(
    self,
    tools: list[str | GateRule],
    *,
    agent_name: str = "",
    ttl_seconds: int | None = None,
    skip_read_only: bool = False,
    remember: str = "once",
) -> None:
    """Configure which tool calls wait for approval.

    Args:
        tools: Tool names, globs or :class:`GateRule` objects. The first
            matching entry applies.
        agent_name: The agent's name.
        ttl_seconds: Optional deadline for each request. An answer arriving
            after it does not run the call.
        skip_read_only: Let tools annotated ``readOnlyHint: true`` run
            without asking.
        remember: Default widest scope: ``once``, ``thread`` or ``always``.

    Raises:
        ValueError: If a scope is not one of :data:`APPROVAL_SCOPES`.
    """
    self._rules = [t if isinstance(t, GateRule) else GateRule(pattern=t) for t in tools]
    for scope in [remember, *(r.remember for r in self._rules if r.remember is not None)]:
        if scope not in APPROVAL_SCOPES:
            raise ValueError(
                f"Unknown approval scope {scope!r}; use one of {list(APPROVAL_SCOPES)}."
            )
    self._agent_name = agent_name
    self._ttl_seconds = ttl_seconds
    self._skip_read_only = skip_read_only
    self._remember = remember

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,
    stamp_actor=False,
)

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
stamp_actor	bool	When True, kaboo/actor is set to {"agent", "thread", "run"} for the calling agent unless the provider already supplied one.	False

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

def __init__(
    self,
    provider: Any | None = None,
    *,
    stamp_tool_call_id: bool = True,
    stamp_actor: bool = False,
) -> 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
    self._stamp_actor = stamp_actor

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 is registered only for inline agents; media the model fetches with fetch_attachment comes back in that tool's result, so it needs no injection.

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 is registered only for ``inline`` agents; media
    the model fetches with ``fetch_attachment`` comes back in that tool's
    result, so it needs no injection.
    """
    registry.add_callback(BeforeInvocationEvent, self._on_before_invocation)
    if self._inline:
        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.

StateEffectCall dataclass

StateEffectCall(tool, input, result, agent, run_id, state)

What a state effect's mapper receives about the call that succeeded.

Attributes:

Name	Type	Description
tool	str	The tool's name.
input	Any	The call's arguments.
result	Any	The tool's result: its text parsed as JSON when it is JSON, otherwise the text.
agent	str	The agent that made the call.
run_id	str	The run the call belongs to.
state	dict[str, Any]	The shared state before this effect, to read and never modify.

StateEffectsHook

StateEffectsHook(*, agent_name='')

Bases: HookProvider

Apply the run's state effects after each successful tool call.

Initialize the hook.

Parameters:

Name	Type	Description	Default
agent_name	str	The agent's name, handed to mappers.	''

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

def __init__(self, *, agent_name: str = "") -> None:
    """Initialize the hook.

    Args:
        agent_name: The agent's name, handed to mappers.
    """
    self._agent_name = agent_name

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)

ToolApprovalHook

ToolApprovalHook(*, agent_name='')

Bases: HookProvider

Let tools ask the person to approve a call.

Wraps each tool the agent calls. When a tool replies with kaboo_approval instead of a result, the run pauses with an approval request: the same one the gate raises, plus the tool's request ID. The reply never reaches the model. On resume:

• Approved: the tool is called once more, with the person's edits if any, and its result is returned. If it asks again, the model reads that the call still needs approval, so the agent cannot loop.
• Denied, skipped or expired: the tool is not called again, and the model reads the same note as for a gated call.

The host must tell the tool's service about the decision before the run continues (kaboo-runtime's onDecision). The service, not the agent, decides whether the second call runs.

Configure the hook.

Parameters:

Name	Type	Description	Default
agent_name	str	Agent name, used to name the approval request.	''

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

def __init__(self, *, agent_name: str = "") -> None:
    """Configure the hook.

    Args:
        agent_name: Agent name, used to name the approval request.
    """
    self._agent_name = agent_name

ToolAttachmentsHook

Bases: HookProvider

Unpack attachments in a tool result into media blocks of that result.

Registered on every agent the config builds; a result without the field is left untouched. An MCP result's structuredContent copy loses the field too, so the base64 is not kept beside the media.

register_hooks

register_hooks(registry, **kwargs)

Register the after-tool-call callback.

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

@override
def register_hooks(self, registry: HookRegistry, **kwargs: Any) -> None:
    """Register the after-tool-call callback."""
    registry.add_callback(AfterToolCallEvent, self._unpack)

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 this run resumes.

Returns whether anything was restored. Only a resume restores: a fresh run's message supersedes any gate the state still holds, and a host that replays an older snapshot can hand over a gate already answered. A run restores at most once, 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 on every invocation without either clobbering the other, or a later invocation bringing back the gate the run has since answered.

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 this run resumes.

    Returns whether anything was restored. Only a resume restores: a fresh run's
    message supersedes any gate the state still holds, and a host that replays an
    older snapshot can hand over a gate already answered. A run restores at most
    once, 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 on every invocation without either clobbering the other, or
    a later invocation bringing back the gate the run has since answered.
    """
    exchange = get_session_exchange()
    if exchange is None:
        return False
    exchange.agent = agent
    exchange.publishes_approvals = True
    if not exchange.resuming or exchange.restored:
        return False
    state = exchange.inbound.get("interrupt_state")
    if not _is_interrupt_state(state):
        return False
    exchange.restored = True
    # 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("paused session restore failed; starting with clean state")
        return False
    logger.debug(
        "paused 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("paused session capture failed; state 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
    snapshot: dict[str, Any] = {"interrupt_state": serialized}
    if exchange.delegates:
        snapshot["delegates"] = exchange.delegates
    return snapshot

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