Python SDK Execution reference

See Commands for usage examples.

Sandbox

sandbox.exec()

async def exec(
    cmd: str,
    args: list[str] | ExecOptions | None = None,
    *,
    cwd: str | None = None,
    user: str | None = None,
    env: Mapping[str, str] | None = None,
    timeout: float | None = None,
    stdin: Stdin | bytes | None = None,
    tty: bool = False,
    rlimits: list[Rlimit] | None = None,
) -> ExecOutput

Example

out = await sandbox.exec(
    "python3",
    ["script.py"],
    cwd="/app",
    env={"PYTHONPATH": "/app/lib"},
    timeout=30.0,
)
print(out.stdout_text)
print(out.exit_code)  # 0

Run a command inside the sandbox and wait for it to complete, buffering all stdout and stderr into memory. The keyword-only options apply to this call alone and don’t change the sandbox’s defaults. For long-running processes or large output, use exec_stream() instead. Raises ExecTimeoutError if timeout elapses and MicrosandboxError if the process can’t be spawned.

Parameters

cmdstr

Command to execute (e.g. “python3”, “/usr/bin/node”).

argslist[str] | ExecOptions | None

Command arguments, or a typed options mapping that may include its own args list.

cwdstr | None

Working directory for this command.

userstr | None

Guest user to run as.

envMapping[str, str] | None

Environment variables, merged on top of the sandbox defaults.

timeoutfloat | None

Seconds before the process is killed.

stdinStdin | bytes | None

Stdin configuration. Raw bytes are sent inline; default is /dev/null.

ttybool

Allocate a pseudo-terminal, merging stdout and stderr.

rlimitslist[Rlimit] | None

POSIX resource limits applied to the process.

Returns

ExecOutput

Collected stdout, stderr, and exit status.

sandbox.shell()

async def shell(
    script: str,
    *,
    cwd: str | None = None,
    user: str | None = None,
    env: Mapping[str, str] | None = None,
    timeout: float | None = None,
    stdin: Stdin | bytes | None = None,
    tty: bool = False,
    rlimits: list[Rlimit] | None = None,
) -> ExecOutput

Example

out = await sandbox.shell("ls -la /app && echo done")
print(out.stdout_text)

Run a command through the sandbox’s configured shell (defaults to /bin/sh). Shell syntax like pipes, redirects, and && chains works. Accepts the same keyword-only options as exec().

Parameters

scriptstr

Shell command string (e.g. “ls -la /app && echo done”).

cwd, user, env, timeout, stdin, tty, rlimits

Same per-call options as exec().

Returns

ExecOutput

Collected stdout, stderr, and exit status.

Stream methods

sandbox.exec_stream()

async def exec_stream(
    cmd: str,
    args: list[str] | ExecOptions | None = None,
    *,
    cwd: str | None = None,
    user: str | None = None,
    env: Mapping[str, str] | None = None,
    timeout: float | None = None,
    stdin: Stdin | bytes | None = None,
    tty: bool = False,
    rlimits: list[Rlimit] | None = None,
) -> ExecHandle

Example

import sys

from microsandbox import ExecEventType

handle = await sandbox.exec_stream("tail", ["-f", "/var/log/app.log"])
async for event in handle:
    if event.event_type is ExecEventType.STDOUT:
        sys.stdout.buffer.write(event.data)
    elif event.event_type is ExecEventType.EXITED:
        break

Run a command with streaming output. Returns an ExecHandle that emits stdout, stderr, and exit events as they happen rather than buffering everything. Takes the same per-call options as exec(). Pass stdin=Stdin.pipe() to write to the process while it runs via take_stdin(). For TTY sessions, call resize(rows, cols) when the terminal dimensions change.

Parameters

cmdstr

Command to execute.

argslist[str] | ExecOptions | None

Command arguments, or a typed options mapping.

cwd, user, env, timeout, stdin, tty, rlimits

Same per-call options as exec().

Returns

ExecHandle

Streaming handle for receiving events and controlling the process.

sandbox.shell_stream()

