Getting Started with OpenRouter on macOS

mkdir -p openrouter-quickstart
cd openrouter-quickstart
touch .gitignore

uv init --name orkit --no-workspace
rm -f main.py hello.py
uv add openai python-dotenv
uv add --dev pytest

[project]
name = "orkit"
version = "0.1.0"
description = "A small OpenRouter toolkit: catalogue, costed calls, and streaming"
readme = "README.md"
requires-python = ">=3.12"
dependencies = [
    "openai>=3.3.1",
    "python-dotenv>=1.2.3",
]

[dependency-groups]
dev = [
    "pytest>=9.1.1",
]

[tool.pytest.ini_options]
pythonpath = ["."]
testpaths = ["tests"]

touch .env.example config.py

# Copy to .env (`make setup` does it) and fill in your key. .env is gitignored.
# Create a key at https://openrouter.ai/keys

OPENROUTER_API_KEY=sk-or-v1-replace-me

# Default model. Free ids end in `:free` and change over time, and a busy free
# model returns 429 from its upstream provider even when your own quota is
# untouched — run `make models` to see what is available right now.
OPENROUTER_MODEL=nvidia/nemotron-3-super-120b-a12b:free

# Tried in order when the model above fails. Free models are heavily shared, so
# a 429 from the upstream provider is routine even on your first call of the
# day; a chain turns that into a slower answer instead of an error. Set this to
# an empty value to disable fallbacks entirely.
OPENROUTER_FALLBACKS=liquid/lfm-2.5-2.6b:free,google/gemma-4-31b-it:free

# Attribution headers. These are optional; they identify your app on
# openrouter.ai's public rankings. Use your real site once you have one.
OPENROUTER_REFERER=https://example.com
OPENROUTER_TITLE=openrouter-quickstart

# Seconds to wait for a response before giving up.
OPENROUTER_TIMEOUT=60

"""Settings for the OpenRouter client, loaded from the environment.

Values come from real environment variables first, then from a local `.env`
file (which is gitignored). Nothing is hard-coded, and a missing key fails
loudly at startup instead of surfacing as a 401 three layers down.
"""

import os
from dataclasses import dataclass

from dotenv import load_dotenv

BASE_URL = "https://openrouter.ai/api/v1"

# Free model ids rotate, and a busy one 429s from its upstream provider even
# when your own quota is untouched. `make models` lists what is live today.
DEFAULT_MODEL = "nvidia/nemotron-3-super-120b-a12b:free"
DEFAULT_FALLBACKS = ("liquid/lfm-2.5-2.6b:free", "google/gemma-4-31b-it:free")


class ConfigError(RuntimeError):
    """Raised when a required setting is absent."""


@dataclass(frozen=True)
class Settings:
    api_key: str
    model: str
    fallbacks: tuple[str, ...]
    referer: str
    title: str
    timeout: float


def _split(raw: str | None, default: tuple[str, ...]) -> tuple[str, ...]:
    """Parse a comma-separated model list, falling back to `default`.

    An explicitly empty value means "no fallbacks", which is different from an
    unset variable meaning "use the defaults".
    """
    if raw is None:
        return default
    return tuple(part.strip() for part in raw.split(",") if part.strip())


def load_settings(env: dict[str, str] | None = None) -> Settings:
    """Build Settings from `env`, defaulting to the process environment.

    Passing `env` explicitly is what makes this testable: the tests hand in a
    plain dict and never touch the developer's real key.
    """
    if env is None:
        load_dotenv()  # no-op when .env is absent; never overrides a real env var
        env = dict(os.environ)

    api_key = env.get("OPENROUTER_API_KEY", "").strip()
    if not api_key:
        raise ConfigError(
            "OPENROUTER_API_KEY is not set. Copy .env.example to .env and put "
            "your key there (get one at https://openrouter.ai/keys)."
        )

    return Settings(
        api_key=api_key,
        model=env.get("OPENROUTER_MODEL", DEFAULT_MODEL).strip(),
        fallbacks=_split(env.get("OPENROUTER_FALLBACKS"), DEFAULT_FALLBACKS),
        referer=env.get("OPENROUTER_REFERER", "https://example.com").strip(),
        title=env.get("OPENROUTER_TITLE", "openrouter-quickstart").strip(),
        timeout=float(env.get("OPENROUTER_TIMEOUT", "60")),
    )

touch catalog.py

"""Query the OpenRouter model catalogue.

`GET /api/v1/models` is public: it needs no API key, which makes it the
cheapest way to answer "what can I actually call today, and what does it
cost?". Prices come back as USD *per token* in string form, so they are
converted to dollars per million tokens for display.

Run it directly:
    uv run python catalog.py            # free models, widest context first
    uv run python catalog.py --all      # every model
    uv run python catalog.py --tools    # only models that support tool calling
"""

import argparse
from dataclasses import dataclass

import httpx2

from config import BASE_URL


