Services¶

The services layer provides infrastructure services for the CLI. These are leaf services with no business logic — they handle external interactions such as authentication, credential storage, Docker operations, and API communication.

Services are instantiated lazily via the dependency injection container (kelvin.sdk.container.Services).

API Client¶

Factory for creating authenticated Kelvin API client instances.

API Client factory - provides authenticated Kelvin API client.

This module provides a factory for creating authenticated API client instances. It handles session validation, token refresh, and client instantiation.

exception kelvin.sdk.services.api_client.APIClientError(message=None, exit_code=None)[source]¶

Bases: AuthenticationError

Base exception for API client factory errors.

Inherits from AuthenticationError (exit_code=77) so that except AuthenticationError catches login-related errors like NotLoggedInError and CredentialsExpiredError.

Parameters:

message (str)
exit_code (int)

exception kelvin.sdk.services.api_client.NotLoggedInError(message=None)[source]¶

Bases: APIClientError

Raised when no active session exists.

Parameters:: message (str)

exception kelvin.sdk.services.api_client.CredentialsExpiredError(message=None)[source]¶

Bases: APIClientError

Raised when credentials are expired and cannot be refreshed.

Parameters:: message (str)

class kelvin.sdk.services.api_client.APIClientFactory(session, credentials, auth)[source]¶

Bases: object

Factory for creating authenticated Kelvin API clients.

This factory handles: - Session validation (ensures user is logged in) - Token expiration checks - Automatic token refresh when possible - Client instantiation with valid credentials

The factory does NOT cache clients - each call creates a new client. Caching should be done at the container level if needed.

Example¶

>>> factory = APIClientFactory(session, credentials, auth)
>>> client = factory.create()  # Returns authenticated Client
>>> workloads = client.app_workloads.list()

__init__(session, credentials, auth)[source]¶

Initialize API client factory.

Parameters¶

session: Session service for getting current session.
credentials: Credential store for retrieving tokens.
auth: Auth service for refreshing tokens.

Parameters:

session (SessionService)
credentials (CredentialStore)
auth (AuthService)

Return type:

None

create(verbose=False)[source]¶

Create an authenticated API client.

Return type:: Client
Parameters:: verbose (bool)

Returns¶

Client: Authenticated Kelvin API client.

Raises¶

NotLoggedInError: If no active session exists or no credentials found.
CredentialsExpiredError: If credentials are expired and cannot be refreshed.

Parameters:

session (SessionService)
credentials (CredentialStore)
auth (AuthService)

Configuration¶

Manages CLI configuration settings persisted to a local YAML file.

ConfigService - Manages CLI configuration file storage.

This service manages the CLI configuration settings by persisting to a local YAML file in the user’s config directory.

exception kelvin.sdk.services.config.ConfigServiceError(message=None, exit_code=None)[source]¶

Bases: CLIError

Base exception for ConfigService errors.

Parameters:

message (str)
exit_code (int)

exit_code: int = 78¶

exception kelvin.sdk.services.config.InvalidConfigKeyError(key, valid_keys)[source]¶

Bases: ConfigServiceError

Raised when an invalid configuration key is provided.

Parameters:

key (str)
valid_keys (list[str])

Return type:

None

key: str¶

valid_keys: list[str]¶

exception kelvin.sdk.services.config.InvalidConfigValueError(key, value, expected_type)[source]¶

Bases: ConfigServiceError

Raised when an invalid configuration value is provided.

Parameters:

key (str)
value (str)
expected_type (str)

Return type:

None

key: str¶

value: str¶

expected_type: str¶

class kelvin.sdk.services.config.CLIConfig(**data)[source]¶

Bases: BaseModel

CLI configuration settings.

This is the single source of truth for configuration defaults. Field defaults here are used everywhere in the CLI.

Parameters:

version_warning (bool)
colored_logs (bool)
verbose (int)
json_output (bool)
no_prompt (bool)

version_warning: bool¶

colored_logs: bool¶

verbose: int¶

json_output: bool¶

no_prompt: bool¶

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

kelvin.sdk.services.config.get_config_keys()[source]¶

Get configuration keys with metadata from the Pydantic model.

Return type:: dict[str, dict[str, object]]
Returns:: Dictionary of key names to metadata (type, description, default).

class kelvin.sdk.services.config.ConfigService[source]¶

Bases: object

Manages CLI configuration file storage.

This is a leaf service with no dependencies. It persists configuration settings to a YAML file in the user’s config directory.

CONFIG_DIR: str = 'kelvin'¶

CONFIG_FILE: str = 'config.yaml'¶

__init__()[source]¶

Initialize ConfigService.

Return type:: None

get_config()[source]¶

Get the current configuration.

Return type:: CLIConfig
Returns:: CLIConfig with current settings.

save_config(config)[source]¶

Save the configuration to file.

Parameters:: config (CLIConfig) – The configuration to save.
Return type:: None

set_value(key, value)[source]¶

Set a configuration value.

Parameters:

key (str) – The configuration key.
value (str) – The value to set (as string, will be converted).

Return type:

CLIConfig

Returns:

Updated CLIConfig.

Raises:

InvalidConfigKeyError – If key is not valid.
InvalidConfigValueError – If value cannot be parsed.

unset_value(key)[source]¶

Reset a configuration value to its default.

Parameters:: key (str) – The configuration key.
Return type:: CLIConfig
Returns:: Updated CLIConfig.
Raises:: InvalidConfigKeyError – If key is not valid.

reset()[source]¶

Reset all configuration to defaults and delete the config file.

Return type:: None

get_config_keys()[source]¶

Get all configuration keys with their metadata.

Return type:: dict[str, dict[str, object]]
Returns:: Dictionary of key names to metadata.

get_config_file_path()[source]¶

Get the path to the config file (for display purposes).

Return type:: Path
Returns:: Path to the config file.

Credential Store¶

Secure OAuth token storage and retrieval via keyring.

Credential store service for secure OAuth token storage.

This service manages secure storage and retrieval of OAuth tokens using keyring. It is a leaf service with no dependencies.

exception kelvin.sdk.services.credential_store.CredentialStoreError(message=None, exit_code=None)[source]¶

Bases: CLIError

Base exception for credential store operations.

Parameters:

message (str)
exit_code (int)

exit_code: int = 77¶

exception kelvin.sdk.services.credential_store.CredentialStorageError(message=None, exit_code=None)[source]¶

Bases: CredentialStoreError

Failed to store credentials.

Parameters:

message (str)
exit_code (int)

exception kelvin.sdk.services.credential_store.CredentialRetrievalError(message=None, exit_code=None)[source]¶

