pip install microsandbox

import asyncio
from microsandbox import Sandbox

async def main():
    # Create a sandbox from an OCI image.
    sandbox = await Sandbox.create(
        "my-sandbox",
        image="alpine",
        cpus=1,
        memory=512,
    )

    # Run a command.
    output = await sandbox.shell("echo 'Hello from microsandbox!'")
    print(output.stdout_text)

    # Stop the sandbox.
    await sandbox.stop_and_wait()

asyncio.run(main())

from microsandbox import Sandbox

# Collected output.
output = await sandbox.exec("python3", ["-c", "print(1 + 1)"])
print(output.stdout_text)   # "2\n"
print(output.exit_code)      # 0

# Shell command (pipes, redirects, etc.).
output = await sandbox.shell("echo hello && pwd")
print(output.stdout_text)

# Full configuration via ExecOptions dict.
output = await sandbox.exec("python3", {
    "args": ["script.py"],
    "cwd": "/app",
    "env": {"PYTHONPATH": "/app/lib"},
    "timeout": 30.0,
})

# Streaming output.
handle = await sandbox.exec_stream("tail", ["-f", "/var/log/app.log"])
async for event in handle:
    match event.event_type:
        case "stdout": sys.stdout.buffer.write(event.data)
        case "stderr": sys.stderr.buffer.write(event.data)
        case "exited": break

await sandbox.stop_and_wait()

fs = sandbox.fs

# Write and read files.
await fs.write("/tmp/config.json", b'{"debug": true}')
content = await fs.read_text("/tmp/config.json")

# List a directory.
entries = await fs.list("/etc")
for entry in entries:
    print(f"{entry.path} ({entry.kind})")

# Copy between host and guest.
await fs.copy_from_host("./local-file.txt", "/tmp/file.txt")
await fs.copy_to_host("/tmp/output.txt", "./output.txt")

# Check existence and metadata.
if await fs.exists("/tmp/config.json"):
    meta = await fs.stat("/tmp/config.json")
    print(f"size: {meta.size}, kind: {meta.kind}")

# Streaming read.
async for chunk in await fs.read_stream("/tmp/large-file.bin"):
    process(chunk)

from microsandbox import Sandbox, Volume

# Create a 100 MiB named volume.
data = await Volume.create("my-data", quota_mib=100)

# Mount it in a sandbox.
writer = await Sandbox.create(
    "writer",
    image="alpine",
    volumes={"/data": Volume.named(data.name)},
    replace=True,
)

await writer.shell("echo 'hello' > /data/message.txt")
await writer.stop_and_wait()

# Mount the same volume in another sandbox (read-only).
reader = await Sandbox.create(
    "reader",
    image="alpine",
    volumes={"/data": Volume.named(data.name, readonly=True)},
    replace=True,
)

output = await reader.shell("cat /data/message.txt")
print(output.stdout_text)  # "hello\n"

await reader.stop_and_wait()

# Cleanup.
await Sandbox.remove("writer")
await Sandbox.remove("reader")
await Volume.remove("my-data")

from microsandbox import Sandbox, Volume, DiskImageFormat

# Mount a host disk image at a guest path. Format is inferred from the
# extension; pass `format=` to override. `fstype` is the inner FS agentd
# will mount; omit to let agentd autodetect.
sb = await Sandbox.create(
    "worker",
    image="alpine",
    volumes={
        "/data": Volume.disk("./data.qcow2", format=DiskImageFormat.QCOW2, fstype="ext4"),
        "/seed": Volume.disk("./seed.raw", readonly=True),
        "/scratch": Volume.tmpfs(size_mib=128, readonly=True),
    },
    replace=True,
)

from microsandbox import Network, Sandbox

# Default: public internet only (blocks private ranges).
sandbox = await Sandbox.create("public", image="alpine")

# Fully airgapped.
sandbox = await Sandbox.create(
    "isolated",
    image="alpine",
    network=Network.none(),
)

# Unrestricted.
sandbox = await Sandbox.create(
    "open",
    image="alpine",
    network=Network.allow_all(),
)

# DNS filtering.
sandbox = await Sandbox.create(
    "filtered",
    image="alpine",
    network=Network(
        deny_domains=("blocked.example.com",),
        deny_domain_suffixes=(".evil.com",),
    ),
)

sandbox = await Sandbox.create(
    "web",
    image="python",
    ports={8080: 80},  # host:8080 → guest:80
)

from microsandbox import Sandbox, Secret

sandbox = await Sandbox.create(
    "agent",
    image="python",
    secrets=[
        Secret.env("OPENAI_API_KEY",
                    value=os.environ["OPENAI_API_KEY"],
                    allow_hosts=["api.openai.com"]),
    ],
)

# Guest sees: OPENAI_API_KEY=$MSB_OPENAI_API_KEY (a placeholder)
# HTTPS to api.openai.com: placeholder is transparently replaced with the real key
# HTTPS to any other host with the placeholder: request is blocked

from microsandbox import Patch, Sandbox

sandbox = await Sandbox.create(
    "patched",
    image="alpine",
    patches=[
        Patch.text("/etc/greeting.txt", "Hello!\n"),
        Patch.mkdir("/app", mode=0o755),
        Patch.text("/app/config.json", '{"debug": true}', mode=0o644),
        Patch.copy_dir("./scripts", "/app/scripts"),
        Patch.append("/etc/hosts", "127.0.0.1 myapp.local\n"),
    ],
)

# Create in detached mode.
sandbox = await Sandbox.create(
    "background",
    image="python",
    detached=True,
)
await sandbox.detach()

# Later, from another process:
handle = await Sandbox.get("background")
reconnected = await handle.connect()
output = await reconnected.shell("echo reconnected")

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

from microsandbox import Network, Sandbox, TlsConfig

sandbox = await Sandbox.create(
    "tls-inspect",
    image="python",
    network=Network(
        tls=TlsConfig(
            bypass=("*.googleapis.com",),
            verify_upstream=True,
            intercepted_ports=(443,),
        ),
    ),
)

from microsandbox import Sandbox, all_sandbox_metrics, MiB

sandbox = await Sandbox.create("metrics-demo", image="python")

# Per-sandbox metrics.
m = await sandbox.metrics()
print(f"CPU: {m.cpu_percent:.1f}%")
print(f"Memory: {m.memory_bytes // MiB} MiB")
print(f"Uptime: {m.uptime_ms / 1000:.1f}s")

# Streaming metrics.
async for m in sandbox.metrics_stream(interval=1.0):
    print(f"CPU: {m.cpu_percent:.1f}%")

# All sandboxes at once.
all_metrics = await all_sandbox_metrics()
for name, metrics in all_metrics.items():
    print(f"{name}: {metrics.cpu_percent:.1f}%")

from microsandbox import is_installed, install

if not is_installed():
    await install()  # Downloads msb + libkrunfw to ~/.microsandbox/

cd sdk/python
uv sync --group dev

maturin develop --release

uv run pytest tests/

uv run --project sdk/python python examples/python/root-oci/main.py

uv run ruff check .