Copy page

Client libraries

Python SDK

unikraft-cloud is the official Python SDK for the Unikraft Cloud platform and control-plane APIs.

It has two layers. The idiomatic layer is what you reach for: envelope-free results, automatic pagination, chainable references, and multi-metro fan-out. The plumbing layer underneath mirrors the OpenAPI specification exactly, and stays available for anything the idiomatic layer does not cover yet.

The SDK is async-only.

Installation

Code
pip install unikraft-cloud

Requires Python 3.10 or newer.

Quickstart

Code
import asyncio

from unikraft_cloud import UnikraftCloud


async def main() -> None:
    async with UnikraftCloud() as ukc:  # token from UKC_TOKEN
        instance = await ukc.metro("fra").instances.create(
            image="nginx:latest", memory_mb=256, autostart=True
        )
        print(instance.name, instance.uuid, instance.metro)


asyncio.run(main())

The client owns a connection pool, so close it when you are done — either with async with, or by awaiting ukc.aclose().

Authentication

Pass a bearer token to the constructor, or set the UKC_TOKEN environment variable. Create a token in the Unikraft Cloud console.

Configuration

Code
ukc = UnikraftCloud(
    token="...",  # falls back to UKC_TOKEN
    metro="fra",  # falls back to UKC_METRO; omit to cover every metro
)

Argument	Purpose
token	Bearer token. Falls back to UKC_TOKEN.
metro	The metro code that operations default to, such as fra, or a full http(s):// URL for a staging or self-hosted deployment. Falls back to UKC_METRO. A metro code leaves the other metros reachable with ukc.metro(...); a URL pins the client to that endpoint, and naming another metro then raises.
metros	The metros operations cover by default: "all", one metro, or a list. Creating a resource needs exactly one, so name a metro somewhere when you create.
base_url	Explicit platform API base URL. It settles where requests go, so it overrides metro and UKC_METRO alike, and pins the client to that one endpoint.
control_plane_url	Override the control-plane API base URL.
headers	Extra headers sent with every request.
user_agent	Override the default User-Agent.
http	An httpx.AsyncClient to send through. Supplying one makes its lifetime yours.
transport	An httpx.AsyncBaseTransport, chiefly for testing with httpx.MockTransport.
trust_env	Honour HTTP_PROXY/HTTPS_PROXY/NO_PROXY. Defaults to True.
timeout	Timeout for every request. Omitted, an injected http client keeps its own; otherwise the default bounds connecting but not reading, because wait operations block for as long as you asked.

Metros

The platform API is metro-scoped. By default the client is account-wide: reads ask every metro the account can reach and merge the answers as they arrive, and each result carries the metro it came from.

Code
# Every metro, merged as the pages arrive. Await the listing instead for a list.
async for inst in ukc.instances.list(details=True):
    print(inst.metro, inst.name, inst.state)
every = await ukc.instances.list(details=True)  # one pass each: call list() again for more

# One metro. Because it is known, no lookup is needed.
await ukc.metro("fra").instances.get(name="web").suspend()

# Several metros, for one call or for a whole client.
async for inst in ukc.instances.list(metros=["fra", "dal"]):
    ...
scoped = ukc.metros(["fra", "dal"])

# A listing you stop reading holds a page of every metro, so close it.
async with ukc.instances.list() as listing:
    async for inst in listing:
        break

# What the account can reach, as the control plane reports it.
for endpoint in await ukc.available_metros():
    print(endpoint.metro, endpoint.base_url)

Naming metros is also how you skip metro discovery, which is otherwise one extra request per client. The metros page lists the regions themselves.

References

A resource is addressed by name or uuid — one or the other, because the API validates whichever field it is given.

Code
await ukc.instances.get(name="web")
await ukc.instances.get(uuid="550e8400-e29b-41d4-a716-446655440000")

A name is only unique within a metro, so the same name can exist in several. Add metro= to say which you mean, which also saves a lookup:

Code
await ukc.instances.get(name="web", metro="fra")

Without it, and with more than one metro in scope, the SDK asks every metro. If the name matches in several it raises AmbiguousRefError rather than picking one — with the matches attached, so recovering costs no further requests:

Code
from unikraft_cloud import AmbiguousRefError

try:
    await ukc.instances.get(name="web")
except AmbiguousRefError as err:
    print(err.metros)  # ("fra", "dal")
    print([m.uuid for m in err.matches])

To act on all of them deliberately, use each():

Code
await ukc.instances.each(name="web").suspend()  # in every metro that has one

Bulk operations take a sequence of references, as Ref objects, plain dicts, or names:

Code
from unikraft_cloud import Ref

await ukc.instances.delete([Ref(uuid="a"), {"name": "b"}, "web"])

An operation the API could only carry out in part raises, naming what failed. What did succeed is on err.results, so a partial failure costs nothing already done:

Code
from unikraft_cloud import NotFoundError

try:
    await ukc.instances.delete(["web", "gone"])
except NotFoundError as err:
    print([deleted.name for deleted in err.results])  # ["web"]

Chainable handles

Single-resource operations return a handle rather than a coroutine, so they compose. A handle is awaitable too, so awaiting one gives you the resource:

Code
inst = await ukc.instances.get(name="web")  # the instance
await ukc.instances.get(name="web").suspend()  # the suspend

