orchestrata_ultimate.providers.adapters.base

Reference › orchestrata_ultimate › providers › adapters › base

base holds the shared HTTP plumbing of the provider adapters. BaseAdapter is the parent of the Anthropic, OpenAI, DeepSeek, Ollama and OpenRouter adapters and is built from a ProviderConfig. _ensure_client returns an httpx.AsyncClient with the configured base URL, headers and timeout. It recreates the client whenever the running event loop differs from the one that created it, because the router can dispatch each volk call on a fresh event loop and a client from a dead loop would fail with "Event loop is closed". Subclasses override _get_headers for authentication. close and aclose close the client, and _build_pricing reads model pricing from the configuration. strip_tools_when_choice_none returns a copy of the call parameters without the tool settings when tool choice is none.

At a glance

Class diagram

Dependency diagram

API

Classes

BaseAdapter

Line 44

Base class for HTTP-based provider adapters.

Functions

Source

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

"""Base adapter with common HTTP logic.

SPEC-011 FR-007: Common functionality for HTTP-based provider adapters.
"""

import logging
from typing import Any

import httpx

from orchestrata_ultimate.core.protocols import (
    ModelPricing,
    ProviderConfig,
)

logger = logging.getLogger(__name__)


def strip_tools_when_choice_none(parameters: dict[str, Any]) -> dict[str, Any]:
    """Return a copy of ``parameters`` with ``tools``/``tool_choice`` omitted
    when the caller asked for ``tool_choice=="none"``.

    Shared by the OpenAI-compatible adapters (openai/deepseek/openrouter),
    which otherwise spread ``parameters`` verbatim into the request body.
    Mirrors ``anthropic_adapter.py``'s existing ``tool_choice=="none"``
    handling (Anthropic has no "none" equivalent, so it omits ``tools``
    outright rather than sending an unsupported value).

    Dogfood run-3e (F12): DeepSeek v4 ignored ``tool_choice="none"`` while
    the ``tools`` schema was still attached to the request and emitted its
    internal tool-call markup as plain text instead of a finished answer.
    Omitting ``tools`` from the request entirely -- not just setting
    ``tool_choice`` -- prevents the model from being drawn into tool syntax
    at all, since there is no schema left to invoke.
    """
    if parameters.get("tool_choice") != "none":
        return parameters
    stripped = dict(parameters)
    stripped.pop("tools", None)
    stripped.pop("tool_choice", None)
    return stripped


class BaseAdapter:
    """Base class for HTTP-based provider adapters.

    Provides shared httpx client management, retry logic,
    and response normalization.
    """

    def __init__(self, config: ProviderConfig):
        self._config = config
        self._client: httpx.AsyncClient | None = None
        self._client_loop: object | None = None

    async def _ensure_client(self) -> httpx.AsyncClient:
        """Get or create the httpx client bound to the running event loop.

        The sync router dispatches each volk call via a fresh ``asyncio.run``
        (a new event loop per call). An ``httpx.AsyncClient`` is bound to the
        loop it was created on, and its ``is_closed`` stays False even after
        that loop is torn down — so reusing it across calls raises
        ``RuntimeError: Event loop is closed``. Recreate whenever the running
        loop differs from the one the client was created on (the stale client
        is abandoned, not aclosed — its loop is already dead).
        """
        import asyncio  # noqa: PLC0415

        loop = asyncio.get_running_loop()
        if (
            self._client is None
            or self._client.is_closed
            or self._client_loop is not loop
        ):
            self._client = httpx.AsyncClient(
                base_url=self._config.base_url or "",
                headers=self._get_headers(),
                timeout=httpx.Timeout(self._config.timeout_seconds),
            )
            self._client_loop = loop
        return self._client

    def _get_headers(self) -> dict[str, str]:
        """Override in subclass to set auth headers."""
        return {}

    async def close(self) -> None:
        """Close the HTTP client."""
        if self._client and not self._client.is_closed:
            await self._client.aclose()
            self._client = None

    async def aclose(self) -> None:
        """Async-close alias for :meth:`close` (httpx/redis naming parity)."""
        await self.close()

    def _build_pricing(self) -> dict[str, ModelPricing]:
        """Return pricing from config."""
        return dict(self._config.pricing)