Bases: CredentialStoreError

Failed to retrieve credentials.

Parameters:

message (str)
exit_code (int)

class kelvin.sdk.services.credential_store.TokenCredentials(access_token, access_expires_at, refresh_token=None, refresh_expires_at=None, oauth_client_id=None)[source]¶

Bases: object

OAuth token credentials.

Parameters:

access_token (str)
access_expires_at (int)
refresh_token (str | None)
refresh_expires_at (int | None)
oauth_client_id (str | None)

access_token: str¶

access_expires_at: int¶

refresh_token: str | None = None¶

refresh_expires_at: int | None = None¶

oauth_client_id: str | None = None¶

property is_service_account: bool¶

True if these credentials were obtained via client credentials grant.

Service account tokens have no refresh token. The Keycloak userinfo endpoint rejects them, so callers should skip remote token validation.

class kelvin.sdk.services.credential_store.CredentialStore[source]¶

Bases: object

Secure credential storage using keyring.

This service handles storage and retrieval of OAuth tokens in the system’s secure credential store (keyring). It does not perform authentication - that responsibility belongs to AuthService.

Example¶

>>> store = CredentialStore()
>>> creds = store.retrieve("https://api.example.com")
>>> if creds and not store.is_token_expired(creds):
...     print("Valid credentials found")

KEYRING_SERVICE_NAME: Final[str] = 'kelvin:kelvin-sdk'¶

EXPIRATION_THRESHOLD: Final[int] = 60¶

store(url, access_token, expires_in, refresh_token=None, refresh_expires_in=None, oauth_client_id=None)[source]¶

Store OAuth tokens in the secure keyring.

Return type:

TokenCredentials

Parameters:

url (str)
access_token (str)
expires_in (int)
refresh_token (str | None)
refresh_expires_in (int | None)
oauth_client_id (str | None)

Parameters¶

url: The platform URL to associate with these credentials.
access_token: The OAuth access token.
expires_in: Number of seconds until the access token expires.
refresh_token: Optional OAuth refresh token.
refresh_expires_in: Number of seconds until the refresh token expires.

Returns¶

TokenCredentials: The stored credentials with calculated expiration timestamps.

Raises¶

CredentialStorageError: If storage fails.

store_token_response(url, tokens, oauth_client_id=None)[source]¶

Store OAuth tokens from a TokenResponse.

Return type:

TokenCredentials

Parameters:

url (str)
tokens (TokenResponse)
oauth_client_id (str | None)

Parameters¶

url: The platform URL to associate with these credentials.
tokens: The TokenResponse from AuthService.
oauth_client_id: The Keycloak OAuth client ID used to obtain these tokens.

Returns¶

TokenCredentials: The stored credentials with calculated expiration timestamps.

Raises¶

CredentialStorageError: If storage fails.

retrieve(url)[source]¶

Retrieve OAuth tokens from the secure keyring.

Return type:: Optional[TokenCredentials]
Parameters:: url (str)

Parameters¶

url: The platform URL to retrieve credentials for.

Returns¶

Optional[TokenCredentials]: The stored credentials if found, None otherwise.

clear(url)[source]¶

Clear stored credentials for a specific URL.

Return type:: bool
Parameters:: url (str)

Parameters¶

url: The platform URL to clear credentials for.

Returns¶

bool: True if credentials were cleared, False if none existed.

is_token_expired(credentials)[source]¶

Check if the access token is expired.

Return type:: bool
Parameters:: credentials (TokenCredentials)

Parameters¶

credentials: The credentials to check.

Returns¶

bool: True if the access token is expired, False otherwise.

is_refresh_token_expired(credentials)[source]¶

Check if the refresh token is expired.

Return type:: bool
Parameters:: credentials (TokenCredentials)

Parameters¶

credentials: The credentials to check.

Returns¶

bool: True if no refresh token or if it’s expired, False otherwise.

Docker¶

Docker operations for building and managing application images.

DockerService - Docker operations using buildx for all builds.

This service manages Docker operations for building and managing application images. Uses buildx for all builds (both single and multi-arch) for consistency. It is a leaf service with no dependencies.

exception kelvin.sdk.services.docker.DockerServiceError(message=None, exit_code=None)[source]¶

Bases: CLIError

Base exception for Docker service operations.

Parameters:

message (str)
exit_code (int)

exit_code: int = 75¶

exception kelvin.sdk.services.docker.DockerNotRunningError(message=None)[source]¶

Bases: DockerServiceError

Docker daemon is not running.

Parameters:: message (str)

exception kelvin.sdk.services.docker.DockerVersionError(message=None, exit_code=None)[source]¶

Bases: DockerServiceError

Docker version does not meet requirements.

Parameters:

message (str)
exit_code (int)

exception kelvin.sdk.services.docker.DockerBuildError(message=None, exit_code=None)[source]¶

Bases: DockerServiceError

Docker build operation failed.

Parameters:

message (str)
exit_code (int)

exception kelvin.sdk.services.docker.DockerImageNotFoundError(message=None, exit_code=None)[source]¶

Bases: DockerServiceError

Docker image not found.

Parameters:

message (str)
exit_code (int)

exit_code: int = 66¶

exception kelvin.sdk.services.docker.DockerPlatformError(message=None, exit_code=None)[source]¶

Bases: DockerServiceError

Requested platform is not supported.

Parameters:

message (str)
exit_code (int)

exception kelvin.sdk.services.docker.DockerRegistryError(message=None, exit_code=None)[source]¶

Bases: DockerServiceError

Docker registry operation failed.

Parameters:

message (str)
exit_code (int)

class kelvin.sdk.services.docker.DockerImage(id, name, tags, size, created_at=None, labels=<factory>)[source]¶

Bases: object

Docker image information.

Parameters:

id (str)
name (str)
tags (list[str])
size (int)
created_at (datetime | None)
labels (dict[str, str])

id: str¶

name: str¶

tags: list[str]¶

size: int¶

created_at: datetime | None = None¶

labels: dict[str, str]¶

classmethod from_api(data)[source]¶

Create DockerImage from Docker API response.

Return type:: DockerImage
Parameters:: data (dict[str, object])

class kelvin.sdk.services.docker.BuildConfig(context_path, dockerfile_path, image_name, build_args=None, platforms=None, labels=None, fresh_build=False)[source]¶

Bases: object

Configuration for a Docker build operation.

Parameters:

context_path (Path)
dockerfile_path (Path)
image_name (str)
build_args (dict[str, str] | None)
platforms (list[str] | None)
labels (dict[str, str] | None)
fresh_build (bool)

context_path: Path¶

