Skip to content

OpenRouter Model

OpenRouter Model class

Full source code 
import json
import logging
import os
from dataclasses import asdict, dataclass, field
from typing import Any, Literal

import requests
from tenacity import (
    before_sleep_log,
    retry,
    retry_if_not_exception_type,
    stop_after_attempt,
    wait_exponential,
)

from minisweagent.models import GLOBAL_MODEL_STATS
from minisweagent.models.utils.cache_control import set_cache_control

logger = logging.getLogger("openrouter_model")


@dataclass
class OpenRouterModelConfig:
    model_name: str
    model_kwargs: dict[str, Any] = field(default_factory=dict)
    set_cache_control: Literal["default_end"] | None = None
    """Set explicit cache control markers, for example for Anthropic models"""


class OpenRouterAPIError(Exception):
    """Custom exception for OpenRouter API errors."""

    pass


class OpenRouterAuthenticationError(Exception):
    """Custom exception for OpenRouter authentication errors."""

    pass


class OpenRouterRateLimitError(Exception):
    """Custom exception for OpenRouter rate limit errors."""

    pass


class OpenRouterModel:
    def __init__(self, **kwargs):
        self.config = OpenRouterModelConfig(**kwargs)
        self.cost = 0.0
        self.n_calls = 0
        self._api_url = "https://openrouter.ai/api/v1/chat/completions"
        self._api_key = os.getenv("OPENROUTER_API_KEY", "")

    @retry(
        stop=stop_after_attempt(int(os.getenv("MSWEA_MODEL_RETRY_STOP_AFTER_ATTEMPT", "10"))),
        wait=wait_exponential(multiplier=1, min=4, max=60),
        before_sleep=before_sleep_log(logger, logging.WARNING),
        retry=retry_if_not_exception_type(
            (
                OpenRouterAuthenticationError,
                KeyboardInterrupt,
            )
        ),
    )
    def _query(self, messages: list[dict[str, str]], **kwargs):
        headers = {
            "Authorization": f"Bearer {self._api_key}",
            "Content-Type": "application/json",
        }

        payload = {
            "model": self.config.model_name,
            "messages": messages,
            "usage": {"include": True},
            **(self.config.model_kwargs | kwargs),
        }

        try:
            response = requests.post(self._api_url, headers=headers, data=json.dumps(payload), timeout=60)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.HTTPError as e:
            if response.status_code == 401:
                error_msg = "Authentication failed. You can permanently set your API key with `mini-extra config set OPENROUTER_API_KEY YOUR_KEY`."
                raise OpenRouterAuthenticationError(error_msg) from e
            elif response.status_code == 429:
                raise OpenRouterRateLimitError("Rate limit exceeded") from e
            else:
                raise OpenRouterAPIError(f"HTTP {response.status_code}: {response.text}") from e
        except requests.exceptions.RequestException as e:
            raise OpenRouterAPIError(f"Request failed: {e}") from e

    def query(self, messages: list[dict[str, str]], **kwargs) -> dict:
        if self.config.set_cache_control:
            messages = set_cache_control(messages, mode=self.config.set_cache_control)
        response = self._query(messages, **kwargs)

        # Extract cost from usage information
        usage = response.get("usage", {})
        cost = usage.get("cost", 0.0)
        assert cost >= 0.0, f"Cost is negative: {cost}"

        # If total_cost is not available, raise an error
        if cost == 0.0:
            raise OpenRouterAPIError(
                f"No cost information available from OpenRouter API for model {self.config.model_name}. "
                "Cost tracking is required but not provided by the API response."
            )

        self.n_calls += 1
        self.cost += cost
        GLOBAL_MODEL_STATS.add(cost)

        return {
            "content": response["choices"][0]["message"]["content"] or "",
            "extra": {
                "response": response,  # already is json
            },
        }

    def get_template_vars(self) -> dict[str, Any]:
        return asdict(self.config) | {"n_model_calls": self.n_calls, "model_cost": self.cost}

Guide

Setting up OpenRouter models is covered in the quickstart guide.

minisweagent.models.openrouter_model

logger module-attribute 

logger = getLogger('openrouter_model')

OpenRouterModelConfig dataclass 

OpenRouterModelConfig(
    model_name: str,
    model_kwargs: dict[str, Any] = dict(),
    set_cache_control: Literal["default_end"] | None = None,
)

model_name instance-attribute 

model_name: str

model_kwargs class-attribute instance-attribute 

model_kwargs: dict[str, Any] = field(default_factory=dict)

set_cache_control class-attribute instance-attribute 

set_cache_control: Literal['default_end'] | None = None

Set explicit cache control markers, for example for Anthropic models

OpenRouterAPIError

Bases: Exception

Custom exception for OpenRouter API errors.

OpenRouterAuthenticationError

Bases: Exception

Custom exception for OpenRouter authentication errors.

OpenRouterRateLimitError

Bases: Exception

Custom exception for OpenRouter rate limit errors.

OpenRouterModel 

OpenRouterModel(**kwargs)
Source code in src/minisweagent/models/openrouter_model.py 
49
50
51
52
53
54
def __init__(self, **kwargs):
    self.config = OpenRouterModelConfig(**kwargs)
    self.cost = 0.0
    self.n_calls = 0
    self._api_url = "https://openrouter.ai/api/v1/chat/completions"
    self._api_key = os.getenv("OPENROUTER_API_KEY", "")

config instance-attribute 

config = OpenRouterModelConfig(**kwargs)

cost instance-attribute 

cost = 0.0

n_calls instance-attribute 

n_calls = 0

query 

query(messages: list[dict[str, str]], **kwargs) -> dict
Source code in src/minisweagent/models/openrouter_model.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def query(self, messages: list[dict[str, str]], **kwargs) -> dict:
    if self.config.set_cache_control:
        messages = set_cache_control(messages, mode=self.config.set_cache_control)
    response = self._query(messages, **kwargs)

    # Extract cost from usage information
    usage = response.get("usage", {})
    cost = usage.get("cost", 0.0)
    assert cost >= 0.0, f"Cost is negative: {cost}"

    # If total_cost is not available, raise an error
    if cost == 0.0:
        raise OpenRouterAPIError(
            f"No cost information available from OpenRouter API for model {self.config.model_name}. "
            "Cost tracking is required but not provided by the API response."
        )

    self.n_calls += 1
    self.cost += cost
    GLOBAL_MODEL_STATS.add(cost)

    return {
        "content": response["choices"][0]["message"]["content"] or "",
        "extra": {
            "response": response,  # already is json
        },
    }

get_template_vars 

get_template_vars() -> dict[str, Any]
Source code in src/minisweagent/models/openrouter_model.py 
123
124
def get_template_vars(self) -> dict[str, Any]:
    return asdict(self.config) | {"n_model_calls": self.n_calls, "model_cost": self.cost}
bug_report Something broken/unclear?

Open an issue on GitHub!
help Open-ended discussions

Join our Slack!
Our projects
SWE-agent SWE-agent SWE-rex SWE-ReX SWE-smith SWE-smith SWE-bench SWE-bench sb-cli sb-cli