kaboo_workflows.config¶

Config schema, loaders, interpolation, and infrastructure resolution.

kaboo_workflows.config ¶

YAML configuration loading, validation, and resolution.

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:

Builds a :class:~kaboo_workflows.types.SessionManifest from the resolved runtime objects.
Auto-resolves stream groups from the original AppConfig so every agent is attributed to a hierarchical dot-path group.
Collects tool labels from MCP clients and agents in the original AppConfig (agent-level overrides MCP-level).
Wires every agent (and orchestrator) with an :class:~kaboo_workflows.hooks.EventPublisher via :func:~kaboo_workflows.wire.make_event_queue.
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, ...)

AgentDef ¶

Bases: BaseModel

Top-level agent definition.

All agents are defined flat under the agents: section. Multi-agent orchestration is configured separately in the orchestrations: section.

tools accepts spec strings:

"module.path:function_name" — single function from module
"module.path" — all @tool functions in module
"./path/to/file.py:function_name" — single function from file
"./path/to/file.py" — all @tool functions in file
"./path/to/dir/" — all @tool functions in directory

hooks accepts import-path strings ("module.path:ClassName" or "./file.py:ClassName") or inline :class:HookDef objects with explicit type + optional params.

type `class-attribute` `instance-attribute` ¶

type = None

Custom agent factory import path.

Format: module.path:ClassName or ./file.py:ClassName. When set, the factory is called instead of strands.Agent() directly. The agent_kwargs dict is spread as **kwargs to this factory.

agent_kwargs `class-attribute` `instance-attribute` ¶

agent_kwargs = Field(default_factory=dict)

Additional keyword arguments passed to strands.Agent() or custom factory.

Valid Agent parameters: messages, callback_handler, record_direct_tool_call, trace_attributes, state, plugins, structured_output_prompt, structured_output_model, tool_executor, retry_strategy, concurrent_invocation_mode, load_tools_from_directory.

Warning: Use at your own risk — no schema-level validation is performed. Agent.init has 24 explicit parameters and no **kwargs; any invalid key will raise TypeError at construction time.

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.

ConversationManagerDef ¶

Bases: BaseModel

Conversation manager configuration.

type must be a module.path:ClassName import path or a ./file.py:ClassName file-based import path. The resolver raises ValueError if there is no colon separator. params are forwarded as constructor kwargs.

Built-in strands classes:

strands.agent:SlidingWindowConversationManager
strands.agent:SummarizingConversationManager
strands.agent:NullConversationManager

DelegateConnectionDef ¶

Bases: BaseModel

A delegation connection: orchestrator calls agent as a tool.

DelegateOrchestrationDef ¶

Bases: BaseModel

Delegate mode: entry agent calls other agents as tools.

A new Agent is constructed from the entry_name agent's blueprint (model, system_prompt, hooks, tools, etc.) with delegate tools added for each connection. The original agent is never mutated.

Entry point is explicit via entry_name (consistent with swarm/graph).

agent_kwargs is merged over the entry agent's agent_kwargs — orchestration values win on conflict, unset keys are inherited.

agent_kwargs `class-attribute` `instance-attribute` ¶

agent_kwargs = Field(default_factory=dict)

Merged over the entry agent's agent_kwargs (orchestration wins).

Common uses: system_prompt, callback_handler, conversation_manager.

reference_fields `classmethod` ¶

reference_fields()

Return mapping of JSON paths to reference types for name sanitization.

Source code in src/kaboo_workflows/config/schema.py

@classmethod
def reference_fields(cls) -> dict[str, str]:
    """Return mapping of JSON paths to reference types for name sanitization."""
    return {
        "entry_name": "node",
        "connections[].agent": "node",
    }

GraphEdgeDef ¶

Bases: BaseModel

An edge in a graph orchestration.

GraphOrchestrationDef ¶

Bases: BaseModel

Graph mode: DAG-based orchestration with conditional edges.

Agents execute in parallel batches based on dependency order. Uses strands Graph under the hood.

