How to Batch Generate 100 Product Videos with Python

What you need

The agent-media CLI handles the actual video generation, and Python drives the batch loop around it. Install the CLI if you have not already, then confirm you are logged in.

agent-media login
agent-media whoami

Create a CSV with one row per video. At minimum you need a product name, a script for the actor to say, and a short description of the actor who will say it. You can export this from Shopify, your CMS, or a simple spreadsheet.

product_name,description,character_description
Hydra Serum,"I have been using this serum for two weeks and my skin has never looked better. It absorbs instantly and does not leave a greasy feel.",woman in her late 20s, warm and natural, bathroom mirror light
Protein Pro,"This protein powder mixes clean with no chalky taste. One scoop after my workout and I am good to go.",man in his early 30s, casual and upbeat, gym locker room
Sleep Stack,"I started taking this before bed and I fall asleep in minutes. No groggy mornings either.",woman in her late 30s, calm and reassuring, bedside lamp evening
Daily Greens,"I hate vegetables but this tastes like a green apple smoothie. I drink it every morning now.",man in his 20s, playful and energetic, kitchen morning light
Glow Oil,"Two drops of this and my skin literally glows. My friends keep asking what I changed.",woman in her early 20s, warm and natural, bright natural window light

Tip: write scripts in first person, conversational tone. These will be spoken by the generated person as a natural UGC-style testimonial.

This script reads your CSV, loops through each row, and shells out to agent-media selfie for every product. It waits for each clip to finish before moving to the next one.

import csv
import subprocess

with open("products.csv") as f:
    reader = csv.DictReader(f)
    for row in reader:
        result = subprocess.run(
            [
                "agent-media", "selfie",
                "--description", row["character_description"],
                "--script", row["description"],
                "--duration", "10",
                "--subtitles", "true",
            ],
            capture_output=True,
            text=True,
            check=True,
        )
        job_id = result.stdout.strip().splitlines()[-1]

        subprocess.run(["agent-media", "status", job_id, "--watch"], check=True)
        subprocess.run(
            ["agent-media", "download", job_id, "-o", f"{row['product_name']}.mp4"],
            check=True,
        )
        print(f"[{row['product_name']}] {job_id} -> {row['product_name']}.mp4")

This works but it is sequential, each clip waits for the previous one to finish rendering. For 100 products that means a long wait. Let us fix that.

You can drive the CLI concurrently with asyncio.create_subprocess_exec() instead of subprocess.run(). The example below batches requests in groups of 10 to avoid overwhelming the API.

import asyncio
import csv

BATCH_SIZE = 10

