How to Automate Image Generation at Scale

The easiest way to automate image generation at scale without managing several different APIs is to separate the workflow from the model provider. Put every image request into one queue, route each job to a current model ID, and send compatible requests through one CometAPI key and the OpenAI-compatible base URL https://api.cometapi.com/v1.

This tutorial builds that pipeline in Python. It accepts product, advertising, and content jobs from a JSON Lines queue; chooses a model; limits concurrency; retries transient failures; stores either URL- or base64-based results; and records per-job usage and estimated cost. The example keeps those production essentials in one compact block so you can test the workflow without turning the article into a code reference.

How a Unified Image API Simplifies Batch Generation

By the end, the workflow will look like this:

jobs.jsonl → bounded worker pool → CometAPI /v1/images/generations → object storage → manifest.jsonl

The queue and storage layer remain yours. Switching image models changes the model value, not the authentication system or the main request route. That is the practical advantage of a unified image API: model choice becomes a routing decision inside one pipeline instead of a separate provider integration.

What Do You Need to Automate Image Generation?

You need Python 3.10 or later, the requests package, a CometAPI key, a writable output location, and at least one current image model ID.

Install the only dependency:

pip install requests

Set your key on the server, never in browser code or a repository:

export COMETAPI_KEY="your-key"

The base URL is https://api.cometapi.com/v1, and compatible text-to-image jobs use POST /images/generations. Before a deployment, verify each model in the live model catalog; the catalog returns the current ID, supported endpoint, features, and pricing metadata without requiring an authorization header.

As of August 20, 2026, the live catalog listed these two useful routes:

The table is a starting point, not a claim that the models have identical capabilities. Size, quality, format, reference-image support, and response behavior remain model-specific. Check the model record and its linked documentation before passing optional parameters.

How to Build a Batch Image Generation Workflow in Python

1. Give every job a durable ID

Use one JSON object per line so a queue, database export, or spreadsheet job can feed the same worker:

{"id":"sku-1001","kind":"product","prompt":"Studio product photo of a ceramic coffee dripper on a warm neutral background"}
{"id":"campaign-204","kind":"ad","prompt":"Editorial summer travel image, vivid natural light, wide composition, no text"}
{"id":"blog-088","kind":"content","prompt":"Minimal illustration of a developer automating a creative workflow, no text"}

The ID becomes the output filename and manifest key. In production, use it as the idempotency key and skip IDs already marked successful before reprocessing a queue.

2. Route by job type, then validate against the live catalog

The example maps product work to gpt-image-2 and ad or content work to doubao-seedream-4-5-251128. A job may override that choice with its own model field. At startup, the worker downloads the public catalog and rejects an ID that is no longer listed.

This is safer than hard-coding a provider-specific SDK throughout the application. You can change a route in one mapping after evaluating quality, latency, and price for your own prompts.

3. Bound concurrency instead of launching the whole batch

The worker starts with four concurrent requests. That number is a conservative application setting, not a universal service limit. Measure latency and 429 responses for your account, then raise or lower MAX_WORKERS deliberately.

Only 408, 429, and 5xx responses are retried with exponential backoff and jitter. Authentication errors, invalid model IDs, and unsupported parameters fail immediately because retrying the same bad request only adds delay.

4. Normalize the result before storage

Image models do not always return the same container. The documented GPT Image response contains data[0].b64_json; other compatible models may return data[0].url. The worker handles both, writes the image to a temporary file, and renames it only after the download or decode succeeds.

For production, replace the local output/ directory with S3, R2, GCS, or another object store. Do not treat a provider-hosted URL as permanent storage unless its retention policy explicitly says so.

5. Record usage, attempts, and estimated cost

Every result becomes a compact manifest row with the job ID, model, saved path, status, and estimated USD cost when the live catalog provides enough pricing data. Failed jobs keep the error instead of disappearing from the batch.

Complete Python Script for Batch Image Generation

Save the following as batch_image_pipeline.py, place the queue beside it as jobs.jsonl, and run python3 batch_image_pipeline.py.

import base64, json, os, random, time
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
import requests

BASE_URL = "https://api.cometapi.com/v1"
KEY = os.environ["COMETAPI_KEY"]
WORKERS = int(os.getenv("MAX_WORKERS", "4"))
OUT = Path("output")
ROUTES = {
    "product": "gpt-image-2",
    "ad": "doubao-seedream-4-5-251128",
    "content": "doubao-seedream-4-5-251128",
}

catalog = requests.get("https://api.cometapi.com/api/models", timeout=30)
catalog.raise_for_status()
CATALOG = {model["id"]: model for model in catalog.json()["data"]}

def generate(job):
    model = job.get("model", ROUTES[job["kind"]])
    if model not in CATALOG:
        raise ValueError(f"Unknown model: {model}")
    payload = {"model": model, "prompt": job["prompt"], "n": 1}
    if model == "gpt-image-2":
        payload.update(quality="low", size="1024x1024", output_format="jpeg")

    for attempt in range(4):
        response = requests.post(
            f"{BASE_URL}/images/generations",
            headers={"Authorization": f"Bearer {KEY}"},
            json=payload,
            timeout=180,
        )
        if response.status_code not in {408, 429} and response.status_code < 500:
            break
        time.sleep(2**attempt + random.random())
    response.raise_for_status()
    body = response.json()
    item = body["data"][0]

    if item.get("b64_json"):
        data = base64.b64decode(item["b64_json"])
        extension = body.get("output_format", "png")
    else:
        download = requests.get(item["url"], timeout=120)
        download.raise_for_status()
        data = download.content
        extension = {"image/png": "png", "image/webp": "webp"}.get(
            download.headers.get("content-type"), "jpg"
        )
    path = OUT / f"{job['id']}.{extension}"
    path.write_bytes(data)

    price, usage = CATALOG[model].get("pricing") or {}, body.get("usage", {})
    cost = price.get("per_request")
    if cost is None and price.get("input") is not None:
        cost = (usage.get("input_tokens", 0) * price["input"] +
                usage.get("output_tokens", 0) * price["output"]) / 1_000_000
    return {"id": job["id"], "model": model, "path": str(path),
            "estimated_usd": cost * price.get("ratio", 1) if cost is not None else None}