@dataclass(frozen=True)
class Model:
    id: str
    name: str
    context_length: int
    prompt_usd_per_mtok: float
    completion_usd_per_mtok: float
    supported_parameters: tuple[str, ...]

    @property
    def is_free(self) -> bool:
        return self.id.endswith(":free")

    @property
    def supports_tools(self) -> bool:
        return "tools" in self.supported_parameters


def parse_models(payload: dict) -> list[Model]:
    """Turn the raw /models JSON into Model records.

    Kept separate from the HTTP call so the tests can feed it a fixture.
    """
    models: list[Model] = []
    for item in payload.get("data", []):
        pricing = item.get("pricing") or {}
        models.append(
            Model(
                id=item["id"],
                name=item.get("name", item["id"]),
                context_length=item.get("context_length") or 0,
                prompt_usd_per_mtok=float(pricing.get("prompt") or 0.0) * 1_000_000,
                completion_usd_per_mtok=float(pricing.get("completion") or 0.0)
                * 1_000_000,
                supported_parameters=tuple(item.get("supported_parameters") or ()),
            )
        )
    return models


def fetch_models(client: httpx2.Client | None = None) -> list[Model]:
    """Fetch the live catalogue. No Authorization header is required."""
    owns_client = client is None
    client = client or httpx2.Client(timeout=30.0)
    try:
        response = client.get(f"{BASE_URL}/models")
        response.raise_for_status()
        return parse_models(response.json())
    finally:
        if owns_client:
            client.close()


def format_table(models: list[Model]) -> str:
    rows = [f"{'MODEL':<44} {'CONTEXT':>9}  {'$/MTOK IN':>10} {'$/MTOK OUT':>10}  TOOLS"]
    for m in models:
        rows.append(
            f"{m.id:<44} {m.context_length:>9,}  "
            f"{m.prompt_usd_per_mtok:>10.4f} {m.completion_usd_per_mtok:>10.4f}  "
            f"{'yes' if m.supports_tools else 'no'}"
        )
    return "\n".join(rows)


def main() -> None:
    parser = argparse.ArgumentParser(description="List OpenRouter models.")
    parser.add_argument("--all", action="store_true", help="include paid models")
    parser.add_argument(
        "--tools", action="store_true", help="only models supporting tool calling"
    )
    args = parser.parse_args()

    models = fetch_models()
    if not args.all:
        models = [m for m in models if m.is_free]
    if args.tools:
        models = [m for m in models if m.supports_tools]
    models.sort(key=lambda m: m.context_length, reverse=True)

    print(format_table(models))
    print(f"\n{len(models)} model(s).")


if __name__ == "__main__":
    main()

uv run python catalog.py

touch client.py

"""An OpenRouter client built on the OpenAI SDK, plus cost-aware completions.

OpenRouter speaks the OpenAI chat-completions wire format, so the `openai`
package works unchanged once its `base_url` points at OpenRouter. Two things
are OpenRouter-specific and are handled here: the attribution headers, and
reading the per-request `cost` that OpenRouter adds to the `usage` object.
"""

from dataclasses import dataclass

import httpx2
from openai import OpenAI

from config import BASE_URL, Settings


@dataclass(frozen=True)
class Completion:
    """One finished call, with what it cost and who actually served it."""

    text: str
    model: str
    prompt_tokens: int
    completion_tokens: int
    reasoning_tokens: int
    cost_usd: float | None
    finish_reason: str | None

    @property
    def was_truncated(self) -> bool:
        """True when the token ceiling cut the answer short."""
        return self.finish_reason == "length"

    @property
    def cost_note(self) -> str:
        if self.cost_usd is None:
            return "cost: not reported"
        return f"cost: ${self.cost_usd:.6f}"


def build_client(
    settings: Settings, http_client: httpx2.Client | None = None
) -> OpenAI:
    """Point the OpenAI SDK at OpenRouter.

    `http_client` exists for the tests, which pass an `httpx2.Client` wrapping a
    `MockTransport` so no request leaves the machine.
    """
    return OpenAI(
        base_url=BASE_URL,
        api_key=settings.api_key,
        timeout=settings.timeout,
        max_retries=0,
        default_headers={
            "HTTP-Referer": settings.referer,
            "X-OpenRouter-Title": settings.title,
        },
        http_client=http_client,
    )


def read_reasoning_tokens(usage) -> int:
    """Count the tokens a reasoning model spent thinking.

    They are billed and they count against `max_tokens`, but they are not in
    the answer, so a large number here with a short answer explains both a
    surprising bill and a truncated reply.
    """
    details = getattr(usage, "completion_tokens_details", None)
    if details is None:
        return 0
    return int(getattr(details, "reasoning_tokens", 0) or 0)


def read_cost(usage) -> float | None:
    """Pull OpenRouter's `cost` off the SDK's typed usage object.

    `cost` is not part of the OpenAI schema, so the SDK parks it in the
    pydantic model's extras rather than a declared field. It is a float of
    USD charged for this request.
    """
    if usage is None:
        return None
    extra = getattr(usage, "model_extra", None) or {}
    cost = extra.get("cost", getattr(usage, "cost", None))
    return float(cost) if cost is not None else None