async def run_cli(*args: str) -> str:
    proc = await asyncio.create_subprocess_exec(
        "agent-media", *args,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    stdout, stderr = await proc.communicate()
    if proc.returncode != 0:
        raise RuntimeError(stderr.decode().strip())
    return stdout.decode().strip()

async def generate_video(row: dict) -> dict:
    out = await run_cli(
        "selfie",
        "--description", row["character_description"],
        "--script", row["description"],
        "--duration", "10",
        "--subtitles", "true",
    )
    job_id = out.splitlines()[-1]
    await run_cli("status", job_id, "--watch")
    await run_cli("download", job_id, "-o", f"{row['product_name']}.mp4")
    return {"product": row["product_name"], "job_id": job_id}

async def main():
    with open("products.csv") as f:
        rows = list(csv.DictReader(f))

    results = []
    for i in range(0, len(rows), BATCH_SIZE):
        batch = rows[i : i + BATCH_SIZE]
        batch_results = await asyncio.gather(
            *[generate_video(row) for row in batch]
        )
        results.extend(batch_results)
        print(f"Batch {i // BATCH_SIZE + 1} complete: {len(batch)} videos")

    for r in results:
        print(f"[{r['product']}] {r['job_id']}")

asyncio.run(main())

10 clips generating in parallel instead of one at a time. For 100 products that is 10 batches, dramatically faster than sequential processing.

When you are generating 100 videos, some will inevitably fail, a network hiccup, a transient API error, a bad description. The production-ready version adds retry logic, structured logging, and writes results to an output CSV so you know exactly which products succeeded and which need a rerun.

import asyncio
import csv
import logging

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s")
logger = logging.getLogger(__name__)

BATCH_SIZE = 10
MAX_RETRIES = 3

async def run_cli(*args: str) -> str:
    proc = await asyncio.create_subprocess_exec(
        "agent-media", *args,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    stdout, stderr = await proc.communicate()
    if proc.returncode != 0:
        raise RuntimeError(stderr.decode().strip())
    return stdout.decode().strip()

async def generate_with_retry(row: dict) -> dict:
    for attempt in range(1, MAX_RETRIES + 1):
        try:
            out = await run_cli(
                "selfie",
                "--description", row["character_description"],
                "--script", row["description"],
                "--duration", "10",
                "--subtitles", "true",
            )
            job_id = out.splitlines()[-1]
            await run_cli("status", job_id, "--watch")
            await run_cli("download", job_id, "-o", f"{row['product_name']}.mp4")
            logger.info(f"OK: {row['product_name']}")
            return {
                "product_name": row["product_name"],
                "status": "success",
                "job_id": job_id,
                "file": f"{row['product_name']}.mp4",
            }
        except Exception as e:
            logger.warning(
                f"Attempt {attempt}/{MAX_RETRIES} failed for "
                f"{row['product_name']}: {e}"
            )
            if attempt < MAX_RETRIES:
                await asyncio.sleep(2 ** attempt)

    logger.error(f"FAILED: {row['product_name']}")
    return {
        "product_name": row["product_name"],
        "status": "failed",
        "job_id": "",
        "file": "",
    }

async def main():
    with open("products.csv") as f:
        rows = list(csv.DictReader(f))

    results = []
    for i in range(0, len(rows), BATCH_SIZE):
        batch = rows[i : i + BATCH_SIZE]
        batch_results = await asyncio.gather(
            *[generate_with_retry(row) for row in batch]
        )
        results.extend(batch_results)

    # Write output CSV
    with open("output.csv", "w", newline="") as f:
        writer = csv.DictWriter(
            f, fieldnames=["product_name", "status", "job_id", "file"]
        )
        writer.writeheader()
        writer.writerows(results)

    succeeded = sum(1 for r in results if r["status"] == "success")
    logger.info(f"Done: {succeeded}/{len(results)} videos generated")

asyncio.run(main())

The output CSV gives you a clear record of what was generated:

product_name,status,job_id,file
Hydra Serum,success,job_abc123,Hydra Serum.mp4
Protein Pro,success,job_def456,Protein Pro.mp4
Sleep Stack,failed,,

Filter the output for failed rows and rerun just those, no need to regenerate the entire batch.

The examples above call agent-media status <job-id> --watch, which blocks until each clip is ready. For production systems you may prefer to submit every job first and poll them separately, so a slow render does not hold up the rest of the batch.

import csv
import subprocess

with open("products.csv") as f:
    reader = csv.DictReader(f)
    job_ids = []
    for row in reader:
        result = subprocess.run(
            [
                "agent-media", "selfie",
                "--description", row["character_description"],
                "--script", row["description"],
                "--duration", "10",
                "--subtitles", "true",
            ],
            capture_output=True,
            text=True,
            check=True,
        )
        job_id = result.stdout.strip().splitlines()[-1]
        job_ids.append((row["product_name"], job_id))
        print(f"Queued: {row['product_name']} -> {job_id}")

with open("job_ids.csv", "w", newline="") as f:
    writer = csv.writer(f)
    writer.writerow(["product_name", "job_id"])
    writer.writerows(job_ids)

Later, poll each job and download the ones that finished:

agent-media list
agent-media download job_abc123 -o clip.mp4

Cost breakdown: 100 product videos

agent-media selfie costs 45 credits plus 28 credits per second of output. A typical 10-second product video costs 325 credits (45 + 28 x 10). Here is what 100 videos looks like:

100 videos x 325 credits = 32,500 credits.

Compare that to hiring 100 UGC creators at $50 to $200 each, roughly $5,000 to $20,000. The API approach runs on a credit plan from $39/month and delivers in minutes, not weeks. See the full tier breakdown on the pricing page.

TypeScript equivalent

Prefer TypeScript? The same batch generation pattern works by shelling out to the agent-media CLI from Node, or by calling POST /v2/generate/video directly with fetch. Here is the async version with batching: