Guides / Python / Developer guide

Python developer guide

The whole kosa8 package, capability by capability: what each call does, what it sends, what comes back, where it can surprise you, and how to put the pieces together into an AI agent that cannot hurt the machine it runs on.

What this is, in plain words

kosa8 gives each piece of untrusted code its own small computer, a sandbox, running on your machine. The Python SDK is a remote control for those sandboxes: a Python program can create one, run programs inside it, read what they print, take a photo of its complete state (a checkpoint), go back to that photo later (a rewind), make several identical copies (a fork) and try different ideas in the copies at once (an explore).

The typical user is an AI agent. The model proposes a shell command; your program runs it in the sandbox instead of on your laptop; if the command goes wrong, you rewind and nothing outside the sandbox was touched. The network is closed unless you open it, so a confused or hijacked agent cannot send your data anywhere.

The SDK does not run anything itself. It sends requests to the kosa8 daemon (kosa8d), the background service that owns the virtual machines, over a local file called a unix socket. Every call on this page is one or a few such requests.

The parts, and the words for them

Connecting

In plain words: the client has to know where the daemon is. On your own machine it finds it by itself. To drive a kosa8 worker on another machine, you give it an address and a password (token).

from kosa8 import Kosa8Client

k = Kosa8Client()                       # local daemon, ~/.kosa8/run/kosa8.sock
print(k.transport.target, k.version())
print(k.health())

unix:///Users/dhruvit/.kosa8/run/kosa8.sock 0.9.36
{'commit': 'f6124c0', 'go': 'go1.26.9', 'phase': '0-foundations', 'status': 'ok', 'uptime_seconds': 4282, 'version': '0.9.36'}

Constructing a client sends nothing; it only resolves the endpoint and can raise Kosa8Error right away for an address it refuses or an unknown actor. All arguments are keyword-only:

Two rules about the token are easy to miss. KOSA8_TOKEN is only picked up when the host also came from KOSA8_HOST; pass host= in code and you must pass token= too. And plain http to another machine is refused before anything is sent, because the token would cross the network in clear. You can check what the client will do without a daemon, with resolve_endpoint:

from kosa8 import resolve_endpoint

print(resolve_endpoint(env={"KOSA8_HOST": "https://worker.example.com/k8/", "KOSA8_TOKEN": "s3cret"}))
print(resolve_endpoint(host="https://worker.example.com", env={"KOSA8_TOKEN": "s3cret"}))
print(resolve_endpoint(env={"KOSA8_HOST": "localhost:7070"}))

Endpoint(kind='tcp', socket_path='', scheme='https', host='worker.example.com', port=None, base_path='/k8', token='s3cret')
Endpoint(kind='tcp', socket_path='', scheme='https', host='worker.example.com', port=None, base_path='', token=None)
Endpoint(kind='tcp', socket_path='', scheme='http', host='localhost', port=7070, base_path='', token=None)

The second line has token=None: the host was explicit, so KOSA8_TOKEN was not used. KOSA8_HOST=unix:///Users/you/.kosa8/run/kosa8.sock also works and points at a socket. HTTPS uses ssl.create_default_context(), so the system's certificate store and normal hostname checks apply; there is no option to turn verification off. (Remote workers were not tested for this page; the rules above are read from kosa8/transport.py.)

How requests are made

The transport is the standard library's http.client. It opens a new connection for every request and closes it after, so there is no pool, nothing to close on the client, and no state shared between calls apart from the checkpoint store. That makes one client safe to share between threads (below). Bodies are JSON; streamed answers (run, exec, attach) are NDJSON, one frame per line, read as they arrive.

Sandboxes

In plain words: create a sandbox, give it a name, decide how much of the network it may use, and delete it when you are done. A sandbox keeps running, and using memory, until you stop or delete it.

c = k.create_sandbox(name="doc-py-g1", egress_allow=["example.com"], cpus=2, mem_mb=1024,
                     labels={"team": "docs"})