def complete(
    client: OpenAI,
    prompt: str,
    model: str,
    fallbacks: list[str] | None = None,
    max_tokens: int = 2048,
) -> Completion:
    """Send one chat completion, optionally with a fallback chain.

    `models` + `route: "fallback"` are OpenRouter extensions to the request
    body, so they travel through `extra_body`, which the SDK merges into the
    JSON it posts.
    """
    extra_body: dict[str, object] = {}
    if fallbacks:
        extra_body["models"] = [model, *fallbacks]
        extra_body["route"] = "fallback"

    response = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        max_tokens=max_tokens,
        extra_body=extra_body or None,
    )

    usage = response.usage
    choice = response.choices[0]
    return Completion(
        text=(choice.message.content or "").strip(),
        model=response.model,
        prompt_tokens=getattr(usage, "prompt_tokens", 0) or 0,
        completion_tokens=getattr(usage, "completion_tokens", 0) or 0,
        reasoning_tokens=read_reasoning_tokens(usage),
        cost_usd=read_cost(usage),
        finish_reason=choice.finish_reason,
    )

touch errors.py

"""Turn OpenRouter failures into one clear line for a human.

The OpenAI SDK raises a named exception for the status codes OpenAI itself
returns. OpenRouter adds meanings of its own, and the one users hit most --
402, out of credits -- has no named class at all, so a handler that only
catches `AuthenticationError` and `RateLimitError` misses it and shows a
traceback instead.

`stream.py` posts with `httpx2` rather than the SDK, so its failures arrive as
`httpx2` exceptions. Both shapes are normalized here so there is one ladder
rather than two that drift apart.
"""

import httpx2
import openai


def describe_failure(exc: Exception) -> str:
    """Map an exception from a completion call to an actionable message."""
    if isinstance(exc, (openai.APIConnectionError, httpx2.TransportError)):
        return "Could not reach openrouter.ai. Check your network, then retry."

    if isinstance(exc, httpx2.HTTPStatusError):
        exc = _as_status_error(exc)

    if isinstance(exc, openai.APIStatusError):
        detail = _detail(exc)
        match exc.status_code:
            case 401:
                return f"Key rejected (401). Check OPENROUTER_API_KEY. {detail}"
            case 402:
                return (
                    "Out of credits (402). Add credits at "
                    f"https://openrouter.ai/credits, or use a `:free` model. {detail}"
                )
            case 429:
                return (
                    "Rate limited (429). Free models allow 20 requests/minute and "
                    f"50/day below 10 purchased credits. {detail}"
                )
            case 502:
                return f"Upstream model failed (502). Try a fallback model. {detail}"
            case 503:
                return (
                    "No provider matched your routing requirements (503). Relax "
                    "the `provider` constraints, allow fallbacks, or retry if "
                    f"every eligible provider was overloaded. {detail}"
                )
            case _:
                return f"OpenRouter returned {exc.status_code}. {detail}"

    return f"Unexpected failure: {exc!r}"


def _as_status_error(exc: httpx2.HTTPStatusError) -> openai.APIStatusError:
    """Rewrap a raw streaming failure so it runs through the same ladder.

    `raise_for_status()` on the streamed response raises `HTTPStatusError`,
    whose body still carries the `{"error": {...}}` envelope the SDK would
    have unwrapped, so unwrap it here.
    """
    try:
        payload = exc.response.json()
    except (ValueError, httpx2.StreamError):
        payload = {}
    inner = payload.get("error") if isinstance(payload, dict) else None
    return openai.APIStatusError(
        str(exc), response=exc.response, body=inner if isinstance(inner, dict) else {}
    )


def _detail(exc: openai.APIStatusError) -> str:
    """Extract OpenRouter's message.

    The SDK unwraps the `{"error": {...}}` envelope, so `exc.body` is already
    the inner object carrying `code` and `message`.
    """
    body = exc.body
    if isinstance(body, dict):
        message = body.get("message")
        if message:
            return f"Server said: {message}"
    return ""

touch stream.py

"""Stream tokens, and catch the two failures that arrive with HTTP 200.

Once the response headers are sent the status is locked at 200, so neither of
these can come back as a 4xx or 5xx:

1. A provider dying mid-generation arrives as an ordinary SSE event carrying an
   `error` object and `finish_reason: "error"`. Code that only concatenates
   deltas treats it as a clean, short answer.
2. A reasoning model spending its whole token budget on `delta.reasoning`
   before it writes a single character of `delta.content` ends with
   `finish_reason: "length"` and an empty answer. Most free models on
   OpenRouter today are reasoning models, so this is the common case, not the
   exotic one.

The raw `httpx2` request here (rather than the SDK's streaming helper) keeps
both visible: neither `reasoning` nor the error event is part of the OpenAI
schema, so reading the JSON directly is the honest way to see them.
"""

