Client Library

Python Client

Submit and track tasks with the official MachGen Python client.

machgen-client is the official Python client. It wraps the REST API with typed models, automatic uploads of local source files, and blocking or streaming waits. Python 3.11 or newer is required.

1. Install

pip install machgen-client
uv add machgen-client

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>")

Required

If no key is supplied and MACHGEN_API_KEY is unset, the constructor raises a ValueError immediately - the client never sends unauthenticated requests.

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).

Tasks with input images

Some task types may involve input images like I2V (with optional last frame) and R2V. Such tasks expect the field src_image_urls to be specified, which supports:

  • local paths like foo/bar.png - they are uploaded automatically on submission
  • http(s):// URLs - they must be publicly accessible

Refer to the examples below for different task types.

text_to_video.py
"""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))
image_to_video.py
"""Image-to-video (I2V): animate a still image into a short clip."""

from __future__ import annotations

import time

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


def run(client: MachGenClient) -> str:
    task = TaskInput(
        prompt="slow dolly-in as the leaves drift in the wind",
        model="Vidu-Q3-Turbo",
        task_type="I2V",
        src_image_urls=["https://example.com/first-frame.png"],
        video_config=VideoConfig(duration_secs=5, height=720),
    )

    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))
reference_to_video.py
"""Reference-to-video (R2V): drive a video from named subject references.

``subject_to_image_ids`` maps a subject name to the indices of its reference
images in ``src_image_urls``. The prompt can then address a subject via
``@name`` (honored by vendors with named subjects, e.g. Vidu reference2video).
"""

from __future__ import annotations

import time

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


def run(client: MachGenClient) -> str:
    task = TaskInput(
        prompt="@alice waves to @bob across a busy street",
        model="Vidu-Q3",
        task_type="R2V",
        src_image_urls=[
            "https://example.com/alice-1.png",
            "https://example.com/alice-2.png",
            "https://example.com/bob.png",
        ],
        subject_to_image_ids={"alice": [0, 1], "bob": [2]},
        video_config=VideoConfig(duration_secs=5, height=720, aspect_ratio="16:9"),
    )

    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))
text_to_image.py
"""Text-to-image (T2I): generate a still image from a prompt."""

from __future__ import annotations

import time

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


def run(client: MachGenClient) -> str:
    task = TaskInput(
        prompt="An isometric illustration of a cozy reading nook, soft lighting",
        model="Nano-Banana-Pro",
        task_type="T2I",
        image_config=ImageConfig(aspect_ratio="1:1", height=1024),
    )

    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}")

    # task_output maps each output kind to its download locator; a T2I task
    # yields a single image. Use client.download_asset(handle.task_id) for bytes.
    assert result.task_output is not None
    return result.task_output[TaskOutputType.IMAGE]


if __name__ == "__main__":
    with MachGenClient() as client:
        print(run(client))
image_editing.py
"""Image editing (I2I): edit an existing image with a text instruction.

Source images may be public ``http(s)://`` URLs (forwarded as-is) or local file
paths (uploaded automatically on submit). This example uses a URL.
"""

from __future__ import annotations

import time

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


def run(client: MachGenClient) -> str:
    task = TaskInput(
        prompt="make the sky a dramatic sunset",
        model="Nano-Banana-Pro",
        task_type="I2I",
        src_image_urls=["https://example.com/landscape.png"],
        # or local file path:
        # src_image_urls=["/path/to/local/image.png"],
        image_config=ImageConfig(aspect_ratio="1:1", height=1024),
    )

    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.IMAGE]


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

Waiting for results

submit_task returns a TaskHandle immediately. Drive it to completion in one of three ways:

  • client.get_task_state(handle) - fetch the current status once (the polling loop used in the examples above).

  • client.wait(handle, timeout=300.0) - block until the task reaches a terminal status (COMPLETED or FAILED), opening a stream under the hood. Raises TimeoutError on timeout.

  • on_update - pass a callback to submit_task to receive live status changes over a server-sent events stream:

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

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.

API reference

Please refer to the API Reference page.

REST API

Call the MachGen REST API directly over HTTPS.

API Reference

The request, configuration, and response models shared by the REST API and the Python client.

On this page

1. Install2. Provide your API key3. ExamplesWaiting for resultsAPI reference