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.

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.

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

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

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

Source code in src/kaboo_workflows/config/resolvers/orchestrations/builders.py

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,
) -> 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.
    """
    self._configs = configs
    self._global_sm_def = global_session_manager_def
    self._session_id = session_id
    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

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

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

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

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

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

    Example::

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

mark_inline_chat_owner

mark_inline_chat_owner()

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

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

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

register_hooks

register_hooks(registry, **kwargs)

Register hook callbacks for agent and multiagent events.

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

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

as_callback_handler

as_callback_handler()

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

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

Returns:

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

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

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

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

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

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

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

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

    return _handler

MaxToolCallsGuard

MaxToolCallsGuard(max_calls=25)

Bases: HookProvider

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

Initialize the MaxToolCallsGuard.

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

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

Parameters:

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

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

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

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

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

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

register_hooks

register_hooks(registry, **kwargs)

Register the BeforeToolCallEvent callback.

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

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

StopGuard

StopGuard(stop_check)

Bases: HookProvider

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

Initialize the StopGuard.

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

Parameters:

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

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

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

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

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

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

register_hooks

register_hooks(registry, **kwargs)

Register the before-tool-call guard.

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

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

ToolNameSanitizer

Bases: HookProvider

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

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

register_hooks

register_hooks(registry, **kwargs)

Register AfterModelCall and BeforeToolCall sanitization callbacks.

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

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

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

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

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

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

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

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()

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.

Raises:

Type	Description
RuntimeError	If any server fails to start or become ready.

Source code in src/kaboo_workflows/mcp/lifecycle.py

def start(self) -> 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.

    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.
    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

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

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

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

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

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

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

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

@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

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

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

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

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

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

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

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

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

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)

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

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())

    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

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.
    """
    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)

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

    normalized = normalize(merged)
    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

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

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

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

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

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

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

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

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

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

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

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

@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

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()
        logger.debug("orchestrator=<%s>, stream_group=<%s> | wired EventPublisher", orch_name, sg)

    return event_queue