dockerfile_path: Path¶

image_name: str¶

build_args: dict[str, str] | None = None¶

platforms: list[str] | None = None¶

labels: dict[str, str] | None = None¶

fresh_build: bool = False¶

class kelvin.sdk.services.docker.BuildResult(success, image_name, logs=<factory>)[source]¶

Bases: object

Result of a Docker build operation.

Parameters:

success (bool)
image_name (str)
logs (list[str])

success: bool¶

image_name: str¶

logs: list[str]¶

class kelvin.sdk.services.docker.DockerService[source]¶

Bases: object

Docker operations using buildx for all builds.

This is a leaf service with no dependencies. It manages: - Docker daemon status checks - Registry authentication - Image builds (via buildx) - Image management (list, get, remove, tag, pull) - File extraction from images

All builds use buildx for consistency, even for single-architecture builds.

Example¶

>>> docker = DockerService()
>>> if docker.is_running():
...     result = docker.build(config, registry_url="registry.example.com")

CLIENT_TIMEOUT: Final[int] = 300¶

__init__()[source]¶

Initialize DockerService.

Return type:: None

is_running()[source]¶

Check if Docker daemon is running.

Return type:: bool

Returns¶

bool: True if Docker is running, False otherwise.

get_version()[source]¶

Get Docker version string.

Return type:: Optional[str]

Returns¶

Optional[str]: Docker version string, or None if unavailable.

validate_version(minimum_version)[source]¶

Check if Docker version meets minimum requirement.

Return type:: bool
Parameters:: minimum_version (str)

Parameters¶

minimum_version: Minimum required Docker version.

Returns¶

bool: True if current version meets requirement.

Raises¶

DockerVersionError: If version check fails or version is insufficient.

get_supported_platforms()[source]¶

Get platforms supported by current buildx builder.

Return type:: list[str]

Returns¶

List[str]: List of supported platforms (e.g., [“linux/amd64”, “linux/arm64”]).

validate_platforms(platforms)[source]¶

Validate that requested platforms are supported by buildx.

Return type:: None
Parameters:: platforms (list[str])

Parameters¶

platforms: List of platforms to validate.

Raises¶

DockerPlatformError: If any platform is not supported.

login(registry_url, username, password)[source]¶

Return type:

None

Parameters:

registry_url (str)
username (str)
password (str)

Parameters¶

registry_url: The registry URL to authenticate with.
username: Registry username.
password: Registry password or token.

Raises¶

DockerRegistryError: If login fails.

build(config, registry_url=None)[source]¶

Build image locally using buildx.

Single platform: loads image to local daemon (–load)
Multi platform: builds but does NOT load (multi-arch can’t be loaded locally)
If registry_url provided, image name is prefixed with it

Parameters¶

config: Build configuration.
registry_url: Optional registry URL to prefix image name.

Returns¶

BuildResult: Result of the build operation.

Raises¶

DockerBuildError: If build fails.

rtype:: BuildResult

Parameters:

config (BuildConfig)
registry_url (str | None)

Return type:

BuildResult

build_and_push(config, registry_url)[source]¶

Build and push image to registry using buildx –push.

Supports multi-arch builds
Image name is prefixed with registry_url
Image is NOT loaded locally (pushed directly to registry)

Parameters¶

config: Build configuration.
registry_url: Registry URL to push to.

Returns¶

BuildResult: Result of the build operation.

Raises¶

DockerBuildError: If build or push fails.

rtype:: BuildResult

Parameters:

config (BuildConfig)
registry_url (str)

Return type:

BuildResult

list_images(labels=None)[source]¶

List local images, optionally filtered by labels.

Return type:: list[DockerImage]
Parameters:: labels (dict[str, str] | None)

Parameters¶

labels: Optional labels to filter images.

Returns¶

List[DockerImage]: List of matching Docker images.

get_image(name)[source]¶

Get a specific image by name/tag.

Return type:: Optional[DockerImage]
Parameters:: name (str)

Parameters¶

name: Image name or tag.

Returns¶

Optional[DockerImage]: The image if found, None otherwise.

image_exists(name)[source]¶

Check if image exists locally.

Return type:: bool
Parameters:: name (str)

Parameters¶

name: Image name or tag.

Returns¶

bool: True if image exists, False otherwise.

remove_image(name)[source]¶

Remove a local image.

Return type:: None
Parameters:: name (str)

Parameters¶

name: Image name or tag to remove.

Raises¶

DockerImageNotFoundError: If image does not exist.

tag_image(source, target)[source]¶

Tag an existing image with a new name.

Return type:

None

Parameters:

source (str)
target (str)

Parameters¶

source: Source image name.
target: Target tag name.

Raises¶

DockerImageNotFoundError: If source image does not exist.

pull_image(name)[source]¶

Pull image from registry.

Return type:: DockerImage
Parameters:: name (str)

Parameters¶

name: Image name with optional tag.

Returns¶

DockerImage: The pulled image.

Raises¶

DockerImageNotFoundError: If image not found in registry.
DockerRegistryError: If pull fails.

extract_from_image(image_name, container_path, output_path)[source]¶

Extract files from an image to local path.

Return type:

None

Parameters:

image_name (str)
container_path (str)
output_path (Path)

Parameters¶

image_name: Name of the image to extract from.
container_path: Path inside the container to extract.
output_path: Local path to extract files to.

Raises¶

DockerImageNotFoundError: If image does not exist.
DockerServiceError: If extraction fails.

MLflow¶

MLflow registry operations for model listing, retrieval, and artifact downloading.

MLflowService - MLflow registry operations.

This is a leaf service that wraps MLflow operations for interacting with MLflow model registries. It handles model listing, version retrieval, and model artifact downloading.

MLflow is an optional dependency. If not installed, operations will raise MLflowNotInstalledError with instructions on how to install it.

exception kelvin.sdk.services.mlflow.MLflowServiceError(message=None, exit_code=None)[source]¶

Bases: CLIError

Base exception for MLflow service errors.

Parameters:

message (str)
exit_code (int)

exit_code: int = 70¶

exception kelvin.sdk.services.mlflow.MLflowNotInstalledError[source]¶

Bases: MLflowServiceError

Raised when MLflow is not installed.

Return type:: None

exit_code: int = 78¶

exception kelvin.sdk.services.mlflow.RegistryConnectionError(registry_uri, reason)[source]¶

Bases: MLflowServiceError

Raised when cannot connect to MLflow registry.

Parameters:

registry_uri (str)
reason (str)

Return type:

None

exit_code: int = 74¶

registry_uri: str¶

reason: str¶

