Sandbox

@staticmethod
async def create(name: str, **kwargs) -> Sandbox

async with await Sandbox.create("my-sandbox", image="alpine") as sb:
    output = await sb.shell("echo hello")
    print(output.stdout_text)
# sandbox is automatically killed and removed on exit

@staticmethod
def create_with_progress(name: str, **kwargs) -> PullSession

@staticmethod
async def start(name: str, *, detached: bool = False) -> Sandbox

@staticmethod
async def get(name: str) -> SandboxHandle

@staticmethod
async def list() -> list[SandboxHandle]

@staticmethod
async def remove(name: str) -> None

@property
async def name(self) -> str

@property
async def owns_lifecycle(self) -> bool

@property
def fs(self) -> SandboxFs

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

async def attach_shell() -> int

async def detach() -> None

async def drain() -> None

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

output = await sb.exec(
    "python",
    ["compute.py"],
    cwd="/app",
    env={"PYTHONPATH": "/app/lib"},
    timeout=30.0,
)

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

async def kill() -> None

async def metrics() -> SandboxMetrics

async def metrics_stream(interval: float = 1.0) -> MetricsStream

async def logs(
    tail: int | None = None,
    since_ms: float | None = None,
    until_ms: float | None = None,
    sources: list[str] | None = None,
) -> list[LogEntry]

import asyncio
import microsandbox

async def main():
    handle = await microsandbox.Sandbox.get("web")

    # Default — all user-program output, regardless of pipe/pty mode
    entries = await handle.logs()

    for e in entries:
        source = {
            "stdout": "OUT",
            "stderr": "ERR",
            "output": "PTY",
            "system": "SYS",
        }[e.source]
        print(
            f"[{e.timestamp_ms / 1000:.3f}] "
            f"{source} {e.session_id}: {e.text().rstrip()}"
        )

    # Filtered: last 50 entries from the past hour, including system lines
    import time
    recent = await handle.logs(
        tail=50,
        since_ms=(time.time() - 3600) * 1000,
        sources=["stdout", "stderr", "output", "system"],
    )

asyncio.run(main())

from datetime import datetime, timezone
dt = datetime.fromtimestamp(e.timestamp_ms / 1000, timezone.utc)

async def remove_persisted() -> None

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 | str | None = None,
    tty: bool = False,
    rlimits: list[Rlimit] | None = None,
) -> ExecOutput

async def shell_stream(script: str) -> ExecHandle

async def stop() -> None

async def stop_and_wait() -> tuple[int, bool]

async def wait() -> tuple[int, bool]

@staticmethod
def append(path: str, content: str) -> PatchConfig

@staticmethod
def copy_dir(src: str, dst: str, *, replace: bool = False) -> PatchConfig

@staticmethod
def copy_file(
    src: str,
    dst: str,
    *,
    mode: int | None = None,
    replace: bool = False,
) -> PatchConfig

@staticmethod
def mkdir(path: str, *, mode: int | None = None) -> PatchConfig

@staticmethod
def remove(path: str) -> PatchConfig

@staticmethod
def symlink(target: str, link: str, *, replace: bool = False) -> PatchConfig

@staticmethod
def text(
    path: str,
    content: str,
    *,
    mode: int | None = None,
    replace: bool = False,
) -> PatchConfig

from microsandbox import Sandbox

session = Sandbox.create_with_progress("my-sandbox", image="ubuntu:latest")
async with session:
    async for event in session.progress:
        if event.event_type == "resolved":
            print(f"{event.layer_count} layers, {event.total_download_bytes} bytes")
        elif event.event_type == "layer_download_progress":
            print(f"layer {event.layer_index}: {event.downloaded_bytes}/{event.total_bytes}")
    sb = await session.result()

session = Sandbox.create_with_progress("my-sandbox", image="ubuntu:latest")
async with session:
    async for event in session.progress:
        print(event)
    sb = await session.result()

from dataclasses import dataclass

@dataclass(frozen=True, slots=True)
class InitConfig:
    cmd: str
    args: tuple[str, ...] = ()
    env: Mapping[str, str] = {}

from microsandbox import InitConfig, Sandbox

# Common case: bare string.
sb = await Sandbox.create("worker", image="jrei/systemd-debian:12", init="auto")

# Argv / env: dataclass.
sb = await Sandbox.create(
    "worker",
    image="jrei/systemd-debian:12",
    init=InitConfig(
        cmd="/lib/systemd/systemd",
        args=("--unit=multi-user.target",),
        env={"container": "microsandbox"},
    ),
)