feat(video): OpenRouter backend covers every model on the live video catalog

The salvaged #103267 plugin hardcoded a single model (minimax/hailuo-3-max) and
rejected any other id. OpenRouter's public GET /api/v1/videos/models already
publishes every generative model with its supported durations, resolutions,
aspect ratios, frame-image support, audio and seed flags, and pricing SKUs, so
the provider now reads that catalog (5-min TTL, offline snapshot fallback):

- list_models(): all 25+ generative models (edit/upscale/avatar rows that take
  no duration are outside the unified video_generate surface and are dropped)
  with a per-second price label where the SKU is per-second
- capabilities(): the CONFIGURED model's surface, so the dynamic schema only
  advertises audio/seed/resolutions the selected model honours
- _build_payload(): clamps duration/resolution/aspect ratio to the model's
  live limits (nearest by value/height/ratio) and drops generate_audio/seed
  for models that lack them (the API 400s otherwise); reference images ride
  in input_references; local file inputs are refused (OpenRouter fetches
  URLs itself), data:image/ URLs from the sandbox chokepoint pass through
- bearer key only ever goes to the configured origin (poll + /content),
  never to a provider-supplied unsigned_urls host (kept from #103267)

Also drops the source-grep `_IGNORES_SEED` escape hatch #103267 added to the
declaration⇄implementation sweep; the provider now implements seed for real.
Docs list OpenRouter and DeepInfra as bundled video backends.

Requested by Don Piedro Savastano (Discord): OpenRouter credit for video_generate.
This commit is contained in:
teknium1
2026-09-12 12:58:29 -07:00
committed by Teknium
parent 387ac50d85
commit c6f87deb2c
7 changed files with 358 additions and 501 deletions

View File

@@ -1,340 +1,339 @@
"""OpenRouter MiniMax H3 Max video generation backend. """OpenRouter video generation backend.
Uses OpenRouter's dedicated asynchronous video API rather than the chat OpenRouter fronts every major video model (Veo, Sora, Kling, Seedance, Wan, Hailuo, Grok
completions API: submit ``POST /api/v1/videos``, poll the returned job, then Imagine, FLUX Video, …) behind one asynchronous API: ``POST /api/v1/videos`` → poll
materialize the completed clip under Hermes' video cache. ``GET /api/v1/videos/{id}`` → download ``GET /api/v1/videos/{id}/content`` with the
bearer key. The catalog and each model's limits come live from ``GET /api/v1/videos/models``
(public, no key), so new releases are selectable without a patch and a retired model drops
out on its own. ``capabilities()`` reflects the *selected* model so the dynamic
``video_generate`` schema advertises only what that model honours.
Docs: https://openrouter.ai/docs/guides/overview/multimodal/video-generation
""" """
from __future__ import annotations from __future__ import annotations
import ipaddress
import logging import logging
import os import os
import time import time
from typing import Any, Dict, List, Optional from typing import Any, Dict, List, Optional, Tuple
from urllib.parse import urlsplit
from agent.video_gen_provider import ( from agent.video_gen_provider import VideoGenProvider, error_response, save_url_video, success_response
VideoGenProvider,
error_response,
save_url_video,
success_response,
)
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
DEFAULT_MODEL = "minimax/hailuo-3-max" DEFAULT_MODEL = "minimax/hailuo-3-max"
DEFAULT_BASE_URL = "https://openrouter.ai/api/v1" DEFAULT_BASE_URL = "https://openrouter.ai/api/v1"
DEFAULT_DURATION = 5 _CATALOG_TTL_S = 300.0
DEFAULT_RESOLUTION = "768p" _CATALOG_TIMEOUT_S = 10.0
DEFAULT_ASPECT_RATIO = "16:9"
SUPPORTED_DURATIONS = tuple(range(5, 16))
SUPPORTED_RESOLUTIONS = ("480p", "768p")
SUPPORTED_ASPECT_RATIOS = ("21:9", "16:9", "4:3", "1:1", "3:4", "9:16")
_TERMINAL_STATUSES = frozenset({"completed", "failed", "cancelled", "canceled", "expired"}) _TERMINAL_STATUSES = frozenset({"completed", "failed", "cancelled", "canceled", "expired"})
# Every value the API accepts (docs "Supported Resolutions / Aspect Ratios"); the union fallback
# when the live catalog is unreachable. Heights drive nearest-match clamping.
_RESOLUTION_HEIGHT = {"480p": 480, "540p": 540, "720p": 720, "768p": 768, "1080p": 1080, "1K": 1024, "2K": 1440, "4K": 2160}
_ALL_ASPECT_RATIOS = ("16:9", "9:16", "1:1", "4:3", "3:4", "3:2", "2:3", "21:9", "9:21")
_MAX_REFERENCE_IMAGES = 3 # image references are accepted by every provider per the API schema
# Offline snapshot so the picker/default work before the first successful catalog fetch.
_FALLBACK_CATALOG: List[Dict[str, Any]] = [
{"id": DEFAULT_MODEL, "name": "MiniMax: Hailuo 3 Max", "supported_durations": list(range(5, 16)),
"supported_resolutions": ["768p", "480p"], "supported_aspect_ratios": ["21:9", "16:9", "4:3", "1:1", "3:4", "9:16"],
"supported_frame_images": ["first_frame", "last_frame"], "generate_audio": False, "seed": False,
"pricing_skus": {"duration_seconds_480p": "0.05", "duration_seconds_768p": "0.08"}},
]
def _is_public_first_frame_url(value: str) -> bool: def _price_label(skus: Any) -> str:
"""Accept provider-fetchable HTTPS URLs, never local/private inputs.""" """Compact per-second price from ``pricing_skus`` (``$0.10/s`` or ``$0.05–0.28/s``); ``""`` when the
try: SKU shape is token- or megapixel-priced (Seedance, FLUX upscale) — a wrong number is worse than none."""
parsed = urlsplit(value) if not isinstance(skus, dict):
host = (parsed.hostname or "").strip().lower().rstrip(".") return ""
if parsed.scheme.lower() != "https" or not host: per_second: List[float] = []
return False for key, value in skus.items():
if host == "localhost" or host.endswith((".localhost", ".local", ".lan", ".internal")):
return False
try: try:
return ipaddress.ip_address(host).is_global amount = float(value)
except ValueError: except (TypeError, ValueError):
return True continue
if key.startswith("duration_seconds"):
per_second.append(amount)
elif key.startswith(("cents_per_second_output", "cents_per_video_output_second")):
per_second.append(amount / 100)
if not per_second:
return ""
lo, hi = min(per_second), max(per_second)
return f"${lo:.2f}/s" if lo == hi else f"${lo:.2f}–{hi:.2f}/s"
def _ratio(value: Any) -> Optional[float]:
try:
w, h = str(value).split(":")
return float(w) / float(h)
except (ValueError, ZeroDivisionError):
return None
def _nearest(value: Any, supported: List[Any], height: Optional[Dict[str, int]] = None) -> Any:
"""Closest supported value (durations by distance, resolutions by pixel height, aspect ratios by
numeric ratio); the request is otherwise a guaranteed 400 because the tool always sends a default
resolution/aspect ratio."""
if not supported or value in supported:
return value
if height is not None:
target = height.get(str(value))
if target is None:
return supported[0]
scored = [(abs(height[str(s)] - target), i, s) for i, s in enumerate(supported) if str(s) in height]
return min(scored)[2] if scored else supported[0]
if ":" in str(value):
target = _ratio(value)
if target is None:
return supported[0]
scored = [(abs(r - target), i, s) for i, s in enumerate(supported) if (r := _ratio(s)) is not None]
return min(scored)[2] if scored else supported[0]
try:
return min(supported, key=lambda s: (abs(int(s) - int(value)), s))
except (TypeError, ValueError): except (TypeError, ValueError):
return False return supported[0]
def _nearest(value: int, supported: tuple[int, ...]) -> int: def _is_generative(entry: Dict[str, Any]) -> bool:
return min(supported, key=lambda candidate: (abs(candidate - value), candidate)) """Text/image-to-video models declare durations; edit, upscale and avatar models (video/audio input)
do not and are outside the unified ``video_generate`` surface."""
return bool(entry.get("supported_durations"))
def _entry_capabilities(entry: Dict[str, Any]) -> Dict[str, Any]:
durations = [int(d) for d in entry.get("supported_durations") or [] if isinstance(d, (int, float))]
return {
"modalities": ["text", "image"] if entry.get("supported_frame_images") else ["text"],
"aspect_ratios": list(entry.get("supported_aspect_ratios") or _ALL_ASPECT_RATIOS),
"resolutions": list(entry.get("supported_resolutions") or _RESOLUTION_HEIGHT),
"max_duration": max(durations) if durations else 30, "min_duration": min(durations) if durations else 1,
"supports_audio": bool(entry.get("generate_audio")), "supports_negative_prompt": False,
"supports_seed": bool(entry.get("seed")), "supports_upscale": False,
"max_reference_images": _MAX_REFERENCE_IMAGES,
}
def _image_part(url: str) -> Dict[str, Any]:
return {"type": "image_url", "image_url": {"url": url}}
def _acceptable_image_ref(value: str) -> bool:
"""OpenRouter fetches the frame itself, so it needs a public HTTPS URL or an inline ``data:image/`` URL
(what the sandbox confinement chokepoint hands us for local files)."""
lowered = value.lower()
return lowered.startswith("https://") or lowered.startswith("data:image/")
def _build_payload( def _build_payload(
*, entry: Dict[str, Any], *, model: str, prompt: str, image_url: Optional[str], reference_image_urls: Optional[List[str]],
prompt: str, duration: Optional[int], aspect_ratio: str, resolution: str, audio: Optional[bool], seed: Optional[int],
image_url: Optional[str],
duration: Optional[int],
aspect_ratio: str,
resolution: str,
) -> Dict[str, Any]: ) -> Dict[str, Any]:
"""Translate the unified Hermes inputs to OpenRouter's video API.""" """Unified inputs → OpenRouter body, clamped to the model's live limits; unsupported toggles are dropped
requested_duration = duration if duration is not None else DEFAULT_DURATION rather than sent (the API 400s on ``seed``/``generate_audio`` for models that lack them)."""
try: payload: Dict[str, Any] = {"model": model, "prompt": prompt}
requested_duration = int(requested_duration) if aspect_ratio:
except (TypeError, ValueError): payload["aspect_ratio"] = _nearest(aspect_ratio, list(entry.get("supported_aspect_ratios") or []))
requested_duration = DEFAULT_DURATION if resolution:
payload["resolution"] = _nearest(resolution, list(entry.get("supported_resolutions") or []), _RESOLUTION_HEIGHT)
payload: Dict[str, Any] = { if duration:
"model": DEFAULT_MODEL, payload["duration"] = int(_nearest(int(duration), list(entry.get("supported_durations") or [])))
"prompt": prompt,
"duration": _nearest(requested_duration, SUPPORTED_DURATIONS),
"resolution": resolution if resolution in SUPPORTED_RESOLUTIONS else DEFAULT_RESOLUTION,
"aspect_ratio": (
aspect_ratio if aspect_ratio in SUPPORTED_ASPECT_RATIOS else DEFAULT_ASPECT_RATIO
),
}
if image_url: if image_url:
payload["frame_images"] = [ payload["frame_images"] = [{**_image_part(image_url), "frame_type": "first_frame"}]
{ if reference_image_urls:
"type": "image_url", payload["input_references"] = [_image_part(u) for u in reference_image_urls[:_MAX_REFERENCE_IMAGES]]
"image_url": {"url": image_url}, if audio is not None and entry.get("generate_audio"):
"frame_type": "first_frame", payload["generate_audio"] = bool(audio)
} if seed is not None and entry.get("seed"):
] payload["seed"] = int(seed)
return payload return payload
class OpenRouterVideoGenProvider(VideoGenProvider): class OpenRouterVideoGenProvider(VideoGenProvider):
"""MiniMax H3 Max text-to-video and first-frame image-to-video.""" """Every generative model on OpenRouter's video API, catalog and limits discovered live."""
_IGNORES_SEED = True # Unified ABC input; H3 Max's OpenRouter SKU rejects it. name = "openrouter"
_poll_interval_s = 5.0 display_name = "OpenRouter"
_poll_interval_s = 10.0
_poll_deadline_s = 900.0 _poll_deadline_s = 900.0
_request_timeout_s = 60.0 _request_timeout_s = 60.0
def __init__(self) -> None:
self._catalog_cache: Optional[Tuple[List[Dict[str, Any]], float]] = None
@property # ---- credentials / transport -------------------------------------------------------------
def name(self) -> str:
return "openrouter"
@property
def display_name(self) -> str:
return "OpenRouter"
def _api_key(self) -> str: def _api_key(self) -> str:
return os.getenv("OPENROUTER_API_KEY", "").strip() return os.environ.get("OPENROUTER_API_KEY", "").strip()
def _base_url(self) -> str: def _base_url(self) -> str:
return os.getenv("OPENROUTER_BASE_URL", DEFAULT_BASE_URL).strip().rstrip("/") return (os.environ.get("OPENROUTER_BASE_URL", "").strip() or DEFAULT_BASE_URL).rstrip("/")
def _session(self):
import requests
return requests.Session()
def _headers(self) -> Dict[str, str]: def _headers(self) -> Dict[str, str]:
return { return {"Authorization": f"Bearer {self._api_key()}", "Content-Type": "application/json",
"Authorization": f"Bearer {self._api_key()}", "HTTP-Referer": "https://github.com/NousResearch/hermes-agent", "X-Title": "Hermes Agent"}
"Content-Type": "application/json",
"User-Agent": "hermes-agent/video_gen", def _session(self) -> Any:
} import requests
return requests.Session()
def is_available(self) -> bool: def is_available(self) -> bool:
return bool(self._api_key()) return bool(self._api_key())
# ---- catalog -------------------------------------------------------------------------------
def _catalog(self) -> List[Dict[str, Any]]:
"""Live ``/videos/models`` entries (public endpoint), cached per TTL; the snapshot when unreachable."""
if self._catalog_cache and time.monotonic() - self._catalog_cache[1] < _CATALOG_TTL_S:
return self._catalog_cache[0]
entries: List[Dict[str, Any]] = []
try:
import requests
response = requests.get(f"{self._base_url()}/videos/models", timeout=_CATALOG_TIMEOUT_S)
response.raise_for_status()
data = response.json().get("data")
entries = [e for e in (data if isinstance(data, list) else []) if isinstance(e, dict) and e.get("id")]
except Exception as exc: # noqa: BLE001 — offline picker keeps working on the snapshot
logger.debug("OpenRouter video catalog unavailable: %s", exc)
if not entries:
return _FALLBACK_CATALOG
self._catalog_cache = (entries, time.monotonic())
return entries
def _entry(self, model_id: str) -> Dict[str, Any]:
"""Catalog row for *model_id*; ``{}`` for an unknown id (request passes through unclamped so a
brand-new model works before our cache refreshes — the API validates)."""
return next((e for e in self._catalog() if e.get("id") == model_id), {})
def _configured_model(self) -> str:
try:
from hermes_cli.config import cfg_get, load_config
value = cfg_get(load_config(), "video_gen", "model")
except Exception as exc: # noqa: BLE001
logger.debug("Could not read video_gen.model: %s", exc)
value = None
return value.strip() if isinstance(value, str) and value.strip() else DEFAULT_MODEL
def list_models(self) -> List[Dict[str, Any]]: def list_models(self) -> List[Dict[str, Any]]:
return [ rows = []
{ for entry in self._catalog():
"id": DEFAULT_MODEL, if not _is_generative(entry):
"display": "MiniMax H3 Max", continue
"speed": "~20-60s", caps = _entry_capabilities(entry)
"strengths": "Fast text-to-video and first-frame image-to-video.", extras = [f"{caps['min_duration']}-{caps['max_duration']}s", "/".join(caps["resolutions"])]
"price": "$0.05/s (480p), $0.08/s (768p)", extras += ["audio"] if caps["supports_audio"] else []
"modalities": ["text", "image"], extras += ["i2v"] if "image" in caps["modalities"] else []
} rows.append({"id": entry["id"], "display": entry.get("name") or entry["id"], "strengths": ", ".join(extras),
] "price": _price_label(entry.get("pricing_skus")), "modalities": caps["modalities"],
"min_duration": caps["min_duration"], "max_duration": caps["max_duration"]})
return rows
def default_model(self) -> Optional[str]: def default_model(self) -> Optional[str]:
return DEFAULT_MODEL return DEFAULT_MODEL
def capabilities(self) -> Dict[str, Any]: def capabilities(self) -> Dict[str, Any]:
return { """The selected model's live surface; the API-wide union when the id is unknown or the catalog is down.
"modalities": ["text", "image"], ``supports_seed``/``supports_audio`` are per-model (Veo/Wan/Seedance: yes; Hailuo/Grok: no)."""
"aspect_ratios": list(SUPPORTED_ASPECT_RATIOS), entry = self._entry(self._configured_model())
"resolutions": list(SUPPORTED_RESOLUTIONS), if entry:
"max_duration": max(SUPPORTED_DURATIONS), return _entry_capabilities(entry)
"min_duration": min(SUPPORTED_DURATIONS), return {"modalities": ["text", "image"], "aspect_ratios": list(_ALL_ASPECT_RATIOS),
"supports_audio": False, "resolutions": list(_RESOLUTION_HEIGHT), "max_duration": 30, "min_duration": 1,
"supports_negative_prompt": False, "supports_audio": True, "supports_negative_prompt": False, "supports_seed": True,
"supports_seed": False, "supports_upscale": False, "max_reference_images": _MAX_REFERENCE_IMAGES}
"supports_upscale": False,
"max_reference_images": 0,
}
def get_setup_schema(self) -> Dict[str, Any]: def get_setup_schema(self) -> Dict[str, Any]:
return { return {"name": "OpenRouter", "badge": "paid",
"name": "OpenRouter", "tag": "Veo 3.1, Sora 2 Pro, Kling 3, Seedance 2, Wan 3, Hailuo 3, Grok Imagine & more — live catalog; "
"badge": "paid", "text-to-video, image-to-video & reference-to-video; uses OPENROUTER_API_KEY",
"tag": "MiniMax H3 Max — text-to-video and first-frame image-to-video", "env_vars": [{"key": "OPENROUTER_API_KEY", "prompt": "OpenRouter API key", "url": "https://openrouter.ai/settings/keys"}]}
"env_vars": [
{
"key": "OPENROUTER_API_KEY",
"prompt": "OpenRouter API key",
"url": "https://openrouter.ai/settings/keys",
}
],
}
# ---- generation ----------------------------------------------------------------------------
def _poll(self, session: Any, job_id: str) -> Dict[str, Any]: def _poll(self, session: Any, job_id: str) -> Dict[str, Any]:
deadline = time.monotonic() + self._poll_deadline_s deadline = time.monotonic() + self._poll_deadline_s
url = f"{self._base_url()}/videos/{job_id}" url = f"{self._base_url()}/videos/{job_id}"
last_status = "unknown" last_status = "unknown"
def raise_timeout() -> None:
raise TimeoutError(
f"video job {job_id} did not finish within {int(self._poll_deadline_s)}s "
f"(last status={last_status})"
)
while True: while True:
remaining = deadline - time.monotonic() remaining = deadline - time.monotonic()
if remaining <= 0: if remaining <= 0:
raise_timeout() raise TimeoutError(f"video job {job_id} did not finish within {int(self._poll_deadline_s)}s (last status={last_status})")
response = session.get( response = session.get(url, headers=self._headers(), timeout=max(0.001, min(self._request_timeout_s, remaining)))
url,
headers=self._headers(),
timeout=max(0.001, min(self._request_timeout_s, remaining)),
)
response.raise_for_status() response.raise_for_status()
payload = response.json() payload = response.json()
last_status = str(payload.get("status") or "").lower() or "unknown" last_status = str(payload.get("status") or "").lower() or "unknown"
remaining = deadline - time.monotonic()
if remaining <= 0:
raise_timeout()
if last_status in _TERMINAL_STATUSES: if last_status in _TERMINAL_STATUSES:
return payload return payload
time.sleep(min(self._poll_interval_s, remaining)) time.sleep(min(self._poll_interval_s, max(0.0, deadline - time.monotonic())))
def _save_completed_video(self, job_id: str) -> str: def _save_completed_video(self, job_id: str) -> str:
"""Stream the authenticated content endpoint through the shared size cap. # The content endpoint is derived from our configured origin, never from ``unsigned_urls``: the
# bearer key must only ever be sent to the host the operator selected.
Do not follow a provider-supplied ``unsigned_urls`` destination: deriving return str(save_url_video(f"{self._base_url()}/videos/{job_id}/content", prefix="openrouter",
the endpoint from the configured API origin keeps the bearer request on headers=self._headers(), require_video_content_type=True))
the same operator-selected host and avoids a second unbounded code path.
"""
return str(
save_url_video(
f"{self._base_url()}/videos/{job_id}/content",
prefix="openrouter-hailuo",
headers=self._headers(),
require_video_content_type=True,
)
)
def generate( def generate(
self, self, prompt: str, *, model: Optional[str] = None, image_url: Optional[str] = None,
prompt: str, reference_image_urls: Optional[List[str]] = None, duration: Optional[int] = None,
*, aspect_ratio: str = "16:9", resolution: str = "720p", negative_prompt: Optional[str] = None,
model: Optional[str] = None, audio: Optional[bool] = None, seed: Optional[int] = None, **kwargs: Any,
image_url: Optional[str] = None,
reference_image_urls: Optional[List[str]] = None,
duration: Optional[int] = None,
aspect_ratio: str = DEFAULT_ASPECT_RATIO,
resolution: str = DEFAULT_RESOLUTION,
negative_prompt: Optional[str] = None,
audio: Optional[bool] = None,
seed: Optional[int] = None,
**kwargs: Any,
) -> Dict[str, Any]: ) -> Dict[str, Any]:
del negative_prompt, audio, seed, kwargs del negative_prompt, kwargs # no top-level negative_prompt on this API; unknown kwargs are ignored per the ABC
cleaned_prompt = (prompt or "").strip() prompt = (prompt or "").strip()
if not cleaned_prompt: model_id = (model or "").strip() or self._configured_model()
return error_response(
error="prompt is required",
error_type="invalid_request",
provider=self.name,
)
model_id = (model or DEFAULT_MODEL).strip()
if model_id != DEFAULT_MODEL:
return error_response(
error=f"OpenRouter video currently supports only {DEFAULT_MODEL}",
error_type="invalid_model",
provider=self.name,
model=model_id,
prompt=cleaned_prompt,
)
if reference_image_urls:
return error_response(
error=f"{DEFAULT_MODEL} does not support reference_image_urls",
error_type="unsupported_input",
provider=self.name,
model=model_id,
prompt=cleaned_prompt,
)
cleaned_image_url = (image_url or "").strip() or None
if cleaned_image_url and not _is_public_first_frame_url(cleaned_image_url):
return error_response(
error="image_url must be a public HTTPS URL",
error_type="invalid_request",
provider=self.name,
model=model_id,
prompt=cleaned_prompt,
)
if not self._api_key():
return error_response(
error="OPENROUTER_API_KEY is not set",
error_type="missing_credentials",
provider=self.name,
model=model_id,
prompt=cleaned_prompt,
)
payload = _build_payload( def fail(error: str, error_type: str) -> Dict[str, Any]:
prompt=cleaned_prompt, return error_response(error=error, error_type=error_type, provider=self.name, model=model_id, prompt=prompt,
image_url=cleaned_image_url, aspect_ratio=aspect_ratio)
duration=duration,
aspect_ratio=aspect_ratio, if not prompt:
resolution=resolution, return fail("prompt is required", "invalid_request")
) if not self._api_key():
return fail("OPENROUTER_API_KEY is not set", "missing_credentials")
image_url = (image_url or "").strip() or None
refs = [r.strip() for r in (reference_image_urls or []) if isinstance(r, str) and r.strip()]
for ref in ([image_url] if image_url else []) + refs:
if not _acceptable_image_ref(ref):
return fail("image inputs must be public HTTPS URLs or data:image/ URLs (OpenRouter fetches them itself)",
"invalid_request")
entry = self._entry(model_id)
payload = _build_payload(entry, model=model_id, prompt=prompt, image_url=image_url, reference_image_urls=refs,
duration=duration, aspect_ratio=aspect_ratio, resolution=resolution, audio=audio, seed=seed)
session = self._session() session = self._session()
try: try:
submitted = session.post( submitted = session.post(f"{self._base_url()}/videos", headers=self._headers(), json=payload,
f"{self._base_url()}/videos", timeout=self._request_timeout_s)
headers=self._headers(), if submitted.status_code >= 400:
json=payload, detail = ""
timeout=self._request_timeout_s, try:
) detail = str((submitted.json().get("error") or {}).get("message") or "")
submitted.raise_for_status() except Exception: # noqa: BLE001 — non-JSON error body
submission = submitted.json() detail = ""
job_id = str(submission.get("id") or "").strip() return fail(f"OpenRouter rejected the request (HTTP {submitted.status_code}): {detail or submitted.text[:300]}",
"api_error")
job_id = str(submitted.json().get("id") or "").strip()
if not job_id: if not job_id:
raise ValueError("OpenRouter submit response did not contain a job id") return fail("OpenRouter submit response did not contain a job id", "api_error")
job = self._poll(session, job_id) job = self._poll(session, job_id)
status = str(job.get("status") or "").lower() status = str(job.get("status") or "").lower()
if status != "completed": if status != "completed":
return error_response( return fail(str(job.get("error") or f"video job ended with status={status!r}"), "job_failed")
error=str(job.get("error") or f"video job ended with status={status!r}"),
error_type="job_failed",
provider=self.name,
model=model_id,
prompt=cleaned_prompt,
aspect_ratio=payload["aspect_ratio"],
)
video_path = self._save_completed_video(job_id) video_path = self._save_completed_video(job_id)
except Exception as exc: # noqa: BLE001 - normalize transport/API failures for tool callers except Exception as exc: # noqa: BLE001 — normalize transport/timeout failures for tool callers
logger.debug("OpenRouter H3 Max video generation failed", exc_info=True) logger.debug("OpenRouter video generation failed", exc_info=True)
return error_response( return fail(f"OpenRouter video generation failed: {exc}", "api_error")
error=f"OpenRouter video generation failed: {exc}", finally:
error_type="api_error", close = getattr(session, "close", None)
provider=self.name, if callable(close):
model=model_id, close()
prompt=cleaned_prompt,
aspect_ratio=payload["aspect_ratio"],
)
raw_usage = job.get("usage") raw_usage = job.get("usage")
usage: Dict[str, Any] = raw_usage if isinstance(raw_usage, dict) else {} usage: Dict[str, Any] = raw_usage if isinstance(raw_usage, dict) else {}
extra: Dict[str, Any] = {"job_id": job_id} extra: Dict[str, Any] = {"job_id": job_id, **({"cost": usage["cost"]} if usage.get("cost") is not None else {})}
if usage.get("cost") is not None:
extra["cost"] = usage["cost"]
return success_response( return success_response(
video=video_path, video=video_path, model=model_id, prompt=prompt, modality="image" if image_url else "text",
model=model_id, aspect_ratio=str(payload.get("aspect_ratio") or ""), duration=int(payload.get("duration") or 0),
prompt=cleaned_prompt, provider=self.name, extra=extra)
modality="image" if cleaned_image_url else "text",
aspect_ratio=payload["aspect_ratio"],
duration=payload["duration"],
provider=self.name,
extra=extra,
)
def register(ctx) -> None: def register(ctx) -> None:
"""Plugin entry point.""" """Plugin entry point — wire ``OpenRouterVideoGenProvider`` into the registry."""
ctx.register_video_gen_provider(OpenRouterVideoGenProvider()) ctx.register_video_gen_provider(OpenRouterVideoGenProvider())

