Automate Higgsfield with a Python Batch Queue Runner

TL;DR: A batch runner for a metered generation API needs five things most tutorials skip — two workers because the queue allows two concurrent slots, a manifest that records failures as well as successes, retries with growing backoff, a disk guard that stops the whole run, and defensive parsing because the CLI's JSON shape is not guaranteed. Here is the script, and why each part exists.

This is post #5 of the AI Video Automation series. It assumes you understand the credit model — the runner below spends real money on every job, which is the reason for half its design.

Prerequisites

• The Higgsfield CLI installed and authenticated, with a workspace selected
• Python 3.9+ — standard library only, no extra packages
• A shot list CSV

Step 1: The shot list

One row per shot. This is the only input the runner takes, and every later stage reads from it:

id,product,category,model,prompt,aspect,duration,mode,resolution,priority
hero_garage,demo,video,kling3_0,"A wide interior of a modern workshop, warm light, slow dolly-in, cinematic, photorealistic",9:16,5,pro,,1
plate_streaks,demo,video,seedance_2_0,"Abstract amber and teal light streaks on dark, premium, cinematic",9:16,5,std,720p,1
still_hero,demo,image,nano_banana_2,"Photoreal modern workshop, warm light, negative space for text",16:9,,,2k,1

category is the important column — it is what splits the list across workers in step 6. priority matters because runs get interrupted; sorting by it means the shots you actually need are finished first.

rows = list(csv.DictReader(open("prompts.csv")))
rows.sort(key=lambda r: int(r.get("priority", 999)))   # essentials first

Step 2: Build the command

Every optional column becomes an optional flag. Passing an empty --resolution is not the same as omitting it, so build the list conditionally:

def build_cmd(p, seed):
    cmd = ["higgsfield", "generate", "create", p["model"],
           "--prompt", p["prompt"], "--wait", "--json"]
    if p.get("aspect"):     cmd += ["--aspect_ratio", p["aspect"]]
    if p.get("resolution"): cmd += ["--resolution", p["resolution"]]
    if p.get("duration"):   cmd += ["--duration", str(p["duration"])]
    if p.get("mode"):       cmd += ["--mode", p["mode"]]
    if seed is not None:    cmd += ["--seed", str(seed)]
    return cmd

--wait blocks until the generation finishes and --json makes the output parseable. Both are what let a single thread drain a queue sequentially without polling.

Step 3: Parse the URL defensively

This is the part I would not have written on day one, and the part that saved the run.

The CLI returns a result URL, not a file. The field name that URL arrives under is not something you should assume — CLI JSON shapes change between versions, and a batch that runs overnight is a bad place to discover it. So try the likely keys, then the likely nested shapes, then fall back to scraping any http token out of the output:

def extract_url(stdout):
    """VERIFY on day one: run one job with --json and confirm the field name."""
    try:
        data = json.loads(stdout)
    except json.JSONDecodeError:
        # some CLIs print a plain URL line; take the last http(s) token
        for tok in reversed(stdout.split()):
            if tok.startswith("http"):
                return tok
        return None

    for key in ("url", "result_url", "output_url", "download_url"):
        if isinstance(data, dict) and data.get(key):
            return data[key]

    # nested shapes: {"result": {"url": ...}} or {"outputs": [{"url": ...}]}
    if isinstance(data, dict):
        r = data.get("result") or {}
        if isinstance(r, dict) and r.get("url"):
            return r["url"]
        outs = data.get("outputs") or data.get("assets") or []
        if outs and isinstance(outs, list) and isinstance(outs[0], dict):
            return outs[0].get("url")
    return None

The alternative — data["result_url"] — raises a KeyError at 2am, after the credits have already been spent. The generation succeeded; only your parsing failed. That is the worst possible failure on a metered API, because a retry pays for the same clip twice.

Step 4: Generate, then download

Because the output is a URL, the runner fetches the file itself rather than trusting the CLI to write it to disk:

ext = {"video": "mp4", "image": "png", "audio": "mp3"}[p["category"]]
filename = f"{p['id']}_s{seed}.{ext}" if seed is not None else f"{p['id']}.{ext}"

proc = subprocess.run(cmd, capture_output=True, text=True, timeout=20 * 60)
if proc.returncode != 0:
    raise RuntimeError(proc.stderr.strip()[:200])

url = extract_url(proc.stdout)
if not url:
    raise RuntimeError("no URL in output: " + proc.stdout[:200])

