Merge pull request #770 from Routstr/feat/venice-provider

feat: venice provider with anthropic web-search translation

routstr/upstream/__init__.py

@@ -13,6 +13,7 @@ from .ppqai import PPQAIUpstreamProvider
 from .routstr import RoutstrUpstreamProvider
 from .tinfoil import TinfoilUpstreamProvider
 from .typesafe import TypeSafeUpstreamProvider
+from .venice import VeniceUpstreamProvider
 from .xai import XAIUpstreamProvider
 
 upstream_provider_classes: list[type[BaseUpstreamProvider]] = [
@@ -30,6 +31,7 @@ upstream_provider_classes: list[type[BaseUpstreamProvider]] = [
     RoutstrUpstreamProvider,
     TinfoilUpstreamProvider,
     TypeSafeUpstreamProvider,
+    VeniceUpstreamProvider,
     XAIUpstreamProvider,
 ]
 """List of all upstream classes"""

routstr/upstream/base.py

@@ -2537,6 +2537,16 @@ class BaseUpstreamProvider:
     ) -> dict:
         return await messages_dispatch.aggregate_anthropic_events_to_message(iterator)
 
+    def adapt_messages_request(self, body: dict, model_obj: Model) -> str:
+        """Rewrite an allowlisted /v1/messages body for this upstream.
+
+        Returns a suffix appended to the upstream model name, empty when the
+        provider needs none. Subclasses override this to express an Anthropic
+        feature the upstream spells differently; the base forwards the body
+        untouched.
+        """
+        return ""
+
     async def _dispatch_anthropic_messages(
         self,
         request_body: bytes | None,
@@ -2551,6 +2561,7 @@ class BaseUpstreamProvider:
             api_key=self.api_key,
             provider_prefix=self.get_litellm_provider_prefix(),
             transform_model_name=self.transform_model_name,
+            adapt_request=lambda body: self.adapt_messages_request(body, model_obj),
             log_extra=log_extra,
         )
 

routstr/upstream/messages_dispatch.py

@@ -458,6 +458,7 @@ async def dispatch_anthropic_messages(
     api_key: str,
     provider_prefix: str,
     transform_model_name: Callable[[str], str],
+    adapt_request: Callable[[dict], str] | None = None,
     log_extra: dict[str, Any] | None = None,
 ) -> tuple[bool, Any, str | None]:
     """Call ``litellm.anthropic.messages.acreate`` and return
@@ -465,6 +466,11 @@ async def dispatch_anthropic_messages(
 
     Shared by the bearer-key and x-cashu paths. Raises :class:`UpstreamError`
     on bad input or upstream failure.
+
+    ``adapt_request`` is the provider's last word on the allowlisted body: it
+    may rewrite it in place and returns a suffix for the upstream model name,
+    which is how a provider expresses a feature litellm would otherwise
+    translate into a parameter the upstream rejects.
     """
     if not request_body:
         raise UpstreamError("Missing request body for /v1/messages", status_code=400)
@@ -499,13 +505,15 @@ async def dispatch_anthropic_messages(
         )
     body = {k: v for k, v in body.items() if k in ALLOWED_MESSAGES_REQUEST_FIELDS}
 
+    model_suffix = adapt_request(body) if adapt_request else ""
+
     # Convention: `model.id` is the canonical upstream model name;
     # `forwarded_model_id` is the public alias the internal API exposes
     # and echoes back to the client.
     requested_model = (
         (model_obj.forwarded_model_id or model_obj.id) if model_obj else None
     )