print(sorted(c), c["sandbox"])

['boot_ms', 'sandbox'] {'id': 'doc-py-g1', 'name': 'doc-py-g1', 'created': '2026-11-03T01:39:06.472971-05:00', 'state': 'running', 'cpus': 2, 'mem_mb': 1024, 'disk_gb': 32, 'egress': 'allow:example.com', 'ip': '192.168.127.2', 'labels': {'team': 'docs'}}

A value that is None, empty or 0 is left out of the request, so cpus=0 means "the default", not zero CPUs.

Egress, in practice

The allow-list above lets the sandbox reach example.com and nothing else. A name that is not on the list does not even resolve:

r = sb.run("alpine:latest", ["sh", "-c",
    "wget -q -T 5 -O- http://example.com | head -c 60; echo; wget -q -T 3 -O- http://kosa8.com || echo blocked"])
print(repr(r.stdout), repr(r.stderr))
print(sb.egress())

'<!doctype html><html lang=en><head><meta charset=utf-8><link\nblocked\n' "wget: bad address 'kosa8.com'\n"
{'policy': 'allow:example.com', 'stats': {'passed': 8, 'denied': 8, 'dropped': 7, 'dns_answered': 6, 'dns_refused': 8}}

egress() (sandbox_egress) reports the rule the host is enforcing and its counters, which is how you prove to someone else that the sandbox was sealed. The daemon checks the rule when you create the sandbox; a wildcard on a public suffix is refused with Kosa8BadRequestError (egress allow "*.com": com is a public suffix…). The full rule language is on Network rules.

Finding, stopping, starting, deleting

print(k.stop_sandbox("doc-py-g2-copy"))
try:
    k.exec("doc-py-g2-copy", srv, ["true"])
except Kosa8Error as e:
    print(type(e).__name__, e.status, e)
started = k.start_sandbox("doc-py-g2-copy")
print(started["boot_ms"], started["sandbox"]["state"], k.sandbox_containers("doc-py-g2-copy"))

{'ok': True, 'sandbox': {'id': 'doc-py-g2-copy', 'name': 'doc-py-g2-copy', 'created': '2026-11-03T01:40:49.86258-05:00', 'state': 'stopped', 'parent': 'doc-py-g2', 'cpus': 2, 'mem_mb': 1024, 'disk_gb': 32, 'egress': 'deny'}, 'warning': ''}
Kosa8ConflictError 409 sandbox is not running: doc-py-g2-copy (start it with `kosa8 sandbox up doc-py-g2-copy`)
830 running []

The Sandbox handle

k.sandbox("name") returns a Sandbox: the same methods with the first argument filled in, and other keyword arguments passed through. It holds only the client and the string, sends nothing when made, and does not check that the sandbox exists. It covers info, start, stop, delete, pull, run, start_container, exec, containers, images, egress, snapshot, fork, explore, checkpoint, checkpoints, rewind, and from 0.1.1 exec_frames, start_interactive, write_stdin, attach. For the snapshot calls (list_snapshots, restore_snapshot, delete_snapshot), use the client.

Running code

In plain words: you choose a packaged program (an image), and the sandbox runs it. A one-off command starts fresh every time and forgets its files. A long-lived container keeps its files, so you can send it one command after another, like a terminal.

Images

Each sandbox has its own container engine. sb.pull(ref) (pull_into_sandbox) fetches the image on the host and copies it in, so it works even with egress="deny". sb.images() lists what is there. k.pull(ref), with no sandbox, pulls into the host's shared engine instead; it does not make an image available to any sandbox.

pull: {'image': 'docker.io/library/alpine:latest', 'ok': True}
images: ['docker.io/library/alpine:latest']

A registry the org policy does not allow is refused with Kosa8PolicyError.

run: one command, fresh container

