Guides / Python / API reference

Python API reference

Every name the kosa8 package exports in version 0.1.1 (its __all__), and the public members of each. Signatures are copied from the source of the published package. Keys in returned dicts are the daemon's own JSON names, unchanged.

Package

from kosa8 import (
    Kosa8Client, Sandbox, ExecResult, ExploreBranch, ExploreResult,
    Transport, default_socket_path, resolve_endpoint,
    CheckpointBook, FileCheckpointStore, MemoryCheckpointStore, MAX_CHECKPOINTS_PER_SANDBOX,
    Kosa8Error, Kosa8ConnectionError, Kosa8ApiError, Kosa8BadRequestError, Kosa8AuthError,
    Kosa8PolicyError, Kosa8ApprovalRequiredError, Kosa8NotFoundError, Kosa8ConflictError,
    Kosa8UnsupportedError, Kosa8StreamError, Kosa8CheckpointError, Kosa8TimeoutError,
    Kosa8ProtocolError, error_from_response, __version__,
)

Python 3.9 or later. No dependencies: only the standard library. The package ships py.typed, so type checkers read its annotations. Type aliases used below, from kosa8.client and kosa8.transport:

Env          = Union[Mapping[str, str], Sequence[str], None]       # {"A": "1"} or ["A=1"]
Chunk        = Callable[[bytes], None]                             # output callbacks
EgressPolicy = Union[str, Mapping[str, Sequence[str]], None]       # "deny" | "any" | "fakenet" | "allow:a,b" | {"allow": [...]}
Body         = Union[bytes, str, Iterable[Union[bytes, str]]]      # write_stdin data
Query        = Mapping[str, Union[str, int, bool, None]]

Kosa8Client

Kosa8Client(
    *,
    socket_path: Optional[str] = None,
    host: Optional[str] = None,
    token: Optional[str] = None,
    insecure: bool = False,
    actor: str = "tool",
    agent_token: Optional[str] = None,
    headers: Optional[Mapping[str, str]] = None,
    timeout: Optional[float] = None,
    checkpoint_store: Any = None,
    env: Optional[Mapping[str, str]] = None,
)

All arguments are keyword-only. Constructing the client sends nothing; it resolves the endpoint (and can raise Kosa8Error for a refused address or a bad actor).

Every method that takes approval: Optional[str] = None sends it, stripped, as the X-Kosa8-Approval header. Methods without it in their signature send none. Every method can raise Kosa8ConnectionError (daemon unreachable), Kosa8TimeoutError (a timeout was set and passed), and the HTTP error class matching the daemon's status; only the extra errors are listed per method. Paths below are shortened: … is /kosa8/v1, and every path segment is percent-encoded.

Status and the shared engine

Sandboxes

create_sandbox parameters

A parameter that is None, empty or zero is left out of the request body, and the daemon applies its default.

Extra errors: Kosa8Error (client side) for egress={"allow": []} ("an egress allow-list needs at least one host…") and for egress_allow with another egress.

Running code

• args is an argv list; no shell is involved unless you run one. env is a mapping (sent as ["K=V", …]) or a sequence of "K=V" strings.
• Callbacks receive each chunk as bytes, in the calling thread, as it arrives. on_status (run only) receives the daemon's progress frames; status bytes are not collected into the result.
• timeout is a socket timeout in seconds for connect and for each read, so a stream that keeps producing output within it can run longer in total. When it passes, the call raises Kosa8TimeoutError; the process in the sandbox keeps running.
• A non-zero exit is not an error. Extra errors: Kosa8StreamError when the stream ends with an error frame, or with no final frame.

A frame is a dict as the daemon sends it, with data base64-decoded to bytes: output frames {"stream": "stdout" | "stderr" | "status", "data": b"…"}, then a final {"done": True, "ok": …, "exit": int}, or {"done": True, "error": "…"}. exec_frames yields them unchanged and raises nothing for an error frame; an HTTP error before streaming starts is raised as its typed error.

Snapshots, fork, explore

explore parameters

Branches run on a ThreadPoolExecutor with one worker per variant. A branch whose error message looks transient (connection refused, EOF, reset, broken pipe, "no such file or directory", "stream ended", unreachable) is retried up to four attempts, waiting 0.5, 1 and 1.5 seconds between them (0.1.0 also waited 2 seconds after the last); a timeout is never retried. The winner is the fastest branch with no error and exit 0; "" when none succeeded. A failed delete of a loser is not raised; removed counts the ones that went. Extra errors: Kosa8Error when the fork returned fewer copies than variants.

Checkpoints

Client-side, over snapshots. The daemon has no checkpoint endpoint.

to is a checkpoint id or a label (case-insensitive; the newest match wins); without it, the latest. Extra errors: Kosa8CheckpointError for no checkpoints, no match, or an invalid sandbox name (empty, ., .., containing / or \, or starting with a dot). If the restore fails after the delete, rewind raises Kosa8CheckpointError (0.1.1; a plain Kosa8Error saying "is now stopped" in 0.1.0) with the message restore failed and NAME no longer exists (it was deleted to free its name for the restore): … — the checkpoint is intact; recover with restore_snapshot('ID', name='NAME'), chained to the daemon's error. checkpoint (0.1.1): when the snapshot is refused with a 409 whose message contains "already exists" (a concurrent checkpoint took the id), it tries the next number, up to ten attempts in all; any other 409, or the tenth, is raised. Failures deleting evicted or dropped snapshots are not raised.