async def shell_stream(
    script: str,
    *,
    cwd: str | None = None,
    user: str | None = None,
    env: Mapping[str, str] | None = None,
    timeout: float | None = None,
    stdin: Stdin | bytes | None = None,
    tty: bool = False,
    rlimits: list[Rlimit] | None = None,
) -> ExecHandle

Example

import sys

from microsandbox import ExecEventType

handle = await sandbox.shell_stream("for i in 1 2 3; do echo $i; sleep 1; done")
async for event in handle:
    if event.event_type is ExecEventType.STDOUT:
        sys.stdout.buffer.write(event.data)

Streaming variant of shell(): runs script through the configured shell but returns an ExecHandle instead of buffering output.

Parameters

scriptstr

Shell command string.

cwd, user, env, timeout, stdin, tty, rlimits

Same per-call options as exec().

Returns

ExecHandle

Streaming handle.

Attach methods

sandbox.attach()

async def attach(
    cmd: str,
    args: list[str] | None = None,
    *,
    cwd: str | None = None,
    user: str | None = None,
    env: Mapping[str, str] | None = None,
    detach_keys: str | None = None,
) -> int

Example

code = await sandbox.attach("bash", cwd="/app", env={"EDITOR": "vim"})

Bridge your terminal directly to a process inside the sandbox for a fully interactive PTY session. Press the configured detach key sequence (default Ctrl+]) to disconnect without stopping the process. Returns the process exit code.

Parameters

cmdstr

Command to run.

argslist[str] | None

Command arguments.

cwdstr | None

Working directory.

userstr | None

Guest user to run as.

envMapping[str, str] | None

Environment variables for the session.

detach_keysstr | None

Detach key sequence (e.g. “ctrl-]” or “ctrl-p,ctrl-q”).

Returns

int

Exit code of the process.

sandbox.attach_shell()

async def attach_shell() -> int

Example

code = await sandbox.attach_shell()

Bridge your terminal to the sandbox’s default shell in a fully interactive PTY session. Returns the shell’s exit code.

Returns

int

Exit code of the shell process.

ExecHandle

Returned by exec_stream() · shell_stream()

A handle to a running streaming execution and its events. A configured streaming timeout is enforced by the SDK even when the caller is not reading output. At expiry, the SDK waits for process startup before sending SIGKILL. Waiting or collecting reports ExecTimeoutError after the guest confirms exit; cancelling a pending wait does not cancel the deadline. Raw event readers receive the actual exit event. If termination cannot be delivered, the SDK reports a separate error without claiming the process exited. The SDK connection must remain open for this client-owned deadline to run.

handle.id

str Correlation ID for this execution

handle.take_stdin()

take_stdin()

Take the stdin writer. Returns None after the first call, or when stdin wasn’t piped

Returns

ExecSink \| None

handle.recv()

recv()

(async) Receive the next event. Returns None when the stream ends

Returns

ExecEvent \| None

handle.wait()

wait()

(async) Wait for the process to exit. Returns (code, success)

Returns

tuple[int, bool]

handle.collect()

collect()

(async) Drain remaining output and wait for exit

Returns

ExecOutput

handle.signal()

signal(sig)

(async) Send a POSIX signal (numeric) to the process

handle.kill()

kill()

(async) Send SIGKILL to the process

handle.resize()

resize(rows, cols)

(async) Resize the PTY to the given uint16 dimensions

ExecSink

Returned by ExecHandle.take_stdin()

Writer for sending data to a running process’s stdin. Obtained from ExecHandle.take_stdin() when the execution was configured with stdin=Stdin.pipe().

sink.write()

write(data)

(async) Write bytes to the process’s stdin

sink.close()

close()

(async) Close the sink. Sends EOF in non-TTY pipe mode; the guest PTY remains open in TTY mode

Stdin

Used by exec() · shell() · exec_stream() · shell_stream()

Factory for process stdin configuration.

Stdin.null()

Stdin.null()

Connect stdin to /dev/null (default)

Stdin.pipe()

Stdin.pipe()

Open a writable pipe. Write via take_stdin() on the handle

Stdin.bytes()

