API

Core app & server

Shared FastAPI app factory for Klea packages.

See devdocs/system/api-sse-sequence.md and devdocs/system/c4-container.md; ADR-0031.

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.storage_root}/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.configure_profile(profile: str | None, config_env_var: str | None, config_dir: str | Path | None, template_writer: Callable[[Path], Path] | None) → None[source]

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.

Kept for backward compatibility; the implementation lives in klea_utils.net.is_loopback_host().

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, app_name: str = 'klea-serve') → 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

  • app_name – Process title to set (e.g. "klea-agent-serve"), so the server is identifiable in ps/top.

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, quiet: bool = False) → 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)

  • quiet – When True, discard the subprocess’s stdout/stderr (DEVNULL) instead of inheriting them. Used by a full-screen frontend (the Textual TUI) whose terminal must not be written to; server logs remain in its rotating log file.

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 runners.

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 runners 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.

The ordered request/lifecycle view of this module is documented in devdocs/system/api-sse-sequence.md; the emitted event catalogue is devdocs/system/streams.md. Keep both in sync when the contract changes.

Supporting helpers live in focused modules: klea_utils.api.chat_common (app-state, thread identity, cancel), klea_utils.api.overrides (model overrides/credentials), and klea_utils.api.hitl (human-in-the-loop, ADR-0046).

File: klea_utils/api/chat_core.py

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

klea_utils.api.chat_core.HEARTBEAT_INTERVAL_SECONDS = 15.0

Idle gap, in seconds, after which _heartbeat() emits a ping SSE frame. A long-running tool (e.g. run_command) emits no graph events while it runs, so without a heartbeat the client’s idle read timeout (300 s in sse.py) and intermediary proxies drop the stream. 15 s is well below both.

