orchestrata_ultimate.providers.adapters.openai_adapter

Reference › orchestrata_ultimate › providers › adapters › openai_adapter

openai_adapter is the adapter for the OpenAI API (specification 011, requirement 030). OpenAIAdapter extends BaseAdapter (base), builds the authorisation headers in _get_headers, and implements the provider protocol: complete sends a chat completion and returns a raw provider response, get_models lists the configured models and supports_capability checks whether any model has a capability. client registers it when its key is configured.

At a glance

Class diagram

Dependency diagram

API

Classes

OpenAIAdapter

Bases: BaseAdapter · Decorators: traces(spec='011', story='US-001', reqs=['REQ-011-030']) · Line 25

OpenAI API provider adapter.

Source

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

"""OpenAI provider adapter.

SPEC-011 FR-007: Implements IProvider for OpenAI API.
"""

import logging
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__)


@traces(spec="011", story="US-001", reqs=["REQ-011-030"])
class OpenAIAdapter(BaseAdapter):
    """OpenAI API provider adapter."""

    def _get_headers(self) -> dict[str, str]:
        return {
            "Authorization": f"Bearer {self._config.api_key}",
            "Content-Type": "application/json",
        }

    async def complete(
        self,
        messages: list[Message],
        model: str,
        parameters: dict[str, Any],
    ) -> RawProviderResponse:
        """Send completion request to OpenAI."""
        client = await self._ensure_client()
        # 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 the model 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": [{"role": m.role, "content": m.content} for m in messages],
            **call_params,
        }
        response = await client.post("/chat/completions", json=payload)
        response.raise_for_status()
        data = response.json()
        choice = data.get("choices", [{}])[0]
        usage = data.get("usage", {})
        return RawProviderResponse(
            content=(choice.get("message", {}).get("content") or ""),
            input_tokens=usage.get("prompt_tokens", 0),
            output_tokens=usage.get("completion_tokens", 0),
            model=data.get("model", model),
            metadata={
                "id": data.get("id", ""),
                "finish_reason": choice.get("finish_reason", ""),
            },
        )

    @traces(spec="011", story="US-001", reqs=["REQ-011-028"])
    def get_models(self) -> list[ModelInfo]:
        """Return available models."""
        models = [
            ModelInfo(
                name=model_name,
                provider="openai",
                capabilities=pricing.capabilities,
                context_window=pricing.context_window,
                pricing=pricing,
            )
            for model_name, pricing in self._config.pricing.items()
        ]
        return models

    @traces(spec="011", story="US-001", reqs=["REQ-011-007"])
    def supports_capability(self, capability: ModelCapability) -> bool:
        """Check if any model supports the capability."""
        for pricing in self._config.pricing.values():
            if capability in pricing.capabilities:
                return True
        return False