Python SDK

pip install katakate

from katakate import Client

k7 = Client(endpoint="https://<your-endpoint>", api_key="<your-key>")

# Create a sandbox
sb = k7.create({
    "name": "my-sandbox",
    "image": "alpine:latest",
    # optional: "namespace": "default",
    # optional: "env_file": ".env",
    # optional: "egress_whitelist": ["1.1.1.1/32", "8.8.8.8/32"],
    # optional: "limits": {"cpu": "1", "memory": "1Gi", "ephemeral-storage": "2Gi"},
    # optional: "before_script": "apk add curl"
})

# Execute a command
result = sb.exec('echo "Hello World"')
print(result["stdout"])  # Also includes stderr and exit_code

# List sandboxes
print(k7.list())

# Delete sandbox
sb.delete()

sb = k7.create({
    "name": "secure-sb",
    "image": "alpine:latest",
    "namespace": "default",

    # Non-root execution
    "pod_non_root": True,          # Pod UID/GID/FSGroup 65532
    "container_non_root": True,    # Container UID 65532, no privilege escalation

    # Capabilities: drop ALL by default, add minimal ones back
    "cap_add": ["CHOWN"],
    "cap_drop": ["NET_RAW"],

    # Network egress control
    #   - Omit key to keep egress open
    #   - [] blocks all egress (DNS to CoreDNS still allowed)
    #   - [CIDRs] allows only those CIDRs (+ DNS)
    "egress_whitelist": [
        "1.1.1.1/32",
        "8.8.8.8/32",
        "203.0.113.0/24"
    ],

    # Resource limits/requests (same values used for both)
    "limits": {"cpu": "500m", "memory": "512Mi", "ephemeral-storage": "2Gi"},

    # Optional setup commands run before Ready (executed with open egress)
    "before_script": "apk add --no-cache curl git"
})

import time

def wait_until_ready(name: str, namespace: str = "default", timeout_seconds: int = 120) -> None:
    deadline = time.time() + timeout_seconds
    while time.time() < deadline:
        for info in k7.list(namespace=namespace):
            if info.get("name") == name and info.get("status") == "Running" and info.get("ready") == "True":
                return
        time.sleep(2)
    raise TimeoutError("Sandbox did not become Ready in time")

wait_until_ready("secure-sb")

res = sb.exec("echo hello && uname -a")
print(res["stdout"])          # command output
print(res["stderr"])          # error stream (if any)
print(res["exit_code"])       # 0 on success

# Example of a failing command
bad = sb.exec("sh -lc 'exit 2'")
if bad["exit_code"] != 0:
    print("Command failed:")
    print("stderr:", bad.get("stderr", ""))

print(k7.list())                 # all namespaces
print(k7.list(namespace="dev")) # only dev

k7.delete("secure-sb")
k7.delete_all(namespace="default")

import asyncio
from katakate import AsyncClient

async def main():
    k7 = AsyncClient(endpoint="https://<your-endpoint>", api_key="<your-key>")
    sandboxes = await k7.list()
    print(sandboxes)
    await k7.aclose()

asyncio.run(main())

import os
import asyncio
from katakate import AsyncClient

K7_ENDPOINT = os.getenv("K7_ENDPOINT")
K7_API_KEY = os.getenv("K7_API_KEY")

async def main():

    try: 
        k7 = AsyncClient(endpoint=K7_ENDPOINT, api_key=K7_API_KEY)

        cfg = {
            "name": "async-sb",
            "image": "alpine:latest",
            "pod_non_root": True,
            "container_non_root": True,
            "cap_add": ["CHOWN"],
            # "before_script": "apk add --no-cache curl" # This is commented out here as it would fail, because 'apk add' needs root access, which we removed with pod_non_root and container_non_root set to True
            "egress_whitelist": [],   # full network lockdown after the before_script
        }

        print("Creating sandbox...")
        await k7.create(cfg)
        print("Sandbox created.")

        # (Optional) Simple readiness wait (poll list). This can be removed, it is just here to illustrate.
        for _ in range(60):
            sbs = await k7.list()
            if any(s.get("name") == "async-sb" and s.get("status") == "Running" and s.get("ready") == "True" for s in sbs):
                break
            await asyncio.sleep(2)

        out = await k7.exec("async-sb", "echo from async")
        print("Output of execution:", out)

    except Exception as e:
        raise e

    # Include a finally block to clean resources even if code fails
    finally:
        print("Deleting sandbox 'async-sb'...")
        try: 
            await k7.delete("async-sb")
            print("Sandbox 'async-sb' deleted.")
        except:
            raise Exception("Failed to delete async-sb, you might need to clean resources manually.")

        print("Closing the client's httpx connection...")
        try:
            await k7.aclose()
            print("Connection closed.)
        except:
            raise Exception("Failed to close the K7 client's httpx connection, you might need to clean resources manually.)

asyncio.run(main())