exception kelvin.sdk.services.mlflow.ModelNotFoundError(model_name)[source]¶

Bases: MLflowServiceError

Raised when a model is not found in the registry.

Parameters:: model_name (str)
Return type:: None

exit_code: int = 66¶

model_name: str¶

exception kelvin.sdk.services.mlflow.ModelVersionNotFoundError(model_name, version)[source]¶

Bases: MLflowServiceError

Raised when a model version is not found.

Parameters:

model_name (str)
version (str)

Return type:

None

exit_code: int = 66¶

model_name: str¶

version: str¶

exception kelvin.sdk.services.mlflow.ModelDownloadError(model_name, version, reason)[source]¶

Bases: MLflowServiceError

Raised when model download fails.

Parameters:

model_name (str)
version (str)
reason (str)

Return type:

None

exit_code: int = 74¶

model_name: str¶

version: str¶

reason: str¶

class kelvin.sdk.services.mlflow.ModelIO(name, data_type)[source]¶

Bases: object

Model output specification.

Parameters:

name (str)
data_type (str)

name: str¶

data_type: str¶

class kelvin.sdk.services.mlflow.DownloadedModel(model_path, inputs=<factory>, outputs=<factory>, description=None)[source]¶

Bases: object

Information about a downloaded model.

Parameters:

model_path (Path)
inputs (list[ModelIO])
outputs (list[ModelIO])
description (str | None)

model_path: Path¶

inputs: list[ModelIO]¶

outputs: list[ModelIO]¶

description: str | None = None¶

class kelvin.sdk.services.mlflow.MLflowService[source]¶

Bases: object

MLflow registry operations.

This is a leaf service with no dependencies on other services. It wraps MLflow client operations and handles the optional MLflow dependency gracefully.

list_models(registry_uri, filter_string=None, max_results=1000)[source]¶

List registered models in the MLflow registry.

Parameters:

registry_uri (str) – URI of the MLflow registry.
filter_string (Optional[str]) – Optional filter string for the search.
max_results (int) – Maximum number of results to return.

Return type:

list[RegisteredModel]

Returns:

List of RegisteredModel objects from MLflow.

Raises:

RegistryConnectionError – If cannot connect to registry.

get_model_versions(registry_uri, model_name)[source]¶

Get all versions of a model.

Parameters:

registry_uri (str) – URI of the MLflow registry.
model_name (str) – Name of the model.

Return type:

list[ModelVersion]

Returns:

List of ModelVersion objects from MLflow.

Raises:

RegistryConnectionError – If cannot connect to registry.
ModelNotFoundError – If model doesn’t exist.

download_model(registry_uri, model_name, model_version, dest_path)[source]¶

Download model artifacts and extract signature information.

Parameters:

registry_uri (str) – URI of the MLflow registry.
model_name (str) – Name of the model.
model_version (str) – Version of the model.
dest_path (Path) – Destination path for the model artifacts.

Return type:

DownloadedModel

Returns:

DownloadedModel with path and signature information.

Raises:

RegistryConnectionError – If cannot connect to registry.
ModelVersionNotFoundError – If model version doesn’t exist.
ModelDownloadError – If download fails.

Schema¶

JSON Schema validation for app configuration files.

SchemaService - JSON Schema validation for app configurations.

This service validates app configuration files against JSON schemas. Schemas are fetched from a remote service based on the spec_version defined in the app configuration.

exception kelvin.sdk.services.schema.SchemaServiceError(message=None, exit_code=None)[source]¶

Bases: CLIError

Base exception for SchemaService errors.

Parameters:

message (str)
exit_code (int)

exit_code: int = 65¶

exception kelvin.sdk.services.schema.SchemaFetchError(message=None, exit_code=None)[source]¶

Bases: SchemaServiceError

Raised when schema cannot be fetched from remote service.

Parameters:

message (str)
exit_code (int)

exception kelvin.sdk.services.schema.SchemaValidationError(message, errors)[source]¶

Bases: SchemaServiceError

Raised when app configuration fails schema validation.

Parameters:

message (str)
errors (list[str])

Return type:

None

__init__(message, errors)[source]¶

Initialize with validation errors.

Parameters:

message (str) – Summary error message.
errors (list[str]) – List of specific validation error messages.

Return type:

None

errors: list[str]¶

class kelvin.sdk.services.schema.SchemaValidationResult(valid, errors=<factory>, warnings=<factory>)[source]¶

Bases: object

Result of schema validation.

Parameters:

valid (bool)
errors (list[str])
warnings (list[str])

valid: bool¶

errors: list[str]¶

warnings: list[str]¶

class kelvin.sdk.services.schema.SchemaService[source]¶

Bases: object

Schema management and validation.

This is a leaf service with no dependencies. It validates app configurations against JSON schemas fetched from a remote service.

The schema URL is determined by the spec_version key in the app configuration. Schemas are available at https://apps.kelvininc.com/schemas/kelvin/{version}/app/app.json

SCHEMA_BASE_URL: Final[str] = 'https://apps.kelvininc.com/schemas/kelvin'¶

SCHEMA_PATH: Final[str] = 'app/app.json'¶

REQUEST_TIMEOUT: Final[float] = 10.0¶

__init__()[source]¶

Initialize SchemaService.

Return type:: None

validate_config_dict(config)[source]¶

Validate an app configuration dictionary against its schema.

The schema version is determined by the ‘spec_version’ key in the config.

Parameters:: config (dict[str, object]) – The configuration dictionary to validate.
Return type:: SchemaValidationResult
Returns:: SchemaValidationResult with validation outcome.
Raises:: SchemaFetchError – If the schema cannot be fetched.

get_schema(version)[source]¶

Get the schema for a specific version.

Parameters:: version (str) – The schema version to retrieve.
Return type:: dict[str, object]
Returns:: The schema as a dictionary.
Raises:: SchemaFetchError – If the schema cannot be fetched.

clear_cache()[source]¶

Clear the schema cache.

Return type:: None

Session¶

Session state management and platform metadata.

SessionService - Manages session state and platform metadata file storage.

This service manages the current session state, including the platform URL, Docker registry information, and platform metadata by persisting to local files.

exception kelvin.sdk.services.session.SessionServiceError(message=None, exit_code=None)[source]¶

Bases: CLIError

Base exception for SessionService errors.

Parameters:

message (str)
exit_code (int)

exit_code: int = 78¶

exception kelvin.sdk.services.session.SessionNotFoundError(message=None, exit_code=None)[source]¶

Bases: SessionServiceError

Raised when no session is found.

Parameters:

message (str)
exit_code (int)

