Skip to content

kaboo_workflows

The top-level package. See API reference overview for the curated map of the public surface grouped by import path.

kaboo_workflows

kaboo-workflows — YAML-driven multi-agent orchestration with AG-UI/CopilotKit support.

Principal dataclass 

Principal(token=None, claims=dict(), headers=dict())

Authenticated caller identity for the current request.

Produced by an inbound auth verifier (see :func:~kaboo_workflows.adapters.agui.create_agui_app) and bound to the request context so outbound MCP auth strategies (relay / OBO) can derive a downstream token from the same identity.

Attributes:

Name Type Description
token str | None

The raw inbound bearer credential (no Bearer prefix), e.g. the user JWT or an AgentCore WorkloadAccessToken. None when the caller was not authenticated (or auth is disabled).
claims dict[str, Any]

Verified claims (decoded JWT payload, introspection result, …). Empty when the verifier only relays without validating.
headers dict[str, str]

Selected inbound headers the verifier chose to carry forward (e.g. WorkloadAccessToken), so strategies can read a token from a non-Authorization header when needed.

M2MClientCredentialsAuth 

M2MClientCredentialsAuth(
    *,
    token_url,
    client_id,
    client_secret,
    scope=None,
    audience=None,
    extra=None,
    header="Authorization",
    scheme="Bearer",
)

Bases: _BearerAuth

OAuth2 client-credentials (machine-to-machine) token for MCP calls.

Fetches a token from token_url using the client-credentials grant and caches it until shortly before expiry. Independent of the inbound caller — represents the workflow service's own machine identity, so it is safe on a long-lived shared MCP client.

Parameters:

Name Type Description Default
token_url str

OAuth2 token endpoint.

 required
client_id str

Client identifier.

 required
client_secret str

Client secret.

 required
scope str | None

Optional space-delimited scopes.

 None
audience str | None

Optional audience parameter (e.g. Auth0).

 None
extra dict[str, str] | None

Extra form fields to include in the token request.

 None
header str | Sequence[str]

Header to set, or several (default Authorization).

 'Authorization'
scheme str

Auth scheme prefix (default Bearer).

 'Bearer'
Source code in src/kaboo_workflows/auth/strategies.py 
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
def __init__(
    self,
    *,
    token_url: str,
    client_id: str,
    client_secret: str,
    scope: str | None = None,
    audience: str | None = None,
    extra: dict[str, str] | None = None,
    header: str | Sequence[str] = "Authorization",
    scheme: str = "Bearer",
) -> None:
    self._token_url = token_url
    self._client_id = client_id
    self._client_secret = client_secret
    self._scope = scope
    self._audience = audience
    self._extra = extra or {}
    self.header_names = _header_names(header)
    self.scheme = scheme
    self._cache: tuple[str, float] | None = None
    self._lock = threading.Lock()

OBOTokenAuth 

OBOTokenAuth(
    *,
    provider,
    region="us-east-1",
    scopes=None,
    workload_name=None,
    workload_token=None,
    custom_parameters=None,
    force_authentication=False,
    header="Authorization",
    scheme="Bearer",
)

Bases: _BearerAuth

AgentCore On-Behalf-Of token exchange for a downstream resource.

Two exchanges happen, and both run inside AgentCore Identity:

  1. The inbound end-user token is exchanged for a workload access token (GetWorkloadAccessTokenForJWT), which is what binds this agent's workload identity to that user. Set workload_name to perform it here; leave it unset when the runtime already hands the workload token in on the request, in which case it is read from the principal.
  2. The workload access token is exchanged for a downstream access token (GetResourceOauth2Token) against the named credential provider.

Downstream tokens are cached per workload token until shortly before expiry.

Provider differences belong in configuration, not in this class: custom_parameters is passed through to the exchange, which is how an Entra ID provider gets its requested_token_use=on_behalf_of, and header / scheme decide how the result is presented downstream.

Parameters:

Name Type Description Default
provider str

Name of the AgentCore OAuth2 credential provider (the downstream resource) to exchange for.

 required
region str

AWS region of the AgentCore control plane.

 'us-east-1'
scopes list[str] | None

OAuth2 scopes to request. The API requires the field, so an empty list is sent when none are given.

 None
workload_name str | None

Workload identity to mint a workload access token for from the inbound user token. Unset means the inbound request is expected to carry the workload token already.

 None
workload_token str | None

Explicit workload token (bypasses the principal).

 None
custom_parameters dict[str, str] | None

Extra provider-specific parameters forwarded to the token exchange.

 None
force_authentication bool

Skip AgentCore's cached token for this identity.

 False
header str | Sequence[str]

Header to set, or several (default Authorization).

 'Authorization'
scheme str

Auth scheme prefix (default Bearer).

 'Bearer'
Source code in src/kaboo_workflows/auth/strategies.py 
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
def __init__(
    self,
    *,
    provider: str,
    region: str = "us-east-1",
    scopes: list[str] | None = None,
    workload_name: str | None = None,
    workload_token: str | None = None,
    custom_parameters: dict[str, str] | None = None,
    force_authentication: bool = False,
    header: str | Sequence[str] = "Authorization",
    scheme: str = "Bearer",
) -> None:
    self._provider = provider
    self._region = region
    self._scopes = scopes or []
    self._workload_name = workload_name
    self._explicit_workload = workload_token
    self._custom_parameters = custom_parameters or {}
    self._force = force_authentication
    self.header_names = _header_names(header)
    self.scheme = scheme
    self._client: Any = None
    self._cache: dict[str, tuple[str, float]] = {}
    self._workload_cache: dict[str, str] = {}
    self._lock = threading.Lock()

RelayTokenAuth 

RelayTokenAuth(
    *, token=None, header="Authorization", scheme="Bearer"
)

Bases: _BearerAuth

Forward the inbound caller token to the MCP unchanged.

Reads the current :class:~kaboo_workflows._context.Principal token unless an explicit token is given (useful when binding a per-request client with a captured token).

Parameters:

Name Type Description Default
token str | None

Explicit token to forward. When None, the inbound principal's token is used at call time.

 None
header str | Sequence[str]

Header to set, or several (default Authorization). A managed gateway that authenticates on Authorization and forwards its own credential needs both it and the header the target reads.

 'Authorization'
scheme str

Auth scheme prefix (default Bearer; "" for a raw value).

 'Bearer'
Source code in src/kaboo_workflows/auth/strategies.py 
110
111
112
113
114
115
116
117
118
119
def __init__(
    self,
    *,
    token: str | None = None,
    header: str | Sequence[str] = "Authorization",
    scheme: str = "Bearer",
) -> None:
    self._explicit = token
    self.header_names = _header_names(header)
    self.scheme = scheme

StaticTokenAuth 

StaticTokenAuth(
    *, token, header="Authorization", scheme="Bearer"
)

Bases: _BearerAuth

Attach a fixed token (e.g. a long-lived API key) to every MCP request.

Parameters:

Name Type Description Default
token str

The credential to send.

 required
header str | Sequence[str]

Header to set, or several (default Authorization).

 'Authorization'
scheme str

Auth scheme prefix (default Bearer; "" for a raw value).

 'Bearer'
Source code in src/kaboo_workflows/auth/strategies.py 
137
138
139
140
141
142
143
144
145
146
def __init__(
    self,
    *,
    token: str,
    header: str | Sequence[str] = "Authorization",
    scheme: str = "Bearer",
) -> None:
    self._value = token
    self.header_names = _header_names(header)
    self.scheme = scheme

AppConfig

Bases: BaseModel

Root YAML configuration.

Orchestrations are defined as a dict of named orchestration blocks that can reference each other for arbitrary nesting.

version class-attribute instance-attribute 

version = '1'

Schema version — omit to use the default "1".

history class-attribute instance-attribute 

history = False

Global default for per-agent history:.

Applies to any agent that does not set its own history:. Defaults to False so sub-agents are stateless per run unless opted in. The entry agent is unaffected — its transcript is always the CopilotKit chat.

attachments class-attribute instance-attribute 

attachments = Field(default_factory=AttachmentsDef)

Global reference/attachment policy (manifest + optional resolver tool).

runtime class-attribute instance-attribute 

runtime = Field(default_factory=RuntimeDef)

Runtime behavior toggles (per-invocation agent overrides, …).

telemetry class-attribute instance-attribute 

telemetry = Field(default_factory=TelemetryDef)

Opt-in OpenTelemetry tracing (see :class:TelemetryDef).

ResolvedConfig dataclass 

ResolvedConfig(
    *,
    agents=dict(),
    orchestrators=dict(),
    entry,
    mcp_lifecycle=MCPLifecycle(),
    app_config=None,
)

Fully resolved config — lifecycle started, agents ready.

After calling :func:~kaboo_workflows.config.loaders.load, use :meth:wire_event_queue to set up event streaming::

resolved = load("config.yaml")
event_queue = resolved.wire_event_queue()

wire_event_queue 

wire_event_queue(*, session_id=None, tool_labels=None)

Wire all agents and orchestrators for event streaming.

This is the recommended way to set up event streaming. It:

  1. Builds a :class:~kaboo_workflows.types.SessionManifest from the resolved runtime objects.
  2. Auto-resolves stream groups from the original AppConfig so every agent is attributed to a hierarchical dot-path group.
  3. Collects tool labels from MCP clients and agents in the original AppConfig (agent-level overrides MCP-level).
  4. Wires every agent (and orchestrator) with an :class:~kaboo_workflows.hooks.EventPublisher via :func:~kaboo_workflows.wire.make_event_queue.
  5. Emits a SESSION_START event carrying the manifest as the first event on the queue.

.. warning::

This **mutates** the agents and orchestrators stored on this
instance by adding hooks and overwriting ``callback_handler``.
Call it only once per ``ResolvedConfig`` instance.

Parameters:

Name Type Description Default
session_id str | None

Optional session ID to embed in events.

 None
tool_labels dict[str, str] | None

Optional runtime tool name → display label mapping. These take highest priority, overriding both agent-level and MCP-client-level labels from YAML config.

 None

Returns:

Type Description
EventQueue

A ready-to-use :class:~kaboo_workflows.wire.EventQueue with
EventQueue

SESSION_START already on it.

Raises:

Type Description
ValueError

If the entry node cannot be resolved by object identity.
Source code in src/kaboo_workflows/config/resolvers/config.py 
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def wire_event_queue(
    self,
    *,
    session_id: str | None = None,
    tool_labels: dict[str, str] | None = None,
) -> EventQueue:
    """Wire all agents and orchestrators for event streaming.

    This is the recommended way to set up event streaming.  It:

    1. Builds a :class:`~kaboo_workflows.types.SessionManifest` from the
       resolved runtime objects.
    2. Auto-resolves stream groups from the original AppConfig so every
       agent is attributed to a hierarchical dot-path group.
    3. Collects tool labels from MCP clients and agents in the original
       AppConfig (agent-level overrides MCP-level).
    4. Wires every agent (and orchestrator) with an
       :class:`~kaboo_workflows.hooks.EventPublisher` via
       :func:`~kaboo_workflows.wire.make_event_queue`.
    5. Emits a SESSION_START event carrying the manifest as the first
       event on the queue.

    .. warning::

        This **mutates** the agents and orchestrators stored on this
        instance by adding hooks and overwriting ``callback_handler``.
        Call it only once per ``ResolvedConfig`` instance.

    Args:
        session_id: Optional session ID to embed in events.
        tool_labels: Optional runtime tool name → display label mapping.
            These take highest priority, overriding both agent-level and
            MCP-client-level labels from YAML config.

    Returns:
        A ready-to-use :class:`~kaboo_workflows.wire.EventQueue` with
        SESSION_START already on it.

    Raises:
        ValueError: If the entry node cannot be resolved by object identity.
    """
    manifest = build_manifest(self.agents, self.orchestrators, self.entry)
    stream_groups = _auto_resolve_stream_groups(self.app_config)
    history = _resolve_history(self.app_config)
    chat_owner = _resolve_chat_owner(self.app_config)
    chat_reply = chat_owner or _resolve_chat_output(self.app_config)
    merged_labels = _collect_tool_labels(self.app_config)
    if tool_labels:
        merged_labels.update(tool_labels)
    event_queue = make_event_queue(
        self.agents,
        orchestrators=self.orchestrators,
        tool_labels=merged_labels or None,
        stream_groups=stream_groups,
        history=history,
        entry_name=manifest.entry.name,
        chat_owner=chat_owner,
        chat_reply=chat_reply,
        session_id=session_id,
    )
    event_queue.emit_session_start(manifest)
    return event_queue