Stdin.bytes(data)

Inline data sent before the process starts, then EOF The factories set the corresponding StdinMode member.

Rlimit

Used by exec() · shell() · exec_stream() · shell_stream()

Frozen dataclass describing a POSIX resource limit. Construct one directly or via a factory, then pass a list as the rlimits argument.

Field	Type	Description
`resource`	`RlimitResource`	Which resource is limited
`soft`	`int`	Soft limit
`hard`	`int`	Hard limit

Rlimit.nofile()

Rlimit.nofile(limit)

Max open file descriptors

Rlimit.cpu()

Rlimit.cpu(secs)

CPU time limit in seconds

Rlimit.as_()

Rlimit.as_(\*, soft, hard)

Virtual memory size

Rlimit.nproc()

Rlimit.nproc(limit)

Max number of processes

Rlimit.fsize()

Rlimit.fsize(limit)

Max file size

Rlimit.memlock()

Rlimit.memlock(limit)

Max locked memory

Rlimit.stack()

Rlimit.stack(limit)

Max stack size

Types

ExecOptions

Accepted by exec() · exec_stream()

Typed dictionary for passing arguments and per-call options together as the second positional argument. Use a Stdin object to select a mode or raw bytes for inline data.

Key	Type	Description
args	`list[str]`	Command arguments
cwd	`str`	Working directory
user	`str`	Guest user
env	`Mapping[str, str]`	Per-command environment variables
timeout	`float`	Timeout in seconds
stdin	`Stdin` `\| bytes`	Stdin configuration or inline bytes
stdin_data	`bytes`	Inline stdin bytes when `stdin` is omitted
tty	`bool`	Whether to allocate a pseudo-terminal
rlimits	`list[Rlimit]`	POSIX resource limits

ExecOutput

Returned by exec() · shell() · ExecHandle.collect()

The result of a completed command execution: collected output plus exit status. All members are properties. In TTY mode, stdout_bytes contains the combined terminal output and stderr_bytes is empty because a PTY doesn’t preserve separate stdout and stderr streams.

Property	Type	Description
exit_code	`int`	Process exit code
success	`bool`	`True` when `exit_code == 0`
stdout_text	`str`	Collected stdout decoded as UTF-8. Raises on invalid encoding
stderr_text	`str`	Collected stderr decoded as UTF-8. Raises on invalid encoding
stdout_bytes	`bytes`	Raw stdout bytes
stderr_bytes	`bytes`	Raw stderr bytes

ExecEvent

Emitted by ExecHandle

Native event object emitted by recv() and by iterating an ExecHandle. Fields that don’t apply to a given event are None.

Property	Type	Description
event_type	`ExecEventType`	Exact event discriminator
pid	`int \| None`	Guest PID, set on `ExecEventType.STARTED`
data	`bytes \| None`	Output bytes on `STDOUT` / `STDERR`, or a UTF-8 failure message on `FAILED` / `STDIN_ERROR`
code	`int \| None`	Exit code on `EXITED`, or errno when available on `FAILED` / `STDIN_ERROR`

ExecEventType

Returned by ExecEvent.event_type

String enum (enum.StrEnum) identifying a streaming execution event.

Member	Value	Description
`ExecEventType.STARTED`	`"started"`	Process started; `pid` is populated
`ExecEventType.STDOUT`	`"stdout"`	Standard-output bytes are available in `data`
`ExecEventType.STDERR`	`"stderr"`	Standard-error bytes are available in `data`
`ExecEventType.EXITED`	`"exited"`	Process exited; `code` is populated
`ExecEventType.FAILED`	`"failed"`	Process startup failed
`ExecEventType.STDIN_ERROR`	`"stdin_error"`	Writing process stdin failed

ExitStatus

Used by ExecHandle.wait()

Frozen dataclass describing a process exit result. ExecHandle.wait() returns the same information as a (code, success) tuple.

Field	Type	Description
code	`int`	Process exit code
success	`bool`	`True` when `code == 0`

StdinMode

Used internally by Stdin

String enum (enum.StrEnum) identifying how command stdin is connected.