def safe_generate(job):
    try:
        return {"status": "success", **generate(job)}
    except Exception as error:
        return {"id": job["id"], "status": "failed", "error": str(error)}

OUT.mkdir(exist_ok=True)
jobs = [json.loads(line) for line in Path("jobs.jsonl").read_text().splitlines() if line]
with ThreadPoolExecutor(max_workers=WORKERS) as pool:
    results = list(pool.map(safe_generate, jobs))
with (OUT / "manifest.jsonl").open("w") as manifest:
    manifest.writelines(json.dumps(result) + "\n" for result in results)

The script uses the current catalog at runtime, while the two fallback mappings are examples verified on August 20, 2026. Recheck them before publishing or deploying the code on another date.

How to Test the Batch Image Generation Workflow

Start with one job and one worker:

MAX_WORKERS=1 python3 batch_image_pipeline.py

A successful GPT Image response follows this structure:

{
  "created": 1776841943,
  "output_format": "jpeg",
  "quality": "low",
  "size": "1024x1024",
  "usage": {
    "input_tokens": 16,
    "output_tokens": 208,
    "total_tokens": 224
  },
  "data": [{"b64_json": "<base64-image-data>"}]
}

The worker decodes the image, writes output/<job-id>.jpeg, and adds a success row to output/manifest.jsonl. If a model returns a URL instead, the worker downloads it and stores the local path in the same manifest format.

The code was syntax-checked locally. A live generation call still requires your CometAPI key, so run the one-job smoke test before increasing concurrency.

How Much Does Batch Image Generation Cost?

Pricing must be time-stamped because model rates change. As of August 20, 2026, the live CometAPI model catalog returned the following base price fields and a 0.8 billing ratio:

• gpt-image-2: $5 per 1M input tokens and $30 per 1M output tokens; applying the listed ratio gives effective rates of $4 and $24 per 1M tokens.
• doubao-seedream-4-5-251128: $0.04 per request; applying the listed ratio gives $0.032 per request.

The CometAPI pricing guide explains token-based billing for models with official pricing and call-based billing for models priced per request. The script reads the catalog when it runs and uses the same rule:

token cost = ratio × (input tokens × input rate + output tokens × output rate) / 1,000,000

request cost = ratio × per-request price

For example, the documented GPT Image response above reports 16 input tokens and 208 output tokens. Using the August 20 catalog values, that illustrative result estimates to about $0.005056. The real total changes with model, quality, size, prompt, retries, and the response usage. Treat the API response and account usage dashboard as the billing record, not a fixed per-image assumption.

Budget for unsuccessful work too. A retry after an unconfirmed timeout may produce a second billable result, and a technically successful image that fails review still consumes budget. Track both API cost and the acceptance rate:

effective cost per accepted image = total batch spend / approved images

Common Image Generation API Errors and How to Fix Them

Do not retry every error. A permanent 400 request will remain invalid, while an unlimited 429 retry loop can turn a traffic spike into a backlog.

Best Practices for Production Image Generation at Scale

Move from JSON Lines to a durable queue when multiple workers are involved. Set a visibility timeout longer than the maximum generation time, acknowledge a job only after the image and manifest are stored, and send exhausted jobs to a dead-letter queue for review.

Keep optional controls in model-specific configuration. A shared payload should contain only common fields such as model, prompt, and n: 1; add quality, size, or output_format only after the selected model documentation confirms them. If you add fallback routing, choose a model that supports the same task and rebuild the payload for that model instead of blindly replaying provider-specific options.

Store the API key in a secret manager, restrict prompt input, scan generated assets according to your policy, and keep provider URLs out of long-term product records. Log job ID, model ID, latency, attempts, usage, storage path, review result, and catalog snapshot date. Those fields let you compare models by accepted-image cost rather than headline price alone.

Finally, set budget guardrails: a maximum batch size, a per-job retry limit, a daily spend alert, and a stop condition when the approval rate falls. Scaling a poor prompt faster is not an optimization.

FAQs About Automating Image Generation at Scale

What is the easiest way to automate image generation at scale without managing several APIs?

Use one queue and storage workflow, then send compatible image requests through one CometAPI key and https://api.cometapi.com/v1/images/generations. Change the model ID in your routing layer instead of maintaining separate authentication and provider SDKs.

Can I send one request and ask several image models to generate at once?

The example sends one model per job. Fan-out is an application workflow: duplicate a job with distinct IDs and model values, then compare the stored outputs. This keeps cost and review status attributable to each model.

What concurrency should I use?

There is no universal number for every account and model. Start with a small bounded pool such as four workers, monitor latency and 429 responses, and tune from evidence.

Should I store the returned URL or the image itself?

Store the image in your own object storage. A returned URL may be temporary, while GPT Image models may return base64 content instead of a URL.

How do I choose the cheapest model?

Calculate cost per accepted image, not just price per call. Include token or request charges, retries, failed downloads, rejected assets, post-processing, and human review. Recheck the live model catalog on the day you publish or deploy.

Where should I verify the endpoint and response format?

Use the CometAPI Quick Start, model catalog documentation, image generation reference, and pricing guide.