Blame for Code-agent-main/ycode/llm/anthropic.py

"""Anthropic Claude provider (Messages API with streaming tool use)."""

from __future__ import annotations

import os
from pathlib import Path
from typing import Any

import anthropic

from ycode.errors import (
    LLMAuthenticationError,
    LLMBadRequestError,
    LLMConnectionError,
    LLMError,
    LLMNotFoundError,
    LLMPermissionError,
    LLMRateLimitError,
    LLMResponseError,
    LLMServerError,
    MissingCredentialsError,
    YCodeError,
)
from ycode.llm.base import (
    AssistantMessage,
    LLMProvider,
    LLMResponse,
    Message,
    StopReason,
    StreamCallbacks,
    ToolCall,
    ToolResultsMessage,
    ToolSpec,
    Usage,
    UserMessage,
)

# Models that take `thinking: {"type": "adaptive"}` and `output_config.effort`.
_ADAPTIVE_PREFIXES = (
    "claude-fable-5",
    "claude-mythos-5",
    "claude-opus-5",
    "claude-opus-4-8",
    "claude-opus-4-7",
    "claude-opus-4-6",
    "claude-sonnet-5",
    "claude-sonnet-4-6",
)
_NO_XHIGH_PREFIXES = ("claude-opus-4-6", "claude-sonnet-4-6")
# Models that accept server-side refusal fallbacks in the `"default"` form.
_FALLBACK_MODELS = {"claude-fable-5-1", "claude-opus-5-5", "claude-opus-5", "claude-sonnet-5-5"}
_FALLBACK_BETA = "server-side-fallback-2026-07-01"

_STOP_REASONS = {
    "end_turn": StopReason.END_TURN,
    "stop_sequence": StopReason.END_TURN,
    "tool_use": StopReason.TOOL_USE,
    "max_tokens": StopReason.MAX_TOKENS,
    "refusal": StopReason.REFUSAL,
    "pause_turn": StopReason.PAUSE,
}

_MAX_JSON_RETRIES = 2


def _base_model(model: str) -> str:
    return model.split(".", 1)[1] if model.startswith("anthropic.") else model


def supports_adaptive_thinking(model: str) -> bool:
    return _base_model(model).startswith(_ADAPTIVE_PREFIXES)


def supports_fallbacks(model: str) -> bool:
    return _base_model(model) in _FALLBACK_MODELS


def has_credentials(env: dict[str, str] | None = None) -> bool:
    """Best-effort check that the SDK will find some credential source."""
    env = dict(os.environ) if env is None else env
    if env.get("ANTHROPIC_API_KEY", "").strip() or env.get("ANTHROPIC_AUTH_TOKEN", "").strip():
        return True
    if env.get("ANTHROPIC_PROFILE") or (
        env.get("ANTHROPIC_FEDERATION_RULE_ID") and env.get("ANTHROPIC_IDENTITY_TOKEN_FILE")
    ):
        return True
    config_dir = Path(env.get("XDG_CONFIG_HOME") or Path.home() / ".config") / "anthropic"
    return config_dir.is_dir() and any(config_dir.iterdir())


def translate_error(exc: Exception) -> YCodeError:
    """Map SDK exceptions to YCode errors (most specific first)."""
    if isinstance(exc, anthropic.AuthenticationError):
        return LLMAuthenticationError("Authentication failed: the API key was rejected.")
    if isinstance(exc, anthropic.PermissionDeniedError):
        return LLMPermissionError(f"Permission denied by the API: {exc.message}")
    if isinstance(exc, anthropic.NotFoundError):
        return LLMNotFoundError(f"Not found: {exc.message}")
    if isinstance(exc, anthropic.RateLimitError):
        retry_after = exc.response.headers.get("retry-after") if exc.response is not None else None
        suffix = f" Retry after {retry_after}s." if retry_after else ""
        return LLMRateLimitError(f"Rate limited by the API.{suffix}")
    if isinstance(exc, anthropic.BadRequestError):
        return LLMBadRequestError(f"The API rejected the request: {exc.message}")
    if isinstance(exc, anthropic.APIStatusError):
        if exc.status_code >= 500:
            return LLMServerError(f"API server error ({exc.status_code}): {exc.message}")
        return LLMError(f"API error ({exc.status_code}): {exc.message}")
    if isinstance(exc, anthropic.APITimeoutError):
        return LLMConnectionError("The request to the API timed out.")
    if isinstance(exc, anthropic.APIConnectionError):
        return LLMConnectionError("Could not connect to the Anthropic API.")
    if isinstance(exc, anthropic.APIResponseValidationError):
        return LLMResponseError(f"The API returned an unexpected response: {exc}")
    if isinstance(exc, TypeError) and "authentication" in str(exc).lower():
        return MissingCredentialsError("No Anthropic API credentials found.")
    return LLMError(f"Unexpected model error: {type(exc).__name__}: {exc}")


