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

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

Parameters:

Name	Type	Description	Default
config_path	str | Path	Path to the kaboo-workflows YAML config.	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

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,
) -> FastAPI:
    """Create a FastAPI app serving AG-UI SSE from a YAML config.

    Args:
        config_path: Path to the kaboo-workflows YAML config.
        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.

    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()
    resolved = load(str(config_path))

    entry = resolved.entry
    mcp_lifecycle = resolved.mcp_lifecycle

    entry_name = _resolve_entry_name(entry, resolved)
    event_queue = resolved.wire_event_queue()
    registry = ActivityRegistry()

    entry_inline = False
    if isinstance(entry, MultiAgentBase):
        # First-class Swarm/Graph entry — kaboo owns the run loop.
        chat_output = _resolve_chat_output(getattr(resolved, "app_config", None))
        agui_agent: StrandsAgent | StrandsMultiAgent = StrandsMultiAgent(
            entry, name=entry_name, chat_output=chat_output
        )
    elif isinstance(entry, Agent):
        # Plain agent or delegate (a forked Agent) — ag-ui-strands drives it.
        # Forward the entry node's kaboo hook providers so interrupt/HITL hooks
        # fire on the per-thread clone ag-ui-strands actually executes (the
        # clone does not inherit the blueprint's HookRegistry).
        forwarded_hooks = get_agent_hook_providers(entry)

        # A plain-agent entry (not a delegate) also forwards its EventPublisher so
        # its own tool calls land in the activity stream and enrich the inline tool
        # rows (labels, formatted results, error status) — otherwise those rows
        # render bare from the raw AG-UI message. The publisher goes FIRST so
        # TOOL_START is emitted before any HITL gate interrupts, registering the
        # tool in run 1 so its result updates on resume. Its group is flagged
        # inline_chat_owner so the UI enriches the rows but never draws a card.
        entry_pub = getattr(entry, "_kaboo_event_publisher", None)
        is_delegate_entry = _resolve_chat_owner(getattr(resolved, "app_config", None)) is not None
        if entry_pub is not None and not is_delegate_entry:
            entry_pub.mark_inline_chat_owner()
            forwarded_hooks = [entry_pub, *forwarded_hooks]
            entry_inline = True

        agui_agent = StrandsAgent(
            agent=entry,
            name=entry_name,
            config=_build_agui_config(resolved),
            hooks=forwarded_hooks,
        )
    else:
        raise TypeError(f"Unsupported entry node type: {type(entry)}")

    @asynccontextmanager
    async def lifespan(app: FastAPI) -> AsyncIterator[None]:
        logger.info("AG-UI server starting — MCP lifecycle already active")
        try:
            yield
        finally:
            logger.info("AG-UI server shutting down — stopping MCP lifecycle")
            await event_queue.close()
            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,
        agui_agent,
        event_queue,
        registry,
        entry_name,
        endpoint,
        entry_inline=entry_inline,
    )
    if ping_path is not None:
        add_ping(app, ping_path)

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

    return app