chat_output names the node whose streamed text becomes the assistant reply when the graph is a first-class AG-UI entry. When omitted, the adapter streams the sole terminal node (a node with no outgoing edges) if there is exactly one, otherwise it emits the final node's text on completion.

node_ids ¶

node_ids()

Return every node id referenced by this graph (entry + edge ends).

Source code in src/kaboo_workflows/config/schema.py

def node_ids(self) -> set[str]:
    """Return every node id referenced by this graph (entry + edge ends)."""
    ids = {self.entry_name}
    for edge in self.edges:
        ids.add(edge.from_agent)
        ids.add(edge.to_agent)
    return ids

terminal_nodes ¶

terminal_nodes()

Return nodes with no outgoing edge (candidate chat-output nodes).

Source code in src/kaboo_workflows/config/schema.py

def terminal_nodes(self) -> list[str]:
    """Return nodes with no outgoing edge (candidate chat-output nodes)."""
    sources = {edge.from_agent for edge in self.edges}
    return sorted(n for n in self.node_ids() if n not in sources)

reference_fields `classmethod` ¶

reference_fields()

Return mapping of JSON paths to reference types for name sanitization.

Source code in src/kaboo_workflows/config/schema.py

@classmethod
def reference_fields(cls) -> dict[str, str]:
    """Return mapping of JSON paths to reference types for name sanitization."""
    return {
        "entry_name": "node",
        "chat_output": "node",
        "edges[].from": "node",
        "edges[].to": "node",
    }

HookDef ¶

Bases: BaseModel

Hook provider reference.

type must be a module.path:ClassName import path or a ./file.py:ClassName file-based import path. The resolver raises ValueError if there is no colon separator. params are forwarded as constructor kwargs.

MCPClientDef ¶

Bases: BaseModel

MCP client connection definition.

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

params are forwarded to strands MCPClient (e.g., startup_timeout, tool_filters, prefix). transport_options are forwarded to the transport factory (e.g., headers, auth, timeout, http_client).

MCPServerDef ¶

Bases: BaseModel

MCP server definition.

ModelDef ¶

Bases: BaseModel

LLM model configuration.

SessionManagerDef ¶

Bases: BaseModel

Session manager configuration.

Built-in providers: "file", "s3", "agentcore".

For a custom class, set type to an import path ("module.path:ClassName"). The class must be a subclass of strands.session.SessionManager. When type is set, provider is ignored.

Session ID resolution order

Runtime override (e.g., HTTP session header)
params.session_id
Random UUID (fresh session per CLI run)

SwarmOrchestrationDef ¶

Bases: BaseModel

Swarm mode: collaborative handoffs between peer agents.

Agents transfer control to each other via handoff_to_agent tool. Uses strands Swarm under the hood.

chat_output names the member agent whose streamed text becomes the assistant reply when the swarm is a first-class AG-UI entry. When omitted, the adapter emits the final active node's text once the run completes.

reference_fields `classmethod` ¶

reference_fields()

Return mapping of JSON paths to reference types for name sanitization.

Source code in src/kaboo_workflows/config/schema.py

@classmethod
def reference_fields(cls) -> dict[str, str]:
    """Return mapping of JSON paths to reference types for name sanitization."""
    return {
        "entry_name": "node",
        "chat_output": "node",
        "agents[]": "node",
    }

interpolate ¶

interpolate(raw, *, variables=None, env=None)

Interpolate ${VAR} and ${VAR:-default} references in a YAML config dict.

Lookup order: variables dict -> env dict -> default value -> raise error.

Uses a two-pass strategy to resolve cross-variable references inside the vars: block before interpolating the rest of the config. Pass 1 resolves each var against env only; Pass 2 resolves each var sequentially against the pass-1 results so chains like B: "${A}y" work correctly. Any ${VAR} pattern still present after two passes indicates a circular or undefined reference and raises ValueError.

Parameters:

Name	Type	Description	Default
`raw`	`dict[str, Any]`	Raw parsed YAML dict (will not be mutated — returns a new dict).	required
`variables`	`dict[str, Any] \| None`	User-defined variables (from the `vars:` block in YAML). Values may be any type; non-string values are preserved when the entire string is a single `${VAR}` reference.	`None`
`env`	`dict[str, str] \| None`	Environment variables (defaults to `os.environ` if None).	`None`

