📦 ConTree SDK

SDK for ConTree: Sandboxes That Branch Like Git. ConTree is a container runtime purpose-built to support research on SWE agents, providing reproducible, versioned filesystem state — like Git for container execution, accessible from Python.

👉 See full feature list and use cases in the documentation →

📥 Get Started

Installation

Install the SDK from a PyPi:

pip install contree-sdk

Quick Start

import asyncio
from contree_sdk import Contree


async def main():
    # Get client
    contree = Contree(token="fake-token")

    # Use image by tag
    image = await contree.images.use("ubuntu:latest")

    # Run command
    result = await image.run(shell='echo "Hello from Contree!"')

    # Output result
    print(result.stdout)


asyncio.run(main())

from contree_sdk import ContreeSync


def main():
    # Get client
    contree = ContreeSync(token="fake-token")

    # Use image by tag
    image = contree.images.use("ubuntu:latest")

    # Run command
    result = image.run(shell='echo "Hello from Contree!"').wait()

    # Output result
    print(result.stdout)


main()

Examples

Ready to explore more? Check out our comprehensive examples:

• Session Management - Working with persistent sessions and state management
• Image Operations - Advanced image pulling, versioning, and management
• Branching Workflows - Complex workflow patterns with image branching

Explore all examples in the examples/ directory

Development Setup

Prerequisites

• Python 3.10 - 3.13
• uv package manager

Env setup

git clone git@github.com:nebius/contree-sdk.git
cd contree-sdk
uv sync

Running Checks

Linting and formatting with Ruff:

uv run ruff check .
uv run ruff format .

Type checking with basedpyright:

uv run basedpyright

Running Tests

uv run pytest

Documentation Dev Server

make rtd-dev

Table of Contents

• Installation
• Quick Start
• Examples
• Development Setup
  • Prerequisites
  • Environment Setup
  • Running Checks
  • Running Tests
  • Documentation Dev Server
• Quick Start (Advanced)
• Core Concepts
  • Sessions and Versioning
  • Subprocess-like interface
  • Stable image UUID
  • Async/sync clients and objects
• Advanced Usage
  • Client configuration
  • Objects reusing
  • File uploading
• License

🚀 Quick Start (Advanced)

import asyncio
import stat

from pathlib import PurePosixPath

from contree_sdk import Contree
from contree_sdk.utils.models.file import UploadFileSpec
from contree_sdk.sdk.objects.image_fs import ImageFile


async def amain():
    # create client
    contree = Contree(token="fake-token")

    # list images
    images = await contree.images()

    # use image by tag (no API call, resolved at execution time)
    ubuntu_image = await contree.images.use("ubuntu:latest")

    # pulling image from a remote registry
    busybox_image = await contree.images.oci("docker://docker.io/busybox:latest")

    # running command
    result0 = await ubuntu_image.run(
        command="/app.sh",
        args=("arg1", "arg2"),
        stdin="input",
        env=dict(http_proxy="http://10.20.30.40:1234"),
        files=[
            UploadFileSpec(source="/local/files/app.sh", mode=stat.S_IXUSR),
            UploadFileSpec(source="/local/files/data_ver1.csv", path=PurePosixPath("/data.csv")),
        ],
    )
    print(result0.stdout)
    print(result0.stderr)

    # running next command
    result1 = await result0.run(shell="echo output.csv | grep something")

    # getting files and directories by path
    items = await result1.ls("files/path")
    print(len(items))

    # iterating through files and directories by path
    for item in await result1.ls("~"):
        print(item.name, item.is_dir)
        if item.is_file:
            # download file
            assert isinstance(item, ImageFile)
            await item.download("/local/files/downloaded/")

    # using session
    session = busybox_image.session()
    await session.run(
        command="/bin/app",
        files=[UploadFileSpec(source="/local/files/app", path="bin/app", mode=stat.S_IXUSR)],
    )
    res = await session.run(command="/bin/cat", args=("result.txt",))
    print(res.stdout)

    # downloading file from session
    await session.download("/tmp/log.jsonl", "/local/logs/session_1.log")

    # or simply reading from file
    content = await session.read("/tmp/log.jsonl")
    print(content.decode())


asyncio.run(amain())

import stat

from contree_sdk import ContreeSync
from contree_sdk.utils.models.file import UploadFileSpec
from contree_sdk.sdk.objects.image_fs import ImageFileSync