exception kelvin.sdk.services.session.SessionStorageError(message=None, exit_code=None)[source]¶

Bases: SessionServiceError

Raised when session storage operations fail.

Parameters:

message (str)
exit_code (int)

class kelvin.sdk.services.session.SessionService[source]¶

Bases: object

Manages session state and platform metadata file storage.

This is a leaf service with no dependencies. It simply persists session and metadata information to local files.

Note: Login state checking and token refresh are NOT handled here. Those responsibilities belong to the command layer (AuthCommands) which orchestrates CredentialStore and AuthService.

CONFIG_DIR: Final[str] = 'kelvin'¶

SESSION_FILE: Final[str] = 'session.yaml'¶

METADATA_FILE: Final[str] = 'metadata.json'¶

__init__()[source]¶

Initialize SessionService.

Return type:: None

get_current_session()[source]¶

Get the current session information.

Return type:: Optional[SessionInfo]
Returns:: SessionInfo if a session exists, None otherwise.

set_current_session(url, metadata=None)[source]¶

Set the current session and optionally store metadata.

Parameters:

url (str) – The platform URL.
metadata (Optional[PlatformMetadata]) – Optional platform metadata. If provided, will also be stored.

Return type:

SessionInfo

Returns:

The created SessionInfo.

Raises:

SessionStorageError – If saving the session fails.

clear_session()[source]¶

Clear the current session and metadata files.

Raises:: SessionStorageError – If clearing the session fails.
Return type:: None

store_metadata(metadata)[source]¶

Store platform metadata.

Parameters:: metadata (PlatformMetadata) – The metadata to store.
Raises:: SessionStorageError – If storing fails.
Return type:: None

get_metadata()[source]¶

Get stored platform metadata.

Return type:: Optional[PlatformMetadata]
Returns:: PlatformMetadata if available, None otherwise.

get_docker_registry_url()[source]¶

Get the Docker registry URL including optional path.

Used for image name prefixes (e.g., registry.example.com:5000/kelvin).

Return type:: Optional[str]
Returns:: Docker registry URL (with optional path) or None if not available.

get_docker_login_url()[source]¶

Get the Docker registry URL for docker login (host:port only, no path).

docker login only accepts host:port, not a full image path prefix.

Return type:: Optional[str]
Returns:: Docker registry login URL or None if not available.

get_documentation_url()[source]¶

Get the documentation URL.

Return type:: Optional[str]
Returns:: Documentation URL or None if not available.

get_config_dir()[source]¶

Get the configuration directory path.

Return type:: Path
Returns:: Path to the configuration directory.

Session Models¶

Data models for session state and platform metadata.

Session and Platform Metadata models.

Pydantic models for session state and platform metadata.

class kelvin.sdk.services.session_models.SessionInfo(**data)[source]¶

Bases: BaseModel

Current session information.

Parameters:

url (str)
docker_url (str | None)
docker_port (int | None)
docker_path (str | None)
last_login (datetime | None)

url: str¶

docker_url: str | None¶

docker_port: int | None¶

docker_path: str | None¶

last_login: datetime | None¶

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class kelvin.sdk.services.session_models.ApiDocumentation(**data)[source]¶

Bases: BaseModel

Documentation section of the API metadata response.

Parameters:: url (str | None)

url: str | None¶

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class kelvin.sdk.services.session_models.ApiDocker(**data)[source]¶

Bases: BaseModel

Docker section of the API metadata response.

Parameters:

url (str | None)
port (int | None)
path (str | None)

url: str | None¶

port: int | None¶

path: str | None¶

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class kelvin.sdk.services.session_models.ApiSdk(**data)[source]¶

Bases: BaseModel

SDK section of the API metadata response.

Parameters:

docker_minimum_version (str | None)
ksdk_latest_version (str | None)
ksdk_minimum_version (str | None)

docker_minimum_version: str | None¶

ksdk_latest_version: str | None¶

ksdk_minimum_version: str | None¶

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class kelvin.sdk.services.session_models.ApiSchema(**data)[source]¶

Bases: BaseModel

Schema section of the API metadata response.

Parameters:

latest_spec_version (str | None)
minimum_spec_version (str | None)

latest_spec_version: str | None¶

minimum_spec_version: str | None¶

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class kelvin.sdk.services.session_models.ApiMetadataResponse(**data)[source]¶

Bases: BaseModel

Pydantic model for parsing the platform metadata API response.

Parameters:

documentation (ApiDocumentation | None)
docker (ApiDocker | None)
sdk (ApiSdk | None)
schema (ApiSchema | None)

model_config: ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

documentation: ApiDocumentation | None¶

docker: ApiDocker | None¶

sdk: ApiSdk | None¶

schema_: ApiSchema | None¶

class kelvin.sdk.services.session_models.PlatformMetadata(**data)[source]¶

Bases: BaseModel

Platform metadata from the API.

Parameters:

url (str)
documentation_url (str | None)
ksdk_minimum_version (str | None)
ksdk_latest_version (str | None)
docker_minimum_version (str | None)
docker_url (str | None)
docker_port (int | None)
docker_path (str | None)
schema_latest_version (str | None)
schema_minimum_version (str | None)

url: str¶

documentation_url: str | None¶

ksdk_minimum_version: str | None¶

ksdk_latest_version: str | None¶

docker_minimum_version: str | None¶

docker_url: str | None¶

docker_port: int | None¶

docker_path: str | None¶

schema_latest_version: str | None¶

schema_minimum_version: str | None¶

classmethod from_api_response(url, response)[source]¶

Create PlatformMetadata from API response.

Parameters:

url (str) – The platform URL.
response (dict[str, object]) – The raw metadata response from the API.

Return type:

Self

Returns:

PlatformMetadata instance.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Template¶

Application template management and project scaffolding.

TemplateService - Application template management and project scaffolding.

This service manages application templates for creating new Kelvin applications. Templates are defined declaratively using template.yaml files that specify: - Static files to copy - Jinja2 templates to render - Shared files from _common directory - Symlinks to create

Template structure:

templates/
├── _common/                    # Shared files across templates
│   ├── .dockerignore.jinja2
│   └── ui_schemas/
├── app/
│   ├── template.yaml           # Template definition
│   ├── app.yaml.jinja2
│   └── ...
└── importer/
    ├── template.yaml
    └── ...

exception kelvin.sdk.services.template.TemplateServiceError(message=None, exit_code=None)[source]¶

Bases: CLIError

Base exception for TemplateService errors.

Parameters:

message (str)
exit_code (int)

exit_code: int = 65¶

exception kelvin.sdk.services.template.TemplateNotFoundError(template_name)[source]¶