ResolvedInfra dataclass 

ResolvedInfra(
    models=dict(),
    clients=dict(),
    mcp_lifecycle=MCPLifecycle(),
)

Infrastructure resolved from config — lifecycle NOT started.

This is the pure result of :func:resolve_infra. Lifecycle is cold, agents are not yet created.

Session managers are NOT stored here — they are built per agent and per orchestration at session time, from config.session_manager (the global def) plus effective_session_id computed by load_session.

Use :func:~kaboo_workflows.config.loaders.load for a fully activated system, or manually::

infra = resolve_infra(config)
infra.mcp_lifecycle.start()
agents = resolve_agents(agent_defs=config.agents, ...)

OrchestrationBuilder 

OrchestrationBuilder(
    configs,
    agents,
    agent_defs,
    models,
    mcp_clients,
    *,
    global_session_manager_def=None,
    session_id=None,
    attachments=None,
)

Builds all named orchestrations in dependency order.

Initialize the OrchestrationBuilder.

Orchestrations are sorted topologically so that each orchestration's dependencies (other orchestrations it references) are built first. The node pool grows as orchestrations are built, making earlier results available to downstream configurations.

Parameters:

Name Type Description Default
configs Mapping[str, OrchestrationDef]

Orchestration definitions keyed by name.

 required
agents Mapping[str, Agent]

Already-resolved agents keyed by name.

 required
agent_defs dict[str, AgentDef]

Agent schema definitions for delegate forking.

 required
models dict[str, Model]

Resolved model objects keyed by name.

 required
mcp_clients dict[str, MCPClient]

Resolved MCP client objects keyed by name.

 required
global_session_manager_def SessionManagerDef | None

Global session manager def from AppConfig.session_manager, used as a fallback when an orchestration declares no session_manager: of its own and has not explicitly opted out.

 None
session_id str | None

Effective session id threaded down from load_session. Passed as session_id_override to every resolve_session_manager call made by leaf builders.

 None
attachments AttachmentsDef | None

Global reference/attachment policy, forwarded to the forked delegate manager so it matches declared agents.

 None
Source code in src/kaboo_workflows/config/resolvers/orchestrations/builders.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def __init__(
    self,
    configs: Mapping[str, OrchestrationDef],
    agents: Mapping[str, Agent],
    agent_defs: dict[str, AgentDef],
    models: dict[str, Model],
    mcp_clients: dict[str, StrandsMCPClient],
    *,
    global_session_manager_def: SessionManagerDef | None = None,
    session_id: str | None = None,
    attachments: AttachmentsDef | None = None,
) -> None:
    """Initialize the OrchestrationBuilder.

    Orchestrations are sorted topologically so that each orchestration's
    dependencies (other orchestrations it references) are built first.
    The node pool grows as orchestrations are built, making earlier results
    available to downstream configurations.

    Args:
        configs: Orchestration definitions keyed by name.
        agents: Already-resolved agents keyed by name.
        agent_defs: Agent schema definitions for delegate forking.
        models: Resolved model objects keyed by name.
        mcp_clients: Resolved MCP client objects keyed by name.
        global_session_manager_def: Global session manager def from
            ``AppConfig.session_manager``, used as a fallback when an
            orchestration declares no ``session_manager:`` of its own and
            has not explicitly opted out.
        session_id: Effective session id threaded down from ``load_session``.
            Passed as ``session_id_override`` to every
            ``resolve_session_manager`` call made by leaf builders.
        attachments: Global reference/attachment policy, forwarded to the
            forked delegate manager so it matches declared agents.
    """
    self._configs = configs
    self._global_sm_def = global_session_manager_def
    self._session_id = session_id
    self._attachments = attachments
    self._nodes: dict[str, Node] = dict(agents)
    self._built: dict[str, Node] = {}
    self._agent_defs = agent_defs
    self._models = models
    self._mcp_clients = mcp_clients

build_all 

build_all()

Build all orchestrations in topological order.

Returns:

Type Description
dict[str, Node]

Dict of orchestration name -> built orchestration (Swarm | Graph | Agent).
Source code in src/kaboo_workflows/config/resolvers/orchestrations/builders.py 
91
92
93
94
95
96
97
98
99
def build_all(self) -> dict[str, Node]:
    """Build all orchestrations in topological order.

    Returns:
        Dict of orchestration name -> built orchestration (Swarm | Graph | Agent).
    """
    for name in topological_sort(self._configs):
        self._build_one(name)
    return self._built

CircularDependencyError

Bases: ConfigurationError

Raised when orchestration definitions contain a dependency cycle.

ConfigurationError

Bases: ValueError

Raised when a kaboo-workflows configuration is invalid.

Subclasses ValueError so callers that catch ValueError still work, but allows more specific except ConfigurationError handling.

ImportResolutionError

Bases: ConfigurationError

Raised when a custom import spec (module.path:Name) cannot be loaded.

SchemaValidationError

Bases: ConfigurationError

Raised when YAML config fails Pydantic schema validation.

UnresolvedReferenceError

Bases: ConfigurationError

Raised when a config references a non-existent model, agent, or MCP resource.

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 
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
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 
241
242
243
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 
264
265
266
267
268
269
270
271
272
273
274
275
276
277
@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 
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
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

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 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
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 
55
56
57
58
@override
def register_hooks(self, registry: HookRegistry, **kwargs: Any) -> None:
    """Register the BeforeToolCallEvent callback."""
    registry.add_callback(BeforeToolCallEvent, self._on_before_tool)

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 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
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 
45
46
47
48
@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 
93
94
95
96
97
@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)

MCPLifecycle 

MCPLifecycle(server_ready_timeout=30)

Manages MCP server and client lifecycle ordering.

Initialize the MCPLifecycle.

Ensures servers are fully ready before clients connect, and clients are stopped before servers on shutdown.

Example::

lifecycle = MCPLifecycle()
lifecycle.add_server("postgres", pg_server)
lifecycle.add_client("pg_client", pg_client)

with lifecycle:
    # All servers started and ready, all clients connected
    agent = Agent(tools=[lifecycle.get_client("pg_client")])
    agent("Query the database")

# All cleaned up

Or without context manager::

lifecycle.start()
try:
    ...
finally:
    lifecycle.stop()

Parameters:

Name Type Description Default
server_ready_timeout float

Seconds to wait for each server to become ready.

 30
Source code in src/kaboo_workflows/mcp/lifecycle.py 
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
def __init__(self, server_ready_timeout: float = 30) -> None:
    """Initialize the MCPLifecycle.

    Ensures servers are fully ready before clients connect, and clients
    are stopped before servers on shutdown.

    Example::

        lifecycle = MCPLifecycle()
        lifecycle.add_server("postgres", pg_server)
        lifecycle.add_client("pg_client", pg_client)

        with lifecycle:
            # All servers started and ready, all clients connected
            agent = Agent(tools=[lifecycle.get_client("pg_client")])
            agent("Query the database")

        # All cleaned up

    Or without context manager::

        lifecycle.start()
        try:
            ...
        finally:
            lifecycle.stop()

    Args:
        server_ready_timeout: Seconds to wait for each server to become ready.
    """
    self._servers: dict[str, MCPServer] = {}
    self._clients: dict[str, StrandsMCPClient] = {}
    self._server_ready_timeout = server_ready_timeout
    self._started = False
    self._consumer_token = f"kaboo-mcp-lifecycle-{id(self)}"

servers property 

servers

Read-only view of registered servers.

clients property 

clients

Read-only view of registered clients.

add_server 

add_server(name, server)

Register an MCP server.

Parameters:

Name Type Description Default
name str

Unique server identifier.

 required
server MCPServer

The MCP server instance.

 required

Raises:

Type Description
ValueError

If a server with this name is already registered.
Source code in src/kaboo_workflows/mcp/lifecycle.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
def add_server(self, name: str, server: MCPServer) -> None:
    """Register an MCP server.

    Args:
        name: Unique server identifier.
        server: The MCP server instance.

    Raises:
        ValueError: If a server with this name is already registered.
    """
    if name in self._servers:
        raise ValueError(f"MCP server '{name}' is already registered")
    self._servers[name] = server

add_client 

add_client(name, client)

Register an MCP client.

Parameters:

Name Type Description Default
name str

Unique client identifier.

 required
client MCPClient

The strands MCP client instance.

 required

Raises:

Type Description
ValueError

If a client with this name is already registered.
Source code in src/kaboo_workflows/mcp/lifecycle.py 
81
82
83
84
85
86
87
88
89
90
91
92
93
def add_client(self, name: str, client: StrandsMCPClient) -> None:
    """Register an MCP client.

    Args:
        name: Unique client identifier.
        client: The strands MCP client instance.

    Raises:
        ValueError: If a client with this name is already registered.
    """
    if name in self._clients:
        raise ValueError(f"MCP client '{name}' is already registered")
    self._clients[name] = client

get_server 

get_server(name)

Get a registered server by name.

Parameters:

Name Type Description Default
name str

Server identifier.

 required

Returns:

Type Description
MCPServer

The registered MCP server.

Raises:

Type Description
KeyError

If no server with this name is registered.
Source code in src/kaboo_workflows/mcp/lifecycle.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def get_server(self, name: str) -> MCPServer:
    """Get a registered server by name.

    Args:
        name: Server identifier.

    Returns:
        The registered MCP server.

    Raises:
        KeyError: If no server with this name is registered.
    """
    if name not in self._servers:
        raise KeyError(f"MCP server '{name}' not registered.\nAvailable: {list(self._servers)}")
    return self._servers[name]

get_client 

get_client(name)

Get a registered client by name.

Parameters:

Name Type Description Default
name str

Client identifier.

 required

Returns:

Type Description
MCPClient

The registered strands MCP client.

Raises:

Type Description
KeyError

If no client with this name is registered.
Source code in src/kaboo_workflows/mcp/lifecycle.py 
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def get_client(self, name: str) -> StrandsMCPClient:
    """Get a registered client by name.

    Args:
        name: Client identifier.

    Returns:
        The registered strands MCP client.

    Raises:
        KeyError: If no client with this name is registered.
    """
    if name not in self._clients:
        raise KeyError(f"MCP client '{name}' not registered.\nAvailable: {list(self._clients)}")
    return self._clients[name]

start 

start(*, pin_clients=True)

Start all servers and wait for readiness.

Idempotent: if already started, returns immediately. load() calls this before creating agents (so MCP clients can connect), and the context manager calls it again on enter — the second call is a no-op. The context manager is still needed for graceful shutdown via stop().

Clients are not started here — strands automatically starts MCPClient instances when they are registered as tool providers on an Agent. Starting them here would cause a "session is currently running" error when the Agent tries to start them again. Instead we pin each client with a permanent lifecycle consumer so its session is not torn down when consuming Agents are garbage-collected.

Parameters:

Name Type Description Default
pin_clients bool

Whether to pin client sessions for the lifetime of this lifecycle. Pass False when clients are owned per run instead of per process: a pinned session outlives the agents that used it, which is exactly what a per-run client must not do.

 True

Raises:

Type Description
RuntimeError

