kanyun-sandbox 0.3.8

Python SDK for Kanyun Sandbox v2 control plane

Kanyun Sandbox Python SDK

Python SDK for the Kanyun Sandbox v2 control plane.

Install

pip install kanyun-sandbox

Install a specific release:

pip install kanyun-sandbox==0.3.8

• Package name: kanyun-sandbox
• Import: kanyun_sandbox

Quick Start

from kanyun_sandbox import Client, CreateSandboxRequest

client = Client(
    control_plane_url="http://127.0.0.1:8080",
    api_key="kanyun_xxx",
)

# Create a sandbox from an existing template
handle = client.create_sandbox(CreateSandboxRequest(
    template_name="my-devbox",
    ttl_seconds=3600,
))

print(f"Sandbox: {handle.info.claim_name}")
print(f"Status:  {handle.info.status}")

# When done
handle.destroy()

Typical Workflow

A common usage flow looks like this:

准备镜像 → 查看可用资源 → 创建模板 → (可选) 创建预热池 → 创建 Sandbox → 使用 → 销毁

Step 1: Explore Available Resources

Before creating resources, check what clusters and runtime profiles are available:

# List clusters
clusters = client.list_clusters()
for c in clusters:
    print(f"  id={c.id}, name={c.display_name}, status={c.status}")

# List runtime profiles (defines where sandboxes run)
profiles = client.list_runtime_profiles()
for p in profiles:
    print(f"  id={p.id}, cluster={p.cluster_id}, ns={p.namespace}")

Step 2: Create a Template

