contree-client 0.1.2

Python client for the Contree API, generated from the OpenAPI specification

contree-client

Python client for the Contree API — sandboxed code execution combining virtual machine isolation with container operational efficiency.

Synchronous clients are named ContreeClient, asynchronous ones are named ContreeAsyncClient; every backend shares the same interface:

Backend	Extra	Import
http.client	— (stdlib)	from contree_client.http import ContreeClient
urllib3	contree-client[urllib3]	from contree_client.urllib3 import ContreeClient
requests	contree-client[requests]	from contree_client.requests import ContreeClient
httpx	contree-client[httpx]	from contree_client.httpx import ContreeClient / ContreeAsyncClient
aiohttp	contree-client[aiohttp]	from contree_client.aiohttp import ContreeAsyncClient

Or let the package pick the first installed backend:

from contree_client.sync import ContreeClient
from contree_client.asyncio import ContreeAsyncClient

Quick start

from contree_client.sync import ContreeClient

# reuse a saved Contree profile,
# or pass credentials directly: ContreeClient("IAM_TOKEN", project=...)
with ContreeClient.from_profile() as client:
    response = client.spawn_instance("uname -a", "tag:ubuntu:latest", shell=True)
    for event in client.iter_operation_events(response.uuid, follow=True):
        print(event.type, event.data)

Use cases

Run a command and read its output:

from contree_client import OperationStatus

response = client.spawn_instance(
    "echo hello",
    "tag:ubuntu:latest",
    shell=True,
    disposable=True,   # do not persist a result image
    timeout=60,
)
operation = client.wait_operation(response.uuid)
assert operation.status is OperationStatus.SUCCESS
print(operation.metadata.result.stdout.as_text())

Follow the live event stream with transparent reconnection (Last-Event-Id resume; ends on the completion event):

import sys

from contree_client import decode_chunk

for event in client.follow_operation_events(response.uuid):
    if event.type in ("stdout", "stderr"):
        sys.stdout.buffer.write(decode_chunk(event.data))

Retry transient failures (network errors, 410/425/5xx) automatically:

from contree_client import RetryPolicy

client = ContreeClient.from_profile(retry=RetryPolicy())

Stage files into the sandbox:

from contree_client import FileSpec

uploaded = client.upload_file(open("data.txt", "rb"))
client.spawn_instance(
    "wc -l /work/data.txt",
    "tag:ubuntu:latest",
    shell=True,
    files={"/work/data.txt": FileSpec(uuid=uploaded.uuid, mode="0644")},
)

Import an image and inspect it without spawning anything:

from contree_client import ImageImportRegistry

operation_id = client.import_image(
    ImageImportRegistry(url="docker://docker.io/library/busybox:latest"),
    tag="busybox:latest",
)
image_uuid = client.inspect_find_image_by_tag("busybox:latest")
print(client.inspect_image_list(image_uuid, "/etc").files)

Download a directory as a tar stream:

with open("etc.tar", "wb") as archive:
    for chunk in client.inspect_image_archive(image_uuid, "/etc"):
        archive.write(chunk)

Asynchronous flavour mirrors the sync API exactly:

from contree_client.asyncio import ContreeAsyncClient

async with ContreeAsyncClient.from_profile() as client:
    me = await client.whoami()
    print(me.permissions)

Test your code without a server — the in-memory double from contree_client.testing implements the same interface as every real backend; unmocked methods raise, calls are recorded:

from contree_client.testing import ContreeClient

client = ContreeClient()
client.mock("spawn_instance", spawn_response)
client.mock("iter_operation_events", [stdout_event, exit_event])

run_my_code(client)

assert client.calls_for("spawn_instance")

Debug logging (redacted, off by default):

import logging

from contree_client.types import set_log_level

set_log_level(logging.DEBUG)

Copyright

Nebius B.V. 2026, Licensed under the Apache License, Version 2.0 (see "LICENSE" file).