If any server fails to start or become ready.
Source code in src/kaboo_workflows/mcp/lifecycle.py 
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
def start(self, *, pin_clients: bool = True) -> None:
    """Start all servers and wait for readiness.

    **Idempotent**: if already started, returns immediately.
    ``load()`` calls this before creating agents (so MCP clients can
    connect), and the context manager calls it again on enter — the
    second call is a no-op.  The context manager is still needed for
    **graceful shutdown** via ``stop()``.

    Clients are **not** started here — strands automatically starts
    MCPClient instances when they are registered as tool providers
    on an Agent. Starting them here would cause a "session is currently
    running" error when the Agent tries to start them again. Instead we
    pin each client with a permanent lifecycle consumer so its session
    is not torn down when consuming Agents are garbage-collected.

    Args:
        pin_clients: Whether to pin client sessions for the lifetime of this
            lifecycle. Pass ``False`` when clients are owned per run instead
            of per process: a pinned session outlives the agents that used
            it, which is exactly what a per-run client must not do.

    Raises:
        RuntimeError: If any server fails to start or become ready.
    """
    if self._started:
        return

    # Phase 1: Start all servers
    for name, server in self._servers.items():
        logger.info("server=<%s> | starting MCP server", name)
        server.start()

    # Phase 2: Wait for all servers to be ready
    for name, server in self._servers.items():
        if not server.wait_ready(timeout=self._server_ready_timeout):
            raise RuntimeError(
                f"MCP server '{name}' did not become ready within {self._server_ready_timeout}s"
            )
        logger.info("server=<%s> | MCP server is ready", name)

    # Phase 3: pin every client with a permanent consumer so its session
    # survives agent garbage collection (esp. across an interrupt/resume
    # boundary). Strands reference-counts clients per consuming Agent and
    # tears the session down when the count hits zero; stop() clears it.
    if pin_clients:
        for name, client in self._clients.items():
            add_consumer = getattr(client, "add_consumer", None)
            if callable(add_consumer):
                add_consumer(self._consumer_token)
                logger.debug("client=<%s> | pinned lifecycle consumer", name)

    self._started = True

stop 

stop()

Stop all clients first, then all servers.

Clients that were never started (e.g., never registered on an Agent) are skipped gracefully.

Source code in src/kaboo_workflows/mcp/lifecycle.py 
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
def stop(self) -> None:
    """Stop all clients first, then all servers.

    Clients that were never started (e.g., never registered on an Agent)
    are skipped gracefully.
    """
    if not self._started:
        return

    # Phase 1: Stop all clients
    for name, client in self._clients.items():
        try:
            # Normal shutdown — no exception context (matches __exit__ protocol)
            client.stop(exc_type=None, exc_val=None, exc_tb=None)
            logger.info("client=<%s> | MCP client stopped", name)
        except Exception:
            logger.warning("client=<%s> | failed to stop MCP client", name, exc_info=True)

    # Phase 2: Stop all servers
    for name, server in self._servers.items():
        try:
            server.stop()
            logger.info("server=<%s> | MCP server stopped", name)
        except Exception:
            logger.warning("server=<%s> | failed to stop MCP server", name, exc_info=True)

    self._started = False

__enter__ 

__enter__()

Start lifecycle on context entry.

Source code in src/kaboo_workflows/mcp/lifecycle.py 
209
210
211
212
def __enter__(self) -> MCPLifecycle:
    """Start lifecycle on context entry."""
    self.start()
    return self

__exit__ 

__exit__(exc_type, exc_val, exc_tb)

Stop lifecycle on context exit.

Source code in src/kaboo_workflows/mcp/lifecycle.py 
214
215
216
217
218
219
220
221
def __exit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> None:
    """Stop lifecycle on context exit."""
    self.stop()

__aenter__ async 

__aenter__()

Async context entry — delegates to sync :meth:start.

Useful with Starlette / ASGI lifespan::

@asynccontextmanager
async def lifespan(app):
    async with lifecycle:
        yield
Source code in src/kaboo_workflows/mcp/lifecycle.py 
223
224
225
226
227
228
229
230
231
232
233
234
async def __aenter__(self) -> MCPLifecycle:
    """Async context entry — delegates to sync :meth:`start`.

    Useful with Starlette / ASGI lifespan::

        @asynccontextmanager
        async def lifespan(app):
            async with lifecycle:
                yield
    """
    self.start()
    return self

__aexit__ async 

__aexit__(exc_type, exc_val, exc_tb)

Async context exit — delegates to sync :meth:stop.

Source code in src/kaboo_workflows/mcp/lifecycle.py 
236
237
238
239
240
241
242
243
async def __aexit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> None:
    """Async context exit — delegates to sync :meth:`stop`."""
    self.stop()

AnsiRenderer 

AnsiRenderer(
    *, file=None, separator_width=None, typewriter_delay=0.0
)

Bases: EventRenderer

Renders events using raw ANSI escape codes.

Initialize the AnsiRenderer.

No third-party dependencies. Colour codes are automatically suppressed when stdout is not a TTY (piped / redirected output).

Token and reasoning events are written inline (no trailing newline) so they appear as a continuous stream. A separator line is printed when the mode changes between reasoning and responding, or when the active agent changes.

Parameters:

Name Type Description Default
file Any | None

Output stream (defaults to sys.stdout).

 None
separator_width int | None

Width of separator lines. Defaults to 70 or the current terminal width, whichever is available.

 None
typewriter_delay float

Seconds to sleep after each printable character in TOKEN and REASONING events. 0.0 (default) disables the effect entirely with no overhead.

 0.0
Source code in src/kaboo_workflows/renderers/ansi.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def __init__(
    self,
    *,
    file: Any | None = None,
    separator_width: int | None = None,
    typewriter_delay: float = 0.0,
) -> None:
    """Initialize the AnsiRenderer.

    No third-party dependencies.  Colour codes are automatically
    suppressed when stdout is not a TTY (piped / redirected output).

    Token and reasoning events are written inline (no trailing newline)
    so they appear as a continuous stream.  A separator line is printed
    when the mode changes between reasoning and responding, or when the
    active agent changes.

    Args:
        file: Output stream (defaults to ``sys.stdout``).
        separator_width: Width of separator lines. Defaults to 70 or the
            current terminal width, whichever is available.
        typewriter_delay: Seconds to sleep after each printable character
            in TOKEN and REASONING events.  ``0.0`` (default) disables
            the effect entirely with no overhead.
    """
    self._out = file or sys.stdout
    self._in_stream = False
    self._mode: str | None = None  # "reasoning" | "responding" | None
    self._active_agent: str | None = None
    self._typewriter_delay = typewriter_delay

    # Cache TTY state and pre-compute ANSI escape strings.
    is_tty = hasattr(self._out, "isatty") and self._out.isatty()
    self._dim = "\033[2m" if is_tty else ""
    self._bold = "\033[1m" if is_tty else ""
    self._cyan = "\033[36m" if is_tty else ""
    self._green = "\033[32m" if is_tty else ""
    self._red = "\033[31m" if is_tty else ""
    self._yellow = "\033[33m" if is_tty else ""
    self._magenta = "\033[95m" if is_tty else ""
    self._reset = "\033[0m" if is_tty else ""

    if separator_width is not None:
        self._separator_width = separator_width
    else:
        self._separator_width = shutil.get_terminal_size((70, 24)).columns

    self._handlers: dict[str, Callable[[StreamEvent], None]] = {
        EventType.SESSION_START: self._handle_session_start,
        EventType.SESSION_END: self._handle_session_end,
        EventType.TOKEN: self._handle_token,
        EventType.AGENT_START: self._handle_agent_start,
        EventType.TOOL_START: self._handle_tool_start,
        EventType.TOOL_END: self._handle_tool_end,
        EventType.AGENT_COMPLETE: self._handle_agent_complete,
        EventType.ERROR: self._handle_error,
        EventType.NODE_START: self._handle_node_start,
        EventType.NODE_STOP: self._handle_node_stop,
        EventType.HANDOFF: self._handle_handoff,
        EventType.MULTIAGENT_START: self._handle_multiagent_start,
        EventType.MULTIAGENT_COMPLETE: self._handle_multiagent_complete,
        EventType.REASONING: self._handle_reasoning,
    }

EventType

Bases: StrEnum

Event type constants for :class:StreamEvent.

StrEnum values are plain strings, so == "token" comparisons work unchanged.

StreamEvent dataclass 

StreamEvent(
    type,
    agent_name,
    timestamp=(lambda: datetime.now(tz=(timezone.utc)))(),
    data=dict(),
)

A typed event from agent or multi-agent execution.

Per-agent activity (AGENT_START, TOKEN, REASONING, TOOL_START, TOOL_END, INTERRUPT, AGENT_COMPLETE, ERROR, NODE_START, NODE_STOP, HANDOFF, MULTIAGENT_START, MULTIAGENT_COMPLETE) is produced by :class:~kaboo_workflows.hooks.EventPublisher. Session-level events (SESSION_START, SESSION_END) are produced by the queue/wiring layer in :mod:kaboo_workflows.wire.

Attributes:

Name Type Description
type str

Event type identifier (one of the :class:EventType values).
agent_name str

Name of the agent or session entry point that produced this event.
timestamp datetime

When the event occurred.
data dict[str, Any]

Event-specific payload.

asdict 

asdict()

Convert this StreamEvent to a flat dict for serialization.

Source code in src/kaboo_workflows/types.py 
88
89
90
91
92
def asdict(self) -> dict[str, Any]:
    """Convert this StreamEvent to a flat dict for serialization."""
    data = asdict(self)
    data["timestamp"] = self.timestamp.isoformat()
    return data

from_dict classmethod 

from_dict(data)

Deserialize a dict into a StreamEvent.

Restores all fields produced by :meth:asdict, including the timestamp (parsed from its ISO-8601 string representation).

Parameters:

Name Type Description Default
data dict[str, Any]

A dict as produced by :meth:asdict.

 required

Returns:

Type Description
StreamEvent

A new StreamEvent instance.
Source code in src/kaboo_workflows/types.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
@classmethod
def from_dict(cls, data: dict[str, Any]) -> StreamEvent:
    """Deserialize a dict into a StreamEvent.

    Restores all fields produced by :meth:`asdict`, including the
    ``timestamp`` (parsed from its ISO-8601 string representation).

    Args:
        data: A dict as produced by :meth:`asdict`.

    Returns:
        A new StreamEvent instance.
    """
    ts_raw = data.get("timestamp")
    if isinstance(ts_raw, str):
        ts = datetime.fromisoformat(ts_raw)
    elif isinstance(ts_raw, datetime):
        ts = ts_raw
    else:
        ts = datetime.now(tz=timezone.utc)

    return cls(
        type=data.get("type", ""),
        agent_name=data.get("agent_name", ""),
        timestamp=ts,
        data=data.get("data", {}),
    )

__eq__ 

__eq__(other)

Compare events by type, agent_name, and data (ignoring timestamp).

Source code in src/kaboo_workflows/types.py 
122
123
124
125
126
def __eq__(self, other: object) -> bool:
    """Compare events by type, agent_name, and data (ignoring timestamp)."""
    if not isinstance(other, StreamEvent):
        return NotImplemented
    return (self.type, self.agent_name, self.data) == (other.type, other.agent_name, other.data)

__hash__ 

__hash__()

Hash based on type and agent_name (data is unhashable).

Source code in src/kaboo_workflows/types.py 
128
129
130
def __hash__(self) -> int:
    """Hash based on type and agent_name (data is unhashable)."""
    return hash((self.type, self.agent_name))

EventQueue 

EventQueue(queue, *, entry_name=None, session_id=None)

Async event queue with hidden end-of-stream sentinel and session lifecycle.

Initialize the EventQueue.

Callers consume events via :meth:get (which returns None when the stream is closed) and signal completion via :meth:close. The sentinel is an implementation detail — user code never sees or owns it.

entry_name and session_id parameterise the SESSION_START and SESSION_END events emitted by :meth:emit_session_start and :meth:close respectively.

Parameters:

Name Type Description Default
queue Queue

The underlying asyncio.Queue to wrap.

 required
entry_name str | None

The configured name of the entry node. Used as agent_name on SESSION_START and SESSION_END events. Defaults to an empty string when not provided.

 None
