orchestrata_ultimate.providers.adapters.openrouter_adapter

Reference › orchestrata_ultimate › providers › adapters › openrouter_adapter

openrouter_adapter is the adapter for OpenRouter, which fronts many vendors' models under VENDOR/MODEL names. OpenRouterAdapter extends BaseAdapter (base) and implements complete, get_models and supports_capability. Besides the authorisation and content-type headers, _get_headers sends the OpenRouter attribution headers HTTP-Referer and X-Title, whose values come from the environment variables OPENROUTER_REFERER and OPENROUTER_TITLE with fixed defaults. client registers it, and the ask_openrouter script imports it.

At a glance

Class diagram

Dependency diagram

API

Classes

OpenRouterAdapter

Bases: BaseAdapter · Line 29

OpenRouter API provider adapter.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/providers/adapters/openrouter_adapter.py

"""OpenRouter provider adapter.

SPEC-011 FR-007: Implements IProvider for OpenRouter API.
Routes to multiple upstream LLM providers via a single API key.
Supports Claude, GPT-4, DeepSeek, Llama, Mistral, Qwen via OpenRouter.
"""

import logging
import os
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.protocols import (
    Message,
    ModelCapability,
    ModelInfo,
    RawProviderResponse,
)
from orchestrata_ultimate.providers.adapters.base import (
    BaseAdapter,
    strip_tools_when_choice_none,
)

logger = logging.getLogger(__name__)

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


class OpenRouterAdapter(BaseAdapter):
    """OpenRouter API provider adapter.

    Routes requests to multiple upstream LLM providers
    (Claude, GPT-4, DeepSeek, Llama, Mistral, Qwen) through
    a single API key and OpenAI-compatible endpoint.
    """

    def _get_headers(self) -> dict[str, str]:
        """Build request headers with OpenRouter-specific fields.

        Returns:
            Headers dict with Authorization, Content-Type,
            HTTP-Referer, and X-Title.
        """
        return {
            "Authorization": f"Bearer {self._config.api_key}",
            "Content-Type": "application/json",
            "HTTP-Referer": os.environ.get(
                "OPENROUTER_REFERER",
                "https://orchestrata.dev",
            ),
            "X-Title": os.environ.get("OPENROUTER_TITLE", "Orchestrata"),
        }

    async def complete(
        self,
        messages: list[Message],
        model: str,
        parameters: dict[str, Any],
    ) -> RawProviderResponse:
        """Send completion request to OpenRouter.

        Args:
            messages: Conversation messages.
            model: Model identifier
                (e.g. "anthropic/claude-sonnet-4-6").
            parameters: Additional parameters
                (temperature, max_tokens, etc.).

        Returns:
            RawProviderResponse with content and token usage.
        """
        client = await self._ensure_client()
        # Build OpenAI-compatible messages, preserving tool-calling fields
        # when present (assistant tool_calls, and tool-result messages) --
        # same handling as deepseek_adapter.py. Before 2026-07-13 this
        # adapter dropped BOTH sides of the tool protocol: assistant
        # tool_calls were not forwarded in requests and response tool_calls
        # were discarded, so every tool-loop volk (architect, codedirector
        # review, U:SD, ...) silently lost its tools under any openrouter
        # profile -- the model's turn ended at the "let me look at..."
        # preamble and the loop saw plain text. Found live in the first
        # openrouter-cheap pipeline run: U:Ar crashed with all design
        # sections blank.
        api_messages: list[dict[str, Any]] = []
        for m in messages:
            msg: dict[str, Any] = {"role": m.role, "content": m.content or ""}
            if m.tool_calls:
                msg["tool_calls"] = m.tool_calls
            if m.tool_call_id:
                msg["tool_call_id"] = m.tool_call_id
            if m.name:
                msg["name"] = m.name
            api_messages.append(msg)
        # Dogfood run-3e (F12): omit `tools`/`tool_choice` from the request
        # entirely when the caller asks for tool_choice=="none" -- mirrors
        # anthropic_adapter.py's handling and prevents whichever upstream
        # model OpenRouter routes to from being drawn into emitting tool-call
        # syntax as text (see strip_tools_when_choice_none's docstring in
        # base.py).
        call_params = strip_tools_when_choice_none(parameters)
        payload = {
            "model": model,
            "messages": api_messages,
            **call_params,
        }
        response = await client.post("/chat/completions", json=payload)
        response.raise_for_status()
        data = response.json()
        choice = data.get("choices", [{}])[0]
        message = choice.get("message", {})
        usage = data.get("usage", {})
        return RawProviderResponse(
            content=(message.get("content") or ""),
            input_tokens=usage.get("prompt_tokens", 0),
            output_tokens=usage.get("completion_tokens", 0),
            model=data.get("model", model),
            tool_calls=message.get("tool_calls") or [],
            metadata={
                "id": data.get("id", ""),
                "finish_reason": choice.get("finish_reason", ""),
                "upstream_model": data.get("model", ""),
                "openrouter_id": data.get("id", ""),
            },
        )

    @traces(spec="011", story="US-001", reqs=["REQ-011-028"])
    def get_models(self) -> list[ModelInfo]:
        """Return available models.

        Returns:
            List of ModelInfo for registered OpenRouter models.
        """
        return [
            ModelInfo(
                name=model_name,
                provider="openrouter",
                capabilities=pricing.capabilities,
                context_window=pricing.context_window,
                pricing=pricing,
            )
            for model_name, pricing in self._config.pricing.items()
        ]

    @traces(spec="011", story="US-001", reqs=["REQ-011-007"])
    def supports_capability(self, capability: ModelCapability) -> bool:
        """Check if any registered model supports the capability.

        Args:
            capability: The capability to check.

        Returns:
            True if at least one model supports it.
        """
        for pricing in self._config.pricing.values():
            if capability in pricing.capabilities:
                return True
        return False