kainguru-sdk

Python SDK for the Kainguru ML platform — run models and fine-tune them from any Python application.

• Python 3.10+
• Sync and async clients with an identical surface (KainguruClient, AsyncKainguruClient)
• Typed pydantic v2 models
• Built-in polling with configurable timeout and backoff
• Automatic retry on 429 / 5xx with exponential backoff (honors Retry-After)

Covers /v1/executions and /v1/fine-tuning.

Installation

pip install kainguru-sdk

Import name is kainguru:

from kainguru import KainguruClient

Authentication

You must pass both an API key and the base URL explicitly. The API key is issued from the Kainguru Dashboard (it begins with kg_); the base URL is the full host including the /api context path:

from kainguru import KainguruClient

client = KainguruClient(
    api_key="kg_your_api_key",
    base_url="https://your-host/api",
)

Both fields are required. If either is missing or blank, the constructor raises KainguruConfigError (fail fast). There are no defaults, environment presets, or environment-variable fallbacks.

KainguruClient is a context manager — use with to close the HTTP pool automatically.

Quick start (sync)

from kainguru import KainguruClient

with KainguruClient(
    api_key="kg_your_api_key",
    base_url="https://your-host/api",
) as client:
    submitted = client.executions.execute(
        "my-mlflow-id",
        {"prompt": "hello world"},
        output_format="json",
    )

    done = client.executions.await_completion(submitted.id)

    print(done.status)              # ModelStatus.COMPLETED / FAILED
    print(done.execution.output)    # model output

ml_flow_id (first positional arg) is the model's MLflow id — the identifier the Dashboard shows for the model.

Quick start (async)

import asyncio
from kainguru import AsyncKainguruClient


async def main():
    async with AsyncKainguruClient(
        api_key="kg_your_api_key", base_url="https://your-host/api"
    ) as client:
        submitted = await client.executions.execute(
            "my-mlflow-id", {"prompt": "hello world"}, output_format="json"
        )
        done = await client.executions.await_completion(submitted.id)
        print(done.status)


asyncio.run(main())

Executions API

exec = client.executions

# Run a model (returns immediately, typically PENDING)
pending = exec.execute(ml_flow_id, input, output_format=None, exec_id=None)

# Poll the current status once
current = exec.get(id)

# Block until COMPLETED or FAILED (default: 2 s interval, 5 min timeout)
done = exec.await_completion(id)

# Block with custom poll options
done = exec.await_completion(id, poll_interval=1.0, timeout=60.0)

Terminal statuses: COMPLETED, FAILED. REGISTERED is treated as non-terminal. A FAILED job returns from await_completion — inspect dto.status; it is not raised.

Fine-Tuning API

ft = client.fine_tuning

# Start fine-tuning
pending = ft.execute(model_id, "my-fine-tuned-variant", input)

# Get status / poll until done
current = ft.get_status(id)
done = ft.await_completion(id)

(The async client exposes the same methods with await.)

Polling options

client.executions.await_completion(
    id,
    poll_interval=2.0,   # base interval between polls in s (default 2.0)
    timeout=300.0,       # total wall-clock timeout in s (default 300.0)
    backoff=1.5,         # multiply interval each attempt (default 1.0 = fixed)
    max_interval=30.0,   # cap on interval after backoff (default 30.0)
)

On timeout, await_completion raises KainguruTimeoutError, which carries the last DTO seen via err.last_dto.

Configuration

client = KainguruClient(
    api_key="kg_...",                    # required, explicit key
    base_url="https://your-host/api",    # required, full host incl. /api context path
    timeout=30.0,                        # per-request timeout in s
    max_retries=3,                       # retries on 429 / 5xx / network
)

Both api_key and base_url are required and must be passed explicitly — there are no defaults or environment-variable fallbacks. The base URL must be the full host including the dashboard's /api context path.

Every method also accepts a per-request api_key= to override the client key for one call.

Error handling

All errors extend KainguruError.

from kainguru import (
    KainguruAPIError,
    KainguruNotFoundError,
    KainguruTimeoutError,
)

try:
    result = client.executions.await_completion(id)
    if result.status.value == "FAILED":
        ...  # FAILED is returned, not raised — inspect the result
except KainguruNotFoundError as e:
    print("not found:", e.status_code)
except KainguruAPIError as e:
    print(f"HTTP {e.status_code}: {e.body}")
except KainguruTimeoutError as e:
    print("timed out; last:", e.last_dto)

The SDK automatically retries 429 and 5xx responses up to 3 times with exponential backoff (base 1 s, doubling per attempt); Retry-After headers are respected.

Limitations (v0.1.0)

• Cancel endpoints are not exposed yet (they require JWT/Keycloak auth, not an API key).
• input parameters are untyped (dict[str, Any]).
• output_format is a free str; allowed values are not yet enumerated by the backend.

Building or publishing the SDK yourself? See MAINTAINERS.md.