from kaitencloud import KaitenClient

client = KaitenClient(token="ksh_...")

for customer in client.customers.list():
    print(customer.name)

client.instances.report_usage("acme-production", "seats", 1)

Why this SDK

• Typed end to end. Every response is a pydantic model generated from Kaiten's OpenAPI contract; the package ships py.typed and passes mypy --strict.
• Sync and async, identical. KaitenClient and AsyncKaitenClient expose the same methods with the same signatures — the sync client is generated from the async one.
• Complete lists. list() walks every page for you, and refuses to hand back a list the API said was incomplete.
• Retries that never double-bill. Only requests that are safe to repeat are retried. A usage report is never sent twice.
• Errors you can act on. Every failure is a typed exception carrying the API's RFC 9457 problem: code, detail, error_id.
• Both APIs. The Core API your product talks to, and the Platform API that provisions organizations and mints their tokens.
• Webhooks included. Svix signatures verified, all 55 event types parsed into models.
• Flags through OpenFeature. Kaiten speaks OFREP, so flags are defined here and evaluated with the OpenFeature SDK — a vendor-neutral API, not ours.
• Contract-tested. All 108 operations the SDK wraps are exercised against the contract, and every request body is validated against the API's own schema.

Contents

• Installation
• Quick start
• Configuration
• Guide
  • Catalog and licenses · Customers and instances · Usage metering
  • Feature flags · Releases and deployment zones · Webhooks
  • The Platform API
• The playbook
• Working with the SDK: pagination · errors · retries and timeouts · models · logging
• API reference
• How the SDK stays true to the API
• Development

Installation

pip install kaitencloud

or, with uv:

uv add kaitencloud

The SDK supports Python 3.10 to 3.14, and depends only on httpx, pydantic v2 and typing-extensions.

Quick start

Create an organization token (ksh_...) from a service account in the Kaiten console, then point the client at your deployment:

from kaitencloud import KaitenClient

client = KaitenClient(
    base_url="https://kaiten.example.com",  # or set KAITEN_BASE_URL
    token="ksh_...",  # or set KAITEN_AUTH_TOKEN
)

# Every customer, however many pages it takes.
for customer in client.customers.list():
    print(customer.slug, customer.name)

# What an instance consumes, next to what its license grants.
for meter in client.instances.list_usage("acme-production"):
    granted = meter.limit.value if meter.limit else "not granted"
    print(f"{meter.entitlement_slug}: {meter.value.value} of {granted}")

With KAITEN_BASE_URL and KAITEN_AUTH_TOKEN set, a client needs no arguments, and the rest of this README assumes they are. The async client is the same API, awaited:

import asyncio

from kaitencloud import AsyncKaitenClient


async def main() -> None:
    async with AsyncKaitenClient() as client:
        customers, instances = await asyncio.gather(
            client.customers.list(),
            client.instances.list(),
        )
        print(f"{len(customers)} customers, {len(instances)} instances")


asyncio.run(main())

The model in one picture

Organization              you: a SaaS vendor using Kaiten
└── Customer              client.customers         a client of yours
    └── Instance          client.instances         a deployment of your product for that customer
        ├── License       client.licenses          the contract it runs under, granting values
        │                 client.entitlements      ...of the entitlements in your catalog
        ├── Usage         client.instances.report_usage()   what it consumed
        └── Zone          client.deployment_zones  where it runs, and the release deployed there
                          client.releases → client.components

Configuration

Both clients are context managers, and closing one closes the connection pool it created — never one you passed in.

Credentials

Kaiten has several kinds of credential, and each API refuses the others with a deliberately vague error. The SDK recognizes them and raises a clear CredentialError before sending anything when one ends up in the wrong place.

A credential that rotates can be supplied as a callable — it is read before every request, so there is no client to rebuild:

client = KaitenClient(token=lambda: vault.read("kaiten/token"))

Guide

Catalog and licenses

An entitlement says what can be granted and how its usage is measured. A license grants values of entitlements. A numeric grant carries its own enforcement: overage_percent=0 is a hard limit, 10 tolerates 10% over the granted value before usage reports are refused.

from kaitencloud import UNLIMITED, KaitenClient

client = KaitenClient()

client.entitlements.create(
    name="Seats",
    slug="seats",
    type="NUMBER",
    unit_singular="seat",
    unit_plural="seats",
    user_facing=True,
)
client.entitlements.create(name="Single sign-on", slug="sso", type="BOOLEAN")
client.entitlements.create(name="API calls", slug="api-calls", type="NUMBER", reset_period="MONTH")

client.licenses.create(name="Growth", slug="growth", type="PAID")
client.licenses.associate_entitlement("growth", "seats", 50, overage_percent=10)
client.licenses.associate_entitlement("growth", "sso", True)
client.licenses.associate_entitlement("growth", "api-calls", UNLIMITED)