View File

@@ -1,6 +1,6 @@
name: openrouter name: openrouter
version: 1.0.0 version: 1.0.0
description: "OpenRouter MiniMax H3 Max video generation via the asynchronous /api/v1/videos API." description: "OpenRouter video generation backend. Every generative model on the /api/v1/videos API (Veo, Sora, Kling, Seedance, Wan, Hailuo, Grok Imagine, FLUX Video, …) — text-to-video, image-to-video and reference-to-video; catalog and per-model limits discovered live."
author: Nous Research author: Nous Research
kind: backend kind: backend
requires_env: requires_env:

View File

@@ -1,247 +1,134 @@
"""OpenRouter video_gen plugin — live-catalog shape, per-model clamping, and the submit→poll→download flow."""
from __future__ import annotations from __future__ import annotations
from dataclasses import dataclass from dataclasses import dataclass, field
import pytest
from tools.video_generation_tool import VIDEO_GENERATE_SCHEMA
from agent import video_gen_registry from agent import video_gen_registry
from plugins.video_gen.openrouter import ( from plugins.video_gen.openrouter import OpenRouterVideoGenProvider, _build_payload
DEFAULT_MODEL,
OpenRouterVideoGenProvider, _VEO = {"id": "google/veo-3.1", "name": "Google: Veo 3.1", "supported_durations": [4, 6, 8],
_build_payload, "supported_resolutions": ["720p", "1080p", "4K"], "supported_aspect_ratios": ["16:9", "9:16"],
) "supported_frame_images": ["first_frame", "last_frame"], "generate_audio": True, "seed": True,
"pricing_skus": {"duration_seconds_with_audio": "0.40", "duration_seconds_without_audio": "0.20"}}
_HAILUO = {"id": "minimax/hailuo-3-max", "name": "MiniMax: Hailuo 3 Max", "supported_durations": list(range(5, 16)),
"supported_resolutions": ["768p", "480p"], "supported_aspect_ratios": ["21:9", "16:9", "4:3", "1:1", "3:4", "9:16"],
"supported_frame_images": ["first_frame"], "generate_audio": False, "seed": False,
"pricing_skus": {"duration_seconds_480p": "0.05", "duration_seconds_768p": "0.08"}}
_EDIT = {"id": "black-forest-labs/flux-video-edit", "name": "FLUX Video Edit", "supported_durations": None,
"supported_resolutions": None, "supported_aspect_ratios": None, "supported_frame_images": None,
"generate_audio": False, "seed": False, "pricing_skus": {"cents_per_second_output": "3"}}
def _provider(monkeypatch, catalog, configured="google/veo-3.1"):
provider = OpenRouterVideoGenProvider()
monkeypatch.setattr(provider, "_catalog", lambda: catalog)
monkeypatch.setattr(provider, "_configured_model", lambda: configured)
return provider
def test_catalog_drives_picker_rows_and_selected_model_capabilities(monkeypatch):
"""Rows come from the live catalog minus edit/upscale models; capabilities() follows the CONFIGURED
model (Veo: audio+seed; Hailuo: neither) so the dynamic schema never advertises a dead toggle."""
provider = _provider(monkeypatch, [_VEO, _HAILUO, _EDIT], configured="google/veo-3.1")
rows = provider.list_models()
assert [r["id"] for r in rows] == ["google/veo-3.1", "minimax/hailuo-3-max"]
assert rows[0]["price"] == "$0.20–0.40/s" and rows[1]["max_duration"] == 15
veo = provider.capabilities()
assert veo["supports_audio"] and veo["supports_seed"] and veo["resolutions"] == ["720p", "1080p", "4K"]
monkeypatch.setattr(provider, "_configured_model", lambda: "minimax/hailuo-3-max")
hailuo = provider.capabilities()
assert not hailuo["supports_audio"] and not hailuo["supports_seed"] and hailuo["max_duration"] == 15
def test_payload_clamps_to_model_limits_and_drops_unsupported_toggles():
payload = _build_payload(_HAILUO, model=_HAILUO["id"], prompt="p", image_url="https://x/a.png",
reference_image_urls=["https://x/r.png"], duration=99, aspect_ratio="2:3",
resolution="720p", audio=True, seed=7)
assert payload["duration"] == 15 and payload["resolution"] == "768p" and payload["aspect_ratio"] == "3:4"
assert payload["frame_images"][0]["frame_type"] == "first_frame"
assert payload["input_references"] == [{"type": "image_url", "image_url": {"url": "https://x/r.png"}}]
assert "generate_audio" not in payload and "seed" not in payload # Hailuo lacks both → would 400
veo = _build_payload(_VEO, model=_VEO["id"], prompt="p", image_url=None, reference_image_urls=None,
duration=5, aspect_ratio="16:9", resolution="1080p", audio=False, seed=7)
assert veo["duration"] == 4 and veo["generate_audio"] is False and veo["seed"] == 7
@dataclass @dataclass
class _Response: class _Response:
payload: dict payload: dict
status_code: int = 200 status_code: int = 200
content: bytes = b"" text: str = ""
def json(self): def json(self):
return self.payload return self.payload
def raise_for_status(self): def raise_for_status(self):
if self.status_code >= 400: if self.status_code >= 400:
raise RuntimeError(f"HTTP {self.status_code}: {self.payload}") raise RuntimeError(f"HTTP {self.status_code}")
@dataclass
class _Session: class _Session:
def __init__(self): posts: list = field(default_factory=list)
self.posts = [] gets: list = field(default_factory=list)
self.gets = [] polls: list = field(default_factory=lambda: [
self.polls = [ _Response({"id": "job-1", "status": "in_progress"}),
_Response({"id": "job-1", "status": "in_progress"}), _Response({"id": "job-1", "status": "completed", "unsigned_urls": ["https://evil.example/steal"],
_Response( "usage": {"cost": 0.4}})])
{
"id": "job-1",
"status": "completed",
"unsigned_urls": ["https://cdn.example/video.mp4"],
"usage": {"cost": 0.4},
}
),
]
def post(self, url, **kwargs): def post(self, url, **kwargs):
self.posts.append((url, kwargs)) self.posts.append((url, kwargs))
return _Response( return _Response({"id": "job-1", "polling_url": f"{url}/job-1", "status": "pending"}, status_code=202)
{
"id": "job-1",
"polling_url": "https://openrouter.ai/api/v1/videos/job-1",
"status": "pending",
},
status_code=202,
)
def get(self, url, **kwargs): def get(self, url, **kwargs):
self.gets.append((url, kwargs)) self.gets.append((url, kwargs))
return self.polls.pop(0) return self.polls.pop(0)
def close(self):
pass
def test_poll_caps_request_timeout_and_rejects_late_terminal_response(monkeypatch):
provider = OpenRouterVideoGenProvider() def test_generate_submits_polls_and_downloads_from_configured_origin(monkeypatch, tmp_path):
provider._poll_deadline_s = 10 """The bearer key goes to the poll URL and to ``{base}/videos/{id}/content`` derived from OUR base URL,
provider._request_timeout_s = 60 never to a provider-supplied ``unsigned_urls`` host."""
monkeypatch.setenv("OPENROUTER_API_KEY", "sk-or-test")
monkeypatch.delenv("OPENROUTER_BASE_URL", raising=False)
provider = _provider(monkeypatch, [_VEO], configured="google/veo-3.1")
session = _Session() session = _Session()
session.polls = [_Response({"id": "job-1", "status": "completed"})]
ticks = iter([100.0, 109.0, 111.0])
monkeypatch.setattr("plugins.video_gen.openrouter.time.monotonic", lambda: next(ticks))
with pytest.raises(TimeoutError, match="did not finish within 10s"):
provider._poll(session, "job-1")
assert session.gets[0][1]["timeout"] == 1.0
def test_unified_video_schema_exposes_hailuo_resolution_and_aspect_ratio():
properties = VIDEO_GENERATE_SCHEMA["parameters"]["properties"]
assert "768p" in properties["resolution"]["enum"]
assert "21:9" in properties["aspect_ratio"]["enum"]
def test_hailuo_catalog_and_capabilities_are_pinned_to_openrouter_contract():
provider = OpenRouterVideoGenProvider()
assert DEFAULT_MODEL == "minimax/hailuo-3-max"
assert provider.default_model() == DEFAULT_MODEL
assert provider.list_models() == [
{
"id": DEFAULT_MODEL,
"display": "MiniMax H3 Max",
"speed": "~20-60s",
"strengths": "Fast text-to-video and first-frame image-to-video.",
"price": "$0.05/s (480p), $0.08/s (768p)",
"modalities": ["text", "image"],
}
]
assert provider.capabilities() == {
"modalities": ["text", "image"],
"aspect_ratios": ["21:9", "16:9", "4:3", "1:1", "3:4", "9:16"],
"resolutions": ["480p", "768p"],
"max_duration": 15,
"min_duration": 5,
"supports_audio": False,
"supports_negative_prompt": False,
"supports_seed": False,
"supports_upscale": False,
"max_reference_images": 0,
}
def test_build_payload_uses_openrouter_video_fields_and_clamps_values():
payload = _build_payload(
prompt="A lighthouse in a storm",
image_url="https://example.com/start.png",
duration=99,
aspect_ratio="2:3",
resolution="720p",
)
assert payload == {
"model": DEFAULT_MODEL,
"prompt": "A lighthouse in a storm",
"duration": 15,
"resolution": "768p",
"aspect_ratio": "16:9",
"frame_images": [
{
"type": "image_url",
"image_url": {"url": "https://example.com/start.png"},
"frame_type": "first_frame",
}
],
}
def test_generate_submits_polls_and_materializes_completed_video(monkeypatch, tmp_path):
provider = OpenRouterVideoGenProvider()
session = _Session()
saved = []
monkeypatch.setenv("OPENROUTER_API_KEY", "test-key")
monkeypatch.setattr(provider, "_session", lambda: session) monkeypatch.setattr(provider, "_session", lambda: session)
monkeypatch.setattr("plugins.video_gen.openrouter.time.sleep", lambda _seconds: None) monkeypatch.setattr("plugins.video_gen.openrouter.time.sleep", lambda s: None)
monkeypatch.setattr( saved = []
"plugins.video_gen.openrouter.save_url_video",
lambda url, prefix, headers, require_video_content_type: saved.append(
(url, prefix, headers, require_video_content_type)
)
or tmp_path / "hailuo.mp4",
)
result = provider.generate( def fake_save(url, **kwargs):
"A slow cinematic push-in", saved.append((url, kwargs))
duration=5, return tmp_path / "clip.mp4"
aspect_ratio="9:16", monkeypatch.setattr("plugins.video_gen.openrouter.save_url_video", fake_save)
resolution="480p",
)
assert result["success"] is True result = provider.generate("a fox", duration=6, resolution="1080p", aspect_ratio="16:9", audio=True)
assert result["video"] == str(tmp_path / "hailuo.mp4")
assert result["provider"] == "openrouter" assert result["success"], result
assert result["model"] == DEFAULT_MODEL assert result["video"] == str(tmp_path / "clip.mp4") and result["cost"] == 0.4 and result["duration"] == 6
assert result["modality"] == "text"
assert result["duration"] == 5
assert result["cost"] == 0.4
assert session.posts[0][0] == "https://openrouter.ai/api/v1/videos" assert session.posts[0][0] == "https://openrouter.ai/api/v1/videos"
assert session.posts[0][1]["json"]["resolution"] == "480p" assert session.posts[0][1]["json"]["model"] == "google/veo-3.1" and session.posts[0][1]["json"]["generate_audio"] is True
assert [url for url, _kwargs in session.gets] == [ assert session.posts[0][1]["headers"]["Authorization"] == "Bearer sk-or-test"
"https://openrouter.ai/api/v1/videos/job-1", assert [g[0] for g in session.gets] == ["https://openrouter.ai/api/v1/videos/job-1"] * 2
"https://openrouter.ai/api/v1/videos/job-1", assert saved[0][0] == "https://openrouter.ai/api/v1/videos/job-1/content"
] assert saved[0][1]["headers"]["Authorization"] == "Bearer sk-or-test" and saved[0][1]["require_video_content_type"]
assert saved == [
(
"https://openrouter.ai/api/v1/videos/job-1/content",
"openrouter-hailuo",
{
"Authorization": "Bearer test-key",
"Content-Type": "application/json",
"User-Agent": "hermes-agent/video_gen",
},
True,
)
]
def test_generate_rejects_non_http_image_input_without_calling_api(monkeypatch): def test_generate_rejects_local_image_paths_before_spending(monkeypatch):
provider = OpenRouterVideoGenProvider() monkeypatch.setenv("OPENROUTER_API_KEY", "sk-or-test")
monkeypatch.setenv("OPENROUTER_API_KEY", "test-key") provider = _provider(monkeypatch, [_VEO])
monkeypatch.setattr(provider, "_session", lambda: (_ for _ in ()).throw(AssertionError("must not submit")))
result = provider.generate("animate this", image_url="/private/start.png") result = provider.generate("p", image_url="/home/me/frame.png")
assert not result["success"] and result["error_type"] == "invalid_request"
assert result["success"] is False
assert result["error_type"] == "invalid_request"
assert "public HTTP" in result["error"]
def test_generate_rejects_private_first_frame_without_calling_api(monkeypatch): def test_register_exposes_openrouter_in_the_video_gen_picker(monkeypatch):
provider = OpenRouterVideoGenProvider() from hermes_cli import plugins as plugin_loader, tools_config
monkeypatch.setenv("OPENROUTER_API_KEY", "test-key")
monkeypatch.setattr(
provider,
"_session",
lambda: (_ for _ in ()).throw(AssertionError("API must not be called")),
)
result = provider.generate("animate this", image_url="https://127.0.0.1/start.png")
assert result["success"] is False
assert result["error_type"] == "invalid_request"
def test_reference_images_are_rejected_before_session_creation(monkeypatch):
provider = OpenRouterVideoGenProvider()
monkeypatch.setenv("OPENROUTER_API_KEY", "test-key")
monkeypatch.setattr(
provider,
"_session",
lambda: (_ for _ in ()).throw(AssertionError("API must not be called")),
)
result = provider.generate(
"use these references",
reference_image_urls=["https://example.com/reference.png"],
)
assert result["success"] is False
assert result["error_type"] == "unsupported_input"
def test_generate_rejects_unknown_model_without_calling_api(monkeypatch):
provider = OpenRouterVideoGenProvider()
monkeypatch.setenv("OPENROUTER_API_KEY", "test-key")
result = provider.generate("test", model="other/video-model")
assert result["success"] is False
assert result["error_type"] == "invalid_model"
assert DEFAULT_MODEL in result["error"]
def test_register_and_picker_discovery_expose_openrouter(monkeypatch):
from hermes_cli import tools_config
from hermes_cli import plugins as plugin_loader
from plugins.video_gen.openrouter import register from plugins.video_gen.openrouter import register
class _Context: class _Context:
@@ -252,35 +139,7 @@ def test_register_and_picker_discovery_expose_openrouter(monkeypatch):
try: try:
register(_Context()) register(_Context())
monkeypatch.setattr(plugin_loader, "_ensure_plugins_discovered", lambda: None) monkeypatch.setattr(plugin_loader, "_ensure_plugins_discovered", lambda: None)
row = next(r for r in tools_config._plugin_video_gen_providers() if r["video_gen_plugin_name"] == "openrouter")
registered = video_gen_registry.get_provider("openrouter") assert row["name"] == "OpenRouter" and row["env_vars"][0]["key"] == "OPENROUTER_API_KEY"
rows = tools_config._plugin_video_gen_providers()
assert isinstance(registered, OpenRouterVideoGenProvider)
row = next(item for item in rows if item["video_gen_plugin_name"] == "openrouter")
assert row["name"] == "OpenRouter"
assert row["env_vars"][0]["key"] == "OPENROUTER_API_KEY"
finally: finally:
video_gen_registry._reset_for_tests() video_gen_registry._reset_for_tests()
def test_dynamic_schema_is_capability_scoped(monkeypatch):
from tools import video_generation_tool
provider = OpenRouterVideoGenProvider()
monkeypatch.setattr(video_generation_tool, "_resolve_active_provider", lambda: provider)
monkeypatch.setattr(video_generation_tool, "_read_configured_video_model", lambda: DEFAULT_MODEL)
schema = video_generation_tool._build_dynamic_video_schema()
properties = schema["parameters"]["properties"]
assert properties["duration"]["minimum"] == 5
assert properties["duration"]["maximum"] == 15
assert properties["resolution"]["enum"] == ["480p", "768p"]
assert properties["aspect_ratio"]["enum"] == [
"21:9", "16:9", "4:3", "1:1", "3:4", "9:16",
]
assert "image_url" in properties
assert "reference_image_urls" not in properties
assert "audio" not in properties
assert "seed" not in properties