session_id str | None

The effective session id. Included in the SESSION_END event payload as data["session_id"].

 None
Source code in src/kaboo_workflows/wire.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def __init__(
    self,
    queue: asyncio.Queue,
    *,
    entry_name: str | None = None,
    session_id: str | None = None,
) -> None:
    """Initialize the EventQueue.

    Callers consume events via :meth:`get` (which returns ``None`` when the
    stream is closed) and signal completion via :meth:`close`.  The sentinel
    is an implementation detail — user code never sees or owns it.

    *entry_name* and *session_id* parameterise the SESSION_START and
    SESSION_END events emitted by :meth:`emit_session_start` and
    :meth:`close` respectively.

    Args:
        queue: The underlying asyncio.Queue to wrap.
        entry_name: The configured name of the entry node.  Used as
            ``agent_name`` on SESSION_START and SESSION_END events.
            Defaults to an empty string when not provided.
        session_id: The effective session id.  Included in the
            SESSION_END event payload as ``data["session_id"]``.
    """
    self._queue = queue
    self._entry_name = entry_name or ""
    self._session_id = session_id
    self._session_start_emitted = False
    self._session_end_emitted = False

flush 

flush()

Discard all currently queued events and reset lifecycle guards.

Call this at the start of each request to clear any stale events left over from a previous invocation. Resets the SESSION_START and SESSION_END guards so the next invocation cycle can re-emit them.

Source code in src/kaboo_workflows/wire.py 
 98
 99
100
101
102
103
104
105
106
107
108
def flush(self) -> None:
    """Discard all currently queued events and reset lifecycle guards.

    Call this at the start of each request to clear any stale events left
    over from a previous invocation.  Resets the SESSION_START and
    SESSION_END guards so the next invocation cycle can re-emit them.
    """
    while not self._queue.empty():
        self._queue.get_nowait()
    self._session_start_emitted = False
    self._session_end_emitted = False

emit_session_start 

emit_session_start(manifest)

Emit a SESSION_START event with the session manifest.

Places a :class:StreamEvent with type=EventType.SESSION_START on the queue. A guard prevents double-emission within the same invocation cycle (reset by :meth:flush).

Parameters:

Name Type Description Default
manifest SessionManifest

The :class:SessionManifest describing the wired session. Placed in the event payload as {"session_id": <session_id>, "manifest": manifest.model_dump()}.

 required
Source code in src/kaboo_workflows/wire.py 
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
def emit_session_start(self, manifest: SessionManifest) -> None:
    """Emit a SESSION_START event with the session manifest.

    Places a :class:`StreamEvent` with ``type=EventType.SESSION_START`` on
    the queue.  A guard prevents double-emission within the same
    invocation cycle (reset by :meth:`flush`).

    Args:
        manifest: The :class:`SessionManifest` describing the wired
            session.  Placed in the event payload as
            ``{"session_id": <session_id>, "manifest": manifest.model_dump()}``.
    """
    if self._session_start_emitted:
        return
    self._session_start_emitted = True
    self._put(
        StreamEvent(
            type=EventType.SESSION_START,
            agent_name=self._entry_name,
            data={
                "session_id": self._session_id,
                "manifest": manifest.model_dump(),
            },
        )
    )
    logger.debug("entry=<%s> | session_start emitted", self._entry_name)

get async 

get()

Wait for the next event.

Returns:

Type Description
StreamEvent | None

The next :class:StreamEvent, or None when the stream
StreamEvent | None

has been closed via :meth:close.
Source code in src/kaboo_workflows/wire.py 
137
138
139
140
141
142
143
144
145
async def get(self) -> StreamEvent | None:
    """Wait for the next event.

    Returns:
        The next :class:`StreamEvent`, or ``None`` when the stream
        has been closed via :meth:`close`.
    """
    item = await self._queue.get()
    return None if item is _SENTINEL else item

get_nowait 

get_nowait()

Return the next already-queued event without waiting.

Returns:

Type Description
StreamEvent | None

The next :class:StreamEvent, or None when the queue is
StreamEvent | None

empty or the next item is the close sentinel.
Source code in src/kaboo_workflows/wire.py 
147
148
149
150
151
152
153
154
155
156
157
158
def get_nowait(self) -> StreamEvent | None:
    """Return the next already-queued event without waiting.

    Returns:
        The next :class:`StreamEvent`, or ``None`` when the queue is
        empty or the next item is the close sentinel.
    """
    try:
        item = self._queue.get_nowait()
    except asyncio.QueueEmpty:
        return None
    return None if item is _SENTINEL else item

put_event 

put_event(event)

Place an event on the queue (non-blocking, thread-safe).

Useful for injecting out-of-band events such as error signals.

Source code in src/kaboo_workflows/wire.py 
160
161
162
163
164
165
def put_event(self, event: StreamEvent) -> None:
    """Place an event on the queue (non-blocking, thread-safe).

    Useful for injecting out-of-band events such as error signals.
    """
    self._put(event)

close async 

close(data=None)

Signal end-of-stream.

Emits a SESSION_END event before placing the sentinel on the queue. A guard prevents double-emission within the same invocation cycle (reset by :meth:flush). Subsequent close() calls are no-ops for the SESSION_END emission but still place the sentinel — the method remains idempotent.

Typically called in a finally block after the agent invocation finishes.

Parameters:

Name Type Description Default
data dict[str, Any] | None

Additional data to include in the SESSION_END event.

 None
Source code in src/kaboo_workflows/wire.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
async def close(self, data: dict[str, Any] | None = None) -> None:
    """Signal end-of-stream.

    Emits a SESSION_END event before placing the sentinel on the queue.
    A guard prevents double-emission within the same invocation cycle
    (reset by :meth:`flush`).  Subsequent ``close()`` calls are no-ops
    for the SESSION_END emission but still place the sentinel — the
    method remains idempotent.

    Typically called in a ``finally`` block after the agent invocation
    finishes.

    Args:
        data: Additional data to include in the SESSION_END event.

    """
    if not self._session_end_emitted:
        self._session_end_emitted = True
        self._put(
            StreamEvent(
                type=EventType.SESSION_END,
                agent_name=self._entry_name,
                data={"session_id": self._session_id, **(data or {})},
            )
        )
        logger.debug(
            "entry=<%s>, session_id=<%s> | session_end emitted",
            self._entry_name,
            self._session_id,
        )
    await self._queue.put(_SENTINEL)

get_auth_context 

get_auth_context()

Return the current request's :class:Principal (or None).

Source code in src/kaboo_workflows/_context.py 
339
340
341
def get_auth_context() -> Principal | None:
    """Return the current request's :class:`Principal` (or ``None``)."""
    return _current_principal.get()

get_forwarded_props 

get_forwarded_props()

Return the current request's forwarded props (empty dict when none).

Source code in src/kaboo_workflows/_context.py 
269
270
271
def get_forwarded_props() -> dict[str, Any]:
    """Return the current request's forwarded props (empty dict when none)."""
    return _current_forwarded_props.get() or {}

set_auth_context 

set_auth_context(principal)

Bind the current request's :class:Principal.

Call at the start of each request handler, before creating the run task, so tasks (and MCP clients started within this context via contextvars.copy_context()) inherit the identity. Outbound MCP auth strategies read it via :func:get_auth_context.

Source code in src/kaboo_workflows/_context.py 
328
329
330
331
332
333
334
335
336
def set_auth_context(principal: Principal | None) -> None:
    """Bind the current request's :class:`Principal`.

    Call at the start of each request handler, before creating the run task, so
    tasks (and MCP clients started within this context via
    ``contextvars.copy_context()``) inherit the identity. Outbound MCP auth
    strategies read it via :func:`get_auth_context`.
    """
    _current_principal.set(principal)

set_forwarded_props 

set_forwarded_props(props)

Bind the current request's RunAgentInput.forwarded_props.

forwardedProps is the AG-UI protocol's free-form side channel: hosts attach per-run context (scoped credentials, per-run agent configuration, tenant ids, …) that is not part of the conversation and must not reach the model. Call at the start of each request handler, before creating the run task, so tasks created afterwards inherit a copy of this context. Host tools and hooks read it back with :func:get_forwarded_props.

Source code in src/kaboo_workflows/_context.py 
256
257
258
259
260
261
262
263
264
265
266
def set_forwarded_props(props: dict[str, Any] | None) -> None:
    """Bind the current request's ``RunAgentInput.forwarded_props``.

    ``forwardedProps`` is the AG-UI protocol's free-form side channel: hosts
    attach per-run context (scoped credentials, per-run agent configuration,
    tenant ids, …) that is not part of the conversation and must not reach the
    model. Call at the start of each request handler, before creating the run
    task, so tasks created afterwards inherit a copy of this context. Host
    tools and hooks read it back with :func:`get_forwarded_props`.
    """
    _current_forwarded_props.set(props if isinstance(props, dict) else {})

build_auth 

build_auth(strategy, params=None)

Build an outbound MCP auth strategy from a name + params.

Parameters:

Name Type Description Default
strategy str

One of relay, static, obo, m2m.

 required
params dict[str, Any] | None

Constructor keyword arguments for the strategy.

 None

Returns:

Type Description
Auth

A configured :class:httpx.Auth instance.

Raises:

Type Description
ValueError

If strategy is not a known strategy name.
Source code in src/kaboo_workflows/auth/strategies.py 
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
def build_auth(strategy: str, params: dict[str, Any] | None = None) -> httpx.Auth:
    """Build an outbound MCP auth strategy from a name + params.

    Args:
        strategy: One of ``relay``, ``static``, ``obo``, ``m2m``.
        params: Constructor keyword arguments for the strategy.

    Returns:
        A configured :class:`httpx.Auth` instance.

    Raises:
        ValueError: If ``strategy`` is not a known strategy name.
    """
    try:
        cls = _STRATEGIES[strategy]
    except KeyError:
        raise ValueError(
            f"Unknown MCP auth strategy '{strategy}'. Available: {', '.join(sorted(_STRATEGIES))}."
        ) from None
    return cls(**(params or {}))

load 

load(config)

Load config from file(s) or YAML string(s) and resolve to live objects.

This is the main entry point for zero-code usage. Accepts a single source or a list of sources. Each source is either a file path (str or Path) or a raw YAML string. File paths are detected by checking if the path exists on disk; anything else is parsed as inline YAML.

Pipeline:
  1. Parse each source (file read or inline YAML)
  2. Per-source: strip anchors, interpolate variables
  3. Sanitize collection keys (spaces/special chars -> underscores)
  4. Merge sources (if multiple), detect duplicate names
  5. Validate against schema (Pydantic)
  6. Resolve infrastructure (models, MCP — pure; no session manager)
  7. Start MCP servers (so clients can connect)
  8. Create agents (Agent.init auto-starts MCP clients)
  9. Wire orchestration / entry point

Parameters:

Name Type Description Default
config ConfigInput | list[ConfigInput]

File path, raw YAML string, or list of either.

 required

Returns:

Type Description
ResolvedConfig

ResolvedConfig with agents, entry (callable), and mcp_lifecycle.

Raises:

Type Description
FileNotFoundError

Config file doesn't exist.
ConfigurationError

Invalid YAML syntax, schema validation failure, or invalid references.
REMARKS:

When multiple sources are provided, collection sections (agents, models, mcp_servers, mcp_clients, orchestrations) are merged. Duplicate names within the same section raise ValueError. Singleton fields (entry, session_manager, log_level) use last-wins semantics.

Side effect: this function starts MCP servers during resolution. Agent.__init__ auto-starts MCP clients (via process_tools() -> MCPClient.load_tools()), and those clients need running servers to connect to. MCPLifecycle.start() is called before agent creation to satisfy this dependency.

MCPLifecycle.start() is idempotent, so the caller's context manager (async with resolved.mcp_lifecycle:) is a no-op on enter but still required for graceful shutdown — __aexit__ stops clients first, then servers.

