hermes-admin/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 28 Packages Projects Releases Wiki Activity

feat(providers): add Ramp Router (router.com) provider plugin

Browse Source 
Ramp Router is an OpenAI Responses-compatible LLM gateway at
https://api.router.com/v1 that routes each request across upstream
providers (OpenAI, Anthropic, xAI, Fireworks, ...) with server-side
fallbacks and spend controls. Nous asked for a PR adding it as a
provider, so:

- plugins/model-providers/router/: RouterProfile plugin —
  api_mode=codex_responses, RAMP_ROUTER_API_KEY auth,
  RAMP_ROUTER_BASE_URL override, live account-scoped catalog via
  GET /v1/models (no hardcoded fallback_models: IDs are key-scoped and
  Router's docs mandate runtime catalog reads).
- hermes_cli/providers.host_mandated_api_mode +
  runtime_provider._detect_api_mode_for_url: api.router.com ->
  codex_responses. The host is Responses-only — POST /v1/chat/completions
  does not exist and 404s — so this is a genuine host mandate (exact
  hostname match per #32243, mirroring the api.meta.ai precedent).
- providers/base.py: new overrideable supported_reasoning_efforts(model)
  hook (tri-state: None=defer, ()=model takes no reasoning params,
  tuple=clamp set). Router validates reasoning.effort per model and
  returns HTTP 400 invalid-argument on levels outside the model's
  published vocabulary, and 400 unsupported_parameter when a
  non-reasoning model receives any reasoning field (both verified live).
  The profile answers from a cached copy of the catalog's
  router.capabilities.reasoning block: cache-only on the hot path,
  seeded for free by fetch_models(), disk-mirrored across processes
  (/cache/router_catalog.json), background-warmed when cold
  — same design as the OpenRouter reasoning-caps clamp on the chat path.
- agent/transports/codex.py: consult the profile-declared vocabulary in
  the generic effort-clamp branch (xai/actual/github branches untouched;
  profiles that do not override the hook see no behavior change).
- cli-config.yaml.example + adding-providers.md + providers/README.md:
  document the provider, the host mandate, and the new hook.
- tests: behavior contracts for the host mandate/URL detection/spoof
  rejection, profile registration + auth auto-registry wiring, catalog
  parsing, and transport clamp/suppression/fallback paths.

Verified live against api.router.com (Aug 2026): one-shot chat,
streaming SSE, tool calls + parallel_tool_calls, encrypted-reasoning
replay on OpenAI-served models, function_call_output follow-up turns on
OpenAI- and Fireworks-served models; store:false / prompt_cache_key /
include:[reasoning.encrypted_content] / reasoning.summary accepted
across backends; effort clamp confirmed to convert a would-be 400
(xhigh on o3) into a successful request via the disk mirror.
This commit is contained in:
Neel Patel
2026-08-23 20:14:10 -04:00
committed by kshitij
parent b4b7727ea0
commit 804f8b4732
11 changed files with 744 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

50
agent/transports/codex.py
View File

@@ -236,6 +236,34 @@ def _content_cache_key(
return f"pck_{digest}"
def _profile_declared_efforts(
provider: Any, model: Optional[str]
) -> Optional[tuple]:
"""Provider-profile-declared reasoning-effort vocabulary, or None.
Thin, fail-open wrapper around
``ProviderProfile.supported_reasoning_efforts`` (see providers/base.py
for the tri-state contract). Lazy import: provider plugins import this
transport during registry discovery, so a module-level import of
``providers`` would cycle.
"""
name = str(provider or "").strip().lower()
if not name:
return None
try:
from providers import get_provider_profile
profile = get_provider_profile(name)
if profile is None:
return None
declared = profile.supported_reasoning_efforts(model)
except Exception:
return None
if declared is None:
return None
return tuple(declared)
def _is_azure_foundry_responses(params: Dict[str, Any]) -> bool:
"""Return True for Microsoft Foundry's OpenAI-compatible Responses API.
@@ -513,10 +541,24 @@ class ResponsesApiTransport(ProviderTransport):
# none/low/medium/high/max.
_supported = ACTUAL_RELAY_EFFORTS
else:
# OpenAI/Codex Responses backend — per-model vocabulary
# (live-verified: "max" is gpt-5.6-only, "minimal" always
# rejected). #68365 premise confirmed.
_supported = codex_supported_efforts(model)
# Profile-declared vocabulary first: gateways that validate
# reasoning.effort per model (Ramp Router reads its live catalog)
# declare it via ProviderProfile.supported_reasoning_efforts.
# ``()`` is the definitive "this model takes no reasoning
# parameters" verdict — such backends 400 on any reasoning field
# rather than ignoring it, so suppress reasoning entirely.
_supported = None
_declared = _profile_declared_efforts(params.get("provider"), model)
if _declared is not None:
if not _declared:
reasoning_enabled = False
else:
_supported = _declared
if _supported is None:
# OpenAI/Codex Responses backend — per-model vocabulary
# (live-verified: "max" is gpt-5.6-only, "minimal" always
# rejected). #68365 premise confirmed.
_supported = codex_supported_efforts(model)
reasoning_effort = clamp_effort(reasoning_effort, _supported)
response_tools = _responses_tools(tools)

8
cli-config.yaml.example
View File

@@ -148,6 +148,14 @@ model:
# # api_mode auto-detected as codex_responses for api.meta.ai; no need to set
# # (the bundled meta-ai provider covers this — a named custom provider is
# # only needed for a non-default Meta-compatible endpoint)
# providers:
# router:
# base_url: https://api.router.com/v1
# api_key: ${RAMP_ROUTER_API_KEY}
# # api_mode auto-detected as codex_responses for api.router.com — the host
# # is Responses-only (/v1/chat/completions 404s). The bundled router
# # provider covers this; a named custom provider is only needed for a
# # non-default Router-compatible endpoint.
# Command-minted credentials (optional): key_cmd

7
hermes_cli/providers.py
View File

@@ -663,6 +663,8 @@ def host_mandated_api_mode(base_url: str = "") -> Optional[str]:
- api.meta.ai only achieves KV-cache hits on /v1/responses with
prompt_cache_retention; /v1/chat/completions returns 0 cached
tokens (measured 0% vs 93-99% on /responses with retention).
- api.router.com (Ramp Router) implements ONLY the Responses API —
POST /v1/chat/completions does not exist on the host and 404s.
- api.anthropic.com / ``…/anthropic`` suffixes speak native Messages.
- Kimi's ``/coding`` endpoint speaks native Messages.
- AWS Bedrock runtime hosts speak Converse.
@@ -695,6 +697,11 @@ def host_mandated_api_mode(base_url: str = "") -> Optional[str]:
# cache-cold (0% vs 93-99% measured). Exact-hostname match per #32243.
if hostname == "api.meta.ai":
return "codex_responses"
# Ramp Router (api.router.com) is Responses-only: the host serves
# GET /v1/models and POST /v1/responses, and /v1/chat/completions 404s
# (docs.router.com/api/endpoint). Exact-hostname match per #32243.
if hostname == "api.router.com":
return "codex_responses"
if hostname.startswith("bedrock-runtime.") and base_url_host_matches(base_url, "amazonaws.com"):
return "bedrock_converse"
return None

6
hermes_cli/runtime_provider.py
View File

@@ -163,6 +163,12 @@ def _detect_api_mode_for_url(base_url: str) -> Optional[str]:
return "codex_responses"
if hostname == "api.actual.inc":
return "codex_responses"
# Ramp Router: Responses-only host — /v1/chat/completions does not
# exist and 404s (docs.router.com/api/endpoint). Mirrors the
# host_mandated_api_mode clause in hermes_cli/providers.py so the
# runtime resolver stays in lockstep. Exact hostname per #32243.
if hostname == "api.router.com":
return "codex_responses"
# Direct native Anthropic host: realign with providers.determine_api_mode,
# which already maps this host to anthropic_messages. The exact-hostname
# match rejects lookalike subdomains (api.anthropic.com.attacker.test) and

357
plugins/model-providers/router/__init__.py Normal file
View File

@@ -0,0 +1,357 @@
"""Ramp Router (router.com) provider plugin for Hermes Agent.
Provider profile for `Ramp Router <https://docs.router.com>`_, Ramp's LLM
gateway: one OpenAI Responses-compatible endpoint at
``https://api.router.com/v1`` that routes each request across upstream
providers (OpenAI, Anthropic, xAI, Fireworks, ...) and handles fallbacks and
spend controls server-side.
Wire notes (verified live against api.router.com, Aug 2026):
* **Responses API only.** Router implements ``GET /v1/models`` and
``POST /v1/responses``; ``POST /v1/chat/completions`` does not exist and
404s. ``api_mode="codex_responses"`` plus the ``api.router.com`` host
mandate in ``hermes_cli/providers.py`` keep every path off the chat wire.
* **Account-scoped catalog.** Valid model IDs are whatever the key's
``GET /v1/models`` returns (BYOK accounts see extra entries), so this
profile ships **no** ``fallback_models`` — the picker relies on the live
fetch, per Router's own guidance to never hardcode model names.
* **Strict reasoning-effort validation.** Router validates
``reasoning.effort`` against each model's catalog-declared vocabulary and
returns HTTP 400 ``invalid-argument`` on a level the model does not accept
(e.g. ``max`` on grok-4.6), and 400 ``unsupported_parameter`` when a
non-reasoning model (gpt-4.1 family, gpt-4o, ...) receives any reasoning
field. The catalog publishes the vocabulary per model
(``router.capabilities.reasoning``), so ``supported_reasoning_efforts``
below feeds the codex transport's clamp from a cached copy of it.
* **Everything else passes through.** ``store: false``, ``prompt_cache_key``,
``include: ["reasoning.encrypted_content"]``, and ``reasoning.summary`` are
accepted on all models (ignored where a backend cannot honor them), tools /
``parallel_tool_calls`` / streaming SSE work across backends, and encrypted
reasoning replay round-trips on OpenAI-served models — so the generic
Responses transport path needs no Router-specific request surgery.
The capability cache mirrors the OpenRouter reasoning-caps design in
``hermes_cli/models.py``: cache-only lookups on the per-request hot path
(never HTTP), seeded for free whenever ``fetch_models()`` runs (picker,
setup, doctor), hydrated from a disk mirror across processes, and refreshed
by a background warmer when cold or stale.
"""
from __future__ import annotations
import json
import logging
import os
import threading
import time
from pathlib import Path
from typing import Any, Optional
from providers import register_provider
from providers.base import ProviderProfile, _profile_user_agent
logger = logging.getLogger(__name__)
ROUTER_DEFAULT_BASE_URL = "https://api.router.com/v1"
#: Efforts-by-model cache: ``model id -> list of accepted effort levels``.
#: ``[]`` means the catalog says the model accepts NO reasoning parameters
#: (``reasoning.supported: false``) — the transport must omit reasoning
#: entirely. A model absent from the dict is unknown (custom/BYOK route or
#: vocabulary not published) and callers fall back to their defaults.
_efforts_cache: Optional[dict[str, list[str]]] = None
_efforts_lock = threading.Lock()
_warm_started = False
_disk_checked = False
#: Disk-mirror staleness bound. Vocabularies change rarely; a stale verdict
#: beats no verdict, so a past-TTL mirror is still served while a background
#: refresh runs (same policy as the OpenRouter caps mirror).
_DISK_TTL_SECONDS = 24 * 60 * 60
def _base_url() -> str:
"""Allow a base-URL override via ``RAMP_ROUTER_BASE_URL``."""
return os.getenv("RAMP_ROUTER_BASE_URL", "").strip().rstrip("/") or ROUTER_DEFAULT_BASE_URL
def _resolve_api_key() -> str:
"""Resolve the Router key from .env / environment, preferring dotenv.
``RAMP_ROUTER_API_KEY`` is Router's documented variable;
``ROUTER_API_KEY`` is accepted as a convenience alias. Falls back to the
raw environment when the hermes_cli helper is unavailable (e.g. stripped
test environments).
"""
resolvers = []
try:
from hermes_cli.config import get_env_value_prefer_dotenv
resolvers.append(get_env_value_prefer_dotenv)
except Exception:
pass
resolvers.append(lambda var: os.environ.get(var, ""))
for resolve in resolvers:
for var in ("RAMP_ROUTER_API_KEY", "ROUTER_API_KEY"):
try:
value = str(resolve(var) or "").strip()
except Exception:
value = ""
if value:
return value
return ""
def _parse_efforts(items: Any) -> Optional[dict[str, list[str]]]:
"""Parse a Router ``/v1/models`` ``data`` array into the efforts map.
Returns None when the array has no usable entries, which callers treat
as a failed fetch rather than caching an empty verdict.
"""
if not isinstance(items, list):
return None
efforts_by_id: dict[str, list[str]] = {}
for item in items:
if not isinstance(item, dict):
continue
mid = str(item.get("id") or "").strip()
if not mid:
continue
router_meta = item.get("router")
reasoning = None
if isinstance(router_meta, dict):
capabilities = router_meta.get("capabilities")
if isinstance(capabilities, dict):
reasoning = capabilities.get("reasoning")
if not isinstance(reasoning, dict):
continue
if reasoning.get("supported") is False:
# Definitive negative: any reasoning field 400s on this model.
efforts_by_id[mid] = []
continue
levels = [
str(entry.get("value") or "").strip()
for entry in reasoning.get("efforts") or []
if isinstance(entry, dict) and str(entry.get("value") or "").strip()
]
if levels:
efforts_by_id[mid] = levels
# supported=True with no published vocabulary -> leave the model out
# (unknown), so the transport keeps its default clamp behavior.
return efforts_by_id or None
def _disk_path() -> Optional[Path]:
try:
from hermes_constants import get_hermes_home
return get_hermes_home() / "cache" / "router_catalog.json"
except Exception:
return None
def _save_disk(efforts_by_id: dict[str, list[str]]) -> None:
path = _disk_path()
if path is None:
return
try:
path.parent.mkdir(parents=True, exist_ok=True)
tmp = path.with_suffix(".tmp")
tmp.write_text(
json.dumps({"ts": time.time(), "efforts": efforts_by_id}),
encoding="utf-8",
)
tmp.replace(path)
except Exception as exc:
logger.debug("router: caps disk mirror write failed: %s", exc)
def _load_disk() -> tuple[Optional[dict[str, list[str]]], float]:
path = _disk_path()
if path is None:
return None, 0.0
try:
data = json.loads(path.read_text(encoding="utf-8"))
efforts = data.get("efforts")
if not isinstance(efforts, dict) or not efforts:
return None, 0.0
parsed = {
str(mid): [str(level) for level in levels]
for mid, levels in efforts.items()
if isinstance(levels, list)
}
try:
age = max(0.0, time.time() - float(data.get("ts") or 0))
except (TypeError, ValueError):
age = float(_DISK_TTL_SECONDS)
return (parsed or None), age
except Exception:
return None, 0.0
def _seed_efforts(items: Any) -> Optional[dict[str, list[str]]]:
"""Seed memory + disk caches from a ``/v1/models`` payload."""
global _efforts_cache
parsed = _parse_efforts(items)
if parsed is None:
return None
with _efforts_lock:
_efforts_cache = parsed
_save_disk(parsed)
return parsed
def _fetch_catalog_items(
*, api_key: str = "", base_url: str = "", timeout: float = 8.0
) -> Optional[list]:
"""Fetch the raw ``/v1/models`` ``data`` array. None on any failure."""
url = (base_url or _base_url()).rstrip("/") + "/models"
import urllib.request
from hermes_cli.urllib_security import open_credentialed_url
req = urllib.request.Request(url)
key = api_key or _resolve_api_key()
if key:
req.add_header("Authorization", f"Bearer {key}")
req.add_header("Accept", "application/json")
# Router sits behind a WAF that rejects the default Python-urllib UA.
req.add_header("User-Agent", _profile_user_agent())
try:
with open_credentialed_url(req, timeout=timeout) as resp:
data = json.loads(resp.read().decode())
except Exception as exc:
logger.debug("router: catalog fetch failed: %s", exc)
return None
items = data if isinstance(data, list) else data.get("data", [])
return items if isinstance(items, list) else None
def _efforts_cache_only() -> Optional[dict[str, list[str]]]:
"""Memory, else the disk mirror. Never HTTP (hot-path safe)."""
global _efforts_cache, _disk_checked
with _efforts_lock:
cached = _efforts_cache
if cached is not None:
return cached
if _disk_checked:
return None
_disk_checked = True
parsed, age = _load_disk()
if parsed is None:
return None
with _efforts_lock:
if _efforts_cache is None:
_efforts_cache = parsed
cached = _efforts_cache
if age >= _DISK_TTL_SECONDS:
_warm_efforts_async()
return cached
def _warm_efforts_async() -> None:
"""Refresh the efforts cache in the background, at most once per process."""
global _warm_started
with _efforts_lock:
if _warm_started:
return
_warm_started = True
if not _resolve_api_key():
# Without a key the fetch would 401; the first authenticated
# fetch_models() (picker/setup/doctor) seeds the cache instead.
return
def _refresh() -> None:
items = _fetch_catalog_items()
if items is not None:
_seed_efforts(items)
try:
threading.Thread(
target=_refresh, name="router-caps-warm", daemon=True
).start()
except Exception as exc:
logger.debug("router: caps warmer failed to start: %s", exc)
class RouterProfile(ProviderProfile):
"""Ramp Router — Responses-only gateway with catalog-declared efforts."""
def fetch_models(
self,
*,
api_key: Optional[str] = None,
base_url: Optional[str] = None,
timeout: float = 8.0,
) -> Optional[list[str]]:
"""Fetch the live, key-scoped catalog and seed the caps cache.
One request serves both consumers: the picker gets the model IDs and
the reasoning-vocabulary mirror is left warm at no extra network
cost (the same document carries both).
"""
items = _fetch_catalog_items(
api_key=api_key or "", base_url=base_url or "", timeout=timeout
)
if items is None:
return None
_seed_efforts(items)
ids = [
str(item["id"])
for item in items
if isinstance(item, dict) and item.get("id")
]
return ids or None
def supported_reasoning_efforts(
self, model: Optional[str]
) -> Optional[tuple[str, ...]]:
"""Catalog-declared effort vocabulary for *model* (cache-only).
Router 400s on efforts outside a model's published set and on any
reasoning field for non-reasoning models, so the codex transport
clamps (or suppresses) from this verdict. Cold cache returns None —
the transport keeps its defaults — and kicks a background warmer so
the next turn is covered.
"""
mid = str(model or "").strip()
if not mid:
return None
efforts_by_id = _efforts_cache_only()
if efforts_by_id is None:
_warm_efforts_async()
return None
levels = efforts_by_id.get(mid)
if levels is None:
return None
return tuple(levels)
router = RouterProfile(
name="router",
aliases=("ramp-router", "ramp", "router.com"),
api_mode="codex_responses",
display_name="Ramp Router",
description="Ramp Router (router.com) — routes each request to the cheapest model that clears your quality bar",
signup_url="https://app.router.com/keys",
# RAMP_ROUTER_API_KEY is Router's documented variable; ROUTER_API_KEY is
# a convenience alias. RAMP_ROUTER_BASE_URL overrides the endpoint
# (auth.py picks it up as the registry's base_url_env_var).
env_vars=("RAMP_ROUTER_API_KEY", "ROUTER_API_KEY", "RAMP_ROUTER_BASE_URL"),
base_url=_base_url(),
auth_type="api_key",
# Most of the catalog's frontier routes accept image input; capability is
# still model-dependent and governed by the live catalog.
supports_vision=True,
# Cheap, reasoning-capable, and vision-capable — safe for auxiliary tasks
# (compaction, titles, vision) when Router is the main provider. Also the
# model Router's own docs use as their example.
default_aux_model="gpt-5.4-mini",
# Deliberately empty: model IDs are account-scoped (BYOK accounts see
# extra entries) and Router's docs say to read the catalog at runtime
# rather than hardcode names. The picker uses fetch_models() above.
fallback_models=(),
)
register_provider(router)

5
plugins/model-providers/router/plugin.yaml Normal file
View File

@@ -0,0 +1,5 @@
name: router-provider
kind: model-provider
version: 1.0.0
description: Ramp Router (router.com) — OpenAI Responses-compatible LLM gateway
author: Ramp

1
providers/README.md
View File

@@ -69,6 +69,7 @@ under `$HERMES_HOME/plugins/model-providers/` for a private plugin).
| `prepare_messages(msgs)` | Provider-specific message preprocessing (Qwen normalises to list-of-parts, injects `cache_control`). |
| `build_extra_body(**ctx)` | Provider-specific `extra_body` (OpenRouter provider prefs, Gemini `thinking_config`). |
| `build_api_kwargs_extras(**ctx)` | `(extra_body_additions, top_level_kwargs)` — Kimi puts reasoning_effort top-level, Qwen splits `enable_thinking`/`thinking_budget`. |
| `supported_reasoning_efforts(model)` | Declared per-model reasoning-effort vocabulary for gateways that 400 on unknown levels (Ramp Router reads its live catalog). `None` = defer to transport defaults, `()` = model takes no reasoning params, tuple = clamp target. Must be cache-only — called on the request hot path. |
| `fetch_models(*, api_key)` | Live catalog fetch — default hits `{models_url or base_url}/models` with Bearer auth. Override for no-REST providers (Bedrock), OAuth catalogs (Anthropic), or public catalogs (OpenRouter). |
---

29
providers/base.py
View File

@@ -194,6 +194,35 @@ class ProviderProfile:
"""
return self.default_max_tokens
def supported_reasoning_efforts(
self, model: str | None
) -> tuple[str, ...] | None:
"""Declared reasoning-effort vocabulary for *model* on this provider.
Overrideable hook for providers whose gateway validates
``reasoning.effort`` per model instead of ignoring or clamping
unknown levels server-side (Ramp Router derives this from its live
``/v1/models`` catalog). The Responses transport consults it before
falling back to its built-in per-backend vocabularies; it is the
profile-declared analog of the OpenRouter catalog clamp on the
chat-completions path (``openrouter_model_reasoning_capabilities``).
Tri-state contract:
- ``None`` — unknown/undeclared: the transport keeps its default
vocabulary for the wire (this base implementation).
- ``()`` — the model accepts NO reasoning parameters at all; the
transport must omit reasoning fields entirely (some gateways
return HTTP 400 rather than ignoring them).
- non-empty tuple — clamp the requested effort onto these levels
(``agent.reasoning_effort.clamp_effort`` semantics: nearest
weaker supported level, never escalate).
Implementations are called on the per-request hot path and must not
block on network I/O — answer from a cache and return None while
cold.
"""
return None
def fetch_models(
self,
*,

156
tests/agent/transports/test_router_codex_efforts.py Normal file
View File

@@ -0,0 +1,156 @@
"""Router catalog-declared reasoning-effort clamping on the codex transport.
Ramp Router (api.router.com) validates ``reasoning.effort`` against each
model's published vocabulary — HTTP 400 ``invalid-argument`` on an
unsupported level, and 400 ``unsupported_parameter`` when a non-reasoning
model receives any reasoning field (both verified live, Aug 2026). The
router profile declares each model's vocabulary from its cached catalog via
``ProviderProfile.supported_reasoning_efforts``; these tests pin how the
codex transport consumes that declaration.
All tests seed the plugin's in-memory cache directly — no network.
"""
import sys
import pytest
from agent.transports import get_transport
def _router_plugin_module():
from providers import get_provider_profile
profile = get_provider_profile("router")
assert profile is not None, "router profile must be registered"
return profile, sys.modules[type(profile).__module__]
@pytest.fixture
def transport():
import agent.transports.codex # noqa: F401
return get_transport("codex_responses")
@pytest.fixture
def seeded_catalog(monkeypatch):
"""Seed the router efforts cache with catalog-shaped verdicts."""
profile, mod = _router_plugin_module()
monkeypatch.setattr(mod, "_efforts_cache", {
# grok via Router: no "none", no "max" (live catalog shape)
"grok-4.6": ["minimal", "low", "medium", "high", "xhigh"],
# non-reasoning model: any reasoning field 400s
"gpt-4.1-mini": [],
# full ladder including max
"accounts/fireworks/models/kimi-k3": [
"minimal", "low", "medium", "high", "xhigh", "max",
],
})
monkeypatch.setattr(mod, "_disk_checked", True)
return profile
class TestProfileContract:
def test_declared_vocabulary(self, seeded_catalog):
assert seeded_catalog.supported_reasoning_efforts("grok-4.6") == (
"minimal", "low", "medium", "high", "xhigh",
)
def test_non_reasoning_model_is_definitive_empty(self, seeded_catalog):
assert seeded_catalog.supported_reasoning_efforts("gpt-4.1-mini") == ()
def test_unknown_model_is_none(self, seeded_catalog):
assert seeded_catalog.supported_reasoning_efforts("some-byok-route") is None
def test_cold_cache_is_none_and_never_blocks(self, monkeypatch):
profile, mod = _router_plugin_module()
monkeypatch.setattr(mod, "_efforts_cache", None)
monkeypatch.setattr(mod, "_disk_checked", True)
monkeypatch.setattr(mod, "_warm_efforts_async", lambda: None)
assert profile.supported_reasoning_efforts("grok-4.6") is None
def test_parse_efforts_catalog_shapes(self):
_, mod = _router_plugin_module()
parsed = mod._parse_efforts([
{
"id": "grok-4.6",
"router": {"capabilities": {"reasoning": {
"supported": True,
"efforts": [{"value": "low"}, {"value": "high"}],
}}},
},
{
"id": "gpt-4.1",
"router": {"capabilities": {"reasoning": {"supported": False, "efforts": []}}},
},
# reasoning supported but vocabulary unpublished -> omitted (unknown)
{
"id": "mystery-model",
"router": {"capabilities": {"reasoning": {"supported": True, "efforts": []}}},
},
# no router metadata at all -> omitted
{"id": "bare-model"},
])
assert parsed == {"grok-4.6": ["low", "high"], "gpt-4.1": []}
class TestTransportClamp:
def _kwargs(self, transport, model, reasoning_config=None):
return transport.build_kwargs(
model=model,
messages=[{"role": "user", "content": "Hi"}],
tools=[],
base_url="https://api.router.com/v1",
session_id="sid",
provider="router",
reasoning_config=reasoning_config,
)
def test_clamps_to_catalog_vocabulary(self, transport, seeded_catalog):
# grok-4.6 via Router has no "max" — nearest weaker supported is xhigh.
kw = self._kwargs(transport, "grok-4.6", {"effort": "max"})
assert kw["reasoning"]["effort"] == "xhigh"
def test_supported_effort_passes_through(self, transport, seeded_catalog):
kw = self._kwargs(
transport, "accounts/fireworks/models/kimi-k3", {"effort": "max"}
)
assert kw["reasoning"]["effort"] == "max"
def test_non_reasoning_model_suppresses_reasoning(self, transport, seeded_catalog):
# Default reasoning_config is enabled — the () verdict must strip the
# reasoning field entirely (Router 400s rather than ignoring it).
kw = self._kwargs(transport, "gpt-4.1-mini")
assert "reasoning" not in kw
assert kw.get("include") == []
def test_unknown_model_falls_back_to_codex_default(self, transport, seeded_catalog):
# Not in the catalog -> default codex vocabulary applies (legacy has
# xhigh but no max: max clamps to xhigh, medium is untouched).
kw = self._kwargs(transport, "some-byok-route", {"effort": "max"})
assert kw["reasoning"]["effort"] == "xhigh"
kw = self._kwargs(transport, "some-byok-route", {"effort": "medium"})
assert kw["reasoning"]["effort"] == "medium"
def test_cold_cache_keeps_default_behavior(self, transport, monkeypatch):
_, mod = _router_plugin_module()
monkeypatch.setattr(mod, "_efforts_cache", None)
monkeypatch.setattr(mod, "_disk_checked", True)
monkeypatch.setattr(mod, "_warm_efforts_async", lambda: None)
kw = self._kwargs(transport, "grok-4.6", {"effort": "xhigh"})
# Cold cache -> no declaration -> default codex vocabulary (xhigh ok).
assert kw["reasoning"]["effort"] == "xhigh"
def test_other_providers_unaffected(self, transport, seeded_catalog):
kw = transport.build_kwargs(
model="gpt-4.1-mini",
messages=[{"role": "user", "content": "Hi"}],
tools=[],
base_url="https://generic.example.com/v1",
session_id="sid",
provider="some-other-provider",
reasoning_config={"effort": "medium"},
)
# The router catalog's () verdict for gpt-4.1-mini must not leak
# into other providers' requests.
assert kw["reasoning"]["effort"] == "medium"

127
tests/hermes_cli/test_router_provider.py Normal file
View File

@@ -0,0 +1,127 @@
"""Behavior contracts for the Ramp Router (api.router.com) provider.
Router is Responses-only: the host implements GET /v1/models and
POST /v1/responses, and POST /v1/chat/completions does not exist (404).
These tests pin the host mandate, the runtime URL detection that mirrors
it, and the profile/auth registry wiring — same contract suite shape as
tests/hermes_cli/test_meta_prompt_cache.py.
"""
import pytest
from hermes_cli.providers import determine_api_mode, host_mandated_api_mode
from hermes_cli import runtime_provider as rp
class TestHostMandatedRouterResponses:
@pytest.mark.parametrize(
"url",
[
"https://api.router.com/v1",
"https://api.router.com/v1/",
"https://api.router.com/v1/chat/completions",
"https://API.ROUTER.COM/v1",
"https://api.router.com",
"https://api.router.com:443/v1",
"https://attacker.test@api.router.com/v1",
],
)
def test_host_mandated_router_returns_codex_responses(self, url):
assert host_mandated_api_mode(url) == "codex_responses"
@pytest.mark.parametrize(
"url",
[
"https://api.router.com.attacker.test/v1",
"https://proxy.test/api.router.com/v1",
"https://api.router.com.evil/v1",
"https://router.com/v1",
"https://www.router.com/v1",
"https://app.router.com/v1",
"https://docs.router.com/v1",
"https://generic.example.com/v1",
"",
],
)
def test_host_mandated_router_rejects_spoofs(self, url):
assert host_mandated_api_mode(url) != "codex_responses"
# Generic/unrelated hosts must stay None (contract: no clobber of an
# explicitly configured api_mode on endpoints we don't recognize).
if url in (
"https://api.router.com.attacker.test/v1",
"https://proxy.test/api.router.com/v1",
"https://generic.example.com/v1",
"https://app.router.com/v1",
"https://docs.router.com/v1",
"",
):
assert host_mandated_api_mode(url) is None
def test_determine_api_mode_router_via_named_custom(self):
assert determine_api_mode("router", "https://api.router.com/v1") == "codex_responses"
assert determine_api_mode("custom", "https://api.router.com/v1") == "codex_responses"
def test_runtime_detect_router(self):
assert rp._detect_api_mode_for_url("https://api.router.com/v1") == "codex_responses"
assert rp._detect_api_mode_for_url("https://api.router.com/v1/chat/completions") == "codex_responses"
assert rp._detect_api_mode_for_url("https://API.ROUTER.COM/v1") == "codex_responses"
def test_runtime_detect_router_rejects_spoofs(self):
assert rp._detect_api_mode_for_url("https://api.router.com.attacker.test/v1") is None
assert rp._detect_api_mode_for_url("https://proxy.test/api.router.com/v1") is None
assert rp._detect_api_mode_for_url("https://router.com/v1") is None
assert rp._detect_api_mode_for_url("https://app.router.com/v1") is None
def test_fallback_api_mode_router(self):
assert rp._fallback_api_mode("router", "https://api.router.com/v1", "gpt-5.4-mini") == "codex_responses"
assert rp._fallback_api_mode("custom", "https://api.router.com/v1", "gpt-5.4-mini") == "codex_responses"
# generic endpoints stay chat_completions
assert rp._fallback_api_mode("custom", "https://generic.example.com/v1", "gpt-5.4-mini") == "chat_completions"
class TestRouterProfileRegistration:
def test_profile_registered_with_responses_mode(self):
from providers import get_provider_profile
profile = get_provider_profile("router")
assert profile is not None
assert profile.api_mode == "codex_responses"
assert profile.auth_type == "api_key"
assert profile.base_url.startswith("https://api.router.com")
def test_profile_aliases_resolve(self):
from providers import get_provider_profile
canonical = get_provider_profile("router")
for alias in ("ramp-router", "ramp", "router.com"):
assert get_provider_profile(alias) is canonical, alias
def test_documented_env_var_is_primary(self):
from providers import get_provider_profile
profile = get_provider_profile("router")
# RAMP_ROUTER_API_KEY is the variable Router's docs tell users to
# set; it must stay first so key resolution prefers it.
assert profile.env_vars[0] == "RAMP_ROUTER_API_KEY"
def test_no_hardcoded_fallback_models(self):
# Router model IDs are account-scoped (BYOK accounts see extra
# entries) and the vendor docs say to read the catalog at runtime —
# an offline fallback list would advertise IDs a key may not have.
from providers import get_provider_profile
profile = get_provider_profile("router")
assert profile.fallback_models == ()
def test_auth_registry_autowired(self):
from hermes_cli.auth import PROVIDER_REGISTRY
config = PROVIDER_REGISTRY.get("router")
assert config is not None
assert config.auth_type == "api_key"
# Key vars must not contain the base-url override var, which is
# split out into base_url_env_var by the auto-registry.
assert "RAMP_ROUTER_API_KEY" in config.api_key_env_vars
assert "RAMP_ROUTER_BASE_URL" not in config.api_key_env_vars
assert config.base_url_env_var == "RAMP_ROUTER_BASE_URL"
assert config.inference_base_url.startswith("https://api.router.com")

3
website/docs/developer-guide/adding-providers.md
View File

@@ -35,6 +35,7 @@ The important abstraction is `api_mode`.
- Most providers use `chat_completions`.
- Codex and Meta Model API (`api.meta.ai` — Muse Spark) use `codex_responses` (auto-sends `prompt_cache_retention: 24h` for prompt caching; `api.meta.ai` achieves 93–99% cache hits only on `/v1/responses`).
- Ramp Router (`api.router.com`) also uses `codex_responses` — the host is Responses-only (`/v1/chat/completions` 404s) and validates `reasoning.effort` per model, which the router profile handles by declaring each model's vocabulary from the live catalog (`ProviderProfile.supported_reasoning_efforts`).
- Anthropic uses `anthropic_messages`.
- A new non-OpenAI protocol usually means adding a new adapter and a new `api_mode` branch.
@@ -65,7 +66,7 @@ Use this when the provider does not behave like OpenAI chat completions.
Examples in-tree today:
- `codex_responses` (OpenAI Codex, xAI Grok, and Meta Muse Spark via `api.meta.ai` — the latter auto-sends `prompt_cache_retention: 24h`)
- `codex_responses` (OpenAI Codex, xAI Grok, Meta Muse Spark via `api.meta.ai` — the latter auto-sends `prompt_cache_retention: 24h` — and Ramp Router via `api.router.com`)
- `anthropic_messages`
This path includes everything from Path A plus: