Python SDK

pip install k7-sdk==0.5.0            # sync client (requests only)pip install "k7-sdk[async]==0.5.0"   # also installs httpx for AsyncClient

from k7_sdk import Clientk7 = Client(endpoint="https://node:31007", api_key="<your-key>", verify_ssl="./k7-ca.crt")sb = k7.create({    "name": "demo",    "image": "alpine:latest",    "before_script": "apk add --no-cache curl",})print(sb.exec("uname -a"))print(k7.list())sb.delete()

sb = k7.create({    "name": "secure-sb",    "image": "alpine:latest",    "namespace": "default",    # Backend selection    "backend": "kata-qemu-longhorn",      # or "kata-firecracker-devmapper" / "k7d" / "k7d-fc"    "root_disk_size": "20Gi",        # kata-qemu-longhorn only    # Docker inside the sandbox (pinned dockerd, overlay2 on a block disk)    "docker": True,    "docker_disk": "40Gi",           # default 20Gi    # Non-root + minimal capabilities    "pod_non_root": True,    "container_non_root": True,    "cap_add": ["CHOWN"],    # FQDN-aware egress (Cilium); falls back to CIDR if all entries are CIDRs    "egress_whitelist": ["api.openai.com", "*.huggingface.co"],    # Ingress is deny-all unless you open ports; scope callers; optionally publish    "ingress_ports": [8000],    "ingress_from": ["sandbox:frontend"],   # or "namespace:<ns>" / "cidr:<cidr>"    # "expose_ports": [8000],               # NodePort; node_ports comes back on list()    # Resource budget    "limits": {"cpu": "1", "memory": "1Gi", "ephemeral-storage": "2Gi"},    # Setup commands run with open egress before lockdown    "before_script": "pip install openai",    # Override entrypoint/cmd    "entrypoint": ["/usr/bin/python3"],    "cmd": ["-c", "import time; time.sleep(3600)"],})

import timedef wait_until_ready(client, name, namespace="default", timeout=120):    deadline = time.time() + timeout    while time.time() < deadline:        for s in client.list(namespace=namespace):            ready = s.get("ready")            if s["name"] == name and s.get("status") == "Running" and ready in (True, "True"):                return        time.sleep(2)    raise TimeoutError(f"{name} did not become Ready")wait_until_ready(k7, "secure-sb")

base = k7.create({    "name": "demo",    "image": "python:3.12-slim",    "backend": "kata-qemu-longhorn",    "before_script": "pip install numpy pandas",})wait_until_ready(k7, "demo")base.pause(snapshot="demo-v1")base.resume()forks = [base.fork(f"exp-{i}") for i in range(8)]for f in forks:    print(f.exec("python -c 'import numpy; print(numpy.__version__)'")["stdout"])    f.delete()

base.snapshot("demo-experiment-1")for s in k7.list_snapshots(sandbox="demo"):    print(s["name"], s["kind"], s["age"])k7.delete_snapshot("demo-experiment-1")k7.gc_snapshots(all_namespaces=True, keep_fork_for="10m", dry_run=True)base.pause(snapshot="demo-v1")base.delete()restored = k7.restore("demo-v1", "demo-restored")wait_until_ready(k7, "demo-restored")  # restore returns once objects exist; boot is asyncprint(restored.exec("ls /")["stdout"])

import asynciofrom k7_sdk import AsyncClientasync def main():    k7 = AsyncClient(endpoint="https://node:31007", api_key="<your-key>", verify_ssl="./k7-ca.crt")    try:        await k7.create({"name": "async-sb", "image": "alpine:latest"})        for _ in range(60):            sbs = await k7.list()            if any(s["name"] == "async-sb" and s["ready"] == "True" for s in sbs):                break            await asyncio.sleep(2)        out = await k7.exec("async-sb", "echo from async")        print(out["stdout"])    finally:        await k7.delete("async-sb")        await k7.aclose()asyncio.run(main())

import asynciofrom k7_sdk import AsyncClientasync def run_one(k7, idx):    name = f"worker-{idx}"    await k7.create({"name": name, "image": "python:3.12-slim"})    try:        for _ in range(60):            if any(                s["name"] == name and s["ready"] == "True"                for s in await k7.list()            ):                break            await asyncio.sleep(2)        result = await k7.exec(name, f"python -c 'print({idx} ** 2)'")        return result["stdout"].strip()    finally:        await k7.delete(name)async def main():    k7 = AsyncClient(endpoint="https://node:31007", api_key="<your-key>", verify_ssl="./k7-ca.crt")    try:        results = await asyncio.gather(*[run_one(k7, i) for i in range(10)])        print(results)    finally:        await k7.aclose()asyncio.run(main())

{ "error": { "code": "BadRequest", "message": "name already exists" } }

import requeststry:    sb = k7.create({"name": "demo", "image": "alpine"})except requests.HTTPError as e:    body = e.response.json()    print(body["error"]["code"], body["error"]["message"])