A granted value is a bool for a BOOLEAN entitlement, a number for a NUMBER one, and a dict for a CONFIG one.

A license is one version of a product, its license family. Creating a license without naming a family opens a new one: growth above is version 1 of the family growth. To change what a plan grants without touching the customers already on it, add a version to the family, build it as a draft, and put it on sale when it is ready:

client.licenses.create(name="Growth", type="PAID", family_slug="growth", lifecycle_state="DRAFT")
client.licenses.associate_entitlement("growth-v2", "seats", 100, overage_percent=10)
client.licenses.publish("growth-v2")

family = client.license_families.get("growth")  # now resolves to growth-v2

A new version's slug is <family>-v<version>, and its number is assigned by the server. A family resolves to its default version if it has one, otherwise to its highest-numbered published version — so store the family slug, which survives renames and new versions, rather than a version's. license_families.get("growth", version=1) pins a version whatever its state, and include_versions=True adds the whole history. licenses.archive() withdraws a version from sale: the instances already on it keep it, and no new one can take it until licenses.unarchive().

Customers and instances

from datetime import datetime, timezone

acme = client.customers.create(name="Acme", slug="acme", domain="acme.com")
growth = client.licenses.get("growth")

client.instances.create(
    name="Acme production",
    slug="acme-production",
    customer_id=acme.id,
    license_id=growth.id,
    start_license_date=datetime(2026, 9, 1, tzinfo=timezone.utc),
    end_license_date=datetime(2027, 9, 1, tzinfo=timezone.utc),
    metadata={"region": "eu-west-1"},
)

client.instances.update_status("acme-production", "DEGRADED")
client.instances.update_lifecycle_stage("acme-production", "AT_RISK")

Note — update() methods are full replacements, as the API's PUT is: send every field you want to keep. Datetimes must be timezone-aware: the SDK refuses to guess the zone of a license term.

Usage metering

from kaitencloud import ThresholdExceededError

try:
    client.instances.report_usage("acme-production", "seats", 1)
except ThresholdExceededError as error:
    # Refused at the cap: the granted value plus its overage allowance. Nothing was recorded.
    print("Time to upgrade:", error.detail)

behavior="append" (the default) folds the value into the total through the entitlement's aggregation method; behavior="set" replaces the total, which is also how to correct it.

kaitencloud.usage computes the same boundaries the server enforces, to render a meter or check a report before sending it:

from kaitencloud import usage

grant = client.licenses.get_entitlement("growth", "seats")
current = client.instances.get_usage("acme-production", "seats")
included, used = grant.value.value, current.value.value

usage.maximum_allowed_usage(included, grant.limit_cap_exceeded_overage_percent)  # 55.0
usage.remaining(included, used, grant.limit_cap_exceeded_overage_percent)  # left before refusal
usage.percentage_used(included, used)  # passes 100 once the allowance is in use

Feature flags

Define flags with the builders in kaitencloud.targeting. Targeting rules are CEL expressions, evaluated in order; the first match wins. Kaiten fills the reserved __kaiten namespace itself: evaluate with a customer's slug as the targeting key, and a rule can read the license and the entitlements that customer holds.

from kaitencloud import targeting

client.feature_flags.create(
    name="New checkout",
    slug="new-checkout",
    type="boolean",
    variants=[targeting.variant("on", True), targeting.variant("off", False)],
    default_variant="off",
    targetings=[
        targeting.rule("Growth plan", "__kaiten.license.familySlug == 'growth'", "on"),
        targeting.split_rule("Progressive rollout", "true", {"on": 10, "off": 90}),
    ],
)

This SDK does not evaluate flags, on purpose. Kaiten implements OFREP, the OpenFeature Remote Evaluation Protocol, so evaluation belongs to OpenFeature itself: install the SDK and the generic OFREP provider, point it at the same base URL with the same token, and your application talks to a vendor-neutral API it already knows.

pip install openfeature-sdk openfeature-provider-ofrep

from openfeature import api
from openfeature.contrib.provider.ofrep import OFREPProvider
from openfeature.evaluation_context import EvaluationContext

api.set_provider(
    OFREPProvider(
        base_url="https://kaiten.example.com/api",  # the provider appends /ofrep/v1/...
        headers_factory=lambda: {"Authorization": "Bearer ksh_..."},
    )
)

flags = api.get_client("checkout")
context = EvaluationContext(targeting_key="acme", attributes={"region": "eu"})

if flags.get_boolean_value("new-checkout", False, context):
    ...

details = flags.get_boolean_details("new-checkout", False, context)
print(details.value, details.variant, details.reason)

