feat(providers): Support DeepInfra as an LLM provider

This commit is contained in:
Georgi Atsev 2026-04-06 12:22:42 +03:00 committed by kshitij
parent ed8ce1f96c
commit fe002eb124
34 changed files with 2499 additions and 84 deletions

View file

@ -115,6 +115,10 @@
# HF_BASE_URL=https://router.huggingface.co/v1 # Override default base URL
# OPENCODE_GO_BASE_URL=https://opencode.ai/zen/go/v1 # Override default base URL
# DeepInfra — 100+ top open models, pay-per-use.
# Get your key at: https://deepinfra.com/dash/api_keys
# DEEPINFRA_API_KEY=
# =============================================================================
# LLM PROVIDER (Qwen OAuth)
# =============================================================================

View file

@ -474,6 +474,10 @@ _API_KEY_PROVIDER_AUX_MODELS_FALLBACK: Dict[str, str] = {
"kilocode": "google/gemini-3-flash-preview",
"ollama-cloud": "nemotron-3-nano:30b",
"tencent-tokenhub": "hy3-preview",
# NB: no "deepinfra" entry — its aux model lives on the ProviderProfile
# (plugins/model-providers/deepinfra: default_aux_model), which
# _get_aux_model_for_provider() reads first. Duplicating it here would be
# dead data that drifts when the profile's value is bumped.
}
# Legacy alias — callers that haven't been updated to _get_aux_model_for_provider()
@ -489,6 +493,33 @@ _PROVIDER_VISION_MODELS: Dict[str, str] = {
"zai": "glm-5v-turbo",
}
def _resolve_provider_vision_default(provider: str) -> Optional[str]:
"""Return the provider's preferred default vision model id, or None.
Static entries in :data:`_PROVIDER_VISION_MODELS` win first (xiaomi /
zai have dedicated vision-only model names that don't live in any
discoverable catalog). Otherwise the provider's :class:`ProviderProfile`
gets a chance to supply one via its ``default_vision_model()`` hook
that's where catalog-backed providers (DeepInfra) resolve a live default,
keeping the discovery logic inside their plugin instead of a name-check
branch here.
"""
static = _PROVIDER_VISION_MODELS.get(provider)
if static:
return static
try:
from providers import get_provider_profile
profile = get_provider_profile(provider)
except Exception:
return None
if profile is None:
return None
try:
return profile.default_vision_model()
except Exception:
return None
# Providers whose endpoint does not accept image input, even though the
# provider's broader ecosystem has vision models available elsewhere. When
# `auxiliary.vision.provider: auto` sees one of these as the main provider,
@ -5178,6 +5209,7 @@ def get_async_text_auxiliary_client(task: str = "", *, main_runtime: Optional[Di
_VISION_AUTO_PROVIDER_ORDER = (
"openrouter",
"nous",
"deepinfra",
)
@ -5234,6 +5266,21 @@ def _resolve_strict_vision_backend(
return resolve_provider_client("openai-codex", model, is_vision=True)
if provider == "anthropic":
return _try_anthropic()
if provider == "deepinfra":
# DeepInfra exposes vision-capable models (Llama-4 Scout/Maverick,
# Qwen3-VL, Gemma 3, Gemini) on the same OpenAI-compatible endpoint
# as its chat models. The default is discovered live via the profile's
# default_vision_model() hook (key-gated, chat-surface + vision tag) so
# we don't pin a hardcoded id that may rot when DeepInfra retires a
# model, and this module stays provider-agnostic.
vision_model = model or _resolve_provider_vision_default("deepinfra")
if not vision_model:
logger.debug(
"Vision auto-detect: deepinfra catalog unreachable or "
"returned no vision-tagged models — skipping"
)
return None, None
return resolve_provider_client("deepinfra", vision_model, is_vision=True)
if provider == "custom":
return _try_custom_endpoint()
return None, None
@ -5319,16 +5366,29 @@ def resolve_vision_provider_client(
# _PROVIDER_VISION_MODELS provides per-provider vision model
# overrides when the provider has a dedicated multimodal model
# that differs from the chat model (e.g. xiaomi → mimo-v2-omni,
# zai → glm-5v-turbo). Nous is the exception: it has a dedicated
# strict vision backend with tier-aware defaults, so it must not
# fall through to the user's text chat model here.
# 2. OpenRouter (vision-capable aggregator fallback)
# zai → glm-5v-turbo). DeepInfra is similar but resolves its
# default vision model live from the catalog (see
# :func:`_resolve_provider_vision_default`). Nous is the
# exception: it has a dedicated strict vision backend with
# tier-aware defaults, so it must not fall through to the
# user's text chat model here.
# 2. OpenRouter (vision-capable aggregator fallback)
# 3. Nous Portal (vision-capable aggregator fallback)
# 4. Stop
# 4. DeepInfra (OpenAI-compatible; vision model discovered
# live from the catalog — tried when
# DEEPINFRA_API_KEY is set)
# 5. Stop
main_provider = _read_main_provider()
main_model = _read_main_model()
if main_provider and main_provider not in {"auto", ""}:
vision_model = _PROVIDER_VISION_MODELS.get(main_provider, main_model)
# A provider-specific vision default wins over the user's chat model:
# static overrides (xiaomi/zai) and catalog-backed discovery (the
# DeepInfra profile hook) both yield a *known* vision-capable model,
# whereas the pinned chat model is usually NOT multimodal (e.g. the
# DeepSeek-V4-Flash default) and _main_model_supports_vision can't be
# trusted to catch that. Only fall back to the chat model when no
# provider default is available (catalog unreachable).
vision_model = _resolve_provider_vision_default(main_provider) or main_model
if main_provider == "nous":
sync_client, default_model = _resolve_strict_vision_backend(
main_provider, vision_model

View file

@ -47,7 +47,7 @@ def _resolve_requests_verify() -> bool | str:
# are preserved so the full model name reaches cache lookups and server queries.
_PROVIDER_PREFIXES: frozenset[str] = frozenset({
"openrouter", "nous", "openai-codex", "copilot", "copilot-acp",
"gemini", "ollama-cloud", "zai", "kimi-coding", "kimi-coding-cn", "stepfun", "minimax", "minimax-oauth", "minimax-cn", "anthropic", "deepseek",
"gemini", "ollama-cloud", "zai", "kimi-coding", "kimi-coding-cn", "stepfun", "minimax", "minimax-oauth", "minimax-cn", "anthropic", "deepseek", "deepinfra",
"opencode-zen", "opencode-go", "kilocode", "alibaba", "novita",
"qwen-oauth",
"xiaomi",
@ -58,7 +58,7 @@ _PROVIDER_PREFIXES: frozenset[str] = frozenset({
# Common aliases
"google", "google-gemini", "google-ai-studio",
"glm", "z-ai", "z.ai", "zhipu", "github", "github-copilot",
"github-models", "kimi", "moonshot", "kimi-cn", "moonshot-cn", "claude", "deep-seek",
"github-models", "kimi", "moonshot", "kimi-cn", "moonshot-cn", "claude", "deep-seek", "deep-infra",
"ollama",
"stepfun", "opencode", "zen", "go", "kilo", "dashscope", "aliyun", "qwen",
"mimo", "xiaomi-mimo",
@ -461,6 +461,7 @@ _URL_TO_PROVIDER: Dict[str, str] = {
"generativelanguage.googleapis.com": "gemini",
"inference-api.nousresearch.com": "nous",
"api.deepseek.com": "deepseek",
"api.deepinfra.com": "deepinfra",
"api.githubcopilot.com": "copilot",
# Enterprise Copilot endpoints look like api.enterprise.githubcopilot.com,
# api.business.githubcopilot.com, etc. Match the suffix so context-window
@ -807,6 +808,24 @@ def _extract_pricing(payload: Dict[str, Any]) -> Dict[str, Any]:
pricing["completion"] = str(float(novita_output) / 10_000 / 1_000_000)
return pricing
# DeepInfra ships pricing under ``metadata.pricing`` with $/MTok values:
# ``input_tokens``, ``output_tokens``, ``cache_read_tokens``. Convert to
# per-token strings so the generic cost machinery (usage_pricing.py)
# consumes them through the same path as OpenRouter / OpenAI.
metadata = payload.get("metadata") if isinstance(payload.get("metadata"), dict) else None
deepinfra_pricing = metadata.get("pricing") if metadata else None
if isinstance(deepinfra_pricing, dict) and any(
k in deepinfra_pricing for k in ("input_tokens", "output_tokens", "cache_read_tokens")
):
result: Dict[str, Any] = {}
if deepinfra_pricing.get("input_tokens") is not None:
result["prompt"] = str(float(deepinfra_pricing["input_tokens"]) / 1_000_000)
if deepinfra_pricing.get("output_tokens") is not None:
result["completion"] = str(float(deepinfra_pricing["output_tokens"]) / 1_000_000)
if deepinfra_pricing.get("cache_read_tokens") is not None:
result["cache_read"] = str(float(deepinfra_pricing["cache_read_tokens"]) / 1_000_000)
return result
alias_map = {
"prompt": ("prompt", "input", "input_cost_per_token", "prompt_token_cost"),
"completion": ("completion", "output", "output_cost_per_token", "completion_token_cost"),

View file

@ -56,6 +56,7 @@ _BUILTIN_NAMES = frozenset({
"neutts",
"kittentts",
"piper",
"deepinfra",
})

View file

@ -244,6 +244,78 @@ def save_bytes_video(
return path
_URL_VIDEO_CONTENT_TYPES = {
"video/mp4": "mp4",
"video/webm": "webm",
"video/quicktime": "mov",
"video/x-matroska": "mkv",
}
def save_url_video(
url: str,
*,
prefix: str = "video",
timeout: float = 180.0,
max_bytes: int = 200 * 1024 * 1024,
) -> Path:
"""Download a video URL and write it under ``$HERMES_HOME/cache/videos/``.
The video twin of :func:`agent.image_gen_provider.save_url_image`: several
backends (DeepInfra, FAL) return an *ephemeral* delivery URL that expires
before a downstream consumer can fetch it, so we materialise the bytes
locally at tool-completion time. Streams with a size cap.
Raises on any network / HTTP / oversize error so callers can fall back to
returning the bare URL.
"""
import requests
response = requests.get(url, timeout=timeout, stream=True)
response.raise_for_status()
content_type = (response.headers.get("Content-Type") or "").split(";", 1)[0].strip().lower()
extension = _URL_VIDEO_CONTENT_TYPES.get(content_type)
if extension is None:
url_path = url.split("?", 1)[0].lower()
for ext in ("mp4", "webm", "mov", "mkv"):
if url_path.endswith(f".{ext}"):
extension = ext
break
if extension is None:
extension = "mp4"
ts = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
short = uuid.uuid4().hex[:8]
path = _videos_cache_dir() / f"{prefix}_{ts}_{short}.{extension}"
bytes_written = 0
with path.open("wb") as fh:
for chunk in response.iter_content(chunk_size=256 * 1024):
if not chunk:
continue
bytes_written += len(chunk)
if bytes_written > max_bytes:
fh.close()
try:
path.unlink()
except OSError:
pass
raise ValueError(
f"Video at {url} exceeds {max_bytes // (1024 * 1024)}MB cap; refusing to cache."
)
fh.write(chunk)
if bytes_written == 0:
try:
path.unlink()
except OSError:
pass
raise ValueError(f"Video at {url} was empty (0 bytes).")
return path
def success_response(
*,
video: str,
@ -297,3 +369,222 @@ def error_response(
"aspect_ratio": aspect_ratio,
"provider": provider,
}
# ---------------------------------------------------------------------------
# Reusable OpenAI-compatible backend
# ---------------------------------------------------------------------------
class OpenAICompatibleVideoGenProvider(VideoGenProvider):
"""Generic text/image-to-video over the OpenAI ``client.videos`` API.
DeepInfra, OpenAI/Sora, and OpenRouter all expose the same
``POST /videos`` async-job shape (``create`` poll ``download_content``),
so the SDK call lives here once. A concrete backend only needs to declare
its identity and credentials::
class FooVideoGenProvider(OpenAICompatibleVideoGenProvider):
name = "foo"
_env_key = "FOO_API_KEY"
_default_base_url = "https://api.foo.com/v1/openai"
def list_models(self):
return [...] # entries with an "id" key; default_model() uses [0]
``image_url`` routes to image-to-video; its absence routes to text-to-video.
Provider-specific fields (``image_url``/``negative_prompt``/``seed``) ride
in ``extra_body`` so they pass through the SDK unchanged.
"""
_env_key: str = "OPENAI_API_KEY"
_default_base_url: str = "https://api.openai.com/v1"
# Polling cadence for the async video job. The OpenAI SDK's
# ``create_and_poll`` defaults to ~1 poll/second and loops forever on a
# non-terminal status, so a multi-minute job issues hundreds of sequential
# requests and a stuck job pins its tool-executor worker thread with no way
# out. We hand-roll a bounded poll instead: a coarse interval plus a hard
# wall-clock deadline that surfaces a timeout error.
_poll_interval_s: float = 5.0
_poll_deadline_s: float = 900.0
def _api_key(self) -> str:
import os
return os.environ.get(self._env_key, "").strip()
def is_available(self) -> bool:
return bool(self._api_key())
def _create_and_poll(self, client: Any, call_kwargs: Dict[str, Any]) -> Any:
"""Create the video job and poll to completion with a hard deadline.
Replaces ``client.videos.create_and_poll`` (unbounded 1/s loop) with a
coarse interval and a wall-clock cap. Returns the terminal video object
(any status); raises :class:`TimeoutError` if the deadline passes
first.
"""
import time
video = client.videos.create(**call_kwargs)
terminal = {"completed", "succeeded", "failed", "error", "cancelled", "canceled"}
deadline = time.monotonic() + self._poll_deadline_s
while getattr(video, "status", None) not in terminal:
if time.monotonic() >= deadline:
raise TimeoutError(
f"video job {getattr(video, 'id', '?')} did not reach a terminal "
f"status within {int(self._poll_deadline_s)}s "
f"(last status={getattr(video, 'status', None)!r})"
)
time.sleep(self._poll_interval_s)
video = client.videos.retrieve(video.id)
return video
def _base_url(self) -> str:
import os
override = os.environ.get(f"{self.name.upper()}_BASE_URL", "").strip()
return override or self._default_base_url
def generate(
self,
prompt: str,
*,
model: Optional[str] = None,
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]:
if not prompt or not prompt.strip():
return error_response(
error="prompt is required", error_type="invalid_request", provider=self.name
)
if not self._api_key():
return error_response(
error=f"{self._env_key} is not set",
error_type="missing_credentials",
provider=self.name,
)
try:
import openai
except ImportError:
return error_response(
error="openai Python package not installed (pip install openai)",
error_type="missing_dependency",
provider=self.name,
)
model_id = model or self.default_model()
if not model_id:
return error_response(
error=f"no {self.name} video model available (live catalog empty?)",
error_type="no_model",
provider=self.name,
)
# Provider-specific fields the OpenAI ``videos.create`` signature does
# not name natively — pass them through ``extra_body``.
extra_body = {
k: v
for k, v in {
"negative_prompt": negative_prompt,
"aspect_ratio": aspect_ratio,
"image_url": image_url, # presence ⇒ image-to-video
"seed": seed,
}.items()
if v is not None
}
call_kwargs: Dict[str, Any] = {"model": model_id, "prompt": prompt}
if duration:
call_kwargs["seconds"] = str(duration)
if resolution:
call_kwargs["size"] = resolution
if extra_body:
call_kwargs["extra_body"] = extra_body
client = openai.OpenAI(api_key=self._api_key(), base_url=self._base_url())
try:
try:
video = self._create_and_poll(client, call_kwargs)
except Exception as exc: # noqa: BLE001 - surface any SDK/API/timeout failure uniformly
logger.debug("%s video generation failed", self.name, exc_info=True)
return error_response(
error=f"{self.name} video generation failed: {exc}",
error_type="api_error",
provider=self.name,
model=model_id,
prompt=prompt,
aspect_ratio=aspect_ratio,
)
# Terminal success status differs across backends: DeepInfra reports
# "succeeded", OpenAI/Sora reports "completed". Accept both.
status = getattr(video, "status", None)
if status not in ("completed", "succeeded"):
# ``video.error`` is a structured SDK object (pydantic
# VideoCreateError), not a string — str() it so the response
# dict stays JSON-serializable for the tool layer.
job_error = getattr(video, "error", None)
return error_response(
error=str(job_error) if job_error else f"video job ended with status={status!r}",
error_type="job_failed",
provider=self.name,
model=model_id,
prompt=prompt,
aspect_ratio=aspect_ratio,
)
# Resolve the output. Providers expose it either as a delivery URL in
# the job's ``data`` list (DeepInfra, FAL-style) or only via the SDK
# download endpoint (OpenAI/Sora). Download the bytes and save locally
# so the caller gets a durable file — DeepInfra's delivery URLs in
# particular are short-lived. Matches plugins/image_gen/deepinfra.
url = None
for item in getattr(video, "data", None) or []:
candidate = item.get("url") if isinstance(item, dict) else getattr(item, "url", None)
if candidate:
url = candidate
break
try:
if url:
# Materialise the (often short-lived) delivery URL locally.
video_ref = str(save_url_video(url, prefix=self.name))
else:
# OpenAI/Sora style: no public URL — pull bytes via the SDK.
raw = client.videos.download_content(video.id).read()
video_ref = str(save_bytes_video(raw, prefix=self.name))
except Exception as exc: # noqa: BLE001
if url:
# Best-effort: hand back the URL rather than fail outright.
logger.debug("%s: saving video locally failed (%s); returning URL", self.name, exc)
video_ref = url
else:
return error_response(
error=f"{self.name} video job succeeded but no output could be retrieved: {exc}",
error_type="empty_response",
provider=self.name,
model=model_id,
prompt=prompt,
aspect_ratio=aspect_ratio,
)
return success_response(
video=video_ref,
model=model_id,
prompt=prompt,
modality="image" if image_url else "text",
aspect_ratio=aspect_ratio,
duration=duration or 0,
provider=self.name,
)
finally:
close = getattr(client, "close", None)
if callable(close):
close()

View file

@ -11,12 +11,15 @@ Active selection
The active provider is chosen by ``video_gen.provider`` in ``config.yaml``.
If unset, :func:`get_active_provider` applies fallback logic:
1. If exactly one provider is registered, use it.
1. If exactly one *available* provider is registered, use it.
2. Otherwise return ``None`` (the tool surfaces a helpful error pointing
the user at ``hermes tools``).
Mirrors ``agent/image_gen_registry.py`` so the two surfaces behave the
same.
same: the unconfigured fallback is filtered by ``is_available()`` so a box
that has credentials for only one backend (e.g. DeepInfra, while the
``fal``/``xai`` plugins also register unconditionally) auto-selects it
instead of returning ``None``.
"""
from __future__ import annotations
@ -104,9 +107,21 @@ def get_active_provider() -> Optional[VideoGenProvider]:
configured,
)
# Fallback: single-provider case
if len(snapshot) == 1:
return next(iter(snapshot.values()))
def _is_available_safe(p: VideoGenProvider) -> bool:
"""Wrap ``is_available()`` so a buggy provider doesn't kill resolution."""
try:
return bool(p.is_available())
except Exception as exc: # noqa: BLE001
logger.debug("video_gen provider %s.is_available() raised %s", p.name, exc)
return False
# Fallback: single *available* provider — filter by is_available() so a
# box with credentials for only one backend auto-selects it even when
# other providers (fal/xai) register unconditionally without keys.
# Mirrors agent/image_gen_registry.get_active_provider().
available = [p for p in snapshot.values() if _is_available_safe(p)]
if len(available) == 1:
return available[0]
return None

View file

@ -28,6 +28,7 @@ model:
# "xiaomi" - Xiaomi MiMo (requires: XIAOMI_API_KEY)
# "arcee" - Arcee AI Trinity models (requires: ARCEEAI_API_KEY)
# "ollama-cloud" - Ollama Cloud (requires: OLLAMA_API_KEY — https://ollama.com/settings)
# "deepinfra" - DeepInfra (requires: DEEPINFRA_API_KEY)
# "kilocode" - KiloCode gateway (requires: KILOCODE_API_KEY)
# "azure-foundry" - Microsoft Foundry / Azure OpenAI (API key or Entra ID)
# "lmstudio" - LM Studio local server (optional: LM_API_KEY, defaults to http://127.0.0.1:1234/v1)
@ -1010,6 +1011,34 @@ stt:
model: "whisper-1" # whisper-1 | gpt-4o-mini-transcribe | gpt-4o-transcribe
# mistral:
# model: "voxtral-mini-latest" # voxtral-mini-latest | voxtral-mini-2602
# deepinfra:
# # Model id is discovered live from the DeepInfra catalog filtered
# # by the `stt` surface tag — leave `model` blank to take the first
# # live result. Pin only when you need a specific Whisper variant.
# model: ""
# Text-to-speech. Only the deepinfra block is documented here — the
# remaining providers (edge, openai, xai, minimax, mistral, gemini,
# elevenlabs, neutts, kittentts, piper) inherit sensible defaults from
# DEFAULT_CONFIG in hermes_cli/config.py.
# tts:
# provider: "deepinfra"
# deepinfra:
# # Model id is discovered live from the DeepInfra catalog filtered
# # by the `tts` surface tag — leave `model` blank to take the first
# # live result.
# model: ""
# voice: "default"
# Image generation. Each provider plugin reads its own ``image_gen.<name>``
# block; deepinfra discovers models live from
# api.deepinfra.com/v1/openai/models filtered by the ``image-gen`` tag —
# no model id is hardcoded, so retired models disappear automatically.
# image_gen:
# provider: "deepinfra"
# deepinfra:
# # Leave `model` blank for the first live `image-gen`-tagged result.
# model: ""
# =============================================================================
# Response Pacing (Messaging Platforms)

View file

@ -442,6 +442,14 @@ PROVIDER_REGISTRY: Dict[str, ProviderConfig] = {
api_key_env_vars=("AZURE_FOUNDRY_API_KEY",),
base_url_env_var="AZURE_FOUNDRY_BASE_URL",
),
"deepinfra": ProviderConfig(
id="deepinfra",
name="DeepInfra",
auth_type="api_key",
inference_base_url="https://api.deepinfra.com/v1/openai",
api_key_env_vars=("DEEPINFRA_API_KEY",),
base_url_env_var="DEEPINFRA_BASE_URL",
),
}
# Auto-extend PROVIDER_REGISTRY with any api-key provider registered in
@ -1692,6 +1700,7 @@ def resolve_provider(
"tencent": "tencent-tokenhub", "tokenhub": "tencent-tokenhub",
"tencent-cloud": "tencent-tokenhub", "tencentmaas": "tencent-tokenhub",
"aws": "bedrock", "aws-bedrock": "bedrock", "amazon-bedrock": "bedrock", "amazon": "bedrock",
"deep-infra": "deepinfra",
"go": "opencode-go", "opencode-go-sub": "opencode-go",
"kilo": "kilocode", "kilo-code": "kilocode", "kilo-gateway": "kilocode",
"lmstudio": "lmstudio", "lm-studio": "lmstudio", "lm_studio": "lmstudio",

View file

@ -2064,7 +2064,11 @@ DEFAULT_CONFIG = {
# limit (OpenAI 4096, xAI 15000, MiniMax 10000, ElevenLabs 5k-40k model-aware,
# Gemini 32000, Edge 5000, Mistral 4000, NeuTTS/KittenTTS 2000).
"tts": {
"provider": "edge", # "edge" (free) | "elevenlabs" (premium) | "openai" | "xai" | "minimax" | "mistral" | "gemini" | "neutts" (local) | "kittentts" (local) | "piper" (local)
# "" (auto) → use the active inference provider when it ships a built-in
# TTS backend (DeepInfra, OpenAI, xAI, …), else fall back to Edge. Set
# explicitly to pin a backend:
# "edge" (free) | "elevenlabs" (premium) | "openai" | "xai" | "minimax" | "mistral" | "gemini" | "deepinfra" | "neutts" (local) | "kittentts" (local) | "piper" (local)
"provider": "",
"edge": {
"voice": "en-US-AriaNeural",
# Popular: AriaNeural, JennyNeural, AndrewNeural, BrianNeural, SoniaNeural
@ -2120,15 +2124,20 @@ DEFAULT_CONFIG = {
# "volume": 1.0,
# "normalize_audio": True,
},
"deepinfra": {
"model": "", # empty = first tts-tagged model from the live catalog
"voice": "default",
# "base_url": "", # override DEEPINFRA_BASE_URL for TTS only
},
},
"stt": {
"enabled": True,
# When true, gateway voice messages are transcribed for the agent and
# the raw transcript is also echoed back to the user as a 🎙️ message.
# Set false to keep STT for the agent while suppressing that user-facing echo.
"echo_transcripts": True,
"provider": "local", # "local" (free, faster-whisper) | "groq" | "openai" (Whisper API) | "mistral" (Voxtral Transcribe) | "elevenlabs" (Scribe)
"provider": "local", # "local" (free, faster-whisper) | "groq" | "openai" (Whisper API) | "mistral" (Voxtral Transcribe) | "elevenlabs" (Scribe) | "deepinfra"
"local": {
"model": "base", # tiny, base, small, medium, large-v3
"language": "", # auto-detect by default; set to "en", "es", "fr", etc. to force
@ -2145,6 +2154,10 @@ DEFAULT_CONFIG = {
"tag_audio_events": False,
"diarize": False,
},
"deepinfra": {
"model": "", # empty = first stt-tagged model from the live catalog
# "base_url": "", # override DEEPINFRA_BASE_URL for STT only
},
},
"voice": {
@ -3721,6 +3734,21 @@ OPTIONAL_ENV_VARS = {
"category": "provider",
"advanced": True,
},
"DEEPINFRA_API_KEY": {
"description": "DeepInfra API key (100+ top models via api.deepinfra.com)",
"prompt": "DeepInfra API Key",
"url": "https://deepinfra.com/dash/api_keys",
"password": True,
"category": "provider",
},
"DEEPINFRA_BASE_URL": {
"description": "DeepInfra base URL override",
"prompt": "DeepInfra base URL (leave empty for default)",
"url": None,
"password": False,
"category": "provider",
"advanced": True,
},
# ── Tool API keys ──
"EXA_API_KEY": {

View file

@ -30,6 +30,7 @@ from utils import base_url_host_matches
_PROVIDER_ENV_HINTS = (
"DEEPINFRA_API_KEY",
"OPENROUTER_API_KEY",
"OPENAI_API_KEY",
"ANTHROPIC_API_KEY",
@ -850,6 +851,10 @@ def run_doctor(args):
# (accounts/fireworks/models/... and .../routers/...), so a "/"
# is expected, not an aggregator vendor prefix.
"fireworks",
# DeepInfra is an aggregator-style gateway: its catalog
# is exclusively ``vendor/model`` slugs (Qwen/Qwen3.5-…,
# meta-llama/Llama-3-…, anthropic/claude-opus-4-7, …).
"deepinfra",
}
provider_accepts_vendor_slug = (
provider_policy_id in providers_accepting_vendor_slugs

View file

@ -3185,6 +3185,7 @@ def select_provider_and_model(args=None):
"ollama-cloud",
"tencent-tokenhub",
"lmstudio",
"deepinfra",
} or _is_profile_api_key_provider(selected_provider):
_model_flow_api_key_provider(config, selected_provider, current_model)

View file

@ -9,6 +9,7 @@ from __future__ import annotations
import json
import os
import re
import urllib.parse
import urllib.request
import urllib.error
@ -516,6 +517,12 @@ _PROVIDER_MODELS: dict[str, list[str]] = {
"glm-4.7",
"MiniMax-M2.5",
],
# DeepInfra: empty by design. The live catalog at
# _fetch_deepinfra_models() (filtered by the ``chat`` surface tag) is
# the only source of truth. Hardcoding ids here would rot as models
# are deprecated upstream; the picker shows "no models" when the
# catalog is unreachable, which is honest.
"deepinfra": [],
# Curated HF model list — only agentic models that map to OpenRouter defaults.
"huggingface": [
"moonshotai/Kimi-K2.5",
@ -1087,6 +1094,7 @@ CANONICAL_PROVIDERS: list[ProviderEntry] = [
ProviderEntry("bedrock", "AWS Bedrock", "AWS Bedrock (Claude, Nova, Llama, DeepSeek; IAM or API key)"),
ProviderEntry("azure-foundry", "Azure Foundry", "Azure Foundry (OpenAI-style or Anthropic-style endpoint, your Azure AI deployment)"),
ProviderEntry("qwen-oauth", "Qwen OAuth (Portal)", "Qwen OAuth (Reuses local Qwen CLI login)"),
ProviderEntry("deepinfra", "DeepInfra", "DeepInfra (100+ top models)"),
]
# Auto-extend CANONICAL_PROVIDERS with any provider registered in providers/
@ -1297,6 +1305,7 @@ _PROVIDER_ALIASES = {
"lm_studio": "lmstudio",
"ollama": "custom", # bare "ollama" = local; use "ollama-cloud" for cloud
"ollama_cloud": "ollama-cloud",
"deep-infra": "deepinfra",
}
@ -1593,6 +1602,8 @@ def get_pricing_for_provider(provider: str, *, force_refresh: bool = False) -> d
)
if normalized == "novita":
return _fetch_novita_pricing(force_refresh=force_refresh)
if normalized == "deepinfra":
return _fetch_deepinfra_pricing(force_refresh=force_refresh)
if normalized == "nous":
api_key, base_url = _resolve_nous_pricing_credentials()
if base_url:
@ -2385,6 +2396,10 @@ def provider_model_ids(provider: Optional[str], *, force_refresh: bool = False)
merged_lower.add(m.lower())
return merged
return list(_PROVIDER_MODELS.get("anthropic", []))
if normalized == "deepinfra":
live = _fetch_deepinfra_models()
if live:
return live
if normalized == "ollama-cloud":
live = fetch_ollama_cloud_models(force_refresh=force_refresh)
if live:
@ -3626,6 +3641,227 @@ def probe_api_models(
}
# Legacy filter — used when an item has no surface tag (rolling out
# 2026-05). Once every model returned by the catalog endpoint carries an
# explicit surface tag (``chat``/``embed``/``image-gen``/``tts``/``stt``)
# the regex path becomes unreachable and can be removed.
_DEEPINFRA_EXCLUDE_RE = re.compile(
r"(?i)(embed|rerank|whisper|stable-diffusion|flux|sdxl|"
r"tts|bark|speech|image-gen|clip|vit-|dpt-)",
)
# Surface tags announce *what kind of model* this is. When none of these
# are present on a catalog entry, the tags array only carries capability
# tags (``reasoning``, ``vision``, ``prompt_cache``, …) and we have to
# fall back to id-regex inference for the chat surface.
_DEEPINFRA_SURFACE_TAGS: frozenset[str] = frozenset({
"chat", "embed", "image-gen", "tts", "stt", "video-gen",
})
_DEEPINFRA_DEFAULT_BASE_URL = "https://api.deepinfra.com/v1/openai"
_DEEPINFRA_MODELS_QUERY = "filter=true&sort_by=hermes"
# Module-level cache for the full tagged catalog response, keyed by base URL.
# Each value is the parsed ``data`` list. Surface-specific filters read from
# this cache so a single network round-trip serves chat / image-gen / tts /
# stt callers across the whole process lifetime.
_deepinfra_catalog_cache: dict[str, list[dict]] = {}
# Negative cache: monotonic timestamp of the last failed fetch, keyed by base
# URL. Without this, an unreachable catalog (offline / DNS / firewall) makes
# every surface helper (chat picker, pricing, image/video/tts/stt defaults,
# vision) re-attempt a fresh blocking fetch that eats the full timeout each
# time — several sequential stalls in one user-visible operation. A short TTL
# lets connectivity recover without a process restart.
_deepinfra_catalog_neg_cache: dict[str, float] = {}
_DEEPINFRA_CATALOG_NEG_TTL = 60.0 # seconds
def _deepinfra_catalog_url() -> tuple[str, str]:
"""Return ``(cache_key, full_url)`` for the DeepInfra catalog endpoint."""
base = os.getenv("DEEPINFRA_BASE_URL", "").strip() or _DEEPINFRA_DEFAULT_BASE_URL
cache_key = base.rstrip("/")
return cache_key, f"{cache_key}/models?{_DEEPINFRA_MODELS_QUERY}"
def _fetch_deepinfra_catalog(
*,
timeout: float = 5.0,
force_refresh: bool = False,
) -> Optional[list[dict]]:
"""Fetch the raw DeepInfra catalog list with module-level caching.
The endpoint serves chat + embed + image-gen + tts + stt models in one
response. Authentication is optional but Bearer-attached when available
so user-scoped catalogs (private fine-tunes etc.) are visible.
"""
cache_key, url = _deepinfra_catalog_url()
if not force_refresh:
if cache_key in _deepinfra_catalog_cache:
return _deepinfra_catalog_cache[cache_key]
last_fail = _deepinfra_catalog_neg_cache.get(cache_key)
if last_fail is not None and (time.monotonic() - last_fail) < _DEEPINFRA_CATALOG_NEG_TTL:
return None
headers: dict[str, str] = {"User-Agent": _HERMES_USER_AGENT}
api_key = os.getenv("DEEPINFRA_API_KEY", "").strip()
if api_key:
headers["Authorization"] = f"Bearer {api_key}"
req = urllib.request.Request(url, headers=headers)
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
payload = json.loads(resp.read().decode())
except Exception:
_deepinfra_catalog_neg_cache[cache_key] = time.monotonic()
return None
data = payload.get("data")
if not isinstance(data, list):
_deepinfra_catalog_neg_cache[cache_key] = time.monotonic()
return None
_deepinfra_catalog_cache[cache_key] = data
_deepinfra_catalog_neg_cache.pop(cache_key, None)
return data
def _fetch_deepinfra_models_by_tag(
tag: str,
*,
timeout: float = 5.0,
force_refresh: bool = False,
) -> Optional[list[dict]]:
"""Return DeepInfra models whose ``metadata.tags`` includes *tag*.
Each returned item is ``{"id": str, "metadata": dict}`` so callers can
inspect context length, pricing, default dimensions (image-gen),
pricing units (tts ``input_characters``, stt ``input_seconds``), etc.
For the chat surface, items without any ``tags`` field fall through
to the legacy name-regex exclusion so this keeps working while the
tag rollout (mid-2026) is still in flight.
Returns ``None`` on network failure.
"""
data = _fetch_deepinfra_catalog(timeout=timeout, force_refresh=force_refresh)
if data is None:
return None
matched: list[dict] = []
for item in data:
mid = item.get("id")
if not mid:
continue
# ``metadata is None`` means DeepInfra returns a stub without
# pricing/context — typically a model that's listed but not
# served. Skip those for every surface.
raw_metadata = item.get("metadata")
if raw_metadata is None:
continue
metadata = raw_metadata if isinstance(raw_metadata, dict) else {}
raw_tags = metadata.get("tags")
tags = raw_tags if isinstance(raw_tags, list) else []
has_surface_tag = any(t in _DEEPINFRA_SURFACE_TAGS for t in tags)
if has_surface_tag:
if tag in tags:
matched.append({"id": mid, "metadata": metadata})
continue
# Surface-tag rollout incomplete — fall back to id-regex inference.
# Only meaningful for the chat surface; embed/image-gen/tts/stt
# cannot be safely inferred from an id alone.
if tag == "chat" and not _DEEPINFRA_EXCLUDE_RE.search(mid):
matched.append({"id": mid, "metadata": metadata})
return matched
def _fetch_deepinfra_models(
timeout: float = 5.0,
*,
force_refresh: bool = False,
) -> Optional[list[str]]:
"""Return DeepInfra chat-model ids (tag-aware, regex fallback).
Thin wrapper over :func:`_fetch_deepinfra_models_by_tag` so historical
callers in :func:`provider_model_ids` keep their string-list contract.
Returns ``None`` on network failure, an empty list if the catalog
contains no chat-tagged ids (which would itself be surprising).
"""
items = _fetch_deepinfra_models_by_tag(
"chat", timeout=timeout, force_refresh=force_refresh
)
if items is None:
return None
return [item["id"] for item in items] or None
def deepinfra_model_ids(tag: str, *, force_refresh: bool = False) -> list[str]:
"""Return DeepInfra model ids carrying surface *tag* (``[]`` on failure).
Single source of truth for the per-surface model shims (TTS/STT/vision),
replacing the copy-pasted ``import _fetch_deepinfra_models_by_tag fetch
[item["id"] ]`` wrapper each of them used to carry.
"""
items = _fetch_deepinfra_models_by_tag(tag, force_refresh=force_refresh)
return [item["id"] for item in items] if items else []
def deepinfra_base_url(section: Optional[dict] = None) -> str:
"""Resolve the DeepInfra OpenAI-compatible base URL, normalized.
Precedence: config-section ``base_url`` ``DEEPINFRA_BASE_URL`` env
default. Always stripped with any trailing slash removed. Single source
of truth for the base-URL chain the TTS/STT/image/video shims each used
to re-code (with subtly divergent normalization).
"""
candidate = section.get("base_url") if isinstance(section, dict) else None
value = candidate or os.getenv("DEEPINFRA_BASE_URL") or _DEEPINFRA_DEFAULT_BASE_URL
return str(value).strip().rstrip("/")
def _fetch_deepinfra_pricing(
timeout: float = 5.0,
*,
force_refresh: bool = False,
) -> dict[str, dict[str, str]]:
"""Return picker-shape pricing for DeepInfra chat models.
DeepInfra publishes ``input_tokens`` / ``output_tokens`` /
``cache_read_tokens`` in $/MTok; the picker expects per-token strings
under ``prompt`` / ``completion`` / ``input_cache_read`` (mirrors the
OpenRouter shape consumed by
:func:`format_model_pricing_table`). Cached via the catalog helper so
repeated picker renders are free.
"""
items = _fetch_deepinfra_models_by_tag(
"chat", timeout=timeout, force_refresh=force_refresh
)
if not items:
return {}
result: dict[str, dict[str, str]] = {}
for item in items:
metadata = item.get("metadata") or {}
pricing = metadata.get("pricing") if isinstance(metadata, dict) else None
if not isinstance(pricing, dict):
continue
entry: dict[str, str] = {}
inp = pricing.get("input_tokens")
out = pricing.get("output_tokens")
cache_read = pricing.get("cache_read_tokens")
if inp is not None:
entry["prompt"] = str(float(inp) / 1_000_000)
if out is not None:
entry["completion"] = str(float(out) / 1_000_000)
if cache_read is not None:
entry["input_cache_read"] = str(float(cache_read) / 1_000_000)
if entry:
result[item["id"]] = entry
return result
def fetch_api_models(
api_key: Optional[str],
base_url: Optional[str],

View file

@ -150,6 +150,7 @@ def show_status(args):
"StepFun Step Plan": "STEPFUN_API_KEY",
"MiniMax": "MINIMAX_API_KEY",
"MiniMax-CN": "MINIMAX_CN_API_KEY",
"DeepInfra": "DEEPINFRA_API_KEY",
"Firecrawl": "FIRECRAWL_API_KEY",
"Tavily": "TAVILY_API_KEY",
"Browser Use": "BROWSER_USE_API_KEY", # Optional — local browser works without this
@ -374,6 +375,7 @@ def show_status(args):
"StepFun Step Plan": ("STEPFUN_API_KEY",),
"MiniMax": ("MINIMAX_API_KEY",),
"MiniMax (China)": ("MINIMAX_CN_API_KEY",),
"DeepInfra": ("DEEPINFRA_API_KEY",),
}
for pname, env_vars in apikey_providers.items():
key_val = ""

View file

@ -322,6 +322,15 @@ TOOL_CATEGORIES = {
"tts_provider": "piper",
"post_setup": "piper",
},
{
"name": "DeepInfra TTS",
"badge": "paid",
"tag": "Chatterbox, Qwen3-TTS, … — live catalog from api.deepinfra.com",
"env_vars": [
{"key": "DEEPINFRA_API_KEY", "prompt": "DeepInfra API key", "url": "https://deepinfra.com/dash/api_keys"},
],
"tts_provider": "deepinfra",
},
],
},
"web": {

View file

@ -0,0 +1,319 @@
"""DeepInfra image generation backend.
Exposes DeepInfra's image-gen catalog (FLUX, Qwen-Image-Edit, …) through
the OpenAI-compatible ``/v1/openai/images/generations`` endpoint as an
:class:`ImageGenProvider` implementation.
**Fully dynamic model discovery.** Unlike the other image-gen plugins in
this tree (which ship a hardcoded ``_MODELS`` dict), DeepInfra publishes
a single tagged catalog at
``https://api.deepinfra.com/v1/openai/models?filter=true&sort_by=hermes``
where each entry's ``metadata.tags`` declares its surface (``image-gen``
here). ``list_models()`` filters that catalog via
:func:`hermes_cli.models._fetch_deepinfra_models_by_tag` so newly added
models show up in ``hermes tools`` automatically. No model ids are
hardcoded in this file if a model is retired upstream, it disappears
from hermes the next time the catalog is fetched, no patch required.
Model selection (first hit wins):
1. ``DEEPINFRA_IMAGE_MODEL`` env var
2. ``image_gen.deepinfra.model`` in ``config.yaml``
3. First model from the live catalog
When all three are absent (catalog unreachable, nothing configured),
``generate()`` returns an :func:`error_response` rather than guessing.
"""
from __future__ import annotations
import logging
import os
from typing import Any, Dict, List, Optional
from agent.image_gen_provider import (
DEFAULT_ASPECT_RATIO,
ImageGenProvider,
error_response,
resolve_aspect_ratio,
save_b64_image,
save_url_image,
success_response,
)
logger = logging.getLogger(__name__)
# DeepInfra accepts standard OpenAI ``size`` strings. Mirrors the
# OpenAI plugin's mapping so aspect_ratio semantics stay consistent
# across the agent's image_generate tool surface.
_SIZES = {
"landscape": "1536x1024",
"square": "1024x1024",
"portrait": "1024x1536",
}
def _load_deepinfra_image_config() -> Dict[str, Any]:
"""Read ``image_gen.deepinfra`` from config.yaml."""
try:
from hermes_cli.config import load_config
cfg = load_config()
section = cfg.get("image_gen") if isinstance(cfg, dict) else None
di_section = section.get("deepinfra") if isinstance(section, dict) else None
return di_section if isinstance(di_section, dict) else {}
except Exception as exc:
logger.debug("Could not load image_gen.deepinfra config: %s", exc)
return {}
def _live_models() -> Optional[List[Dict[str, Any]]]:
"""Fetch ``image-gen``-tagged models from the DeepInfra catalog."""
try:
from hermes_cli.models import _fetch_deepinfra_models_by_tag
except Exception as exc:
logger.debug("Cannot import _fetch_deepinfra_models_by_tag: %s", exc)
return None
return _fetch_deepinfra_models_by_tag("image-gen")
def _format_catalog_row(item: Dict[str, Any]) -> Dict[str, Any]:
"""Format a catalog item into the picker row shape."""
mid = item.get("id", "")
metadata = item.get("metadata") or {}
pricing = metadata.get("pricing") if isinstance(metadata, dict) else None
price = ""
if isinstance(pricing, dict) and pricing.get("per_image_unit") is not None:
try:
price = f"${float(pricing['per_image_unit']):.4f}/image"
except (TypeError, ValueError):
price = ""
row: Dict[str, Any] = {
"id": mid,
"display": mid.split("/", 1)[-1] if "/" in mid else mid,
"strengths": metadata.get("description", "") if isinstance(metadata, dict) else "",
}
if price:
row["price"] = price
if isinstance(metadata, dict):
for key in ("default_width", "default_height", "default_iterations"):
if metadata.get(key) is not None:
row[key] = metadata[key]
return row
def _resolve_model(catalog: List[Dict[str, Any]], cfg: Dict[str, Any]) -> Optional[str]:
"""Pick the model id (env > config > first live result, else None).
Takes the already-loaded ``image_gen.deepinfra`` config so ``generate()``
reads config once instead of via a second ``load_config`` deepcopy.
"""
env_override = os.environ.get("DEEPINFRA_IMAGE_MODEL", "").strip()
if env_override:
return env_override
cfg_model = cfg.get("model") if isinstance(cfg, dict) else None
if isinstance(cfg_model, str) and cfg_model.strip():
return cfg_model.strip()
if catalog:
first = catalog[0].get("id")
if isinstance(first, str) and first:
return first
return None
class DeepInfraImageGenProvider(ImageGenProvider):
"""DeepInfra ``images.generations`` backend.
Catalog is discovered live from the DeepInfra ``/models`` endpoint
filtered by the ``image-gen`` surface tag.
"""
@property
def name(self) -> str:
return "deepinfra"
@property
def display_name(self) -> str:
return "DeepInfra"
def is_available(self) -> bool:
return bool(os.environ.get("DEEPINFRA_API_KEY", "").strip())
def list_models(self) -> List[Dict[str, Any]]:
live = _live_models()
if not live:
return []
return [_format_catalog_row(item) for item in live]
def default_model(self) -> Optional[str]:
rows = self.list_models()
if rows:
return rows[0].get("id")
return None
def get_setup_schema(self) -> Dict[str, Any]:
return {
"name": "DeepInfra",
"badge": "paid",
"tag": "FLUX, Qwen-Image, … — live catalog from api.deepinfra.com",
"env_vars": [
{
"key": "DEEPINFRA_API_KEY",
"prompt": "DeepInfra API key",
"url": "https://deepinfra.com/dash/api_keys",
},
],
}
def generate(
self,
prompt: str,
aspect_ratio: str = DEFAULT_ASPECT_RATIO,
**kwargs: Any,
) -> Dict[str, Any]:
prompt = (prompt or "").strip()
aspect = resolve_aspect_ratio(aspect_ratio)
if not prompt:
return error_response(
error="Prompt is required and must be a non-empty string",
error_type="invalid_argument",
provider="deepinfra",
aspect_ratio=aspect,
)
api_key = os.environ.get("DEEPINFRA_API_KEY", "").strip()
if not api_key:
return error_response(
error=(
"DEEPINFRA_API_KEY not set. Run `hermes tools` → Image "
"Generation → DeepInfra to configure, or `hermes setup` "
"to add the key."
),
error_type="auth_required",
provider="deepinfra",
aspect_ratio=aspect,
)
di_cfg = _load_deepinfra_image_config()
catalog = _live_models() or []
model_id = _resolve_model(catalog, di_cfg)
if not model_id:
return error_response(
error=(
"No DeepInfra image-gen model available. Pin one in "
"config.yaml under image_gen.deepinfra.model, set "
"DEEPINFRA_IMAGE_MODEL, or check connectivity to "
"api.deepinfra.com so the live catalog can be fetched."
),
error_type="no_model_available",
provider="deepinfra",
prompt=prompt,
aspect_ratio=aspect,
)
size = _SIZES.get(aspect, _SIZES["square"])
from hermes_cli.models import deepinfra_base_url
base_url = deepinfra_base_url(di_cfg)
# DeepInfra's /images/generations is OpenAI-compatible — use the
# openai SDK so we inherit its retry, timeout, and error mapping
# (mirrors the existing OpenAI image-gen plugin).
try:
import openai
except ImportError:
return error_response(
error="openai Python package not installed (pip install openai)",
error_type="missing_dependency",
provider="deepinfra",
aspect_ratio=aspect,
)
client = openai.OpenAI(api_key=api_key, base_url=base_url)
try:
response = client.images.generate(
model=model_id,
prompt=prompt,
size=size,
n=1,
)
except Exception as exc:
logger.debug("DeepInfra image generation failed", exc_info=True)
return error_response(
error=f"DeepInfra image generation failed: {exc}",
error_type="api_error",
provider="deepinfra",
model=model_id,
prompt=prompt,
aspect_ratio=aspect,
)
finally:
close = getattr(client, "close", None)
if callable(close):
close()
data = getattr(response, "data", None) or []
if not data:
return error_response(
error="DeepInfra returned no image data",
error_type="empty_response",
provider="deepinfra",
model=model_id,
prompt=prompt,
aspect_ratio=aspect,
)
first = data[0]
b64 = getattr(first, "b64_json", None)
url = getattr(first, "url", None)
# Drop the ``vendor/`` prefix and any colons so the saved filename
# stays a single path component on every OS.
short = model_id.split("/", 1)[-1].replace(":", "_")
if b64:
try:
saved_path = save_b64_image(b64, prefix=f"deepinfra_{short}")
except Exception as exc:
return error_response(
error=f"Could not save image to cache: {exc}",
error_type="io_error",
provider="deepinfra",
model=model_id,
prompt=prompt,
aspect_ratio=aspect,
)
image_ref = str(saved_path)
elif url:
# Materialise the (often short-lived) delivery URL locally so a
# downstream consumer (Telegram send_photo, browser fetch) doesn't
# get a dead link — mirrors the openai/xai/krea image plugins.
# Best-effort: fall back to the bare URL if the download fails.
try:
image_ref = str(save_url_image(url, prefix=f"deepinfra_{short}"))
except Exception as exc:
logger.debug("DeepInfra: caching delivery URL failed (%s); returning URL", exc)
image_ref = url
else:
return error_response(
error="DeepInfra response contained neither b64_json nor URL",
error_type="empty_response",
provider="deepinfra",
model=model_id,
prompt=prompt,
aspect_ratio=aspect,
)
return success_response(
image=image_ref,
model=model_id,
prompt=prompt,
aspect_ratio=aspect,
provider="deepinfra",
extra={"size": size},
)
def register(ctx) -> None:
"""Plugin entry point — wire ``DeepInfraImageGenProvider`` into the registry."""
ctx.register_image_gen_provider(DeepInfraImageGenProvider())

View file

@ -0,0 +1,7 @@
name: deepinfra
version: 1.0.0
description: "DeepInfra image generation backend (FLUX, Qwen-Image, …) via OpenAI-compatible /v1/images/generations. Catalog discovered live from api.deepinfra.com."
author: Georgi Atsev
kind: backend
requires_env:
- DEEPINFRA_API_KEY

View file

@ -0,0 +1,93 @@
"""DeepInfra provider profile.
DeepInfra is an OpenAI-compatible inference gateway that hosts 100+ open
models (Step, GLM, Kimi, DeepSeek, MiniMax, Nemotron, Mistral, Qwen, ) as
well as image-gen / TTS / STT / embedding endpoints. The chat surface is
wired in through this profile; non-chat surfaces are wired in through
their respective plugin subsystems (``plugins/image_gen/deepinfra`` and
the TTS/STT dispatchers in ``tools/``).
"""
from providers import register_provider
from providers.base import ProviderProfile
class _DeepInfraProfile(ProviderProfile):
"""DeepInfra profile with live vision-default discovery.
Owns its own vision default so shared vision resolution in
``agent/auxiliary_client.py`` stays provider-agnostic (a
``default_vision_model()`` hook call instead of an ``if provider ==
"deepinfra"`` branch reaching into the catalog helpers).
"""
def default_vision_model(self): # type: ignore[override]
"""First vision-capable *chat* model from the live catalog, or None.
Key-gated so a box without ``DEEPINFRA_API_KEY`` never pays the
catalog round-trip. Requires the ``chat`` surface tag (not just the
``vision`` capability) so an image-gen/edit model that merely carries
a ``vision`` tag can't be picked as a chat-completions vision backend.
"""
import os
if not (os.environ.get("DEEPINFRA_API_KEY") or "").strip():
return None
try:
from hermes_cli.models import _fetch_deepinfra_models_by_tag
items = _fetch_deepinfra_models_by_tag("chat")
except Exception:
return None
for item in items or []:
metadata = item.get("metadata") or {}
tags = metadata.get("tags") if isinstance(metadata, dict) else None
if isinstance(tags, list) and "vision" in tags:
model_id = item.get("id")
if model_id:
return model_id
return None
deepinfra = _DeepInfraProfile(
name="deepinfra",
aliases=("deep-infra", "deepinfra-ai"),
display_name="DeepInfra",
description="DeepInfra — 100+ open models, pay-per-use",
signup_url="https://deepinfra.com/dash/api_keys",
env_vars=("DEEPINFRA_API_KEY", "DEEPINFRA_BASE_URL"),
base_url="https://api.deepinfra.com/v1/openai",
auth_type="api_key",
# Default output cap when the user hasn't set ``agent.max_tokens``.
# Without this the profile inherits ``None`` and the transport sends no
# ``max_tokens`` (chat_completions.py: the ``elif profile_max`` branch
# is skipped because ``None`` is falsy), so DeepInfra applies its small
# server-side default (~8-16K). Tool-heavy runs (e.g. cron jobs doing
# many web_search calls before composing) then exhaust that budget on
# tool results, hit ``finish_reason='length'``, and fail after the
# 3 continuation retries in conversation_loop.py.
#
# 128K gives ample room for tool-result processing + final output.
# Safe across the whole catalog despite per-model limits ranging from
# 4K (Gryphe/MythoMax) to 1M (DeepSeek-V4): DeepInfra silently CLAMPS
# max_tokens to each model's own limit rather than rejecting it
# (verified live). Users who set ``agent.max_tokens`` still win — their
# value takes priority over this profile default in the transport.
default_max_tokens=131072,
# Auxiliary model — cheap/fast chat model the same provider uses for
# side tasks (context compression, session search, web extract,
# vision). This is the *only* hardcoded DeepInfra model in the
# integration: aux resolution is synchronous (no time for a catalog
# round-trip on every agent turn), so we need one explicit choice.
# Every other surface (chat picker, image-gen, tts, stt, pricing)
# discovers models live from
# ``api.deepinfra.com/v1/openai/models?filter=true&sort_by=hermes``.
default_aux_model="deepseek-ai/DeepSeek-V4-Flash",
# ``fallback_models`` deliberately empty — the live catalog at
# ``hermes_cli/models.py::_fetch_deepinfra_models`` is the source of
# truth. When the live fetch fails (network/DNS), the picker shows
# no options, which is preferable to silently routing the user to a
# model that may have been retired upstream.
fallback_models=(),
)
register_provider(deepinfra)

View file

@ -0,0 +1,5 @@
name: deepinfra-provider
kind: model-provider
version: 1.0.0
description: DeepInfra — 100+ open models, pay-per-use
author: Georgi Atsev

View file

@ -0,0 +1,90 @@
"""DeepInfra video generation backend.
DeepInfra serves video over the OpenAI-compatible ``/v1/openai/videos``
endpoint (async job: ``create`` poll ``download_content``), so all the
SDK plumbing lives in
:class:`agent.video_gen_provider.OpenAICompatibleVideoGenProvider`. This
plugin only declares DeepInfra's identity, credentials, and live model
discovery no hardcoded model ids, so retired models drop out of hermes the
next time the catalog is fetched without a patch.
Mirrors ``plugins/image_gen/deepinfra`` (which does the same for
``/v1/openai/images/generations``).
"""
from __future__ import annotations
import logging
from typing import Any, Dict, List
from agent.video_gen_provider import OpenAICompatibleVideoGenProvider
logger = logging.getLogger(__name__)
class DeepInfraVideoGenProvider(OpenAICompatibleVideoGenProvider):
"""Text-to-video and image-to-video via DeepInfra's OpenAI-compatible API."""
name = "deepinfra"
_env_key = "DEEPINFRA_API_KEY"
_default_base_url = "https://api.deepinfra.com/v1/openai"
@property
def display_name(self) -> str:
return "DeepInfra"
def list_models(self) -> List[Dict[str, Any]]:
"""Return ``video-gen``-tagged DeepInfra models from the live catalog.
Empty list when the catalog is unreachable the picker then shows no
options rather than routing to a possibly-retired model.
"""
try:
from hermes_cli.models import _fetch_deepinfra_models_by_tag
except Exception as exc: # noqa: BLE001 — never break the picker
logger.debug("Cannot import _fetch_deepinfra_models_by_tag: %s", exc)
return []
items = _fetch_deepinfra_models_by_tag("video-gen") or []
out: List[Dict[str, Any]] = []
for item in items:
mid = item.get("id")
if not mid:
continue
meta = item.get("metadata", {}) if isinstance(item, dict) else {}
out.append({
"id": mid,
"display": mid.split("/")[-1],
"strengths": (meta.get("description") or "")[:80],
})
return out
def capabilities(self) -> Dict[str, Any]:
return {
"modalities": ["text", "image"],
"aspect_ratios": ["16:9", "9:16", "1:1"],
"resolutions": ["480p", "720p", "1080p"],
"max_duration": 10,
"min_duration": 1,
"supports_audio": False,
"supports_negative_prompt": True,
"max_reference_images": 0,
}
def get_setup_schema(self) -> Dict[str, Any]:
return {
"name": "DeepInfra",
"badge": "paid",
"tag": "Wan, p-video, … — live catalog from api.deepinfra.com; text-to-video & image-to-video",
"env_vars": [
{
"key": "DEEPINFRA_API_KEY",
"prompt": "DeepInfra API key",
"url": "https://deepinfra.com/dash/api_keys",
},
],
}
def register(ctx) -> None:
"""Plugin entry point — wire ``DeepInfraVideoGenProvider`` into the registry."""
ctx.register_video_gen_provider(DeepInfraVideoGenProvider())

View file

@ -0,0 +1,7 @@
name: deepinfra
version: 1.0.0
description: "DeepInfra video generation (text-to-video & image-to-video) via the OpenAI-compatible /v1/openai/videos endpoint. Catalog discovered live from api.deepinfra.com."
author: Georgi Atsev
kind: backend
requires_env:
- DEEPINFRA_API_KEY

View file

@ -145,6 +145,19 @@ class ProviderProfile:
"""
return {}, {}
def default_vision_model(self) -> str | None:
"""Return a default vision model id for this provider, or None.
Overrideable hook for providers that discover their vision default at
runtime (e.g. from a live catalog) rather than pinning one in code.
Keeps provider-specific vision discovery inside the provider's plugin
instead of a name-check branch in shared vision resolution.
Default: None (no provider-specific vision model the caller falls
back to the user's chat model or the aggregator chain).
"""
return None
def get_max_tokens(self, model: str | None) -> int | None:
"""Return the default max_tokens cap for *model*.

View file

@ -80,14 +80,27 @@ class TestGetActiveProvider:
def test_multi_without_config_returns_none(self, tmp_path, monkeypatch):
"""Unlike image_gen (which falls back to 'fal'), video_gen has no
legacy default when there are multiple providers and no config,
the registry returns None and the tool surfaces a helpful error.
legacy default when there are multiple *available* providers and no
config, the registry returns None and the tool surfaces a helpful error.
"""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
video_gen_registry.register_provider(_FakeProvider("xai"))
video_gen_registry.register_provider(_FakeProvider("fal"))
assert video_gen_registry.get_active_provider() is None
def test_single_available_among_many_autoresolves(self, tmp_path, monkeypatch):
"""When several providers are registered but only one has credentials
(is_available()), that one is auto-selected without config. This is the
DeepInfra-only-box case: fal/xai register unconditionally but lack keys.
Mirrors agent/image_gen_registry's availability-filtered fallback.
"""
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
video_gen_registry.register_provider(_FakeProvider("fal", available=False))
video_gen_registry.register_provider(_FakeProvider("xai", available=False))
video_gen_registry.register_provider(_FakeProvider("deepinfra", available=True))
active = video_gen_registry.get_active_provider()
assert active is not None and active.name == "deepinfra"
def test_config_selects_provider(self, tmp_path, monkeypatch):
import yaml

View file

@ -1,5 +1,6 @@
"""Tests for API-key provider support (z.ai/GLM, Kimi, MiniMax)."""
import json
import os
import pytest
@ -110,6 +111,17 @@ class TestProviderRegistry:
assert pconfig.api_key_env_vars == ("HF_TOKEN",)
assert pconfig.base_url_env_var == "HF_BASE_URL"
def test_deepinfra_registration(self):
pconfig = PROVIDER_REGISTRY["deepinfra"]
assert pconfig.id == "deepinfra"
assert pconfig.name == "DeepInfra"
assert pconfig.auth_type == "api_key"
def test_deepinfra_env_vars(self):
pconfig = PROVIDER_REGISTRY["deepinfra"]
assert pconfig.api_key_env_vars == ("DEEPINFRA_API_KEY",)
assert pconfig.base_url_env_var == "DEEPINFRA_BASE_URL"
def test_base_urls(self):
assert PROVIDER_REGISTRY["copilot"].inference_base_url == "https://api.githubcopilot.com"
assert PROVIDER_REGISTRY["copilot-acp"].inference_base_url == "acp://copilot"
@ -121,6 +133,7 @@ class TestProviderRegistry:
assert PROVIDER_REGISTRY["kilocode"].inference_base_url == "https://api.kilo.ai/api/gateway"
assert PROVIDER_REGISTRY["gmi"].inference_base_url == "https://api.gmi-serving.com/v1"
assert PROVIDER_REGISTRY["huggingface"].inference_base_url == "https://router.huggingface.co/v1"
assert PROVIDER_REGISTRY["deepinfra"].inference_base_url == "https://api.deepinfra.com/v1/openai"
def test_oauth_providers_unchanged(self):
"""Ensure we didn't break the existing OAuth providers."""
@ -241,6 +254,12 @@ class TestResolveProvider:
def test_alias_huggingface_hub(self):
assert resolve_provider("huggingface-hub") == "huggingface"
def test_explicit_deepinfra(self):
assert resolve_provider("deepinfra") == "deepinfra"
def test_alias_deep_infra(self):
assert resolve_provider("deep-infra") == "deepinfra"
def test_unknown_provider_raises(self):
with pytest.raises(AuthError):
resolve_provider("nonexistent-provider-xyz")
@ -285,6 +304,10 @@ class TestResolveProvider:
monkeypatch.setenv("HF_TOKEN", "hf_test_token")
assert resolve_provider("auto") == "huggingface"
def test_auto_detects_deepinfra_key(self, monkeypatch):
monkeypatch.setenv("DEEPINFRA_API_KEY", "test-di-key")
assert resolve_provider("auto") == "deepinfra"
def test_openrouter_takes_priority_over_glm(self, monkeypatch):
"""OpenRouter API key should win over GLM in auto-detection."""
monkeypatch.setenv("OPENROUTER_API_KEY", "or-key")
@ -1305,3 +1328,271 @@ class TestMinimaxOAuthProvider:
"doesn't fire the 'No auxiliary LLM provider configured' warning "
"for every minimax-oauth session."
)
# =============================================================================
# DeepInfra provider tests
# =============================================================================
# Registration / alias / env-var invariants are asserted in
# TestProviderRegistry + TestResolveProvider above. The classes below
# cover the catalog/tag/pricing/profile machinery added on top of the
# baseline provider wiring.
@pytest.fixture
def _deepinfra_cache_isolation(monkeypatch):
"""Reset the module-level catalog cache around each DeepInfra test.
The cache is keyed by base URL and would otherwise leak fixture data
from one test into the next in the same session. The negative cache is
reset too, so a test that simulates an unreachable catalog can't suppress
a later test's fetch within the failure TTL.
"""
import hermes_cli.models as _models_mod
monkeypatch.setattr(_models_mod, "_deepinfra_catalog_cache", {})
monkeypatch.setattr(_models_mod, "_deepinfra_catalog_neg_cache", {})
yield
@pytest.mark.usefixtures("_deepinfra_cache_isolation")
class TestFetchDeepInfraModels:
"""Tests for _fetch_deepinfra_models() live model discovery."""
def test_returns_filtered_models_on_success(self, monkeypatch):
monkeypatch.setenv("DEEPINFRA_API_KEY", "test-key")
class _Resp:
def __enter__(self):
return self
def __exit__(self, *a):
return False
def read(self):
return json.dumps({"data": [
{"id": "meta-llama/Llama-3-70B-Instruct", "metadata": {}},
{"id": "mistralai/Mistral-Nemo-Instruct-2407", "metadata": {}},
{"id": "BAAI/bge-large-en-v1.5-embed", "metadata": {}},
{"id": "stabilityai/stable-diffusion-xl-base-1.0", "metadata": {}},
]}).encode()
import urllib.request
monkeypatch.setattr(urllib.request, "urlopen", lambda *a, **kw: _Resp())
from hermes_cli.models import _fetch_deepinfra_models
result = _fetch_deepinfra_models()
assert result is not None
assert "meta-llama/Llama-3-70B-Instruct" in result
assert "mistralai/Mistral-Nemo-Instruct-2407" in result
# Embedding and image models should be excluded
assert not any("embed" in m.lower() for m in result)
assert not any("stable-diffusion" in m.lower() for m in result)
def test_works_without_api_key(self, monkeypatch):
monkeypatch.delenv("DEEPINFRA_API_KEY", raising=False)
class _Resp:
def __enter__(self):
return self
def __exit__(self, *a):
return False
def read(self):
return json.dumps({"data": [
{"id": "meta-llama/Llama-3-70B-Instruct", "metadata": {}},
]}).encode()
import urllib.request
monkeypatch.setattr(urllib.request, "urlopen", lambda *a, **kw: _Resp())
from hermes_cli.models import _fetch_deepinfra_models
result = _fetch_deepinfra_models()
assert result == ["meta-llama/Llama-3-70B-Instruct"]
def test_returns_none_on_network_failure(self, monkeypatch):
monkeypatch.setenv("DEEPINFRA_API_KEY", "test-key")
import urllib.request
monkeypatch.setattr(urllib.request, "urlopen", lambda *a, **kw: (_ for _ in ()).throw(Exception("timeout")))
from hermes_cli.models import _fetch_deepinfra_models
assert _fetch_deepinfra_models() is None
def test_excludes_non_chat_models(self, monkeypatch):
monkeypatch.setenv("DEEPINFRA_API_KEY", "test-key")
class _Resp:
def __enter__(self):
return self
def __exit__(self, *a):
return False
def read(self):
return json.dumps({"data": [
{"id": "Qwen/Qwen3-235B-A22B-Instruct-2507", "metadata": {}},
{"id": "openai/whisper-large-v3", "metadata": {}},
{"id": "some-org/flux-dev", "metadata": {}},
{"id": "sentence-transformers/clip-ViT-B-32", "metadata": {}},
{"id": "microsoft/vit-base-patch16-224", "metadata": {}},
{"id": "some-org/rerank-v2", "metadata": {}},
{"id": "some-org/bark-large", "metadata": {}},
{"id": "nvidia/sdxl-turbo", "metadata": {}},
]}).encode()
import urllib.request
monkeypatch.setattr(urllib.request, "urlopen", lambda *a, **kw: _Resp())
from hermes_cli.models import _fetch_deepinfra_models
result = _fetch_deepinfra_models()
assert result == ["Qwen/Qwen3-235B-A22B-Instruct-2507"]
def _make_urlopen_returning(payload):
"""Helper: build a urlopen() shim returning a fixed JSON payload."""
import json as _json
class _Resp:
def __enter__(self):
return self
def __exit__(self, *a):
return False
def read(self):
return _json.dumps(payload).encode()
return lambda *a, **kw: _Resp()
@pytest.mark.usefixtures("_deepinfra_cache_isolation")
class TestDeepInfraTagFiltering:
"""Contract tests for the shared _fetch_deepinfra_models_by_tag helper."""
def test_filters_by_surface_tag_and_handles_rollout_states(self, monkeypatch):
# One payload, several invariants in one test:
# - explicit surface tags are honored (chat / image-gen / tts / stt / embed)
# - capability-tags-only items fall through to the regex fallback
# (used during the surface-tag rollout)
# - the regex excludes id-name matches (whisper, embed, …)
# - a surface tag takes priority over the regex
# - ``metadata: None`` stubs are dropped
payload = {"data": [
{"id": "vendor/chat-tagged", "metadata": {"tags": ["chat"]}},
{"id": "vendor/image-tagged", "metadata": {"tags": ["image-gen"]}},
{"id": "vendor/tts-tagged", "metadata": {"tags": ["tts"]}},
{"id": "vendor/stt-tagged", "metadata": {"tags": ["stt"]}},
{"id": "vendor/embed-tagged", "metadata": {"tags": ["embed"]}},
# capability-only — rolls through regex fallback
{"id": "Qwen/Qwen3-30B", "metadata": {"tags": ["reasoning", "vision"]}},
{"id": "openai/whisper-large", "metadata": {"tags": ["reasoning"]}},
# surface tag overrides legacy regex exclusion
{"id": "some-org/whisper-finetune-chat", "metadata": {"tags": ["chat"]}},
# null metadata — stub model, must be skipped
{"id": "stub-model", "metadata": None},
]}
import urllib.request
from hermes_cli.models import _fetch_deepinfra_models_by_tag
for surface in ("chat", "image-gen", "tts", "stt", "embed"):
monkeypatch.setattr(urllib.request, "urlopen", _make_urlopen_returning(payload))
# Reset cache between iterations so each surface re-parses the payload.
import hermes_cli.models as _m
_m._deepinfra_catalog_cache.clear()
got = _fetch_deepinfra_models_by_tag(surface)
assert got is not None
ids = {item["id"] for item in got}
assert "stub-model" not in ids # null-metadata always skipped
if surface == "chat":
# explicit chat + capability-only (Qwen) + surface-tag-over-regex
assert "vendor/chat-tagged" in ids
assert "Qwen/Qwen3-30B" in ids
assert "some-org/whisper-finetune-chat" in ids
# regex still excludes capability-only items that match the excluder
assert "openai/whisper-large" not in ids
else:
# non-chat surfaces only see explicit surface-tagged items
for item in got:
assert surface in item["metadata"]["tags"]
def test_returns_none_on_network_failure(self, monkeypatch):
import urllib.request
monkeypatch.setattr(
urllib.request, "urlopen",
lambda *a, **kw: (_ for _ in ()).throw(Exception("timeout")),
)
from hermes_cli.models import _fetch_deepinfra_models_by_tag, _fetch_deepinfra_pricing
assert _fetch_deepinfra_models_by_tag("chat") is None
# Pricing rides the same catalog cache — same failure mode.
assert _fetch_deepinfra_pricing() == {}
@pytest.mark.usefixtures("_deepinfra_cache_isolation")
class TestDeepInfraPricingFetcher:
"""_fetch_deepinfra_pricing reshapes $/MTok values into per-token strings
and is wired into the get_pricing_for_provider dispatch."""
def test_pricing_shape_and_dispatch(self, monkeypatch):
payload = {"data": [
{
"id": "vendor/model-a",
"metadata": {
"tags": ["chat", "prompt_cache"],
"pricing": {
"input_tokens": 0.1,
"output_tokens": 0.3,
"cache_read_tokens": 0.02,
},
},
},
{
"id": "vendor/model-b",
"metadata": {"tags": ["chat"], "pricing": {"input_tokens": 1.0, "output_tokens": 5.0}},
},
# non-chat — must not appear
{"id": "vendor/model-image", "metadata": {"tags": ["image-gen"], "pricing": {"per_image_unit": 0.05}}},
]}
import urllib.request
monkeypatch.setattr(urllib.request, "urlopen", _make_urlopen_returning(payload))
from hermes_cli.models import get_pricing_for_provider
# get_pricing_for_provider → _fetch_deepinfra_pricing dispatch path
result = get_pricing_for_provider("deepinfra")
assert set(result) == {"vendor/model-a", "vendor/model-b"}
# Picker-shape: per-token strings under prompt/completion (+ cache_read when source had it)
assert float(result["vendor/model-a"]["prompt"]) == pytest.approx(0.1 / 1_000_000)
assert float(result["vendor/model-a"]["completion"]) == pytest.approx(0.3 / 1_000_000)
assert "input_cache_read" in result["vendor/model-a"]
assert "input_cache_read" not in result["vendor/model-b"]
class TestDeepInfraProviderProfile:
"""plugins/model-providers/deepinfra registration + aux resolution."""
def test_profile_registered_with_alias_and_aux(self):
from providers import get_provider_profile
from agent.auxiliary_client import _get_aux_model_for_provider
profile = get_provider_profile("deepinfra")
assert profile is not None
assert profile.name == "deepinfra"
assert profile.auth_type == "api_key"
# Alias resolves to the same profile.
assert get_provider_profile("deep-infra") is profile
# Aux model is resolved via the profile (not via the legacy
# _API_KEY_PROVIDER_AUX_MODELS_FALLBACK dict, which has no
# deepinfra entry).
assert _get_aux_model_for_provider("deepinfra")
# Fallback list intentionally empty — live catalog is the source
# of truth. Pin the shape only, not contents.
assert isinstance(profile.fallback_models, tuple)
def test_profile_sets_default_max_tokens(self):
"""A non-None default_max_tokens must be advertised so the transport's
``elif profile_max`` branch fires when the user hasn't configured
``agent.max_tokens``. Without it DeepInfra applies a small server
default and tool-heavy runs truncate (finish_reason='length')."""
from providers import get_provider_profile
profile = get_provider_profile("deepinfra")
assert profile.default_max_tokens is not None
assert profile.default_max_tokens > 0
# get_max_tokens() returns the static default regardless of model
# (DeepInfra clamps per-model server-side, so one value is safe).
assert profile.get_max_tokens(None) == profile.default_max_tokens
assert (
profile.get_max_tokens("deepseek-ai/DeepSeek-V4-Flash")
== profile.default_max_tokens
)

View file

@ -0,0 +1,97 @@
"""Tests for the bundled DeepInfra image_gen plugin.
Invariants only no snapshots of specific model ids. Most surface-level
contracts (network-failure empty list, tag filtering, no-model error)
are covered by the shared tag-filter test in
``tests/hermes_cli/test_api_key_providers.py``; these two tests pin the
plugin-specific bits that wrapper doesn't reach.
"""
from __future__ import annotations
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
import pytest
import plugins.image_gen.deepinfra as deepinfra_plugin
# 1×1 transparent PNG — valid bytes for save_b64_image()
_PNG_HEX = (
"89504e470d0a1a0a0000000d49484452000000010000000108060000001f15c4"
"890000000d49444154789c6300010000000500010d0a2db40000000049454e44"
"ae426082"
)
def _b64_png() -> str:
import base64
return base64.b64encode(bytes.fromhex(_PNG_HEX)).decode()
@pytest.fixture(autouse=True)
def _isolation(tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
import hermes_cli.models as _models_mod
monkeypatch.setattr(_models_mod, "_deepinfra_catalog_cache", {})
monkeypatch.setenv("DEEPINFRA_API_KEY", "test-key")
yield
def test_list_models_filters_by_image_gen_tag(monkeypatch):
"""Plugin-side wiring: list_models() returns only ``image-gen``-tagged
catalog entries and surfaces pricing + default dims when present."""
import json
import urllib.request
class _Resp:
def __enter__(self): return self
def __exit__(self, *a): return False
def read(self):
return json.dumps({"data": [
{"id": "vendor/chat", "metadata": {"tags": ["chat"]}},
{"id": "vendor/img", "metadata": {
"tags": ["image-gen"],
"pricing": {"per_image_unit": 0.005},
"default_width": 1024,
}},
]}).encode()
monkeypatch.setattr(urllib.request, "urlopen", lambda *a, **kw: _Resp())
rows = deepinfra_plugin.DeepInfraImageGenProvider().list_models()
ids = {row["id"] for row in rows}
assert ids == {"vendor/img"}
img = next(row for row in rows if row["id"] == "vendor/img")
assert "price" in img and img["default_width"] == 1024
def test_generate_calls_openai_sdk_with_deepinfra_base_url(monkeypatch):
"""Happy path: pinned model → openai SDK called with DeepInfra
base_url + Bearer key b64 saved to cache."""
monkeypatch.setenv("DEEPINFRA_IMAGE_MODEL", "vendor/test-img")
captured: dict = {}
class _FakeImages:
def generate(self, **kwargs):
captured["kwargs"] = kwargs
return SimpleNamespace(data=[SimpleNamespace(b64_json=_b64_png(), url=None)])
class _FakeClient:
def __init__(self, api_key=None, base_url=None):
captured["api_key"] = api_key
captured["base_url"] = base_url
self.images = _FakeImages()
fake_openai = MagicMock()
fake_openai.OpenAI = _FakeClient
with patch.dict("sys.modules", {"openai": fake_openai}):
result = deepinfra_plugin.DeepInfraImageGenProvider().generate(
prompt="a cat", aspect_ratio="square",
)
assert result["success"] is True
assert "deepinfra" in captured["base_url"]
assert captured["api_key"] == "test-key"
assert captured["kwargs"]["model"] == "vendor/test-img"

View file

@ -0,0 +1,218 @@
"""Tests for the bundled DeepInfra video_gen plugin.
Invariants only no snapshots of specific model ids. The plugin is a thin
subclass of ``agent.video_gen_provider.OpenAICompatibleVideoGenProvider``;
these tests pin the plugin-specific bits (tag filtering, identity) and the
shared base behaviour exercised through it (OpenAI ``videos`` call shape,
t2v vs i2v routing, download save).
"""
from __future__ import annotations
from contextlib import contextmanager
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
import pytest
import plugins.video_gen.deepinfra as deepinfra_plugin
@pytest.fixture(autouse=True)
def _isolation(tmp_path, monkeypatch):
monkeypatch.setenv("HERMES_HOME", str(tmp_path))
import hermes_cli.models as _models_mod
monkeypatch.setattr(_models_mod, "_deepinfra_catalog_cache", {})
monkeypatch.setenv("DEEPINFRA_API_KEY", "test-key")
yield
def test_identity_and_availability(monkeypatch):
p = deepinfra_plugin.DeepInfraVideoGenProvider()
assert p.name == "deepinfra"
assert p.display_name == "DeepInfra"
assert p._base_url() == "https://api.deepinfra.com/v1/openai"
assert p.is_available() is True
monkeypatch.delenv("DEEPINFRA_API_KEY", raising=False)
assert p.is_available() is False
def test_list_models_filters_by_video_gen_tag(monkeypatch):
"""list_models() returns only ``video-gen``-tagged catalog entries."""
import hermes_cli.models as _models_mod
def _fake_by_tag(tag, **kw):
assert tag == "video-gen"
return [
{"id": "vendor/p-video", "metadata": {"description": "fast t2v"}},
{"id": "vendor/wan-t2v", "metadata": {}},
]
monkeypatch.setattr(_models_mod, "_fetch_deepinfra_models_by_tag", _fake_by_tag)
rows = deepinfra_plugin.DeepInfraVideoGenProvider().list_models()
ids = {row["id"] for row in rows}
assert ids == {"vendor/p-video", "vendor/wan-t2v"}
assert all("display" in r for r in rows)
def _fake_openai_with_capture(captured: dict, *, status="succeeded",
data=None, download=b"\x00\x00mp4bytes"):
"""Build a fake ``openai`` module whose videos resource records the call.
Defaults mirror the real DeepInfra job shape: status ``"succeeded"`` and a
``data`` list carrying the delivery URL.
"""
if data is None:
data = [{"url": "https://cdn.example/out.mp4"}]
class _FakeVideos:
def create(self, **kwargs):
captured["kwargs"] = kwargs
# Return a terminal status immediately so the bounded poll in
# OpenAICompatibleVideoGenProvider._create_and_poll exits without
# calling retrieve() or sleeping.
return SimpleNamespace(status=status, id="vid_123", error=None, data=data)
def retrieve(self, video_id):
return SimpleNamespace(status=status, id=video_id, error=None, data=data)
def download_content(self, video_id):
captured["downloaded_id"] = video_id
return SimpleNamespace(read=lambda: download)
class _FakeClient:
def __init__(self, api_key=None, base_url=None):
captured["api_key"] = api_key
captured["base_url"] = base_url
self.videos = _FakeVideos()
fake = MagicMock()
fake.OpenAI = _FakeClient
return fake
@contextmanager
def _mock_url_download(captured: dict, raise_exc: Exception | None = None):
"""Patch the shared ``save_url_video`` helper the base provider calls."""
import agent.video_gen_provider as base
from pathlib import Path
def _fake_save_url_video(url, *, prefix="video", **kw):
captured["url"] = url
if raise_exc:
raise raise_exc
return Path(f"/home/x/.hermes/cache/videos/{prefix}_test.mp4")
with patch.object(base, "save_url_video", _fake_save_url_video):
yield
def test_generate_text_to_video_downloads_url_and_saves_locally():
"""t2v happy path: SDK called with DeepInfra base_url + key; status
'succeeded' + data[].url bytes downloaded and saved to a local file."""
captured: dict = {}
with patch.dict("sys.modules", {"openai": _fake_openai_with_capture(captured)}), \
_mock_url_download(captured):
result = deepinfra_plugin.DeepInfraVideoGenProvider().generate(
prompt="a red cube rotating", model="vendor/test-vid", duration=5,
)
assert result["success"] is True
assert result["modality"] == "text"
assert result["video"].endswith(".mp4") and "cache/videos" in result["video"]
assert captured["url"] == "https://cdn.example/out.mp4"
assert "deepinfra" in captured["base_url"]
assert captured["api_key"] == "test-key"
assert captured["kwargs"]["model"] == "vendor/test-vid"
assert captured["kwargs"]["seconds"] == "5"
# No image_url ⇒ no image-to-video field passed through.
assert "image_url" not in captured["kwargs"].get("extra_body", {})
def test_generate_returns_url_when_local_save_fails():
"""If downloading the delivery URL fails, fall back to returning the URL."""
captured: dict = {}
with patch.dict("sys.modules", {"openai": _fake_openai_with_capture(captured)}), \
_mock_url_download(captured, raise_exc=OSError("network down")):
result = deepinfra_plugin.DeepInfraVideoGenProvider().generate(
prompt="x", model="vendor/test-vid",
)
assert result["success"] is True
assert result["video"] == "https://cdn.example/out.mp4"
def test_generate_falls_back_to_download_when_no_url():
"""OpenAI/Sora style: no data[].url → download_content bytes saved locally."""
captured: dict = {}
fake = _fake_openai_with_capture(captured, status="completed", data=[])
with patch.dict("sys.modules", {"openai": fake}):
result = deepinfra_plugin.DeepInfraVideoGenProvider().generate(
prompt="x", model="vendor/test-vid",
)
assert result["success"] is True
assert captured["downloaded_id"] == "vid_123"
assert result["video"].endswith(".mp4")
def test_generate_image_to_video_routes_via_extra_body():
"""Presence of image_url routes to i2v and rides in extra_body."""
captured: dict = {}
with patch.dict("sys.modules", {"openai": _fake_openai_with_capture(captured)}), \
_mock_url_download(captured):
result = deepinfra_plugin.DeepInfraVideoGenProvider().generate(
prompt="animate this", model="vendor/test-vid",
image_url="https://example.com/cat.jpg", negative_prompt="blurry",
)
assert result["success"] is True
assert result["modality"] == "image"
extra = captured["kwargs"]["extra_body"]
assert extra["image_url"] == "https://example.com/cat.jpg"
assert extra["negative_prompt"] == "blurry"
def test_generate_errors_when_key_missing(monkeypatch):
monkeypatch.delenv("DEEPINFRA_API_KEY", raising=False)
result = deepinfra_plugin.DeepInfraVideoGenProvider().generate(
prompt="x", model="vendor/test-vid",
)
assert result["success"] is False
assert result["error_type"] == "missing_credentials"
def test_generate_errors_when_job_not_completed():
"""A non-completed job status surfaces a JSON-serializable job_failed error.
``video.error`` is a structured SDK object (pydantic ``VideoCreateError``),
not a string the provider must str() it so the response dict survives the
tool layer's ``json.dumps``. We simulate that with a non-serializable object.
"""
import json
captured: dict = {}
fake = _fake_openai_with_capture(captured)
class _NonSerializableError:
def __str__(self):
return "content policy violation"
class _FailingVideos:
def create(self, **kwargs):
return SimpleNamespace(
status="failed", id="vid_x", error=_NonSerializableError(), data=None
)
def retrieve(self, video_id): # pragma: no cover - status already terminal
return SimpleNamespace(status="failed", id=video_id, error=None, data=None)
def _client(api_key=None, base_url=None):
return SimpleNamespace(videos=_FailingVideos())
fake.OpenAI = _client
with patch.dict("sys.modules", {"openai": fake}):
result = deepinfra_plugin.DeepInfraVideoGenProvider().generate(
prompt="x", model="vendor/test-vid",
)
assert result["success"] is False
assert result["error_type"] == "job_failed"
assert "content policy violation" in result["error"]
# Must not raise — this is the regression the str() guard prevents.
json.dumps(result)

View file

@ -17,14 +17,18 @@ class TestTTSProviderNullGuard:
"""YAML ``tts: {provider: null}`` should fall back to default."""
from tools.tts_tool import _get_provider, DEFAULT_PROVIDER
result = _get_provider({"provider": None})
# Pin the active inference provider to a non-TTS one so the
# active-provider fallback doesn't fire — isolates the null guard.
with patch("tools.tts_tool._active_model_provider", return_value="anthropic"):
result = _get_provider({"provider": None})
assert result == DEFAULT_PROVIDER.lower().strip()
def test_missing_provider_returns_default(self):
"""No ``provider`` key at all should also return default."""
"""No ``provider`` key + non-TTS active provider should return default."""
from tools.tts_tool import _get_provider, DEFAULT_PROVIDER
result = _get_provider({})
with patch("tools.tts_tool._active_model_provider", return_value="anthropic"):
result = _get_provider({})
assert result == DEFAULT_PROVIDER.lower().strip()
def test_valid_provider_passed_through(self):
@ -33,6 +37,37 @@ class TestTTSProviderNullGuard:
result = _get_provider({"provider": "OPENAI"})
assert result == "openai"
def test_falls_back_to_active_tts_capable_provider_when_available(self):
"""No explicit tts.provider + a TTS-capable, credentialled active
provider use it. DeepInfra/OpenAI are in BUILTIN_TTS_PROVIDERS, so a
single-provider deployment gets matching TTS without configuring
tts.provider but only when the backend can authenticate."""
from tools.tts_tool import _get_provider
with patch("tools.tts_tool._active_model_provider", return_value="deepinfra"), \
patch("tools.tts_tool._tts_provider_available", return_value=True):
assert _get_provider({}) == "deepinfra"
with patch("tools.tts_tool._active_model_provider", return_value="openai"), \
patch("tools.tts_tool._tts_provider_available", return_value=True):
assert _get_provider({"provider": None}) == "openai"
def test_active_provider_without_credentials_keeps_edge(self):
"""A TTS-capable active provider that can't authenticate must NOT
silently displace the free Edge default (no surprise billing / hard
errors for a credential-less deployment)."""
from tools.tts_tool import _get_provider, DEFAULT_PROVIDER
with patch("tools.tts_tool._active_model_provider", return_value="openai"), \
patch("tools.tts_tool._tts_provider_available", return_value=False):
assert _get_provider({}) == DEFAULT_PROVIDER.lower().strip()
def test_explicit_provider_wins_over_active(self):
"""An explicit tts.provider always overrides the active-provider fallback."""
from tools.tts_tool import _get_provider
with patch("tools.tts_tool._active_model_provider", return_value="deepinfra"):
assert _get_provider({"provider": "edge"}) == "edge"
# ── Web tools ─────────────────────────────────────────────────────────────

View file

@ -97,3 +97,92 @@ class TestPluginDispatch:
assert payload["success"] is True
assert payload["provider"] == "codex"
assert payload["aspect_ratio"] == "portrait"
def test_auto_dispatches_to_matching_provider_when_image_gen_unset(self, monkeypatch):
"""``image_gen.provider`` unset → dispatch to the registry's active
provider (resolved via _resolve_active_image_provider), else fall
through (None)."""
from tools import image_generation_tool
from agent import image_gen_registry as registry_module
from hermes_cli import plugins as plugins_module
monkeypatch.setattr(image_generation_tool, "_read_configured_image_provider", lambda: None)
monkeypatch.setattr(image_generation_tool, "_resolve_active_image_provider", lambda: "codex")
monkeypatch.setattr(plugins_module, "_ensure_plugins_discovered", lambda *a, **kw: None)
image_gen_registry.register_provider(_FakeCodexProvider())
monkeypatch.setattr(
registry_module, "get_provider",
lambda name: _FakeCodexProvider() if name == "codex" else None,
)
# Active provider resolved → auto-dispatch.
dispatched = image_generation_tool._dispatch_to_plugin_provider("draw cat", "landscape")
assert dispatched is not None
assert json.loads(dispatched)["provider"] == "codex"
# Nothing available (or legacy FAL) → returns None (caller drops to
# the in-tree FAL pipeline).
monkeypatch.setattr(image_generation_tool, "_resolve_active_image_provider", lambda: None)
assert image_generation_tool._dispatch_to_plugin_provider("draw cat", "landscape") is None
monkeypatch.setattr(image_generation_tool, "_resolve_active_image_provider", lambda: "fal")
assert image_generation_tool._dispatch_to_plugin_provider("draw cat", "landscape") is None
def test_deepinfra_bootstrap_no_config_changes_needed(self, monkeypatch):
"""Bootstrap regression: with ``DEEPINFRA_API_KEY`` set and no FAL
credentials, the dispatcher must route to the bundled DeepInfra plugin
without any ``image_gen.provider`` entry i.e. the user never sees the
FAL ``FAL_KEY isn't set`` fallback. The unset-config path now goes
through the availability-filtered registry (get_active_provider), so
the single credentialled backend (DeepInfra) is selected automatically."""
from tools import image_generation_tool
from hermes_cli import plugins as plugins_module
from plugins.image_gen import deepinfra as deepinfra_plugin
from plugins.image_gen.deepinfra import DeepInfraImageGenProvider
# Simulate: DEEPINFRA_API_KEY set, no FAL_KEY, fresh-out-of-box config.
monkeypatch.setenv("DEEPINFRA_API_KEY", "sk-test-bootstrap")
monkeypatch.delenv("FAL_KEY", raising=False)
monkeypatch.setattr(image_generation_tool, "_read_configured_image_provider", lambda: None)
monkeypatch.setattr(image_generation_tool, "_read_configured_image_model", lambda: None)
monkeypatch.setattr(plugins_module, "_ensure_plugins_discovered", lambda *a, **kw: None)
# Only DeepInfra is registered (autouse fixture reset the registry), so
# the registry's single-available fallback selects it — no bespoke
# model.provider inference needed.
image_gen_registry.register_provider(DeepInfraImageGenProvider())
# Stub the live catalog so DeepInfra has at least one model to pick.
from hermes_cli import models as models_mod
monkeypatch.setattr(
models_mod, "_fetch_deepinfra_models_by_tag",
lambda tag, **kw: (
[{"id": "black-forest-labs/FLUX.1-dev", "metadata": {}}]
if tag == "image-gen" else []
),
)
# Avoid a real network fetch when caching the (stubbed) delivery URL.
monkeypatch.setattr(
deepinfra_plugin, "save_url_image",
lambda url, **kw: __import__("pathlib").Path("/tmp/deepinfra_test.png"),
)
# Stub openai so we don't hit the network.
import openai
class _Images:
def generate(self, **kw):
class _Resp:
class _Data:
b64_json = None
url = "https://example.com/img.png"
data = [_Data()]
return _Resp()
class _Client:
def __init__(self, **kw):
self.images = _Images()
monkeypatch.setattr(openai, "OpenAI", _Client)
dispatched = image_generation_tool._dispatch_to_plugin_provider("a cat", "square")
assert dispatched is not None, "auto-resolution must dispatch to DeepInfra — falling through to FAL is the bug"
payload = json.loads(dispatched)
assert payload["provider"] == "deepinfra"
assert payload["model"] == "black-forest-labs/FLUX.1-dev"

View file

@ -0,0 +1,66 @@
"""Tests for the DeepInfra STT provider.
``_transcribe_deepinfra`` is a thin shim that resolves credentials/model
then delegates to ``_transcribe_openai``. These two tests pin the
STT-specific gating (so an unset DEEPINFRA_API_KEY refuses dispatch) and
the delegation happy path; shared catalog/tag-filter behavior is covered
in ``tests/hermes_cli/test_api_key_providers.py``.
"""
from __future__ import annotations
from unittest.mock import MagicMock, patch
import pytest
@pytest.fixture(autouse=True)
def _isolation(monkeypatch):
import hermes_cli.models as _models_mod
monkeypatch.setattr(_models_mod, "_deepinfra_catalog_cache", {})
yield
def test_get_provider_gating_keys_on_deepinfra_api_key(monkeypatch):
"""Explicit-provider gate: DEEPINFRA_API_KEY presence flips ``deepinfra`` on/off."""
monkeypatch.delenv("DEEPINFRA_API_KEY", raising=False)
from tools.transcription_tools import _get_provider
assert _get_provider({"provider": "deepinfra"}) == "none"
monkeypatch.setenv("DEEPINFRA_API_KEY", "test-key")
assert _get_provider({"provider": "deepinfra"}) == "deepinfra"
def test_delegates_to_openai_handler_with_deepinfra_creds(monkeypatch, tmp_path):
"""Happy path: pinned model → openai SDK invoked with DeepInfra base_url + key,
and the response carries ``provider="deepinfra"`` (not openai)."""
monkeypatch.setenv("DEEPINFRA_API_KEY", "test-key")
audio = tmp_path / "speech.wav"
audio.write_bytes(b"\x00" * 16)
captured: dict = {}
class _FakeClient:
def __init__(self, api_key=None, base_url=None, timeout=None, max_retries=None):
captured["api_key"] = api_key
captured["base_url"] = base_url
transcriptions = MagicMock()
transcriptions.create = MagicMock(return_value=MagicMock(text="ok"))
self.audio = MagicMock(transcriptions=transcriptions)
def close(self):
pass
fake_openai = MagicMock()
fake_openai.OpenAI = _FakeClient
fake_openai.APIError = Exception
fake_openai.APIConnectionError = ConnectionError
fake_openai.APITimeoutError = TimeoutError
with patch.dict("sys.modules", {"openai": fake_openai}), \
patch("tools.transcription_tools._load_stt_config", return_value={}):
from tools.transcription_tools import _transcribe_deepinfra
result = _transcribe_deepinfra(str(audio), "vendor/test-stt")
assert result["success"] is True
assert result["provider"] == "deepinfra"
assert "deepinfra" in captured["base_url"]
assert captured["api_key"] == "test-key"

View file

@ -0,0 +1,59 @@
"""Tests for the DeepInfra TTS provider.
``_generate_deepinfra_tts`` is a thin shim that resolves credentials/model
then delegates to ``_generate_openai_tts``. These two tests pin the
delegation happy path and the no-hardcoded-fallback contract; shared
infrastructure (catalog fetch + tag filter) is covered in
``tests/hermes_cli/test_api_key_providers.py``.
"""
from __future__ import annotations
from unittest.mock import MagicMock, patch
import pytest
@pytest.fixture(autouse=True)
def _isolation(monkeypatch):
import hermes_cli.models as _models_mod
monkeypatch.setattr(_models_mod, "_deepinfra_catalog_cache", {})
monkeypatch.setenv("DEEPINFRA_API_KEY", "test-key")
yield
def test_raises_when_no_model_resolvable(monkeypatch, tmp_path):
"""No-fallback contract: empty config + unreachable catalog → ValueError."""
import urllib.request
monkeypatch.setattr(
urllib.request, "urlopen",
lambda *a, **kw: (_ for _ in ()).throw(Exception("offline")),
)
from tools.tts_tool import _generate_deepinfra_tts
with pytest.raises(ValueError, match="No DeepInfra TTS model available"):
_generate_deepinfra_tts("hi", str(tmp_path / "out.mp3"), {})
def test_delegates_to_openai_handler_with_deepinfra_creds(monkeypatch, tmp_path):
"""Happy path: pinned model → openai SDK invoked with DeepInfra base_url + key."""
captured: dict = {}
class _FakeClient:
def __init__(self, api_key=None, base_url=None):
captured["api_key"] = api_key
captured["base_url"] = base_url
speech = MagicMock()
speech.create = MagicMock(return_value=MagicMock(stream_to_file=lambda p: None))
self.audio = MagicMock(speech=speech)
def close(self):
pass
with patch("tools.tts_tool._import_openai_client", return_value=_FakeClient):
from tools.tts_tool import _generate_deepinfra_tts
_generate_deepinfra_tts(
"hello", str(tmp_path / "out.mp3"),
{"deepinfra": {"model": "vendor/test-tts"}},
)
assert "deepinfra" in captured["base_url"]
assert captured["api_key"] == "test-key"

View file

@ -88,7 +88,10 @@ class TestDynamicSchemaBuilder:
from tools.video_generation_tool import _build_dynamic_video_schema
desc = _build_dynamic_video_schema()["description"]
assert "No video backend is configured" in desc
# No provider configured AND none available → description says so. The
# wording reflects the *resolved* active provider (mirrors execution),
# so it reads "available" rather than "configured".
assert "No video backend is available" in desc
assert "hermes tools" in desc
def test_generic_description_keeps_edit_extend_out_of_surface(self, cfg_home):

View file

@ -1248,7 +1248,7 @@ def _read_configured_image_model():
def _read_configured_image_provider():
"""Return the value of ``image_gen.provider`` from config.yaml, or None.
"""Return ``image_gen.provider`` from config.yaml, or None.
We only consult the plugin registry when this is explicitly set an
unset value keeps users on the in-tree FAL fallback even when other
@ -1271,6 +1271,29 @@ def _read_configured_image_provider():
return None
def _resolve_active_image_provider() -> Optional[str]:
"""Return the registry's active image-gen provider name, or ``None``.
Used only when ``image_gen.provider`` is unset. Delegates to
:func:`agent.image_gen_registry.get_active_provider`, which selects the
single provider that has credentials (else the legacy FAL preference)
the same availability-filtered fallback the video surface uses. So a
box whose only image credential is ``DEEPINFRA_API_KEY`` auto-selects
DeepInfra, while a box with FAL credentials keeps FAL, without this tool
re-implementing provider inference or env auto-detect.
"""
try:
from agent.image_gen_registry import get_active_provider
from hermes_cli.plugins import _ensure_plugins_discovered
_ensure_plugins_discovered()
provider = get_active_provider()
if provider is not None:
return provider.name
except Exception as exc:
logger.debug("image_gen active-provider resolution skipped: %s", exc)
return None
def _dispatch_to_plugin_provider(
prompt: str,
aspect_ratio: str,
@ -1293,8 +1316,17 @@ def _dispatch_to_plugin_provider(
route to its edit endpoint.
"""
configured = _read_configured_image_provider()
if configured == "fal":
return None # explicit opt-in to legacy FAL
if not configured:
return None
# Let the registry pick the active backend (single available provider,
# else legacy FAL preference). ``fal`` or nothing → fall through to the
# in-tree FAL pipeline; any other available backend (e.g. DeepInfra on
# a box whose only image credential is DEEPINFRA_API_KEY) dispatches.
active = _resolve_active_image_provider()
if not active or active == "fal":
return None
configured = active
# Also read configured model so we can pass it to the plugin
configured_model = _read_configured_image_model()

View file

@ -99,6 +99,7 @@ GROQ_BASE_URL = os.getenv("GROQ_BASE_URL", "https://api.groq.com/openai/v1")
OPENAI_BASE_URL = os.getenv("STT_OPENAI_BASE_URL", "https://api.openai.com/v1")
XAI_STT_BASE_URL = os.getenv("XAI_STT_BASE_URL", "https://api.x.ai/v1")
ELEVENLABS_STT_BASE_URL = os.getenv("ELEVENLABS_STT_BASE_URL", "https://api.elevenlabs.io/v1")
# DeepInfra STT base URL now resolved via hermes_cli.models.deepinfra_base_url (shared).
SUPPORTED_FORMATS = {".mp3", ".mp4", ".mpeg", ".mpga", ".m4a", ".wav", ".webm", ".ogg", ".aac", ".flac"}
LOCAL_NATIVE_AUDIO_FORMATS = {".wav", ".aiff", ".aif"}
@ -827,11 +828,24 @@ def _get_provider(stt_config: dict) -> str:
)
return "none"
if provider == "deepinfra":
if _HAS_OPENAI and (get_env_value("DEEPINFRA_API_KEY") or "").strip():
return "deepinfra"
logger.warning(
"STT provider 'deepinfra' configured but DEEPINFRA_API_KEY not set "
"(or openai package missing)"
)
return "none"
return provider # Unknown — let it fail downstream
# --- Auto-detect (no explicit provider): local > groq > openai > xai > elevenlabs -
# mistral is intentionally skipped while `mistralai` is quarantined on
# PyPI (malicious 2.4.6 release on 2026-05-12).
# --- Auto-detect (no explicit provider):
# local > groq > openai > mistral > xai > elevenlabs > deepinfra ---
# DeepInfra is tried LAST so adding DEEPINFRA_API_KEY (commonly set for the
# chat surface) never silently displaces an existing xAI/ElevenLabs STT
# auto-selection; a DeepInfra-only box still resolves to it. mistral is
# intentionally skipped while `mistralai` is quarantined on PyPI (malicious
# 2.4.6 release on 2026-05-12).
if _HAS_FASTER_WHISPER:
return "local"
@ -863,6 +877,9 @@ def _get_provider(stt_config: dict) -> str:
if get_env_value("ELEVENLABS_API_KEY"):
logger.info("No local STT available, using ElevenLabs Scribe STT API")
return "elevenlabs"
if _HAS_OPENAI and (get_env_value("DEEPINFRA_API_KEY") or "").strip():
logger.info("No local STT available, using DeepInfra Whisper API")
return "deepinfra"
return "none"
@ -1327,22 +1344,36 @@ def _transcribe_groq(file_path: str, model_name: str) -> Dict[str, Any]:
# ---------------------------------------------------------------------------
def _transcribe_openai(file_path: str, model_name: str) -> Dict[str, Any]:
"""Transcribe using OpenAI Whisper API (paid)."""
try:
api_key, base_url = _resolve_openai_audio_client_config()
except ValueError as exc:
return {
"success": False,
"transcript": "",
"error": str(exc),
}
def _transcribe_openai(
file_path: str,
model_name: str,
*,
api_key: Optional[str] = None,
base_url: Optional[str] = None,
provider_label: str = "openai",
) -> Dict[str, Any]:
"""Transcribe via the OpenAI ``audio.transcriptions.create`` SDK shape.
Also serves as the shared backend for every OpenAI-compatible STT
endpoint (DeepInfra etc.) callers pass an explicit ``api_key`` /
``base_url`` to skip the OpenAI-only auth chain, and a
``provider_label`` so the response carries the right ``provider``
name.
"""
if api_key is None:
try:
api_key, fallback_base = _resolve_openai_audio_client_config()
except ValueError as exc:
return {"success": False, "transcript": "", "error": str(exc)}
base_url = base_url or fallback_base
if not _HAS_OPENAI:
return {"success": False, "transcript": "", "error": "openai package not installed"}
# Auto-correct model if caller passed a Groq-only model
if model_name in GROQ_MODELS:
# Auto-correct model if caller passed a Groq-only model. Only applies
# to the native OpenAI path — third-party endpoints may legitimately
# serve a whisper-large-v3 variant.
if provider_label == "openai" and model_name in GROQ_MODELS:
logger.info("Model %s not available on OpenAI, using %s", model_name, DEFAULT_STT_MODEL)
model_name = DEFAULT_STT_MODEL
@ -1358,10 +1389,12 @@ def _transcribe_openai(file_path: str, model_name: str) -> Dict[str, Any]:
)
transcript_text = _extract_transcript_text(transcription)
logger.info("Transcribed %s via OpenAI API (%s, %d chars)",
Path(file_path).name, model_name, len(transcript_text))
logger.info(
"Transcribed %s via %s (%s, %d chars)",
Path(file_path).name, provider_label, model_name, len(transcript_text),
)
return {"success": True, "transcript": transcript_text, "provider": "openai"}
return {"success": True, "transcript": transcript_text, "provider": provider_label}
finally:
close = getattr(client, "close", None)
if callable(close):
@ -1376,7 +1409,7 @@ def _transcribe_openai(file_path: str, model_name: str) -> Dict[str, Any]:
except APIError as e:
return {"success": False, "transcript": "", "error": f"API error: {e}"}
except Exception as e:
logger.error("OpenAI transcription failed: %s", e, exc_info=True)
logger.error("%s transcription failed: %s", provider_label, e, exc_info=True)
return {"success": False, "transcript": "", "error": f"Transcription failed: {e}"}
# ---------------------------------------------------------------------------
@ -1616,6 +1649,59 @@ def _transcribe_elevenlabs(file_path: str, model_name: str) -> Dict[str, Any]:
return {"success": False, "transcript": "", "error": f"ElevenLabs STT transcription failed: {e}"}
# ---------------------------------------------------------------------------
# Provider: DeepInfra (OpenAI-compatible /v1/audio/transcriptions)
# ---------------------------------------------------------------------------
def _transcribe_deepinfra(file_path: str, model_name: str) -> Dict[str, Any]:
"""Resolve DeepInfra credentials/model, then delegate to the OpenAI handler.
DeepInfra's STT endpoint is OpenAI-compatible, so the actual SDK
call lives in :func:`_transcribe_openai` this wrapper only owns
DeepInfra-specific credential and model resolution, using the shared
``hermes_cli.models`` helpers so every DeepInfra surface resolves the
base URL and model ids identically.
"""
api_key = (get_env_value("DEEPINFRA_API_KEY") or "").strip()
if not api_key:
return {"success": False, "transcript": "", "error": "DEEPINFRA_API_KEY not set"}
from hermes_cli.models import deepinfra_base_url, deepinfra_model_ids
stt_config = _load_stt_config()
# ``stt.deepinfra: null`` in YAML yields None, not {} — coalesce so the
# ``.get`` calls don't raise (no stt.deepinfra block in DEFAULT_CONFIG to
# deep-merge over the null).
di_config = stt_config.get("deepinfra") if isinstance(stt_config, dict) else None
if not isinstance(di_config, dict):
di_config = {}
base_url = deepinfra_base_url(di_config)
if not model_name:
candidates = deepinfra_model_ids("stt")
if not candidates:
return {
"success": False,
"transcript": "",
"error": (
"No DeepInfra STT model available. Pin one in "
"config.yaml under stt.deepinfra.model, or check "
"connectivity to api.deepinfra.com so the live catalog "
"can be fetched."
),
}
model_name = candidates[0]
return _transcribe_openai(
file_path,
model_name,
api_key=api_key,
base_url=base_url,
provider_label="deepinfra",
)
# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------
@ -1694,6 +1780,12 @@ def transcribe_audio(file_path: str, model: Optional[str] = None) -> Dict[str, A
model_name = model or elevenlabs_cfg.get("model_id", DEFAULT_ELEVENLABS_STT_MODEL)
return _transcribe_elevenlabs(file_path, model_name)
if provider == "deepinfra":
di_config = stt_config.get("deepinfra") # may be None (YAML null)
di_config = di_config if isinstance(di_config, dict) else {}
model_name = model or di_config.get("model") or ""
return _transcribe_deepinfra(file_path, model_name)
# User-declared command-type provider
# (``stt.providers.<name>: type: command``). Fires after the built-in
# elif chain — built-in names short-circuit upstream so a user's

View file

@ -205,6 +205,8 @@ DEFAULT_GEMINI_TTS_VOICE = "Kore"
DEFAULT_GEMINI_TTS_BASE_URL = "https://generativelanguage.googleapis.com/v1beta"
DEFAULT_GEMINI_AUDIO_TAGS = False
GEMINI_AUDIO_TAG_REWRITE_TASK = "tts_audio_tags"
# Base URL now resolved via hermes_cli.models.deepinfra_base_url (shared).
DEFAULT_DEEPINFRA_TTS_VOICE = "default"
# PCM output specs for Gemini TTS (fixed by the API)
GEMINI_TTS_SAMPLE_RATE = 24000
GEMINI_TTS_CHANNELS = 1
@ -350,9 +352,72 @@ def _load_tts_config() -> Dict[str, Any]:
return {}
def _active_model_provider() -> str:
"""Return the active inference provider name.
Reuses :func:`agent.auxiliary_client._read_main_provider` so a runtime
provider switch (``--provider`` flag, per-session gateway provider,
fallback-model activation) is honored reading ``model.provider`` from
disk directly would miss those. Empty string when unavailable. Used only
as a TTS default hint see :func:`_get_provider`.
"""
try:
from agent.auxiliary_client import _read_main_provider
return (_read_main_provider() or "").lower().strip()
except Exception:
try:
from hermes_cli.config import load_config_readonly
return str((load_config_readonly().get("model") or {}).get("provider") or "").lower().strip()
except Exception:
return ""
# API-key env vars for the cloud TTS backends. Local backends (edge, piper,
# neutts, kittentts) are always available. Used to gate the auto-default: the
# active inference provider's TTS backend is only inherited when it can
# actually authenticate — otherwise we fall back to free Edge instead of
# silently switching a deployment onto a backend that will error at call time.
_TTS_PROVIDER_KEY_ENV_VARS: Dict[str, tuple] = {
"elevenlabs": ("ELEVENLABS_API_KEY",),
"openai": ("OPENAI_API_KEY",),
"minimax": ("MINIMAX_API_KEY",),
"xai": ("XAI_API_KEY",),
"mistral": ("MISTRAL_API_KEY",),
"gemini": ("GEMINI_API_KEY", "GOOGLE_API_KEY"),
"deepinfra": ("DEEPINFRA_API_KEY",),
}
def _tts_provider_available(name: str) -> bool:
"""Return True when TTS backend *name* has usable credentials.
Local/command backends (not in the key map) are always considered
available.
"""
env_vars = _TTS_PROVIDER_KEY_ENV_VARS.get(name)
if env_vars is None:
return True
return any((get_env_value(v) or "").strip() for v in env_vars)
def _get_provider(tts_config: Dict[str, Any]) -> str:
"""Get the configured TTS provider name."""
return (tts_config.get("provider") or DEFAULT_PROVIDER).lower().strip()
"""Get the configured TTS provider name.
When ``tts.provider`` is set it always wins. With no explicit provider,
fall back to the active inference provider *if* it ships a built-in TTS
backend AND that backend can authenticate so a single-provider
deployment gets matching TTS out of the box without silently switching a
credential-less deployment off the free Edge default. Anything else keeps
the historical Edge default. Provider-agnostic: no backend is
special-cased here.
"""
explicit = (tts_config.get("provider") or "").lower().strip()
if explicit:
return explicit
active = _active_model_provider()
if active and active in BUILTIN_TTS_PROVIDERS and _tts_provider_available(active):
return active
return DEFAULT_PROVIDER
# ===========================================================================
@ -397,6 +462,7 @@ BUILTIN_TTS_PROVIDERS = frozenset({
"neutts",
"kittentts",
"piper",
"deepinfra",
})
DEFAULT_COMMAND_TTS_TIMEOUT_SECONDS = 120
@ -1009,36 +1075,92 @@ def _generate_elevenlabs(text: str, output_path: str, tts_config: Dict[str, Any]
return output_path
def _tts_response_format_from_path(output_path: str) -> str:
"""Pick an OpenAI-compatible TTS response format from the output extension."""
if output_path.endswith(".ogg"):
return "opus"
if output_path.endswith(".wav"):
return "wav"
if output_path.endswith(".flac"):
return "flac"
return "mp3"
# ===========================================================================
# Provider: OpenAI TTS
# Provider: OpenAI TTS (also used by every OpenAI-compatible TTS endpoint —
# DeepInfra delegates here via _generate_deepinfra_tts).
# ===========================================================================
def _generate_openai_tts(text: str, output_path: str, tts_config: Dict[str, Any]) -> str:
"""
Generate audio using OpenAI TTS.
def _generate_openai_tts(
text: str,
output_path: str,
tts_config: Dict[str, Any],
*,
api_key: Optional[str] = None,
base_url: Optional[str] = None,
model: Optional[str] = None,
voice: Optional[str] = None,
speed: Optional[float] = None,
) -> str:
"""Generate audio via the OpenAI ``audio.speech.create`` SDK shape.
Optional kwargs let OpenAI-compatible backends (DeepInfra etc.) reuse
this function they resolve credentials/model themselves and pass
them through, skipping the OpenAI-only ``_resolve_openai_audio_client_config``.
Args:
text: Text to convert.
output_path: Where to save the audio file.
tts_config: TTS config dict.
tts_config: TTS config dict (used for ``tts.openai`` sub-block
and the global ``speed`` default).
api_key: Bearer token. When None, resolved from the OpenAI auth
chain (config env managed gateway).
base_url: API base URL. When None, falls back to
``tts.openai.base_url`` then the OpenAI default.
model: Model id. When None, reads ``tts.openai.model``.
voice: Voice id. When None, reads ``tts.openai.voice``.
speed: Playback speed. When None, reads ``tts.openai.speed`` /
``tts.speed``.
Returns:
Path to the saved audio file.
"""
api_key, base_url, is_managed = _resolve_openai_audio_client_config()
# Only resolve the OpenAI auth chain when the caller didn't pass explicit
# credentials. OpenAI-compatible backends (DeepInfra) pass api_key /
# base_url / model / voice through and never hit the managed-gateway path.
fallback_base: Optional[str] = None
is_managed = False
explicit_base_url = base_url is not None
if api_key is None:
api_key, fallback_base, is_managed = _resolve_openai_audio_client_config()
oai_config = tts_config.get("openai") or {}
model = oai_config.get("model", DEFAULT_OPENAI_MODEL)
voice = oai_config.get("voice", DEFAULT_OPENAI_VOICE)
custom_base_url = oai_config.get("base_url")
if custom_base_url:
base_url = custom_base_url
speed = float(oai_config.get("speed", tts_config.get("speed", 1.0)))
# ``tts.openai: null`` in YAML yields None — coalesce so .get() is safe.
oai_config = (tts_config.get("openai") if isinstance(tts_config, dict) else None) or {}
if model is None:
model = oai_config.get("model", DEFAULT_OPENAI_MODEL)
if voice is None:
voice = oai_config.get("voice", DEFAULT_OPENAI_VOICE)
config_base_url = oai_config.get("base_url")
if base_url is None:
# Config override wins over the auth-chain fallback (restores the
# pre-refactor precedence, where tts.openai.base_url beat the resolved
# default); the auth-chain value is the last-resort default. An
# explicit base_url arg from an OpenAI-compatible caller (DeepInfra)
# skips this block entirely and always wins.
base_url = config_base_url or fallback_base or DEFAULT_OPENAI_BASE_URL
if speed is None:
speed_default = tts_config.get("speed", 1.0) if isinstance(tts_config, dict) else 1.0
speed = float(oai_config.get("speed", speed_default))
# The managed OpenAI audio gateway only proxies MANAGED_OPENAI_TTS_MODELS.
# A model set for direct OpenAI (e.g. "tts-1-hd") 400s there with
# "Unsupported managed OpenAI speech model", so coerce it — unless the user
# redirected base_url to their own endpoint, in which case respect it.
if is_managed and not custom_base_url and model not in MANAGED_OPENAI_TTS_MODELS:
if (
is_managed
and not explicit_base_url
and not config_base_url
and model not in MANAGED_OPENAI_TTS_MODELS
):
logger.warning(
"TTS: managed OpenAI audio gateway does not support model %r; "
"falling back to %s. Set VOICE_TOOLS_OPENAI_KEY or OPENAI_API_KEY "
@ -1047,16 +1169,12 @@ def _generate_openai_tts(text: str, output_path: str, tts_config: Dict[str, Any]
)
model = DEFAULT_OPENAI_MODEL
# Determine response format from extension
if output_path.endswith(".ogg"):
response_format = "opus"
else:
response_format = "mp3"
response_format = _tts_response_format_from_path(output_path)
OpenAIClient = _import_openai_client()
client = OpenAIClient(api_key=api_key, base_url=base_url)
try:
create_kwargs = {
create_kwargs: Dict[str, Any] = {
"model": model,
"voice": voice,
"input": text,
@ -1075,6 +1193,64 @@ def _generate_openai_tts(text: str, output_path: str, tts_config: Dict[str, Any]
close()
# ===========================================================================
# Provider: DeepInfra TTS
# ===========================================================================
#
# DeepInfra serves TTS over an OpenAI-compatible /v1/openai/audio/speech
# endpoint. Models are discovered live via the shared catalog helper
# (filtered by the ``tts`` surface tag) — no hardcoded model ids in this
# file, so retired models disappear from hermes the next time the
# catalog is fetched without a patch.
def _generate_deepinfra_tts(text: str, output_path: str, tts_config: Dict[str, Any]) -> str:
"""Resolve DeepInfra credentials/model, then delegate to the OpenAI handler.
DeepInfra's audio endpoint is OpenAI-compatible, so there's no need
to duplicate the SDK call we just pass an explicit api_key /
base_url / model / voice through. Model ids and the base URL come from
the shared ``hermes_cli.models`` helpers so every DeepInfra surface
resolves them identically.
"""
api_key = (get_env_value("DEEPINFRA_API_KEY") or "").strip()
if not api_key:
raise ValueError(
"DEEPINFRA_API_KEY not set. Run `hermes setup` to configure, "
"or set the env var directly."
)
# ``tts.deepinfra: null`` in YAML yields None, not {} — coalesce so the
# ``.get`` calls below don't raise AttributeError (there is no
# tts.deepinfra block in DEFAULT_CONFIG to deep-merge over the null).
di_config = tts_config.get("deepinfra") if isinstance(tts_config, dict) else None
if not isinstance(di_config, dict):
di_config = {}
from hermes_cli.models import deepinfra_base_url, deepinfra_model_ids
model = di_config.get("model")
if not isinstance(model, str) or not model.strip():
candidates = deepinfra_model_ids("tts")
if not candidates:
raise ValueError(
"No DeepInfra TTS model available. Pin one in config.yaml "
"under tts.deepinfra.model, or check connectivity to "
"api.deepinfra.com so the live catalog can be fetched."
)
model = candidates[0]
return _generate_openai_tts(
text,
output_path,
tts_config,
api_key=api_key,
base_url=deepinfra_base_url(di_config),
model=model,
voice=di_config.get("voice", DEFAULT_DEEPINFRA_TTS_VOICE),
speed=float(di_config.get("speed", tts_config.get("speed", 1.0))),
)
# ===========================================================================
# Provider: xAI TTS
# ===========================================================================
@ -2294,6 +2470,17 @@ def text_to_speech_tool(
logger.info("Generating speech with OpenAI TTS...")
_generate_openai_tts(text, file_str, tts_config)
elif provider == "deepinfra":
try:
_import_openai_client()
except ImportError:
return json.dumps({
"success": False,
"error": "DeepInfra TTS uses the 'openai' SDK but it isn't installed."
}, ensure_ascii=False)
logger.info("Generating speech with DeepInfra TTS...")
_generate_deepinfra_tts(text, file_str, tts_config)
elif provider == "minimax":
logger.info("Generating speech with MiniMax TTS...")
_generate_minimax_tts(text, file_str, tts_config)

View file

@ -472,29 +472,19 @@ def _build_dynamic_video_schema() -> Dict[str, Any]:
"""
parts: List[str] = [_GENERIC_DESCRIPTION]
configured = _read_configured_video_provider()
configured_model = _read_configured_video_model()
if not configured:
parts.append(
"\nNo video backend is configured. Calls will return an error "
"until the user picks one via `hermes tools` → Video Generation."
)
return {"description": "\n".join(parts)}
try:
from agent.video_gen_registry import get_provider
from hermes_cli.plugins import _ensure_plugins_discovered
_ensure_plugins_discovered()
provider = get_provider(configured)
except Exception:
provider = None
# Reflect the *resolved* active provider (same resolution the handler uses
# in _resolve_active_provider): an explicit ``video_gen.provider``, or —
# when unset — the single available registered backend. Keeping the
# description in sync with execution stops the agent from being told
# "no backend configured" while a call would actually succeed.
provider = _resolve_active_provider()
if provider is None:
parts.append(
f"\nActive backend: {configured} (plugin not yet loaded — the "
f"tool will retry discovery on first call)."
"\nNo video backend is available. Calls will return an error "
"until the user picks one via `hermes tools` → Video Generation."
)
return {"description": "\n".join(parts)}
@ -548,7 +538,7 @@ def _build_dynamic_video_schema() -> Dict[str, Any]:
max_refs = caps.get("max_reference_images") or 0
if max_refs:
parts.append(f"- reference_image_urls: up to {max_refs} images")
if configured == "xai":
if provider.name == "xai":
parts.append(
"- chaining: for edit/extend pass the public HTTPS MP4 in `video` "
"or `public_url` from the prior Imagine result (files-cdn). For "