View File

@@ -168,13 +168,10 @@ class TestFleetCapabilityCoverage(unittest.TestCase):
f"({implements_upscale})", f"({implements_upscale})",
) )
declares_seed = '"supports_seed": True' in src declares_seed = '"supports_seed": True' in src
# Merely accepting ``seed`` in generate() is part of the ABC implements_seed = ("seed" in impl_src
# contract, not proof that the backend implements it. Providers and ("payload[\"seed\"]" in impl_src
# can mark that contract-only parameter explicitly. or "seed: Optional[int]" in impl_src
implements_seed = ("payload[\"seed\"]" in impl_src or "\"seed\": seed" in impl_src))
or "\"seed\": seed" in impl_src
or ("seed: Optional[int]" in impl_src
and "_IGNORES_SEED = True" not in impl_src))
self.assertEqual( self.assertEqual(
declares_seed, implements_seed, declares_seed, implements_seed,
f"{name}: supports_seed declaration ({declares_seed}) " f"{name}: supports_seed declaration ({declares_seed}) "

View File

@@ -134,7 +134,7 @@ def _missing_provider_error(configured: Optional[str]) -> str:
error_type="provider_not_registered", provider=configured)) error_type="provider_not_registered", provider=configured))
return json.dumps(error_response( return json.dumps(error_response(
error=("No video generation backend is configured. Run `hermes tools` → " error=("No video generation backend is configured. Run `hermes tools` → "
"Video Generation to enable one (xAI, FAL, or Google Veo)."), "Video Generation to enable one (xAI, FAL, OpenRouter, or DeepInfra)."),
error_type="no_provider_configured")) error_type="no_provider_configured"))