logs = await (
    ukc.metro("fra")
    .instances.create(image="nginx:latest")
    .wait(state="running", timeout_seconds=30)
    .logs(offset=-4096)
)

Nothing is sent until a handle is awaited or an operation is chained onto it. With one metro in scope, get(name=...).suspend() is a single request; when the scope spans metros, the instance is located first so the operation reaches the metro that holds it.

A handle that is dropped without ever being awaited emits a RuntimeWarning: unlike a forgotten await on a coroutine, nothing else would tell you no request was sent.

A handle is awaitable but is not a coroutine, so asyncio.gather(...) takes one while asyncio.create_task(...) does not; wrap it in asyncio.ensure_future(...) for a task.

Updating a resource

Properties are keyword arguments. A value sets the property, REMOVE clears it out, and anything omitted is left alone — all in one request.

Code
from unikraft_cloud import REMOVE

await ukc.instances.get(name="web").update(memory_mb=512, vcpus=2, autokill=REMOVE)

When set is not what you mean — merging into a property, or removing individual members — stage the operations and apply them together:

Code
await (
    ukc.instances.get(name="web")
    .edit()
    .set(memory_mb=512)
    .add(env={"LOG_LEVEL": "debug"}, tags=["prod"])
    .delete(env=["OLD_FLAG"])
    .apply()
)

apply() returns a handle, so the chain continues. For anything keyword arguments cannot express, patch() takes the raw triples.

Resources

instances, volumes, services, certificates and users hang off any scope — ukc, ukc.metro("fra") or ukc.metros([...]).

Creating one takes the properties the API describes as keyword arguments, and a property it does not have is a TypeError rather than a field the server quietly ignores.

Code
await ukc.volumes.get(name="data").attach(to="web", at="/data")
await ukc.services.get(name="web").update(hard_limit=10)
await ukc.certificates.get(name="tls").update(chain=chain_pem, pkey=key_pem)

for quota in await ukc.users.quotas():
    print(quota.metro, quota.used, quota.hard)

Errors

Every failure is an UnikraftCloudError, so one except catches the lot. Its kind says which layer failed ("http", "network", "parse" or "fanout") and status carries the HTTP status where there was one.

Code
from unikraft_cloud import NotFoundError, UnikraftCloudError

try:
    await ukc.instances.get(name="web")
except NotFoundError:
    ...
except UnikraftCloudError as err:
    print(err.kind, err.status, err.errors)

AuthenticationError (401/403), NotFoundError (404), AlreadyExistsError (409), RateLimitError (429) and ServerError (5xx) are raised for the statuses they name, and all subclass UnikraftCloudError. The API reports some failures inside an otherwise-200 envelope, per item; those carry the API's own code on err.errors[n].code and are raised with the status that says the same thing.

A wait() that runs out of time raises WaitTimeoutError, which is also a builtin TimeoutError, and carries the state the API last saw:

Code
try:
    await ukc.instances.get(name="web").wait(state="running", timeout_seconds=30)
except TimeoutError as err:
    print(err.state)  # e.g. "starting"

When the API attaches a warning to an answer — a deprecated field, say — the SDK issues it as a Python UnikraftCloudWarning, so the standard warnings filters apply.

A multi-metro operation that only partly succeeded raises MetroFanoutError. An iteration yields everything the healthy metros returned before raising, so a partial failure never costs you the whole answer; operations that cannot yield as they go attach what did arrive to err.results.

Code
from unikraft_cloud import MetroFanoutError

try:
    async for inst in ukc.instances.list():
        ...
except MetroFanoutError as err:
    print([failure.metro for failure in err.failures])

The plumbing layer

Every operation in the specification is available raw, returning the response envelope untouched. Each client talks to exactly one metro, and a single call can be redirected with base_url=.

Code
res = await ukc.api.platform.instances.get_instances(count=10)
print(res.status, res.op_time_us, res.data.instances)

await ukc.api.controlplane.metros.list_metros()

# Or per resource, alongside its idiomatic client.
await ukc.instances.api.get_instance_metrics(uuid=["..."])

It can also be used on its own, without the idiomatic layer:

Code
from unikraft_cloud import ApiClientConfig
from unikraft_cloud.api.platform import PlatformApi

config = ApiClientConfig(base_url="https://api.fra.unikraft.cloud", token=token)
async with PlatformApi(config) as api:
    res = await api.instances.get_instances(count=10)

Self-hosted and staging deployments

metro (and UKC_METRO) also accepts a full http(s):// base URL, which is used verbatim instead of being expanded into https://api.<metro>.unikraft.cloud. A named endpoint is then the only endpoint there is: no metro discovery is attempted, and naming another metro raises. Point the control plane at a matching deployment with control_plane_url.

Code
ukc = UnikraftCloud(
    token=token,
    metro="https://api.staging.example.internal",
    control_plane_url="https://controlplane.staging.example.internal",
)

Source

The SDK source lives at github.com/unikraft-cloud/python-sdk, and the published package at pypi.org/project/unikraft-cloud. See examples/ for complete programs.

Learn more

• JavaScript SDK: the client library for the same platform API in JavaScript and TypeScript.
• Go SDK: the client library for the same platform API in Go.
• Metros: the regions that the SDK fans out across.
• Instances: create and manage the instances that the SDK drives.
• Unikraft Cloud's REST API reference, which the plumbing layer mirrors.