Machine API Reference

One Machine class, two backends. This is the reference for the Machine class in the smolmachines SDK (Node) / smol module (Python). The same API runs against the embedded in-process engine (default — no server) or the smolfleet cloud, selected via ConnectOptions ({ target: 'cloud', apiKey } or the SMOL_CLOUD_TOKEN env var). TypeScript methods are async; the Python Machine API is synchronous, with an awaitable AsyncMachine (same methods, non-blocking) for driving many machines from one event loop — see AsyncMachine. For the standalone hosted REST API at api.smolmachines.com, see the Cloud API reference.

Complete API reference for the Machine class.

Static Methods

create()

Create and start a machine in one call.

static async create(
  config?: MachineConfig,
  conn?: ConnectOptions,
): Promise<Machine>

// Usage
const machine = await Machine.create({ name: 'my-machine' });

@classmethod
def create(
    cls,
    config: Optional[MachineConfig] = None,
    conn: Optional[ConnectOptions] = None,
) -> Machine

# Usage
machine = Machine.create(MachineConfig(name="my-machine"))

connect()

Attach to an existing machine (created elsewhere) without creating a new one. Locally, re-opens a persisted machine by name; on cloud, looks up the machine by id.

static async connect(
  id: string,
  conn?: ConnectOptions,
): Promise<Machine>

@classmethod
def connect(
    cls,
    machine_id: str,
    conn: Optional[ConnectOptions] = None,
) -> Machine

Properties

name

The machine’s name / identifier (a getter/property, not a method).

get name(): string

@property
def name(self) -> str

Instance Methods

state()

Get the current lifecycle state as a string ("created" | "running" | "stopped").

async state(): Promise<string>

def state(self) -> str

url()

Public ingress URL for the machine’s first published port (cloud). null/None on local, before a host port is allocated, or with no published port.

async url(): Promise<string | null>

def url(self) -> Optional[str]

ready() / readyAt()

Whether the machine is ready to do work, and when it became so (cloud). state() becoming "started" means only that the VM process launched — the guest is still booting and is not usable yet. ready becomes true once the in-VM agent is reachable and any published port accepts connections. Gate on this, not on state. (On the local target a machine reports ready once running.)

async ready(): Promise<boolean>
async readyAt(): Promise<string | null>   // RFC3339, or null if not yet ready

def ready(self) -> bool
def ready_at(self) -> Optional[str]        # RFC3339, or None if not yet ready

waitUntilReady() / wait_until_ready()

Block until the machine is ready (or raise on a failed/stopped state or timeout). create() already waits for readiness, so this is for machines attached via connect(), or to re-assert readiness before use. Poll ceiling and interval are configurable (defaults: 120 s / 1 s).

async waitUntilReady(opts?: {
  timeoutMs?: number;   // default 120000
  intervalMs?: number;  // default 1000
}): Promise<void>

def wait_until_ready(self, timeout_s: float = 120.0, interval_s: float = 1.0) -> None

endpoint() / fetch()

Reach a published guest port through the control plane’s authenticated connect bridge — no tunnel, no public exposure, no egress allow-list (cloud). endpoint() builds the authed URL + headers (pure, no I/O) for your own WebSocket/HTTP client; fetch() (Node) / request() (Python) is a convenience for a one-shot HTTP call. The machine must publish the port (ports at create).

endpoint(port: number, path?: string): {
  httpUrl: string;   // https://…/v1/machines/:id/connect/:port[/path]
  wsUrl: string;     // wss://…  — plug into a WebSocket client with headers
  headers: Record<string, string>;
}

async fetch(port: number, path?: string, init?: RequestInit): Promise<Response>

def endpoint(self, port: int, path: Optional[str] = None) -> PortEndpoint
# PortEndpoint has .http_url, .ws_url, .headers

def request(
    self, port: int, path: Optional[str] = None,
    method: str = "GET", data: Optional[bytes] = None,
) -> bytes

Have the in-VM worker listen on the port and connect inbound — this inverts the fragile “worker dials out to a relay” pattern into a robust “orchestrator dials in”, and doubles as a readiness probe.