View File

@@ -6,7 +6,7 @@ description: "How to build a video-generation backend plugin for Hermes Agent"
# Building a Video Generation Provider Plugin # Building a Video Generation Provider Plugin
Video-gen provider plugins register a backend that services every `video_generate` tool call. Built-in providers (xAI, FAL, DeepInfra) ship as plugins. Add a new one, or override a bundled one, by dropping a directory into `plugins/video_gen/<name>/`. Video-gen provider plugins register a backend that services every `video_generate` tool call. Built-in providers (xAI, FAL, OpenRouter, DeepInfra) ship as plugins. Add a new one, or override a bundled one, by dropping a directory into `plugins/video_gen/<name>/`.
:::tip :::tip
Video-gen mirrors [Image Generation Provider Plugins](/developer-guide/image-gen-provider-plugin) almost line-for-line — if you've built an image-gen backend, you already know the shape. The main differences: a `capabilities()` method advertising modalities/aspect-ratios/durations, and a routing convention (pass `image_url` to use image-to-video, omit it to use text-to-video — the provider picks the right endpoint internally). Video-gen mirrors [Image Generation Provider Plugins](/developer-guide/image-gen-provider-plugin) almost line-for-line — if you've built an image-gen backend, you already know the shape. The main differences: a `capabilities()` method advertising modalities/aspect-ratios/durations, and a routing convention (pass `image_url` to use image-to-video, omit it to use text-to-video — the provider picks the right endpoint internally).

