orchestrata_ultimate.providers.adapters.anthropic_adapter

Reference › orchestrata_ultimate › providers › adapters › anthropic_adapter

anthropic_adapter is the adapter for the Anthropic API (specification 011, requirement 031). AnthropicAdapter extends BaseAdapter (base) and implements the provider protocol: complete, get_models and supports_capability. client registers it when Anthropic credentials are configured. Anthropic's message format differs from the OpenAI shape the rest of the system uses, so three helpers translate: _build_anthropic_messages separates system messages from the conversation and converts the rest, _translate_tools_to_anthropic converts OpenAI-style tool definitions, and _translate_tool_choice_to_anthropic maps a tool-choice value to Anthropic's form.

At a glance

Class diagram

Dependency diagram

API

Classes

AnthropicAdapter

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

Anthropic API provider adapter.

Functions

Source

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

"""Anthropic provider adapter.

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

import json
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

logger = logging.getLogger(__name__)


@traces(spec="011", story="US-001", reqs=["REQ-011-031"])
def _translate_tools_to_anthropic(tools: list[dict[str, Any]]) -> list[dict[str, Any]]:
    """Translate OpenAI-shaped tool defs into Anthropic's tool schema.

    OpenAI/SPEC-103 shape: ``{"type": "function", "function": {"name",
    "description", "parameters"}}``. Anthropic wants the function's fields
    flattened to the top level with ``parameters`` renamed ``input_schema``.
    """
    translated = []
    for tool in tools:
        fn = tool.get("function", tool)
        translated.append(
            {
                "name": fn.get("name", ""),
                "description": fn.get("description", ""),
                "input_schema": fn.get("parameters", {}),
            }
        )
    return translated


@traces(spec="011", story="US-001", reqs=["REQ-011-031"])
def _translate_tool_choice_to_anthropic(tool_choice: Any) -> dict[str, Any] | None:
    """Map an OpenAI-style ``tool_choice`` onto Anthropic's ``tool_choice`` shape.

    ``"auto"`` -> ``{"type": "auto"}``; ``"required"`` -> ``{"type": "any"}``
    (Anthropic has no "required", "any" is the closest equivalent); the
    OpenAI dict form ``{"type": "function", "function": {"name": N}}`` ->
    ``{"type": "tool", "name": N}``. ``"none"``/``None``/anything
    unrecognized returns ``None`` — Anthropic has no "none" equivalent, so
    the caller omits ``tool_choice`` (and, for "none", the ``tools`` key
    itself) from the payload instead.
    """
    if tool_choice == "auto":
        return {"type": "auto"}
    if tool_choice == "required":
        return {"type": "any"}
    if isinstance(tool_choice, dict):
        fn = tool_choice.get("function", {})
        name = fn.get("name") or tool_choice.get("name")
        if name:
            return {"type": "tool", "name": name}
    return None


@traces(spec="011", story="US-001", reqs=["REQ-011-031"])
def _build_anthropic_messages(
    messages: list[Message],
) -> tuple[str, list[dict[str, Any]]]:
    """Split system messages out and translate the rest to Anthropic's shape.

    Plain user/assistant text messages pass through unchanged. Assistant
    messages carrying OpenAI-style ``tool_calls`` become an assistant turn
    whose content is a list of ``tool_use`` blocks (preceded by a ``text``
    block if there was also assistant text). ``tool``-role messages become
    ``tool_result`` blocks; consecutive ``tool`` messages are merged into a
    single ``user`` turn (one ``tool_result`` block per call) because
    Anthropic requires strictly alternating user/assistant turns — one
    ``user`` message per assistant tool-call turn, not one per result.
    """
    system_content = ""
    api_messages: list[dict[str, Any]] = []
    for m in messages:
        if m.role == "system":
            system_content += m.content + "\n"
        elif m.role == "tool":
            block = {
                "type": "tool_result",
                "tool_use_id": m.tool_call_id or "",
                "content": m.content,
            }
            if (
                api_messages
                and api_messages[-1]["role"] == "user"
                and isinstance(api_messages[-1]["content"], list)
            ):
                api_messages[-1]["content"].append(block)
            else:
                api_messages.append({"role": "user", "content": [block]})
        elif m.role == "assistant" and m.tool_calls:
            content_blocks: list[dict[str, Any]] = []
            if m.content:
                content_blocks.append({"type": "text", "text": m.content})
            for call in m.tool_calls:
                fn = call.get("function", {})
                try:
                    tool_input = json.loads(fn.get("arguments") or "{}")
                except json.JSONDecodeError, TypeError:
                    tool_input = {}
                content_blocks.append(
                    {
                        "type": "tool_use",
                        "id": call.get("id", ""),
                        "name": fn.get("name", ""),
                        "input": tool_input,
                    }
                )
            api_messages.append({"role": "assistant", "content": content_blocks})
        else:
            api_messages.append({"role": m.role, "content": m.content})
    return system_content, api_messages


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

    def _get_headers(self) -> dict[str, str]:
        return {
            "x-api-key": self._config.api_key,
            "anthropic-version": "2023-06-01",
            "Content-Type": "application/json",
        }

    @traces(spec="011", story="US-001", reqs=["REQ-011-008"])
    async def complete(
        self,
        messages: list[Message],
        model: str,
        parameters: dict[str, Any],
    ) -> RawProviderResponse:
        """Send completion request to Anthropic."""
        client = await self._ensure_client()
        # Separate system messages from the rest, and translate any tool-use
        # turns (assistant tool_calls / tool-result messages) into Anthropic's
        # content-block shape. See _build_anthropic_messages.
        system_content, api_messages = _build_anthropic_messages(messages)
        payload: dict[str, Any] = {
            "model": model,
            "messages": api_messages,
            "max_tokens": parameters.get("max_tokens", 4096),
        }
        if system_content:
            payload["system"] = system_content.strip()
        for key in ("temperature", "top_p", "top_k"):
            if key in parameters:
                payload[key] = parameters[key]
        # request.tools/tool_choice (SPEC-103) arrive here via `parameters`
        # (ProviderManager._build_call_params forwards them). Anthropic has no
        # "none" tool_choice: when the caller asks for "none" we omit `tools`
        # entirely rather than send a tool_choice Anthropic doesn't support.
        tools = parameters.get("tools")
        tool_choice = parameters.get("tool_choice")
        if tools and tool_choice != "none":
            payload["tools"] = _translate_tools_to_anthropic(tools)
            mapped_choice = _translate_tool_choice_to_anthropic(tool_choice)
            if mapped_choice is not None:
                payload["tool_choice"] = mapped_choice
        response = await client.post("/messages", json=payload)
        response.raise_for_status()
        data = response.json()
        content_blocks = data.get("content", [])
        text = "".join(
            block.get("text", "")
            for block in content_blocks
            if block.get("type") == "text"
        )
        # Anthropic returns tool calls as "tool_use" content blocks rather than
        # a top-level tool_calls field. Translate them into the OpenAI-compatible
        # envelope (id / type=function / function.name / function.arguments as a
        # JSON string) that RawProviderResponse.tool_calls uses across adapters
        # (see DeepSeekAdapter.complete and SPEC-103's tool-call envelope), so
        # they are not silently dropped when a model requests a tool.
        tool_calls = [
            {
                "id": block.get("id", ""),
                "type": "function",
                "function": {
                    "name": block.get("name", ""),
                    "arguments": json.dumps(block.get("input", {})),
                },
            }
            for block in content_blocks
            if block.get("type") == "tool_use"
        ]
        usage = data.get("usage", {})
        return RawProviderResponse(
            content=text,
            input_tokens=usage.get("input_tokens", 0),
            output_tokens=usage.get("output_tokens", 0),
            model=data.get("model", model),
            tool_calls=tool_calls,
            metadata={
                "id": data.get("id", ""),
                "stop_reason": data.get("stop_reason", ""),
            },
        )

    @traces(spec="011", story="US-005", reqs=["REQ-011-028"])
    def get_models(self) -> list[ModelInfo]:
        """Return available models."""
        return [
            ModelInfo(
                name=model_name,
                provider="anthropic",
                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 model supports the capability."""
        for pricing in self._config.pricing.values():
            if capability in pricing.capabilities:
                return True
        return False