MiniMax Python Polling and Download

pip install requests
export MODELSELL_BASE_URL="https://api.modelsell.com"
export MODELSELL_API_KEY="<your-modelsell-api-key>"

import os
import time
from pathlib import Path
from urllib.parse import quote, urlparse

import requests

BASE = os.environ.get("MODELSELL_BASE_URL", "https://api.modelsell.com").rstrip("/")
KEY = os.environ["MODELSELL_API_KEY"]
HEADERS = {"Authorization": f"Bearer {KEY}"}
PROMPT = "A red ceramic teapot sits on a wooden table in warm morning sunlight."


def create_task(path, body, family):
    response = requests.post(BASE + path, headers=HEADERS, json=body, timeout=300)
    response.raise_for_status()
    data = response.json()
    task_id = data.get("task_id") or (data.get("task") or {}).get("id") or data.get("id")
    if not isinstance(task_id, str) or not task_id.startswith("task_"):
        raise RuntimeError("Creation response has no ModelSell public task ID")
    print(f"Save this task ID: {task_id}", flush=True)
    return task_id, family


def wait_for_task(task_id, family, max_wait=900, interval=12):
    prefix = {
        "native": "/v2/query/video_generation/",
        "platform": "/v1/video/tasks/",
        "generic": "/v1/videos/",
    }[family]
    deadline = time.monotonic() + max_wait
    while time.monotonic() < deadline:
        remaining = deadline - time.monotonic()
        response = requests.get(
            BASE + prefix + quote(task_id, safe=""),
            headers=HEADERS,
            timeout=min(300, max(0.1, remaining)),
        )
        response.raise_for_status()
        payload = response.json()
        task = payload if family == "generic" else payload["task"]
        status = task.get("status")
        if status in {"succeeded", "completed"}:
            return task
        if status in {"failed", "cancelled"}:
            raise RuntimeError(f"Task {task_id} ended as {status}: {task.get('error')}")
        if status not in {"queued", "running", "pending", "processing", "in_progress"}:
            raise RuntimeError(f"Unexpected task status: {status!r}; retain ID {task_id}")
        print(f"{task_id}: {status}", flush=True)
        time.sleep(min(interval, max(0, deadline - time.monotonic())))
    raise TimeoutError(f"Local wait expired; query this task later: {task_id}")


def video_url(task):
    content = task.get("content") or {}
    metadata = task.get("metadata") or {}
    outputs = task.get("outputs") or metadata.get("outputs") or []
    url = content.get("url") or task.get("video_url") or metadata.get("url")
    url = url or (outputs[0] if outputs else None)
    if not isinstance(url, str) or urlparse(url).scheme not in {"http", "https"}:
        raise RuntimeError("Successful task did not contain a downloadable video URL")
    return url


def download_video(url, destination):
    target = Path(destination)
    temporary = target.with_suffix(target.suffix + ".part")
    try:
        # The result may be on a CDN: do not send the ModelSell API Key to it.
        with requests.get(url, stream=True, timeout=300) as response:
            response.raise_for_status()
            with temporary.open("wb") as output:
                for chunk in response.iter_content(65536):
                    if chunk:
                        output.write(chunk)
        temporary.replace(target)
    except Exception:
        temporary.unlink(missing_ok=True)
        raise


if __name__ == "__main__":
    task_id, family = create_task("/v1/videos", {
        "model": "minimax-h3", "prompt": PROMPT, "seconds": "4", "size": "768P",
        "metadata": {"ratio": "16:9", "aigc_watermark": False},
    }, "generic")
    result = wait_for_task(task_id, family)
    download_video(video_url(result), "minimax-h3-output.mp4")

# Native H3 generation: family="native"
task_id, family = create_task("/v2/video_generation", {
    "model": "MiniMax-H3", "content": [{"type": "text", "text": PROMPT}],
    "resolution": "768P", "duration": 4, "ratio": "16:9",
}, "native")

# Context-IR returns text: do not call video_url() / download_video().
ir_id, ir_family = create_task("/v2/h3_context_ir", {
    "model": "MiniMax-H3", "content": [{"type": "text", "text": PROMPT}],
    "duration": 4, "ratio": "16:9",
}, "native")
enhanced_prompt = wait_for_task(ir_id, ir_family)["content"]["prompt"]
if not enhanced_prompt.strip():
    raise RuntimeError("Context-IR returned an empty prompt")
print(enhanced_prompt)

# Platform Max: change model to minimax-h3-max-turbo for Turbo.
max_id, max_family = create_task("/v1/video/generate", {
    "model": "minimax-h3-max", "content": [{"type": "text", "text": PROMPT}],
    "resolution": "480P", "duration": 5, "ratio": "16:9",
}, "platform")
download_video(video_url(wait_for_task(max_id, max_family)), "minimax-max-output.mp4")

# Regeneration: task_id must refer to the successful H3 task above, not Max / Turbo or IR.
source = wait_for_task(task_id, "native")
regen_id, regen_family = create_task("/v2/video_regeneration", {
    "model": "MiniMax-H3", "source_task_id": task_id, "resolution": "2K",
}, "native")
download_video(video_url(wait_for_task(regen_id, regen_family)), "minimax-h3-2k.mp4")