Seedance 2 asynchronous video

Copy this capability to an Agent.

All controls below are generated from [email protected]; model, endpoint, lifecycle, recovery, and verification cannot drift independently.

Copy for Codex Copy for Claude Code Copy API request

Runnable Python from the same source.

#!/usr/bin/env python3
"""Generated from scripts/agent-recipes.json. Do not put API keys in this file."""
import json, os, sys, time
from pathlib import Path
from urllib.error import HTTPError, URLError
from urllib.parse import urlsplit
from urllib.request import Request, urlopen

API = os.environ.get("CHINAAPI_API_BASE", "https://api.chinaapi.ai").rstrip("/")
MODEL = os.environ.get("CHINAAPI_MODEL", "doubao-seedance-2-0-mini-260615")
KEY = os.environ.get("CHINAAPI_API_KEY")
OUT = Path("artifacts/seedance-2.mp4")
SUCCESS = ('completed', 'succeeded', 'success')
FAILURE = ('failed', 'failure', 'cancelled', 'canceled', 'error')

if not KEY:
    raise SystemExit("Missing CHINAAPI_API_KEY. Export it in this shell; never save it in the project.")

def request(method, path, payload=None, retries=0):
    body = json.dumps(payload).encode() if payload is not None else None
    req = Request(API + path, data=body, method=method, headers={"Authorization": "Bearer " + KEY, "Content-Type": "application/json"})
    attempt = 0
    while True:
        try:
            with urlopen(req, timeout=60) as response:
                return json.loads(response.read())
        except HTTPError as exc:
            detail = exc.read().decode("utf-8", "replace")[:1000]
            hints = {'400': 'Validate the exact model ID and request fields shown by the response before retrying.', '401': 'Confirm CHINAAPI_API_KEY is exported in this shell and has not been pasted into a file.', '402': 'Check balance and live pricing in Dash; do not loop a billable create request.', '429': 'Honor Retry-After when present and retry the poll; do not create a duplicate job.', '5xx': 'Retry polling the existing task with backoff. Create a new task only after confirming the old task failed.', 'network': 'The runner retries a dropped poll or download against the same task automatically. If it still exits, resume polling that task ID; a network failure never means the job was lost.', 'timeout': 'Print the task ID and resume polling that task; do not silently submit another billable job.'}
            raise SystemExit(f"HTTP {exc.code}: {detail}\nRecovery: {hints.get(str(exc.code), hints.get('5xx') if exc.code >= 500 else 'Correct the response error before retrying.')}" )
        except (URLError, OSError) as exc:
            # A create request is billable and may have already reached the API
            # before the connection dropped, so it never retries automatically.
            # Polling an existing task is side-effect free: a transient blip
            # (flaky link, local TLS-intercepting proxy) must not strand a job
            # that is already running and already billed.
            if attempt >= retries:
                raise SystemExit(f"Network error: {exc}. Retry polling the same task; do not create a duplicate job.")
            attempt += 1
            print(f"network error on {method}, retry {attempt}/{retries} against the same task: {exc}", flush=True)
            time.sleep(min(5 * attempt, 30))

created = request("POST", "/v1/video/generations", {"model": MODEL, "prompt": 'A paper boat crossing a calm blue pond, one continuous shot'})
task_id = created.get("id") or created.get("task_id") or (created.get("data") or {}).get("id") or (created.get("data") or {}).get("task_id")
if not task_id:
    raise SystemExit("Create response did not contain id/task_id: " + json.dumps(created)[:1000])
print("created task", task_id)

deadline = time.time() + 900
result = None
while time.time() < deadline:
    result = request("GET", "/v1/video/generations/{task_id}".format(task_id=task_id), retries=20)
    data = result.get("data") if isinstance(result.get("data"), dict) else {}
    status = str(result.get("status") or data.get("status") or "").lower()
    print("status", status or "unknown")
    if status in SUCCESS:
        break
    if status in FAILURE:
        raise SystemExit("Task failed: " + json.dumps(result)[:1000])
    retry_after = result.get("retry_after") or data.get("retry_after") or 10
    time.sleep(max(2, min(int(retry_after), 60)))
else:
    raise SystemExit(f"Timed out. Resume polling task {task_id}; do not submit a duplicate billable job.")

data = result.get("data") if isinstance(result.get("data"), dict) else {}
output = data.get("output") if isinstance(data.get("output"), dict) else {}
url = result.get("video_url") or result.get("output_url") or result.get("result_url") or result.get("url") or data.get("video_url") or data.get("output_url") or data.get("result_url") or data.get("url") or output.get("video_url") or output.get("result_url") or output.get("url")
if not url:
    raise SystemExit("Completed task did not contain a downloadable video URL: " + json.dumps(result)[:1000])
OUT.parent.mkdir(parents=True, exist_ok=True)
parts = urlsplit(url)
if parts.scheme not in {"http", "https"}:
    raise SystemExit(f"Artifact URL uses unsupported scheme: {parts.scheme!r}")
download_headers = {"Authorization": "Bearer " + KEY} if parts.netloc == urlsplit(API).netloc else {}
download = Request(url, headers=download_headers)
# The job already completed and was billed by this point; a dropped connection
# here must not discard the artifact. Downloading is a safe idempotent GET.
content_type, payload = "", b""
for attempt in range(1, 6):
    try:
        with urlopen(download, timeout=120) as response:
            content_type = response.headers.get("Content-Type", "")
            payload = response.read()
        break
    except (URLError, OSError) as exc:
        if attempt == 5:
            raise SystemExit(f"Artifact download failed after {attempt} attempts: {exc}. The task already succeeded; re-download {url} before creating another job.")
        print(f"download error, retry {attempt}/5: {exc}", flush=True)
        time.sleep(min(5 * attempt, 30))
if len(payload) < 1024 or not (payload[4:8] == b"ftyp" or payload.startswith(b"\x1aE\xdf\xa3")):
    raise SystemExit(f"Artifact validation failed: {len(payload)} bytes, content-type={content_type!r}")
OUT.write_bytes(payload)
print(f"saved and verified {OUT}: {len(payload)} bytes")

Recovery and acceptance.

• 400: Validate the exact model ID and request fields shown by the response before retrying.
• 401: Confirm CHINAAPI_API_KEY is exported in this shell and has not been pasted into a file.
• 402: Check balance and live pricing in Dash; do not loop a billable create request.
• 429: Honor Retry-After when present and retry the poll; do not create a duplicate job.
• 5xx: Retry polling the existing task with backoff. Create a new task only after confirming the old task failed.
• network: The runner retries a dropped poll or download against the same task automatically. If it still exits, resume polling that task ID; a network failure never means the job was lost.
• timeout: Print the task ID and resume polling that task; do not silently submit another billable job.

python3 -c "from pathlib import Path; p=Path('artifacts/seedance-2.mp4'); b=p.read_bytes(); assert len(b)>=1024 and (b[4:8]==b'ftyp' or b.startswith(b'\x1aE\xdf\xa3')), (len(b), b[:12]); print(f'verified {p}: {len(b)} bytes')"

Runtime acceptance status: live-verified. A copy or click is not an activation; success requires a completed API call.