You need a container image first (built by yourself or via the platform's image build system). Then create a template that references it.

Before creating a template, you need to know which runtime profile to use. A runtime profile defines which cluster and namespace your sandboxes will run in:

# List available runtime profiles
profiles = client.list_runtime_profiles()
for p in profiles:
    print(f"  id={p.id}, cluster={p.cluster_id}, ns={p.namespace}")

# Use the first available profile
runtime_profile_id = profiles[0].id

Then create the template:

template = client.apply_template("my-devbox", {
    "runtimeProfileId": "rp-default",
    "displayName": "My Dev Box",
    "description": "Custom development environment",
    "defaultTTLSeconds": 3600,
    "maxTTLSeconds": 86400,
    "enabled": True,
    "podTemplate": {
        "spec": {
            "containers": [{
                "name": "main",
                "image": "registry.example.com/my-org/my-devbox:latest",
                "resources": {
                    "requests": {"cpu": "1", "memory": "2Gi"},
                    "limits": {"cpu": "2", "memory": "4Gi"},
                },
            }],
        },
    },
    "moxt": {
        "enabled": True,
        "apiKey": "moxt_api_key",
    },
})
print(f"Template created: {template.name}")

When Moxt mount is enabled, the platform injects the sidecar and the main container can access the mounted content at /moxt/data.

Step 3: Create a Warm Pool (Optional)

If you need sandboxes to start quickly, pre-warm a pool of standby instances:

warm_pool = client.apply_warm_pool("my-devbox-pool", {
    "templateName": "my-devbox",
    "desiredReplicas": 3,
})
print(f"Warm pool: {warm_pool.name}, ready: {warm_pool.ready_replicas}/{warm_pool.desired_replicas}")

Step 4: Create a Sandbox

handle = client.create_sandbox(CreateSandboxRequest(
    template_name="my-devbox",
    ttl_seconds=3600,
    metadata={"user": "alice", "purpose": "code-review"},
))

print(f"Sandbox: {handle.info.claim_name}")
print(f"IP:      {handle.info.ip}")
print(f"Expires: {handle.info.expires_at}")

If you did not create a warm pool, the sandbox starts cold (Pod is created on-demand). You may need to wait for it to become Running:

from time import sleep

handle = client.create_sandbox(CreateSandboxRequest(
    template_name="my-devbox",
    ttl_seconds=3600,
))

# Cold start: poll until Running
while handle.info.status.lower() != "running":
    sleep(3)
    handle.refresh()
    print(f"  status={handle.info.status}")

print(f"Sandbox ready! IP: {handle.info.ip}")

With a warm pool, create_sandbox typically returns immediately with status=Running since a pre-warmed Pod is adopted instantly.

Step 5: Manage Sandbox Lifecycle

# Extend TTL by another 30 minutes
handle.extend(1800)

# Refresh to get latest status
handle.refresh()
print(f"Status: {handle.info.status}")

# Get detailed info (includes metadata)
detail = handle.detail()
print(f"Metadata: {detail.metadata}")

# List events
events = handle.events()
for event in events:
    print(f"  {event.event_type} at {event.occurred_at}")

# Destroy when done
handle.destroy()

Client Options

client = Client(
    control_plane_url="http://127.0.0.1:8080",
    api_key="kanyun_xxx",
    timeout=60.0,               # HTTP timeout in seconds
)

API Reference

Sandbox

Method	Description
create_sandbox(request)	Create a sandbox, returns SandboxHandle
get_sandbox(id)	Get sandbox info by ID
list_sandboxes()	List all running sandboxes
get_sandbox_detail(id)	Get detailed sandbox info (includes metadata)
list_sandbox_events(id)	List sandbox lifecycle events
extend_sandbox(id, ttl_seconds)	Extend a running sandbox's TTL
delete_sandbox(id)	Delete a sandbox
connect_sandbox(id)	Connect to an existing sandbox, returns SandboxHandle

Template

Method	Description
list_templates()	List all templates
get_template(name)	Get template by name
apply_template(name, request)	Create or update a template
delete_template(name)	Delete a template
list_template_materializations(name)	List template materializations across clusters
resync_template_materialization(name)	Trigger re-sync of a template materialization

Warm Pool

Method	Description
list_warm_pools()	List all warm pools
get_warm_pool(name)	Get warm pool by name
apply_warm_pool(name, request)	Create or update a warm pool
delete_warm_pool(name)	Delete a warm pool

SandboxHandle

Method	Description
handle.info	Current SandboxInfo snapshot
handle.daemon	Direct-connect daemon client for process, file, git, and info APIs
handle.extend(ttl_seconds)	Extend sandbox TTL from now
handle.refresh_activity()	Explicitly refresh sandbox activity while using direct-connect daemon APIs
handle.refresh()	Refresh sandbox info from server
handle.detail()	Get sandbox detail
handle.events()	List sandbox events
handle.destroy()	Delete the sandbox

Direct-Connect Daemon

When daemon injection is enabled, sandbox responses include daemon_endpoint and daemon_auth_token. The SDK wraps those credentials in handle.daemon.

Since 0.3.6, handle.daemon covers:

• info: version(), health(), work_dir()
• process: one-shot command execution plus persistent shell sessions
• files: list/info/download/upload, binary-safe bytes upload, bulk upload/download, folder/delete/move/search/find/replace/permissions
• git: clone/status/add/commit/push/pull

In 0.3.7, iter_session_command_logs() adds true chunk-by-chunk log following for asynchronous session commands.

Prefer handle.daemon.info.work_dir() over hard-coding /workspace; the control plane can override the daemon work directory with KANYUN_TOOLBOX_DAEMON_WORK_DIR.

handle = client.create_sandbox(CreateSandboxRequest(
    template_name="aio-devbox",
    ttl_seconds=3600,
))

work_dir = handle.daemon.info.work_dir()
result = handle.daemon.process.execute_command(
    command="pwd && whoami",
    cwd=work_dir,
    timeout=10,
)
print(result["stdout"])

handle.daemon.files.upload(f"{work_dir}/hello.txt", b"Hello from SDK!\n")
handle.daemon.files.upload(f"{work_dir}/blob.bin", bytes([0, 1, 2, 255]))
handle.daemon.files.bulk_upload([
    {"path": f"{work_dir}/a.txt", "content": b"alpha"},
    {"path": f"{work_dir}/b.txt", "content": b"beta"},
])
downloaded = handle.daemon.files.bulk_download([f"{work_dir}/a.txt", f"{work_dir}/b.txt"])
print(downloaded[f"{work_dir}/a.txt"].decode())
print(handle.daemon.files.list(work_dir))

status = handle.daemon.git.status(work_dir)
print(status.get("branch"))

# Direct daemon calls do not pass through the control plane.
handle.refresh_activity()

Persistent sessions keep shell state between commands:

session = handle.daemon.process.create_session(cwd=work_dir)
session_id = session["id"]

handle.daemon.process.execute_in_session(
    session_id,
    command="export DEMO=direct && mkdir -p demo",
)
async_command = handle.daemon.process.execute_in_session(
    session_id,
    command='printf "$DEMO\\n" && pwd',
    async_=True,
)
for chunk in handle.daemon.process.iter_session_command_logs(session_id, async_command["commandId"]):
    print(chunk, end="", flush=True)
handle.daemon.process.delete_session(session_id)

handle.daemon raises SandboxDaemonUnavailableError when the sandbox does not have daemon direct-connect credentials.

Session (Context Manager)

with client.run_session(template_name="my-devbox", ttl_seconds=600) as session:
    print(session.info.ip)
    # sandbox auto-destroyed on exit

History

Method	Description
list_sandbox_history(page, page_size, status, template_name)	Query sandbox history
get_sandbox_history(id)	Get historical sandbox detail with events

Platform Profiles

Method	Description
list_clusters() / get_cluster(id) / create_cluster(req) / upsert_cluster(id, req) / delete_cluster(id)	Cluster management
list_runtime_profiles() / get_runtime_profile(id) / create_runtime_profile(req) / upsert_runtime_profile(id, req) / delete_runtime_profile(id)	Runtime profile management
list_build_profiles() / get_build_profile(id) / create_build_profile(req) / upsert_build_profile(id, req) / delete_build_profile(id)	Build profile management
list_registry_profiles() / get_registry_profile(id) / create_registry_profile(req) / upsert_registry_profile(id, req) / delete_registry_profile(id)	Registry profile management
list_secret_profiles() / get_secret_profile(id) / create_secret_profile(req) / upsert_secret_profile(id, req) / delete_secret_profile(id)	Secret profile management
list_secret_materializations(profile_id) / resync_secret_materialization(profile_id, mat_id) / rotate_secret_profile(profile_id, req)	Secret materialization

Image Build

Method	Description
create_image_build(req) / list_image_builds() / get_image_build(id)	Image build management
list_image_build_logs(build_id) / cancel_image_build(build_id)	Build logs and cancellation
list_image_assets() / create_image_asset(req) / get_image_asset(id) / upsert_image_asset(id, req) / delete_image_asset(id)	Reusable image assets

Error Handling

Error	Cause
ValidationError	Invalid request (400)
AuthenticationError	API key invalid (401)
NotFoundError	Resource not found (404)
ConflictError	State conflict (409)
APIError	Other control-plane errors
SandboxDaemonUnavailableError	.daemon accessed without daemon direct-connect credentials

Authentication

All control-plane requests use X-API-Key header.