Python Polling and Download

Create Seedance 2.0 / 2.5 tasks, handle channel-specific success responses, and stop on failure, cancellation, or expiry.

This native-route example handles Ark succeeded / content.video_url, Service Inference / Max top-level completed / outputs[0], and generic video_url responses used by some channels. Poll with the returned ModelSell public id.

The default model is doubao-seedance-2-5-260628; set MODELSELL_VIDEO_MODEL to an available model. For edit/extend request bodies, see Request and Response Parameters. A local timeout does not cancel the upstream task; retain its ID for later queries.

pip install requests

import os
import time
import requests

BASE_URL = os.environ.get("MODELSELL_BASE_URL", "https://api.modelsell.com").rstrip("/")
API_KEY = os.environ["MODELSELL_API_KEY"]
MODEL = os.environ.get("MODELSELL_VIDEO_MODEL", "doubao-seedance-2-5-260628")
HEADERS = {"Authorization": f"Bearer {API_KEY}"}


def create_task(prompt: str) -> str:
    response = requests.post(
        f"{BASE_URL}/api/v3/contents/generations/tasks",
        headers=HEADERS,
        json={
            "model": MODEL,
            "content": [{"type": "text", "text": prompt}],
            "resolution": "720p",
            "ratio": "16:9",
            "duration": 5,
            "generate_audio": True,
        },
        timeout=180,
    )
    response.raise_for_status()
    return response.json()["id"]


def wait_for_result(task_id: str, max_wait: int = 900, interval: int = 10) -> str:
    deadline = time.monotonic() + max_wait
    while time.monotonic() < deadline:
        response = requests.get(
            f"{BASE_URL}/api/v3/contents/generations/tasks/{task_id}",
            headers=HEADERS,
            timeout=60,
        )
        response.raise_for_status()
        data = response.json()
        status = data.get("status")
        print(f"status={status}")

        if status in {"succeeded", "completed"}:
            content = data.get("content") or {}
            metadata = data.get("metadata") or {}
            outputs = data.get("outputs") or []
            video_url = (
                content.get("video_url")
                or data.get("video_url")
                or metadata.get("url")
                or (outputs[0] if outputs else None)
            )
            if not video_url:
                raise RuntimeError("Task succeeded but no video URL was returned")
            return video_url
        if status in {"failed", "cancelled", "expired"}:
            error = data.get("error")
            if isinstance(error, dict):
                raise RuntimeError(f"{error.get('code', status)}: {error.get('message', '')}")
            raise RuntimeError(f"{status}: {error or 'Task did not complete'}")
        time.sleep(min(interval, max(0, deadline - time.monotonic())))

    raise TimeoutError(f"Local polling deadline exceeded: {task_id}")


def download_video(video_url: str, save_path: str) -> None:
    with requests.get(video_url, stream=True, timeout=120) as response:
        response.raise_for_status()
        with open(save_path, "wb") as file:
            for chunk in response.iter_content(chunk_size=8192):
                if chunk:
                    file.write(chunk)


if __name__ == "__main__":
    task_id = create_task("A golden Shiba Inu runs under cherry blossoms as the camera rises")
    video_url = wait_for_result(task_id)
    download_video(video_url, "seedance-output.mp4")

Official video URLs expire after 24 hours; 2.5 URLs permit at most 100 downloads. Other channel limits may differ; save results promptly.