Helper

Sandbox

Sandbox(client: Kosa8Client, sandbox_id: str)
client: Kosa8Client
id: str
repr(sb)  # "Sandbox('doc-py-guide')"

Each method calls the client method with the sandbox filled in, passing other keyword arguments through (**kw), so the options and errors are the client's.

Not on the handle: restore_snapshot, delete_snapshot, list_snapshots, and the host-engine pull. Call them on the client. (In 0.1.0, exec_frames, start_interactive, write_stdin and attach were missing from the handle too.)

Dataclasses

ExecResult

@dataclass
class ExecResult:
    exit_code: int
    stdout_bytes: bytes = b""
    stderr_bytes: bytes = b""

    stdout: str     # property: stdout_bytes decoded as UTF-8, errors replaced
    stderr: str     # property: the same for stderr_bytes
    ok: bool        # property: exit_code == 0

When the final frame carries no integer exit, exit_code is 0.

ExploreBranch

@dataclass
class ExploreBranch:
    name: str                    # the branch sandbox's id
    variant: str                 # the shell command it ran
    exit: int = 0
    output: str = ""             # stdout + stderr, trailing newlines stripped
    duration_ms: int = 0         # including retries
    error: Optional[str] = None  # the last error, if the run never finished

ExploreResult

@dataclass
class ExploreResult:
    parent: str
    branches: List[ExploreBranch] = field(default_factory=list)
    winner: str = ""             # "" when no branch succeeded
    removed: int = 0             # branches deleted

Result shapes

Methods other than the three above return the daemon's JSON as plain dicts and lists. The keys you will meet most, as kosa8 0.9.36 sends them:

Transport and endpoints

Transport

Transport(
    *,
    socket_path: Optional[str] = None,
    host: Optional[str] = None,
    token: Optional[str] = None,
    insecure: bool = False,
    actor: str = "tool",
    agent_token: Optional[str] = None,
    headers: Optional[Mapping[str, str]] = None,
    timeout: Optional[float] = None,
    env: Optional[Mapping[str, str]] = None,
)

The same connection parameters as Kosa8Client, without checkpoint_store. It opens a new connection per request.

Response (returned by raw, in kosa8.transport) has status, headers, read() -> bytes, lines() -> Iterator[bytes] and close(), and is a context manager; read and lines close it when done.

resolve_endpoint

resolve_endpoint(
    host: Optional[str] = None,
    socket_path: Optional[str] = None,
    token: Optional[str] = None,
    insecure: bool = False,
    env: Optional[Mapping[str, str]] = None,
) -> Endpoint

Where to connect, by the kosa8 CLI's rules:

1. The value is host if given (even ""), else KOSA8_HOST, stripped of spaces and trailing slashes. Empty: the unix socket at socket_path or default_socket_path().
2. The scheme is lower-cased. unix:///path is a socket. tcp:// is read as http://. No scheme (host:port) is read as http://. Any other scheme raises Kosa8Error (KOSA8_HOST scheme "ftp" is not supported; use http://, https://, tcp:// or unix://). No host name raises Kosa8Error.
3. Plain http to a host other than localhost, ::1 or 127.* raises Kosa8Error (refusing to talk to … over plain http …) unless insecure or KOSA8_INSECURE=1.
4. The token is token if given; otherwise KOSA8_TOKEN, but only when the host came from KOSA8_HOST (never with an explicit host). A path in the URL is kept as a prefix for every request.

Endpoint (in kosa8.transport) is a frozen dataclass: kind ("unix" or "tcp"), socket_path, scheme, host, port, base_path, token, and the property target.

Other transport names

These six live in kosa8.transport and are not re-exported from kosa8.

Checkpoint stores

The store data is {sandbox_name: [checkpoint record, …]}. Any object with load() and update(fn) of these shapes can be passed as checkpoint_store. kosa8.checkpoints.validate_sandbox_name(name) is the name check used by next_id and record.

Errors

Every class has the attributes message, status, daemon_message, remediation, method, path and body (each None when not known). The constructor is Kosa8Error(message, *, status=None, daemon_message=None, remediation=None, method=None, path=None, body=None); Kosa8ApprovalRequiredError(message, approval, **kw); Kosa8StreamError(message, stdout="", stderr="", **kw).

error_from_response(status: int, body: Any, *, method: Optional[str] = None, path: Optional[str] = None) -> Kosa8ApiError

Builds the typed error for an HTTP error answer, as the client does. daemon_message is the body's error, or else its message, or a non-empty text body; remediation is the body's remediation. The message is "<daemon_message> (<remediation>)", or the daemon message alone, or "request failed with HTTP <status>". Details and real messages: Python errors and troubleshooting.