VMs

solari-desktop: classes DesktopClient and Desktop. A VM is a GUI desktop you drive with mouse, keyboard, and screenshots, and stream live. See the Python SDK hub for install and configuration.

pip install solari-desktop

DesktopClient: DesktopClient(), create(), connect(), get(), pause(), resume(), attach(), destroy(), aclose(), volumes
Desktop: health(), screenshot(), mouse.*, keyboard.*, display.*, clipboard.*, open(), stream.*, record.*, process.*, ports.*, pkg.install(), exec(), exec_stream(), fs.*, inherited members
Types: CreateDesktopResponse, GetDesktopResponse, DesktopLifecycleResponse, DeleteDesktopResponse, DesktopStatus, HealthResult, ExecResult, ExecStreamChunk, ProcessInfo, PortInfo, PkgInstallResult

Result dataclasses keep camelCase fields

Method arguments are idiomatic snake_case (mem_mb, timeout_ms), but the dataclasses they return mirror the wire and stay camelCase: result.exitCode, view.memMb, stat.modTimeMs. This applies to all of solari-desktop.

DesktopClient

The VM entry point. Talks the gateway HTTP API.

Properties

volumes VolumeClient: persistent volume CRUD. Same surface as SandboxClient.volumes.

Constructors

DesktopClient()

DesktopClient(
    *,
    api_key: str,
    base_url: str,
    http: httpx.AsyncClient | None = None,
    call_timeout_ms: int | None = None,
) -> DesktopClient

Creates a client. Does not open any connection. All arguments are keyword-only.

Parameters:

api_key str: required.
base_url str: required, e.g. https://api.getsolari.com.
http httpx.AsyncClient | None: reuse an existing client; otherwise one is created and owned here.
call_timeout_ms int | None: per-RPC timeout for handles this client creates. Default 300_000.

Returns: DesktopClient

Raises: SolariError if api_key or base_url is missing.

Example:

from solari_desktop import DesktopClient

client = DesktopClient(
    api_key="slr_live_...",
    base_url="https://api.getsolari.com",
)

# also an async context manager, which calls aclose() on exit
async with DesktopClient(api_key="slr_live_...", base_url="...") as client:
    ...

A synchronous flavour exists

SyncDesktopClient takes the same arguments (minus http) and drives a private event loop, so it works outside asyncio. The Desktop handle it returns is still async.

Methods

create()

async create(
    *,
    template: str = "default",
    ttl_seconds: int | None = None,
    resolution: str | None = None,
    cpu: int | None = None,
    mem_mb: int | None = None,
    metadata: dict[str, str] | None = None,
    record: bool | None = None,
    timeout_ms: int | None = None,
    lifecycle: dict | None = None,
    volumes: list | None = None,
) -> Desktop

Creates a VM and returns a live handle. The call sends an idempotency key, so retrying is safe.

Parameters:

template str: e.g. "office". Default "default".
resolution str | None: initial display size, e.g. "1280x720".
cpu int | None: vCPUs, 1 to 16. Grown on assign via vCPU hot-add.
mem_mb int | None: RAM in MiB, 2048 to 65536. Grown via virtio-mem hotplug.
record bool | None: record server-side. See the caveat below, because the Python SDK does not hand you the presigned playback URL.
timeout_ms int | None: rolling idle window; resets on every use. Overrides ttl_seconds and the 30-minute default.
lifecycle dict | None: idle policy with camelCase wire keys, e.g. {"onTimeout": "pause", "autoResume": True}.
metadata dict[str, str] | None: opaque labels.
volumes list | None: attachments, e.g. [{"volumeId": "vol_x", "path": "/data"}].
ttl_seconds int | None: legacy TTL; prefer timeout_ms.

Returns: Desktop

Raises: AuthError, PlanError, ConcurrencyLimitError, NoCapacityError, or GatewayError.

Example:

vm = await client.create(
    template="office",
    resolution="1280x720",
    cpu=2,
    mem_mb=4096,
)

record=True gives you no playback URL here

The gateway returns a presigned recordingUrl on the 201, but create() returns a Desktop handle, which drops it. get() never populates GetDesktopResponse.recordingUrl either, so it is always None. For a retrievable capture use record.start() / record.stop() plus download_url(), which write an mp4 inside the guest.

connect()

async connect(session_id: str) -> Desktop

Re-attaches by id, resuming the session first if it is paused. The canonical way back to a VM when all you have is its id.

Parameters:

session_id str: the session id.

Returns: Desktop

Raises: GatewayError if the id is unknown.

Example:

vm = await client.connect(session_id)  # resumes if paused
await vm.keyboard.type("back again")

get()

async get(session_id: str) -> GetDesktopResponse

Fetches a session’s current status. Returns the raw record, not a handle. The gateway does not re-issue control URLs here.

Parameters:

session_id str: the session id.

Returns: GetDesktopResponse: sessionId, status, expiresAt.

Example:

st = await client.get(session_id)
print(st.status)  # "ready" | "paused" | ...

pause()

async pause(session_id: str) -> DesktopLifecycleResponse

Saves full RAM + disk state so it resumes exactly where it left off. Billing for compute stops and it stops counting against your concurrency limit. Prefer Desktop.pause() when you hold a live handle.

Parameters:

session_id str: the session id.

Returns: DesktopLifecycleResponse: sessionId, status.

Example:

await client.pause(session_id)

resume()

async resume(session_id: str) -> Desktop

Restores full RAM + disk state and returns a new handle. Always resumes. Use connect() if the session may already be running.

Parameters:

session_id str: the session id.

Returns: Desktop. Its expiresAt is ""; call get() for the new deadline.

Example:

vm = await client.resume(session_id)

attach()

attach(session: CreateDesktopResponse) -> Desktop

Rebuilds a handle from a saved create response. Synchronous, with no network call. Use it to carry a session across processes.

Parameters:

session CreateDesktopResponse: sessionId, streamUrl, controlUrl, expiresAt.

Returns: Desktop

Example:

import json
from solari_desktop import CreateDesktopResponse

with open("session.json") as f:
    vm = client.attach(CreateDesktopResponse(**json.load(f)))

destroy()

async destroy(session_id: str) -> DeleteDesktopResponse

Destroys a session. Idempotent.

Parameters:

session_id str: the session id.

Returns: DeleteDesktopResponse: ok.

Example:

await client.destroy(session_id)

aclose()

async aclose() -> None

Closes the underlying HTTP clients, if this client owns them. Does not touch live sessions.

Returns: None

Example:

await client.aclose()

volumes

Identical to SandboxClient.volumes: create(), list(), get(), delete().

Example:

vol = await client.volumes.create(name="assets")
vm = await client.create(
    template="office",
    volumes=[{"volumeId": vol["volumeId"], "path": "/data"}],
)

Desktop

A live GUI session. Construct via DesktopClient.create().

Properties

id str: the session id.
sessionId str: alias of id.
streamUrl str: wss:// URL serving RFB (VNC) bytes for the live view.
controlUrl str: wss:// JSON-RPC control channel.
expiresAt str: ISO 8601 expiry.
connected bool: whether the control channel is open.

Desktop extends SessionHandle

Commands, code, files, git, snapshots, pause/resume, and the rest of the shared surface are documented once on the Sandboxes page. This page covers only the GUI additions. See inherited members for the index.

Methods

health()

async health() -> HealthResult

Readiness probe for the display, VNC server, and guest agent.

Returns: HealthResult: ready, display, vnc.

Example:

h = await vm.health()
if not h.ready:
    raise RuntimeError("desktop not ready")

screenshot()

async screenshot(*, format: "png" | "jpeg" = "png", quality: int | None = None) -> bytes

Captures the current screen as decoded image bytes.

Parameters:

format "png" | "jpeg": default "png".
quality int | None: JPEG quality 1 to 100; ignored for PNG.

Returns: bytes

Example:

with open("shot.png", "wb") as f:
    f.write(await vm.screenshot())

jpeg = await vm.screenshot(format="jpeg", quality=80)

mouse.*

async mouse.move(x: int, y: int, *, humanize: bool | None = None) -> None
async mouse.click(x: int, y: int, *, button: MouseButton | None = None, humanize: bool | None = None) -> None
async mouse.double_click(x: int, y: int, *, button: MouseButton | None = None) -> None
async mouse.down(x: int, y: int, button: MouseButton = "left") -> None
async mouse.up(x: int, y: int, button: MouseButton = "left") -> None
async mouse.scroll(x: int, y: int, *, button: MouseButton | None = None, humanize: bool | None = None) -> None
async mouse.drag(frm: dict, to: dict, button: MouseButton = "left") -> None

Drive the pointer at absolute screen coordinates. The origin is the top-left of the display.

Parameters:

x, y int: absolute coordinates.
button MouseButton: "left", "middle", or "right". The SDK maps the name to the X11 code the guest expects.
humanize bool | None: use a humanized trajectory instead of teleporting.
frm, to dict: drag endpoints, each {"x": int, "y": int}.

Returns: None

Example:

await vm.mouse.move(400, 300, humanize=True)
await vm.mouse.click(400, 300)
await vm.mouse.double_click(120, 80)
await vm.mouse.scroll(400, 300)
await vm.mouse.drag({"x": 100, "y": 100}, {"x": 500, "y": 400})

keyboard.*

async keyboard.type(text: str) -> None
async keyboard.press(keys: str | list[str]) -> None
async keyboard.hotkey(*keys: str) -> None
async keyboard.down(keys: str | list[str]) -> None
async keyboard.up(keys: str | list[str]) -> None

Type literal text, or press key chords. hotkey() is press() with variadic arguments.

Parameters:

text str: literal text to type.
keys str | list[str]: key names, e.g. "Return" or ["ctrl", "c"].

Returns: None

Example:

await vm.keyboard.type("hello world")
await vm.keyboard.press("Return")
await vm.keyboard.hotkey("ctrl", "c")

await vm.keyboard.down("shift")   # hold
await vm.keyboard.press("Left")
await vm.keyboard.up("shift")

display.*

async display.set(w: int, h: int) -> None
async display.size() -> dict   # {"w": int, "h": int}
async display.cursor() -> dict # {"x": int, "y": int}

Read or change the display resolution and read the cursor position.

Parameters:

w, h int: new resolution in pixels.

Returns: None / dict / dict, the two getters return raw dicts, not dataclasses.

Example:

await vm.display.set(1920, 1080)
size = await vm.display.size()      # {"w": 1920, "h": 1080}
cursor = await vm.display.cursor()  # {"x": 400, "y": 300}

clipboard.*

async clipboard.get() -> str
async clipboard.set(text: str) -> None

Read or write the guest clipboard. get() returns "" when the clipboard is empty.

Parameters:

text str: content to place on the clipboard.

Returns: str / None

Example:

await vm.clipboard.set("pasted from the host")
await vm.keyboard.hotkey("ctrl", "v")
print(await vm.clipboard.get())

open()

async open(name: str, args: list[str] | None = None) -> int

Launches a GUI application by name and returns its pid.

Parameters:

name str: executable name, e.g. "firefox".
args list[str] | None: arguments.

Returns: int, the pid.

Example:

pid = await vm.open("firefox", ["https://example.com"])

stream.*

async stream.start() -> dict  # {"streamUrl": str}
async stream.stop() -> None

Returns the embeddable live-view URL. The stream URL is minted at create time, so start() is a getter and stop() is a no-op. The RFB socket belongs to the caller.

Returns: dict, only streamUrl, the same value as vm.streamUrl / None.

Example:

s = await vm.stream.start()
print(s["streamUrl"])  # hand it to a noVNC client

record.*

async record.start(fps: int | None = None, format: str | None = None, path: str | None = None) -> dict
async record.stop() -> dict

Records the session to an mp4 inside the guest. Retrieve it afterwards with download_url().

Parameters:

fps int | None: capture frame rate.
format, path str | None: output format and in-guest path.

Returns: dict, raw RPC results, not dataclasses. start() carries path and fps; stop() carries path and sizeBytes.

Example:

await vm.record.start(fps=15)
await vm.keyboard.type("recorded work")
result = await vm.record.stop()

dl = await vm.download_url(result["path"])

process.*

async process.list() -> list[ProcessInfo]
async process.start(cmd: str, *, args: list[str] | None = None, cwd: str | None = None) -> int
async process.kill(pid: int) -> None
async process.signal(pid: int, signal: int | None = None) -> None

Inspect and control guest processes by pid. For output capture use commands.run() instead.

Parameters:

cmd str: program to start.
args list[str] | None, cwd str | None.
pid int: target process.
signal int | None: signal number.

Returns: list[ProcessInfo] / int (the pid) / None

Example:

pid = await vm.process.start("xterm")
for p in await vm.process.list():
    print(p.pid, p.name)
await vm.process.kill(pid)

ports.*

async ports.list() -> list[PortInfo]

Lists TCP sockets listening inside the guest.

Returns: list[PortInfo], each with port, addr, optional pid.

Example:

for p in await vm.ports.list():
    print(p.port, p.addr, p.pid)

pkg.install()

async pkg.install(manager: "apt" | "pip" | "npm", packages: list[str]) -> PkgInstallResult

Installs packages in the guest.

Parameters:

manager PackageManager: "apt", "pip", or "npm".
packages list[str]: package names.

Returns: PkgInstallResult: exitCode, stdout, stderr. A failed install returns a non-zero exitCode; it does not raise.

Example:

