Python Client

machgen-client is the official Python client. It wraps the REST API with typed models, automatic uploads of local source files, polling, and pushed status updates over server-sent events (SSE). You can receive updates through a callback or block until the task finishes. Python 3.11 or newer is required.

1. Install

2. Provide your API key

By default the client reads your key (see Get Started) from the MACHGEN_API_KEY environment variable:

export MACHGEN_API_KEY="MGA_<key_id>:<secret>"

from machgen.client import MachGenClient

with MachGenClient() as client:
    ...

Or pass it explicitly - useful for multi-tenant code:

client = MachGenClient(api_key="MGA_<key_id>:<secret>")

3. Examples

Each script submits a task, polls until it completes, and returns the asset locator from result.task_output (keyed by output kind, e.g. image/video).

Refer to the examples below for different task types.

"""Text-to-video (T2V): generate a short clip from a text prompt."""

from __future__ import annotations

import time

from machgen.client import (
    MachGenClient,
    TaskInput,
    TaskOutputType,
    TaskStatus,
    VideoConfig,
)


def run(client: MachGenClient) -> str:
    task = TaskInput(
        prompt="A red panda exploring a misty forest at dawn",
        model="Wan2.2-A14B",
        task_type="T2V",
        video_config=VideoConfig(
            duration_secs=5,
            height=480,
            aspect_ratio="16:9",
            fps=16,
        ),
    )

    handle = client.submit_task(task)

    result = client.get_task_state(handle)
    while result.status not in (TaskStatus.COMPLETED, TaskStatus.FAILED):
        time.sleep(2)
        result = client.get_task_state(handle)
    if result.status == TaskStatus.FAILED:
        raise RuntimeError(f"Generation failed: {result.error_msg}")

    assert result.task_output is not None
    return result.task_output[TaskOutputType.VIDEO]


if __name__ == "__main__":
    with MachGenClient() as client:
        print(run(client))

Waiting for results

submit_task returns a TaskHandle without waiting for completion. Pass on_update to open a background SSE connection: the server pushes status updates over this connection, and the client calls your callback as they arrive.

handle = client.submit_task(task, on_update=lambda state: print(state.status))
result = client.wait(handle, timeout=300.0)

client.wait(handle) uses the same stream and blocks until the task reaches COMPLETED or FAILED. It also opens an SSE connection when no callback is supplied. The default timeout is 300 seconds; exceeding it raises TimeoutError without cancelling the task.

For manual polling, client.get_task_state(handle) fetches the current status once. Repeat that call as needed, as shown in the examples above. Polling alone does not open an SSE connection.

Once COMPLETED, client.download_asset(handle.task_id) returns the raw asset bytes, or read the asset locator from result.task_output (keyed by output kind) directly.

client.get_account() returns your balance together with how many of your tasks are queued and running, if you want to pace submissions against your account's concurrency limit.

API reference

Please refer to the API Reference page.