Source code in src/kaboo_workflows/config/loaders/loaders.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
def load(config: ConfigInput | list[ConfigInput]) -> ResolvedConfig:
    """Load config from file(s) or YAML string(s) and resolve to live objects.

    This is the main entry point for zero-code usage.
    Accepts a single source or a list of sources. Each source is either
    a file path (``str`` or ``Path``) or a raw YAML string. File paths
    are detected by checking if the path exists on disk; anything else
    is parsed as inline YAML.

    ### Pipeline:

    1. Parse each source (file read or inline YAML)
    2. Per-source: strip anchors, interpolate variables
    3. Sanitize collection keys (spaces/special chars -> underscores)
    4. Merge sources (if multiple), detect duplicate names
    5. Validate against schema (Pydantic)
    6. Resolve infrastructure (models, MCP — pure; no session manager)
    7. Start MCP servers (so clients can connect)
    8. Create agents (Agent.__init__ auto-starts MCP clients)
    9. Wire orchestration / entry point

    Args:
        config: File path, raw YAML string, or list of either.

    Returns:
        ResolvedConfig with agents, entry (callable), and mcp_lifecycle.

    Raises:
        FileNotFoundError: Config file doesn't exist.
        ConfigurationError: Invalid YAML syntax, schema validation failure, or invalid references.

    ---
    ## REMARKS:

    When multiple sources are provided, collection sections (``agents``,
    ``models``, ``mcp_servers``, ``mcp_clients``, ``orchestrations``) are
    merged. Duplicate names within the same section raise ``ValueError``.
    Singleton fields (``entry``, ``session_manager``, ``log_level``) use
    last-wins semantics.

    **Side effect**: this function starts MCP servers during resolution.
    ``Agent.__init__`` auto-starts MCP clients (via ``process_tools()`` ->
    ``MCPClient.load_tools()``), and those clients need running servers to
    connect to. ``MCPLifecycle.start()`` is called **before** agent
    creation to satisfy this dependency.

    ``MCPLifecycle.start()`` is idempotent, so the caller's context
    manager (``async with resolved.mcp_lifecycle:``) is a no-op on enter
    but **still required for graceful shutdown** — ``__aexit__`` stops
    clients first, then servers.
    """
    app_config = load_config(config)

    logging.getLogger("kaboo_workflows").setLevel(app_config.log_level.upper())

    from ...telemetry import init_telemetry

    init_telemetry(app_config.telemetry)

    infra = resolve_infra(app_config)

    # Start MCP servers BEFORE creating agents.
    # Initializing Agent starts MCP clients, so we need servers up first.
    infra.mcp_lifecycle.start()

    return load_session(app_config, infra)

load_config 

load_config(config)

Parse and validate config from file(s) or YAML string(s).

Accepts a single source or a list. Each source is auto-detected:

  • Path objects are always treated as file paths.
  • str values are treated as file paths if the file exists on disk; otherwise they are parsed as inline YAML content.

When multiple sources are provided, their collection sections (agents, models, mcp_servers, mcp_clients, orchestrations) are merged. Duplicate names within the same section raise ValueError. Singleton fields (entry, session_manager, log_level) use last-wins semantics.

Each source's vars: block is applied only to that source (interpolation is per-source).

Parameters:

Name Type Description Default
config ConfigInput | list[ConfigInput]

File path, raw YAML string, or list of either.

 required

Returns:

Type Description
AppConfig

Validated AppConfig instance.

Raises:

Type Description
FileNotFoundError

A Path source doesn't exist.
ConfigurationError

Invalid YAML, schema validation failure, or invalid references.
Source code in src/kaboo_workflows/config/loaders/loaders.py 
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
def load_config(config: ConfigInput | list[ConfigInput]) -> AppConfig:
    """Parse and validate config from file(s) or YAML string(s).

    Accepts a single source or a list. Each source is auto-detected:

    - ``Path`` objects are always treated as file paths.
    - ``str`` values are treated as file paths if the file exists on disk;
      otherwise they are parsed as inline YAML content.

    When multiple sources are provided, their collection sections
    (``agents``, ``models``, ``mcp_servers``, ``mcp_clients``,
    ``orchestrations``) are merged. Duplicate names within the same
    section raise ``ValueError``. Singleton fields (``entry``,
    ``session_manager``, ``log_level``) use last-wins semantics.

    Each source's ``vars:`` block is applied only to that source
    (interpolation is per-source).

    Args:
        config: File path, raw YAML string, or list of either.

    Returns:
        Validated AppConfig instance.

    Raises:
        FileNotFoundError: A ``Path`` source doesn't exist.
        ConfigurationError: Invalid YAML, schema validation failure, or invalid references.
    """
    return validate_raw_config(parse_config_sources(config))

load_session 

load_session(config, infra, *, session_id=None)

Create agents and orchestration from already-started infrastructure.

This is the session-level counterpart to :func:load. Use it when you want to share MCP servers across multiple sessions (e.g. one session per HTTP request) while creating isolated agents per session.

infra does NOT carry a session manager; instances are built per agent/orchestration from config.session_manager plus an effective_session_id computed here.

Typical server pattern::

app_config = load_config("config.yaml")
infra = resolve_infra(app_config)
infra.mcp_lifecycle.start()

# Per request:
resolved = load_session(app_config, infra, session_id="abc")

infra.mcp_lifecycle must already be started before calling this.

Parameters:

Name Type Description Default
config AppConfig

The validated AppConfig.

 required
infra ResolvedInfra

Resolved infrastructure with servers already started.

 required
session_id str | None

Optional runtime session ID. Combined with config.session_manager.params.session_id (if any) and a uuid.uuid4() fallback to derive a single effective_session_id that is threaded into every per-agent and per-orchestration session-manager resolution. When None and no global session_manager: is configured, leaves fall back to their own UUIDs per resolve_session_manager.

 None

Returns:

Type Description
ResolvedConfig

ResolvedConfig with freshly created agents and entry point.
Note

No SessionManager instance is built in this function; instances are constructed at the leaves (build_agent_from_def, OrchestrationBuilder._build_one).

Source code in src/kaboo_workflows/config/loaders/loaders.py 
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
def load_session(
    config: AppConfig,
    infra: ResolvedInfra,
    *,
    session_id: str | None = None,
) -> ResolvedConfig:
    """Create agents and orchestration from already-started infrastructure.

    This is the session-level counterpart to :func:`load`. Use it when
    you want to share MCP servers across multiple sessions (e.g. one
    session per HTTP request) while creating **isolated** agents per
    session.

    ``infra`` does NOT carry a session manager; instances are built per
    agent/orchestration from ``config.session_manager`` plus an
    ``effective_session_id`` computed here.

    Typical server pattern::

        app_config = load_config("config.yaml")
        infra = resolve_infra(app_config)
        infra.mcp_lifecycle.start()

        # Per request:
        resolved = load_session(app_config, infra, session_id="abc")

    ``infra.mcp_lifecycle`` must already be started before calling this.

    Args:
        config: The validated AppConfig.
        infra: Resolved infrastructure with servers already started.
        session_id: Optional runtime session ID. Combined with
            ``config.session_manager.params.session_id`` (if any) and a
            ``uuid.uuid4()`` fallback to derive a single
            ``effective_session_id`` that is threaded into every per-agent
            and per-orchestration session-manager resolution. When ``None``
            and no global ``session_manager:`` is configured, leaves fall
            back to their own UUIDs per ``resolve_session_manager``.

    Returns:
        ResolvedConfig with freshly created agents and entry point.

    Note:
        No ``SessionManager`` instance is built in this function; instances
        are constructed at the leaves (``build_agent_from_def``,
        ``OrchestrationBuilder._build_one``).
    """
    # Compute a single effective session id for every leaf that will resolve a global SM.
    # CLI parity:
    # when no real session_id is supplied but the config declares a global session_manager,
    # all leaves that fall back to that def share one fresh UUID
    effective_session_id: str | None = session_id
    if effective_session_id is None and config.session_manager is not None:
        yaml_sid = (config.session_manager.params or {}).get("session_id")
        effective_session_id = yaml_sid or str(uuid.uuid4())

    # Agents used in Swarm or Graph orchestrations cannot have session_manager set.
    # Temporary until strands-agents supports session persistence for orchestration node agents.
    orchestration_agent_names: set[str] = set()
    for orch in config.orchestrations.values():
        if isinstance(orch, SwarmOrchestrationDef):
            orchestration_agent_names.update(orch.agents)
        elif isinstance(orch, GraphOrchestrationDef):
            orchestration_agent_names.add(orch.entry_name)
            for edge in orch.edges:
                orchestration_agent_names.add(edge.from_agent)
                orchestration_agent_names.add(edge.to_agent)

    agents = resolve_agents(
        agent_defs=config.agents,
        models=infra.models,
        mcp_clients=infra.clients,
        global_session_manager_def=config.session_manager,
        session_id=effective_session_id,
        orchestration_agent_names=orchestration_agent_names,
        attachments=config.attachments,
    )
    orchestrators = resolve_orchestrations(
        config,
        agents,
        agent_defs=config.agents,
        models=infra.models,
        mcp_clients=infra.clients,
        global_session_manager_def=config.session_manager,
        session_id=effective_session_id,
    )

    all_nodes = dict(agents) | orchestrators
    entry = all_nodes[config.entry]

    return ResolvedConfig(
        agents=agents,
        orchestrators=orchestrators,
        entry=entry,
        mcp_lifecycle=infra.mcp_lifecycle,
        app_config=config,
    )

load_session_config 

load_session_config(
    base_raw, overlay=None, *, allowed_mcp_hosts=None
)

Validate a per-run config layered over the service's base config.

This is how a run brings its own workflow — its own agents, delegates, graph or swarm — without the service restarting or reading a file. The overlay replaces the base's agents / orchestrations / entry and adds to its models / mcp_clients; see :func:~kaboo_workflows.config.loaders.helpers.merge_session_overlay.

Interpolation runs on the overlay as its own source, so a ${API_KEY} in a database-authored config resolves from this process's environment. The secret never has to live wherever the config was written.

Parameters:

Name Type Description Default
base_raw dict

The service's config, parsed once by :func:parse_config_sources. Not mutated, so it is safe to share across concurrent runs.

 required
overlay ConfigInput | None

The submitted config — normally a raw YAML string. None validates the base alone, so callers need only one code path.

 None
allowed_mcp_hosts list[str] | None

Hosts an overlay-declared MCP client URL may point at. None disables the check.

 None

Returns:

Type Description
AppConfig

Validated AppConfig for this run.

Raises:

Type Description
ConfigurationError

Invalid YAML, schema failure, bad references, or an MCP client the overlay is not permitted to declare.
Source code in src/kaboo_workflows/config/loaders/loaders.py 
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
def load_session_config(
    base_raw: dict,
    overlay: ConfigInput | None = None,
    *,
    allowed_mcp_hosts: list[str] | None = None,
) -> AppConfig:
    """Validate a per-run config layered over the service's base config.

    This is how a run brings its own workflow — its own agents, delegates, graph
    or swarm — without the service restarting or reading a file. The overlay
    replaces the base's ``agents`` / ``orchestrations`` / ``entry`` and adds to its
    ``models`` / ``mcp_clients``; see
    :func:`~kaboo_workflows.config.loaders.helpers.merge_session_overlay`.

    Interpolation runs on the overlay as its own source, so a ``${API_KEY}`` in a
    database-authored config resolves from this process's environment. The secret
    never has to live wherever the config was written.

    Args:
        base_raw: The service's config, parsed once by :func:`parse_config_sources`.
            Not mutated, so it is safe to share across concurrent runs.
        overlay: The submitted config — normally a raw YAML string. ``None``
            validates the base alone, so callers need only one code path.
        allowed_mcp_hosts: Hosts an overlay-declared MCP client URL may point at.
            ``None`` disables the check.

    Returns:
        Validated AppConfig for this run.

    Raises:
        ConfigurationError: Invalid YAML, schema failure, bad references, or an
            MCP client the overlay is not permitted to declare.
    """
    if overlay is None:
        return validate_raw_config(base_raw)

    overlay_raw = parse_single_source(overlay)
    sanitize_collection_keys(overlay_raw)
    merged = merge_session_overlay(base_raw, overlay_raw, allowed_mcp_hosts=allowed_mcp_hosts)
    return validate_raw_config(merged)