def main():
    # Create client
    contree = ContreeSync(token="fake-token")

    # list images
    images = contree.images()

    # Use image by tag (no API call, resolved at execution time)
    ubuntu_image = contree.images.use("ubuntu:latest")

    # Pulling image from a remote registry
    busybox_image = contree.images.oci("docker://docker.io/busybox:latest")

    # running command
    result0 = ubuntu_image.run(
        command="/app.sh",
        args=("arg1", "arg2"),
        stdin="input",
        env=dict(http_proxy="http://10.20.30.40:1234"),
        files=[
            UploadFileSpec(source="/local/files/app.sh", mode=stat.S_IXUSR),
            UploadFileSpec(source="/local/files/data_ver1.csv", path="/data.csv"),
        ],
    ).wait()
    print(result0.stdout)
    print(result0.stderr)

    # running next command
    result1 = result0.run(shell="echo output.csv | grep something").wait()

    # getting files and directories by path
    items = result1.ls("files/path")
    print(len(items))

    # iterating through files and directories by path
    for item in result1.ls("~"):
        print(item.name, item.is_dir)
        if item.is_file:
            assert isinstance(item, ImageFileSync)
            # download file
            item.download("/local/files/downloaded/")

    # using session
    session = busybox_image.session()
    session.run(
        command="/bin/app",
        files=[UploadFileSpec(source="/local/files/app", path="/bin/app", mode=stat.S_IXUSR)],
    ).wait()
    res = session.run(command="cat", args=("result.txt",)).wait()
    print(res.stdout)

    # downloading file from session
    session.download("/tmp/log.jsonl", "/local/logs/session_1.log")

    # or simply reading from file
    content = session.read("/tmp/log.jsonl")
    print(content.decode())


main()

🧠 Core Concepts

Sessions and Versioning

A session is essentially an image whose version automatically updates after each command execution. When you run commands, you're not modifying the original image - instead, each command creates a new version of the image with your changes applied.

import asyncio
from contree_sdk import Contree


async def amain():
    contree = Contree(token="fake-token")

    # Each command creates a new image version
    image = await contree.images.use("busybox:latest")  # busybox:latest
    result1 = await image.run(shell="apt update")  # some-uuid
    result2 = await result1.run(shell="apt install python3")  # another-uuid

    # Sessions work the same way
    session = image.session()  # busybox:latest
    await session.run(shell="touch /app/file1.txt")  # some-uuid
    await session.run(shell="echo 'hello' > /app/file1.txt")  # another-uuid


asyncio.run(amain())

Subprocess-like interface

Any session can provide Subprocess-like interface

proc = session.popen(
    ["cat"],
    text=True,
)
stdout, stderr = proc.communicate("a\nb\nc\n")

import subprocess

proc = session.popen(
    "echo hello && ls -la",
    shell=True,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True,
)
returncode = proc.wait()
print(proc.stdout)

Stable image UUID

Basically one UUID refers to one state of FS, so in case if after running commands on the image, no FS changes are detected, UUID stays the same.

result0 = image.run("echo CHANGES > file.txt").wait()
result1 = result0.run("sleep 5").wait()

assert result1.uuid == result0.uuid

Async/sync clients and objects

Basically every object that is produced by async client is async-friendly and every object is produced by sync client is sync friendly. For example

import asyncio
from contree_sdk import Contree, ContreeSync


async def amain():
    contree_async = Contree(token="fake-token")

    # async client produces async-friendly images objects, so they can be used in async code
    images = await contree_async.images()
    await images[0].run(shell="some command")


asyncio.run(amain())

contree_sync = ContreeSync(token="fake-token")

# while sync client produces sync-friendly images objects, so they can be used in sync code
images = contree_sync.images()
images[0].run(shell="some command").wait()

Advanced Usage

Client configuration

You can create configuration object and use it later in client

from contree_sdk.auth import IAMAuth
from contree_sdk.config import ContreeConfig
from contree_sdk import Contree, ContreeSync

config = ContreeConfig(
    auth=IAMAuth(token="my-token", base_url="https://contree.host.com"),
    transport_timeout=10.0,  # timeout for transport operations
)

client_async = Contree(config)
client = ContreeSync(config)

Authentication

The SDK resolves credentials in the following priority order:

1. Explicit values passed to IAMAuth / JWTAuth constructors.
2. Environment variables — field defaults like NEBIUS_API_KEY and NEBIUS_PROJECT_ID are substituted automatically if the corresponding variable is set.
3. auth.ini — if the contree CLI is installed, credentials written by contree auth are read from ~/.config/contree/auth.ini (or $CONTREE_HOME / $XDG_CONFIG_HOME/contree).

The active profile is taken from the [DEFAULT] section of auth.ini and can be overridden with the CONTREE_PROFILE environment variable.

Objects reusing

You can preconfigure run and then reuse it, for example:

import asyncio
from contree_sdk import Contree


async def amain():
    contree = Contree(token="fake-token")
    image = await contree.images.use("busybox:latest")

    # preconfigure a run that generates random string and writes to file
    preconfigured_run = image.run(shell="echo $RANDOM > /tmp/random.txt")

    # reuse it multiple times
    result1 = await preconfigured_run
    result2 = await preconfigured_run
    result3 = await preconfigured_run

    # each execution will generate different uuid, because each result is gonna be unique


asyncio.run(amain())

File uploading

import asyncio
from contree_sdk import Contree


async def amain():
    contree = Contree(token="fake-token")

    # upload file
    file = await contree.files.upload("/some/local/file.txt")
    print(file.uuid)


asyncio.run(amain())

License

Copyright 2026 Nebius B.V.

Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

Apache and the Apache logo are either registered trademarks or trademarks of The Apache Software Foundation in the United States and/or other countries.