exec()

Execute a command (an array of argv) directly in the microVM.

async exec(
  command: string[],
  opts?: ExecOptions
): Promise<ExecResult>

def exec(
    self,
    command: list[str],
    opts: Optional[ExecOptions] = None,
) -> ExecResult

run()

Pull an image (if needed) and run a command in a container of it (local).

async run(
  image: string,
  command: string[],
  opts?: ExecOptions
): Promise<ExecResult>

def run(
    self,
    image: str,
    command: list[str],
    opts: Optional[ExecOptions] = None,
) -> ExecResult

execStream() / exec_stream()

Execute a command and stream stdout/stderr/exit events as they occur (local).

execStream(
  command: string[],
  opts?: ExecOptions
): AsyncGenerator<ExecEvent>

type ExecEvent =
  | { kind: 'stdout'; data: string }
  | { kind: 'stderr'; data: string }
  | { kind: 'exit'; exitCode: number }
  | { kind: 'error'; message: string };

def exec_stream(
    self,
    command: list[str],
    opts: Optional[ExecOptions] = None,
)
# Yields dicts:
#   {"kind": "stdout" | "stderr", "data": str}
#   {"kind": "exit", "exit_code": int}
#   {"kind": "error", "message": str}

readFile() / read_file()

Read a file from the machine as raw bytes.

async readFile(path: string): Promise<Buffer>

def read_file(self, path: str) -> bytes

writeFile() / write_file()

Write a file into the machine, with an optional file mode.

async writeFile(
  path: string,
  data: string | Uint8Array,
  mode?: number
): Promise<void>

def write_file(
    self,
    path: str,
    data: bytes | str,
    mode: Optional[int] = None,
) -> None

pullImage() / pull_image()

Pull an OCI image into the machine’s storage (local).

async pullImage(image: string): Promise<ImageInfo>

def pull_image(self, image: str) -> ImageInfo

listImages() / list_images()

List cached OCI images (local).

async listImages(): Promise<ImageInfo[]>

def list_images(self) -> list[ImageInfo]

stop()

Stop the machine (it can be restarted).

async stop(): Promise<void>

def stop(self) -> None

delete()

Stop the machine and delete its storage (permanent).

async delete(): Promise<void>

def delete(self) -> None

fork()

Fork this running, forkable machine into a new clone via copy-on-write live RAM + disks (cloud). Requires the golden to have been created with forkable: true.

async fork(name: string, ports?: PortSpec[]): Promise<Machine>

def fork(self, name: str, ports: Optional[list[PortSpec]] = None) -> Machine

AsyncMachine (Python)

The Python Machine is synchronous — each call blocks the calling thread. When you’re driving many machines from one event loop (a fleet of disposable workers), use AsyncMachine: the same API, but every I/O method is a coroutine that runs off the loop, so launches and calls overlap instead of serializing. create/connect/exec/wait_until_ready/request/fork/… all have awaitable counterparts, plus async with for auto-delete. endpoint(port) stays synchronous (it only builds a URL).

import asyncio
from smol import AsyncMachine, MachineConfig, ConnectOptions, PortSpec

async def main():
    cfg = MachineConfig(image="alpine:3.20", ports=[PortSpec(host=8080, guest=8080)])
    conn = ConnectOptions(target="cloud")  # or SMOL_CLOUD_TOKEN
    # Launch a fleet concurrently — none blocks the loop.
    machines = await asyncio.gather(*(AsyncMachine.create(cfg, conn) for _ in range(8)))
    try:
        await asyncio.gather(*(m.wait_until_ready() for m in machines))
        health = await machines[0].request(8080, "healthz")  # authed connect bridge, no tunnel
    finally:
        await asyncio.gather(*(m.delete() for m in machines))

asyncio.run(main())

The Node SDK is already fully async, so there is no separate async class there — every Machine method returns a Promise.

Types

MachineConfig

interface MachineConfig {
  name?: string;
  image?: string;          // required for cloud, optional for local
  mounts?: MountSpec[];    // local
  ports?: PortSpec[];      // local
  resources?: ResourceSpec;
  persistent?: boolean;    // local; default false
  autoStopSeconds?: number; // cloud
  ttlSeconds?: number;      // cloud
  forkable?: boolean;       // cloud; default false
}

