How to Build a Batch Video Generation Pipeline with Seedance 2.0 Mini

Seedance 2.0 Mini is designed for volume. Its speed and cost advantages compound at scale — but getting from "single clip generation" to a production batch pipeline requires thinking through concurrency, error handling, and cost controls. This guide covers all of it.

The Core Loop

Every Seedance generation follows the same two-step pattern: submit a job, then poll for completion. A minimal implementation:

import requests
import time

ATLAS_API_KEY = "YOUR_ATLASCLOUD_API_KEY"
BASE_URL = "https://api.atlascloud.ai/api/v1/model"

def generate_video(prompt: str, **kwargs) -> str:
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {ATLAS_API_KEY}"
    }

    payload = {
        "model": "bytedance/seedance-2.0/text-to-video",
        "prompt": prompt,
        "width": kwargs.get("width", 1280),
        "height": kwargs.get("height", 720),
        "duration": kwargs.get("duration", 6),
        "fps": 24,
    }
    if "image_url" in kwargs:
        payload["image_url"] = kwargs["image_url"]

    r = requests.post(f"{BASE_URL}/generateVideo", headers=headers, json=payload)
    r.raise_for_status()
    pid = r.json()["data"]["id"]

    while True:
        result = requests.get(
            f"{BASE_URL}/prediction/{pid}",
            headers={"Authorization": f"Bearer {ATLAS_API_KEY}"}
        ).json()
        status = result["data"]["status"]
        if status in ["completed", "succeeded"]:
            return result["data"]["outputs"][0]
        elif status == "failed":
            raise Exception(result["data"].get("error", "Generation failed"))
        time.sleep(2)

Note: Update "model" to "bytedance/seedance-2.0-mini/text-to-video" when Mini API is generally available.

Parallel Batch Generation

For batch workloads, submit all jobs concurrently and poll in parallel:

from concurrent.futures import ThreadPoolExecutor, as_completed

prompts = [
    "A coffee cup steaming on a wooden desk, morning light, cinematic close-up",
    "A smartphone on a marble surface, slow rotation, soft studio lighting",
    "Hands typing on a laptop, overhead shot, shallow depth of field",
    "A pair of headphones on a clean desk, minimal lifestyle shot",
    "A leather wallet on a stone surface, macro detail shot",
]

results = []
with ThreadPoolExecutor(max_workers=5) as executor:
    futures = {executor.submit(generate_video, p): p for p in prompts}
    for future in as_completed(futures):
        try:
            url = future.result()
            results.append({"prompt": futures[future], "url": url, "status": "ok"})
            print(f"✓ Done: {url}")
        except Exception as e:
            results.append({"prompt": futures[future], "error": str(e), "status": "failed"})
            print(f"✗ Failed: {e}")

print(f"\n{len([r for r in results if r['status'] == 'ok'])}/{len(prompts)} succeeded")

Adding Retry Logic

Network blips and transient API errors happen. Add automatic retry with exponential backoff:

import time

def generate_with_retry(prompt: str, max_retries: int = 3, **kwargs) -> str:
    for attempt in range(max_retries):
        try:
            return generate_video(prompt, **kwargs)
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            wait = 2 ** attempt  # 1s, 2s, 4s
            print(f"Attempt {attempt + 1} failed ({e}), retrying in {wait}s...")
            time.sleep(wait)

Cost Controls

For large batches, add a dry-run cost estimate before submitting:

PRICE_PER_SECOND = 0.048  # Mini at 720p

def estimate_cost(prompts: list, avg_duration: float = 6.0) -> dict:
    total_seconds = len(prompts) * avg_duration
    total_cost = total_seconds * PRICE_PER_SECOND
    return {
        "clips": len(prompts),
        "total_seconds": total_seconds,
        "estimated_cost_usd": round(total_cost, 2),
    }

estimate = estimate_cost(prompts, avg_duration=6)
print(f"Estimated cost: ${estimate['estimated_cost_usd']} for {estimate['clips']} clips")

# Confirm before running
confirm = input("Proceed? (y/n): ")
if confirm.lower() != "y":
    print("Cancelled.")
    exit()

Saving Results

Persist generation results to a file as jobs complete:

import json
from datetime import datetime

output_file = f"generation_results_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"

results = []
with ThreadPoolExecutor(max_workers=5) as executor:
    futures = {executor.submit(generate_with_retry, p): i for i, p in enumerate(prompts)}
    for future in as_completed(futures):
        idx = futures[future]
        try:
            url = future.result()
            results.append({"index": idx, "prompt": prompts[idx], "url": url, "status": "ok"})
        except Exception as e:
            results.append({"index": idx, "prompt": prompts[idx], "error": str(e), "status": "failed"})

        # Save incrementally so partial results aren't lost
        with open(output_file, "w") as f:
            json.dump(results, f, indent=2)

print(f"Results saved to {output_file}")

Rate Limits and Concurrency

Atlas Cloud enforces rate limits per API key. For large batches:

• Start with max_workers=3 and increase if you don't see rate limit errors
• If you hit a 429 response, back off and retry after the Retry-After header duration
• For very large runs (10,000+ clips), contact Atlas Cloud about enterprise rate limits

Full Pipeline Summary

Component	What it handles
generate_video()	Single clip, submit + poll
generate_with_retry()	Transient failures, exponential backoff
ThreadPoolExecutor	Parallel concurrent generation
estimate_cost()	Pre-run cost check
Incremental save	Preserves partial results on failure

Get started with Atlas Cloud → | Pricing breakdown