kosa8 for Python

Python client for kosa8: microVM sandboxes for AI agents, with checkpoint, rewind, fork and explore. The counterpart of @kosa8/sdk, method for method.

No dependencies. It talks to the local kosa8 daemon over its unix socket, or to a remote kosa8 worker.

pip install kosa8

Quickstart

from kosa8 import Kosa8Client

k = Kosa8Client()
k.create_sandbox(name="demo", egress="deny")   # a microVM with its own kernel, no network
sb = k.sandbox("demo")

sb.pull("alpine:latest")
print(sb.run("alpine:latest", ["sh", "-c", "echo hello"]).stdout)

sb.checkpoint(label="before the risky bit")    # memory, processes and disk
sb.run("alpine:latest", ["sh", "-c", "rm -rf /data"])
sb.rewind()                                    # back, processes resumed

forks = sb.fork(4)                             # four live copies
sb.delete()

What's in it

k.sandbox(id) returns a handle with the same methods bound to one sandbox.

Checkpoints share the CLI's store (~/.kosa8/checkpoints.json, under the same flock), so kosa8 rewind sees checkpoints taken here and the other way round.

Toolbox

sbx = client.sandbox("agent-1")
r = sbx.run_code("import math; print(math.pi)")          # CodeResult: stdout, exit_code, artifacts
sbx.shell("cd /workspace && git init demo")               # cd and export persist
sbx.write_file("demo/app.py", "print('hi')")
print(sbx.read_text("demo/app.py"))
sbx.shell("cd demo && python3 -m http.server 8000 >/dev/null 2>&1 &")
print(sbx.preview(8000)["url"])                           # signed, expiring link
client.create_sandbox(name="scratch", ephemeral=True, auto_stop_min=15)

Agent frameworks

from kosa8.integrations.openai_agents import sandbox_tools   # pip install 'kosa8[openai-agents]'
agent = Agent(name="coder", tools=sandbox_tools(client, "agent-1"))

from kosa8.integrations.langchain import sandbox_tools       # pip install 'kosa8[langchain]'
agent = create_react_agent(model, sandbox_tools(client, "agent-1"))

Guides: https://docs.kosa8.com/guides/py-frameworks/ and https://docs.kosa8.com/guides/py-rl-eval/.

Connecting

Where requests go, in order: host= or KOSA8_HOST; else the connection the CLI saved (kosa8 wsl setup on Windows, where the engine runs in WSL2, or kosa8 worker join anywhere), with its token; else the local daemon's socket. So on Windows, once kosa8 wsl setup has run, Kosa8Client() reaches the WSL2 engine with no configuration.

kosa8 Cloud

The same sandboxes on kosa8 Cloud instead of the local daemon. Make an API key in the cloud dashboard and set KOSA8_API_KEY.

from kosa8 import Kosa8Cloud, Kosa8PaymentRequiredError

cloud = Kosa8Cloud()  # or Kosa8Cloud("k8c_live_...")

box = cloud.sandboxes.create(template="python", sleep_after_s=600)
r = box.exec(["python3", "-c", "print(6 * 7)"])
print(r.exit_code, r.stdout)    # 0 42
box.sleep()                     # costs only snapshot storage while asleep
box.exec("ls /workspace")       # wakes it, files and all
box.delete()

cloud.functions.deploy("hello", {
    "runtime": "python3.12",
    "code": 'def handler(event):\n    return {"hello": event.get("name", "world")}\n',
})
out = cloud.functions.invoke("hello", {"name": "kosa8"})
print(out["ok"], out["result"])  # True {'hello': 'kosa8'}

A new organization runs nothing until a payment method is on file; until then creating a sandbox raises Kosa8PaymentRequiredError (code payment_required), whose message says what to do. Every error from the cloud carries its stable code.

Errors

Every daemon error is raised as a typed exception carrying the daemon's own message and remediation: Kosa8BadRequestError (400), Kosa8AuthError (401), Kosa8PolicyError (403), Kosa8ApprovalRequiredError (403 needing a signed approval: repeat the call with approval=<id>), Kosa8NotFoundError (404), Kosa8ConflictError (409), Kosa8UnsupportedError (501), Kosa8ConnectionError, Kosa8StreamError (output kept), Kosa8TimeoutError.

Tests

PYTHONPATH=src:tests python3 -m unittest discover -s tests

License

MIT