API¶

Core app & server¶

Shared FastAPI app factory for Klea packages.

File: klea_utils/api/app.py

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:

Instantiates and sets up the graph via graph_factory
Opens a persistent SessionStore at {graph.paths.user_data_dir}/sessions.db alongside the graph’s checkpoints.
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

Apply a --profile value for the current process.

The profile is carried into the app module through the environment: config_env_var is set to <name>.json and left in place (the env var takes precedence over the env file in pydantic-settings). The profile name is validated against the working directory / config_dir first so a typo fails fast.

template is special: it writes a scaffold config into the working directory and exits, never launching anything.

Parameters:

profile – Raw --profile value (a trailing .json is stripped); None is a no-op
config_env_var – Environment variable that names the config file, or None to skip setting one
config_dir – Config directory used for validation; None skips the fast-fail check
template_writer – Callable that writes a config template into the working directory and returns its path; None disables --profile template

Raises:

typer.BadParameter – If the profile does not resolve to a file
typer.Exit – After writing a template

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, config_env_var: str | None = None, config_dir: str | Path | None = None, template_writer: Callable[[Path], Path] | None = None) → 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".

The serve command accepts a --profile option that selects the JSON config file for the app. The value is validated (see configure_profile()); when config_env_var is given it is forwarded to the app module through that environment variable, so a profile dropped in the working directory or the config directory is used without any other wiring. The special profile template scaffolds a new config and exits instead of launching.

Parameters:

app_module – Uvicorn module string
default_port – Default port number
config_env_var – Environment variable that carries the config file name into the app process (e.g. "KLEA_RAG_APP_CONFIG_FILE")
config_dir – Config directory searched after the working directory when validating --profile
template_writer – Callable that writes a config template into the working directory for --profile template

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, profile: str | None = None) → 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.

When profile is given but a server is already running, the profile cannot take effect (the running server was started with its own config) and a warning is printed.

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
profile – Config profile the caller requested (used only to warn when the requested profile cannot apply)

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 plumbing for Klea packages.

This module provides the generic plumbing the chat endpoints (/query and /query/stream) all need: readiness checks, session persistence, per-request model overrides, SSE framing, and error handling. The API contract itself is app-specific – each app’s api/chat.py defines its own ChatPayload model and its own endpoint functions, wired through the helpers here, so the router can expose whichever fields the app contract needs (e.g. the agent’s mode) without growing this shared module.

The enrich hook lets an app inject app-level events into the SSE stream (for example a context event carrying the agent operating mode), while the shared framing stays here.

File: klea_utils/api/chat_core.py

async klea_utils.api.chat_core.run_query(request: Request, *, query: str, user_id: str, chat_id: str, extra_state: dict[str, Any] | None = None, context_fields: dict[str, Any] | None = None) → str[source]¶

Run the graph via run_graph_invoke and persist the exchange.

Applies the stored per-chat model overrides for the duration of the call (via the LangGraph Runtime context, ADR-0033), maps graph errors onto HTTP status codes, and writes the user query + assistant answer to the session store.

Parameters:

request – Request carrying app.state.graph / chat_sessions
query – User query text
user_id – Persistent user identifier
chat_id – Chat conversation identifier
extra_state – Optional app-specific initial state fields passed to the graph invocation (e.g. the agent’s operating mode request).
context_fields – Optional app-defined per-run context fields that the plumbing forwards together with the framework-provided model_overrides slice (ADR-0033). The assembled dict is coerced/validated against the app’s registered context_schema at the graph boundary; KleaRunContext is extra="allow" (or the app subclasses it for typed fields). Apps wire frontend payload fields through this generic hook instead of forking chat_core.

Returns:

The assistant’s answer text

Note:

POST /query returns only the answer string; the session context (operating mode etc., ADR-0032) is not included. It is not produced on the bare ainvoke path (run_graph_invoke) – fetch it via /query/stream context events or the hydration endpoint GET /chat/{user_id}/{chat_id}/context.

Raises:

HTTPException – 400 on ValueError, 503 on RuntimeError, 500 on any other failure

klea_utils.api.chat_core.stream_response(request: Request, *, query: str, user_id: str, chat_id: str, enrich: Callable[[AsyncIterator[dict]], AsyncIterator[dict]] | None = None, extra_state: dict[str, Any] | None = None, context_fields: dict[str, Any] | None = None) → StreamingResponse[source]¶

Return a /query/stream SSE response for the graph’s events.

Applies the stored per-chat model overrides for the duration of the stream, persists the user query + final assistant answer on the complete event, and converts graph failures into error SSE events instead of dropping the stream.

Parameters:

request – Request carrying app.state.graph / chat_sessions
query – User query text
user_id – Persistent user identifier
chat_id – Chat conversation identifier
enrich – Optional async-generator wrapper applied to the raw run_graph_astream_events event stream before framing. Apps use it to inject app-specific events (e.g. a context event with the operating mode) or filter events. When None, every graph event is emitted unchanged.
extra_state – Optional app-specific initial state fields passed to the graph invocation (e.g. the agent’s operating mode request).
context_fields – Optional app-defined per-run context fields that the plumbing forwards together with the framework-provided model_overrides slice (ADR-0033). The assembled dict is coerced/validated against the app’s registered context_schema at the graph boundary; KleaRunContext is extra="allow" (or the app subclasses it for typed fields). Apps wire frontend payload fields through this generic hook instead of forking chat_core.

Returns:

A fastapi.responses.StreamingResponse SSE stream

klea_utils.api.chat_core.thread_id_for(user_id: str, chat_id: str) → str[source]¶: Return the checkpoint thread id for a {user_id}:{chat_id} pair.

Shared health check endpoint factory for Klea packages.

File: klea_utils/api/health.py

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

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

class klea_utils.api.sessions.CreateChatPayload(*, chat_id: Annotated[str, _PydanticGeneralMetadata(pattern='^[^:]+$')], 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

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.

Session context¶

Session-context projection endpoint.

Returns the session context for a chat – the app-defined projection of checkpointed graph state via BaseLangGraph.context_snapshot() (ADR-0032) – so frontends can restore the mode badge / selector on hydration without waiting for the next streamed query.

Design: the mode lives in checkpointed graph state, which is the single source of truth (ADR-0030 / ADR-0032). This endpoint reads it from the checkpoint directly (graph.checkpointer / graph.graph); nothing is duplicated into the sessions database. An unsent query is not a chat yet, so a thread with no checkpoint simply reports null context – the chat entity exists, only its projection is unset, so 200 with null is returned rather than a 404.

File: klea_utils/api/context.py

klea_utils.api.context.create_context_router() → APIRouter[source]¶

Create an APIRouter exposing the graph-level context projection.

GET /chat/{user_id}/{chat_id}/context: Return {"context": {...}} – the checkpointed session context projected by the app’s context_snapshot hook – or {"context": null} when the thread has no checkpoint yet (a chat that never ran a query) or the graph uses no checkpointer.

Model configuration¶

Per-session model configuration endpoints for runtime model switching.

File: klea_utils/api/models.py

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

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 = '', extra: dict | None = None) → 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": {...}}
context     {"type": "context", "data": {...}}  (graph-level session context)
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.
extra – Optional extra request fields merged into the POST body (e.g. an app-specific mode request, ADR-0030).

klea_utils.api.sse.stream_events_sync(query: str, chat_id: str, server_url: str, user_id: str = '', extra: dict | None = None) → 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.
extra – Optional extra request fields merged into the POST body (e.g. an app-specific mode request, ADR-0030).

Utilities¶

Utility functions for the Klea API layer.

File: klea_utils/api/utils.py

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.