Skills › Scientific Agent Skills › generate-image › generate_image.py
generate_image.py
802 lines · skills/generate-image/scripts/generate_image.py at 5f2bc03
1#!/usr/bin/env python3
2"""
3Generate and edit images through the OpenRouter Image API (POST /api/v1/images).
5The Image API is model-agnostic: the same request shape reaches Gemini,
6Seedream, Recraft, GPT-Image, Riverflow, and the rest of the image catalogue.
7Responses carry base64 payloads in ``data[].b64_json`` alongside the concrete
8``media_type``, so the output extension follows what the model actually
9returned rather than an assumption.
11Reference images (image-to-image and editing) go in ``input_references`` as
12HTTP(S) URLs or base64 data URLs.
14Parameter support differs by model and endpoint. Every generation is preceded
15by free model and endpoint metadata lookups that validate advertised support
16locally before anything is billed (``--no-preflight`` skips them). Only the
17standard library is required.
18"""
20from __future__ import annotations
22import argparse
23import base64
24import binascii
25import difflib
26import http.client
27import json
28import math
29import os
30import sys
31import time
32import urllib.error
33import urllib.request
34from pathlib import Path
35from typing import Any
37API_BASE = "https://openrouter.ai/api/v1"
38IMAGES_URL = f"{API_BASE}/images"
39MODELS_URL = f"{API_BASE}/images/models"
41DEFAULT_MODEL = "google/gemini-3.1-flash-image"
43# media_type -> file extension. Vector-capable models return image/svg+xml.
44EXTENSIONS = {
45 "image/png": ".png",
46 "image/jpeg": ".jpg",
47 "image/jpg": ".jpg",
48 "image/webp": ".webp",
49 "image/gif": ".gif",
50 "image/svg+xml": ".svg",
51}
53# Suffixes that name the same format, so `-o out.jpeg` for an image/jpeg
54# response is not reported as a mismatch.
55SUFFIX_ALIASES = {
56 ".jpg": {".jpg", ".jpeg"},
57 ".jpeg": {".jpg", ".jpeg"},
58 ".svg": {".svg"},
59}
61# Local file extension -> MIME type, for encoding reference images.
62INPUT_MIME = {
63 ".png": "image/png",
64 ".jpg": "image/jpeg",
65 ".jpeg": "image/jpeg",
66 ".gif": "image/gif",
67 ".webp": "image/webp",
68}
70# Request parameters that carry a per-model capability spec. `model`, `prompt`,
71# and `input_references` are handled separately.
72VALIDATED_PARAMETERS = (
73 "n",
74 "aspect_ratio",
75 "resolution",
76 "quality",
77 "output_format",
78 "background",
79 "output_compression",
80 "seed",
81)
83RETRY_STATUS = frozenset({429, 500, 502, 503, 504})
85# Ceiling for an image the response asks us to download rather than inlining.
86# A 4K PNG is a few tens of megabytes; anything past this is not an image we want.
87MAX_DOWNLOAD_BYTES = 64 * 1024 * 1024
89class ApiError(RuntimeError):
90 """An error surfaced by the OpenRouter API or the transport beneath it."""
93class RequestRejected(ApiError):
94 """The request cannot succeed as built, caught before it is billed."""
97def find_api_key(explicit: str | None = None) -> str:
98 """Resolve the API key from --api-key, the environment, then any .env file.
100 The .env scan walks up from the working directory and finally checks the
101 script's own directory, so running from anywhere inside a project picks up
102 the key at its root. Only the standard library is used: python-dotenv is a
103 common omission, and a missing optional dependency should not read as a
104 missing credential.
105 """
106 if explicit:
107 return explicit
109 from_env = os.environ.get("OPENROUTER_API_KEY", "").strip()
110 if from_env:
111 return from_env
113 cwd = Path.cwd()
114 for directory in [cwd, *cwd.parents, Path(__file__).resolve().parent]:
115 env_file = directory / ".env"
116 if not env_file.is_file():
117 continue
118 try:
119 content = env_file.read_text(encoding="utf-8", errors="replace")
120 except OSError:
121 continue
122 for raw in content.splitlines():
123 line = raw.strip()
124 if line.startswith("export "):
125 line = line[len("export "):].strip()
126 if line.startswith("#") or "=" not in line:
127 continue
128 name, _, value = line.partition("=")
129 if name.strip() == "OPENROUTER_API_KEY":
130 value = value.strip().strip('"').strip("'")
131 if value:
132 return value
134 raise ApiError(
135 "OPENROUTER_API_KEY not found.\n"
136 " export OPENROUTER_API_KEY=your-key\n"
137 " or add OPENROUTER_API_KEY=your-key to a .env file\n"
138 " or pass --api-key\n"
139 "Keys: https://openrouter.ai/keys"
140 )
143def encode_reference(source: str) -> str:
144 """Return an image reference as a URL the API accepts.
146 HTTP(S) URLs and existing data URLs pass through; local paths are read and
147 encoded as base64 data URLs.
148 """
149 if source.startswith(("http://", "https://", "data:")):
150 return source
152 path = Path(source)
153 if not path.is_file():
154 raise ApiError(f"Reference image not found: {source}")
156 mime = INPUT_MIME.get(path.suffix.lower())
157 if mime is None:
158 supported = ", ".join(sorted(INPUT_MIME))
159 raise ApiError(
160 f"Unsupported reference image type '{path.suffix}' ({source}). "
161 f"Supported: {supported}"
162 )
164 encoded = base64.b64encode(path.read_bytes()).decode("ascii")
165 return f"data:{mime};base64,{encoded}"
168def request_bytes(
169 url: str,
170 api_key: str | None,
171 payload: dict | None,
172 timeout: float,
173 retries: int = 2,
174 accept: str = "application/json",
175 max_bytes: int | None = None,
176) -> bytes:
177 """POST (or GET when payload is None), retrying transient failures.
179 Rate limits and selected 5xx responses retry with bounded backoff. Other
180 HTTP errors are surfaced for inspection. Ambiguous transport failures only
181 retry GET requests, never generation POSTs.
182 """
183 headers = {"Accept": accept}
184 if api_key:
185 headers["Authorization"] = f"Bearer {api_key}"
187 data = None
188 if payload is not None:
189 data = json.dumps(payload).encode("utf-8")
190 headers["Content-Type"] = "application/json"
192 attempt = 0
193 while True:
194 req = urllib.request.Request(url, data=data, headers=headers)
195 try:
196 with urllib.request.urlopen(req, timeout=timeout) as resp:
197 if max_bytes is None:
198 return resp.read()
199 body = resp.read(max_bytes + 1)
200 if len(body) > max_bytes:
201 raise ApiError(f"Refusing a response larger than {max_bytes} bytes: {url}")
202 return body
203 except urllib.error.HTTPError as exc:
204 body = exc.read().decode("utf-8", errors="replace")
205 retryable = exc.code in RETRY_STATUS
206 if retryable and attempt < retries:
207 delay = retry_delay(exc.headers.get("Retry-After"), attempt)
208 print(
209 f"HTTP {exc.code} from OpenRouter; retrying in {delay:.0f}s "
210 f"({attempt + 1}/{retries})",
211 file=sys.stderr,
212 )
213 time.sleep(delay)
214 attempt += 1
215 continue
216 raise ApiError(f"OpenRouter returned HTTP {exc.code}: {error_detail(exc.code, body)}") from exc
217 except (OSError, http.client.HTTPException) as exc:
218 reason = getattr(exc, "reason", str(exc))
219 # A dropped connection does not prove a POST was never processed.
220 # Do not automatically submit another potentially billed generation.
221 if payload is None and attempt < retries:
222 delay = retry_delay(None, attempt)
223 print(
224 f"Could not reach OpenRouter ({reason}); retrying in {delay:.0f}s "
225 f"({attempt + 1}/{retries})",
226 file=sys.stderr,
227 )
228 time.sleep(delay)
229 attempt += 1
230 continue
231 hint = " Generation outcome is unknown; check activity before resubmitting." if payload is not None else ""
232 raise ApiError(f"Could not reach OpenRouter: {reason}.{hint}") from exc
235def retry_delay(retry_after: str | None, attempt: int) -> float:
236 """Seconds to wait before the next attempt: Retry-After, else backoff."""
237 if retry_after:
238 try:
239 return max(1.0, min(60.0, float(retry_after)))
240 except ValueError:
241 pass
242 return float(2 ** min(attempt, 6)) if attempt < 6 else 60.0
245def error_detail(status: int, body: str) -> str:
246 """Extract the API's error message and add a hint for the usual causes."""
247 detail = body
248 try:
249 parsed = json.loads(body)
250 detail = parsed.get("error", {}).get("message") or body
251 except (json.JSONDecodeError, AttributeError):
252 pass
254 hint = ""
255 if status == 400:
256 hint = (
257 "\nParameter support and allowed values are per-model. Inspect this model:\n"
258 " python generate_image.py --model-info MODEL"
259 )
260 elif status == 402:
261 hint = "\nCheck credits, key spending limits, and any in-flight budget Retry-After header."
262 elif status in (401, 403):
263 hint = (
264 "\nCheck the key and its credit balance: https://openrouter.ai/keys\n"
265 "A 403 can also reflect permissions, guardrails, or content policy; inspect the error."
266 )
267 return f"{detail}{hint}"
270def request_json(url: str, api_key: str | None, payload: dict | None, timeout: float,
271 retries: int = 2) -> Any:
272 """POST or GET and decode the JSON response."""
273 raw = request_bytes(url, api_key, payload, timeout, retries)
274 try:
275 result = json.loads(raw.decode("utf-8"))
276 except (json.JSONDecodeError, UnicodeDecodeError) as exc:
277 raise ApiError(f"OpenRouter returned a non-JSON response: {exc}") from exc
278 if not isinstance(result, dict):
279 raise ApiError("OpenRouter returned a JSON response that is not an object.")
280 if result.get("error"):
281 raise ApiError(f"OpenRouter returned an error: {error_detail(0, json.dumps(result))}")
282 return result
285def fetch_catalogue(timeout: float, retries: int = 2) -> dict[str, dict]:
286 """Return the image catalogue keyed by model id. Needs no API key."""
287 result = request_json(MODELS_URL, None, None, timeout, retries)
288 data = result.get("data")
289 if not isinstance(data, list) or any(not isinstance(m, dict) for m in data):
290 raise ApiError("Image model discovery returned no valid data array.")
291 return {m["id"]: m for m in data if m.get("id")}
294def fetch_endpoints(model_id: str, timeout: float, retries: int = 2) -> list[dict]:
295 """Public endpoint records contain definitive capabilities, unlike the union."""
296 result = request_json(f"{MODELS_URL}/{model_id}/endpoints", None, None, timeout, retries)
297 endpoints = result.get("endpoints")
298 if not isinstance(endpoints, list) or any(not isinstance(e, dict) for e in endpoints):
299 raise ApiError("Image endpoint discovery returned no endpoints array.")
300 return endpoints
303def describe_spec(spec: dict) -> str:
304 """Render one capability spec as the values or range it permits."""
305 if not isinstance(spec, dict):
306 return "supported"
307 if spec.get("type") == "enum":
308 return ", ".join(str(v) for v in spec.get("values", []))
309 if spec.get("type") == "range":
310 return f"{spec.get('min', '?')}-{spec.get('max', '?')}"
311 return "supported"
314def spec_problem(name: str, value: Any, spec: dict) -> str | None:
315 """Return a message if `value` violates `spec`, else None."""
316 if not isinstance(spec, dict):
317 return None
318 kind = spec.get("type")
319 if kind == "enum":
320 allowed = [str(v) for v in spec.get("values", [])]
321 if allowed and str(value) not in allowed:
322 return f"{name}={value} is not allowed; this model accepts: {', '.join(allowed)}"
323 elif kind == "range":
324 low, high = spec.get("min"), spec.get("max")
325 if isinstance(value, (int, float)):
326 if low is not None and value < low:
327 return f"{name}={value} is below this model's minimum of {low}"
328 if high is not None and value > high:
329 return f"{name}={value} exceeds this model's maximum of {high}"
330 return None
333def models_supporting(catalogue: dict[str, dict], parameter: str, value: Any = None) -> list[str]:
334 """Model ids that accept `parameter`, and `value` for it when given."""
335 matches = []
336 for model_id, model in catalogue.items():
337 spec = (model.get("supported_parameters") or {}).get(parameter)
338 if spec is None:
339 continue
340 if value is not None and spec_problem(parameter, value, spec):
341 continue
342 matches.append(model_id)
343 return sorted(matches)
346def preflight(payload: dict, catalogue: dict[str, dict], endpoints: list[dict] | None = None) -> None:
347 """Validate the request against the model's advertised capabilities.
349 This is a conservative check of advertised metadata, not a guarantee of
350 provider behavior or account access. Errors are collected so one run reports
351 all parameter problems.
352 """
353 model_id = payload["model"]
354 model = catalogue.get(model_id)
355 if model is None:
356 close = difflib.get_close_matches(model_id, sorted(catalogue), n=3, cutoff=0.4)
357 suggestion = f" Did you mean: {', '.join(close)}?" if close else ""
358 raise RequestRejected(
359 f"'{model_id}' is not in the OpenRouter image catalogue.{suggestion}\n"
360 " python generate_image.py --list-models"
361 )
363 supported = model.get("supported_parameters") or {}
364 problems: list[str] = []
366 for name in VALIDATED_PARAMETERS:
367 if name not in payload:
368 continue
369 spec = supported.get(name)
370 if spec is None:
371 others = models_supporting(catalogue, name, payload[name])
372 alternatives = f"\n Models that accept it: {', '.join(others[:6])}" if others else ""
373 problems.append(
374 f"{name} is not supported by {model_id}"
375 f" (it accepts: {', '.join(sorted(supported)) or 'nothing but model and prompt'})"
376 f"{alternatives}"
377 )
378 continue
379 issue = spec_problem(name, payload[name], spec)
380 if issue:
381 problems.append(issue)
383 references = payload.get("input_references") or []
384 spec = supported.get("input_references")
385 if spec is None:
386 if references:
387 problems.append(f"{model_id} does not accept reference images")
388 elif isinstance(spec, dict):
389 minimum, maximum = spec.get("min", 0), spec.get("max")
390 if minimum is not None and len(references) < minimum:
391 problems.append(f"{model_id} requires at least {minimum} reference images")
392 if maximum is not None and len(references) > maximum:
393 problems.append(f"{len(references)} reference images given; {model_id} accepts at most {maximum}")
395 if payload.get("background") == "transparent" and payload.get("output_format") == "jpeg":
396 problems.append("transparent background requires PNG or WebP, not JPEG")
398 if not problems and endpoints is not None:
399 # A union may advertise A and B even when no single endpoint accepts both.
400 for endpoint in endpoints:
401 try:
402 preflight(payload, {model_id: {"supported_parameters": endpoint.get("supported_parameters", {})}})
403 except RequestRejected:
404 continue
405 break
406 else:
407 problems.append("no advertised provider endpoint accepts this complete parameter combination")
409 if problems:
410 listed = "\n - ".join(problems)
411 raise RequestRejected(
412 f"Request rejected before billing ({len(problems)} problem"
413 f"{'s' if len(problems) > 1 else ''}):\n - {listed}\n"
414 f"Inspect the model: python generate_image.py --model-info {model_id}\n"
415 "Bypass this check with --no-preflight."
416 )
419def build_payload(args: argparse.Namespace) -> dict:
420 """Assemble the request body, omitting every parameter the caller left unset.
422 Omission matters: providers advertise different parameter sets, and an
423 unsupported field may be rejected or ignored.
424 """
425 payload: dict[str, Any] = {"model": args.model, "prompt": args.prompt}
427 optional = {
428 "n": args.n,
429 "aspect_ratio": args.aspect_ratio,
430 "resolution": args.resolution,
431 "quality": args.quality,
432 "output_format": args.output_format,
433 "background": args.background,
434 "output_compression": args.output_compression,
435 "seed": args.seed,
436 }
437 payload.update({key: value for key, value in optional.items() if value is not None})
439 if args.input:
440 payload["input_references"] = [
441 {"type": "image_url", "image_url": {"url": encode_reference(src)}}
442 for src in args.input
443 ]
445 return payload
448def redacted_payload(payload: dict) -> dict:
449 """Copy the payload with base64 reference blobs shortened, for printing."""
450 shown = dict(payload)
451 references = shown.get("input_references")
452 if references:
453 trimmed = []
454 for ref in references:
455 url = ((ref.get("image_url") or {}).get("url")) or ""
456 if url.startswith("data:") and len(url) > 80:
457 head = url.split(",", 1)[0]
458 url = f"{head},<{len(url)} chars of base64>"
459 trimmed.append({"type": ref.get("type"), "image_url": {"url": url}})
460 shown["input_references"] = trimmed
461 return shown
464def acceptable_suffixes(media_type: str) -> set[str]:
465 """Suffixes that correctly name `media_type`, including spelling variants."""
466 extension = EXTENSIONS.get(normalise_media_type(media_type), ".bin")
467 return SUFFIX_ALIASES.get(extension, {extension})
470def normalise_media_type(media_type: str) -> str:
471 """Strip parameters and case from a media type: 'IMAGE/PNG; x=1' -> 'image/png'."""
472 return media_type.split(";", 1)[0].strip().lower()
475def output_paths(requested: str | None, media_type: str, count: int) -> list[Path]:
476 """Choose filenames for the returned images.
478 The extension follows the model's media_type unless the caller named a file
479 explicitly, in which case their choice is kept and a real mismatch -- not a
480 mere spelling variant -- is reported.
481 """
482 extension = EXTENSIONS.get(normalise_media_type(media_type), ".bin")
484 if requested is None:
485 stem, suffix = Path("generated_image"), extension
486 else:
487 given = Path(requested)
488 stem, suffix = given.with_suffix(""), given.suffix or extension
489 if given.suffix and given.suffix.lower() not in acceptable_suffixes(media_type):
490 print(
491 f"Note: model returned {media_type} but --output ends in "
492 f"'{given.suffix}'; writing the returned bytes under that name.",
493 file=sys.stderr,
494 )
496 if count == 1:
497 return [stem.with_suffix(suffix)]
498 return [stem.parent / f"{stem.name}_{i + 1}{suffix}" for i in range(count)]
501def image_bytes(item: dict, timeout: float) -> bytes | None:
502 """Decode one response entry, downloading it if it arrived as a URL."""
503 payload = item.get("b64_json")
504 if payload:
505 if not isinstance(payload, str):
506 raise ApiError("Image response b64_json must be a string.")
507 if payload.startswith("data:") and "," in payload:
508 payload = payload.split(",", 1)[1]
509 try:
510 content = base64.b64decode(payload, validate=True)
511 except (binascii.Error, ValueError) as exc:
512 raise ApiError("Image response contains invalid base64 data.") from exc
513 if not content:
514 raise ApiError("Image response contains empty image data.")
515 return content
517 url = item.get("url")
518 if not url:
519 return None
521 # The documented response inlines base64, so this branch only runs if a
522 # provider hands back a link instead. The URL comes from the response rather
523 # than from the caller, so require HTTPS and cap what we are willing to read.
524 if not url.startswith("https://"):
525 raise ApiError(f"Refusing to fetch a non-HTTPS image URL from the response: {url}")
526 return request_bytes(
527 url, None, None, timeout, retries=1, accept="image/*", max_bytes=MAX_DOWNLOAD_BYTES
528 )
531def save_images(result: dict, requested_output: str | None, timeout: float = 60.0) -> list[Path]:
532 """Decode data[] into files and return the paths written."""
533 items = result.get("data")
534 if not isinstance(items, list) or not items:
535 raise ApiError(
536 "Response contained no images.\n"
537 f"Raw response: {json.dumps(result, indent=2)[:800]}"
538 )
540 decoded: list[tuple[dict, bytes]] = []
541 for item in items:
542 if not isinstance(item, dict):
543 raise ApiError("Image response contains a non-object data entry.")
544 content = image_bytes(item, timeout)
545 if content is None:
546 print(f"Skipping an entry with no image data: {list(item)}", file=sys.stderr)
547 continue
548 decoded.append((item, content))
550 if not decoded:
551 raise ApiError("Response carried image entries but none contained data.")
553 # Numbering is assigned after skipping empty entries so the written files
554 # are always _1.._n with no gaps.
555 written: list[Path] = []
557 for index, (item, content) in enumerate(decoded):
558 media_type = item.get("media_type") or "application/octet-stream"
559 path = output_paths(requested_output, media_type, len(decoded))[index]
560 if media_type == "application/octet-stream":
561 print("Note: response omitted media_type; verify the saved file's format.", file=sys.stderr)
562 if str(path.parent) not in ("", "."):
563 path.parent.mkdir(parents=True, exist_ok=True)
564 path.write_bytes(content)
565 written.append(path)
567 return written
570def format_cost(cost: Any) -> str:
571 """Render a USD amount in decimal, never scientific notation."""
572 if not isinstance(cost, (int, float)):
573 return str(cost)
574 return f"{cost:.8f}".rstrip("0").rstrip(".") or "0"
577def format_pricing(pricing: list[dict] | None) -> list[str]:
578 """Render an endpoint's pricing entries as readable lines."""
579 if not pricing:
580 return [" pricing: not published for this model"]
581 lines = []
582 for entry in pricing:
583 variant = f" ({entry['variant']})" if entry.get("variant") else ""
584 lines.append(
585 f" pricing: ${format_cost(entry.get('cost_usd'))} per {entry.get('unit', '?')}"
586 f" of {entry.get('billable', '?')}{variant}"
587 )
588 return lines
591def cost_lines(usage: dict) -> list[str]:
592 """Report OpenRouter charges and upstream cost separately, without double-counting."""
593 cost = usage.get("cost")
594 upstream = (usage.get("cost_details") or {}).get("upstream_inference_cost")
596 lines = []
597 if cost is not None:
598 lines.append(f"OpenRouter cost: ${format_cost(cost)} reported")
599 if upstream is not None:
600 label = " (BYOK)" if usage.get("is_byok") else ""
601 lines.append(f"Upstream inference cost{label}: ${format_cost(upstream)} reported")
603 image_tokens = (usage.get("completion_tokens_details") or {}).get("image_tokens")
604 if image_tokens:
605 lines.append(f"Image tokens: {image_tokens}")
606 return lines
609def print_model(model: dict) -> None:
610 """Print one catalogue entry with its per-parameter allowed values."""
611 params = model.get("supported_parameters") or {}
612 n_spec = params.get("n") or {}
613 refs = params.get("input_references") or {}
614 print(model.get("id", "?"))
615 per_request = f"up to {n_spec.get('max', 1)}" if n_spec else "1 (n not accepted)"
616 minimum_refs = refs.get("min", 0)
617 print(
618 f" images/request: {per_request}"
619 f" | reference images: {minimum_refs}-{refs.get('max', 0)}"
620 f" | streaming: {'yes' if model.get('supports_streaming') else 'no'}"
621 )
622 for name in sorted(params):
623 if name in ("n", "input_references"):
624 continue
625 print(f" {name}: {describe_spec(params[name])}")
628def list_models(timeout: float, substring: str = "") -> int:
629 """Print the image catalogue with each model's parameters and their values."""
630 catalogue = fetch_catalogue(timeout)
631 if not catalogue:
632 print("No models returned.", file=sys.stderr)
633 return 1
635 selected = [m for mid, m in sorted(catalogue.items()) if substring.lower() in mid.lower()]
636 if not selected:
637 print(f"No image model id contains '{substring}'.", file=sys.stderr)
638 return 1
640 scope = f" matching '{substring}'" if substring else ""
641 print(f"{len(selected)} image models available{scope}\n")
642 for model in selected:
643 print_model(model)
644 return 0
647def model_info(model_id: str, timeout: float) -> int:
648 """Print one model's allowed values, passthrough parameters, and pricing."""
649 catalogue = fetch_catalogue(timeout)
650 if model_id not in catalogue:
651 close = difflib.get_close_matches(model_id, sorted(catalogue), n=3, cutoff=0.4)
652 hint = f" Did you mean: {', '.join(close)}?" if close else ""
653 print(f"'{model_id}' is not in the image catalogue.{hint}", file=sys.stderr)
654 return 1
656 print_model(catalogue[model_id])
657 endpoints = fetch_endpoints(model_id, timeout)
658 if not endpoints:
659 print(" No provider endpoints currently advertised.")
660 for endpoint in endpoints:
661 print(f" provider: {endpoint.get('provider_name', '?')}")
662 print(f" options key: {endpoint.get('provider_slug')} | routing tag: {endpoint.get('provider_tag')}")
663 print(f" endpoint streaming: {bool(endpoint.get('supports_streaming'))}")
664 for name, spec in sorted((endpoint.get("supported_parameters") or {}).items()):
665 print(f" endpoint {name}: {describe_spec(spec)}")
666 for line in format_pricing(endpoint.get("pricing")):
667 print(line)
668 passthrough = endpoint.get("allowed_passthrough_parameters") or []
669 if passthrough:
670 print(f" passthrough (direct API only): {', '.join(passthrough)}")
671 return 0
674def main(argv: list[str] | None = None) -> int:
675 parser = argparse.ArgumentParser(
676 description="Generate or edit images via the OpenRouter Image API.",
677 formatter_class=argparse.RawDescriptionHelpFormatter,
678 epilog="""
679Examples:
680 # Generate with the default model
681 python generate_image.py "A beautiful sunset over mountains"
683 # Pick a model and shape
684 python generate_image.py "A cat in space" -m google/gemini-3-pro-image --aspect-ratio 16:9
686 # Edit an existing image
687 python generate_image.py "Make the sky purple" -i photo.jpg -o edited.png
689 # Composite from several references (model-dependent limit)
690 python generate_image.py "Blend these styles" -i a.png -i b.png -o blended.png
692 # Vector output
693 python generate_image.py "Minimal fox logo" -m recraft/recraft-v4-vector -o logo.svg
695 # Validate the request and print the payload without generating or billing
696 python generate_image.py "A cat astronaut" --resolution 4K --dry-run
698 # Inspect the catalogue, or one model's allowed values and pricing
699 python generate_image.py --list-models gemini
700 python generate_image.py --model-info openai/gpt-image-1
701""",
702 )
704 parser.add_argument("prompt", nargs="?", help="Description of the image, or the edit to apply")
705 parser.add_argument("--model", "-m", default=DEFAULT_MODEL,
706 help=f"Model slug (default: {DEFAULT_MODEL})")
707 parser.add_argument("--output", "-o",
708 help="Output path; extension defaults to the returned media type")
709 parser.add_argument("--input", "-i", action="append", metavar="IMAGE",
710 help="Reference image: local path, HTTP(S) URL, or data URL. Repeatable.")
711 parser.add_argument("--n", type=int, help="Number of images (model-dependent maximum)")
712 parser.add_argument("--aspect-ratio", help="e.g. 1:1, 16:9, 9:16, 4:3 (enum differs per model)")
713 parser.add_argument("--resolution", help="Tier: 512, 1K, 2K, 4K (support differs per model)")
714 parser.add_argument("--quality", help="Quality value; validated against live model capabilities")
715 parser.add_argument("--output-format", choices=["png", "jpeg", "webp", "svg"])
716 parser.add_argument("--background", choices=["auto", "transparent", "opaque"])
717 parser.add_argument("--output-compression", type=int, metavar="0-100",
718 help="Compression for webp/jpeg output")
719 parser.add_argument("--seed", type=int, help="Seed for deterministic output, where supported")
720 parser.add_argument("--api-key", help="Overrides OPENROUTER_API_KEY and any .env file")
721 parser.add_argument("--timeout", type=float, default=300.0,
722 help="Request timeout in seconds (default: 300)")
723 parser.add_argument("--retries", type=int, default=2, metavar="N",
724 help="Retries for rate limits and 5xx responses (default: 2)")
725 parser.add_argument("--no-preflight", action="store_true",
726 help="Skip the free capability check before the billed request")
727 parser.add_argument("--dry-run", action="store_true",
728 help="Validate and print the request, then exit without generating")
729 parser.add_argument("--list-models", nargs="?", const="", metavar="SUBSTRING",
730 help="List image models and their allowed values, then exit")
731 parser.add_argument("--model-info", metavar="MODEL",
732 help="Print one model's allowed values and pricing, then exit")
734 args = parser.parse_args(argv)
736 try:
737 if not math.isfinite(args.timeout) or args.timeout <= 0:
738 parser.error("--timeout must be a positive finite number")
739 if args.retries < 0:
740 parser.error("--retries must be nonnegative")
741 if args.list_models is not None:
742 return list_models(args.timeout, args.list_models)
744 if args.model_info:
745 return model_info(args.model_info, args.timeout)
747 if not args.prompt:
748 parser.error("a prompt is required unless --list-models or --model-info is given")
750 if args.output_compression is not None and not 0 <= args.output_compression <= 100:
751 parser.error("--output-compression must be between 0 and 100")
752 if args.n is not None and not 1 <= args.n <= 10:
753 parser.error("--n must be between 1 and 10")
755 payload = build_payload(args)
757 if not args.no_preflight:
758 catalogue = fetch_catalogue(args.timeout, args.retries)
759 preflight(payload, catalogue)
760 endpoints = fetch_endpoints(args.model, args.timeout, args.retries)
761 preflight(payload, catalogue, endpoints)
763 if args.dry_run:
764 print(f"POST {IMAGES_URL}")
765 print(json.dumps(redacted_payload(payload), indent=2))
766 print("Dry run: nothing generated, nothing billed.")
767 return 0
769 api_key = find_api_key(args.api_key)
771 action = "Editing" if args.input else "Generating"
772 print(f"{action} with {args.model}")
773 print(f"Prompt: {args.prompt}")
774 if args.input:
775 print(f"References: {len(args.input)}")
777 result = request_json(IMAGES_URL, api_key, payload, args.timeout, args.retries)
778 written = save_images(result, args.output, args.timeout)
779 if args.n is not None and len(written) < args.n:
780 print(f"Note: requested up to {args.n} images; received {len(written)}.", file=sys.stderr)
782 for path in written:
783 print(f"Saved {path}")
785 for line in cost_lines(result.get("usage") or {}):
786 print(line)
787 print("Open the image and check it before using it.")
788 return 0
790 except ApiError as exc:
791 print(f"Error: {exc}", file=sys.stderr)
792 return 1
793 except OSError as exc:
794 print(f"Error reading or writing an image: {exc}", file=sys.stderr)
795 return 1
796 except KeyboardInterrupt:
797 print("Interrupted.", file=sys.stderr)
798 return 130
801if __name__ == "__main__":
802 sys.exit(main())