import json
from collections.abc import Iterator
from dataclasses import dataclass

import httpx2

from config import BASE_URL, Settings

CONTENT = "content"
REASONING = "reasoning"


class StreamInterrupted(RuntimeError):
    """The stream ended with an error event rather than a normal stop."""


class StreamTruncated(RuntimeError):
    """The stream hit the token ceiling before the model finished."""


@dataclass(frozen=True)
class Delta:
    """One piece of streamed text, and which channel it came from."""

    kind: str  # CONTENT or REASONING
    text: str


def stream_completion(
    settings: Settings,
    prompt: str,
    model: str,
    http_client: httpx2.Client | None = None,
    max_tokens: int = 2048,
) -> Iterator[Delta]:
    """Yield Deltas, raising on a mid-stream error or a truncated answer."""
    owns_client = http_client is None
    client = http_client or httpx2.Client(timeout=settings.timeout)
    payload = {
        "model": model,
        "messages": [{"role": "user", "content": prompt}],
        "max_tokens": max_tokens,
        "stream": True,
    }
    headers = {
        "Authorization": f"Bearer {settings.api_key}",
        "HTTP-Referer": settings.referer,
        "X-OpenRouter-Title": settings.title,
        "Content-Type": "application/json",
    }

    finish_reason: str | None = None
    try:
        with client.stream(
            "POST", f"{BASE_URL}/chat/completions", json=payload, headers=headers
        ) as response:
            if response.status_code >= 400:
                # A streamed body is not read yet, and the error JSON is the
                # only place OpenRouter says what went wrong.
                response.read()
            response.raise_for_status()
            for line in response.iter_lines():
                if not line.startswith("data: "):
                    continue  # SSE comments (": OPENROUTER PROCESSING") and blanks
                data = line.removeprefix("data: ")
                if data == "[DONE]":
                    break
                event = json.loads(data)
                finish_reason = _finish_reason(event) or finish_reason
                yield from _deltas(event)
    finally:
        if owns_client:
            client.close()

    if finish_reason == "length":
        raise StreamTruncated(
            f"hit the {max_tokens}-token ceiling before finishing. On a "
            "reasoning model the budget covers reasoning tokens too, so raise "
            "max_tokens."
        )


def _finish_reason(event: dict) -> str | None:
    return ((event.get("choices") or [{}])[0]).get("finish_reason")


def _deltas(event: dict) -> Iterator[Delta]:
    """Yield the text on one SSE event, or raise if it is an error event."""
    error = event.get("error")
    choice = (event.get("choices") or [{}])[0]

    if error or choice.get("finish_reason") == "error":
        code = (error or {}).get("code", "unknown")
        message = (error or {}).get("message", "provider failed mid-stream")
        raise StreamInterrupted(f"[{code}] {message}")

    delta = choice.get("delta") or {}
    # `reasoning` first: a reasoning model emits it before any answer text.
    if reasoning := delta.get("reasoning"):
        yield Delta(REASONING, reasoning)
    if content := delta.get("content"):
        yield Delta(CONTENT, content)

touch ask.py

"""Command line entry point: ask a question, print the answer and its cost.

    uv run python ask.py "Name three uses for a paperclip."
    uv run python ask.py --stream "Write a haiku about port 8080."
    uv run python ask.py --stream --show-reasoning "Why is the sky blue?"
    uv run python ask.py --model z-ai/glm-5.2:free "Same question, other model."
"""

import argparse
import sys

from client import build_client, complete
from config import ConfigError, load_settings
from errors import describe_failure
from stream import CONTENT, StreamInterrupted, StreamTruncated, stream_completion


def main() -> int:
    parser = argparse.ArgumentParser(description="Ask a model via OpenRouter.")
    parser.add_argument("prompt", help="the question to send")
    parser.add_argument("--model", help="override OPENROUTER_MODEL")
    parser.add_argument(
        "--fallback",
        action="append",
        default=[],
        help="model to try if the primary fails (repeatable); "
        "overrides OPENROUTER_FALLBACKS",
    )
    parser.add_argument("--stream", action="store_true", help="stream the answer")
    parser.add_argument(
        "--show-reasoning",
        action="store_true",
        help="with --stream, also print the model's reasoning to stderr",
    )
    args = parser.parse_args()

    try:
        settings = load_settings()
    except ConfigError as exc:
        print(exc, file=sys.stderr)
        return 2

    model = args.model or settings.model
    fallbacks = args.fallback or list(settings.fallbacks)

    if args.stream:
        try:
            for delta in stream_completion(settings, args.prompt, model):
                if delta.kind == CONTENT:
                    print(delta.text, end="", flush=True)
                elif args.show_reasoning:
                    print(delta.text, end="", file=sys.stderr, flush=True)
            print()
        except StreamInterrupted as exc:
            print(f"\nstream failed: {exc}", file=sys.stderr)
            return 1
        except StreamTruncated as exc:
            print(f"\nanswer incomplete: {exc}", file=sys.stderr)
            return 1
        except Exception as exc:  # noqa: BLE001 - mapped to a human message below
            print(f"\n{describe_failure(exc)}", file=sys.stderr)
            return 1
        return 0

    client = build_client(settings)
    try:
        result = complete(client, args.prompt, model, fallbacks=fallbacks)
    except Exception as exc:  # noqa: BLE001 - mapped to a human message below
        print(describe_failure(exc), file=sys.stderr)
        return 1

    print(result.text)
    reasoning = (
        f" ({result.reasoning_tokens} reasoning)" if result.reasoning_tokens else ""
    )
    print(
        f"\n-- served by {result.model} | "
        f"{result.prompt_tokens} in / {result.completion_tokens} out{reasoning} | "
        f"{result.cost_note}",
        file=sys.stderr,
    )
    if result.was_truncated:
        print(
            "-- warning: hit the token ceiling; raise max_tokens for a full answer",
            file=sys.stderr,
        )
        return 1
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