class AnthropicProvider(LLMProvider):
    name = "anthropic"

    def __init__(
        self,
        model: str,
        *,
        max_tokens: int = 64000,
        effort: str | None = "high",
        refusal_fallback: bool = True,
        client: Any = None,
    ) -> None:
        super().__init__(model)
        self.max_tokens = max_tokens
        self.effort = effort
        self.refusal_fallback = refusal_fallback
        self._client = client
        # Through a custom gateway, stick to the plainest request shape.
        self._custom_base_url = bool(os.environ.get("ANTHROPIC_BASE_URL"))

    @property
    def client(self) -> Any:
        if self._client is None:
            self._client = anthropic.Anthropic(max_retries=3)
        return self._client

    @property
    def display_name(self) -> str:
        return f"Claude ({self.model})"

    # ---------------------------------------------------------- conversion

    def _tool_param(self, spec: ToolSpec) -> dict[str, Any]:
        param: dict[str, Any] = {
            "name": spec.name,
            "description": spec.description,
            "input_schema": spec.parameters,
        }
        if not self._custom_base_url:
            # Stream large inputs (file contents) as they are generated. The
            # agent validates every input before running a tool.
            param["eager_input_streaming"] = True
        return param

    def convert_messages(self, messages: list[Message]) -> list[dict[str, Any]]:
        out: list[dict[str, Any]] = []

        def push(role: str, content: Any) -> None:
            if out and out[-1]["role"] == role:
                prev = out[-1]["content"]
                if isinstance(prev, str):
                    prev = [{"type": "text", "text": prev}]
                if isinstance(content, str):
                    content = [{"type": "text", "text": content}]
                out[-1]["content"] = [*prev, *content]
            else:
                out.append({"role": role, "content": content})

        for msg in messages:
            if isinstance(msg, UserMessage):
                push("user", msg.text)
            elif isinstance(msg, ToolResultsMessage):
                push("user", [
                    {
                        "type": "tool_result",
                        "tool_use_id": r.tool_call_id,
                        "content": r.content or "(no output)",
                        **({"is_error": True} if r.is_error else {}),
                    }
                    for r in msg.results
                ])
            elif isinstance(msg, AssistantMessage):
                if msg.provider == self.name and msg.provider_data:
                    # Replay native blocks unchanged (keeps thinking blocks valid).
                    push("assistant", list(msg.provider_data))
                    continue
                blocks: list[dict[str, Any]] = []
                if msg.text:
                    blocks.append({"type": "text", "text": msg.text})
                for call in msg.tool_calls:
                    args = call.arguments if isinstance(call.arguments, dict) else {}
                    blocks.append({"type": "tool_use", "id": call.id, "name": call.name, "input": args})
                push("assistant", blocks or [{"type": "text", "text": "(no response)"}])
        return out

    def build_request(self, system: str, messages: list[Message], tools: list[ToolSpec]) -> dict[str, Any]:
        params: dict[str, Any] = {
            "model": self.model,
            "max_tokens": self.max_tokens,
            "system": system,
            "messages": self.convert_messages(messages),
            "tools": [self._tool_param(t) for t in tools],
            # Automatic prompt caching: the growing conversation prefix is reused.
            "cache_control": {"type": "ephemeral"},
        }
        if supports_adaptive_thinking(self.model):
            params["thinking"] = {"type": "adaptive"}
            if self.effort:
                effort = self.effort
                if effort == "xhigh" and _base_model(self.model).startswith(_NO_XHIGH_PREFIXES):
                    effort = "high"
                params["output_config"] = {"effort": effort}
        if self.refusal_fallback and not self._custom_base_url and supports_fallbacks(self.model):
            params["fallbacks"] = "default"
            params["betas"] = [_FALLBACK_BETA]
        return params

    # ------------------------------------------------------------- request

    def complete(
        self,
        *,
        system: str,
        messages: list[Message],
        tools: list[ToolSpec],
        callbacks: StreamCallbacks | None = None,
    ) -> LLMResponse:
        callbacks = callbacks or StreamCallbacks()
        params = self.build_request(system, messages, tools)
        json_failures = 0
        while True:
            try:
                with self.client.beta.messages.stream(**params) as stream:
                    for event in stream:
                        etype = getattr(event, "type", None)
                        if etype == "text":
                            callbacks.text(event.text)
                        elif etype == "content_block_start":
                            block = getattr(event, "content_block", None)
                            if getattr(block, "type", None) == "tool_use":
                                callbacks.tool_call_start(block.name)
                    final = stream.get_final_message()
                break
            except ValueError as exc:
                # Tool-input JSON the SDK could not parse at all: re-issue the turn.
                if isinstance(exc, anthropic.AnthropicError):
                    raise translate_error(exc) from None
                json_failures += 1
                if json_failures > _MAX_JSON_RETRIES:
                    raise LLMResponseError("The model produced malformed tool-call JSON repeatedly.") from None
            except (anthropic.AnthropicError, TypeError) as exc:
                raise translate_error(exc) from None
        return self._parse(final)

    def _parse(self, final: Any) -> LLMResponse:
        content = list(getattr(final, "content", None) or [])
        texts: list[str] = []
        calls: list[ToolCall] = []
        for block in content:
            btype = getattr(block, "type", None)
            if btype == "text":
                texts.append(block.text)
            elif btype == "tool_use":
                calls.append(ToolCall(id=block.id, name=block.name, arguments=block.input))
        raw_stop = getattr(final, "stop_reason", None)
        stop = _STOP_REASONS.get(raw_stop or "", StopReason.OTHER)
        detail = None
        if stop is StopReason.REFUSAL:
            details = getattr(final, "stop_details", None)
            detail = getattr(details, "explanation", None) or getattr(details, "category", None)
        usage_obj = getattr(final, "usage", None)
        usage = Usage(
            input_tokens=getattr(usage_obj, "input_tokens", 0) or 0,
            output_tokens=getattr(usage_obj, "output_tokens", 0) or 0,
            cache_read_tokens=getattr(usage_obj, "cache_read_input_tokens", 0) or 0,
            cache_write_tokens=getattr(usage_obj, "cache_creation_input_tokens", 0) or 0,
        )
        message = AssistantMessage(
            text="\n\n".join(t for t in texts if t),
            tool_calls=calls,
            provider=self.name,
            provider_data=content,
        )
        return LLMResponse(message=message, stop_reason=stop, usage=usage, stop_detail=detail)