Aerol.ai MicroVM Python SDK

A lightweight Python SDK for the Aerol.ai MicroVM sandbox API.

Install

pip install aerolvm-sdk

For the asyncio client, AsyncMicroVM, add the async extra. It pulls in httpx; the plain install needs only websocket-client:

pip install "aerolvm-sdk[async]"

For local development from this repository:

cd sdk/python
pip install .

Usage

from microvm import MicroVM

client = MicroVM(pat_token="${SB_PAT_TOKEN}", api_url="https://sandbox.example.com")
health = client.health()
print(health)
print(health["sshGateway"])

sandbox = client.create(
	{
		"image": "ghcr.io/aerol-ai/ubuntu:22.04",
		"lifecycle": {
			"stopIfIdleFor": 3_600_000_000_000,
			"destroyAtAge": 86_400_000_000_000,
		},
		"failover": {"policy": "recreate"},
		"mounts": [
			{
				"type": "s3",
				"target": "/workspace",
				"source": "s3://bucket/prefix",
				"credentials": {"access_key_id": "AKIA...", "secret_access_key": "SECRET"},
			}
		],
	}
)
print(sandbox.sshPublicKey)
print(sandbox.sshPrivateKey)  # only returned by create()
print(client.mounts(sandbox.id))

sandbox.update_lifecycle(
	{
		"stopIfIdleFor": 7_200_000_000_000,
		"destroyAtAge": 172_800_000_000_000,
	}
)
print(sandbox.lifecycle)

Async client

AsyncMicroVM has every call MicroVM has, as coroutines, built from the same request code. Use it with async with so its connection pool is closed. Stream callbacks (onStdout, file watch events) run on the event loop.

import asyncio

from microvm import AsyncMicroVM


async def main() -> None:
    async with AsyncMicroVM() as client:
        sandboxes = await asyncio.gather(
            *(client.create(image="python:3.12", destroy_on_exit=True) for _ in range(3))
        )
        for sandbox in sandboxes:
            async with sandbox:
                result = await sandbox.exec("python -c 'print(42)'", timeout=60)
                print(result["stdout"])


asyncio.run(main())

Build Images

from microvm import Image, MicroVM

client = MicroVM(pat_token="${SB_PAT_TOKEN}", api_url="https://sandbox.example.com")
image = Image.base("ubuntu:22.04").run_commands("apt-get update", "apt-get install -y curl")

# create() accepts an Image directly
sandbox = client.create({"image": image})

# or build it explicitly and reuse the content-addressed tag
tag = client.build_image(image)
print(tag)

Example

python examples/create_sandbox.py --api-url http://127.0.0.1:21212 --pat-token test --image ghcr.io/aerol-ai/ubuntu:22.04

Streaming exec

import sys

handle = sandbox.exec_stream(
	{
		"command": "bash",
		"tty": True,
		"cols": 120,
		"rows": 40,
		"onStdout": lambda chunk: sys.stdout.buffer.write(chunk),
	}
)

handle.write("echo hello\n")
result = handle.wait()
print(result)

Watching files

Works on gVisor and Firecracker sandboxes. Other runtimes answer with a MicroVMHTTPError whose status_code is 501.

handle = sandbox.watch_files(
	"/workspace",
	lambda event: print(event.type, event.path),
	recursive=True,
)
sandbox.exec("touch /workspace/new.txt")
handle.stop()

An event with type overflow means some changes were lost. List the folder again to catch up.

Sessions

import sys

session = sandbox.create_session(
	{
		"name": "shell",
		"command": "bash",
		"workDir": "/workspace",
		"pty": True,
		"cols": 120,
		"rows": 40,
	}
)

print(session)
print(sandbox.list_sessions())
print(sandbox.session_log(session["id"]).decode("utf-8"))

handle = sandbox.attach_session(
	session["id"],
	{
		"cols": 120,
		"rows": 40,
		"onStdout": lambda chunk: sys.stdout.buffer.write(chunk),
	}
)
handle.write("echo attached\n")
print(handle.wait())