openrouter

openrouter

OpenRouter HTTP backend for multi-model access via OpenAI-compatible API.

Enables Marianne to use any model available on OpenRouter (free and paid) through a single HTTP backend. Uses the OpenAI-compatible chat completions endpoint at https://openrouter.ai/api/v1/chat/completions.

Key design decisions: - Extends Backend ABC with HttpxClientMixin for lazy httpx client lifecycle. - Model is specified per-request, allowing per-sheet instrument overrides. - Rate limit detection from HTTP 429 status and Retry-After header. - Token usage extracted from the standard OpenAI usage response field. - Free-tier model support (no cost for many models).

Security: API keys are NEVER logged. The key is read from environment and passed only in the Authorization header. The logging infrastructure uses SENSITIVE_PATTERNS to automatically redact fields containing 'api_key', 'token', 'secret', etc.

Classes

OpenRouterBackend

OpenRouterBackend(model=_DEFAULT_MODEL, api_key_env='OPENROUTER_API_KEY', max_tokens=16384, temperature=0.7, timeout_seconds=300.0, base_url=_OPENROUTER_BASE_URL)

Bases: HttpxClientMixin, Backend

Run prompts via the OpenRouter API (OpenAI-compatible).

Provides direct HTTP access to 300+ models including free-tier options. Uses HttpxClientMixin for lazy, connection-pooled httpx client lifecycle.

Example usage::

backend = OpenRouterBackend(
    model="minimax/minimax-m1-80k",
    api_key_env="OPENROUTER_API_KEY",
)
result = await backend.execute("Explain quicksort")

Initialize OpenRouter backend.

Parameters:

Name	Type	Description	Default
model	str	Model ID (e.g., 'minimax/minimax-m1-80k', 'google/gemma-4').	_DEFAULT_MODEL
api_key_env	str	Environment variable containing API key.	'OPENROUTER_API_KEY'
max_tokens	int	Maximum tokens for response.	16384
temperature	float	Sampling temperature (0.0-2.0).	0.7
timeout_seconds	float	Maximum time for API request.	300.0
base_url	str	OpenRouter API base URL (without endpoint path).	_OPENROUTER_BASE_URL

Source code in src/marianne/backends/openrouter.py

def __init__(
    self,
    model: str = _DEFAULT_MODEL,
    api_key_env: str = "OPENROUTER_API_KEY",
    max_tokens: int = 16384,
    temperature: float = 0.7,
    timeout_seconds: float = 300.0,
    base_url: str = _OPENROUTER_BASE_URL,
) -> None:
    """Initialize OpenRouter backend.

    Args:
        model: Model ID (e.g., 'minimax/minimax-m1-80k', 'google/gemma-4').
        api_key_env: Environment variable containing API key.
        max_tokens: Maximum tokens for response.
        temperature: Sampling temperature (0.0-2.0).
        timeout_seconds: Maximum time for API request.
        base_url: OpenRouter API base URL (without endpoint path).
    """
    if not model:
        raise ValueError("model must be a non-empty string")
    if not api_key_env:
        raise ValueError("api_key_env must be a non-empty string")
    if max_tokens < 1:
        raise ValueError(f"max_tokens must be >= 1, got {max_tokens}")
    if timeout_seconds <= 0:
        raise ValueError(f"timeout_seconds must be > 0, got {timeout_seconds}")

    self.model = model
    self.api_key_env = api_key_env
    self.max_tokens = max_tokens
    self.temperature = temperature
    self.timeout_seconds = timeout_seconds
    self._working_directory: Path | None = None

    # Read API key from environment (may be None — checked at execute time)
    self._api_key: str | None = os.environ.get(api_key_env)

    # Error classifier for rate limit wait extraction
    self._error_classifier = ErrorClassifier()

    # Per-sheet overrides — saved originals for clear_overrides()
    self._saved_model: str | None = None
    self._saved_temperature: float | None = None
    self._saved_max_tokens: int | None = None
    self._has_overrides: bool = False

    # Real-time output logging (set per-sheet by runner)
    self._stdout_log_path: Path | None = None
    self._stderr_log_path: Path | None = None

    # Preamble and extensions for prompt injection
    self._preamble: str | None = None
    self._prompt_extensions: list[str] = []

    # Build auth headers
    headers: dict[str, str] = {
        "Content-Type": "application/json",
        "HTTP-Referer": "https://github.com/Mzzkc/marianne-ai-compose",
        "X-Title": "Marianne AI Compose",
    }
    if self._api_key:
        headers["Authorization"] = f"Bearer {self._api_key}"

    # HTTP client lifecycle via shared mixin
    self._init_httpx_mixin(
        base_url.rstrip("/"),
        timeout_seconds,
        connect_timeout=10.0,
        headers=headers,
    )

Attributes

name property

name

Human-readable backend name including model.

Functions

from_config classmethod

from_config(config)

Create backend from a BackendConfig.