Member	Value	Description
`StdinMode.NULL`	`"null"`	Connect stdin to `/dev/null`
`StdinMode.PIPE`	`"pipe"`	Expose a writable `ExecSink`
`StdinMode.BYTES`	`"bytes"`	Send inline bytes, then EOF

RlimitResource

Used by Rlimit.resource

String enum (enum.StrEnum) naming a limitable POSIX resource.

Member	Value	Description
`RlimitResource.CPU`	`"cpu"`	CPU time
`RlimitResource.FSIZE`	`"fsize"`	File size
`RlimitResource.DATA`	`"data"`	Data segment size
`RlimitResource.STACK`	`"stack"`	Stack size
`RlimitResource.CORE`	`"core"`	Core file size
`RlimitResource.RSS`	`"rss"`	Resident set size
`RlimitResource.NPROC`	`"nproc"`	Number of processes
`RlimitResource.NOFILE`	`"nofile"`	Open file descriptors
`RlimitResource.MEMLOCK`	`"memlock"`	Locked memory
`RlimitResource.AS`	`"as"`	Virtual memory
`RlimitResource.LOCKS`	`"locks"`	File locks
`RlimitResource.SIGPENDING`	`"sigpending"`	Pending signals
`RlimitResource.MSGQUEUE`	`"msgqueue"`	Message queue size
`RlimitResource.NICE`	`"nice"`	Nice priority ceiling
`RlimitResource.RTPRIO`	`"rtprio"`	Real-time priority ceiling
`RlimitResource.RTTIME`	`"rttime"`	Real-time CPU time

Managed jobs

Run and reconnect to background commands. See Background jobs for lifecycle, retention, and runtime limits.

sandbox.exec_detached()

async def exec_detached(
    cmd: str,
    args: list[str] | ExecOptions | None = None,
    *,
    cwd: str | None = None,
    user: str | None = None,
    env: Mapping[str, str] | None = None,
    timeout: float | None = None,
    stdin: Stdin | bytes | None = None,
    tty: bool = False,
    rlimits: list[Rlimit] | None = None,
) -> Job

Start a job and return after guest startup is confirmed. Uses the same options as exec(). Stdin defaults to a retained pipe; Stdin.null() closes it, and bytes supply at most 64 KiB followed by EOF.

Example

job = await sb.exec_detached("cat", cwd="/tmp")
io = await job.attach()
await io.write_stdin(b"hello\n")
await io.detach()
job = await sb.get_job(job.id)
await job.eof()
result = await job.wait()
output = await job.logs()
page = await sb.list_jobs(all=True, limit=50)

sandbox.get_job()

async def get_job(id: str) -> Job

Retrieve a retained job by ID. Also available on SandboxHandle.

sandbox.list_jobs()

async def list_jobs(*, all: bool = False, limit: int = 50, cursor: str | None = None) -> JobPage

List active jobs, or include retained history with the all option. Returns items and a next-page cursor. Default page size: 50; range: 1–100. Also available on SandboxHandle.

Job

job.id

id: str

The job ID within its sandbox.

job.inspect()

async def inspect() -> JobInfo

Read current metadata without starting a stopped sandbox.

job.wait()

async def wait() -> JobExit

Wait for a confirmed exit without consuming output. The result contains the exit code, success flag, and timeout flag. Cancelling the await leaves the job running.

job.signal()

async def signal(signal: int) -> None

Send a Linux signal to the process group.

job.kill()

async def kill() -> None

Request SIGKILL. Wait separately to confirm exit.

job.eof()

async def eof() -> None

Permanently close pipe input after accepted bytes. PTYs reject this operation; use terminal control characters instead.

job.logs()

async def logs(
    *, tail: int | None = None, since: int | None = None,
    until: int | None = None, sources: list[str] | None = None,
    from_cursor: str | None = None
) -> list[JobLogEntry]

Read retained output. Entries contain original bytes, a source, a Unix-millisecond timestamp, and an opaque cursor belonging to this job. since is inclusive and until exclusive, both in Unix milliseconds; from_cursor cannot be combined with since.