Bases: TemplateServiceError

Raised when a requested template does not exist.

Parameters:: template_name (str)
Return type:: None

template_name: str¶

exception kelvin.sdk.services.template.ManifestError(template_name, reason)[source]¶

Bases: TemplateServiceError

Raised when a manifest file is invalid or missing.

Parameters:

template_name (str)
reason (str)

Return type:

None

template_name: str¶

reason: str¶

exception kelvin.sdk.services.template.TemplateRenderError(file_path, reason)[source]¶

Bases: TemplateServiceError

Raised when a template fails to render.

Parameters:

file_path (str)
reason (str)

Return type:

None

file_path: str¶

reason: str¶

exception kelvin.sdk.services.template.FileCreationError(file_path, reason)[source]¶

Bases: TemplateServiceError

Raised when file creation fails.

Parameters:

file_path (str)
reason (str)

Return type:

None

file_path: str¶

reason: str¶

class kelvin.sdk.services.template.FileMapping(**data)[source]¶

Bases: BaseModel

File mapping in template.yaml.

Parameters:

source (str)
dest (str)

source: str¶

dest: str¶

property is_common: bool¶: Check if this file comes from the common directory.

property source_path: str¶

prefix.

Type:: Get the actual source path without the common

property is_jinja2: bool¶: Check if this is a Jinja2 template file.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class kelvin.sdk.services.template.SymlinkMapping(**data)[source]¶

Bases: BaseModel

Symlink mapping in template.yaml.

Parameters:

link (str)
target (str)

link: str¶

target: str¶

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class kelvin.sdk.services.template.TemplateManifest(**data)[source]¶

Bases: BaseModel

Parsed template.yaml manifest file.

Parameters:

name (str)
description (str)
app_type (str)
files (list[FileMapping])
symlinks (list[SymlinkMapping])

name: str¶

description: str¶

app_type: str¶

files: list[FileMapping]¶

symlinks: list[SymlinkMapping]¶

classmethod from_dict(data, template_name)[source]¶

Create a TemplateManifest from a dictionary.

Parameters:

data (dict[str, object]) – The manifest data dictionary.
template_name (str) – Name of the template (for error messages).

Return type:

TemplateManifest

Returns:

Parsed TemplateManifest.

Raises:

ManifestError – If validation fails.

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class kelvin.sdk.services.template.TemplateInfo(name, description, app_type)[source]¶

Bases: object

Information about an available template.

Parameters:

name (str)
description (str)
app_type (str)

name: str¶

description: str¶

app_type: str¶

class kelvin.sdk.services.template.TemplateContext(app_name, app_version, app_description='', title='', app_type='', spec_version='5.0.0', inputs=<factory>, outputs=<factory>)[source]¶

Bases: object

Context for rendering an application template.

This contains all the variables available to Jinja2 templates.

Parameters:

app_name (str)
app_version (str)
app_description (str)
title (str)
app_type (str)
spec_version (str)
inputs (list[dict[str, str]])
outputs (list[dict[str, str]])

app_name: str¶

app_version: str¶

app_description: str = ''¶

title: str = ''¶

app_type: str = ''¶

spec_version: str = '5.0.0'¶

inputs: list[dict[str, str]]¶

outputs: list[dict[str, str]]¶

__post_init__()[source]¶

Set defaults for computed fields.

Return type:: None

class kelvin.sdk.services.template.CreateAppResult(app_path, files_created=<factory>, symlinks_created=<factory>)[source]¶

Bases: object

Result of app creation from template.

Parameters:

app_path (Path)
files_created (list[str])
symlinks_created (dict[str, str])

app_path: Path¶

files_created: list[str]¶

symlinks_created: dict[str, str]¶

class kelvin.sdk.services.template.SyncAgentResult(agents_md_path, symlinks_created=<factory>)[source]¶

Bases: object

Result of syncing agent files.

Parameters:

agents_md_path (Path)
symlinks_created (dict[str, str])

agents_md_path: Path¶

symlinks_created: dict[str, str]¶

class kelvin.sdk.services.template.TemplateService(templates_dir=None)[source]¶

Bases: object

Application template management and project scaffolding.

This is a leaf service with no dependencies. It manages templates stored in the templates/ directory and creates new applications from those templates.

Templates are defined using manifest.yaml files that specify what files to copy/render and what symlinks to create.

Parameters:: templates_dir (Optional[Path])

__init__(templates_dir=None)[source]¶

Initialize TemplateService.

Parameters:: templates_dir (Optional[Path]) – Path to templates directory. Defaults to the built-in templates directory.
Return type:: None

property templates_dir: Path¶: Get the templates directory path.

property common_dir: Path¶: Get the common files directory path.

list_templates()[source]¶

List all available application templates.

Return type:: list[TemplateInfo]
Returns:: List of TemplateInfo objects.

get_template(name)[source]¶

Get template info by name.

Parameters:

name (str) – Template name.

Return type:

TemplateInfo

Returns:

TemplateInfo for the template.

Raises:

TemplateNotFoundError – If template doesn’t exist.
ManifestError – If manifest is invalid.

create_app(template_name, target_dir, context)[source]¶

Create an application from template.

Creates target_dir/<context.app_name>/ with all template files.

Parameters:

template_name (str) – Name of the template to use.
target_dir (Path) – Parent directory where app folder will be created.
context (TemplateContext) – Template context with variables for rendering.

Return type:

CreateAppResult

Returns:

CreateAppResult with paths to created files.

Raises:

TemplateNotFoundError – If template doesn’t exist.
ManifestError – If manifest is invalid.
TemplateRenderError – If template rendering fails.
FileCreationError – If file creation fails.

preview_files(template_name, context)[source]¶

Preview file paths that would be created.

Parameters:

template_name (str) – Name of the template to use.
context (TemplateContext) – Application context (only app_name is used).

Return type:

list[str]

Returns:

List of relative file paths that would be created.

Raises:

TemplateNotFoundError – If template doesn’t exist.
ManifestError – If manifest is invalid.

clear_cache()[source]¶

Clear the manifest cache.

Return type:: None

sync_agent_file(target_dir, template_name='app')[source]¶

Sync AGENTS.md, .skills directory, and related symlinks to an existing app directory.

Reads the template manifest to find the AGENTS.md file, .skills directory, and related symlinks, then copies/creates them. This avoids duplicating the file/symlink definitions — template.yaml is the single source of truth.

Parameters:

target_dir (Path) – Directory to sync the agent file into.
template_name (str) – Template to read manifest from (default: “app”).

Return type:

SyncAgentResult

Returns:

SyncAgentResult with paths to created files and symlinks.