@dataclass
class MachineConfig:
    name: Optional[str] = None
    image: Optional[str] = None          # required for cloud, optional for local
    mounts: Optional[list[MountSpec]] = None
    ports: Optional[list[PortSpec]] = None
    resources: Optional[ResourceSpec] = None
    persistent: bool = False             # local
    auto_stop_seconds: Optional[int] = None  # cloud
    ttl_seconds: Optional[int] = None        # cloud
    forkable: bool = False               # cloud

ConnectOptions

Selects and configures the backend. Local (embedded) is the default.

interface ConnectOptions {
  target?: 'local' | 'cloud'; // default 'local'
  baseUrl?: string;           // cloud base URL
  apiKey?: string;            // cloud API key, smk_…
}

@dataclass
class ConnectOptions:
    target: Optional[Literal["local", "cloud"]] = None  # default local
    base_url: Optional[str] = None
    api_key: Optional[str] = None  # cloud API key, smk_…

Cloud is selected when target == 'cloud', or when an API key is present (via apiKey/api_key or the SMOL_CLOUD_TOKEN env var) and target is not 'local'. The cloud base URL defaults to https://api.smolmachines.com (override with baseUrl or SMOL_CLOUD_URL).

ExecOptions

interface ExecOptions {
  env?: Record<string, string>;
  workdir?: string;
  timeout?: number;  // seconds
}

@dataclass
class ExecOptions:
    env: Optional[dict[str, str]] = None
    workdir: Optional[str] = None
    timeout: Optional[int] = None  # seconds

ExecResult

interface ExecResult {
  exitCode: number;
  stdout: string;
  stderr: string;
  success: boolean;   // exitCode === 0
  output: string;     // stdout + stderr
  assertSuccess(): void;  // throws ExecutionError if exitCode !== 0
}

@dataclass
class ExecResult:
    exit_code: int
    stdout: str
    stderr: str

    @property
    def success(self) -> bool: ...   # exit_code == 0

    @property
    def output(self) -> str: ...     # stdout + stderr

    def assert_success(self, command="") -> "ExecResult": ...
    # raises ExecutionError if exit_code != 0

MountSpec

interface MountSpec {
  source: string;      // absolute host path
  target: string;      // absolute path in the machine
  readonly?: boolean;  // default true
}

@dataclass
class MountSpec:
    source: str          # absolute host path
    target: str          # absolute path in the machine
    readonly: bool = True

PortSpec

interface PortSpec {
  host: number;
  guest: number;
}

@dataclass
class PortSpec:
    host: int
    guest: int

ResourceSpec

See Resources for defaults and per-target notes.

interface ResourceSpec {
  cpus?: number;
  memoryMb?: number;
  network?: boolean;      // default false
  allowCidrs?: string[];  // cloud
  allowHosts?: string[];  // cloud
  storageGb?: number;     // default 20
  overlayGb?: number;     // default 10
  gpu?: boolean;          // local; default false
  gpuVramMib?: number;    // local
}

@dataclass
class ResourceSpec:
    cpus: Optional[int] = None
    memory_mb: Optional[int] = None
    network: Optional[bool] = None       # default False
    allow_cidrs: Optional[list[str]] = None  # cloud
    allow_hosts: Optional[list[str]] = None  # cloud
    storage_gb: Optional[int] = None     # default 20
    overlay_gb: Optional[int] = None     # default 10
    gpu: Optional[bool] = None           # local; default False
    gpu_vram_mib: Optional[int] = None   # local

ImageInfo

interface ImageInfo {
  reference: string;
  digest: string;
  size: number;          // bytes
  architecture: string;
  os: string;
}

@dataclass
class ImageInfo:
    reference: str
    digest: str
    size: int            # bytes
    architecture: str
    os: str

MachineState

type MachineState = 'created' | 'running' | 'stopped';

# MachineState is an alias for str: "created" | "running" | "stopped"
MachineState = str