Parameters:

Name	Type	Description	Default
config	object	A BackendConfig instance (typed as object to avoid circular import — BackendConfig lives in core.config).	required

Returns:

Type	Description
OpenRouterBackend	Configured OpenRouterBackend instance.

Source code in src/marianne/backends/openrouter.py

@classmethod
def from_config(cls, config: object) -> OpenRouterBackend:
    """Create backend from a BackendConfig.

    Args:
        config: A BackendConfig instance (typed as object to avoid
            circular import — BackendConfig lives in core.config).

    Returns:
        Configured OpenRouterBackend instance.
    """
    model = getattr(config, "model", _DEFAULT_MODEL) or _DEFAULT_MODEL
    timeout = getattr(config, "timeout_seconds", 300.0)
    api_key_env = getattr(config, "api_key_env", "OPENROUTER_API_KEY")
    max_tokens = getattr(config, "max_tokens", 16384)
    temperature = getattr(config, "temperature", 0.7)
    return cls(
        model=model,
        api_key_env=api_key_env,
        max_tokens=max_tokens,
        temperature=temperature,
        timeout_seconds=timeout,
    )

apply_overrides

apply_overrides(overrides)

Apply per-sheet overrides for the next execution.

Source code in src/marianne/backends/openrouter.py

def apply_overrides(self, overrides: dict[str, object]) -> None:
    """Apply per-sheet overrides for the next execution."""
    if not overrides:
        return
    self._saved_model = self.model
    self._saved_temperature = self.temperature
    self._saved_max_tokens = self.max_tokens
    self._has_overrides = True
    if "model" in overrides:
        self.model = str(overrides["model"])
    if "temperature" in overrides:
        self.temperature = float(overrides["temperature"])  # type: ignore[arg-type]
    if "max_tokens" in overrides:
        self.max_tokens = int(overrides["max_tokens"])  # type: ignore[call-overload]

clear_overrides

clear_overrides()

Restore original backend parameters after per-sheet execution.

Source code in src/marianne/backends/openrouter.py

def clear_overrides(self) -> None:
    """Restore original backend parameters after per-sheet execution."""
    if not self._has_overrides:
        return
    self.model = self._saved_model  # type: ignore[assignment]
    self.temperature = self._saved_temperature  # type: ignore[assignment]
    self.max_tokens = self._saved_max_tokens  # type: ignore[assignment]
    self._saved_model = None
    self._saved_temperature = None
    self._saved_max_tokens = None
    self._has_overrides = False

set_preamble

set_preamble(preamble)

Set the dynamic preamble for the next execution.

Source code in src/marianne/backends/openrouter.py

def set_preamble(self, preamble: str | None) -> None:
    """Set the dynamic preamble for the next execution."""
    self._preamble = preamble

set_prompt_extensions

set_prompt_extensions(extensions)

Set prompt extensions for the next execution.

Source code in src/marianne/backends/openrouter.py

def set_prompt_extensions(self, extensions: list[str]) -> None:
    """Set prompt extensions for the next execution."""
    self._prompt_extensions = [e for e in extensions if e.strip()]

set_output_log_path

set_output_log_path(path)

Set base path for real-time output logging.

Parameters:

Name	Type	Description	Default
path	Path | None	Base path for log files (without extension), or None to disable.	required

Source code in src/marianne/backends/openrouter.py

def set_output_log_path(self, path: Path | None) -> None:
    """Set base path for real-time output logging.

    Args:
        path: Base path for log files (without extension), or None to disable.
    """
    if path is None:
        self._stdout_log_path = None
        self._stderr_log_path = None
    else:
        self._stdout_log_path = path.with_suffix(".stdout.log")
        self._stderr_log_path = path.with_suffix(".stderr.log")

execute async

execute(prompt, *, timeout_seconds=None)

Execute a prompt via the OpenRouter API.

Sends a chat completion request to OpenRouter's OpenAI-compatible endpoint and returns the result.

Parameters:

Name	Type	Description	Default
prompt	str	The prompt to send.	required
timeout_seconds	float | None	Per-call timeout override. Logged but not enforced (httpx client timeout from init is used).	None

Returns:

Type	Description
ExecutionResult	ExecutionResult with API response and metadata.

Source code in src/marianne/backends/openrouter.py