Raises:

TemplateServiceError – If the target directory is not a Kelvin app (no app.yaml), or if AGENTS.md is not found in the template manifest.

Authentication¶

OAuth/PKCE authentication service package.

Auth Service¶

Core authentication service handling OAuth flows and token operations.

AuthService - Pure authentication service for OAuth flows.

This service handles OAuth flows and token operations but does NOT store credentials. The calling code (Commands layer) is responsible for deciding when/if to persist tokens using CredentialStore.

class kelvin.sdk.services.auth.auth_service.AuthService[source]¶

Bases: object

Pure authentication service - returns tokens, no storage.

This service handles OAuth flows and token operations but does NOT store credentials. The calling code (Commands layer) is responsible for deciding when/if to persist tokens using CredentialStore.

REALM: Final[str] = 'kelvin'¶

BROWSER_CLIENT_ID: Final[str] = 'kelvin-sdk'¶

USER_CLIENT_ID: Final[str] = 'kelvin-client'¶

CALLBACK_SERVER_PORT: Final[int] = 38638¶

__init__()[source]¶

Initialize AuthService. No dependencies required.

Return type:: None

login_browser(url, open_browser=True, on_auth_url=None)[source]¶

Start browser-based OAuth flow, return tokens.

Parameters:

url (str) – The base URL of the Kelvin platform.
open_browser (bool) – Whether to automatically open the browser.
on_auth_url (Optional[Callable[[str], None]]) – Optional callback invoked with the authorization URL before waiting for the browser callback.

Return type:

TokenResponse

Returns:

TokenResponse containing access and refresh tokens.

Raises:

OAuthFlowError – If the OAuth flow fails.

get_authorization_url(url)[source]¶

Get the authorization URL for manual browser-based login.

This is useful when automatic browser opening is not desired.

Parameters:: url (str) – The base URL of the Kelvin platform.
Return type:: str
Returns:: The authorization URL to open in a browser.
Raises:: OAuthFlowError – If fetching the authorization endpoint fails.

login_password(url, username, password, totp=None)[source]¶

Authenticate with username/password, return tokens.

Parameters:

url (str) – The base URL of the Kelvin platform.
username (str) – The user’s username.
password (str) – The user’s password.
totp (Optional[str]) – Optional TOTP code for 2FA.

Return type:

TokenResponse

Returns:

TokenResponse containing access and refresh tokens.

Raises:

OAuthFlowError – If authentication fails.

login_client_credentials(url, client_id, client_secret)[source]¶

Authenticate with client credentials, return tokens.

Parameters:

url (str) – The base URL of the Kelvin platform.
client_id (str) – The OAuth client ID.
client_secret (str) – The OAuth client secret.

Return type:

TokenResponse

Returns:

TokenResponse containing access token (no refresh token for this grant).

Raises:

OAuthFlowError – If authentication fails.

refresh_tokens(url, refresh_token, oauth_client_id=None)[source]¶

Refresh expired tokens, return new tokens.

Parameters:

url (str) – The base URL of the Kelvin platform.
refresh_token (str) – The refresh token.
oauth_client_id (Optional[str]) – The Keycloak OAuth client ID that was used to obtain the tokens. Falls back to USER_CLIENT_ID if not provided.

Return type:

TokenResponse

Returns:

TokenResponse containing new tokens.

Raises:

OAuthFlowError – If token refresh fails.

verify_token(url, access_token)[source]¶

Verify token validity, return user info.

Parameters:

url (str) – The base URL of the Kelvin platform.
access_token (str) – The access token to verify.

Return type:

dict[str, object]

Returns:

Dictionary containing user information (sub, email, name, etc.).

Raises:

OAuthFlowError – If the token is invalid or verification fails.

logout(url, refresh_token, oauth_client_id=None)[source]¶

Revoke tokens on the server.

Parameters:

url (str) – The base URL of the Kelvin platform.
refresh_token (str) – The refresh token to revoke.
oauth_client_id (Optional[str]) – The Keycloak OAuth client ID that was used to obtain the tokens. Falls back to USER_CLIENT_ID if not provided.

Raises:

OAuthFlowError – If logout fails.

Return type:

None

fetch_platform_metadata(url)[source]¶

Fetch platform metadata from the /metadata endpoint.

Parameters:: url (str) – The base URL of the Kelvin platform.
Return type:: PlatformMetadata
Returns:: PlatformMetadata containing docker registry info and other platform details.
Raises:: OAuthFlowError – If fetching metadata fails.

Callback Server¶

Local HTTP server for handling OAuth redirect callbacks.

Local HTTP server for handling OAuth callbacks.

class kelvin.sdk.services.auth.callback_server.CallbackResult(code=None, state=None)[source]¶

Bases: object

Result from OAuth callback.

Parameters:

code (str | None)
state (str | None)

code: str | None = None¶

state: str | None = None¶

class kelvin.sdk.services.auth.callback_server.CallbackHandler(request, client_address, server)[source]¶

Bases: BaseHTTPRequestHandler

HTTP request handler for OAuth callbacks.

authorization_code: str | None = None¶

state: str | None = None¶

error: str | None = None¶

error_description: str | None = None¶

callback_received: Event | None = None¶

do_GET()[source]¶

Handle GET requests to the callback endpoint.

Return type:: None

finish()[source]¶

Override finish to ensure connection is properly closed.

Return type:: None

log_message(format, *args)[source]¶

Suppress log messages.

Return type:

None

Parameters:

format (str)
args (object)

class kelvin.sdk.services.auth.callback_server.CallbackServer(port=8080)[source]¶

Bases: object

Local HTTP server for handling OAuth callbacks.

Parameters:: port (int)

ADDRESS: Final[str] = 'localhost'¶

__init__(port=8080)[source]¶

Initialize the callback server.

Parameters:: port (int) – The port to listen on.
Return type:: None

port: int¶

server: HTTPServer | None¶

thread: Thread | None¶

get_callback_url()[source]¶

Get the callback URL for the OAuth flow.

Return type:: str
Returns:: The callback URL.

start()[source]¶

Start the callback server.

Raises:: OAuthFlowError – If the server fails to start.
Return type:: None

wait_for_callback(timeout=120.0)[source]¶

Wait for the OAuth callback.

Parameters:: timeout (float) – Maximum time to wait in seconds.
Return type:: CallbackResult
Returns:: CallbackResult with the authorization code and state.
Raises:: OAuthFlowError – If timeout or error occurs.

stop()[source]¶

Stop the callback server.

Return type:: None

Auth Errors¶

Authentication-specific exception hierarchy.

Authentication service errors.