touch Makefile

.DEFAULT_GOAL := help
.PHONY: help setup models models-all models-tools ask stream test key clean

Q ?= Name three uses for a paperclip.

help:  ## Show this help screen
	@echo "OpenRouter quickstart — available targets:"
	@grep -E '^[a-zA-Z_-]+:.*?## .*$$' $(MAKEFILE_LIST) \
	  | awk 'BEGIN {FS = ":.*?## "}; {printf "  %-14s %s\n", $$1, $$2}'
	@echo ""
	@echo "Override the prompt with Q=..., e.g.  make ask Q='Why is the sky blue?'"

setup:  ## Install dependencies and create .env from the template
	uv sync
	@test -f .env || (cp .env.example .env && echo "created .env — add your key")

models:  ## List free models, widest context first
	uv run python catalog.py

models-all:  ## List every model with per-million-token prices
	uv run python catalog.py --all

models-tools:  ## List free models that support tool calling
	uv run python catalog.py --tools

ask:  ## Ask one question (make ask Q='...')
	uv run python ask.py "$(Q)"

stream:  ## Ask one question and stream the answer (make stream Q='...')
	uv run python ask.py --stream "$(Q)"

test:  ## Run the test suite (no network, no API key)
	uv run pytest -v

key:  ## Show the credit and rate-limit status of your API key
	@set -a && . ./.env && set +a && \
	  curl -s https://openrouter.ai/api/v1/key \
	    -H "Authorization: Bearer $$OPENROUTER_API_KEY" | uv run python -m json.tool

clean:  ## Remove caches and build artifacts
	rm -rf .pytest_cache __pycache__ tests/__pycache__

make

mkdir -p tests
touch tests/test_config.py tests/test_catalog.py

"""Settings tests. Every case passes an explicit env dict, so the developer's
real key is never read."""

import pytest

from config import ConfigError, load_settings


def test_defaults_apply_when_only_the_key_is_set():
    settings = load_settings({"OPENROUTER_API_KEY": "sk-or-v1-abc"})

    assert settings.api_key == "sk-or-v1-abc"
    assert settings.model == "nvidia/nemotron-3-super-120b-a12b:free"
    assert settings.fallbacks == (
        "liquid/lfm-2.5-2.6b:free",
        "google/gemma-4-31b-it:free",
    )
    assert settings.timeout == 60.0


def test_environment_overrides_every_default():
    settings = load_settings(
        {
            "OPENROUTER_API_KEY": "sk-or-v1-abc",
            "OPENROUTER_MODEL": "z-ai/glm-5.2:free",
            "OPENROUTER_REFERER": "https://scriptable.com",
            "OPENROUTER_TITLE": "my-app",
            "OPENROUTER_TIMEOUT": "12.5",
            "OPENROUTER_FALLBACKS": "a/one:free, b/two:free",
        }
    )

    assert settings.model == "z-ai/glm-5.2:free"
    assert settings.referer == "https://scriptable.com"
    assert settings.title == "my-app"
    assert settings.timeout == 12.5
    assert settings.fallbacks == ("a/one:free", "b/two:free")


def test_an_empty_fallback_list_disables_fallbacks():
    """Unset means "use the defaults"; explicitly empty means "none"."""
    settings = load_settings(
        {"OPENROUTER_API_KEY": "sk-or-v1-abc", "OPENROUTER_FALLBACKS": ""}
    )

    assert settings.fallbacks == ()


@pytest.mark.parametrize(
    "env", [{}, {"OPENROUTER_API_KEY": ""}, {"OPENROUTER_API_KEY": "   "}]
)
def test_a_missing_or_blank_key_fails_immediately(env):
    with pytest.raises(ConfigError, match="OPENROUTER_API_KEY is not set"):
        load_settings(env)

"""Catalogue tests: price conversion and filtering, against a fixture that
mirrors the real /models payload."""

import httpx2
import pytest

from catalog import fetch_models, parse_models