async def execute(
    self, prompt: str, *, timeout_seconds: float | None = None,
) -> ExecutionResult:
    """Execute a prompt via the OpenRouter API.

    Sends a chat completion request to OpenRouter's OpenAI-compatible
    endpoint and returns the result.

    Args:
        prompt: The prompt to send.
        timeout_seconds: Per-call timeout override. Logged but not
            enforced (httpx client timeout from __init__ is used).

    Returns:
        ExecutionResult with API response and metadata.
    """
    if timeout_seconds is not None:
        _logger.debug(
            "timeout_override_ignored",
            backend="openrouter",
            requested=timeout_seconds,
            actual=self.timeout_seconds,
        )

    start_time = time.monotonic()
    started_at = utc_now()

    _logger.debug(
        "openrouter_execute_start",
        model=self.model,
        prompt_length=len(prompt),
        max_tokens=self.max_tokens,
    )

    # Check API key before making request
    if not self._api_key:
        duration = time.monotonic() - start_time
        msg = (
            f"API key not found in environment variable: {self.api_key_env}. "
            "Set it with: export OPENROUTER_API_KEY=your-key"
        )
        _logger.error(
            "configuration_error",
            api_key_env=self.api_key_env,
        )
        self._write_log_file(self._stderr_log_path, msg)
        return ExecutionResult(
            success=False,
            exit_code=1,
            stdout="",
            stderr=msg,
            duration_seconds=duration,
            started_at=started_at,
            error_type="configuration",
            error_message=msg,
            model=self.model,
        )

    assembled_prompt = self._build_prompt(prompt)

    payload: dict[str, Any] = {
        "model": self.model,
        "messages": [{"role": "user", "content": assembled_prompt}],
        "max_tokens": self.max_tokens,
        "temperature": self.temperature,
    }

    try:
        client = await self._get_client()

        response = await client.post(
            "/chat/completions",
            json=payload,
        )

        duration = time.monotonic() - start_time

        # Handle rate limiting via HTTP status
        if response.status_code == 429:
            return self._handle_rate_limit(response, duration, started_at)

        # Handle other HTTP errors
        if response.status_code >= 400:
            return self._handle_http_error(response, duration, started_at)

        # Parse successful response
        data = response.json()
        return self._parse_success_response(data, duration, started_at)

    except httpx.ConnectError as e:
        duration = time.monotonic() - start_time
        _logger.error("openrouter_connection_error", error=str(e))
        self._write_log_file(self._stderr_log_path, str(e))
        return ExecutionResult(
            success=False,
            exit_code=503,
            stdout="",
            stderr=f"Connection error: {e}",
            duration_seconds=duration,
            started_at=started_at,
            error_type="connection",
            error_message=str(e),
            model=self.model,
        )

    except httpx.TimeoutException as e:
        duration = time.monotonic() - start_time
        _logger.error(
            "openrouter_timeout",
            timeout_seconds=self.timeout_seconds,
            error=str(e),
        )
        self._write_log_file(self._stderr_log_path, str(e))
        return ExecutionResult(
            success=False,
            exit_code=408,
            stdout="",
            stderr=f"Timeout after {self.timeout_seconds}s: {e}",
            duration_seconds=duration,
            started_at=started_at,
            exit_reason="timeout",
            error_type="timeout",
            error_message=f"API timeout after {self.timeout_seconds}s: {e}",
            model=self.model,
        )

    except Exception as e:
        duration = time.monotonic() - start_time
        _logger.exception(
            "openrouter_execute_error",
            model=self.model,
            error=str(e),
        )
        self._write_log_file(self._stderr_log_path, str(e))
        raise

health_check async

health_check()

Check if the OpenRouter API is reachable and authenticated.

Uses the /models endpoint (lightweight, no token consumption) to verify connectivity and authentication.

Returns:

Type	Description
bool	True if healthy, False otherwise.

Source code in src/marianne/backends/openrouter.py

async def health_check(self) -> bool:
    """Check if the OpenRouter API is reachable and authenticated.

    Uses the /models endpoint (lightweight, no token consumption)
    to verify connectivity and authentication.

    Returns:
        True if healthy, False otherwise.
    """
    if not self._api_key:
        _logger.warning(
            "health_check_failed",
            error_type="MissingAPIKey",
            error=f"No API key configured — set {self.api_key_env}",
        )
        return False

    try:
        client = await self._get_client()
        response = await client.get("/models", timeout=10.0)
        if response.status_code == 200:
            return True
        _logger.warning(
            "openrouter_health_check_failed",
            status_code=response.status_code,
        )
        return False
    except (httpx.HTTPError, OSError, ValueError) as e:
        _logger.warning(
            "openrouter_health_check_error",
            error=str(e),
            error_type=type(e).__name__,
        )
        return False

availability_check async

availability_check()

Check if the backend can be initialized without consuming API quota.

Verifies that the API key is present and the httpx client can be created. Does NOT make any HTTP requests.

Source code in src/marianne/backends/openrouter.py

async def availability_check(self) -> bool:
    """Check if the backend can be initialized without consuming API quota.

    Verifies that the API key is present and the httpx client can be
    created. Does NOT make any HTTP requests.
    """
    if not self._api_key:
        return False
    try:
        await self._get_client()
        return True
    except Exception:
        return False

close async

close()

Close HTTP client and release resources.

Source code in src/marianne/backends/openrouter.py

async def close(self) -> None:
    """Close HTTP client and release resources."""
    await self._close_httpx_client()

Functions