job.log_stream()

async def log_stream(
    *, tail: int | None = None, since: int | None = None,
    until: int | None = None, sources: list[str] | None = None,
    from_cursor: str | None = None, follow: bool = False
) -> JobLogStream

Read retained output and optionally follow new records. Supports async iteration; call await stream.close() when finished.

job.follow_logs()

async def follow_logs(
    *, tail: int | None = None, since: int | None = None,
    until: int | None = None, sources: list[str] | None = None,
    from_cursor: str | None = None
) -> JobLogStream

Stream logs with following enabled. Accepts the same filters as logs().

job.attach()

async def attach(
    *, read_only: bool = False, replay_bytes: int | None = None,
    cursor: str | None = None,
) -> JobAttachment

Attach to live I/O. Only one attachment may own input; read-only observers may coexist. Recent-byte replay and cursor replay are alternatives.

JobAttachment

attachment.recv()

async def recv() -> JobEvent | None

Read an output, history-gap, or completion event. Also supports async iteration. One reader may run alongside input writes.

attachment.write_stdin()

async def write_stdin(data: bytes) -> None

Write up to 16 KiB through the input-owning attachment.

attachment.resize()

async def resize(rows: int, cols: int) -> None

Resize the job terminal. Each dimension must fit an unsigned 16-bit integer.

attachment.detach()

async def detach() -> None

Release the attachment when finished. The job and its input remain open.

JobLogStream

stream.close()

async def close() -> None

Stop observing logs. This does not stop the job.

​Sandbox

​sandbox.exec()

​sandbox.shell()

​sandbox.exec_stream()

​sandbox.shell_stream()

​sandbox.attach()

​sandbox.attach_shell()

​ExecHandle

​handle.id

​handle.take_stdin()

​handle.recv()

​handle.wait()

​handle.collect()

​handle.signal()

​handle.kill()

​handle.resize()

​ExecSink

​sink.write()

​sink.close()

​Stdin

​Stdin.null()

​Stdin.pipe()

​Stdin.bytes()

​Rlimit

​Rlimit.nofile()

​Rlimit.cpu()

​Rlimit.as_()

​Rlimit.nproc()

​Rlimit.fsize()

​Rlimit.memlock()

​Rlimit.stack()

​Types

​ExecOptions

​ExecOutput

​ExecEvent

​ExecEventType

​ExitStatus

​StdinMode

​RlimitResource

​Managed jobs

​sandbox.exec_detached()

​sandbox.get_job()

​sandbox.list_jobs()

​Job

​job.id

​job.inspect()

​job.wait()

​job.signal()

​job.kill()

​job.eof()

​job.logs()

​job.log_stream()

​job.follow_logs()

​job.attach()

​JobAttachment

​attachment.recv()

​attachment.write_stdin()

​attachment.resize()

​attachment.detach()

​JobLogStream

​stream.close()

Sandbox

sandbox.exec()

sandbox.shell()

sandbox.exec_stream()

sandbox.shell_stream()

sandbox.attach()

sandbox.attach_shell()

ExecHandle

handle.id

handle.take_stdin()

handle.recv()

handle.wait()

handle.collect()

handle.signal()

handle.kill()

handle.resize()

ExecSink

sink.write()

sink.close()

Stdin

Stdin.null()

Stdin.pipe()

Stdin.bytes()

Rlimit

Rlimit.nofile()

Rlimit.cpu()

Rlimit.as_()

Rlimit.nproc()

Rlimit.fsize()

Rlimit.memlock()

Rlimit.stack()

Types

ExecOptions

ExecOutput

ExecEvent

ExecEventType

ExitStatus

StdinMode

RlimitResource

Managed jobs

sandbox.exec_detached()

sandbox.get_job()

sandbox.list_jobs()

Job

job.id

job.inspect()

job.wait()

job.signal()

job.kill()

job.eof()

job.logs()

job.log_stream()

job.follow_logs()

job.attach()

JobAttachment

attachment.recv()

attachment.write_stdin()

attachment.resize()

attachment.detach()

JobLogStream

stream.close()