Celesto Python class reference

Celesto(
    config: VMConfig | None = None,
    *,
    vm_id: str | None = None,
    image: str | None = None,
    data_dir: Path | None = None,
    socket_dir: Path | None = None,
    backend: str | None = None,
    os: GuestOS | str | None = None,
    qemu_machine: Literal["auto", "microvm", "q35"] = "auto",
    memory: int | None = None,
    disk_size: int | None = None,
    ssh_user: str = "root",
    ssh_key_path: str | None = None,
    ssh_password: str | None = None,
    comm_channel: Literal["ssh", "vsock"] | None = None,
    internet_settings: InternetSettings | dict | None = None,
    mounts: list[str] | None = None,
    writable_mounts: bool = False,
    callbacks: list[Callback] | None = None,
    state_manager: StateManagerProtocol | None = None,
)

@classmethod
Celesto.from_id(
    vm_id: str,
    *,
    data_dir: Path | None = None,
    socket_dir: Path | None = None,
    backend: str | None = None,
    ssh_user: str = "root",
    ssh_key_path: str | None = None,
    ssh_password: str | None = None,
    comm_channel: Literal["ssh", "vsock"] | None = None,
    state_manager: StateManagerProtocol | None = None,
) -> Celesto

@classmethod
Celesto.from_image(
    image: BootImage,
    *,
    vm_id: str | None = None,
    name_prefix: str = "sbx",
    data_dir: Path | None = None,
    socket_dir: Path | None = None,
    state_manager: StateManagerProtocol | None = None,
    backend: str | None = None,
    arch: str | None = None,
    qemu_machine: Literal["auto", "microvm", "q35"] = "auto",
    vcpus: int = 1,
    memory_mb: int = 512,
    guest_os: GuestOS | str = GuestOS.ALPINE,
    network: Literal["tap", "slirp"] | None = None,
    port_forwards: list[PortForwardConfig | dict] | None = None,
    vsock: VsockConfig | dict | None = None,
    comm_channel: Literal["ssh", "vsock"] | None = None,
    disk_mode: Literal["isolated", "shared"] = "isolated",
    disk_size_mb: int | None = None,
    grow_filesystem: bool = False,
    ssh_user: str = "root",
    ssh_key_path: str | None = None,
    ssh_password: str | None = None,
    internet_settings: InternetSettings | dict | None = None,
    mounts: list[str] | None = None,
    writable_mounts: bool = False,
    callbacks: list[Callback] | None = None,
) -> Celesto

from celesto import Celesto, DirectKernelBoot, DockerRootfsBuilder

builder = DockerRootfsBuilder(
    name="my-app",
    dockerfile="""
        FROM alpine:3.20
        RUN apk add --no-cache python3
        COPY init /init
        RUN chmod +x /init
    """,
    context={
        "init": "#!/bin/sh\nwhile true; do sleep 3600; done\n",
    },
)

image = builder.ensure(
    backend="qemu",
    arch="host",
    boot=DirectKernelBoot(root="/dev/vda", init="/init"),
)

with Celesto.from_image(image, memory_mb=1024, vcpus=2) as vm:
    print(f"Started {vm.vm_id}")

@classmethod
Celesto.browser(
    *,
    headless: bool = True,
    session_id: str | None = None,
    backend: Literal["firecracker", "qemu", "libkrun", "auto"] = "auto",
    profile_id: str | None = None,
    persistent: bool = False,
    timeout_minutes: int = 30,
    viewport: BrowserViewport | dict[str, Any] | None = None,
    viewport_width: int = 1280,
    viewport_height: int = 720,
    record_video: bool = False,
    allow_downloads: bool = True,
    env_vars: dict[str, str] | None = None,
    workspace_mounts: list[WorkspaceMount] | None = None,
    memory_mb: int = 2048,
    disk_size_mb: int = 4096,
    data_dir: Path | None = None,
    socket_dir: Path | None = None,
    ssh_key_path: str | None = None,
    boot_timeout: float = 90.0,
    on_progress: Callable[[str], None] | None = None,
) -> DisplaySandboxProtocol

@classmethod
Celesto.desktop(
    *,
    session_id: str | None = None,
    backend: Literal["firecracker", "qemu", "libkrun", "auto"] = "auto",
    profile_id: str | None = None,
    persistent: bool = False,
    timeout_minutes: int = 30,
    viewport: BrowserViewport | dict[str, Any] | None = None,
    viewport_width: int = 1280,
    viewport_height: int = 720,
    record_video: bool = False,
    allow_downloads: bool = True,
    env_vars: dict[str, str] | None = None,
    workspace_mounts: list[WorkspaceMount] | None = None,
    memory_mb: int = 2048,
    disk_size_mb: int = 4096,
    data_dir: Path | None = None,
    socket_dir: Path | None = None,
    ssh_key_path: str | None = None,
    boot_timeout: float = 90.0,
    on_progress: Callable[[str], None] | None = None,
) -> DisplaySandboxProtocol

@classmethod
Celesto.from_snapshot(
    snapshot_id: str,
    *,
    data_dir: Path | None = None,
    socket_dir: Path | None = None,
    backend: str | None = None,
    resume_vm: bool = False,
    force: bool = False,
    ssh_user: str = "root",
    ssh_key_path: str | None = None,
    comm_channel: Literal["ssh", "vsock"] | None = None,
    state_manager: StateManagerProtocol | None = None,
) -> Celesto

def start(self, boot_timeout: float = 30.0) -> Celesto

def stop(self, timeout: float = 3.0) -> Celesto

def delete(self) -> None

def pause(self) -> Celesto

def resume(self) -> Celesto