chunks = []
r = sb.run("alpine:latest", ["sh", "-c", "echo $GREETING; echo oops >&2"],
           env={"GREETING": "hi"},
           on_stdout=lambda b: chunks.append(("out", b)),
           on_stderr=lambda b: chunks.append(("err", b)),
           on_status=lambda b: chunks.append(("status", b)))
print(r.exit_code, repr(r.stdout), repr(r.stderr), chunks)

0 'hi\n' 'oops\n' [('err', b'oops\n'), ('out', b'hi\n')]

• args is an argument list, passed as is. There is no shell unless you run one, as here with sh -c. Quoting is then the shell's business, so never paste untrusted text into a sh -c string you also use for something else.
• env is a dict ({"K": "V"}) or a list of "K=V" strings. Only run and start_interactive take it; start_container and exec do not.
• Callbacks get bytes, in the calling thread, as each frame arrives. A chunk can be part of a line or several lines; the order between stdout and stderr is the order the daemon sent them. on_status gets the daemon's progress frames, which are not collected into the result (none were sent for this run).
• The image must already be in the sandbox. run does not pull.

ExecResult has exit_code, stdout_bytes, stderr_bytes, and the properties stdout and stderr (UTF-8, invalid bytes replaced) and ok (exit_code == 0). A non-zero exit is a result, never an exception. Output is kept in memory in full; for a command that prints gigabytes, use the callbacks or exec_frames and do not keep the result.

start_container and exec: a long-lived workspace

long = sb.start_container("alpine:latest", ["sleep", "3600"])   # returns the container id
r = sb.exec(long, ["sh", "-c", "echo v1 > /note; cat /note"])
print(sb.containers())

The container's process must keep running (sleep 3600, or a while loop); when it exits, the container's status becomes exited, so there is nothing to exec into. containers() lists {id, image, status, ip, exit?}, with status running or exited. An exec into a container id that does not exist is not a 404: the stream ends with an error frame, raised as Kosa8StreamError (no such container: nope).

exec_frames: the raw stream

for f in k.exec_frames(sb.id, long, ["sh", "-c", "for i in 1 2 3; do echo tick $i; sleep 0.3; done; exit 7"]):
    print(f)

{'stream': 'stdout', 'data': b'tick 1\n'}
{'stream': 'stdout', 'data': b'tick 2\n'}
{'stream': 'stdout', 'data': b'tick 3\n'}
{'done': True, 'ok': True, 'exit': 7}

It is a generator: nothing is sent until you iterate. data is already decoded from base64. The last frame has done: True and either exit or error. exec_frames does not raise for an error frame, it yields it; check f.get("error") yourself. Note ok: True next to exit: 7: ok means the stream worked, not that the command succeeded. Breaking out of the loop closes the connection; the SDK sends nothing to stop the command in the sandbox.

Interactive: start_interactive, write_stdin, attach

For a program that reads input, such as a REPL, a filter, or a tool that takes a prompt on stdin:

cid = sb.start_interactive("alpine:latest", ["sh", "-c", "tr a-z A-Z"])
print(sb.write_stdin(cid, [b"hello ", "from python\n"]))   # an iterable is sent chunked
print(sb.attach(cid, timeout=10))

{'ok': True}
ExecResult(exit_code=0, stdout_bytes=b'HELLO FROM PYTHON\n', stderr_bytes=b'')

data can be bytes, str (UTF-8 encoded) or any iterable of them, such as a generator reading a file; an iterable is sent with chunked transfer encoding, so it is never held in memory whole. tty=True asks for a terminal. attach follows the container's output until it exits and returns an ExecResult.

(On 0.1.0 these three are called on the client with sb.id first, as in k.write_stdin(sb.id, cid, data); that form still works.)

One write_stdin call is the whole of stdin. The daemon treats one request body as the program's entire input and closes stdin when the body ends. So the SDK allows one call per container on a client, and raises Kosa8Error on a second, before sending anything:

cid = sb.start_interactive("alpine:latest", ["cat"])
print(sb.write_stdin(cid, b"one\n"))
try:
    sb.write_stdin(cid, b"two\n")