parse_config_sources 

parse_config_sources(config)

Parse and merge config source(s) into a processed raw dict.

The first half of :func:load_config, split out so a server can parse its base config once at boot and reuse the result for every run — see :func:load_session_config. Interpolation has already happened, so the returned dict may contain resolved secrets; it is not for logging.

Parameters:

Name Type Description Default
config ConfigInput | list[ConfigInput]

File path, raw YAML string, or list of either.

 required

Returns:

Type Description
dict

The merged raw config dict.
Source code in src/kaboo_workflows/config/loaders/loaders.py 
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
def parse_config_sources(config: ConfigInput | list[ConfigInput]) -> dict:
    """Parse and merge config source(s) into a processed raw dict.

    The first half of :func:`load_config`, split out so a server can parse its
    base config once at boot and reuse the result for every run — see
    :func:`load_session_config`. Interpolation has already happened, so the
    returned dict may contain resolved secrets; it is not for logging.

    Args:
        config: File path, raw YAML string, or list of either.

    Returns:
        The merged raw config dict.
    """
    sources = config if isinstance(config, list) else [config]
    raw_configs = [parse_single_source(s) for s in sources]

    for raw in raw_configs:
        sanitize_collection_keys(raw)

    return merge_raw_configs(raw_configs) if len(raw_configs) > 1 else raw_configs[0]

resolve_infra 

resolve_infra(config)

Resolve infrastructure from an AppConfig (pure, no I/O).

Creates model objects, MCP server/client objects, and a lifecycle manager. Nothing is started.

Resolution order:

  1. Models (no dependencies)
  2. MCP servers (no dependencies)
  3. MCP clients (depend on servers)
  4. MCP lifecycle (assembles servers + clients, not started)
  5. Session manager validation only — agentcore provider rejected globally; no instance is constructed (instances are built per-leaf at session time).

Agents and orchestration are resolved in :func:load after mcp_lifecycle.start() because Agent.__init__ auto-starts MCP clients which need servers to be running first. The lifecycle start in load() is idempotent — the context manager is still used for graceful shutdown.

Parameters:

Name Type Description Default
config AppConfig

Parsed AppConfig from YAML.

 required

Returns:

Name Type Description
A ResolvedInfra

class:ResolvedInfra with models, clients, and a cold MCP lifecycle.
Source code in src/kaboo_workflows/config/resolvers/config.py 
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
def resolve_infra(config: AppConfig) -> ResolvedInfra:
    """Resolve infrastructure from an AppConfig (pure, no I/O).

    Creates model objects, MCP server/client objects, and a lifecycle
    manager.  Nothing is started.

    Resolution order:

    1. Models (no dependencies)
    2. MCP servers (no dependencies)
    3. MCP clients (depend on servers)
    4. MCP lifecycle (assembles servers + clients, **not** started)
    5. Session manager validation only — ``agentcore`` provider rejected
       globally; no instance is constructed (instances are built per-leaf
       at session time).

    Agents and orchestration are resolved in :func:`load` after
    ``mcp_lifecycle.start()`` because ``Agent.__init__`` auto-starts
    MCP clients which need servers to be running first.  The lifecycle
    start in ``load()`` is idempotent — the context manager is still
    used for graceful shutdown.

    Args:
        config: Parsed AppConfig from YAML.

    Returns:
        A :class:`ResolvedInfra` with models, clients, and a cold MCP lifecycle.
    """
    # Models
    models: dict[str, Model] = {}
    for name, model_def in config.models.items():
        models[name] = resolve_model(model_def)
        logger.info("model=<%s>, provider=<%s> | resolved model", name, model_def.provider)

    # MCP servers
    servers: dict[str, MCPServer] = {}
    for name, server_def in config.mcp_servers.items():
        servers[name] = resolve_mcp_server(server_def, name=name)
        logger.info("server=<%s> | resolved MCP server", name)

    # MCP clients (resolved but NOT started)
    clients: dict[str, StrandsMCPClient] = {}
    for name, client_def in config.mcp_clients.items():
        clients[name] = resolve_mcp_client(client_def, servers, name=name)
        logger.info("client=<%s> | resolved MCP client", name)

    # MCP lifecycle (cold — not started)
    lifecycle = MCPLifecycle()
    for name, server in servers.items():
        lifecycle.add_server(name, server)
    for name, client in clients.items():
        lifecycle.add_client(name, client)

    # Session manager — validation only
    # Instances are built per leaf in load_session / agents / orchestrations.
    # Provider 'agentcore' cannot be set globally -
    # it requires a unique 'actor_id' per agent. Fail fast at boot.
    if (
        config.session_manager is not None
        and config.session_manager.provider.lower() == "agentcore"
    ):
        raise ValueError(
            "The 'agentcore' session manager cannot be set globally.\n"
            "Configure it per-agent — 'actor_id' must be unique per agent."
        )

    return ResolvedInfra(
        models=models,
        clients=clients,
        mcp_lifecycle=lifecycle,
    )

resolve_run_clients 

resolve_run_clients(config, infra)

Resolve this run's MCP clients as fresh objects it exclusively owns.

Client objects are cheap and hold no connection until an agent registers them as a tool provider, so a run can afford its own. Owning them is what makes the session's lifetime the connection's lifetime: the returned lifecycle is stopped when the run ends, and a session that cannot outlive its run cannot go stale between runs.

Servers are not re-resolved — those are processes, owned by the boot-time lifecycle in infra — so a stdio client in config binds to the already-running server of the same name.

Parameters:

Name Type Description Default
config AppConfig

The config for this run, which may declare clients the base config does not.

 required
infra ResolvedInfra

The process-wide infra, consulted for running servers.

 required

Returns:

Type Description
MCPLifecycle

A lifecycle holding only this run's clients, not yet started.

Raises:

Type Description
ValueError

If a stdio client references a server that boot did not start.
Source code in src/kaboo_workflows/config/resolvers/config.py 
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
def resolve_run_clients(config: AppConfig, infra: ResolvedInfra) -> MCPLifecycle:
    """Resolve this run's MCP clients as fresh objects it exclusively owns.

    Client *objects* are cheap and hold no connection until an agent registers
    them as a tool provider, so a run can afford its own. Owning them is what
    makes the session's lifetime the connection's lifetime: the returned
    lifecycle is stopped when the run ends, and a session that cannot outlive
    its run cannot go stale between runs.

    Servers are not re-resolved — those are processes, owned by the boot-time
    lifecycle in ``infra`` — so a ``stdio`` client in ``config`` binds to the
    already-running server of the same name.

    Args:
        config: The config for this run, which may declare clients the base
            config does not.
        infra: The process-wide infra, consulted for running servers.

    Returns:
        A lifecycle holding only this run's clients, not yet started.

    Raises:
        ValueError: If a ``stdio`` client references a server that boot did not
            start.
    """
    servers = infra.mcp_lifecycle.servers
    lifecycle = MCPLifecycle()
    for name, client_def in config.mcp_clients.items():
        lifecycle.add_client(name, resolve_mcp_client(client_def, servers, name=name))
        logger.debug("client=<%s> | resolved MCP client for this run", name)
    return lifecycle

validate_raw_config 

validate_raw_config(raw)

Normalize, schema-validate and reference-check a raw config dict.

Parameters:

Name Type Description Default
raw dict

A processed raw config dict from :func:parse_config_sources.

 required

Returns:

Type Description
AppConfig

Validated AppConfig instance.

Raises:

Type Description
ConfigurationError

Schema validation failure, or invalid references.
Source code in src/kaboo_workflows/config/loaders/loaders.py 
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
def validate_raw_config(raw: dict) -> AppConfig:
    """Normalize, schema-validate and reference-check a raw config dict.

    Args:
        raw: A processed raw config dict from :func:`parse_config_sources`.

    Returns:
        Validated AppConfig instance.

    Raises:
        ConfigurationError: Schema validation failure, or invalid references.
    """
    normalized = normalize(raw)
    try:
        app_config = AppConfig.model_validate(normalized)
    except ValidationError as exc:
        first_error = exc.errors()[0]
        field_path = " -> ".join(str(loc) for loc in first_error["loc"])
        raise SchemaValidationError(
            f"Invalid config at '{field_path}': {first_error['msg']}\n"
            f"Check your YAML configuration file."
        ) from None
    validate_references(app_config)

    return app_config

create_mcp_client 

create_mcp_client(
    *,
    server=None,
    url=None,
    command=None,
    transport=DEFAULT_TRANSPORT,
    transport_options=None,
    **kwargs,
)

Create a strands MCPClient from connection configuration.

Exactly one of server, url, or command must be provided.

Parameters:

Name Type Description Default
server MCPServer | None

A managed MCPServer instance (connects via its URL).

 None
url str | None

External MCP server URL (for SSE or streamable-http).

 None
command list[str] | None

Command to start an MCP server subprocess (stdio transport).

 None
transport MCP_TRANSPORT

Override transport type ("stdio", "sse", "streamable-http"). Auto-detected if not specified.

 DEFAULT_TRANSPORT
transport_options dict[str, Any] | None

Extra kwargs forwarded to the transport factory. These are transport-specific — see each transport function for available options:

stdio: env, cwd, encoding, encoding_error_handler

sse: headers, timeout, sse_read_timeout, auth, httpx_client_factory

streamable-http: headers, http_client (pre-configured httpx.AsyncClient), terminate_on_close

 None
**kwargs Any

Additional kwargs forwarded to strands MCPClient (startup_timeout, tool_filters, prefix, elicitation_callback, tasks_config, etc.).

 {}

Returns:

Type Description
MCPClient

A strands MCPClient instance.

Raises:

Type Description
ValueError

If connection parameters are ambiguous.
Source code in src/kaboo_workflows/mcp/client.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
def create_mcp_client(
    *,
    server: MCPServer | None = None,
    url: str | None = None,
    command: list[str] | None = None,
    transport: MCP_TRANSPORT = DEFAULT_TRANSPORT,
    transport_options: dict[str, Any] | None = None,
    **kwargs: Any,
) -> MCPClient:
    """Create a strands MCPClient from connection configuration.

    Exactly one of server, url, or command must be provided.

    Args:
        server: A managed MCPServer instance (connects via its URL).
        url: External MCP server URL (for SSE or streamable-http).
        command: Command to start an MCP server subprocess (stdio transport).
        transport: Override transport type ("stdio", "sse", "streamable-http").
            Auto-detected if not specified.
        transport_options: Extra kwargs forwarded to the transport factory.
            These are transport-specific — see each transport function for
            available options:

            **stdio**: ``env``, ``cwd``, ``encoding``, ``encoding_error_handler``

            **sse**: ``headers``, ``timeout``, ``sse_read_timeout``, ``auth``,
            ``httpx_client_factory``

            **streamable-http**: ``headers``, ``http_client`` (pre-configured
            ``httpx.AsyncClient``), ``terminate_on_close``

        **kwargs: Additional kwargs forwarded to strands MCPClient
            (startup_timeout, tool_filters, prefix, elicitation_callback,
            tasks_config, etc.).

    Returns:
        A strands MCPClient instance.

    Raises:
        ValueError: If connection parameters are ambiguous.
    """
    modes = sum(x is not None for x in [server, url, command])
    if modes != 1:
        raise ValueError(
            f"Exactly one of server, url, or command must be provided (got {modes}).\n"
            "server=MCPServer for managed servers, url=str for external HTTP, "
            "command=list[str] for subprocess stdio."
        )

    opts = transport_options or {}

    if server is not None:
        transport_callable = _transport_for_http(server.url, transport, opts, allow_stdio=False)
    elif url is not None:
        transport_callable = _transport_for_http(url, transport, opts, allow_stdio=True)
    else:
        # command is guaranteed non-None by the modes == 1 check above.
        transport_callable = stdio_transport(command, **opts)  # ty: ignore

    return _make_strands_client(transport_callable=transport_callable, **kwargs)