r = await vm.pkg.install("apt", ["imagemagick"])
if r.exitCode != 0:
    print(r.stderr)

exec()

async exec(
    cmd: str,
    *,
    args: list[str] | None = None,
    cwd: str | None = None,
    timeout_ms: int | None = None,
    stream: bool | None = None,
) -> ExecResult

Runs a command to completion. A v1 convenience. Prefer commands.run(), which adds streaming callbacks plus per-command env and user.

Parameters:

cmd str: program to run.
args list[str] | None, cwd str | None, timeout_ms int | None: passed to the guest.

Returns: ExecResult: exitCode, stdout, stderr.

Example:

r = await vm.exec("xdotool", args=["getactivewindow"])

exec_stream()

async exec_stream(
    cmd: str,
    on_chunk: Callable[[ExecStreamChunk], None],
    *,
    args: list[str] | None = None,
    cwd: str | None = None,
    timeout_ms: int | None = None,
) -> ExecResult

Runs a command, delivering output chunks as they arrive. A v1 convenience. Prefer commands.run() with on_stdout.

Parameters:

cmd str: program to run.
on_chunk Callable[[ExecStreamChunk], None]: called with stream, text, bytes. Positional, not keyword.
args, cwd, timeout_ms: as exec().

Returns: ExecResult

Example:

import sys

await vm.exec_stream(
    "apt-get",
    lambda c: sys.stdout.write(c.text),
    args=["update"],
)

fs.*

async fs.read(path) / fs.read_text(path) / fs.write(path, data, mode=None)
async fs.list(path) / fs.stat(path) / fs.remove(path, recursive=False) / fs.mkdir(path)

A v1 alias for a subset of files.*, which is the canonical surface and has more methods.

Example:

await vm.fs.write("/tmp/note.txt", "hi")  # same as vm.files.write(...)

Inherited members

Desktop extends the same SessionHandle base as Sandbox. These all work on a VM:

Member	Purpose
`connect()`, `reconnect()`, `close()`	Control-channel lifecycle.
`commands.run()`, `commands.start()`, `pty.create()`	Run commands and open PTYs.
`run_code()`, `create_code_context()`	Stateful kernel execution with rich outputs.
`files.*`, `download_url()`, `upload_url()`	Filesystem access and signed transfer URLs.
`git.*`	Clone, status, commit, push, pull, checkout, log.
`env()`, `volumes.mount()`	Session env vars and volume mounts.
`metrics()`, `snapshot()`, `revert()`	Usage, checkpoints, in-place restore.
`pause()`, `resume()`, `set_timeout()`, `kill()`	Session lifecycle.
`preview_url()`	Public URL for an in-guest port.

Types

CreateDesktopResponse

sessionId str: signed, opaque session capability.
streamUrl str: wss:// RFB stream.
controlUrl str: wss:// JSON-RPC control channel.
expiresAt str: ISO 8601 expiry.
recordingUrl str | None: parsed from the 201 when record=True. Only reachable if you build the dataclass yourself; create() consumes the response and returns a Desktop.

GetDesktopResponse

sessionId str.
status DesktopStatus: starting, ready, paused, releasing, gone.
expiresAt str.
recordingUrl str | None: declared on the dataclass but never filled in by get(); always None in practice.

DesktopLifecycleResponse / DeleteDesktopResponse

DesktopLifecycleResponse: sessionId str, status DesktopStatus.
DeleteDesktopResponse: ok bool.

HealthResult

ready bool, display bool, vnc bool.

ExecResult / ExecStreamChunk

ExecResult: exitCode int, stdout str, stderr str.
ExecStreamChunk: stream "stdout" | "stderr", text str (decoded UTF-8), bytes bytes (raw).

ProcessInfo / PortInfo / PkgInstallResult

ProcessInfo: pid int, name str, cmd str | None.
PortInfo: port int, addr str, pid int | None.
PkgInstallResult: exitCode int, stdout str, stderr str.

Errors are shared across products. See Errors.

VMs

Contents

DesktopClient

Properties

Constructors

DesktopClient()

Methods

create()

connect()

get()

pause()

resume()

attach()

destroy()

aclose()

volumes

Desktop

Properties

Methods

health()

screenshot()

mouse.*

keyboard.*

display.*

clipboard.*

open()

stream.*

record.*

process.*

ports.*

pkg.install()

exec()

exec_stream()

fs.*

Inherited members

Types

CreateDesktopResponse

GetDesktopResponse

DesktopLifecycleResponse / DeleteDesktopResponse

HealthResult

ExecResult / ExecStreamChunk

ProcessInfo / PortInfo / PkgInstallResult