- Router(*, jev_api_key=None, openrouter_api_key=None, providers=None, models=None, limits=Limits(), models_per_provider=None) | Keyword-only. Loads live model data (cached for 24h) and works out the candidate list. Raises UnknownModelError for unknown model ids or providers. |
+ Router(*, jev_api_key=None, openrouter_api_key=None, providers=None, models=None, limits=Limits(), models_per_provider=None, timeout=60) | Keyword-only. Loads live model data (cached for 24h) and works out the candidate list. timeout is the HTTP timeout in seconds for each request. Raises UnknownModelError for unknown model ids or providers. |
router.route(task, limits=None) → str | The best model id for task. |
+ await router.aroute(task, limits=None) → str | Async route() for asyncio apps; doesn't block the event loop. |
router.fitting(task, limits=None) → list[ModelInfo] | The models that pass the limits, without calling Jev (free). |
router.api_key_for(model_id) → str | None | The key you passed in providers={...} for that model's provider. |
Limits(output_tokens=1024, max_cost_usd=None) | The output size you expect and an optional cost cap for each call. |
diff --git a/pyproject.toml b/pyproject.toml
index 3cbff2a..275a50b 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -1,6 +1,6 @@
[project]
name = "model-router-python"
-version = "0.1.0"
+version = "0.2.0"
description = "Pick the right LLM for every task: filter by context, output and cost limits, then let Jev choose the most efficient model."
readme = "README.md"
requires-python = ">=3.12"
diff --git a/src/model_router/_http.py b/src/model_router/_http.py
index 38bb318..344b43b 100644
--- a/src/model_router/_http.py
+++ b/src/model_router/_http.py
@@ -14,7 +14,7 @@ def request_json(url, api_key=None, body=None, timeout=60):
req = urllib.request.Request(url, data=data, headers=headers)
try:
with urllib.request.urlopen(req, timeout=timeout) as r:
- return json.load(r)
+ res = json.load(r)
except urllib.error.HTTPError as e:
raise RouterError(f"{url} -> HTTP {e.code}: {e.read().decode(errors='replace')[:500]}") from e
except urllib.error.URLError as e:
@@ -25,3 +25,6 @@ def request_json(url, api_key=None, body=None, timeout=60):
raise RouterError(f"{url} -> {e}") from e
except (json.JSONDecodeError, UnicodeDecodeError) as e:
raise RouterError(f"{url} -> response was not valid JSON: {e}") from e
+ if not isinstance(res, dict):
+ raise RouterError(f"{url} -> expected a JSON object, got {type(res).__name__}")
+ return res
diff --git a/src/model_router/catalog.py b/src/model_router/catalog.py
index 9279497..f615d71 100644
--- a/src/model_router/catalog.py
+++ b/src/model_router/catalog.py
@@ -1,6 +1,7 @@
import time
from ._http import request_json
+from .errors import RouterError
from .models import ModelInfo
MODELS_URL = "https://openrouter.ai/api/v1/models"
@@ -17,9 +18,12 @@ def fetch_catalog(max_age=CATALOG_TTL_SECONDS, *, timeout=60):
"""
# ponytail: in-process cache only; persist to disk if many short-lived processes each pay the fetch
if _cache["data"] is None or time.time() - _cache["at"] > max_age:
+ data = request_json(MODELS_URL, timeout=timeout).get("data")
+ if not isinstance(data, list):
+ raise RouterError(f"{MODELS_URL} -> response has no 'data' list")
_cache["data"] = {
m["id"]: ModelInfo.from_openrouter(m)
- for m in request_json(MODELS_URL, timeout=timeout)["data"]
+ for m in data
if "text" in ((m.get("architecture") or {}).get("output_modalities") or ["text"])
}
_cache["at"] = time.time()
diff --git a/src/model_router/router.py b/src/model_router/router.py
index 61ebdc7..ef4c90e 100644
--- a/src/model_router/router.py
+++ b/src/model_router/router.py
@@ -33,6 +33,12 @@ def __init__(
uses every current model from those providers (or the newest `models_per_provider`)
models exact OpenRouter-style ids, e.g. ["anthropic/claude-opus-5.5"]; overrides the auto-pick
"""
+ if (
+ isinstance(timeout, bool)
+ or not isinstance(timeout, (int, float))
+ or not 0 < timeout < float("inf")
+ ):
+ raise ValueError("timeout must be a finite number of seconds > 0")
if jev_api_key:
self._choose = partial(jev.choose_via_jev, jev_api_key, timeout=timeout)
elif openrouter_api_key:
diff --git a/tests/test_catalog.py b/tests/test_catalog.py
index e8b376f..85e634b 100644
--- a/tests/test_catalog.py
+++ b/tests/test_catalog.py
@@ -1,7 +1,7 @@
import unittest
from unittest.mock import patch
-from model_router import ModelInfo, catalog
+from model_router import ModelInfo, RouterError, catalog
def raw(id, output=("text",), **extra):
@@ -49,6 +49,12 @@ def test_custom_max_age_zero_always_refetches(self, req):
catalog.fetch_catalog(max_age=0)
self.assertEqual(req.call_count, 2)
+ def test_response_without_data_list_is_router_error(self):
+ for res in ({}, {"error": {"message": "down"}}, {"data": None}):
+ with self.subTest(res=res), patch("model_router.catalog.request_json", return_value=res):
+ with self.assertRaises(RouterError):
+ catalog.fetch_catalog()
+
def info(id, created=0, expires=None):
return ModelInfo(id, 1000, None, 0.0, 0.0, "", created=created, expires=expires)
diff --git a/tests/test_http.py b/tests/test_http.py
index 9e067b7..aa5632f 100644
--- a/tests/test_http.py
+++ b/tests/test_http.py
@@ -76,6 +76,15 @@ def test_connection_reset_becomes_router_error(self):
request_json("https://x.test/a")
self.assertIsInstance(ctx.exception.__cause__, ConnectionResetError)
+ def test_json_that_is_not_an_object_becomes_router_error(self):
+ for body in (b"null", b"[1, 2]", b'"ok"'):
+ with self.subTest(body=body):
+ resp = MagicMock()
+ resp.__enter__.return_value = io.BytesIO(body)
+ with patch("urllib.request.urlopen", return_value=resp):
+ with self.assertRaises(RouterError):
+ request_json("https://x.test/a")
+
if __name__ == "__main__":
unittest.main()
diff --git a/tests/test_router.py b/tests/test_router.py
index 0e4a879..1da4b68 100644
--- a/tests/test_router.py
+++ b/tests/test_router.py
@@ -40,6 +40,11 @@ def make_router(models=("cheap/small", "big/smart"), limits=None, **kw):
class ConfigTest(unittest.TestCase):
+ def test_invalid_timeout_is_rejected(self):
+ for bad in (0, -1, None, "5", True, float("nan"), float("inf")):
+ with self.subTest(timeout=bad), self.assertRaises(ValueError):
+ make_router(timeout=bad)
+
def test_timeout_reaches_both_routing_backends(self):
for backend, payload in [
("jev_api_key", {"code": 0, "data": {"decision": "cheap/small"}}),
diff --git a/uv.lock b/uv.lock
index 317f121..3b2bac2 100644
--- a/uv.lock
+++ b/uv.lock
@@ -22,7 +22,7 @@ wheels = [
[[package]]
name = "model-router-python"
-version = "0.1.0"
+version = "0.2.0"
source = { editable = "." }
[package.dev-dependencies]