PAYLOAD = {
    "data": [
        {
            "id": "vendor/cheap-model",
            "name": "Vendor: Cheap Model",
            "context_length": 8192,
            "pricing": {"prompt": "0.000000044", "completion": "0.000000177"},
            "supported_parameters": ["max_tokens", "temperature"],
        },
        {
            "id": "vendor/free-model:free",
            "name": "Vendor: Free Model",
            "context_length": 262144,
            "pricing": {"prompt": "0", "completion": "0"},
            "supported_parameters": ["tools", "temperature"],
        },
        {
            "id": "vendor/sparse-model",
            "name": "Vendor: Sparse Model",
            "context_length": None,
            "pricing": {},
            "supported_parameters": None,
        },
    ]
}


def test_prices_convert_from_per_token_to_per_million():
    cheap = parse_models(PAYLOAD)[0]

    # approx, not ==: multiplying a tiny per-token float by 1e6 lands at
    # 0.17700000000000002, which is right to every digit that matters.
    assert cheap.prompt_usd_per_mtok == pytest.approx(0.044)
    assert cheap.completion_usd_per_mtok == pytest.approx(0.177)


def test_free_and_tool_support_are_derived_from_the_payload():
    _, free, _ = parse_models(PAYLOAD)

    assert free.is_free is True
    assert free.supports_tools is True


def test_paid_model_is_not_free_even_when_cheap():
    cheap = parse_models(PAYLOAD)[0]

    assert cheap.is_free is False
    assert cheap.supports_tools is False


def test_missing_fields_do_not_raise():
    sparse = parse_models(PAYLOAD)[2]

    assert sparse.context_length == 0
    assert sparse.prompt_usd_per_mtok == 0.0
    assert sparse.supported_parameters == ()


def test_fetch_models_calls_the_public_models_endpoint():
    seen = {}

    def handler(request: httpx2.Request) -> httpx2.Response:
        seen["url"] = str(request.url)
        seen["auth"] = request.headers.get("authorization")
        return httpx2.Response(200, json=PAYLOAD)

    client = httpx2.Client(transport=httpx2.MockTransport(handler))
    models = fetch_models(client=client)

    assert seen["url"] == "https://openrouter.ai/api/v1/models"
    assert seen["auth"] is None  # the catalogue needs no API key
    assert [m.id for m in models] == [
        "vendor/cheap-model",
        "vendor/free-model:free",
        "vendor/sparse-model",
    ]

touch tests/test_client.py tests/test_errors.py tests/test_stream.py

"""Client tests. Every request is answered by a MockTransport, so no key and
no network are involved."""

import json

import httpx2
import pytest
from openai import AuthenticationError

from client import build_client, complete, read_cost, read_reasoning_tokens
from config import Settings

SETTINGS = Settings(
    api_key="test-key",
    model="nvidia/nemotron-3-super-120b-a12b:free",
    fallbacks=(),
    referer="https://example.com",
    title="orkit-tests",
    timeout=5.0,
)


def completion_body(
    model: str = "nvidia/nemotron-3-super-120b-a12b:free",
    cost: float = 0.000123,
    finish_reason: str = "stop",
    reasoning_tokens: int = 0,
):
    return {
        "id": "gen-test",
        "object": "chat.completion",
        "created": 1,
        "model": model,
        "choices": [
            {
                "index": 0,
                "message": {"role": "assistant", "content": " a paperclip \n"},
                "finish_reason": finish_reason,
            }
        ],
        "usage": {
            "prompt_tokens": 11,
            "completion_tokens": 3,
            "total_tokens": 14,
            "cost": cost,
            "completion_tokens_details": {"reasoning_tokens": reasoning_tokens},
        },
    }


def client_returning(body: dict, status: int = 200, captured: dict | None = None):
    def handler(request: httpx2.Request) -> httpx2.Response:
        if captured is not None:
            captured["headers"] = dict(request.headers)
            captured["body"] = json.loads(request.content)
        return httpx2.Response(status, json=body)

    transport = httpx2.MockTransport(handler)
    return build_client(SETTINGS, http_client=httpx2.Client(transport=transport))


def test_completion_text_is_stripped_and_usage_is_reported():
    result = complete(client_returning(completion_body()), "why", SETTINGS.model)

    assert result.text == "a paperclip"
    assert (result.prompt_tokens, result.completion_tokens) == (11, 3)
    assert result.cost_usd == pytest.approx(0.000123)
    assert result.cost_note == "cost: $0.000123"


def test_attribution_headers_and_auth_are_sent():
    captured: dict = {}
    complete(
        client_returning(completion_body(), captured=captured), "why", SETTINGS.model
    )

    headers = captured["headers"]
    assert headers["authorization"] == "Bearer test-key"
    assert headers["http-referer"] == "https://example.com"
    assert headers["x-openrouter-title"] == "orkit-tests"