download(url, filename)

The 20-minute timeout is not arbitrary. --wait blocks, and a hung job with no timeout holds its worker thread forever — which on a two-slot queue means half your throughput disappears silently.

Step 5: A manifest that records failures too

Most resume implementations record what succeeded. Recording what failed matters just as much, because it is the difference between "this shot has not been attempted" and "this shot has been attempted three times and costs money each time":

MAX_RETRIES   = 3
RETRY_BACKOFF = 20   # seconds, grows per attempt

for attempt in range(1, MAX_RETRIES + 1):
    try:
        ...                                   # generate + download
        append_manifest({..., "status": "done",   "filename": filename, "url": url})
        return
    except Exception as e:
        log(f"FAIL {filename}: {e}")
        if attempt < MAX_RETRIES:
            time.sleep(RETRY_BACKOFF * attempt)   # 20s, 40s, 60s
        else:
            append_manifest({..., "status": "failed", "notes": str(e)[:200]})

Backoff grows with the attempt number rather than staying flat — if the first retry failed because the service was busy, retrying at the same interval mostly just burns the remaining attempts.

On startup, the runner reads the manifest and skips anything already marked done:

def load_done_ids():
    done = set()
    if os.path.exists(MANIFEST):
        for r in csv.DictReader(open(MANIFEST)):
            if r["status"] == "done":
                done.add(r["id"])
    return done

A run that cannot resume is a run you cannot afford to interrupt — and on a job measured in hours, something always interrupts it.

Step 6: Two workers, because there are two slots

This is the design decision people get wrong first, myself included.

The relaxed queue allows one video and one image at a time. So concurrency here is not "how many threads can I spawn" — it is two slots, one per media type. Four workers do not go faster; three of them queue behind each other while the image slot sits idle.

video_jobs = [(p, s) for p in rows if p["category"] == "video"
                                   for s in range(1, SEEDS + 1)]
image_jobs = [(p, None) for p in rows if p["category"] != "video"]

threads = [
    threading.Thread(target=worker, args=("video", video_jobs, dry_run)),
    threading.Thread(target=worker, args=("image", image_jobs, dry_run)),
]
for t in threads: t.start()
for t in threads: t.join()

Note that video jobs multiply by seed and image jobs do not. Multiple seeds give you alternate takes of the same shot, which is worth it for a hero clip and wasteful for a still.

Step 7: The disk guard

An overnight run that fills the volume at 3am fails far worse than one that stops cleanly, so check before each job and set a shared stop event when space runs low:

MIN_FREE_GB = 3

def free_gb(path="."):
    return shutil.disk_usage(path).free / (1024 ** 3)

if free_gb(OUT_DIR) < MIN_FREE_GB:
    log(f"LOW DISK ({free_gb(OUT_DIR):.1f} GB free) — stopping before {filename}")
    _stop.set()          # both workers see this and drain
    return

The event is shared across threads deliberately. Stopping one worker while the other keeps writing just delays the same failure by a few minutes.

Step 8: Run it in three phases

The first two phases exist entirely because the third spends money:

# 1. Dry run — prints every command, generates nothing, costs nothing
python queue_runner.py --prompts prompts.csv --dry-run

# 2. Measure one — check the credit delta against your estimate
python queue_runner.py --prompts prompts.csv --slot video --limit 1

# 3. Full run — both slots, overnight-safe
python queue_runner.py --prompts prompts.csv --seeds 3

The dry run catches malformed rows and bad model names for free. The single measured clip is how you discover a row costs 45 credits rather than 22.5 before multiplying it by forty. Skipping either is how a batch becomes an expensive mistake.

What this design is really about

Almost every non-obvious decision here traces back to one property: each job costs money and is not idempotent. Retrying a failed HTTP request is free. Retrying a failed generation is not — the clip may already have been produced and paid for, and only the parsing failed.

That is why the URL parsing is paranoid, why the manifest records failures, why the retry count is bounded at three, and why nothing runs before a dry run. Write a batch runner for a free API and none of this matters. Write one for a metered API and all of it does.

Related posts

• How to Automate AI Video Creation with Claude & Higgsfield — the pipeline this runner sits inside
• Higgsfield Credits vs Unlimited: Real Per-Clip Cost Breakdown — why the runner is built around cost
• Cinematic AI Video Prompts: A Step-by-Step Formula — what goes in the prompt column
• Browse the AI Video Automation series