kaboo_workflows.adapters

The serving layer: turn a resolved config into an AG-UI / CopilotKit app.

kaboo_workflows.adapters

AG-UI adapter for kaboo-workflows.

create_agui_app

create_agui_app(
    config_path,
    *,
    endpoint="/invocations",
    ping_path="/ping",
    cors_origins=None,
    cors_allow_credentials=True,
    auth=None,
    session_config_key=None,
    allowed_mcp_hosts=None,
)

Create a FastAPI app serving AG-UI SSE from a YAML config.

By default the process serves one config: it is loaded and checked here, and each conversation gets its own agents the first time it runs, keeping them for its later turns, so no two conversations share a sub-agent's state.

Setting session_config_key makes the service behave like a function instead. Each run submits its own config in forwardedProps under that key, layered over config_path as an overlay (see :func:~kaboo_workflows.config.load_session_config), and gets its own agents, orchestration, entry and MCP client sessions, all released when the run ends. Different runs can then be different workflows, and a restart or a second replica behaves like a cold instance because nothing is kept between runs.

That is only safe because conversation state does not live in the objects being rebuilt: history and pending interrupts arrive with each turn on the AG-UI state channel (see :class:~kaboo_workflows.hooks.SessionStateHook).

Parameters:

Name	Type	Description	Default
config_path	str | Path	Path to the kaboo-workflows YAML config. With session_config_key set this is the base every submitted config layers over — typically shared models, MCP clients and defaults.	required
endpoint	str	Path for the AG-UI agent endpoint.	'/invocations'
ping_path	str | None	Path for the health check endpoint. None to disable.	'/ping'
cors_origins	list[str] | None	Allowed CORS origins. Defaults to ["*"]. Pass an explicit allow-list for production deployments.	None
cors_allow_credentials	bool	Whether to allow credentialed CORS requests.	True
auth	AuthVerifier | None	Optional inbound auth verifier. Receives each /invocations request and returns a :class:~kaboo_workflows._context.Principal (or None for an anonymous caller); raise to reject (e.g. :class:fastapi.HTTPException -> 401). The resolved identity is bound to the request context so outbound MCP auth strategies can relay/exchange from it. When None (default) the endpoint trusts its caller — only safe behind an authenticating proxy or the AgentCore Runtime authorizer.	None
session_config_key	str | None	forwardedProps key carrying this run's config. None (default) serves config_path alone, unchanged.	None
allowed_mcp_hosts	list[str] | None	Hosts a submitted config's MCP client URL may point at. Set this whenever configs are authored anywhere but this repository: an arbitrary URL needs no code to exfiltrate.	None

Returns:

Type	Description
FastAPI	A FastAPI application with AG-UI SSE streaming.

Raises:

Type	Description
TypeError	If the entry node is an unsupported orchestration type.

Source code in src/kaboo_workflows/adapters/agui.py