-    upstream_model = transform_model_name(model_obj.id)
+    upstream_model = f"{transform_model_name(model_obj.id)}{model_suffix}"
     litellm_model = f"{provider_prefix}{upstream_model}"
 
     kwargs: dict = {

routstr/upstream/venice.py

@@ -0,0 +1,312 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+import httpx
+
+from ..core.exceptions import UpstreamError
+from ..core.logging import get_logger
+from ..payment.models import Architecture, Model, Pricing, TopProvider
+from .base import BaseUpstreamProvider
+
+if TYPE_CHECKING:
+    from ..core.db import UpstreamProviderRow
+
+logger = get_logger(__name__)
+
+# ``GET /models`` defaults to ``type=text``, which is why a Venice account
+# configured as a generic upstream never sees the rest of its catalog.
+_MODELS_TYPE_PARAM = "all"
+
+# Families this proxy can both route and price. Image, audio, music and video
+# are billed per clip or per second and return no usage object to settle
+# against, so exposing them would hand out unpriced inference.
+_SUPPORTED_TYPES = frozenset({"text", "embedding"})
+
+# Venice prices text in USD per million tokens; Routstr prices per token.
+_USD_PER_MILLION = 1_000_000.0
+
+_ARCHITECTURES: dict[str, tuple[str, list[str], list[str]]] = {
+    "text": ("text->text", ["text"], ["text"]),
+    "embedding": ("text->embedding", ["text"], ["embedding"]),
+}
+
+# Venice runs search itself and reports it back through ``venice_parameters``;
+# it has no Anthropic-shaped server tool and rejects the ``web_search_options``
+# that litellm's Anthropic adapter derives from one. ``auto`` matches Anthropic
+# semantics, where declaring the tool leaves the decision to the model.
+# Citations are asked for because litellm's Anthropic response translation
+# carries no ``venice_parameters``, so the inline ``^n^`` markers Venice writes
+# into the text are the only way a caller sees that sources were used.
+_WEB_SEARCH_SUFFIX = ":enable_web_search=auto&enable_web_citations=true"
+
+# Anthropic web-search constraints with no Venice equivalent. Honouring the
+# request means enforcing them, so a request that sets one is refused rather
+# than answered by a search that ignored it. ``max_uses`` is absent on purpose:
+# ``auto`` runs at most one search per request, so any cap of 1 or more is
+# already met, while domain filters and location would be silently ignored.
+# Only ``max_uses: 0``, a request for no search at all, cannot be honoured.
+_UNENFORCEABLE_WEB_SEARCH_KEYS = frozenset(
+    {"allowed_domains", "blocked_domains", "user_location"}
+)
+
+
+def _is_web_search_tool(tool: Any) -> bool:
+    """An Anthropic server-side web-search tool, by either of its markers.
+
+    Matches litellm's own detection (``litellm/llms/anthropic/
+    experimental_pass_through/adapters/transformation.py``), so every tool it
+    would turn into ``web_search_options`` is caught here first.
+    """
+    if not isinstance(tool, dict):
+        return False
+    tool_type = tool.get("type")
+    return (
+        isinstance(tool_type, str) and tool_type.startswith("web_search")
+    ) or tool.get("name") == "web_search"
+
+
+def _usd(entry: Any) -> float | None:
+    """Read the USD leg of a Venice ``{usd, diem}`` price pair."""
+    if isinstance(entry, dict):
+        value = entry.get("usd")
+        if isinstance(value, (int, float)) and not isinstance(value, bool):
+            return float(value)
+    return None
+
+
+class VeniceUpstreamProvider(BaseUpstreamProvider):
+    """Upstream provider for the Venice.ai API.
+
+    Venice publishes a complete price book on its own catalog, so models are
+    built from that rather than matched against OpenRouter, which has never
+    heard of most of Venice's catalog.
+    """
+
+    provider_type = "venice"
+    default_base_url = "https://api.venice.ai/api/v1"
+    platform_url = "https://venice.ai/settings/api"
+
+    def __init__(self, api_key: str, provider_fee: float = 1.01):
+        super().__init__(
+            base_url=self.default_base_url, api_key=api_key, provider_fee=provider_fee
+        )
+
+    @classmethod
+    def _build_from_row(
+        cls, provider_row: "UpstreamProviderRow"
+    ) -> "VeniceUpstreamProvider":
+        return cls(
+            api_key=provider_row.api_key,
+            provider_fee=provider_row.provider_fee,
+        )
+
+    @classmethod
+    def get_provider_metadata(cls) -> dict[str, object]:
+        return {
+            "id": cls.provider_type,
+            "name": "Venice AI",
+            "default_base_url": cls.default_base_url,
+            "fixed_base_url": True,
+            "platform_url": cls.platform_url,
+        }
+
+    def transform_model_name(self, model_id: str) -> str:
+        return model_id.removeprefix("venice/")
+
+    def adapt_messages_request(self, body: dict, model_obj: Model) -> str:
+        """Trade an Anthropic web-search tool for Venice's own search switch.
+
+        Left in the body, litellm's Anthropic adapter rewrites the tool into a
+        top-level ``web_search_options``, which Venice answers with a 400. The
+        tool is lifted out here and the same intent re-expressed as a model
+        feature suffix, the one form of ``venice_parameters`` that survives
+        that adapter.
+        """
+        tools = body.get("tools")
+        if not isinstance(tools, list):
+            return ""
+        search_tools = [tool for tool in tools if _is_web_search_tool(tool)]
+        if not search_tools:
+            return ""
+
+        # A key carrying null or an empty list states no constraint, so it is
+        # read as absent rather than refused. ``auto`` runs at most one search,
+        # so only an integer ``max_uses`` of one or more is known to be met.
+        unenforceable = sorted(
+            {
+                key
+                for tool in search_tools
+                for key, value in tool.items()
+                if (
+                    key in _UNENFORCEABLE_WEB_SEARCH_KEYS
+                    and value is not None
+                    and value != []
+                )
+                or (
+                    key == "max_uses"
+                    and value is not None
+                    and not (
+                        isinstance(value, int)
+                        and not isinstance(value, bool)
+                        and value >= 1
+                    )
+                )
+            }
+        )
+        if unenforceable:
+            raise UpstreamError(
+                "Venice web search cannot honour these Anthropic web_search "
+                f"options: {', '.join(unenforceable)}",
+                status_code=400,
+                code="UNSUPPORTED_WEB_SEARCH_OPTION",
+                details={"unsupported_options": unenforceable},
+            )
+
+        tool_choice = body.get("tool_choice")
+        if isinstance(tool_choice, dict) and tool_choice.get("name") == "web_search":
+            raise UpstreamError(
+                "Venice web search cannot be forced through tool_choice; it is "
+                "decided by the model",
+                status_code=400,
+                code="UNSUPPORTED_WEB_SEARCH_OPTION",
+                details={"unsupported_options": ["tool_choice"]},
+            )
+
+        remaining = [tool for tool in tools if not _is_web_search_tool(tool)]
+        if remaining:
+            # A caller's ``tool_choice: any`` is kept and litellm maps it to
+            # OpenAI ``required``, so one of the remaining function tools must
+            # now be called where Anthropic would have let a search satisfy it.
+            # Deliberate: OpenRouter never rewrites tool_choice for web search
+            # either, and guessing an alternative would change caller intent.
+            body["tools"] = remaining
+        else:
+            body.pop("tools", None)
+            # tool_choice without tools is rejected by OpenAI-shaped upstreams.
+            body.pop("tool_choice", None)
+
+        return _WEB_SEARCH_SUFFIX
+
+    async def _fetch_provider_models(self) -> dict:
+        url = f"{self.base_url.rstrip('/')}/models"
+        headers = {"Authorization": f"Bearer {self.api_key}"} if self.api_key else None
+        async with httpx.AsyncClient(timeout=30.0) as client:
+            response = await client.get(
+                url, params={"type": _MODELS_TYPE_PARAM}, headers=headers
+            )
+            response.raise_for_status()
+            return response.json()
+
+    async def fetch_models(self) -> list[Model]:
+        try:
+            payload = await self._fetch_provider_models()
+        except Exception as e:
+            logger.error(
+                "Error fetching Venice models",
+                extra={"error": str(e), "error_type": type(e).__name__},
+            )
+            return []
+
+        models: list[Model] = []
+        skipped: list[str] = []
+        for entry in payload.get("data", []):
+            if not isinstance(entry, dict):
+                continue
+            try:
+                model = self._parse_model(entry)
+            except Exception as e:
+                logger.warning(
+                    "Failed to parse Venice model",
+                    extra={
+                        "model_id": entry.get("id", "unknown"),
+                        "error": str(e),
+                        "error_type": type(e).__name__,
+                    },
+                )
+                continue
+            if model is None:
+                skipped.append(str(entry.get("id", "unknown")))
+                continue
+            models.append(model)
+
+        if skipped:
+            logger.debug(
+                f"({len(skipped)}) Venice models skipped as unsupported or unpriced",
+                extra={"skipped_models": skipped},
+            )
+        return models
+
+    def _parse_model(self, entry: dict[str, Any]) -> Model | None:
+        model_type = entry.get("type")
+        model_id = entry.get("id")
+        spec = entry.get("model_spec")
+        if not model_id or model_type not in _SUPPORTED_TYPES:
+            return None
+        if not isinstance(spec, dict) or spec.get("offline"):
+            return None
+
+        pricing = self._parse_pricing(spec.get("pricing"), str(model_type))
+        if pricing is None:
+            return None
+
+        modality, input_modalities, output_modalities = _ARCHITECTURES[str(model_type)]
+        capabilities = spec.get("capabilities")
+        if (
+            model_type == "text"
+            and isinstance(capabilities, dict)
+            and capabilities.get("supportsVision")
+        ):
+            input_modalities = [*input_modalities, "image"]
+            modality = "text+image->text"
+
+        context_length = spec.get("availableContextTokens")
+        max_completion_tokens = spec.get("maxCompletionTokens")
+        name = spec.get("name") or str(model_id)
+
+        return Model(
+            id=str(model_id),
+            name=str(name),
+            created=int(entry.get("created") or 0),
+            description=str(spec.get("description") or f"Venice {model_type} model"),
+            context_length=int(context_length) if context_length else 0,
+            architecture=Architecture(
+                modality=modality,
+                input_modalities=input_modalities,
+                output_modalities=output_modalities,
+                tokenizer="Unknown",
+                instruct_type=None,
+            ),
+            pricing=pricing,
+            top_provider=TopProvider(
+                context_length=int(context_length) if context_length else None,
+                max_completion_tokens=int(max_completion_tokens)
+                if max_completion_tokens
+                else None,
+            ),
+        )
+
+    def _parse_pricing(self, raw: Any, model_type: str) -> Pricing | None:
+        if not isinstance(raw, dict):
+            return None
+
+        # The ``extended`` tier some models charge past a context threshold is
+        # ignored: billing it would overcharge every request staying under it.
+        input_usd = _usd(raw.get("input"))
+        output_usd = _usd(raw.get("output"))
+        # Embeddings produce no completion tokens, so only they may omit an
+        # output price. Anywhere else a missing or all-zero price would serve
+        # completions free and a negative one would credit the caller, the
+        # same guards ``generic.py`` applies to this price book.
+        if output_usd is None and model_type == "embedding":
+            output_usd = 0.0
+        if input_usd is None or output_usd is None:
+            return None
+        if input_usd < 0 or output_usd < 0 or (input_usd == 0 and output_usd == 0):
+            return None
+        return Pricing(
+            prompt=input_usd / _USD_PER_MILLION,
+            completion=output_usd / _USD_PER_MILLION,
+            input_cache_read=(_usd(raw.get("cache_input")) or 0.0) / _USD_PER_MILLION,
+            input_cache_write=(_usd(raw.get("cache_write")) or 0.0) / _USD_PER_MILLION,
+        )

tests/integration/test_venice_web_search_wire.py

@@ -0,0 +1,131 @@
+"""What Routstr actually puts on the wire for a Venice web-search request.
+
+The unit tests stop at the kwargs handed to litellm. Everything that produced
+the reported ``400 Unrecognized key(s) in object: 'web_search_options'``
+happened *after* that point, inside litellm's Anthropic adapter, so this test
+runs the whole dispatch against a loopback OpenAI-compatible server and reads
+the bytes Venice would have received.
+"""
+
+from __future__ import annotations
+
+import json
+import threading
+from http.server import BaseHTTPRequestHandler, HTTPServer
+from typing import Any, Iterator
+
+import pytest
+
+from routstr.payment.models import Architecture, Model, Pricing
+from routstr.upstream.litellm_routing import configure_litellm
+from routstr.upstream.venice import VeniceUpstreamProvider
+
+_CHUNKS = [
+    {
+        "id": "chatcmpl-1",
+        "object": "chat.completion.chunk",
+        "created": 0,
+        "model": "deepseek-v4-flash-0731",
+        "choices": [{"index": 0, "delta": {"role": "assistant", "content": "ok"}}],
+    },
+    {
+        "id": "chatcmpl-1",
+        "object": "chat.completion.chunk",
+        "created": 0,
+        "model": "deepseek-v4-flash-0731",
+        "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}],
+        "usage": {"prompt_tokens": 5, "completion_tokens": 2, "total_tokens": 7},
+    },
+]
+
+
+@pytest.fixture
+def upstream() -> Iterator[tuple[str, dict[str, Any]]]:
+    """A loopback stand-in for ``api.venice.ai`` that records one request."""
+    captured: dict[str, Any] = {}
+
+    class Handler(BaseHTTPRequestHandler):
+        def do_POST(self) -> None:  # noqa: N802 - http.server's spelling
+            length = int(self.headers.get("Content-Length", 0))
+            captured["path"] = self.path
+            captured["body"] = json.loads(self.rfile.read(length))
+
+            self.send_response(200)
+            self.send_header("Content-Type", "text/event-stream")
+            self.end_headers()
+            for chunk in _CHUNKS:
+                self.wfile.write(f"data: {json.dumps(chunk)}\n\n".encode())
+            self.wfile.write(b"data: [DONE]\n\n")
+
+        def log_message(self, *args: Any) -> None:
+            return None
+
+    server = HTTPServer(("127.0.0.1", 0), Handler)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    try:
+        yield f"http://127.0.0.1:{server.server_address[1]}/v1", captured
+    finally:
+        server.shutdown()
+        thread.join(timeout=5)
+
+
+def _model() -> Model:
+    return Model(
+        id="deepseek-v4-flash-0731",
+        name="deepseek-v4-flash-0731",
+        created=0,
+        description="",
+        context_length=8192,
+        architecture=Architecture(
+            modality="text->text",
+            input_modalities=["text"],
+            output_modalities=["text"],
+            tokenizer="Unknown",
+            instruct_type=None,
+        ),
+        pricing=Pricing(prompt=0.0, completion=0.0),
+    )
+
+
+@pytest.mark.asyncio
+async def test_web_search_request_reaches_venice_in_its_own_shape(
+    upstream: tuple[str, dict[str, Any]],
+) -> None:
+    base_url, captured = upstream
+    # The app applies this at startup; without it litellm posts the Anthropic
+    # body to /responses, which Venice serves only in alpha.
+    configure_litellm()
+
+    provider = VeniceUpstreamProvider(api_key="sk-test")
+    provider.base_url = base_url
+
+    await provider._dispatch_anthropic_messages(
+        request_body=json.dumps(
+            {
+                "model": "venice/deepseek-v4-flash-0731",
+                "messages": [{"role": "user", "content": "what shipped today?"}],
+                "max_tokens": 64,
+                "stream": True,
+                "tools": [
+                    {"type": "web_search_20250305", "name": "web_search"},
+                    {
+                        "name": "lookup",
+                        "description": "Look something up",
+                        "input_schema": {"type": "object", "properties": {}},
+                    },
+                ],
+            }
+        ).encode(),
+        model_obj=_model(),
+    )
+
+    body = captured["body"]
+    assert captured["path"] == "/v1/chat/completions"
+    # The reported 400, at the only place it could be observed.
+    assert "web_search_options" not in body
+    assert body["model"] == (
+        "deepseek-v4-flash-0731:enable_web_search=auto&enable_web_citations=true"
+    )
+    # The function tool still travels, in OpenAI's shape.
+    assert [tool["function"]["name"] for tool in body["tools"]] == ["lookup"]

tests/unit/test_upstream_venice.py

@@ -0,0 +1,293 @@
+"""Unit tests for ``VeniceUpstreamProvider.fetch_models``.
+
+Venice answers ``/models`` with only its text catalog unless ``type`` is
+passed, which is why the same account configured as a generic upstream sees a
+different catalog. These tests pin that query parameter, the per-token pricing
+shape, and the families dropped as unpriceable.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import patch
+
+import pytest
+
+from routstr.upstream.venice import VeniceUpstreamProvider
+
+
+class _FakeResponse:
+    def __init__(self, payload: dict[str, Any]) -> None:
+        self._payload = payload
+
+    def raise_for_status(self) -> None:
+        return None
+
+    def json(self) -> dict[str, Any]:
+        return self._payload
+
+
+class _FakeAsyncClient:
+    def __init__(self, payload: dict[str, Any], calls: list[dict[str, Any]]) -> None:
+        self._payload = payload
+        self._calls = calls
+
+    async def __aenter__(self) -> "_FakeAsyncClient":
+        return self
+
+    async def __aexit__(self, *_: object) -> None:
+        return None
+
+    async def get(
+        self,
+        url: str,
+        params: dict[str, Any] | None = None,
+        headers: dict[str, str] | None = None,
+    ) -> _FakeResponse:
+        self._calls.append({"url": url, "params": params, "headers": headers})
+        return _FakeResponse(self._payload)
+
+
+CATALOG: dict[str, Any] = {
+    "object": "list",
+    "data": [
+        {
+            "id": "venice-uncensored-1-2",
+            "type": "text",
+            "created": 1727966436,
+            "model_spec": {
+                "name": "Venice Uncensored 1.2",
+                "availableContextTokens": 128000,
+                "maxCompletionTokens": 8192,
+                "capabilities": {"supportsVision": True},
+                "pricing": {
+                    "input": {"usd": 0.2, "diem": 0.2},
+                    "output": {"usd": 0.9, "diem": 0.9},
+                    "cache_input": {"usd": 0.02, "diem": 0.02},
+                    "cache_write": {"usd": 0.25, "diem": 0.25},
+                },
+            },
+        },
+        {
+            "id": "text-embedding-bge-m3",
+            "type": "embedding",
+            "created": 1727966436,
+            "model_spec": {
+                "name": "BGE m3",
+                "availableContextTokens": 8192,
+                "pricing": {"input": {"usd": 0.01, "diem": 0.01}},
+            },
+        },
+        {
+            "id": "unpriced-text",
+            "type": "text",
+            "created": 1727966436,
+            "model_spec": {"name": "Unpriced", "pricing": {}},
+        },
+        {
+            "id": "offline-model",
+            "type": "text",
+            "created": 1727966436,
+            "model_spec": {
+                "name": "Offline",
+                "offline": True,
+                "pricing": {"input": {"usd": 0.2, "diem": 0.2}},
+            },
+        },
+        {
+            "id": "venice-sd35",
+            "type": "image",
+            "created": 1727966436,
+            "model_spec": {
+                "name": "Venice SD35",
+                "pricing": {"generation": {"usd": 0.01, "diem": 0.01}},
+            },
+        },
+        {
+            "id": "flux-2-max-edit",
+            "type": "inpaint",
+            "created": 1727966436,
+            "model_spec": {
+                "name": "FLUX.2 Max Edit",
+                "pricing": {"inpaint": {"usd": 0.12, "diem": 0.12}},
+            },
+        },
+        {
+            "id": "tts-kokoro",
+            "type": "tts",
+            "created": 1727966436,
+            "model_spec": {
+                "name": "Kokoro",
+                "pricing": {"input": {"usd": 3.5, "diem": 3.5}},
+            },
+        },
+        {
+            "id": "unpriced-video",
+            "type": "video",
+            "created": 1727966436,
+            "model_spec": {"name": "Video"},
+        },
+    ],
+}
+
+
+def _fetch(payload: dict[str, Any] = CATALOG) -> tuple[list[Any], list[dict[str, Any]]]:
+    import asyncio
+
+    calls: list[dict[str, Any]] = []
+    provider = VeniceUpstreamProvider(api_key="sk-test")
+    with patch(
+        "routstr.upstream.venice.httpx.AsyncClient",
+        lambda *a, **kw: _FakeAsyncClient(payload, calls),
+    ):
+        models = asyncio.run(provider.fetch_models())
+    return models, calls
+
+
+def test_requests_every_model_family() -> None:
+    _, calls = _fetch()
+    assert calls[0]["params"] == {"type": "all"}
+    assert calls[0]["url"] == "https://api.venice.ai/api/v1/models"
+    assert calls[0]["headers"] == {"Authorization": "Bearer sk-test"}
+
+
+def test_text_pricing_is_per_token() -> None:
+    models, _ = _fetch()
+    model = next(m for m in models if m.id == "venice-uncensored-1-2")
+    assert model.pricing.prompt == pytest.approx(0.2 / 1_000_000)
+    assert model.pricing.completion == pytest.approx(0.9 / 1_000_000)
+    assert model.pricing.input_cache_read == pytest.approx(0.02 / 1_000_000)
+    assert model.pricing.input_cache_write == pytest.approx(0.25 / 1_000_000)
+    assert model.context_length == 128000
+    assert model.top_provider is not None
+    assert model.top_provider.max_completion_tokens == 8192
+    assert model.architecture.input_modalities == ["text", "image"]
+    assert model.architecture.modality == "text+image->text"
+
+
+def test_embedding_models_are_listed() -> None:
+    models, _ = _fetch()
+    model = next(m for m in models if m.id == "text-embedding-bge-m3")
+    assert model.architecture.output_modalities == ["embedding"]
+    assert model.pricing.prompt == pytest.approx(0.01 / 1_000_000)
+    assert model.pricing.completion == 0.0
+
+
+def test_families_billed_per_clip_are_dropped() -> None:
+    """Image, audio and video return no usage to settle against, so listing
+    them here would hand out inference this provider cannot price."""
+    models, _ = _fetch()
+    ids = {m.id for m in models}
+    assert "venice-sd35" not in ids
+    assert "flux-2-max-edit" not in ids
+    assert "tts-kokoro" not in ids
+    assert "unpriced-video" not in ids
+
+
+def test_offline_and_unpriced_models_are_dropped() -> None:
+    models, _ = _fetch()
+    ids = {m.id for m in models}
+    assert "offline-model" not in ids
+    assert "unpriced-text" not in ids
+
+
+def _priced_entry(model_id: str, model_type: str, pricing: dict[str, Any]) -> dict:
+    return {
+        "id": model_id,
+        "type": model_type,
+        "created": 1727966436,
+        "model_spec": {"name": model_id, "pricing": pricing},
+    }
+
+
+@pytest.mark.parametrize(
+    "pricing",
+    [
+        pytest.param({"input": {"usd": 0.2, "diem": 0.2}}, id="missing-output"),
+        pytest.param(
+            {"input": {"usd": 0.0, "diem": 0.0}, "output": {"usd": 0.0, "diem": 0.0}},
+            id="both-zero",
+        ),
+        pytest.param(
+            {"input": {"usd": -0.2, "diem": 0.2}, "output": {"usd": 0.9, "diem": 0.9}},
+            id="negative-input",
+        ),
+        pytest.param(
+            {"input": {"usd": 0.2, "diem": 0.2}, "output": {"usd": -0.9, "diem": 0.9}},
+            id="negative-output",
+        ),
+    ],
+)
+def test_text_models_that_would_bill_free_or_negative_are_dropped(
+    pricing: dict[str, Any],
+) -> None:
+    models, _ = _fetch(
+        {"object": "list", "data": [_priced_entry("bad-text", "text", pricing)]}
+    )
+    assert models == []
+
+
+def test_embedding_with_only_an_input_price_is_listed() -> None:
+    models, _ = _fetch(
+        {
+            "object": "list",
+            "data": [
+                _priced_entry("emb", "embedding", {"input": {"usd": 0.05, "diem": 0}})
+            ],
+        }
+    )
+    assert [m.id for m in models] == ["emb"]
+    assert models[0].pricing.prompt == pytest.approx(0.05 / 1_000_000)
+    assert models[0].pricing.completion == 0.0
+
+
+def test_embedding_with_a_negative_price_is_dropped() -> None:
+    models, _ = _fetch(
+        {
+            "object": "list",
+            "data": [
+                _priced_entry("emb", "embedding", {"input": {"usd": -0.05, "diem": 0}})
+            ],
+        }
+    )
+    assert models == []
+
+
+def test_text_model_with_one_zero_price_is_listed() -> None:
+    """Only both-zero is free; a free prompt with a paid completion is priced."""
+    pricing = {"input": {"usd": 0.0, "diem": 0}, "output": {"usd": 0.9, "diem": 0}}
+    models, _ = _fetch(
+        {"object": "list", "data": [_priced_entry("t", "text", pricing)]}
+    )
+    assert [m.id for m in models] == ["t"]
+    assert models[0].pricing.completion == pytest.approx(0.9 / 1_000_000)
+
+
+def test_model_name_drops_the_venice_prefix() -> None:
+    provider = VeniceUpstreamProvider(api_key="sk-test")
+    assert provider.transform_model_name("venice/venice-uncensored-1-2") == (
+        "venice-uncensored-1-2"
+    )
+    assert provider.transform_model_name("venice-uncensored-1-2") == (
+        "venice-uncensored-1-2"
+    )
+
+
+def test_provider_metadata_pins_the_base_url() -> None:
+    metadata = VeniceUpstreamProvider.get_provider_metadata()
+    assert metadata["id"] == "venice"
+    assert metadata["default_base_url"] == "https://api.venice.ai/api/v1"
+    assert metadata["fixed_base_url"] is True
+
+
+def test_fetch_returns_empty_on_upstream_failure() -> None:
+    provider = VeniceUpstreamProvider(api_key="sk-test")
+
+    with patch.object(
+        VeniceUpstreamProvider,
+        "_fetch_provider_models",
+        side_effect=RuntimeError("boom"),
+    ):
+        import asyncio
+
+        assert asyncio.run(provider.fetch_models()) == []

tests/unit/test_venice_web_search.py

@@ -0,0 +1,281 @@
+"""Venice web search over ``/v1/messages``.
+
+litellm's Anthropic adapter rewrites an Anthropic server-side web-search tool
+into a top-level ``web_search_options``, which Venice rejects with
+``400 Unrecognized key(s) in object: 'web_search_options'``. These tests pin
+the trade: the tool is lifted out of the body and the same intent re-expressed
+as a Venice model feature suffix.
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Any, AsyncIterator
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from routstr.core.exceptions import UpstreamError
+from routstr.payment.models import Architecture, Model, Pricing
+from routstr.upstream.base import BaseUpstreamProvider
+from routstr.upstream.venice import VeniceUpstreamProvider
+
+WEB_SEARCH_TOOL = {"type": "web_search_20250305", "name": "web_search"}
+FUNCTION_TOOL = {
+    "name": "lookup",
+    "description": "Look something up",
+    "input_schema": {"type": "object", "properties": {}},
+}
+
+
+def _model(model_id: str = "deepseek-v4-flash-0731") -> Model:
+    return Model(
+        id=model_id,
+        name=model_id,
+        created=0,
+        description="",
+        context_length=8192,
+        architecture=Architecture(
+            modality="text->text",
+            input_modalities=["text"],
+            output_modalities=["text"],
+            tokenizer="Unknown",
+            instruct_type=None,
+        ),
+        pricing=Pricing(prompt=0.0, completion=0.0),
+    )
+
+
+def _body(**extra: Any) -> dict[str, Any]:
+    return {
+        "messages": [{"role": "user", "content": "what shipped today?"}],
+        "max_tokens": 64,
+        **extra,
+    }
+
+
+async def _dispatch(provider: BaseUpstreamProvider, body: dict[str, Any]) -> dict:
+    """Run the real dispatcher, capturing the kwargs litellm would receive."""
+    captured: dict[str, Any] = {}
+
+    async def empty_iter() -> AsyncIterator[dict]:
+        if False:
+            yield {}
+
+    async def fake_acreate(**kwargs: Any) -> AsyncIterator[dict]:
+        captured.update(kwargs)
+        return empty_iter()
+
+    with patch(
+        "litellm.anthropic.messages.acreate",
+        new=AsyncMock(side_effect=fake_acreate),
+    ):
+        await provider._dispatch_anthropic_messages(
+            request_body=json.dumps(
+                {"model": "venice/x", "stream": True, **body}
+            ).encode(),
+            model_obj=_model(),
+        )
+    return captured
+
+
+@pytest.mark.asyncio
+async def test_web_search_tool_never_reaches_venice_as_web_search_options() -> None:
+    """The reported 400: the derived parameter must not be sent at all."""
+    provider = VeniceUpstreamProvider(api_key="sk-test")
+
+    kwargs = await _dispatch(provider, _body(tools=[WEB_SEARCH_TOOL]))
+
+    assert "web_search_options" not in kwargs
+    assert "tools" not in kwargs
+    assert kwargs["model"] == (
+        "openai/deepseek-v4-flash-0731:enable_web_search=auto&enable_web_citations=true"
+    )
+    assert kwargs["api_base"] == "https://api.venice.ai/api/v1"
+
+
+@pytest.mark.asyncio
+async def test_function_tools_survive_alongside_web_search() -> None:
+    provider = VeniceUpstreamProvider(api_key="sk-test")
+
+    kwargs = await _dispatch(
+        provider,
+        _body(
+            tools=[WEB_SEARCH_TOOL, FUNCTION_TOOL],
+            tool_choice={"type": "tool", "name": "lookup"},
+        ),
+    )
+
+    assert kwargs["tools"] == [FUNCTION_TOOL]
+    assert kwargs["tool_choice"] == {"type": "tool", "name": "lookup"}
+    assert "web_search_options" not in kwargs
+    assert kwargs["model"].endswith(":enable_web_search=auto&enable_web_citations=true")
+
+
+@pytest.mark.asyncio
+async def test_requests_without_web_search_are_untouched() -> None:
+    provider = VeniceUpstreamProvider(api_key="sk-test")
+
+    kwargs = await _dispatch(provider, _body(tools=[FUNCTION_TOOL]))
+
+    assert kwargs["model"] == "openai/deepseek-v4-flash-0731"
+    assert kwargs["tools"] == [FUNCTION_TOOL]
+
+
+@pytest.mark.asyncio
+async def test_other_providers_keep_their_existing_behaviour() -> None:
+    """The base hook is a no-op, so no non-Venice upstream changes shape."""
+    provider = BaseUpstreamProvider(base_url="http://test", api_key="k")
+
+    kwargs = await _dispatch(provider, _body(tools=[WEB_SEARCH_TOOL]))
+
+    assert kwargs["model"] == "openai/deepseek-v4-flash-0731"
+    assert kwargs["tools"] == [WEB_SEARCH_TOOL]
+
+
+@pytest.mark.parametrize(
+    "tool",
+    [
+        {
+            "type": "web_search_20250305",
+            "name": "web_search",
+            "allowed_domains": ["example.com"],
+        },
+        {"type": "web_search_20250305", "name": "web_search", "blocked_domains": ["x"]},
+        {
+            "type": "web_search_20250305",
+            "name": "web_search",
+            "user_location": {"type": "approximate", "country": "DE"},
+        },
+    ],
+)
+def test_constraints_venice_cannot_enforce_are_refused(tool: dict[str, Any]) -> None:
+    """Better an explicit 400 than a search that quietly ignored the limit."""
+    provider = VeniceUpstreamProvider(api_key="sk-test")
+
+    with pytest.raises(UpstreamError) as excinfo:
+        provider.adapt_messages_request(_body(tools=[tool]), _model())
+
+    assert excinfo.value.status_code == 400
+    assert excinfo.value.code == "UNSUPPORTED_WEB_SEARCH_OPTION"
+
+
+@pytest.mark.parametrize(
+    "tool",
+    [
+        {"type": "web_search_20250305", "name": "web_search", "max_uses": None},
+        {"type": "web_search_20250305", "name": "web_search", "allowed_domains": []},
+    ],
+)
+def test_constraint_keys_stating_nothing_are_read_as_absent(
+    tool: dict[str, Any],
+) -> None:
+    provider = VeniceUpstreamProvider(api_key="sk-test")
+
+    assert provider.adapt_messages_request(_body(tools=[tool]), _model()) != ""
+
+
+def test_forcing_web_search_through_tool_choice_is_refused() -> None:
+    provider = VeniceUpstreamProvider(api_key="sk-test")
+    body = _body(
+        tools=[WEB_SEARCH_TOOL],
+        tool_choice={"type": "tool", "name": "web_search"},
+    )
+
+    with pytest.raises(UpstreamError) as excinfo:
+        provider.adapt_messages_request(body, _model())
+
+    assert excinfo.value.status_code == 400
+    assert excinfo.value.code == "UNSUPPORTED_WEB_SEARCH_OPTION"
+    assert excinfo.value.details == {"unsupported_options": ["tool_choice"]}
+
+
+def test_web_search_only_request_drops_tool_choice() -> None:
+    """Without tools left, a surviving tool_choice is rejected upstream."""
+    provider = VeniceUpstreamProvider(api_key="sk-test")
+    body = _body(tools=[WEB_SEARCH_TOOL], tool_choice={"type": "auto"})
+
+    provider.adapt_messages_request(body, _model())
+
+    assert "tools" not in body
+    assert "tool_choice" not in body
+
+
+@pytest.mark.asyncio
+async def test_claude_code_web_search_tool_is_accepted() -> None:
+    """Claude Code always sends ``max_uses: 8``; Venice's single ``auto``
+    search already stays under any cap of one or more."""
+    provider = VeniceUpstreamProvider(api_key="sk-test")
+    tool = {
+        "type": "web_search_20250305",
+        "name": "web_search",
+        "allowed_domains": None,
+        "blocked_domains": None,
+        "max_uses": 8,
+    }
+
+    kwargs = await _dispatch(provider, _body(tools=[tool]))
+
+    assert "web_search_options" not in kwargs
+    assert "tools" not in kwargs
+    assert kwargs["model"] == (
+        "openai/deepseek-v4-flash-0731:enable_web_search=auto&enable_web_citations=true"
+    )
+
+
+@pytest.mark.parametrize("max_uses", [1, None])
+def test_max_uses_of_one_or_absent_is_accepted(max_uses: Any) -> None:
+    provider = VeniceUpstreamProvider(api_key="sk-test")
+    tool = {"type": "web_search_20250305", "name": "web_search", "max_uses": max_uses}
+
+    assert provider.adapt_messages_request(_body(tools=[tool]), _model()) != ""
+
+
+@pytest.mark.parametrize("max_uses", [0, -1, 1.5, True, "0", "8"])
+def test_max_uses_other_than_a_positive_integer_is_refused(max_uses: Any) -> None:
+    """``auto`` may still search, so a cap below one cannot be met, and a
+    malformed cap cannot be shown to be met."""
+    provider = VeniceUpstreamProvider(api_key="sk-test")
+    tool = {"type": "web_search_20250305", "name": "web_search", "max_uses": max_uses}
+
+    with pytest.raises(UpstreamError) as excinfo:
+        provider.adapt_messages_request(_body(tools=[tool]), _model())
+
+    assert excinfo.value.status_code == 400
+    assert excinfo.value.code == "UNSUPPORTED_WEB_SEARCH_OPTION"
+    assert excinfo.value.details == {"unsupported_options": ["max_uses"]}
+
+
+def test_tool_named_web_search_without_the_type_marker_is_caught() -> None:
+    """litellm matches on either marker, so this one would also be rewritten."""
+    provider = VeniceUpstreamProvider(api_key="sk-test")
+    body = _body(tools=[{"name": "web_search"}])
+
+    assert provider.adapt_messages_request(body, _model()) != ""
+    assert "tools" not in body
+
+
+def test_litellm_adapter_derives_no_web_search_options_from_the_adapted_body() -> None:
+    """The fix at its cause: run the real litellm translation over the body
+    this provider produces and assert the rejected key is never derived."""
+    from litellm.llms.anthropic.experimental_pass_through.adapters.transformation import (  # noqa: E501
+        LiteLLMAnthropicMessagesAdapter,
+    )
+
+    provider = VeniceUpstreamProvider(api_key="sk-test")
+    adapter = LiteLLMAnthropicMessagesAdapter()  # type: ignore[no-untyped-call]
+    body = _body(tools=[WEB_SEARCH_TOOL, FUNCTION_TOOL])
+
+    def translate(request: dict[str, Any]) -> dict:
+        # litellm types the request as a TypedDict; these bodies are built
+        # from client JSON, so they are plain dicts at this seam.
+        translated, _ = adapter.translate_anthropic_to_openai(request)  # type: ignore[arg-type]
+        return dict(translated)
+
+    # Unadapted, litellm derives the parameter Venice rejects.
+    before = translate({"model": "m", **_body(tools=[WEB_SEARCH_TOOL])})
+    assert "web_search_options" in before
+
+    provider.adapt_messages_request(body, _model())
+
+    assert "web_search_options" not in translate({"model": "m", **body})