async klea_utils.api.chat_core.run_query(request: Request, *, query: str | None = None, user_id: str, chat_id: str, resume: bool = False, interrupt_response: Mapping[str, Any] | None = None, interrupt_cancel: bool = False, interrupt_id: str | None = None, extra_state: dict[str, Any] | None = None, context_fields: dict[str, Any] | None = None, client_context: ClientContext | 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 turn when the run starts and the assistant reply on success (so a failed turn is recorded and a resume completes it without duplication).

Parameters:

  • request – Request carrying app.state.graph / chat_sessions

  • query – User query text; required unless resume is true, and must be empty when resume is true

  • user_id – Persistent user identifier

  • chat_id – Chat conversation identifier

  • resume – Resume the thread’s last failed run from its checkpoint (invoked with None; no user turn is written)

  • interrupt_response – Answer a pending HITL interrupt (ADR-0046); the mapping is passed as Command(resume=...) and recorded as the user turn.

  • interrupt_cancel – Cancel a pending HITL interrupt (a terminal run).

  • interrupt_id – Optional id of the interrupt being answered; a mismatch with the pending interrupt is rejected as stale.

  • extra_state – Optional app-specific initial state fields passed to the graph invocation (e.g. the agent’s operating mode request). Ignored on resume.

  • 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.

  • client_context – Optional client-supplied environment for this run (cwd/env/interpreter). Accepted from loopback peers only; placed in the runtime context (never in graph state) for the bundled tool server.

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 a missing/extra query, an empty resume, or ValueError; 503 on RuntimeError; 500 on any other failure

async klea_utils.api.chat_core.stream_response(request: Request, *, query: str | None = None, user_id: str, chat_id: str, resume: bool = False, interrupt_response: Mapping[str, Any] | None = None, interrupt_cancel: bool = False, interrupt_id: str | None = None, enrich: Callable[[AsyncIterator[dict]], AsyncIterator[dict]] | None = None, extra_state: dict[str, Any] | None = None, context_fields: dict[str, Any] | None = None, client_context: ClientContext | 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 and writes the user turn when the run starts and the assistant reply on the complete event, so a turn that fails mid-run is still recorded and can be retried. Graph failures become error SSE events (with a resumable hint) instead of dropping the stream.

Parameters:

  • request – Request carrying app.state.graph / chat_sessions

  • query – User query text; required unless resume is true, and must be empty when resume is true

  • user_id – Persistent user identifier

  • chat_id – Chat conversation identifier

  • resume – Resume the thread’s last failed run from its checkpoint. The graph is invoked with None (no query) and the failed node is re-run; no user turn is written.

  • interrupt_response – Answer a pending HITL interrupt (ADR-0046); the mapping is passed as Command(resume=...) and recorded as the user turn.

  • interrupt_cancel – Cancel a pending HITL interrupt (a terminal run).

  • interrupt_id – Optional id of the interrupt being answered; a mismatch with the pending interrupt is rejected as stale.

  • 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). Ignored on resume.

  • 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.

  • client_context – Optional client-supplied environment for this run (cwd/env/interpreter). Accepted from loopback peers only; placed in the runtime context (never in graph state) for the bundled tool server.

Returns:

A fastapi.responses.StreamingResponse SSE stream

Shared chat request identity and app-state helpers.

These are the pieces every chat endpoint needs regardless of its contract: resolving the graph/session store from app.state, deriving a chat’s checkpoint thread id, the per-thread single-flight registry, and cancelling an in-flight run. The endpoint runners themselves live in klea_utils.api.chat_core.

See devdocs/system/api-sse-sequence.md and devdocs/system/streams.md.

File: klea_utils/api/chat_common.py

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

class klea_utils.api.chat_common.CancelPayload(*, chat_id: Annotated[str, _PydanticGeneralMetadata(pattern='^[^:]+$')], user_id: Annotated[str, _PydanticGeneralMetadata(pattern='^[^:]*$')] = '')[source]

Bases: BaseModel

Identity payload for POST /query/cancel (shared by both apps).

Cancellation addresses a chat’s thread, so it needs only the same (user_id, chat_id) identity the query payloads carry.

model_config = {}

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

klea_utils.api.chat_common.cancel_run(request: Request, user_id: str, chat_id: str) → bool[source]

Cancel the active run for a chat’s thread, if any.

Idempotent: an absent or already-completed run is a no-op. Cancelling the driving task raises CancelledError at the node’s next await, leaving the checkpoint resumable at the failed node; no assistant turn is persisted (only the complete event writes one).

Parameters:

  • request – The incoming request; carries the active-run registry.

  • user_id – Persistent user identifier.

  • chat_id – Chat conversation identifier.

Returns:

True when a live run was found and cancellation requested.

klea_utils.api.chat_common.thread_id_for(user_id: str, chat_id: str) → str[source]

Return the checkpoint thread id for a {user_id}:{chat_id} pair.

Per-run model override and credential resolution.

Assembles the model_overrides slice the endpoint runners pass to the graph as per-run runtime context (ADR-0033), and migrates legacy inline plaintext keys into the credential store. Kept separate from the endpoint runners (klea_utils.api.chat_core) because the concern is credential/override management, not request handling.

See devdocs/system/api-sse-sequence.md; ADRs 0033 and 0042.

File: klea_utils/api/overrides.py

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

klea_utils.api.overrides.migrate_legacy_overrides(store: SessionStore) → int[source]

Move legacy per-chat plaintext api_key values into credentials.

Pre-credential deployments stored the key inline in a chat’s role override. Move each into user_credentials (scoped to the model’s provider) and strip it from the override, so the unused-key TTL applies and override payloads no longer carry secrets. Returns the number of roles migrated.

klea_utils.api.overrides.resolve_model_overrides(graph: Any, store: SessionStore, user_id: str, chat_id: str) → dict[str, dict[str, Any]][source]

Assemble the per-run model_overrides slice.

Merges the per-session default overrides with the per-chat overrides (role-level: a chat’s role dict wins entirely), then injects the user’s provider credential (looked up by the model’s provider/endpoint scope) as api_key for every modifiable role with a resolved model. Locked roles are left to the graph/env so admin-managed models are not affected by per-user credentials.

Parameters:

  • graph – The graph, whose llm_models supply the role set and locked (modifiable=False) flags, plus env-default models.

  • store – Session store with the override and credential layers.

  • user_id – Persistent user identifier.

  • chat_id – Chat conversation identifier.

Returns:

The effective {role: config} for the runtime context.

Human-in-the-loop interrupt helpers for the chat endpoints (ADR-0046).

These read a thread’s pending interrupt from its checkpoint, render the question(s) and the user’s answer for the transcript, and validate an incoming request against the thread’s paused state (a paused thread may only be continued by an answer or a cancel). The endpoint runners in klea_utils.api.chat_core call _prepare_chat_request() before invoking the graph.

See devdocs/system/api-sse-sequence.md; ADR-0046.

File: klea_utils/api/hitl.py

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

klea_utils.api.hitl.render_interrupt_response(value: Mapping[str, Any]) → str[source]

Render an interrupt resume mapping as the user turn for the transcript.

Each interrupt kind has a renderer (see _RESPONSE_RENDERERS): input answers are joined; path-permission decisions are shown as “Allowed once: /tmp”, “Allowed for session: /proj”, “Denied: <key>” or “Not a path: <key>”; a review decision becomes “Plan approved” / “Revision requested: <feedback>”. An unknown shape falls back to its JSON form. This is display only: the wire values are unchanged.

Shared by chat_core (the persisted user row) and the web UI (the live transcript) so the two cannot drift.

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: 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, its messages, and its LangGraph checkpoint history.

DELETE /chat/{user_id}

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

Persistent SQLite-backed store for chat session data.

Manages four tables alongside the LangGraph checkpoint DB:

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

  • session_overrides (per-user default model overrides)

  • user_credentials (per-user provider API credentials)

  • 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.

See devdocs/adr/0023-sqlite-checkpointer.md and devdocs/system/api-sse-sequence.md; ADR-0042 (credentials/defaults).

File: klea_utils/api/sessions_db.py

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

klea_utils.api.sessions_db.CHAT_PREVIEW_MAX_CHARS = 200

Maximum characters of a chat’s first user query returned as its list preview. Bounds the session-list payload and the frontends’ display.

klea_utils.api.sessions_db.CREDENTIAL_PURGE_INTERVAL_SECONDS = 3600

Minimum gap between opportunistic credential purges.

klea_utils.api.sessions_db.CREDENTIAL_TOUCH_INTERVAL_SECONDS = 3600

Minimum gap between last_used_at bumps for a credential.

klea_utils.api.sessions_db.CREDENTIAL_TTL_ENV_VAR = 'KLEA_CREDENTIAL_TTL_DAYS'

Env var overriding the credential TTL, in days (0 disables expiry).

klea_utils.api.sessions_db.DEFAULT_CREDENTIAL_TTL_SECONDS = 604800

Default lifetime for an unused per-user API credential. Credentials untouched for longer are purged so plaintext keys are not retained indefinitely.

class klea_utils.api.sessions_db.SessionStore(db_path: str | Path, credential_ttl_seconds: float | None = None)[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_chat_events(user_id: str, chat_id: str, events: Sequence[dict[str, Any]]) → None[source]

Append UI events to a chat atomically, preserving insertion order.

Each event is stored whole (its JSON) so the frontend can replay it through the shared reducer; type is stored alongside for optional filtering. A created_at timestamp is stamped per batch.

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.

all_chat_overrides() → list[dict[str, Any]][source]

Return every chat that has (parsed) overrides.

Each entry is {user_id, chat_id, overrides}. Used by the startup migration that moves legacy per-chat api_key values into the provider credential store.

clear_credential(user_id: str, provider: str, endpoint: str = '') → None[source]

Remove the stored secret for a provider(+endpoint).

clear_credentials(user_id: str) → None[source]

Remove all stored credentials for user_id.

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.

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

Remove the per-user default override for a single role.

clear_session_overrides(user_id: str) → None[source]

Remove all per-user default overrides for user_id.

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, messages, overrides, and credentials for a user.

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

Return a single chat or None.

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

Return a chat’s stored UI events, oldest first.

get_credential(user_id: str, provider: str, endpoint: str = '') → str | None[source]

Return the stored secret for a provider(+endpoint), 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": "..."}, ...}

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

Return per-role default model overrides for user_id.

These are the user’s “last used” defaults, applied to every chat below any per-chat overrides. Returns {"rag": {"model": "...", "provider": "..."}, ...}.

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

Return all chats for user_id, newest first.

Each row carries a preview: the first persisted user message, truncated to CHAT_PREVIEW_MAX_CHARS. It gives the frontends a human-readable idea of what the chat is about without fetching every chat’s messages. None when the chat has no user message yet (a freshly created chat).

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

Return credentials for user_id with timestamps.

Each entry is {provider, endpoint, secret, created_at, last_used_at}. Callers must mask secret before returning it to a client.

purge_expired_credentials() → int[source]

Delete credentials unused for longer than the configured TTL.

Called at startup and (opportunistically) on credential access, so short-lived processes still clean up without a background task. A non-positive TTL disables expiry. Returns the number of rows removed.

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

Update the display title of a chat.

set_credential(user_id: str, provider: str, endpoint: str, secret: str) → None[source]

Store (or replace) the secret for a provider(+endpoint).

Records created_at/last_used_at so the unused-key TTL can apply. The secret is never logged.

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.

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

Set or replace the per-user default override for a given role.

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

Bump updated_at without changing any other field.

touch_credential(user_id: str, provider: str, endpoint: str = '', min_interval_seconds: float = 3600) → None[source]

Mark a stored credential as used now (throttled).

Bumps last_used_at so the TTL measures unused time. Writes are skipped while the previous bump is younger than min_interval_seconds to avoid write amplification on hot paths.

klea_utils.api.sessions_db.resolve_credential_ttl_seconds() → float[source]

Resolve the credential TTL (seconds) from the environment.

Reads CREDENTIAL_TTL_ENV_VAR (days). A missing/invalid value falls back to DEFAULT_CREDENTIAL_TTL_SECONDS; 0 or a negative value disables expiry.

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

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

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": {...}, "pending_interrupt": {...} | null, "usage": {...} | null} – the checkpointed session context projected by the app’s context_snapshot hook, plus any pending HITL interrupt (ADR-0046) so a reloaded frontend can re-present the ask, and the checkpointed cumulative usage_metrics (ADR-0032 update) so the token totals survive a reload without storing the per-node usage events. All are 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.

Model selection has three layers, lowest priority first:

  1. graph/env defaults (graph.llm_models)

  2. per-session default overrides (the user’s “last used” models)

  3. per-chat overrides (the distinguishing per-conversation feature)

Overrides store only the model string; API credentials live in the provider-scoped credential store (/credentials) and are resolved separately. Any legacy plaintext api_key still present in stored overrides is masked before it is returned.

See devdocs/system/api-sse-sequence.md; ADRs 0014 (runtime model switching) and 0042 (session model defaults).

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.ModelPayload(*, model: str, promote_default: bool = True)[source]

Bases: BaseModel

Request body for setting a role’s model.

Only the model string is persisted; credentials are managed through /credentials. promote_default (per-chat writes) also updates the per-session default so a new chat inherits the last-used model. Unknown fields (e.g. a legacy api_key) are ignored.

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-session and per-chat model configuration.

Model catalogue (suggestions for the model selection UIs):

GET    /chat/{user_id}/models/catalogue
GET    /chat/{user_id}/models/catalogue?provider=...

Per-session defaults:

GET    /chat/{user_id}/models/overrides
GET    /chat/{user_id}/models/active
POST   /chat/{user_id}/models/overrides/{role}
DELETE /chat/{user_id}/models/overrides/{role}

Per-chat overrides:

GET    /chat/{user_id}/{chat_id}/models/overrides
GET    /chat/{user_id}/{chat_id}/models/active
POST   /chat/{user_id}/{chat_id}/models/overrides/{role}
DELETE /chat/{user_id}/{chat_id}/models/overrides/{role}

A per-chat POST also promotes the model into the session default unless promote_default is false (the “last used” behaviour).

Provider credentials

Per-user provider credential endpoints.

Credentials are stored per user, keyed by provider (plus the endpoint for custom / explicit-URL models), and are never returned raw: the API only reports a masked suffix and, when no credential is stored, whether the provider’s environment variable is set.

See devdocs/adr/0042-session-model-defaults-and-provider-credentials.md.

File: klea_utils/api/credentials.py

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

class klea_utils.api.credentials.CredentialPayload(*, provider: str, endpoint: str = '', secret: str)[source]

Bases: BaseModel

Request body for storing a provider credential.

model_config = {}

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

klea_utils.api.credentials.create_credentials_router() → APIRouter[source]

Create an APIRouter for per-user provider credentials.

GET /credentials/{user_id}

List stored credentials (masked, with timestamps).

PUT /credentials/{user_id}

Store/replace a credential (provider, endpoint, secret).

DELETE /credentials/{user_id}/{provider}

Remove a provider’s credential (optional endpoint query).

klea_utils.api.credentials.credential_status(store: SessionStore, user_id: str, model_name: str) → dict[str, Any][source]

Resolve how model_name’s credential is supplied.

Checks, in order: a stored per-user credential for the model’s provider(+endpoint), then the provider’s API-key environment variable. Providers that need no key (e.g. Ollama) report source="none" with requires_key=False so callers do not treat them as unconfigured.

Parameters:

  • store – The session store holding per-user credentials.

  • user_id – Persistent user identifier.

  • model_name – Provider-prefixed model string.

Returns:

{provider, endpoint, requires_key, source, masked} where source is "user", "env" or "none".

klea_utils.api.credentials.mask_secret(secret: str | None) → str[source]

Return a masked representation of secret (last 4 characters).

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.

The ordered request/lifecycle view is devdocs/system/api-sse-sequence.md; the event catalogue is devdocs/system/streams.md.

File: klea_utils/api/sse.py

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

klea_utils.api.sse.CATALOGUE_FETCH_TIMEOUT_SECONDS = 30

the first call may have to download the models.dev catalog on the server.

Type:

Timeout for the model catalogue calls. Longer than the default

klea_utils.api.sse.STREAM_READ_TIMEOUT_SECONDS = 300.0

Idle read timeout for the streaming request, in seconds. Must comfortably exceed the server’s SSE heartbeat interval (klea_utils.api.chat_core.HEARTBEAT_INTERVAL_SECONDS, 15 s) so a heartbeat arrives and resets the timer well before it fires while a long-running node emits no events.

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 + per-session + per-chat 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": "...", "credential": {...}}, ...}

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.

async klea_utils.api.sse.fetch_catalogue_models(server_url: str, user_id: str, provider: str) → list[str][source]

Fetch the model ids offered for one provider.

Calls GET /chat/{user_id}/models/catalogue?provider=.... Returns an empty list on any error, and for providers whose models are free text (ollama, custom).

Parameters:

  • server_url – Base URL of the backend API server.

  • user_id – Opaque persistent user identifier.

  • provider – Klea provider id, e.g. "openai".

Returns:

Sorted model ids, e.g. ["gpt-4o", "gpt-4o-mini", ...].

klea_utils.api.sse.fetch_catalogue_models_sync(server_url: str, user_id: str, provider: str) → list[str][source]

Synchronous counterpart of fetch_catalogue_models().

async klea_utils.api.sse.fetch_catalogue_providers(server_url: str, user_id: str) → list[str][source]

Fetch the providers offered for model selection.

Calls GET /chat/{user_id}/models/catalogue. Returns an empty list on any error, so the model fields fall back to plain free text.

Parameters:

  • server_url – Base URL of the backend API server.

  • user_id – Opaque persistent user identifier.

Returns:

Sorted provider ids, e.g. ["anthropic", "custom", ...].

klea_utils.api.sse.fetch_catalogue_providers_sync(server_url: str, user_id: str) → list[str][source]

Synchronous counterpart of fetch_catalogue_providers().

async klea_utils.api.sse.fetch_credentials(server_url: str, user_id: str) → list[dict[str, Any]][source]

Fetch the user’s stored provider credentials (masked).

Calls GET /credentials/{user_id}.

Parameters:

  • server_url – Base URL of the backend API server.

  • user_id – Opaque persistent user identifier.

Returns:

List of {provider, endpoint, source, masked, ...}.

klea_utils.api.sse.fetch_credentials_sync(server_url: str, user_id: str) → list[dict[str, Any]][source]

Synchronous counterpart of fetch_credentials().

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

Fetch defaults + per-session overrides for a user (no chat).

Calls GET /chat/{user_id}/models/active.

Parameters:

  • server_url – Base URL of the backend API server.

  • user_id – Opaque persistent user identifier.

Returns:

{"chat": {"model": "...", "credential": {...}}, ...}

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

Synchronous counterpart of fetch_session_models().

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.request_cancel(server_url: str, chat_id: str, user_id: str = '', timeout: float = 10.0) → bool[source]

Ask the server to cancel the chat’s active run (idempotent).

Best-effort: a 204 (or any non-error response) means the request reached the server; network failures are logged and return False so the frontend can still stop locally.

Parameters:

  • server_url – Base URL of the backend API server.

  • chat_id – Chat conversation identifier.

  • user_id – Opaque persistent user identifier.

  • timeout – Request timeout in seconds.

Returns:

True when the server accepted the cancel request.

async klea_utils.api.sse.stream_events(query: str, chat_id: str, server_url: str, user_id: str = '', extra: dict | None = None, resume: bool = False, interrupt_response: dict | None = None, interrupt_id: str | None = None, interrupt_cancel: bool = False) → 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>",
             "data": {"heading": "<text>"}}  (heading is the line shown;
             it changes for an LLM invoke retry)
inspect     {"type": "inspect", "node": "<label>", "data": {...}}
token       {"type": "token", "content": "<chunk>", "node": "<label>"}
            (only for opted-in free-text nodes)
usage       {"type": "usage", "node": "<label>", "data": {...}}
context     {"type": "context", "data": {...}}  (graph-level session context)
interrupt   {"type": "interrupt", "node": "<label>", "data": {...}}
            (the run paused for human input; ``data`` carries ``kind``,
            the ``question``/``questions``, the ``interrupt_id`` and a
            ``hitl_response_schema`` when the node supplied one.  No
            ``complete`` follows a pause.)
complete    {"type": "complete", "message_for_user": "<text>"}
error       {"type": "error", "message": "<text>", "error_type": "<class>",
             "node": "<label>", "resumable": <bool>}
ping        {"type": "ping"}  (periodic server heartbeat while a
             long-running node emits no events; consumers may ignore
             it -- its only purpose is to keep the SSE connection and
             any intermediaries from timing out)

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

  • resume – Resume the chat’s last failed run from its checkpoint instead of starting a new turn (the query is not sent).

  • interrupt_response – Answer a pending HITL interrupt (ADR-0046); the resume mapping (e.g. {"answers": [...]} or {"decision": "approve", "feedback": "..."}). Pass an empty query.

  • interrupt_id – Id of the interrupt being answered.

  • interrupt_cancel – Cancel the pending interrupt instead of answering.

klea_utils.api.sse.stream_events_sync(query: str, chat_id: str, server_url: str, user_id: str = '', extra: dict | None = None, resume: bool = False, interrupt_response: dict | None = None, interrupt_id: str | None = None, interrupt_cancel: bool = False) → 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).

  • resume – Resume the chat’s last failed run from its checkpoint instead of starting a new turn (the query is not sent).

  • interrupt_response – Answer a pending HITL interrupt (ADR-0046); the resume mapping (e.g. {"answers": [...]}). Pass an empty query.

  • interrupt_id – Id of the interrupt being answered.

  • interrupt_cancel – Cancel the pending interrupt instead of answering.

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.