create_mcp_server 

create_mcp_server(
    *,
    name,
    tools,
    host="127.0.0.1",
    port=8000,
    transport=DEFAULT_TRANSPORT,
    server_params=None,
)

Create an MCP server from a list of callables — no subclassing needed.

Each callable (sync or async) is registered as a tool on the underlying FastMCP instance. For advanced use (custom state, routes, resources), subclass :class:MCPServer directly.

Example::

def get_weather(city: str) -> str:
    return f"Sunny in {city}"


async def query_db(sql: str) -> str: ...


server = create_mcp_server(name="weather", tools=[get_weather, query_db], port=8001)
server.start()

Parameters:

Name Type Description Default
name str

Unique server identifier.

 required
tools list[Callable[..., Any]]

Callables to register as MCP tools.

 required
host str

Bind address (default 127.0.0.1).

 '127.0.0.1'
port int

Bind port (default 8000).

 8000
transport MCP_SERVER_TRANSPORT

Server transport type (streamable-http or sse).

 DEFAULT_TRANSPORT
server_params dict[str, Any] | None

Extra kwargs forwarded to FastMCP().

 None

Returns:

Type Description
MCPServer

A ready-to-use :class:MCPServer instance.
Source code in src/kaboo_workflows/mcp/server.py 
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
def create_mcp_server(
    *,
    name: str,
    tools: list[Callable[..., Any]],
    host: str = "127.0.0.1",
    port: int = 8000,
    transport: MCP_SERVER_TRANSPORT = DEFAULT_TRANSPORT,
    server_params: dict[str, Any] | None = None,
) -> MCPServer:
    """Create an MCP server from a list of callables — no subclassing needed.

    Each callable (sync or async) is registered as a tool on the underlying
    ``FastMCP`` instance.  For advanced use (custom state, routes, resources),
    subclass :class:`MCPServer` directly.

    Example::

        def get_weather(city: str) -> str:
            return f"Sunny in {city}"


        async def query_db(sql: str) -> str: ...


        server = create_mcp_server(name="weather", tools=[get_weather, query_db], port=8001)
        server.start()

    Args:
        name: Unique server identifier.
        tools: Callables to register as MCP tools.
        host: Bind address (default ``127.0.0.1``).
        port: Bind port (default ``8000``).
        transport: Server transport type (``streamable-http`` or ``sse``).
        server_params: Extra kwargs forwarded to ``FastMCP()``.

    Returns:
        A ready-to-use :class:`MCPServer` instance.
    """
    tool_fns = list(tools)

    class _FactoryServer(MCPServer):
        def _register_tools(self, mcp: FastMCP) -> None:
            for fn in tool_fns:
                mcp.tool()(fn)

    return _FactoryServer(
        name=name,
        host=host,
        port=port,
        transport=transport,
        server_params=server_params,
    )

current_trace_id 

current_trace_id(*, span=None)

Return the active trace id as 32-char hex, or None.

Reads the ambient current span (strands activates the agent span around the run loop) unless an explicit span is given. Returns None when telemetry is disabled or no valid span is recording, so callers can unconditionally call this with zero overhead in the disabled case.

Source code in src/kaboo_workflows/telemetry.py 
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
def current_trace_id(*, span: Any | None = None) -> str | None:
    """Return the active trace id as 32-char hex, or ``None``.

    Reads the ambient current span (strands activates the agent span around
    the run loop) unless an explicit *span* is given. Returns ``None`` when
    telemetry is disabled or no valid span is recording, so callers can
    unconditionally call this with zero overhead in the disabled case.
    """
    if not _enabled:
        return None
    target = span if span is not None else trace_api.get_current_span()
    try:
        ctx = target.get_span_context()
    except AttributeError:
        return None
    if ctx is None or not ctx.trace_id:
        return None
    return format(ctx.trace_id, "032x")

init_telemetry 

init_telemetry(config)

Initialize process-wide tracing from a config's telemetry section.

Idempotent — the first call wins; later calls (e.g. per-run session configs) return the already-decided state without touching the tracer provider. When disabled this configures nothing: no SDK provider, no exporters, no overhead.

Resolution order for enabled: the KABOO_TELEMETRY_ENABLED env var (either direction) overrides the config flag.

Parameters:

Name Type Description Default
config TelemetryDef | None

The telemetry: section of an :class:AppConfig (None behaves as the all-defaults section).

 required

Returns:

Type Description
bool

Whether telemetry is enabled for this process.

Raises:

Type Description
ImportError

Telemetry is enabled but the OTLP exporter is not installed (pip install 'kaboo-workflows[telemetry]').
Source code in src/kaboo_workflows/telemetry.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
def init_telemetry(config: TelemetryDef | None) -> bool:
    """Initialize process-wide tracing from a config's ``telemetry`` section.

    Idempotent — the first call wins; later calls (e.g. per-run session
    configs) return the already-decided state without touching the tracer
    provider. When disabled this configures nothing: no SDK provider, no
    exporters, no overhead.

    Resolution order for ``enabled``: the ``KABOO_TELEMETRY_ENABLED`` env var
    (either direction) overrides the config flag.

    Args:
        config: The ``telemetry:`` section of an :class:`AppConfig`
            (``None`` behaves as the all-defaults section).

    Returns:
        Whether telemetry is enabled for this process.

    Raises:
        ImportError: Telemetry is enabled but the OTLP exporter is not
            installed (``pip install 'kaboo-workflows[telemetry]'``).
    """
    global _initialized, _enabled
    if _initialized:
        return _enabled
    _initialized = True

    from .config.schema import TelemetryDef

    cfg = config or TelemetryDef()
    override = _env_flag(os.getenv(_ENV_ENABLED))
    enabled = cfg.enabled if override is None else override
    if not enabled:
        _enabled = False
        return False

    try:
        from opentelemetry.exporter.otlp.proto.http import (  # noqa: F401
            trace_exporter,
        )
    except ImportError as exc:
        raise ImportError(
            "telemetry.enabled is true but the OTLP exporter is not installed. "
            "Install it with: pip install 'kaboo-workflows[telemetry]'"
        ) from exc

    from opentelemetry.sdk.trace import TracerProvider as SDKTracerProvider
    from strands.telemetry import StrandsTelemetry
    from strands.telemetry.config import get_otel_resource

    os.environ.setdefault("OTEL_SERVICE_NAME", cfg.service_name)

    # Respect an existing global SDK TracerProvider (a host app may have
    # configured OTel already); otherwise create one — with sampling when
    # requested — and set it as the global provider.
    current = trace_api.get_tracer_provider()
    if isinstance(current, SDKTracerProvider):
        provider = current
        telemetry = StrandsTelemetry(tracer_provider=provider)
    elif cfg.sample_ratio < 1.0:
        from opentelemetry.sdk.trace.sampling import ParentBasedTraceIdRatio

        provider = SDKTracerProvider(
            resource=get_otel_resource(),
            sampler=ParentBasedTraceIdRatio(cfg.sample_ratio),
        )
        trace_api.set_tracer_provider(provider)
        telemetry = StrandsTelemetry(tracer_provider=provider)
    else:
        telemetry = StrandsTelemetry()
        provider = telemetry.tracer_provider

    provider.add_span_processor(KabooContextSpanProcessor(cfg.trace_attributes))

    exporter_kwargs: dict[str, Any] = {}
    endpoint = resolve_traces_endpoint(cfg.otlp.endpoint)
    if endpoint:
        exporter_kwargs["endpoint"] = endpoint
    headers = _resolve_headers(cfg.otlp.headers)
    if headers:
        exporter_kwargs["headers"] = headers
    telemetry.setup_otlp_exporter(**exporter_kwargs)

    if cfg.console:
        telemetry.setup_console_exporter()

    logger.info(
        "telemetry enabled service=<%s> endpoint=<%s> sample_ratio=<%s>",
        cfg.service_name,
        endpoint or "(env-default)",
        cfg.sample_ratio,
    )
    _enabled = True
    return True

telemetry_enabled 

telemetry_enabled()

Return whether telemetry was enabled by :func:init_telemetry.

Source code in src/kaboo_workflows/telemetry.py 
53
54
55
def telemetry_enabled() -> bool:
    """Return whether telemetry was enabled by :func:`init_telemetry`."""
    return _enabled

node_as_async_tool 

node_as_async_tool(node, *, name=None, description)

Wrap an Agent or MultiAgentBase as an async AgentTool for delegation.

Source code in src/kaboo_workflows/tools/wrappers.py 
151
152
153
154
155
156
157
158
def node_as_async_tool(
    node: Node,
    *,
    name: str | None = None,
    description: str,
) -> DecoratedFunctionTool:
    """Wrap an Agent or MultiAgentBase as an async ``AgentTool`` for delegation."""
    return _node_as_tool(node, name=name, description=description, is_async=True)

node_as_tool 

node_as_tool(node, *, name=None, description)

Wrap an Agent or MultiAgentBase as a sync AgentTool for delegation.

Source code in src/kaboo_workflows/tools/wrappers.py 
141
142
143
144
145
146
147
148
def node_as_tool(
    node: Node,
    *,
    name: str | None = None,
    description: str,
) -> DecoratedFunctionTool:
    """Wrap an Agent or MultiAgentBase as a sync ``AgentTool`` for delegation."""
    return _node_as_tool(node, name=name, description=description, is_async=False)

serialize_multiagent_result 

serialize_multiagent_result(result)

Serialize a MultiAgentResult with execution metadata omitted by to_dict().

Extends result.to_dict() with fields only available on the live object:

  • last_node_id — id of the truly last executing node, derived from node_history / execution_order (not dict insertion order).
  • response — plain-text answer from that node, ready to use directly without any further extraction.
  • swarm.node_history — full ordered execution trace including repeated visits (SwarmResult only).
  • graph.execution_order, graph.edges, graph.entry_points, and node counts (GraphResult only).

Parameters:

Name Type Description Default
result MultiAgentResult

A live MultiAgentResult, SwarmResult, or GraphResult returned directly by invoke_async.

 required

Returns:

Type Description
dict[str, Any]

A JSON-serializable dict extending result.to_dict().
Source code in src/kaboo_workflows/tools/extractors.py 
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
def serialize_multiagent_result(result: MultiAgentResult) -> dict[str, Any]:
    """Serialize a ``MultiAgentResult`` with execution metadata omitted by ``to_dict()``.

    Extends ``result.to_dict()`` with fields only available on the live object:

    - ``last_node_id`` — id of the truly last executing node, derived from
      ``node_history`` / ``execution_order`` (not dict insertion order).
    - ``response`` — plain-text answer from that node, ready to use
      directly without any further extraction.
    - ``swarm.node_history`` — full ordered execution trace including repeated
      visits (``SwarmResult`` only).
    - ``graph.execution_order``, ``graph.edges``, ``graph.entry_points``, and
      node counts (``GraphResult`` only).

    Args:
        result: A live ``MultiAgentResult``, ``SwarmResult``, or ``GraphResult``
            returned directly by ``invoke_async``.

    Returns:
        A JSON-serializable dict extending ``result.to_dict()``.
    """
    data = result.to_dict()

    last_node_id = resolve_last_node_id(result)
    data["last_node_id"] = last_node_id

    final_message = extract_last_message(result)
    data["response"] = extract_text(final_message)

    # SwarmResult extras — node_history is a list[SwarmNode]
    node_history: list[Any] | None = getattr(result, "node_history", None)
    if node_history is not None:
        data["swarm"] = {
            "node_history": [str(n.node_id) for n in node_history],
        }

    # GraphResult extras — execution_order, edges, node counts
    execution_order: list[Any] | None = getattr(result, "execution_order", None)
    if execution_order is not None:
        edges_raw: list[Any] = getattr(result, "edges", []) or []
        entry_points_raw: list[Any] = getattr(result, "entry_points", []) or []

        edges: list[list[str]] = []
        for edge in edges_raw:
            if isinstance(edge, tuple) and len(edge) == 2:
                edges.append([str(edge[0].node_id), str(edge[1].node_id)])
            else:
                # GraphEdge dataclass with from_node / to_node attributes
                from_id = str(getattr(getattr(edge, "from_node", None), "node_id", edge))
                to_id = str(getattr(getattr(edge, "to_node", None), "node_id", edge))
                edges.append([from_id, to_id])

        data["graph"] = {
            "execution_order": [str(n.node_id) for n in execution_order],
            "edges": edges,
            "entry_points": [str(getattr(ep, "node_id", ep)) for ep in entry_points_raw],
            "completed_nodes": getattr(result, "completed_nodes", 0),
            "failed_nodes": getattr(result, "failed_nodes", 0),
            "interrupted_nodes": getattr(result, "interrupted_nodes", 0),
        }

    return data