def test_fallbacks_become_a_models_array_and_fallback_route():
    captured: dict = {}
    complete(
        client_returning(completion_body(), captured=captured),
        "why",
        "primary/model",
        fallbacks=["backup/one", "backup/two"],
    )

    body = captured["body"]
    assert body["model"] == "primary/model"
    assert body["models"] == ["primary/model", "backup/one", "backup/two"]
    assert body["route"] == "fallback"


def test_no_fallbacks_means_no_routing_fields():
    captured: dict = {}
    complete(client_returning(completion_body(), captured=captured), "why", "solo/model")

    body = captured["body"]
    assert "models" not in body
    assert "route" not in body


def test_served_model_comes_from_the_response_not_the_request():
    body = completion_body(model="backup/one")
    result = complete(client_returning(body), "why", "primary/model")

    assert result.model == "backup/one"


def test_missing_cost_is_reported_as_unknown_not_zero():
    body = completion_body()
    del body["usage"]["cost"]
    result = complete(client_returning(body), "why", SETTINGS.model)

    assert result.cost_usd is None
    assert result.cost_note == "cost: not reported"


def test_read_cost_handles_absent_usage():
    assert read_cost(None) is None


def test_reasoning_tokens_are_reported_when_the_model_thinks():
    body = completion_body(reasoning_tokens=815)
    result = complete(client_returning(body), "why", SETTINGS.model)

    assert result.reasoning_tokens == 815


def test_reasoning_tokens_default_to_zero_without_details():
    body = completion_body()
    del body["usage"]["completion_tokens_details"]
    result = complete(client_returning(body), "why", SETTINGS.model)

    assert result.reasoning_tokens == 0
    assert read_reasoning_tokens(None) == 0


def test_a_length_finish_reason_is_surfaced_as_truncation():
    """A reasoning model can burn the whole budget before writing an answer;
    `length` is the only signal that the reply is cut off."""
    body = completion_body(finish_reason="length")
    result = complete(client_returning(body), "why", SETTINGS.model)

    assert result.finish_reason == "length"
    assert result.was_truncated is True


def test_a_normal_stop_is_not_truncation():
    result = complete(client_returning(completion_body()), "why", SETTINGS.model)

    assert result.was_truncated is False


def test_status_errors_still_propagate():
    body = {"error": {"code": 401, "message": "No auth credentials found"}}
    with pytest.raises(AuthenticationError):
        complete(client_returning(body, status=401), "why", SETTINGS.model)

"""The failure ladder: every status OpenRouter documents maps to a message
that tells the reader what to do next."""

import httpx2
import openai
import pytest

from errors import describe_failure


def status_error(status: int, message: str) -> openai.APIStatusError:
    """Build the exception the SDK would raise for this status."""
    request = httpx2.Request("POST", "https://openrouter.ai/api/v1/chat/completions")
    body = {"error": {"code": status, "message": message}}
    response = httpx2.Response(status, json=body, request=request)
    return openai.APIStatusError(message, response=response, body=body["error"])


@pytest.mark.parametrize(
    ("status", "expected"),
    [
        (401, "Key rejected (401)"),
        (402, "Out of credits (402)"),
        (429, "Rate limited (429)"),
        (502, "Upstream model failed (502)"),
        (503, "No provider matched your routing requirements (503)"),
    ],
)
def test_documented_statuses_get_actionable_messages(status: int, expected: str):
    described = describe_failure(status_error(status, "upstream said no"))

    assert described.startswith(expected)
    assert "Server said: upstream said no" in described


def test_unmapped_status_still_reports_the_code():
    assert "418" in describe_failure(status_error(418, "teapot"))


def stream_status_error(status: int, message: str) -> httpx2.HTTPStatusError:
    """Build the exception the raw streaming path raises.

    `stream.py` calls `raise_for_status()` on an `httpx2` response, so the body
    still carries the `{"error": {...}}` envelope the SDK would have unwrapped.
    """
    request = httpx2.Request("POST", "https://openrouter.ai/api/v1/chat/completions")
    body = {"error": {"code": status, "message": message}}
    response = httpx2.Response(status, json=body, request=request)
    return httpx2.HTTPStatusError(f"Client error '{status}'", request=request, response=response)


def test_streaming_failures_run_through_the_same_ladder():
    """A 4xx while streaming is an httpx2 error, not an SDK one; without this
    the reader gets a raw HTTPStatusError repr instead of the 401 message."""
    described = describe_failure(stream_status_error(401, "User not found."))

    assert described.startswith("Key rejected (401)")
    assert "Server said: User not found." in described


def test_a_dropped_connection_while_streaming_is_a_network_problem():
    request = httpx2.Request("POST", "https://openrouter.ai/api/v1/chat/completions")
    exc = httpx2.ConnectError("connection refused", request=request)

    assert "Could not reach openrouter.ai" in describe_failure(exc)


def test_connection_error_is_named_as_a_network_problem():
    request = httpx2.Request("POST", "https://openrouter.ai/api/v1/chat/completions")
    exc = openai.APIConnectionError(request=request)

    assert "Could not reach openrouter.ai" in describe_failure(exc)