except Kosa8Error as e:
    print(type(e).__name__, e)
print(sb.attach(cid, timeout=10))

{'ok': True}
Kosa8Error stdin of container k1-18b52980 in doc-py-v11 was already sent and is closed: the daemon treats one write_stdin body as the whole of stdin. Pass an iterable to stream input
ExecResult(exit_code=0, stdout_bytes=b'one\n', stderr_bytes=b'')

To feed input over time, pass one generator that yields as you go. The check is per client object, keyed by the sandbox and container strings: a second Kosa8Client, or another process, is not stopped, and its input is simply never read. (0.1.0 had no check: the second call answered {'ok': True} and its bytes were lost.)

Timeouts

In plain words: a timeout says how long to wait for an answer before giving up. It does not stop the program in the sandbox.

timeout (client-wide, or per call on run, exec, exec_frames, attach) is a socket timeout in seconds: it applies to connecting and to each read separately. A command that keeps printing within the limit can run far longer in total; a silent one fails after the limit. The default, None, waits forever.

try:
    sb.exec(cid, ["sh", "-c", "echo started; sleep 6; touch /done"], timeout=2)
except Kosa8TimeoutError as e:
    print("timed out:", e)
print("right after:", sb.exec(cid, ["sh", "-c", "ls /done 2>&1"]).stdout.strip())
time.sleep(6)
print("6s later:", sb.exec(cid, ["sh", "-c", "ls /done 2>&1"]).stdout.strip())
r = sb.exec(cid, ["sh", "-c", "for i in 1 2 3 4 5; do echo $i; sleep 1; done"], timeout=2,
            on_stdout=lambda b: print("  got", b))
print("streamed run finished:", r.exit_code)

timed out: POST /kosa8/v1/sandboxes/doc-py-g4/exec timed out mid-stream
right after: ls: /done: No such file or directory
6s later: /done
  got b'1\n'
  …
  got b'5\n'
streamed run finished: 0

The command kept running after the timeout and created /done; the five-second loop finished under a two-second timeout because it printed every second. Two consequences:

• To bound a command's total time, bound it inside the sandbox: ["timeout", "30", "sh", "-c", cmd] (BusyBox and coreutils both have it). The exit code is then 124 or 143, a normal result.
• The client-wide timeout also applies to create_sandbox, snapshot, fork and restore_snapshot, which take no per-call timeout and can be slow (a boot took 14 to 21 seconds on this busy machine; a restore 17). Leave the client timeout at None or generous, and set short ones per call. A timeout on create_sandbox sends no cancel, so the sandbox may still appear; check get_sandbox(name) before retrying.

Snapshots

In plain words: a snapshot is a saved copy of a running sandbox, with everything in memory, so programs carry on from where they were when you bring it back. You can bring it back under a new name as many times as you like.

s = sb.snapshot(id="doc-py-g2-manual")                      # id optional; the daemon picks one
r = k.restore_snapshot("doc-py-g2-manual", name="doc-py-g2-copy")
print(r["sandbox"]["name"], r["sandbox"]["state"], r["restore_ms"], r["sandbox"]["parent"])
print(k.exec("doc-py-g2-copy", srv, ["cat", "/count"]).stdout.strip())   # same container id
print(k.delete_snapshot("doc-py-g2-manual"))

doc-py-g2-copy running 17360 doc-py-g2
18
{'encrypted': False, 'ok': True, 'removed': 'doc-py-g2-manual'}

The copy above is a separate sandbox with parent set, and the container from the original (srv, a counter writing /count every second) was running in it under the same id. A snapshot keeps the egress rule of the sandbox it came from.

Fork

f = sb.fork(2, prefix="doc-py-g2-f")
print({k_: v for k_, v in f.items() if k_ != "forks"})
print([(x["id"], x["state"], x.get("parent")) for x in f["forks"]])

