Python SDK

Use the Dedalus Python SDK to create and manage persistent Linux machines from your application. Create one or many machines for a project, run commands, and reuse their files across sessions.

Your application calls the SDK. Commands submitted through the SDK run inside the machine. You can also change machine resources, control sleep and wake, request SSH access, and inspect usage.

The library provides synchronous and asynchronous clients. Use Dedalus for sequential code. Use AsyncDedalus when your application needs to do other work while it waits for API responses.

Install

Use Python 3.9 or later.

pip install dedalus-sdk

uv add dedalus-sdk

Authenticate

Create an API key in the Dedalus Dashboard, then set it in the environment:

export DEDALUS_API_KEY="your-api-key"

Dedalus() and AsyncDedalus() read DEDALUS_API_KEY automatically.

Create a machine

This creates a machine with 1 vCPU, 4,096 MiB of memory, 10 GiB of storage, and autosleep after five idle minutes:

from dedalus_sdk import Dedalus

client = Dedalus()
machine = client.machines.create(
    vcpu=1,
    memory_mib=4096,
    storage_gib=10,
    autosleep="5m",
)
print(machine.machine_id)

import asyncio

from dedalus_sdk import AsyncDedalus


async def create_machine() -> None:
    async with AsyncDedalus() as client:
        machine = await client.machines.create(
            vcpu=1,
            memory_mib=4096,
            storage_gib=10,
            autosleep="5m",
        )
        print(machine.machine_id)


asyncio.run(create_machine())

Use the saved machine.machine_id in later requests for this machine.

Machine creation is asynchronous. The response contains the requested state and the machine's current lifecycle status. You can submit an execution immediately after creation. Dedalus waits for the machine to start before running the command.

Run a command

An execution is one command run inside a machine. Creating an execution returns an execution_id before the command finishes. Use that ID to check the command's status and retrieve its output.

This example writes a file, reads it back, and waits for the execution to finish:

import time

execution = client.machines.executions.create(
    machine_id=machine.machine_id,
    command=[
        "/bin/sh",
        "-c",
        "printf 'hello from Dedalus\\n' > /root/sdk-example.txt && cat /root/sdk-example.txt",
    ],
    timeout_ms=30_000,
)
finished = {"succeeded", "failed", "cancelled", "expired"}
for attempt in range(120):
    if execution.status in finished:
        break
    time.sleep(0.5)
    execution = client.machines.executions.retrieve(
        machine_id=machine.machine_id,
        execution_id=execution.execution_id,
    )
if execution.status not in finished:
    raise TimeoutError(f"Still waiting for execution {execution.execution_id}")
output = client.machines.executions.output(
    machine_id=machine.machine_id,
    execution_id=execution.execution_id,
)
if execution.status != "succeeded":
    raise RuntimeError(output.stderr or execution.error_message or execution.status)
print(output.stdout or "", end="")

import asyncio


async def run_command(client: AsyncDedalus, machine_id: str) -> None:
    execution = await client.machines.executions.create(
        machine_id=machine_id,
        command=[
            "/bin/sh",
            "-c",
            "printf 'hello from Dedalus\\n' > /root/sdk-example.txt && cat /root/sdk-example.txt",
        ],
        timeout_ms=30_000,
    )
    finished = {"succeeded", "failed", "cancelled", "expired"}
    for attempt in range(120):
        if execution.status in finished:
            break
        await asyncio.sleep(0.5)
        execution = await client.machines.executions.retrieve(
            machine_id=machine_id,
            execution_id=execution.execution_id,
        )
    if execution.status not in finished:
        raise TimeoutError(f"Still waiting for execution {execution.execution_id}")
    output = await client.machines.executions.output(
        machine_id=machine_id,
        execution_id=execution.execution_id,
    )
    if execution.status != "succeeded":
        raise RuntimeError(output.stderr or execution.error_message or execution.status)
    print(output.stdout or "", end="")

Example output:

hello from Dedalus

Execution status is separate from API status. A successful API request can return an execution whose status is failed. Check the execution status before using its output.

timeout_ms limits how long the command can run inside the machine. The execution continues independently of your application's polling loop until it finishes, is cancelled, expires, or reaches that timeout. See Executions for working directories, environment variables, input, cancellation, and output limits.

Reuse the machine

Dedalus Machines are persistent. Save the machine ID to reuse the same filesystem across application sessions.

In a later application session, set MACHINE_ID to the saved ID and retrieve it:

import os

from dedalus_sdk import Dedalus

resumed_client = Dedalus()
saved_machine = resumed_client.machines.retrieve(machine_id=os.environ["MACHINE_ID"])

The returned saved_machine contains the machine's current state.

Submit another execution with saved_machine.machine_id to continue working. Executions wake a sleeping machine automatically. Sleep preserves files on the root filesystem and resets processes and memory. See persistence for the complete lifecycle contract.

When you no longer need the machine or its files, explicitly request deletion:

client.machines.delete(machine_id=machine.machine_id)

Deletion removes the machine and its files. Sleep and autosleep preserve the machine for another session.

Watch lifecycle changes

watch streams machine state changes using server-sent events (SSE).

desired_state is the requested state. status.phase is the machine's current state. These examples request sleep and read updates until the machine is sleeping:

client.machines.sleep(machine_id=machine.machine_id)

with client.machines.watch(machine_id=machine.machine_id) as stream:
    for update in stream:
        print(update.status.phase)
        if update.status.phase == "sleeping":
            break

