API

Core app & server

Shared FastAPI app factory for Klea packages.

File: klea_utils/api/app.py

Copyright 2026 Ankur Sinha Author: Ankur Sinha <sanjay DOT ankur AT gmail DOT com>

klea_utils.api.app.make_app(graph_factory: Callable[[], BaseLangGraph], title: str = 'Klea API', version: str = '0.1.0', routers: list[APIRouter] | None = None) → FastAPI[source]

Create a FastAPI instance with a standard lifespan.

The lifespan:

  1. Instantiates and sets up the graph via graph_factory

  2. Opens a persistent SessionStore at {graph.paths.user_data_dir}/sessions.db alongside the graph’s checkpoints.

  3. Stores the graph and session store on app.state

Parameters:

  • graph_factory – Callable that returns a configured BaseLangGraph instance

  • title – API title (appears in OpenAPI docs)

  • version – API version (appears in OpenAPI docs)

  • routers – List of APIRouters to include on the app

Returns:

Configured FastAPI app

Shared server launcher factory for Klea packages.

File: klea_utils/api/server.py

Copyright 2026 Ankur Sinha Author: Ankur Sinha <sanjay DOT ankur AT gmail DOT com>

klea_utils.api.server.is_loopback_host(host: str) → bool[source]

Return True when host refers to the local machine.

Parameters:

host – Hostname from a server URL (e.g. "127.0.0.1")

Returns:

True for loopback addresses, False otherwise

klea_utils.api.server.make_serve_app(app_module: str, default_port: int = 8005) → Typer[source]

Create a Typer app that runs uvicorn on the given app_module.

The module string should be the importable path to a FastAPI app instance, e.g. "klea_rag.api.main:app".

Parameters:

  • app_module – Uvicorn module string

  • default_port – Default port number

Returns:

A typer.Typer app for use as a CLI entry point

klea_utils.api.server.spawn_server(app_module: str, host: str = '127.0.0.1', port: int = 8005, timeout: float = 180.0) → Iterator[Popen | None][source]

Context manager that runs an API server in a subprocess.

If a healthy server is already listening at host:port (probed via /health/ready), nothing is spawned and None is yielded, so the caller does not own the server’s lifecycle. Otherwise a uvicorn subprocess is spawned (stdout and stderr inherited so startup errors and app output are visible; access logs are disabled to keep the shared terminal clean, with full logging preserved in the server’s rotating log file), readiness is waited on, and the subprocess is terminated when the with block exits.

Parameters:

  • app_module – Uvicorn module string (e.g. "klea_rag.api.main:app")

  • host – Host to bind

  • port – Port to bind

  • timeout – Total seconds to wait for readiness after spawning

Returns:

The spawned subprocess.Popen (or None if an existing server was reused)

klea_utils.api.server.split_server_url(url: str, default_port: int = 8005) → tuple[str, int][source]

Return (host, port) parsed from url.

Falls back to 127.0.0.1 and default_port when the URL does not carry a hostname or port.