{'count': 2, 'fork_ms': 2985, 'snapshot': 'doc-py-g2-fork-1793688049891573000'}
[('doc-py-g2-f1', 'running', 'doc-py-g2'), ('doc-py-g2-f2', 'running', 'doc-py-g2')]

fork(n, prefix=None, label=None) snapshots the sandbox once and restores it n times, named <prefix>1 … <prefix>n. The parent keeps running. The forks are ordinary sandboxes: you delete each one yourself. The intermediate snapshot was not left in kosa8 snapshot ls after the forks were deleted.

Explore: best of N

In plain words: you have several ideas and do not know which works. Explore makes one copy of the sandbox per idea, tries them all at the same time, keeps the copy where the idea worked fastest and throws the others away. The original is not touched.

r = sb.explore(["sleep 2; echo slow", "echo fast", "exit 1"], prefix=name + "-x")
print("winner:", r.winner, "removed:", r.removed)
for b in r.branches:
    print(" ", b)

winner: sbx1-x2 removed: 2
  ExploreBranch(name='sbx1-x1', variant='sleep 2; echo slow', exit=0, output='slow', duration_ms=2959, error=None)
  ExploreBranch(name='sbx1-x2', variant='echo fast', exit=0, output='fast', duration_ms=993, error=None)
  ExploreBranch(name='sbx1-x3', variant='exit 1', exit=1, output='', duration_ms=968, error=None)

What it does, step by step (it is client-side, in Kosa8Client.explore):

1. Raises Kosa8Error if there are fewer than two variants.
2. If pull=True (default), pulls image (default "alpine:latest") into the parent.
3. Forks the parent once per variant, with prefix (default <sandbox>-explore-<5 digits of the time>).
4. Runs every variant as ["sh", "-c", variant] in its own branch, all at once on a thread pool. Each run has a socket timeout of timeout (default 120 s). An error that looks like a connection problem is retried up to four attempts (0.5, 1, 1.5 s apart; from 0.1.1 there is no extra wait after the last one); a timeout is not.
5. The winner is the fastest branch with exit 0 and no error; "" if none. keep="branch-name" overrides the choice.
6. Unless keep_all=True, deletes every other branch. The winner is left running for you to use, and to delete.

The whole call took 54 seconds on this machine, almost all of it the pull and the fork; the branches themselves took one to three seconds. duration_ms includes retries.

Checkpoints and rewind

In plain words: a checkpoint is an undo point. Take one before anything risky. If it goes wrong, rewind, and the sandbox is exactly as it was, with its programs still running. kosa8 keeps the last ten per sandbox.

cp1 = sb.checkpoint(label="first")
print(cp1["checkpoint"], cp1["evicted"], sorted(cp1["snapshot"]))
cp2 = sb.checkpoint(label="second")
print([(c["id"], c["label"]) for c in sb.checkpoints()])
print(subprocess.run(["kosa8", "checkpoint", "ls", "doc-py-g2"], capture_output=True, text=True).stdout)
rw = sb.rewind(to="FIRST")                  # an id, or a label in any case
print(rw["checkpoint"]["id"], rw["dropped"], rw["restore_ms"], rw["sandbox"]["state"])
print([c["id"] for c in sb.checkpoints()])

{'id': 'doc-py-g2-ckpt-1', 'sandbox': 'doc-py-g2', 'label': 'first', 'created': '2026-11-03T06:40:07.019767Z'} [] ['snapshot', 'snapshot_ms']
[('doc-py-g2-ckpt-1', 'first'), ('doc-py-g2-ckpt-2', 'second')]
CHECKPOINT          WHEN       TAKEN   WHY
 doc-py-g2-ckpt-1   just now   you     first
*doc-py-g2-ckpt-2   just now   you     second

* is where `kosa8 rewind doc-py-g2` goes

doc-py-g2-ckpt-1 ['doc-py-g2-ckpt-2'] 574 running
['doc-py-g2-ckpt-1']