def create_agui_app(
    config_path: str | Path,
    *,
    endpoint: str = "/invocations",
    ping_path: str | None = "/ping",
    cors_origins: list[str] | None = None,
    cors_allow_credentials: bool = True,
    auth: AuthVerifier | None = None,
    session_config_key: str | None = None,
    allowed_mcp_hosts: list[str] | None = None,
) -> FastAPI:
    """Create a FastAPI app serving AG-UI SSE from a YAML config.

    By default the process serves one config: it is loaded and checked here, and
    each conversation gets its own agents the first time it runs, keeping them
    for its later turns, so no two conversations share a sub-agent's state.

    Setting ``session_config_key`` makes the service behave like a function
    instead. Each run submits its own config in ``forwardedProps`` under that key,
    layered over ``config_path`` as an overlay (see
    :func:`~kaboo_workflows.config.load_session_config`), and gets its own agents,
    orchestration, entry and MCP client sessions, all released when the run ends.
    Different runs can then be different workflows, and a restart or a second
    replica behaves like a cold instance because nothing is kept between runs.

    That is only safe because conversation state does not live in the objects
    being rebuilt: history and pending interrupts arrive with each turn on the
    AG-UI state channel (see
    :class:`~kaboo_workflows.hooks.SessionStateHook`).

    Args:
        config_path: Path to the kaboo-workflows YAML config. With
            ``session_config_key`` set this is the base every submitted config
            layers over — typically shared models, MCP clients and defaults.
        endpoint: Path for the AG-UI agent endpoint.
        ping_path: Path for the health check endpoint. ``None`` to disable.
        cors_origins: Allowed CORS origins. Defaults to ``["*"]``. Pass an
            explicit allow-list for production deployments.
        cors_allow_credentials: Whether to allow credentialed CORS requests.
        auth: Optional inbound auth verifier. Receives each ``/invocations``
            request and returns a
            :class:`~kaboo_workflows._context.Principal` (or ``None`` for an
            anonymous caller); raise to reject (e.g.
            :class:`fastapi.HTTPException` -> 401). The resolved identity is
            bound to the request context so outbound MCP auth strategies can
            relay/exchange from it. When ``None`` (default) the endpoint trusts
            its caller — only safe behind an authenticating proxy or the
            AgentCore Runtime authorizer.
        session_config_key: ``forwardedProps`` key carrying this run's config.
            ``None`` (default) serves ``config_path`` alone, unchanged.
        allowed_mcp_hosts: Hosts a submitted config's MCP client URL may point
            at. Set this whenever configs are authored anywhere but this
            repository: an arbitrary URL needs no code to exfiltrate.

    Returns:
        A FastAPI application with AG-UI SSE streaming.

    Raises:
        TypeError: If the entry node is an unsupported orchestration type.
    """
    config_path = Path(config_path).resolve()
    per_run_configs = session_config_key is not None

    # Parse the base once. With per-run configs it stays raw, because each run
    # validates its own merge of it; a run's overlay may legitimately replace
    # sections the base leaves incomplete.
    base_raw = parse_config_sources(str(config_path))
    app_config = validate_raw_config(base_raw)
    _install_reference_fetching(app_config)

    # Process-wide, idempotent, and a no-op when telemetry.enabled is false.
    # Initialized from the base config only — per-run overlays cannot toggle
    # the global tracer provider.
    init_telemetry(app_config.telemetry)

    infra = resolve_infra(app_config)
    # Servers are processes and stay process-wide either way. Clients are only
    # pinned for the process when the process owns them; per-run clients are
    # deliberately allowed to end with their run.
    infra.mcp_lifecycle.start(pin_clients=not per_run_configs)

    resolve_session: Callable[[RunAgentInput], AguiSession]
    static_session: AguiSession | None = None
    conversations: _ConversationSessions | None = None
    if session_config_key is None:
        # One config for the process: build it here so a broken config fails at
        # startup rather than on the first request. Each conversation then runs
        # on its own session.
        static_session = _build_session(app_config, infra)
        resolve_session = conversations = _ConversationSessions(app_config, infra)
    else:
        resolve_session = _make_session_resolver(
            base_raw,
            infra,
            session_config_key=session_config_key,
            allowed_mcp_hosts=allowed_mcp_hosts,
        )

    @asynccontextmanager
    async def lifespan(app: FastAPI) -> AsyncGenerator[None]:
        logger.info("AG-UI server starting — MCP lifecycle already active")
        try:
            yield
        finally:
            logger.info("AG-UI server shutting down — stopping MCP lifecycle")
            if static_session is not None:
                await static_session.event_queue.close()
            for session in conversations.kept() if conversations is not None else []:
                await session.event_queue.close()
            infra.mcp_lifecycle.stop()

    app = FastAPI(title="kaboo-workflows", lifespan=lifespan)

    app.add_middleware(
        CORSMiddleware,
        allow_origins=cors_origins if cors_origins is not None else ["*"],
        allow_credentials=cors_allow_credentials,
        allow_methods=["*"],
        allow_headers=["*"],
    )

    _add_kaboo_endpoint(app, resolve_session, endpoint, auth=auth)
    if ping_path is not None:
        add_ping(app, ping_path)

    @app.get("/manifest")
    async def manifest() -> dict:
        return {"entry": static_session.entry_name if static_session else app_config.entry}

    capabilities = agent_capabilities(app_config, per_run_configs=per_run_configs)

    @app.get(f"{endpoint.rstrip('/')}/capabilities")
    async def get_capabilities() -> dict:
        return capabilities.model_dump(by_alias=True, exclude_none=True)

    return app