cli_errors 

cli_errors(*, exit_code=1)

Catch unhandled exceptions and print a clean, user-friendly message.

.. warning::

**CLI-only** — this context manager calls ``sys.exit()`` on errors,
which raises ``SystemExit``.  Do **not** use it in ASGI/WSGI server
code (FastAPI, Flask, etc.) where ``SystemExit`` would kill the
worker process.  For server code, catch exceptions directly.

Intended for CLI entry points — wraps the body so that configuration errors, auth failures, network problems, etc. are displayed without a full Python traceback::

with cli_errors():
    asyncio.run(main())

KeyboardInterrupt and SystemExit are not caught.

Parameters:

Name Type Description Default
exit_code int

Process exit code used after printing the error. Set to 0 to suppress sys.exit (useful in tests).

 1
Source code in src/kaboo_workflows/utils.py 
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
@contextlib.contextmanager
def cli_errors(*, exit_code: int = 1) -> Generator[None]:
    """Catch unhandled exceptions and print a clean, user-friendly message.

    .. warning::

        **CLI-only** — this context manager calls ``sys.exit()`` on errors,
        which raises ``SystemExit``.  Do **not** use it in ASGI/WSGI server
        code (FastAPI, Flask, etc.) where ``SystemExit`` would kill the
        worker process.  For server code, catch exceptions directly.

    Intended for CLI entry points — wraps the body so that configuration
    errors, auth failures, network problems, etc. are displayed without a
    full Python traceback::

        with cli_errors():
            asyncio.run(main())

    ``KeyboardInterrupt`` and ``SystemExit`` are **not** caught.

    Args:
        exit_code: Process exit code used after printing the error.
                   Set to ``0`` to suppress ``sys.exit`` (useful in tests).
    """
    _asyncio = logging.getLogger("asyncio")
    _filter = _SuppressTaskExceptions()
    _asyncio.addFilter(_filter)
    try:
        yield
    except (KeyboardInterrupt, SystemExit):
        raise
    except Exception as exc:
        msg = f"\n{_format_exception(exc)}\n"
        if sys.stderr.isatty():
            msg = f"\033[31m{msg}\033[0m"
        print(msg, file=sys.stderr)  # noqa: T201
        if exit_code:
            sys.exit(exit_code)
    finally:
        _asyncio.removeFilter(_filter)

make_event_queue 

make_event_queue(
    agents,
    *,
    orchestrators=None,
    tool_labels=None,
    stream_groups=None,
    history=None,
    entry_name=None,
    chat_owner=None,
    chat_reply=None,
    session_id=None,
)

Attach :class:~kaboo_workflows.hooks.EventPublisher hooks to agents.

Every agent in agents receives an :class:.EventPublisher hook and a matching callback_handler so all per-agent event types flow into the returned :class:EventQueue. Orchestrators (Swarm / Graph / delegate Agent) in orchestrators also get a publisher for NODE_START, NODE_STOP, HANDOFF, and MULTIAGENT_COMPLETE events.

This function does not emit SESSION_START. Callers that own a :class:~kaboo_workflows.types.SessionManifest should call :meth:EventQueue.emit_session_start themselves; the common :class:ResolvedConfig workflow does this for you via :meth:ResolvedConfig.wire_event_queue.

.. warning::

This function **mutates** the passed-in agents and orchestrators by
adding hooks and overwriting ``callback_handler``.  Call it only once
per set of agents.

Parameters:

Name Type Description Default
agents dict[str, Agent]

Agents to wire, keyed by name.

 required
orchestrators dict[str, Node] | None

Built orchestrations keyed by name.

 None
tool_labels dict[str, str] | None

Tool name → display label mapping forwarded to each :class:.EventPublisher. Defaults to {name: "Delegating work to agent: <Name>"} for every agent.

 None
stream_groups dict[str, tuple[str, str]] | None

agent_name → (group, title) mapping for hierarchical activity attribution. When an agent is not in this dict it uses its own name as the group.

 None
history dict[str, tuple[str, bool]] | None

agent_name → (key, enabled) mapping for client-driven conversation history. Every agent except the chat-owning agent receives a :class:~kaboo_workflows.hooks.HistoryHook so its transcript is seeded from and captured back into the request's history exchange. The chat owner is skipped — its transcript is the chat itself, owned by the AG-UI adapter.

 None
entry_name str | None

The configured name of the entry node. Stored on the EventQueue and used as agent_name on SESSION_START / SESSION_END events.

 None
chat_owner str | None

Name of the agent whose transcript IS the chat and must not receive a HistoryHook. For a delegate entry this is the forked manager's blueprint agent (e.g. coordinator), which differs from entry_name (the orchestration name). Defaults to entry_name.

 None
chat_reply str | None

Name of the agent whose streamed text becomes the chat reply (the chat bubble). Its activity card is flagged is_chat_reply so the UI suppresses only its duplicate text. For swarm/graph this is the chat_output node; for delegate it is the manager agent.

 None
session_id str | None

The effective session id. Stored on the EventQueue and included in the SESSION_END event payload.

 None

Returns:

Type Description
EventQueue

A ready-to-use :class:EventQueue.
Source code in src/kaboo_workflows/wire.py 
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
def make_event_queue(
    agents: dict[str, Agent],
    *,
    orchestrators: dict[str, Node] | None = None,
    tool_labels: dict[str, str] | None = None,
    stream_groups: dict[str, tuple[str, str]] | None = None,
    history: dict[str, tuple[str, bool]] | None = None,
    entry_name: str | None = None,
    chat_owner: str | None = None,
    chat_reply: str | None = None,
    session_id: str | None = None,
) -> EventQueue:
    """Attach :class:`~kaboo_workflows.hooks.EventPublisher` hooks to agents.

    Every agent in *agents* receives an :class:`.EventPublisher` hook and a
    matching ``callback_handler`` so all per-agent event types flow into the
    returned :class:`EventQueue`.  Orchestrators (Swarm / Graph / delegate
    Agent) in *orchestrators* also get a publisher for NODE_START, NODE_STOP,
    HANDOFF, and MULTIAGENT_COMPLETE events.

    This function does **not** emit SESSION_START.  Callers that own a
    :class:`~kaboo_workflows.types.SessionManifest` should call
    :meth:`EventQueue.emit_session_start` themselves; the common
    :class:`ResolvedConfig` workflow does this for you via
    :meth:`ResolvedConfig.wire_event_queue`.

    .. warning::

        This function **mutates** the passed-in agents and orchestrators by
        adding hooks and overwriting ``callback_handler``.  Call it only once
        per set of agents.

    Args:
        agents: Agents to wire, keyed by name.
        orchestrators: Built orchestrations keyed by name.
        tool_labels: Tool name → display label mapping forwarded to each
            :class:`.EventPublisher`.  Defaults to
            ``{name: "Delegating work to agent: <Name>"}`` for every agent.
        stream_groups: agent_name → (group, title) mapping for
            hierarchical activity attribution. When an agent is not in this
            dict it uses its own name as the group.
        history: agent_name → (key, enabled) mapping for client-driven
            conversation history. Every agent except the chat-owning agent
            receives a :class:`~kaboo_workflows.hooks.HistoryHook` so its
            transcript is seeded from and captured back into the request's
            history exchange. The chat owner is skipped — its transcript is the
            chat itself, owned by the AG-UI adapter.
        entry_name: The configured name of the entry node.  Stored on the
            EventQueue and used as ``agent_name`` on SESSION_START /
            SESSION_END events.
        chat_owner: Name of the agent whose transcript IS the chat and must not
            receive a ``HistoryHook``. For a delegate entry this is the forked
            manager's blueprint agent (e.g. ``coordinator``), which differs from
            *entry_name* (the orchestration name). Defaults to *entry_name*.
        chat_reply: Name of the agent whose streamed text becomes the chat reply
            (the chat bubble). Its activity card is flagged ``is_chat_reply`` so
            the UI suppresses only its duplicate text. For swarm/graph this is
            the ``chat_output`` node; for delegate it is the manager agent.
        session_id: The effective session id.  Stored on the EventQueue and
            included in the SESSION_END event payload.

    Returns:
        A ready-to-use :class:`EventQueue`.
    """
    event_queue = EventQueue(
        asyncio.Queue(maxsize=10000),
        entry_name=entry_name,
        session_id=session_id,
    )

    labels = {
        **{name: f"Delegating work to agent: {name.title()}" for name in agents},
        **(tool_labels or {}),
    }

    groups = stream_groups or {}
    history_map = history or {}
    history_skip = chat_owner or entry_name

    for name, agent in agents.items():
        sg, st = groups.get(name, (name, name.replace("_", " ").title()))
        pub = EventPublisher(
            callback=event_queue._put,
            agent_name=name,
            tool_labels=labels,
            stream_group=sg,
            stream_title=st,
            is_chat_reply=(chat_reply is not None and name == chat_reply),
        )
        agent.hooks.add_hook(pub)
        agent.callback_handler = pub.as_callback_handler()
        # Stash the instance so the AG-UI adapter can forward it to the per-thread
        # clone ag-ui-strands executes for a plain-agent entry (that clone does not
        # inherit this blueprint's HookRegistry). See create_agui_app.
        setattr(agent, "_kaboo_event_publisher", pub)
        logger.debug("agent=<%s>, stream_group=<%s> | wired EventPublisher", name, sg)

        # Client-driven history for every agent except the chat owner, whose
        # transcript is the chat itself (owned by the AG-UI adapter). For a
        # delegate entry the chat owner is the manager's blueprint agent, which
        # differs from the orchestration's entry_name.
        if name != history_skip:
            key, enabled = history_map.get(name, (sg, False))
            agent.hooks.add_hook(HistoryHook(key, enabled=enabled))
            logger.debug(
                "agent=<%s>, history_key=<%s>, enabled=<%s> | wired HistoryHook",
                name,
                key,
                enabled,
            )

    for orch_name, orch in (orchestrators or {}).items():
        if not isinstance(orch, (Swarm, Graph, Agent)):
            continue
        sg, st = groups.get(orch_name, (orch_name, orch_name.replace("_", " ").title()))
        orch_pub = EventPublisher(
            callback=event_queue._put,
            agent_name=orch_name,
            tool_labels=labels,
            stream_group=sg,
            stream_title=st,
        )
        orch.hooks.add_hook(orch_pub)
        if isinstance(orch, Agent):
            orch.callback_handler = orch_pub.as_callback_handler()
            # Stash for the AG-UI adapter, same as declared agents above: the
            # per-thread clone it executes for a delegate entry does not inherit
            # this HookRegistry, so the publisher must be forwarded explicitly
            # (its AGENT_COMPLETE carries the manager's token usage).
            setattr(orch, "_kaboo_event_publisher", orch_pub)
        logger.debug("orchestrator=<%s>, stream_group=<%s> | wired EventPublisher", orch_name, sg)

    return event_queue