Parameters:

  • url – Server URL (e.g. http://127.0.0.1:8005)

  • default_port – Port to use when the URL omits one

Returns:

(host, port) suitable for binding a local server

Chat endpoints

Shared chat endpoint factory for Klea packages.

File: klea_utils/api/chat.py

Copyright 2026 Ankur Sinha Author: Ankur Sinha <sanjay DOT ankur AT gmail DOT com>

class klea_utils.api.chat.ChatPayload(*, query: str, chat_id: str, user_id: str = '')[source]

Bases: BaseModel

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

klea_utils.api.chat.create_chat_router() → APIRouter[source]

Create an APIRouter with /query and /query/stream endpoints.

The router reads the graph instance from request.app.state.graph (set by klea_utils.api.app.make_app()). Chat-session data is stored via the SessionStore on app.state.chat_sessions.

Shared health check endpoint factory for Klea packages.

File: klea_utils/api/health.py

Copyright 2026 Ankur Sinha Author: Ankur Sinha <sanjay DOT ankur AT gmail DOT com>

klea_utils.api.health.create_health_router() → APIRouter[source]

Create an APIRouter with /health/live and /health/ready endpoints.

Message history endpoints for chat sessions.

File: klea_utils/api/messages.py

Copyright 2026 Ankur Sinha Author: Ankur Sinha <sanjay DOT ankur AT gmail DOT com>

klea_utils.api.messages.create_messages_router() → APIRouter[source]

Create an APIRouter for chat message history.

GET /chat/{user_id}/{chat_id}/messages

Return all messages for a chat, oldest first.

Session management

Chat session CRUD endpoints.

NOTE: user_id is currently a browser-generated UUID taken from the URL path. For multi-user deployments this must be replaced with an authenticated identity (JWT / OAuth) extracted from the request context — otherwise any user can delete or rename another user’s chats by modifying the user_id in the URL.

File: klea_utils/api/sessions.py

Copyright 2026 Ankur Sinha Author: Ankur Sinha <sanjay DOT ankur AT gmail DOT com>

class klea_utils.api.sessions.CreateChatPayload(*, chat_id: str, title: str = '')[source]

Bases: BaseModel

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class klea_utils.api.sessions.UpdateChatPayload(*, title: str)[source]

Bases: BaseModel

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

klea_utils.api.sessions.create_sessions_router() → APIRouter[source]

Create an APIRouter for chat session CRUD.

Endpoints:

GET /chat/{user_id}

List all chats for the user.

POST /chat/{user_id}

Create a new chat. Title is auto-generated from coolname if not provided.

PATCH /chat/{user_id}/{chat_id}

Update a chat’s metadata (title).

DELETE /chat/{user_id}/{chat_id}

Remove a chat and all associated data.

DELETE /chat/{user_id}

Remove all chats, messages, and checkpoints for the user.

Persistent SQLite-backed store for chat session data.

Manages two tables alongside the LangGraph checkpoint DB:

  • chat_sessions (chat metadata, listing, and model overrides)

  • messages (curated Q&A history for chat display)

There is no separate state table. Graph state (plan, goal, tool_status, …) is read directly from the latest LangGraph checkpoint via graph.aget_state(thread_id) – the checkpoint DB is the canonical source and already stores the full deserialised state with no serialization round-trip.

File: klea_utils/api/sessions_db.py

Copyright 2026 Ankur Sinha Author: Ankur Sinha <sanjay DOT ankur AT gmail DOT com>

class klea_utils.api.sessions_db.SessionStore(db_path: str | Path)[source]

Bases: object

SQLite-backed persistent store for chat session data.

All public methods are thread-safe. The store auto-creates its schema on first connection.

Parameters:

db_path – Filesystem path to the SQLite database file.

add_message(user_id: str, chat_id: str, role: str, content: str, metadata: dict[str, Any] | None = None) → None[source]

Append a single message to a chat’s history.

add_messages(user_id: str, chat_id: str, messages: Sequence[dict[str, Any]]) → None[source]

Append multiple messages atomically.

Each dict must have role and content keys, and may have an optional metadata key.

clear_override(user_id: str, chat_id: str, role: str) → None[source]

Remove the model override for a single role in a chat.

clear_overrides(user_id: str, chat_id: str) → None[source]

Remove all model overrides for a chat.

close() → None[source]

Close the underlying SQLite connection.

create_chat(user_id: str, chat_id: str, title: str = '') → None[source]

Insert a chat row if it does not already exist.

delete_chat(user_id: str, chat_id: str) → None[source]

Remove a chat and all its associated data.

delete_user_chats(user_id: str) → None[source]

Remove all chats and messages for a user.

get_chat(user_id: str, chat_id: str) → dict[str, Any] | None[source]

Return a single chat or None.

get_messages(user_id: str, chat_id: str) → list[dict[str, Any]][source]

Return all messages for a chat, oldest first.

get_overrides(user_id: str, chat_id: str) → dict[str, dict[str, Any]][source]

Return per-role model overrides keyed by role.

Returns {"rag": {"model": "...", "provider": "..."}, ...}

list_chats(user_id: str) → list[dict[str, Any]][source]

Return all chats for user_id, newest first.

rename_chat(user_id: str, chat_id: str, title: str) → None[source]

Update the display title of a chat.

set_override(user_id: str, chat_id: str, role: str, config: dict[str, Any]) → None[source]

Set or replace model overrides for a given role.

touch_chat(user_id: str, chat_id: str) → None[source]

Bump updated_at without changing any other field.

Model configuration

Per-session model configuration endpoints for runtime model switching.

File: klea_utils/api/models.py

Copyright 2026 Ankur Sinha Author: Ankur Sinha <sanjay DOT ankur AT gmail DOT com>

class klea_utils.api.models.ChatModelConfigPayload(*, model: str, api_key: str | None = None, base_url: str | None = None, provider: str | None = None, user_id: str = '')[source]

Bases: BaseModel

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

klea_utils.api.models.create_models_router() → APIRouter[source]

Create an APIRouter for per-chat model configuration.

GET /chat/{user_id}/{chat_id}/models/overrides

Returns stored model overrides for a chat.

GET /chat/{user_id}/{chat_id}/models/active

Returns resolved model config (defaults merged with overrides).

POST /chat/{user_id}/{chat_id}/models/overrides/{role}

Stores per-chat model overrides.

SSE streaming

Shared SSE streaming client for Klea frontends.

Provides both an async generator (for NiceGUI and TUI) and a synchronous generator (for Streamlit) that consume the /query/stream SSE endpoint and yield parsed event dicts.

File: klea_utils/api/sse.py

Copyright 2026 Ankur Sinha Author: Ankur Sinha <sanjay DOT ankur AT gmail DOT com>

async klea_utils.api.sse.fetch_active_models(server_url: str, user_id: str, chat_id: str) → dict[str, dict[str, str]][source]

Fetch the resolved model config per role for a chat.

Calls GET /chat/{user_id}/{chat_id}/models/active and returns the merged default + override config dict.

Parameters:

  • server_url – Base URL of the backend API server.

  • user_id – Opaque persistent user identifier.

  • chat_id – Chat conversation identifier.

Returns:

{"chat": {"model": "...", "provider": "..."}, "guard": ..., "embedding": ...}

klea_utils.api.sse.fetch_active_models_sync(server_url: str, user_id: str, chat_id: str) → dict[str, dict[str, str]][source]

Synchronous counterpart of fetch_active_models().

Intended for frontends that cannot use asyncio.

Parameters:

  • server_url – Base URL of the backend API server.

  • user_id – Opaque persistent user identifier.

  • chat_id – Chat conversation identifier.

klea_utils.api.sse.format_model_info(info: dict[str, dict[str, str]]) → str[source]

Build a compact one-line model summary from active models config.

Strips provider prefixes and joins roles, e.g.:

Chat:deepseek-v4-flash | Guard:llama-guard3 | Embedding:bge-m3
Parameters:

info – The dict returned by fetch_active_models / fetch_active_models_sync.

Returns:

Empty string if no models are configured.

async klea_utils.api.sse.stream_events(query: str, chat_id: str, server_url: str, user_id: str = '') → AsyncGenerator[dict, None][source]

POST to /query/stream and yield parsed SSE event dicts.

Each yielded dict has at least a "type" key. Known types:

progress    {"type": "progress", "node": "<label>"}
info        {"type": "info", "node": "<label>", "data": {...}}
debug       {"type": "debug", "node": "<label>", "data": {...}}
token       {"type": "token", "content": "<chunk>", "node": "<label>"}
usage       {"type": "usage", "node": "<label>", "data": {...}}
complete    {"type": "complete", "message_for_user": "<text>"}
error       {"type": "error", "message": "<text>", "error_type": "<class>", "node": "<label>"}

This async generator is intended for NiceGUI and TUI frontends.

Parameters:

  • query – User’s query string.

  • chat_id – Chat conversation identifier.

  • server_url – Base URL of the backend API server.

  • user_id – Opaque persistent user identifier.

klea_utils.api.sse.stream_events_sync(query: str, chat_id: str, server_url: str, user_id: str = '') → Generator[dict, None, None][source]

Synchronous counterpart of stream_events().

Intended for frontends that cannot use asyncio. Async frontends (NiceGUI, TUI) should use stream_events() instead.

Parameters:

  • query – User’s query string.

  • chat_id – Chat conversation identifier.

  • server_url – Base URL of the backend API server.

  • user_id – Opaque persistent user identifier.

Utilities

Utility functions for the Klea API layer.

File: klea_utils/api/utils.py

Copyright 2026 Ankur Sinha Author: Ankur Sinha <sanjay DOT ankur AT gmail DOT com>

async klea_utils.api.utils.check_api_is_ready(url: str, attempts: int | None = None, timeout: float = 180.0)[source]

Exponentially back off checking that the API is ready.

Parameters:

  • url – Health check endpoint URL

  • attempts – If set, maximum number of probe attempts (overrides timeout)

  • timeout – Total wall-clock seconds to keep probing when attempts is unset

klea_utils.api.utils.validate_url(value: str) → str[source]

Return value if it is a valid HTTP(S) URL, else raise ValueError.