async def sleep_machine(client: AsyncDedalus, machine_id: str) -> None:
    await client.machines.sleep(machine_id=machine_id)
    stream = await client.machines.watch(machine_id=machine_id)
    async with stream:
        async for update in stream:
            print(update.status.phase)
            if update.status.phase == "sleeping":
                break

Each update contains a complete machine response. The context manager closes the open stream when the loop exits.

The stream closes when the machine reaches its desired state, reports a terminal error, or the five-minute watch limit expires. Retrieve the machine after the stream closes to confirm its current state:

current = client.machines.retrieve(machine_id=machine.machine_id)
if current.status.phase != "sleeping":
    raise RuntimeError(f"Sleep not complete: {current.status.phase}")

List machines

Machine lists are paginated. limit sets the number of machines requested in each page, and cursor continues from a previous page.

Use iteration to fetch pages automatically:

for item in client.machines.list(limit=20):
    print(item.machine_id)

async def list_machines(client: AsyncDedalus) -> None:
    async for item in client.machines.list(limit=20):
        print(item.machine_id)

To request one page at a time, inspect items and call get_next_page() only when you want another request:

first_page = client.machines.list(limit=20)
for item in first_page.items:
    print(item.machine_id)
if first_page.has_next_page():
    second_page = first_page.get_next_page()
    print(second_page.items)

With the asynchronous client, await client.machines.list(...) to get a page and await get_next_page() to request the next one.

Configure requests

Constructor options apply to every request. Use client.with_options(...) to create a client with overrides for a particular call without changing the original client.

This example disables retries for the client and enables two retries for one request:

configured_client = Dedalus(max_retries=0, timeout=20.0)
request_client = configured_client.with_options(max_retries=2, timeout=5.0)
configured_machine = request_client.machines.retrieve(machine_id=machine.machine_id)
print(configured_machine.machine_id)

Retries and timeouts

The SDK retries connection failures, HTTP 408, 409, 429, and 5xx responses twice by default, with increasing delays between attempts. An explicit x-should-retry response header overrides the status-based decision.

A timeout raises APITimeoutError and can also be retried. HTTPX timeouts limit individual network operations, not the total duration of a call including retries.

SDK methods

The SDK groups related methods under resources. All methods accept keyword arguments. Methods that act on an existing machine require machine_id. The asynchronous client exposes the same methods with await.

Use the API reference for every request field and response field.

Handle errors

An API request and a command can fail independently. A successful API request to executions.retrieve can describe a failed command: check execution.status, exit_code, error_message, and captured stderr before using its results.

After an HTTP timeout, use the execution ID to retrieve the existing execution. Submitting another command creates another execution. The client timeout controls HTTP waiting, execution timeout_ms controls command runtime, and machine autosleep controls idle compute.

Connection failures raise APIConnectionError. Unsuccessful HTTP responses raise APIStatusError with status_code, response, and body. Both inherit from APIError.

import dedalus_sdk

try:
    client.machines.retrieve(machine_id=machine.machine_id)
except dedalus_sdk.APIConnectionError as error:
    print("The server could not be reached:", error.__cause__)
    raise
except dedalus_sdk.APIStatusError as error:
    print("Request failed with HTTP status:", error.status_code)
    print(error.body)
    raise

Other unsuccessful status codes use the base APIStatusError type.

Advanced client options

Work with types

Nested request parameters use TypedDict. Responses are Pydantic models with named fields, editor autocomplete, and methods to convert them to JSON (to_json()) or a dictionary (to_dict()).

A missing response field and an explicit JSON null both appear as None. Check whether the field name belongs to model_fields_set to distinguish them. If it is absent from that set, the API omitted the field.

Inspect HTTP responses

Use .with_raw_response to receive the HTTP status and headers with the parsed result:

response = client.machines.with_raw_response.retrieve(machine_id=machine.machine_id)
print(response.status_code)
print(response.parse().machine_id)

Use .with_streaming_response when you need to read the raw response body as it arrives:

import sys

with client.machines.with_streaming_response.retrieve(
    machine_id=machine.machine_id,
) as streaming_response:
    for chunk in streaming_response.iter_bytes():
        sys.stdout.buffer.write(chunk)

The asynchronous client returns AsyncAPIResponse. Await the request and use async for to read iter_bytes().

Logging

The SDK uses Python's standard logging module. Set DEDALUS_LOG=info for diagnostic logging or DEDALUS_LOG=debug for more detail.

Customize HTTP requests

The clients use HTTPX by default. Pass DefaultHttpxClient through http_client to configure proxies, transports, or connection pools. Request-specific options passed through with_options(...) override client options.

Use aiohttp with the asynchronous client

Install dedalus-sdk[aiohttp] and pass DefaultAioHttpClient() through http_client to use aiohttp instead of HTTPX.

Requests beyond the generated types

client.get, client.post, and the other HTTP methods let you call endpoints that do not have resource methods. Pass cast_to to choose the response type. Use extra_query, extra_body, or extra_headers for additional request fields.

Close HTTP connections

Use with Dedalus() as client or async with AsyncDedalus() as client to close connections when the block ends. Long-lived applications can keep a client open across requests, then call client.close() or await client.close() during shutdown.

Closing the SDK client releases your application's HTTP connections.

Versioning

Review the release notes before upgrading. If you depend on an internal interface, or find a compatibility issue, report it in the SDK issue tracker.