The targeting key is the subject rollouts bucket on — a customer slug, a user id — and it is what the reserved __kaiten namespace resolves against, so a rule can read the license and the entitlements that subject holds. Splits are sticky: a subject keeps its variant from one evaluation to the next. A failed evaluation returns the default you passed rather than raise, which is OpenFeature's contract, not ours.

client.feature_flags.manifest() returns the OpenFeature manifest — every flag, its type and its default — for tooling that generates typed accessors from flag definitions.

Releases and deployment zones

gateway = client.components.create(name="api-gateway", version="v1.4.0")
release = client.releases.create(version="v1.4.0", component_ids=[gateway.id])

# Pointing a zone at a release records the deployment.
client.deployment_zones.update(
    "eu-west-1", name="AWS eu-west-1", type="production", release_id=release.id
)

Releases are immutable: changing one means deleting it and creating another.

Webhooks

Kaiten delivers webhooks through Svix. verify_webhook() checks the signature and the timestamp of a delivery, then returns its event as a typed model:

import os

from fastapi import FastAPI, Request, Response
from kaitencloud.webhooks import (
    CustomerCreatedEvent,
    InstanceEntitlementCapExceededEvent,
    WebhookVerificationError,
    verify_webhook,
)

app = FastAPI()


@app.post("/webhooks/kaiten")
async def kaiten_webhook(request: Request) -> Response:
    try:
        event = verify_webhook(
            await request.body(),  # the raw body: parsing it first would change the signed bytes
            request.headers,
            secret=os.environ["KAITEN_WEBHOOK_SECRET"],
        )
    except WebhookVerificationError:
        return Response(status_code=400)

    match event:
        case InstanceEntitlementCapExceededEvent(data=cap):
            alert_account_manager(cap.entitlement_slug, cap.overage)
        case CustomerCreatedEvent(data=customer):
            send_welcome(customer.name)

    return Response(status_code=204)

Every event the API publishes has a model named after it, from ComponentCreatedEvent to ReleaseDeletedEvent. An event newer than the SDK you run parses as UnknownWebhookEvent instead of failing.

The Platform API

The Platform API administers a whole Kaiten deployment. It listens on its own port, on no public route, and accepts only a platform credential (ksm_...).

from datetime import timedelta

from kaitencloud import KaitenPlatformClient

platform = KaitenPlatformClient(base_url="http://kaiten-api.kaiten.svc:6001", token="ksm_...")

# Idempotent: the organization's id derives from the identity provider's id.
org = platform.organizations.ensure(external_id="org_2abcDEF", name="Acme")

# A platform credential never acts inside an organization: it mints a token that does.
minted = platform.tokens.mint(
    org.id, name="billing-sync-v2", scopes=["write:instances"], ttl=timedelta(days=90)
)
store_secret(minted.token)  # the plaintext exists in this response only

Revocation cascades. Revoking a platform credential revokes every token it minted, so rotate by minting the new token, adopting it, then revoking the old one. And revoke() takes the token's server-generated slug, which tokens.list() maps back from the name you chose.

old = next(token for token in platform.tokens.list(org.id) if token.name == "billing-sync-v1")
platform.tokens.revoke(org.id, old.slug)

A connector hosted outside Kaiten registers itself on every start — registration is an upsert:

platform.connectors.register(
    name="kaiten.integration.crm.acme",
    version="1.2.0",
    settings_schema={
        "type": "object",
        "properties": {
            "apiKey": {"type": "string", "writeOnly": True}
        },  # a secret: never read back
    },
    entitlement_slug="crm-sync",  # the BOOLEAN entitlement a license must grant to activate it
)

The playbook

examples/playbook.py is a guided tour of a deployment of your own: thirteen steps that build a catalog, a license, a customer and an instance, meter usage until the license refuses a report, evaluate a feature flag, mint and revoke a token and verify a webhook signature — then delete everything they made. Each step prints the call before it makes it, so the script reads as documentation that runs, one step at a time.

uv run python examples/playbook.py --list

examples/README.md is the guide to it: what each step covers, how to pause and resume a run, what it records about itself, and how cleanup deletes only what that run created.

Working with the SDK

Pagination

Every list() returns the whole collection: the SDK walks the cursor at the largest page size the API accepts. If the API ever reports more rows without a usable cursor, list() raises PaginationError rather than return a list it knows is partial.

The audit trail grows without bound, so it takes a ceiling instead:

entries = client.instances.list_audit_trails(
    "acme-production", event_name="INSTANCE_STATUS_CHANGED", limit=100
)

Errors

Every exception derives from KaitenError:

KaitenError
├── CredentialError              a credential is missing, or in the wrong place (raised before sending)
├── APIConnectionError           the API could not be reached
│   └── APITimeoutError
├── APIStatusError               the API answered with a non-2xx status
│   ├── BadRequestError              400
│   ├── AuthenticationError          401  missing, expired or revoked credential
│   ├── PermissionDeniedError        403  a scope is missing
│   ├── NotFoundError                404
│   ├── ConflictError                409
│   │   └── ThresholdExceededError       a usage report refused at the cap
│   ├── UnprocessableEntityError     422
│   ├── RateLimitError               429
│   ├── InternalServerError          5xx
│   │   └── ServiceUnavailableError      503
│   └── FlagEvaluationError          a flag could not be evaluated
├── APIResponseValidationError   a 2xx response that breaks the contract
├── PaginationError              a list the API reported as incomplete
├── WebhookVerificationError     a webhook that fails verification
└── WebhookPayloadError          an authentic webhook that breaks its event contract

An APIStatusError keeps everything the API said. Branch on code rather than on the status: a 409 for a reached limit and a 409 for a taken slug share a status, not a meaning.

from kaitencloud import APIStatusError

try:
    client.customers.create(name="Acme", slug="acme")
except APIStatusError as error:
    print(error.status_code, error.code, error.detail)
    print("Quote this in a support request:", error.error_id)

Retries and timeouts

A request that fails on a connection error, a 408, a 429 or a 5xx is retried with exponential backoff and jitter, honouring Retry-After — but only when repeating it is safe:

Tune it per client, or per call with with_options(), which shares the connection pool:

import httpx

client = KaitenClient(timeout=10.0, max_retries=4)
client.with_options(timeout=1.0, max_retries=0).customers.get("acme")

proxied = KaitenClient(http_client=httpx.Client(proxy="http://proxy.internal:3128"))

Models

Responses are pydantic models with snake_case attributes. They are lenient where the API grows: a field added after your SDK release lands in model_extra, and an unknown enum value parses as a plain string instead of failing.

instance = client.instances.get("acme-production")

instance.customer_slug  # typed attributes
instance.start_license_date  # a timezone-aware datetime
instance.to_dict()  # the exact wire shape, camelCase keys included
instance.model_extra  # fields newer than this SDK

All models, and the typed dictionaries methods accept, are importable from kaitencloud.types.

Logging

The SDK logs to the kaitencloud logger: retries at INFO, flag evaluations that fell back to their default at WARNING. Credentials are never logged.

import logging

logging.getLogger("kaitencloud").setLevel(logging.INFO)

API reference

The Core API clients cover 98 of its 107 operations. Of the nine left out, three are rule-editor tooling for the Kaiten console, four are a signed-in person's own notification feed and preferences — which an API token has no equivalent of — and two are the OFREP evaluation endpoints, which an OpenFeature provider calls. The Platform API clients cover all 10 of its operations. The async clients have the same methods.

How the SDK stays true to the API

An SDK written by hand drifts from the API it wraps, one renamed field at a time — and a request body with one key too many is refused on every call. This one is built so that it cannot drift silently:

1. The contract is vendored. openapi/ holds a snapshot of Kaiten's two OpenAPI documents, copied from kaitencloud/kaiten by task sync:openapi, with the exact commit recorded in openapi/source.yaml.
2. Models are generated from it, by scripts/generate_models.py.
3. Every operation is accounted for. openapi/coverage.yaml maps each operation to the method that calls it; the tests fail when an operation is left unwrapped or a method maps to nothing.
4. Every method is tested against the contract. Each one is called against a fake API that answers like the contract says, and every request body it sends is validated against the operation's JSON Schema — once with all arguments, once with only the required ones.
5. One implementation, two clients. The sync client is generated from the async source by scripts/unasync.py, and the tests check both have the same signatures.

CI checks all of it on every change, on Python 3.10 to 3.14 and on the oldest supported versions of the dependencies.

Development

The workflows are Task targets over uv:

task setup      # install the development environment
task check      # everything CI runs

The playbook is part of the test suite: every step runs against the fake API, so a renamed method or an argument the contract no longer accepts fails in CI rather than halfway through somebody's walkthrough.

Never edit src/kaitencloud/_sync/ or the generated src/kaitencloud/types/ modules by hand: change the async source, the contract or a generator, then run task generate.

Contributions are welcome — CONTRIBUTING.md says how, and every commit carries a DCO sign-off (git commit -s). Report a vulnerability privately, as SECURITY.md describes, never in a public issue.

Versioning

The SDK follows Semantic Versioning: only a major release changes the public API — the changelog says how — while a minor release adds to it and a patch release fixes it.

License

Kaiten SDK for Python is open source and licensed under the Apache License, Version 2.0.

By contributing, you agree to certify your contribution under the Developer Certificate of Origin 1.1.

The Kaiten name and logos are not licensed under Apache-2.0. See the Kaiten trademark policy in the main Kaiten repository.