A counter process that wrote /count every second read 3 at the checkpoint, 3 right after the rewind, and 4 a second later: memory and the process came back and carried on.

How it works

The daemon has no checkpoint endpoint. A checkpoint is a snapshot with the id <sandbox>-ckpt-<n>, plus an entry in a list the client keeps. The SDK does this the same way the Go tools do (pkg/checkpoint), on the same file:

• checkpoint(label=""): works out the next number (one more than the larger of the list's length and the highest number used), takes the snapshot, then, if the list now has more than MAX_CHECKPOINTS_PER_SANDBOX (10), deletes the oldest snapshots, then records the new entry. Eviction happens only after the new snapshot exists. Returns {checkpoint, evicted, snapshot}.
• rewind(to=None): finds the target (the latest, or the newest entry whose id equals to or whose label equals it ignoring case) before touching anything, so a bad to raises Kosa8CheckpointError and changes nothing. Then it deletes the sandbox, restores the snapshot under the same name (if that restore fails, Kosa8CheckpointError says the sandbox was deleted, the checkpoint is intact, and which restore_snapshot call brings it back; see a rewind that failed halfway), cuts the list after the target and deletes the snapshots of the later checkpoints (dropped). The state since the target is gone; take a checkpoint first if you might want it.
• list_checkpoints(sandbox) / sb.checkpoints(): the list, oldest first, read from the file; no request to the daemon.
• delete_sandbox removes the sandbox's list and its snapshots, unless keep_checkpoints=True. With the list kept, a later sandbox with the same name inherits it.

The list is keyed by the string you pass as sandbox. Ids and names are the same in kosa8 0.9.36 (an unnamed sandbox got id and name sbx1), so either works, but use one consistently.

The shared file: ~/.kosa8/checkpoints.json

The default store, FileCheckpointStore(), is the file the kosa8 CLI, the MCP server and the Mac app use. That is why kosa8 checkpoint ls above showed the checkpoints taken from Python, and why kosa8 rewind doc-py-g2 would have gone to the newest one. The format is {"sandbox-name": [{"id", "sandbox", "label", "created"}, …]}.

• Each change takes an in-process lock, then an exclusive fcntl.flock on ~/.kosa8/checkpoints.json.lock (the same lock the Go tools take), reads the file, changes it, writes checkpoints.json.tmp with mode 0600 and renames it into place. Readers never see a half-written file. The directory is created with mode 0700.
• From 0.1.1 the flock is retried every 50 ms for at most lock_timeout seconds (default 10), then Kosa8CheckpointError is raised instead of waiting forever. Tested by holding the lock from another process:

  Kosa8CheckpointError timed out after 1s waiting for …/ck/checkpoints.json.lock; another kosa8 process is holding it

  with FileCheckpointStore(path, lock_timeout=1). Nothing was written.
• Reads (list_checkpoints, finding the next id) do not take the lock.
• A missing or corrupt file reads as empty, so a damaged file never makes a sandbox unusable; the next write replaces it.
• On Windows there is no fcntl, and only the in-process lock is taken. (kosa8 sandboxes do not run on Windows; this matters only for a client talking to a remote worker.)

Other stores:

from kosa8 import Kosa8Client, FileCheckpointStore, MemoryCheckpointStore

isolated = Kosa8Client(checkpoint_store=MemoryCheckpointStore())                # this process only
per_job  = Kosa8Client(checkpoint_store=FileCheckpointStore("/srv/job-42/checkpoints.json"))
patient  = Kosa8Client(checkpoint_store=FileCheckpointStore(lock_timeout=60))   # 0.1.1: wait longer for the lock

MemoryCheckpointStore forgets everything when the process ends, and the CLI cannot see its checkpoints; the snapshots themselves still exist on the daemon, so delete the sandbox through the same client to clean them up. Any object with load() -> dict and update(fn) -> result (call fn on the data, save it, return what fn returned) can be a store, for example one backed by your database.

Two checkpoints of one sandbox at once

The next checkpoint number is read before the snapshot is taken, outside the lock, so two checkpoints of the same sandbox at the same moment (two threads, two processes, or your program and the CLI) can pick the same id. From 0.1.1, the one whose snapshot is refused with snapshot already exists steps to the next number and tries again, up to ten attempts; a 409 for any other reason is raised as before. Two threads, on 0.1.1:

def cp(i):
    return sb.checkpoint(label=f"t{i}")["checkpoint"]["id"]

with ThreadPoolExecutor(2) as p:
    print(list(p.map(cp, range(2))))
print([(c["id"], c["label"]) for c in sb.checkpoints()])

['doc-py-v11-ckpt-1', 'doc-py-v11-ckpt-2']
[('doc-py-v11-ckpt-1', 't0'), ('doc-py-v11-ckpt-2', 't1')]

On 0.1.0 the same test gave one checkpoint and a Kosa8ConflictError: snapshot already exists: doc-py-race-ckpt-1; on that version, serialize checkpoints per sandbox. Which checkpoint gets the lower number is a matter of timing, and the list is ordered by creation time, so with parallel checkpoints "latest" means "finished last".

Threads and asyncio

The SDK is synchronous; there is no async client. Because every request uses its own connection, one Kosa8Client can be shared by many threads:

def job(i):
    return sb.run("alpine:latest", ["sh", "-c", f"sleep 1; echo job {i}"]).stdout.strip()

with ThreadPoolExecutor(4) as pool:
    print(list(pool.map(job, range(4))))

['job 0', 'job 1', 'job 2', 'job 3']    # 1.7 s for four 1-second jobs

From asyncio, run each call in a worker thread with asyncio.to_thread (Python 3.9+), so the event loop is never blocked:

async def main():
    with temp_sandbox(k, prefix="doc-py-async", egress="deny") as sb:     # recipe below
        await asyncio.to_thread(sb.pull, "alpine:latest")
        results = await asyncio.gather(*[
            asyncio.to_thread(sb.run, "alpine:latest", ["sh", "-c", f"echo task {i}"]) for i in range(3)
        ])
        print("async:", [r.stdout.strip() for r in results])

asyncio.run(main())

async: ['task 0', 'task 1', 'task 2']

Output callbacks then run in the worker thread, not the event loop. To pass chunks to async code, use loop.call_soon_threadsafe(queue.put_nowait, chunk). Cancelling the awaiting task does not stop the thread or the command; bound the command inside the sandbox (Timeouts). What is not safe in parallel: a rewind of a sandbox while other threads are running commands in it (they fail while it is deleted and restored).

Recipes

A sandbox that always cleans up

import contextlib, uuid
from kosa8 import Kosa8NotFoundError

@contextlib.contextmanager
def temp_sandbox(client, prefix="agent", **create):
    """A sandbox that is deleted when the block ends, whatever happens inside."""
    name = f"{prefix}-{uuid.uuid4().hex[:6]}"
    client.create_sandbox(name=name, **create)
    sb = client.sandbox(name)
    try:
        yield sb
    finally:
        try:
            sb.delete()
        except Kosa8NotFoundError:
            pass  # already gone

A random suffix means two runs never fight over a name. If create_sandbox itself raises, there is nothing to clean up. A process killed with SIGKILL skips finally; sweep leftovers at start-up with [s for s in k.list_sandboxes() if s["name"].startswith(prefix)], or label your sandboxes and filter on the label.

An AI agent loop with undo

The pattern: one long-lived container per task so files persist between the model's commands; a checkpoint before each step; a rewind when your own check says the step was bad. The stand-in "model" below proposes three commands, one of them destructive.

k = Kosa8Client(actor="agent")

class Workspace:
    """One long-lived container that an agent's commands run in, so files persist."""

    def __init__(self, sb, image="alpine:latest"):
        self.sb = sb
        sb.pull(image)
        self.cid = sb.start_container(image, ["sh", "-c", "mkdir -p /work; while :; do sleep 3600; done"])

    def sh(self, cmd, timeout=60):
        r = self.sb.exec(self.cid, ["sh", "-c", f"cd /work && {cmd}"], timeout=timeout)
        return r.exit_code, (r.stdout + r.stderr)[-4000:]   # cap what goes back to the model

proposals = [
    "echo 'draft 1' > notes.txt && cat notes.txt",
    "rm -rf /work/*",                  # a destructive step
    "ls /work",
]

with temp_sandbox(k, prefix="doc-py-agent", egress="deny") as sb:
    ws = Workspace(sb)
    for turn, cmd in enumerate(proposals, 1):
        ws.sb.checkpoint(label=f"turn {turn}")          # a restore point before every step
        code, out = ws.sh(cmd)
        print(f"turn {turn}: exit {code}: {out.strip()!r}")
        if "rm -rf" in cmd:                           # your own check: tests, a reviewer, a policy
            rw = ws.sb.rewind(to=f"turn {turn}")
            print(f"  undone: back to {rw['checkpoint']['id']}, dropped {rw['dropped']}")
            print("  /work now:", ws.sh("ls /work")[1].strip())
    print("checkpoints:", [(c["id"], c["label"]) for c in sb.checkpoints()])

turn 1: exit 0: 'draft 1'
turn 2: exit 0: ''
  undone: back to doc-py-agent-2b097e-ckpt-2, dropped []
  /work now: notes.txt
turn 3: exit 0: 'notes.txt'
checkpoints: [('doc-py-agent-2b097e-ckpt-1', 'turn 1'), ('doc-py-agent-2b097e-ckpt-2', 'turn 2'), ('doc-py-agent-2b097e-ckpt-3', 'turn 3')]

Notes for real agents:

• The container id survives the rewind, so Workspace keeps working after it.
• Give the model the exit code and the tail of the output as the tool result, not an exception message. Reserve exceptions for "the sandbox is broken": catch Kosa8StreamError and Kosa8TimeoutError per step and report them to the model as a failed step; let Kosa8ConnectionError stop the loop.
• A checkpoint took under a second in these tests (746 ms in the quickstart). Every turn is affordable for a coding agent; for hundreds of tiny steps, checkpoint before the risky ones only. Past ten, the oldest are evicted automatically.
• Keep egress="deny" unless the task needs the network, then use an allow-list. API keys the agent needs belong in the broker, not in env; see Securing an agent.
• actor="agent" marks these calls as an agent's in the audit log; an agent_token names which agent.

Stream output to a user as it happens

import sys
r = sb.exec(cid, ["sh", "-c", "make test"],
            on_stdout=lambda b: sys.stdout.buffer.write(b) or sys.stdout.flush(),
            on_stderr=lambda b: sys.stderr.buffer.write(b) or sys.stderr.flush())
print("\nexit", r.exit_code)

(Not run as shown; the callback mechanism is the one tested above.)

Try several fixes, keep the one that passes

r = sb.explore([
    "cd /work && patch -p1 < /fixes/a.diff && make test",
    "cd /work && patch -p1 < /fixes/b.diff && make test",
], image="my-build-image:latest", timeout=600)
if r.winner:
    print("keep", r.winner)          # a running sandbox with the fix applied and tested
else:
    for b in r.branches:
        print(b.name, b.exit, b.error or b.output[-500:])

(A pattern, not run as shown: the paths and image are placeholders. explore runs each variant in a fresh container of image, so the files it needs must be in that image or on the sandbox's disk.)

Errors, briefly

Every exception the SDK raises is a Kosa8Error. An error from the daemon keeps the daemon's own words in daemon_message and its advice in remediation, and its class says what kind of problem it is (Kosa8NotFoundError, Kosa8ConflictError, Kosa8PolicyError, …). Every class, with real messages and what to do: Python errors and troubleshooting.