--- title: "Python SDK" description: "Use the katakate client to manage sandboxes" --- Install the SDK: ```bash pip install katakate ``` ## Synchronous client ```python from katakate import Client k7 = Client(endpoint="https://", api_key="") # Create a sandbox sb = k7.create({ "name": "my-sandbox", "image": "alpine:latest", # optional: "namespace": "default", # optional: "env_file": ".env", # optional: "egress_whitelist": ["10.0.0.5/32"], # private egress proxy # 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() ``` ### Client configuration - `endpoint`: Base URL of your API, e.g. `https://`. - `api_key`: Your API key. The SDK sends it via `X-API-Key` automatically. Get your endpoint and API key using the CLI: `k7 api-status`, `k7 get-api-endpoint`, `k7 generate-api-key `. See the CLI guide: `/guides/cli`. ### Create with non-root, capabilities, egress controls, limits By default, all Linux capabilities are dropped. You can add back minimal ones if needed. ```python 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 blocked) # - [CIDRs] allows only those CIDRs (DNS still blocked) "egress_whitelist": [ "10.0.0.5/32", # private egress proxy "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" }) ``` `env_file` points to a file on the API host filesystem (server-side), not the client machine. If you need environment variables and you’re calling a remote API, pass values directly for now. ### Wait until sandbox is Ready ```python 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") ``` ### Execute commands and handle errors ```python 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", "")) ``` ### List, filter by namespace ```python print(k7.list()) # all namespaces print(k7.list(namespace="dev")) # only dev ``` ### Delete and delete all ```python k7.delete("secure-sb") k7.delete_all(namespace="default") ``` ## Async client ```python import asyncio from katakate import AsyncClient async def main(): k7 = AsyncClient(endpoint="https://", api_key="") sandboxes = await k7.list() print(sandboxes) await k7.aclose() asyncio.run(main()) ``` ### Async examples Create, wait, exec, delete: ```python 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()) ``` ## Errors and responses - Successful responses are wrapped as `{ "data": ... }` by the API; the SDK unwraps them. - Errors are returned as `{ "error": { "code": string, "message": string } }` with appropriate HTTP status codes. ## Tips - Provide a `namespace` explicitly if you use non-default namespaces. - Keep API keys secret; rotate via `k7 revoke-api-key` and `k7 generate-api-key`.