kanyun-sandbox 0.3.7

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.7

• 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"},
                },
            }],
        },
    },
})
print(f"Template created: {template.name}")

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",
    agent_port=8080,            # Agent service port (default: 8080)
    agent_client_factory=None,  # Optional, see "Agent Access" section
    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

Agent Access (Optional)

By default, the SDK only communicates with the control plane. If your sandbox image runs a compatible agent service (e.g., the AIO sandbox image), you can opt into in-sandbox command execution by providing an agent_client_factory.

pip install agent-sandbox

from agent_sandbox import Sandbox
from kanyun_sandbox import Client, CreateSandboxRequest

client = Client(
    control_plane_url="http://127.0.0.1:8080",
    api_key="kanyun_xxx",
    agent_client_factory=lambda base_url: Sandbox(base_url=base_url),
)

handle = client.create_sandbox(CreateSandboxRequest(
    template_name="aio-devbox",  # Must use an AIO image
    ttl_seconds=3600,
))

# Access the agent client (lazy initialization)
agent: Sandbox = handle.agent

# Execute shell commands
result = agent.shell.exec_command(command="whoami")
print(result.data.output)   # => "root"
print(result.data.exit_code)  # => 0

# Run a multi-line script
result = agent.shell.exec_command(command="""
python3 -c "
import platform
print(f'Python {platform.python_version()}')
"
""")
print(result.data.output)

# File operations
agent.file.write_file(file="/tmp/hello.txt", content="Hello from SDK!")
content = agent.file.read_file(file="/tmp/hello.txt")
print(content.data.content)

# List directory
listing = agent.file.list_path(path="/tmp")
for f in listing.data.files:
    print(f"  {f.name}")

handle.destroy()

Important:

• Agent access requires a specific sandbox image that runs the AIO agent protocol.
• Install the agent-sandbox package separately: pip install agent-sandbox
• The agent_client_factory receives the sandbox's internal http://{ip}:{agent_port} URL and returns a Sandbox instance.
• handle.agent is created lazily — sandbox creation and control-plane operations never require agent support.
• Accessing .agent without a factory raises SandboxAgentUnavailableError.
• Accessing .agent before the sandbox has an IP raises SandboxNotReadyError.

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
SandboxAgentUnavailableError	.agent accessed without agent_client_factory
SandboxNotReadyError	.agent accessed before sandbox has an IP

Authentication

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