def test_unknown_exception_is_not_swallowed():
    assert "Unexpected failure" in describe_failure(ValueError("something else"))

"""Streaming tests: reasoning deltas, and the two failures that arrive as
HTTP 200."""

import httpx2
import pytest

from config import Settings
from stream import (
    CONTENT,
    REASONING,
    Delta,
    StreamInterrupted,
    StreamTruncated,
    stream_completion,
)

SETTINGS = Settings(
    api_key="test-key",
    model="nvidia/nemotron-3-super-120b-a12b:free",
    fallbacks=(),
    referer="https://example.com",
    title="orkit-tests",
    timeout=5.0,
)


def event(delta: str = "", reasoning: str = "", finish=None) -> str:
    parts = []
    if delta:
        parts.append(f'"content":"{delta}"')
    if reasoning:
        parts.append(f'"reasoning":"{reasoning}"')
    finish_json = f'"{finish}"' if finish else "null"
    return (
        '{"id":"g","choices":[{"index":0,"delta":{'
        + ",".join(parts)
        + f'}},"finish_reason":{finish_json}}}]}}'
    )


ERROR_EVENT = (
    '{"id":"g","model":"m","provider":"P",'
    '"error":{"code":502,"message":"upstream died"},'
    '"choices":[{"index":0,"delta":{"content":""},"finish_reason":"error"}]}'
)


def sse_client(events: list[str], status: int = 200) -> httpx2.Client:
    payload = "".join(f"data: {e}\n\n" for e in events) + "data: [DONE]\n\n"

    def handler(request: httpx2.Request) -> httpx2.Response:
        return httpx2.Response(
            status,
            content=payload.encode(),
            headers={"content-type": "text/event-stream"},
        )

    return httpx2.Client(transport=httpx2.MockTransport(handler))


def test_content_deltas_are_yielded_in_order():
    deltas = list(
        stream_completion(
            SETTINGS,
            "hi",
            "m",
            http_client=sse_client(
                [event(delta="Hel"), event(delta="lo"), event(finish="stop")]
            ),
        )
    )

    assert deltas == [Delta(CONTENT, "Hel"), Delta(CONTENT, "lo")]


def test_reasoning_is_a_separate_channel_from_the_answer():
    """Most free models are reasoning models: the answer arrives after the
    thinking, on a different key."""
    deltas = list(
        stream_completion(
            SETTINGS,
            "hi",
            "m",
            http_client=sse_client(
                [
                    event(reasoning="the user wants"),
                    event(reasoning=" a greeting"),
                    event(delta="Hello"),
                    event(finish="stop"),
                ]
            ),
        )
    )

    assert [d.kind for d in deltas] == [REASONING, REASONING, CONTENT]
    assert "".join(d.text for d in deltas if d.kind == CONTENT) == "Hello"


def test_one_event_carrying_both_yields_reasoning_first():
    deltas = list(
        stream_completion(
            SETTINGS,
            "hi",
            "m",
            http_client=sse_client([event(delta="A", reasoning="think"), event(finish="stop")]),
        )
    )

    assert deltas == [Delta(REASONING, "think"), Delta(CONTENT, "A")]


def test_mid_stream_error_raises_instead_of_truncating_silently():
    stream = stream_completion(
        SETTINGS, "hi", "m", http_client=sse_client([event(delta="Hel"), ERROR_EVENT])
    )

    collected = []
    with pytest.raises(StreamInterrupted, match=r"\[502\] upstream died"):
        for delta in stream:
            collected.append(delta)

    assert collected == [Delta(CONTENT, "Hel")]  # partial answer before the failure


def test_hitting_the_token_ceiling_raises_after_the_deltas():
    """finish_reason "length" means the answer is cut off, not finished. On a
    reasoning model the budget can be gone before any content is emitted."""
    stream = stream_completion(
        SETTINGS,
        "hi",
        "m",
        max_tokens=64,
        http_client=sse_client([event(reasoning="thinking hard"), event(finish="length")]),
    )

    collected = []
    with pytest.raises(StreamTruncated, match="64-token ceiling"):
        for delta in stream:
            collected.append(delta)

    assert collected == [Delta(REASONING, "thinking hard")]


def test_a_normal_stop_does_not_raise():
    list(
        stream_completion(
            SETTINGS, "hi", "m", http_client=sse_client([event(delta="ok", finish="stop")])
        )
    )


def test_processing_comments_are_ignored():
    payload = (
        ": OPENROUTER PROCESSING\n\n"
        + f"data: {event(delta='Hel', finish='stop')}\n\n"
        + "data: [DONE]\n\n"
    )

    def handler(request: httpx2.Request) -> httpx2.Response:
        return httpx2.Response(
            200,
            content=payload.encode(),
            headers={"content-type": "text/event-stream"},
        )

    client = httpx2.Client(transport=httpx2.MockTransport(handler))
    assert list(stream_completion(SETTINGS, "hi", "m", http_client=client)) == [
        Delta(CONTENT, "Hel")
    ]