View File

@@ -309,6 +309,8 @@ Backends ship as plugins under `plugins/video_gen/<name>/`:
- **xAI Grok-Imagine** — text-to-video and image-to-video (SuperGrok OAuth or `XAI_API_KEY`). - **xAI Grok-Imagine** — text-to-video and image-to-video (SuperGrok OAuth or `XAI_API_KEY`).
- **FAL.ai** — Veo 3.1, Pixverse v6, Kling O3 (requires `FAL_KEY`). - **FAL.ai** — Veo 3.1, Pixverse v6, Kling O3 (requires `FAL_KEY`).
- **OpenRouter** — every generative model on OpenRouter's video API (Veo 3.1, Sora 2 Pro, Kling 3, Seedance 2, Wan 3, Hailuo 3, Grok Imagine, FLUX 3 Video, …); text-to-video, image-to-video and reference-to-video; catalog and per-model limits fetched live (requires `OPENROUTER_API_KEY`, billed to your OpenRouter credit).
- **DeepInfra** — live `video-gen` catalog over the OpenAI-compatible videos endpoint (requires `DEEPINFRA_API_KEY`).
The single `video_generate` tool covers both modalities — pass `image_url` to animate a still, omit it to generate from text alone. The active backend auto-routes to the right endpoint. The tool's description is rebuilt at session start to reflect the active backend's actual capabilities (modalities, aspect ratios, resolutions, duration range, max reference images, audio support). See [Video Generation Provider Plugins](/developer-guide/video-gen-provider-plugin) for backend authoring. The single `video_generate` tool covers both modalities — pass `image_url` to animate a still, omit it to generate from text alone. The active backend auto-routes to the right endpoint. The tool's description is rebuilt at session start to reflect the active backend's actual capabilities (modalities, aspect ratios, resolutions, duration range, max reference images, audio support). See [Video Generation Provider Plugins](/developer-guide/video-gen-provider-plugin) for backend authoring.