exception kelvin.sdk.services.auth.errors.AuthServiceError(message=None, exit_code=None)[source]¶

Bases: AuthenticationError

Base exception for AuthService errors.

Parameters:

message (str)
exit_code (int)

exception kelvin.sdk.services.auth.errors.KeycloakError(message=None, exit_code=None)[source]¶

Bases: AuthServiceError

Raised when there is an error related to Keycloak operations.

Parameters:

message (str)
exit_code (int)

exception kelvin.sdk.services.auth.errors.OAuthFlowError(message=None, exit_code=None)[source]¶

Bases: AuthServiceError

Raised when there is an error during the OAuth flow.

Parameters:

message (str)
exit_code (int)

Keycloak Client¶

Keycloak OAuth/OIDC client implementation.

class kelvin.sdk.services.auth.keycloak_client.KeycloakClient(base_url, realm, client_id, client_secret=None)[source]¶

Bases: object

Client for interacting with Keycloak OAuth/OIDC endpoints.

Parameters:

base_url (str)
realm (str)
client_id (str)
client_secret (Optional[str])

TIMEOUT: Final[float] = 10.0¶

__init__(base_url, realm, client_id, client_secret=None)[source]¶

Initialize the Keycloak client.

Parameters:

base_url (str) – The base URL of the Kelvin platform.
realm (str) – The Keycloak realm name.
client_id (str) – The OAuth client ID.
client_secret (Optional[str]) – Optional client secret for confidential clients.

Return type:

None

get_oidc_configuration()[source]¶

Fetch the OIDC configuration from Keycloak’s well-known endpoint.

Return type:: dict[str, object]
Returns:: Dictionary containing OIDC configuration.
Raises:: KeycloakError – If the request fails.

get_authorization_endpoint()[source]¶

Get the authorization endpoint URL.

Return type:: str
Returns:: Authorization endpoint URL.
Raises:: KeycloakError – If endpoint is not found.

get_token_endpoint()[source]¶

Get the token endpoint URL.

Return type:: str
Returns:: Token endpoint URL.
Raises:: KeycloakError – If endpoint is not found.

get_userinfo_endpoint()[source]¶

Get the userinfo endpoint URL.

Return type:: str
Returns:: Userinfo endpoint URL.
Raises:: KeycloakError – If endpoint is not found.

get_logout_endpoint()[source]¶

Get the logout endpoint URL.

Return type:: str
Returns:: Logout endpoint URL.
Raises:: KeycloakError – If endpoint is not found.

get_revocation_endpoint()[source]¶

Get the token revocation endpoint URL (RFC 7009).

Tries revocation_endpoint from OIDC config first, then falls back to the standard Keycloak path.

Return type:: str
Returns:: Revocation endpoint URL.

exchange_code_for_tokens(code, redirect_uri, code_verifier)[source]¶

Exchange an authorization code for access and refresh tokens.

Parameters:

code (str) – The authorization code from the callback.
redirect_uri (str) – The redirect URI used in the authorization request.
code_verifier (str) – The PKCE code verifier.

Return type:

TokenResponse

Returns:

TokenResponse containing access and refresh tokens.

Raises:

KeycloakError – If the token exchange fails.

authenticate_with_password(username, password, totp=None)[source]¶

Authenticate using Resource Owner Password Credentials Grant.

Parameters:

username (str) – The user’s username.
password (str) – The user’s password.
totp (Optional[str]) – Optional TOTP code for 2FA.

Return type:

TokenResponse

Returns:

TokenResponse containing access and refresh tokens.

Raises:

KeycloakError – If the authentication fails.

authenticate_with_client_credentials(client_id, client_secret)[source]¶

Authenticate using Client Credentials Grant.

Parameters:

client_id (str) – The OAuth client ID.
client_secret (str) – The OAuth client secret.

Return type:

TokenResponse

Returns:

TokenResponse containing access token (no refresh token for this grant).

Raises:

KeycloakError – If the authentication fails.

refresh_access_token(refresh_token)[source]¶

Refresh the access token using a refresh token.

Parameters:: refresh_token (str) – The refresh token.
Return type:: TokenResponse
Returns:: TokenResponse containing new tokens.
Raises:: KeycloakError – If the token refresh fails.

get_user_info(access_token)[source]¶

Fetch user information from Keycloak’s userinfo endpoint.

Parameters:: access_token (str) – The access token.
Return type:: dict[str, object]
Returns:: Dictionary containing user information.
Raises:: KeycloakError – If the request fails.

logout(refresh_token)[source]¶

Logout by revoking the refresh token (RFC 7009).

Parameters:: refresh_token (str) – The refresh token to revoke.
Raises:: KeycloakError – If the revocation fails.
Return type:: None

Auth Models¶

Authentication data models.

Authentication models.

class kelvin.sdk.services.auth.models.TokenResponse(**data)[source]¶

Bases: BaseModel

OAuth token response from Keycloak.

This represents the raw token response from the OAuth server. Pydantic handles validation automatically.

Parameters:

access_token (str)
expires_in (int)
refresh_token (str | None)
refresh_expires_in (int | None)

access_token: str¶

expires_in: int¶

refresh_token: str | None¶

refresh_expires_in: int | None¶

model_config: ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

PKCE Utilities¶

PKCE (Proof Key for Code Exchange) utilities for OAuth 2.0.

kelvin.sdk.services.auth.pkce.generate_code_verifier()[source]¶

Generate a cryptographically random code verifier for PKCE.

Return type:: str
Returns:: A URL-safe base64 encoded random string (43 characters).

kelvin.sdk.services.auth.pkce.generate_code_challenge(code_verifier)[source]¶

Generate a code challenge from a code verifier for PKCE.

Parameters:: code_verifier (str) – The code verifier string.
Return type:: str
Returns:: URL-safe base64 encoded SHA256 hash of the verifier.

kelvin.sdk.services.auth.pkce.generate_state()[source]¶

Generate a random state parameter for CSRF protection.

Return type:: str
Returns:: A random URL-safe string.

kelvin.sdk.services.auth.pkce.build_authorization_url(authorization_endpoint, client_id, redirect_uri, code_challenge, state, scope='openid profile email')[source]¶

Build the OAuth authorization URL with PKCE parameters.

Parameters:

authorization_endpoint (str) – The Keycloak authorization endpoint.
client_id (str) – The OAuth client ID.
redirect_uri (str) – The redirect URI (callback URL).
code_challenge (str) – The PKCE code challenge.
state (str) – The state parameter for CSRF protection.
scope (str) – The OAuth scopes to request.

Return type:

str

Returns:

The complete authorization URL.