Returns:

Type	Description
`dict[str, Any]`	New dict with all string values interpolated.

Raises:

Type	Description
`ValueError`	If a variable is referenced but not found and has no default, or if a circular reference is detected in the vars block.

Source code in src/kaboo_workflows/config/interpolation.py

def interpolate(
    raw: dict[str, Any],
    *,
    variables: dict[str, Any] | None = None,
    env: dict[str, str] | None = None,
) -> dict[str, Any]:
    """Interpolate ${VAR} and ${VAR:-default} references in a YAML config dict.

    Lookup order: variables dict -> env dict -> default value -> raise error.

    Uses a two-pass strategy to resolve cross-variable references inside the
    ``vars:`` block before interpolating the rest of the config.  Pass 1
    resolves each var against env only; Pass 2 resolves each var sequentially
    against the pass-1 results so chains like ``B: "${A}y"`` work correctly.
    Any ``${VAR}`` pattern still present after two passes indicates a circular
    or undefined reference and raises ``ValueError``.

    Args:
        raw: Raw parsed YAML dict (will not be mutated — returns a new dict).
        variables: User-defined variables (from the ``vars:`` block in YAML).
            Values may be any type; non-string values are preserved when the
            entire string is a single ``${VAR}`` reference.
        env: Environment variables (defaults to ``os.environ`` if None).

    Returns:
        New dict with all string values interpolated.

    Raises:
        ValueError: If a variable is referenced but not found and has no
            default, or if a circular reference is detected in the vars block.
    """
    resolved_vars = dict(variables or {})
    resolved_env = env if env is not None else dict(os.environ)

    # Pass 1: resolve vars against env only (lenient — keeps ${VAR} if absent)
    resolved_vars = {k: _walk_lenient(v, {}, resolved_env) for k, v in resolved_vars.items()}

    # Pass 2: resolve vars sequentially so each resolved var is immediately
    # available to subsequent entries (handles chains like A -> B -> C).
    pass2: dict[str, Any] = {}
    for k, v in resolved_vars.items():
        pass2[k] = _walk_lenient(v, pass2, resolved_env)
    resolved_vars = pass2

    # Validate: remaining ${...} means circular or undefined reference.
    for val in resolved_vars.values():
        if isinstance(val, str):
            m = _VAR_PATTERN.search(val)
            if m:
                var_name = m.group(1).split(":-")[0]
                raise ValueError(
                    f"Unresolved variable reference '${{{var_name}}}' in vars block.\n"
                    f"Check for circular references or undefined variables."
                )

    return _walk(raw, resolved_vars, resolved_env)

strip_anchors ¶

strip_anchors(raw)

Remove x-* keys (YAML anchor scratch pads) from the top level.

Parameters:

Name	Type	Description	Default
`raw`	`dict[str, Any]`	Raw parsed YAML dict.	required

Returns:

Type	Description
`dict[str, Any]`	New dict without top-level `x-*` keys.

Source code in src/kaboo_workflows/config/interpolation.py

def strip_anchors(raw: dict[str, Any]) -> dict[str, Any]:
    """Remove x-* keys (YAML anchor scratch pads) from the top level.

    Args:
        raw: Raw parsed YAML dict.

    Returns:
        New dict without top-level ``x-*`` keys.
    """
    return {k: v for k, v in raw.items() if not k.startswith("x-")}

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:¶

Parse each source (file read or inline YAML)
Per-source: strip anchors, interpolate variables
Sanitize collection keys (spaces/special chars -> underscores)
Merge sources (if multiple), detect duplicate names
Validate against schema (Pydantic)
Resolve infrastructure (models, MCP — pure; no session manager)
Start MCP servers (so clients can connect)
Create agents (Agent.init auto-starts MCP clients)
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:

Models (no dependencies)
MCP servers (no dependencies)
MCP clients (depend on servers)
MCP lifecycle (assembles servers + clients, not started)
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,
    )

kaboo_workflows.config¶