def snapshot(
    self,
    snapshot_id: str | None = None,
    *,
    snapshot_type: SnapshotType | str = SnapshotType.FULL,
    resume_source: bool = False,
) -> SnapshotInfo

def run(
    self,
    command: str,
    timeout: int = 30,
    shell: Literal["login", "raw"] = "login",
) -> CommandResult

def add_callback(self, callback: Callback) -> Celesto

def wait_for_ssh(self, timeout: float = 60.0) -> Celesto

def ssh_commands(
    self,
    *,
    ssh_user: str | None = None,
    key_path: str | Path | None = None,
    public_host: str | None = None,
) -> dict[str, str]

def set_env_vars(
    self,
    env_vars: dict[str, str],
    *,
    merge: bool = True
) -> list[str]

def unset_env_vars(self, keys: list[str]) -> dict[str, str]

def list_env_vars(self) -> dict[str, str]

def expose_local(
    self,
    guest_port: int,
    host_port: int | None = None,
    *,
    guest_loopback: bool = False,
) -> int

def unexpose_local(self, host_port: int, guest_port: int) -> Celesto

@property
def vm_id(self) -> str

@property
def info(self) -> VMInfo

@property
def status(self) -> VMState

@property
def data_dir(self) -> Path

def get_ip(self) -> str

def refresh(self) -> Celesto

def can_run_commands(self) -> bool

def close(self) -> None

async def async_start(self, boot_timeout: float = 30.0) -> Celesto

async def async_stop(self, timeout: float = 3.0) -> Celesto

async def async_run(
    self,
    command: str,
    timeout: int = 30,
    shell: Literal["login", "raw"] = "login",
) -> CommandResult

async def async_wait_for_ssh(self, timeout: float = 60.0) -> Celesto

@classmethod
async def async_create_many(
    cls,
    count: int,
    *,
    memory: int | None = None,
    disk_size: int | None = None,
    backend: str | None = None,
) -> list[Celesto]

import asyncio
from celesto import Celesto

async def main():
    vms = await Celesto.async_create_many(3)
    for vm in vms:
        result = await vm.async_run("hostname")
        print(result.stdout.strip())
    for vm in vms:
        await vm.async_stop()
        vm.delete()

asyncio.run(main())

import asyncio
from celesto import Celesto

async def main():
    async with Celesto() as vm:
        result = await vm.async_run("echo 'Hello from async'")
        print(result.stdout.strip())

asyncio.run(main())

with Celesto() as vm:
    # VM auto-starts on context entry
    result = vm.run("uname -r")
    print(result.stdout)
# VM auto-stops and auto-deletes on context exit

from celesto import Celesto

with Celesto() as vm:
    result = vm.run("echo 'Hello from Celesto'")
    print(result.stdout.strip())

from celesto import Celesto

with Celesto(memory=1024, disk_size=1024) as vm:
    result = vm.run("free -m")
    print(result.stdout)

from celesto import Celesto, VMConfig
from pathlib import Path

config = VMConfig(
    vm_id="my-vm",
    vcpu_count=2,
    memory=512,
    kernel_path=Path("/path/to/kernel"),
    rootfs_path=Path("/path/to/rootfs"),
    boot_args="console=ttyS0 reboot=k panic=1 init=/init"
)

vm = Celesto(config)
vm.start()

result = vm.run("hostname")
print(result.stdout)

vm.stop()
vm.delete()
vm.close()

from celesto import Celesto

with Celesto(
    os="windows",
    image="~/win11-vm/disk-baseline.qcow2",
    ssh_user="celesto",
    ssh_password="celesto",
    memory=4096,
) as vm:
    vm.wait_for_ssh()

    result = vm.run("Write-Output 'hello from windows'")
    print(result.stdout)  # 'hello from windows\r\n'

    vm.upload_file("./hello.ps1", "C:\\Users\\celesto\\hello.ps1")

from celesto import Celesto

from pathlib import Path
from celesto.storage import MemoryStateManager

inventory = MemoryStateManager(Path(".smolvm-state"))
original = Celesto(state_manager=inventory)
original.start()
vm_id = original.vm_id
original.close()

# Reuse the same inventory when creating a new handle.
vm = Celesto.from_id(vm_id, state_manager=inventory)
result = vm.run("uptime")
print(result.stdout)
vm.close()

from celesto import Celesto

with Celesto() as vm:
    # Install and start a web server
    vm.run("apk add python3")
    vm.run("python3 -m http.server 8000 &")
    
    # Expose guest port 8000 on localhost
    host_port = vm.expose_local(guest_port=8000)
    print(f"Server available at http://localhost:{host_port}")
    
    # Access the server from host
    # ...
    
    # Clean up port forward
    vm.unexpose_local(host_port=host_port, guest_port=8000)

from celesto import Celesto

vm = Celesto()
vm.start()

# Run some setup
vm.run("echo 'setup complete'")

# Pause the VM to save resources
vm.pause()

# Resume when you need it again
vm.resume()
result = vm.run("echo 'back online'")
print(result.stdout.strip())

vm.stop()
vm.delete()
vm.close()

from celesto import Celesto

with Celesto() as vm:
    # Set environment variables
    vm.set_env_vars({"API_KEY": "secret", "ENV": "production"})
    
    # List current variables
    env_vars = vm.list_env_vars()
    print(env_vars)  # {'API_KEY': 'secret', 'ENV': 'production'}
    
    # Variables persist across sessions
    result = vm.run("echo $API_KEY")
    print(result.stdout.strip())  # secret
    
    # Remove variables
    removed = vm.unset_env_vars(["API_KEY"])
    print(removed)  # {'API_KEY': 'secret'}