vtrix-sandbox 0.1.0

Vtrix sandbox SDK — async Python client for Vtrix sandbox environments

vtrix-sandbox

Async Python SDK for Vtrix sandbox — run commands and manage files in isolated Linux environments over a persistent WebSocket connection.

Requires Python 3.10+

Installation

pip install vtrix-sandbox

Or from source:

pip install .

Quick Start

import asyncio
from src import Client, ClientOptions, CreateOptions

async def main():
    client = Client(ClientOptions(
        base_url="http://your-hermes-host:8080",
        token="your-token",
        project_id="your-project-id",
    ))

    # Create a sandbox and wait for it to become active
    sb = await client.create(CreateOptions(user_id="user-123"))

    # Run a command and get the result
    result = await sb.run_command("echo hello && uname -a")
    print(f"exit_code={result.exit_code}")
    print(result.output)

    await sb.close()

asyncio.run(main())

Core classes

Class	What it does
Client	Creates and manages sandbox instances
Sandbox	Runs commands and manages files in an isolated environment
Command	Handles a running or completed process
CommandFinished	Result after a command completes — extends Command with exit_code and output

---

Client

Client(opts: ClientOptions)

Creates a new client. The client is reusable and safe for concurrent use across multiple sandbox sessions.

Field	Type	Required	Description
base_url	str	Yes	Hermes gateway URL (e.g. http://host:8080).
token	str	No	Bearer token for authentication.
project_id	str	No	Value sent as X-Project-ID header.

client = Client(ClientOptions(
    base_url="http://your-hermes-host:8080",
    token="your-token",
    project_id="your-project-id",
))

await client.create(opts) → Sandbox

Use client.create() to launch a new sandbox, poll until it is active, and open a WebSocket connection. This is the primary entry point for starting a sandbox session. Pass env to set default environment variables that all commands in this sandbox will inherit.

Returns: Sandbox

Parameter	Type	Required	Description
opts.user_id	str	Yes	Owner of the sandbox.
opts.spec	Spec	No	Resource spec (cpu, memory, image).
opts.labels	dict[str, str]	No	Arbitrary key-value metadata attached to the sandbox.
opts.payloads	list[Payload]	No	Initialisation calls sent to the pod after creation.
opts.ttl_hours	int	No	Sandbox lifetime in hours. Uses the server default when 0.
opts.env	dict[str, str]	No	Default environment variables inherited by all commands. Per-command RunOptions.env values override these.

sb = await client.create(CreateOptions(
    user_id="user-123",
    spec=Spec(cpu="2", memory="4Gi"),
    ttl_hours=2,
    env={"NODE_ENV": "production"},
))

await client.attach(sandbox_id) → Sandbox

Use client.attach() to connect to an existing sandbox without creating a new one. Use this to resume a session after a restart or to connect from a different process. Auth uses the client-level token and project ID.

Returns: Sandbox

Parameter	Type	Required	Description
sandbox_id	str	Yes	ID of the sandbox to connect to.

sb = await client.attach("sandbox-id-abc")

await client.list(opts?) → ListResult

Use client.list() to enumerate sandboxes visible to the current credentials. Filter by user_id or status to scope results.

Returns: ListResult — .items is list[SandboxInfo], .pagination has total, limit, offset, has_more.

Parameter	Type	Required	Description
opts.user_id	str	No	Return only sandboxes owned by this user.
opts.status	str	No	Filter by status: "active", "stopped", etc.
opts.limit	int	No	Maximum number of results.
opts.offset	int	No	Pagination offset.

result = await client.list(ListOptions(user_id="user-123", status="active"))
print(f"Found {result.pagination.total} sandboxes")

await client.get(sandbox_id) → SandboxInfo

Use client.get() to fetch metadata for a sandbox by ID without opening a WebSocket connection.

Returns: SandboxInfo

info = await client.get("sandbox-id-abc")
print(info.status)

await client.delete(sandbox_id)

Call client.delete() to permanently delete a sandbox. This cannot be undone.

Returns: None

await client.delete("sandbox-id-abc")

await client.aclose()

Call client.aclose() to close the underlying HTTP client and release connections. Or use the client as an async context manager to close automatically.

async with Client(ClientOptions(...)) as client:
    sb = await client.create(CreateOptions(user_id="user-123"))
    ...

---

Sandbox

A Sandbox instance gives you full control over an isolated environment. You receive one from client.create() or client.attach().

Properties

sandbox.created_at → datetime

The created_at property returns the sandbox creation time as a timezone-aware datetime (UTC). Returns datetime.min if the field is empty or unparsable.

Returns: datetime

print(sb.created_at.isoformat())

sandbox.status → str

The status property reports the cached lifecycle state of the sandbox. Call await sandbox.refresh() first if you need a live value.

Returns: str — "active", "stopped", "destroying", etc.

print(sb.status)

sandbox.expire_at → str

The expire_at property returns the cached expiry timestamp. Call await sandbox.refresh() first for an accurate value.

Returns: str — RFC 3339 timestamp.

print(sb.expire_at)

sandbox.timeout → int

The timeout property returns the remaining sandbox lifetime in milliseconds based on the cached expire_at. Returns 0 if the sandbox has already expired.

Returns: int — milliseconds remaining; 0 if expired.

if sb.timeout < 60_000:
    await sb.extend(1)  # extend by 1 hour

---

Running Commands

await sandbox.run_command(cmd, args?, opts?) → CommandFinished

sandbox.run_command() executes a command inside the sandbox and blocks until it finishes.

Set opts.stdout or opts.stderr to receive output in real time while still blocking — useful for progress logging.

Returns: CommandFinished

Parameter	Type	Required	Description
cmd	str	Yes	Shell command to run.
args	list[str]	No	Arguments shell-quoted and appended to cmd. Prevents injection.
opts.working_dir	str	No	Working directory inside the sandbox.
opts.timeout_sec	int	No	Kill the command after this many seconds.
opts.env	dict[str, str]	No	Per-command environment variables. Merges with sandbox defaults.
opts.sudo	bool	No	Prepend sudo -E to the command.
opts.stdin	str	No	Data written to the command's stdin before reading output.
opts.stdout	IO	No	Receives stdout chunks as they arrive.
opts.stderr	IO	No	Receives stderr chunks as they arrive.

import sys

result = await sb.run_command("npm install", opts=RunOptions(
    working_dir="/app",
    stdout=sys.stdout,
    stderr=sys.stderr,
))
print(f"exit_code={result.exit_code}")

await sandbox.run_command_detached(cmd, args?, opts?) → Command

Use sandbox.run_command_detached() to start a command in the background and return immediately. Use this for long-running processes such as servers where you want to do other work while the command runs, then call cmd.wait() when you need the result.

Returns: Command

cmd = await sb.run_command_detached("python server.py", opts=RunOptions(
    working_dir="/app",
    env={"PORT": "8080"},
))
# ... do other work ...
finished = await cmd.wait()

async for ev in sandbox.run_command_stream(cmd, args?, opts?) → AsyncIterator[ExecEvent]

Use sandbox.run_command_stream() to run a command and stream ExecEvent values in real time. Use this instead of run_command when you need to process stdout and stderr as separate, typed events — for example, to display them with different colours or route them to different log streams.

Returns: AsyncIterator[ExecEvent]

ev.type	Meaning
"start"	Command has started executing.
"stdout"	A chunk of standard output. Read from ev.data.
"stderr"	A chunk of standard error. Read from ev.data.
"done"	Command has finished.

async for ev in sb.run_command_stream("make build"):
    if ev.type == "stdout":
        print(ev.data, end="")
    elif ev.type == "stderr":
        print(ev.data, end="", file=sys.stderr)

async for ev in sandbox.exec_logs(cmd_id) → AsyncIterator[ExecEvent]

Use sandbox.exec_logs() to attach to a running or completed command and stream its output. It replays buffered output first (up to 512 KB), then streams live events for commands still running. Use this to replay logs from a detached command or to attach a second observer.

Returns: AsyncIterator[ExecEvent]

Parameter	Type	Required	Description
cmd_id	str	Yes	ID of the command to attach to.

async for ev in sb.exec_logs(cmd.cmd_id):
    print(f"[{ev.type}] {ev.data}")

sandbox.get_command(cmd_id) → Command

Use sandbox.get_command() to reconstruct a Command handle from a known cmd_id. Use this to reconnect to a command started in a previous call without going through run_command_detached again.

Returns: Command

Parameter	Type	Required	Description
cmd_id	str	Yes	ID of the command to retrieve.

cmd = sb.get_command("cmd-id-abc")
result = await cmd.wait()

await sandbox.kill(cmd_id, signal="SIGTERM")

Call sandbox.kill() to send a signal to a running command by ID. The signal is sent to the entire process group, so child processes are also terminated. Send SIGTERM for graceful shutdown or SIGKILL for immediate termination.

Returns: None

Parameter	Type	Required	Description
cmd_id	str	Yes	ID of the command to signal.
signal	str	No	Signal name: "SIGTERM" (default), "SIGKILL", "SIGINT", "SIGHUP".

await sb.kill(cmd.cmd_id, "SIGTERM")

---

Command

A Command represents a running or completed process. You receive one from run_command_detached() or get_command(). CommandFinished extends Command and adds exit_code and output.

Properties

Property	Type	Description
cmd_id	str	Unique identifier for this command execution.
pid	int	Process ID inside the sandbox.
cwd	str	Working directory where the command is executing.
started_at	str	RFC 3339 timestamp when the command started.
exit_code	int | None	Exit status. None while the command is still running.

await command.wait() → CommandFinished

Use command.wait() to block until a detached command finishes and get the resulting CommandFinished object.

Returns: CommandFinished — exit_code, output, cmd_id.

cmd = await sb.run_command_detached("python server.py")
# ... do other work ...
result = await cmd.wait()
if result.exit_code != 0:
    print("Command failed:", result.output)

async for ev in command.logs() → AsyncIterator[LogEvent]

Call command.logs() to stream structured log entries as they arrive. Each LogEvent has stream ("stdout" or "stderr") and data. Use this instead of sandbox.exec_logs() when you already have a Command handle.

Returns: AsyncIterator[LogEvent]

async for ev in cmd.logs():
    if ev.stream == "stdout":
        sys.stdout.write(ev.data)
    else:
        sys.stderr.write(ev.data)

await command.stdout() → str

Use command.stdout() to collect the full standard output as a string.

Returns: str

output = await cmd.stdout()
data = json.loads(output)

await command.stderr() → str

Use command.stderr() to collect the full standard error output as a string.

Returns: str

errors = await cmd.stderr()
if errors:
    print("Command errors:", errors, file=sys.stderr)

await command.collect_output(stream) → str

Use command.collect_output() to collect stdout, stderr, or both as a single string.

Returns: str

Parameter	Type	Required	Description
stream	str	Yes	"stdout", "stderr", or "both".

combined = await cmd.collect_output("both")

await command.kill(signal="SIGTERM")

Call command.kill() to send a signal to this command. See sandbox.kill() for valid signal names.

Returns: None

Parameter	Type	Required	Description
signal	str	No	Signal name: "SIGTERM" (default), "SIGKILL", "SIGINT", "SIGHUP".

await cmd.kill("SIGKILL")

---

File Operations

await sandbox.read(path) → ReadResult

Use sandbox.read() to read a file from the sandbox. Text files up to 200 KB are returned in full; larger files are truncated (truncated=True). Image files are detected automatically and returned as base64-encoded data with a MIME type. Raises an exception if the file does not exist.

Returns: ReadResult

Field	Type	Description
type	Literal["text", "image"]	Type of the file.
content	str	File content (text files).
truncated	bool	True if the file was larger than 200 KB. Use read_stream for the full content.
data	str	Base64-encoded bytes (image files).
mime_type	str	MIME type (image files, e.g. "image/png").

result = await sb.read("/app/config.json")
if result.truncated:
    # use read_stream for the full file
    pass
print(result.content)

await sandbox.write(path, content) → WriteResult

Use sandbox.write() to write a text string to a file. Creates parent directories automatically. Returns the number of bytes written.

Returns: WriteResult — .bytes_written.

Parameter	Type	Required	Description
path	str	Yes	Destination path inside the sandbox.
content	str	Yes	Text content to write.

result = await sb.write("/app/config.json", json.dumps(config))
print(f"Wrote {result.bytes_written} bytes")

await sandbox.edit(path, old_text, new_text) → EditResult

Use sandbox.edit() to replace an exact occurrence of old_text with new_text inside a file. Raises an exception if old_text appears zero times or more than once — ensuring the edit is unambiguous.

Returns: EditResult — .message.

Parameter	Type	Required	Description
path	str	Yes	Path to the file inside the sandbox.
old_text	str	Yes	The exact text to find and replace.
new_text	str	Yes	The text to substitute in its place.

await sb.edit("/app/config.json", '"port": 3000', '"port": 8080')

await sandbox.write_files(files)

Use sandbox.write_files() to upload one or more binary files in a single round trip. Creates parent directories automatically. Use this for uploading compiled binaries, images, or executable scripts.

Returns: None

Parameter	Type	Required	Description
files[].path	str	Yes	Destination path inside the sandbox.
files[].content	bytes	Yes	Raw file bytes.
files[].mode	int	No	Unix permission bits (e.g. 0o755 for executable). Uses server default when None.

await sb.write_files([
    WriteFileEntry(path="/app/run.sh", content=script_bytes, mode=0o755),
    WriteFileEntry(path="/app/data.bin", content=data_bytes),
])

await sandbox.read_to_buffer(path) → bytes | None

Use sandbox.read_to_buffer() to read a file into memory as raw bytes. Returns None (not an exception) when the file does not exist, making it easy to check for optional files without try/except.

Returns: bytes or None if the file does not exist.

Parameter	Type	Required	Description
path	str	Yes	File path inside the sandbox.

buf = await sb.read_to_buffer("/app/output.bin")
if buf is not None:
    process(buf)

async for chunk in sandbox.read_stream(path, chunk_size?) → AsyncIterator[bytes]

Use sandbox.read_stream() to read a large file in chunks. Use this instead of read when the file exceeds 200 KB or you need complete binary content without truncation. Each chunk is already decoded bytes (base64 decoding is handled internally).

Returns: AsyncIterator[bytes]

Parameter	Type	Required	Description
path	str	Yes	File path inside the sandbox.
chunk_size	int	No	Bytes per chunk. Defaults to 65536.

with open("large.csv", "wb") as f:
    async for chunk in sb.read_stream("/data/large.csv"):
        f.write(chunk)

await sandbox.mkdir(path)

Use sandbox.mkdir() to create a directory and all parent directories. Safe to call on paths that already exist.

Returns: None

Parameter	Type	Required	Description
path	str	Yes	Directory to create.

await sb.mkdir("/app/logs")

await sandbox.list_files(path) → list[FileEntry]

Use sandbox.list_files() to list the contents of a directory. Raises an exception if the path does not exist or is not a directory.

Returns: list[FileEntry] — each entry has name, path, size, is_dir, modified_at (RFC 3339 string or None).

Parameter	Type	Required	Description
path	str	Yes	Directory path inside the sandbox.

entries = await sb.list_files("/app")
for entry in entries:
    print(f"{'d' if entry.is_dir else 'f'} {entry.name}")

await sandbox.stat(path) → FileInfo

Use sandbox.stat() to get metadata for a path. Unlike most operations, this does not raise an exception when the path does not exist — check info.exists instead.

Returns: FileInfo

Field	Type	Description
exists	bool	False when the path does not exist.
is_file	bool	True for regular files.
is_dir	bool	True for directories.
size	int	File size in bytes.
modified_at	str | None	RFC 3339 timestamp, or None.

info = await sb.stat("/app/config.json")
if not info.exists:
    await sb.write("/app/config.json", "{}")

await sandbox.exists(path) → bool

Use sandbox.exists() to check whether a path exists. A convenient shorthand for stat when you only need the existence check.

Returns: bool

Parameter	Type	Required	Description
path	str	Yes	Path to check.

if await sb.exists("/app/config.json"):
    # ...
    pass

await sandbox.upload_file(local_path, sandbox_path, opts?)

Use sandbox.upload_file() to upload a file from the local filesystem into the sandbox.

Returns: None

Parameter	Type	Required	Description
local_path	str	Yes	Absolute path on the local machine.
sandbox_path	str	Yes	Destination path inside the sandbox.
opts.mkdir_recursive	bool	No	Create parent directories on the sandbox side if they do not exist.

await sb.upload_file("/local/model.bin", "/app/model.bin", FileOptions(mkdir_recursive=True))

await sandbox.download_file(sandbox_path, local_path, opts?) → str | None

Use sandbox.download_file() to download a file from the sandbox to the local filesystem. Returns the absolute local path on success, or None when the sandbox file does not exist.

Returns: str (absolute local path) or None if the file does not exist.

Parameter	Type	Required	Description
sandbox_path	str	Yes	Path to the file inside the sandbox.
local_path	str	Yes	Destination path on the local machine.
opts.mkdir_recursive	bool	No	Create local parent directories if they do not exist.

dst = await sb.download_file("/app/output.json", "/tmp/output.json")
if dst is not None:
    print(f"Saved to {dst}")

await sandbox.download_files(entries, opts?) → dict[str, str]

Use sandbox.download_files() to download multiple files in one call (up to 8 concurrent). Returns a mapping of sandbox_path → absolute local path for each downloaded file.

Returns: dict[str, str] — sandbox_path → absolute local path.

results = await sb.download_files([
    DownloadEntry(sandbox_path="/app/out.json", local_path="/tmp/out.json"),
    DownloadEntry(sandbox_path="/app/log.txt", local_path="/tmp/log.txt"),
])
for sandbox_path, local_path in results.items():
    print(f"{sandbox_path} → {local_path}")

sandbox.domain(port) → str

Use sandbox.domain() to get the publicly accessible URL for an exposed port. The sandbox must be created with this port declared.

Returns: str

Parameter	Type	Required	Description
port	int	Yes	Port number to resolve.

url = sb.domain(3000)
print(f"App running at {url}")

---

Lifecycle

await sandbox.refresh()

Call sandbox.refresh() to re-fetch sandbox metadata from the server and update sb.info. Call this before reading sb.status or sb.expire_at if you need current values.

Returns: None

await sb.refresh()
print(sb.status)

await sandbox.stop(opts?)

Call sandbox.stop() to pause the sandbox without deleting it. Set opts.blocking=True to wait until the sandbox reaches "stopped" or "failed" status before returning.

Returns: None

Parameter	Type	Required	Description
opts.blocking	bool	No	Poll until the sandbox has stopped.
opts.poll_interval	float	No	How often to poll in seconds. Defaults to 2.0.
opts.timeout	float	No	Maximum time to wait in seconds. Defaults to 300.

await sb.stop(StopOptions(blocking=True))

await sandbox.start()

Use sandbox.start() to resume a stopped sandbox.

Returns: None

await sb.start()

await sandbox.restart()

Use sandbox.restart() to stop and restart the sandbox.

Returns: None

await sb.restart()

await sandbox.extend(hours=0)

Use sandbox.extend() to extend the sandbox TTL by hours. Pass 0 to use the server default (12 hours).

Returns: None

Parameter	Type	Required	Description
hours	int	No	Number of hours to add. Pass 0 for the server default (12 hours).

await sb.extend(2)  # extend by 2 hours

await sandbox.extend_timeout(hours=0)

Use sandbox.extend_timeout() to extend the TTL and immediately refresh sb.info in one call.

Returns: None

await sb.extend_timeout(1)  # +1 hour, then refresh

await sandbox.update(opts)

Use sandbox.update() to change the sandbox spec, image, or payloads. Changing payloads triggers a sandbox restart.

Returns: None

Parameter	Type	Required	Description
opts.spec	Spec	No	New resource spec.
opts.image	str	No	New container image tag.
opts.payloads	list[Payload]	No	Replaces all stored payloads and triggers a restart.

await sb.update(UpdateOptions(spec=Spec(cpu="4", memory="8Gi")))

await sandbox.configure(payloads?)

Call sandbox.configure() to immediately apply the current configuration to the running pod. Optionally override the stored payloads for this apply only.

Returns: None

await sb.configure()

await sandbox.delete()

Call sandbox.delete() to permanently delete the sandbox. This cannot be undone.

Returns: None

await sb.delete()

await sandbox.close()

Call sandbox.close() to close the WebSocket connection. Use try/finally or the async with context manager to ensure the connection is always freed.

Returns: None

try:
    result = await sb.run_command("python train.py")
finally:
    await sb.close()

Sandbox also supports the async context manager protocol — __aenter__ returns the sandbox itself, __aexit__ calls close():

async with await client.create(CreateOptions(user_id="user-123")) as sb:
    result = await sb.run_command("python train.py")
# close() called automatically

---

Examples

File	Description
examples/basic.py	Create a sandbox, run commands, use detached execution
examples/stream.py	Real-time streaming, exec_logs replay, Command.logs/stdout
examples/attach.py	Reconnect to an existing sandbox by ID
examples/files.py	Read, write, edit, upload, download, and stream files
examples/lifecycle.py	Stop, start, extend, update, and delete sandboxes

python examples/basic.py

License

MIT — see LICENSE.