any_llm.py

# Inspired by https://github.com/andrewyng/aisuite/tree/main/aisuite
from __future__ import annotations

import importlib
import os
import warnings
from abc import ABC, abstractmethod
from typing import IO, TYPE_CHECKING, Any, ClassVar, Literal, TypeVar, cast, overload

from openresponses_types import ResponseResource
from pydantic import BaseModel

from any_llm.constants import INSIDE_NOTEBOOK, LLMProvider
from any_llm.exceptions import (
    ContentFilterFinishReasonError,
    LengthFinishReasonError,
    MissingApiKeyError,
    UnsupportedProviderError,
)
from any_llm.tools import prepare_tools
from any_llm.types.audio import AudioSpeechParams, AudioTranscriptionParams, Transcription
from any_llm.types.completion import (
    ChatCompletion,
    ChatCompletionMessage,
    CompletionParams,
    ParsedChatCompletion,
    ReasoningEffort,
)
from any_llm.types.image import ImageGenerationParams, ImagesResponse
from any_llm.types.messages import (
    ContentBlockStopEvent,
    MessageDelta,
    MessageDeltaEvent,
    MessageDeltaUsage,
    MessageResponse,
    MessagesParams,
    MessageStopEvent,
    MessageStreamEvent,
    ParsedMessage,
    StopReason,
)
from any_llm.types.provider import ProviderMetadata
from any_llm.types.responses import (
    ParsedResponse,
    Response,
    ResponseInputParam,
    ResponsesParams,
    ResponseStreamEvent,
)
from any_llm.utils.aio import async_coro_to_sync_iter, async_iter_to_sync_iter, run_async_in_sync
from any_llm.utils.exception_handler import handle_exceptions
from any_llm.utils.structured_output import (
    build_parsed_message,
    is_structured_output_type,
    parse_json_content,
    parse_responses_output,
)

ResponseFormatT = TypeVar("ResponseFormatT", bound=BaseModel)

if TYPE_CHECKING:
    from collections.abc import AsyncIterator, Callable, Coroutine, Iterator, Sequence

    from any_llm.types.batch import Batch, BatchResult
    from any_llm.types.completion import ChatCompletionChunk, CreateEmbeddingResponse
    from any_llm.types.model import Model
    from any_llm.types.moderation import ModerationResponse
    from any_llm.types.rerank import RerankResponse


class AnyLLM(ABC):
    """Provider for the LLM."""

    # === Provider-specific configuration (to be overridden by subclasses) ===
    PROVIDER_NAME: str
    """Must match the name of the provider directory  (case sensitive)"""

    PROVIDER_DOCUMENTATION_URL: str
    """Link to the provider's documentation"""

    ENV_API_KEY_NAME: str
    """Environment variable name for the API key"""

    ENV_API_BASE_NAME: str | None = None
    """Environment variable name for the API base URL. Optional."""

    # === Feature support flags (to be set by subclasses) ===
    SUPPORTS_COMPLETION_STREAMING: bool
    """OpenAI Streaming Completion API"""

    SUPPORTS_COMPLETION: bool
    """OpenAI Completion API"""

    SUPPORTS_COMPLETION_REASONING: bool
    """Reasoning Content attached to Completion API Response"""

    SUPPORTS_COMPLETION_IMAGE: bool
    """Image Support for Completion API"""

    SUPPORTS_COMPLETION_PDF: bool
    """PDF Support for Completion API"""

    SUPPORTS_EMBEDDING: bool
    """OpenAI Embedding API"""

    SUPPORTS_MODERATION: bool = False
    """OpenAI-compatible moderation API."""

    SUPPORTS_RESPONSES: bool
    """OpenAI Responses API"""

    SUPPORTS_LIST_MODELS: bool
    """OpenAI Models API"""

    SUPPORTS_BATCH: bool
    """OpenAI Batch Completion API"""

    SUPPORTS_IMAGE_GENERATION: bool = False
    """Image Generation API (e.g., OpenAI DALL-E)"""

    SUPPORTS_AUDIO_TRANSCRIPTION: bool = False
    """Audio Transcription API (e.g., OpenAI Whisper)"""

    SUPPORTS_AUDIO_SPEECH: bool = False
    """Audio Speech / TTS API (e.g., OpenAI TTS)"""

    SUPPORTS_RERANK: bool = False
    """Rerank API - reorder documents by relevance to a query."""

    SUPPORTS_MESSAGES: bool = True
    """Anthropic Messages API (all providers support it via conversion)"""

    API_BASE: str | None = None
    """This is used to set the API base for the provider.
    It is not required but may prove useful for providers that have overridable api bases.
    """

    # === Internal Flag Checks ===
    MISSING_PACKAGES_ERROR: ImportError | None = None
    """Some providers use SDKs that are not installed by default.
    This flag is used to check if the packages are installed before instantiating the provider.
    """

    BUILT_IN_TOOLS: ClassVar[list[Any] | None] = None
    """Some providers have built-in tools that can be used as-is without conversion.
    This should be a list of the allowed built-in tool instances.
    For example, in `gemini` provider, this could include `google.genai.types.Tool`.
    """

    def __init__(self, api_key: str | None = None, api_base: str | None = None, **kwargs: Any) -> None:
        self._verify_no_missing_packages()
        self._init_client(
            api_key=self._verify_and_set_api_key(api_key),
            api_base=self._resolve_api_base(api_base),
            **kwargs,
        )

    def _verify_no_missing_packages(self) -> None:
        if self.MISSING_PACKAGES_ERROR is not None:
            msg = f"{self.PROVIDER_NAME} required packages are not installed. Please install them with `pip install any-llm-sdk[{self.PROVIDER_NAME}]`. Specific error message: {self.MISSING_PACKAGES_ERROR}"
            raise ImportError(msg) from self.MISSING_PACKAGES_ERROR

    def _verify_and_set_api_key(self, api_key: str | None = None) -> str | None:
        # Standardized API key handling. Splitting into its own function so that providers
        # can easily override this method if they don't want verification (for instance, LMStudio)
        if not api_key:
            api_key = os.getenv(self.ENV_API_KEY_NAME)

        if not api_key:
            raise MissingApiKeyError(self.PROVIDER_NAME, self.ENV_API_KEY_NAME)
        return api_key

    def _resolve_api_base(self, api_base: str | None = None) -> str | None:
        """Resolve API base URL from parameter or environment variable.

        Resolution order:
        1. Explicit api_base parameter (if provided)
        2. Environment variable (if ENV_API_BASE_NAME is defined and set)
        3. None (allowing _init_client to use API_BASE class default)
        """
        if api_base:
            return api_base
        if self.ENV_API_BASE_NAME:
            return os.getenv(self.ENV_API_BASE_NAME)
        return None

    @classmethod
    def create(
        cls, provider: str | LLMProvider, api_key: str | None = None, api_base: str | None = None, **kwargs: Any
    ) -> AnyLLM:
        """Create a provider instance using the given provider name and config.

        Args:
            provider: The provider name (e.g., 'openai', 'anthropic')
            api_key: API key for the provider
            api_base: Base URL for the provider API
            **kwargs: Additional provider-specific arguments

        Returns:
            Provider instance for the specified provider

        """
        return cls._create_provider(provider, api_key=api_key, api_base=api_base, **kwargs)

    @classmethod
    def _create_provider(
        cls, provider_key: str | LLMProvider, api_key: str | None = None, api_base: str | None = None, **kwargs: Any
    ) -> AnyLLM:
        """Dynamically load and create an instance of a provider based on the naming convention."""
        provider_key = LLMProvider.from_string(provider_key).value

        provider_class_name = f"{provider_key.capitalize()}Provider"
        provider_module_name = f"{provider_key}"

        module_path = f"any_llm.providers.{provider_module_name}"

        try:
            module = importlib.import_module(module_path)
        except ImportError as e:
            msg = f"Could not import module {module_path}: {e!s}. Please ensure the provider is supported by doing AnyLLM.get_supported_providers()"
            raise ImportError(msg) from e

        provider_class: type[AnyLLM] = getattr(module, provider_class_name)

        return provider_class(api_key=api_key, api_base=api_base, **kwargs)

    @classmethod
    def get_provider_class(cls, provider_key: str | LLMProvider) -> type[AnyLLM]:
        """Get the provider class without instantiating it.

        Args:
            provider_key: The provider key (e.g., 'anthropic', 'openai')

        Returns:
            The provider class

        """
        provider_key = LLMProvider.from_string(provider_key).value

        provider_class_name = f"{provider_key.capitalize()}Provider"
        provider_module_name = f"{provider_key}"

        module_path = f"any_llm.providers.{provider_module_name}"

        try:
            module = importlib.import_module(module_path)
        except ImportError as e:
            msg = f"Could not import module {module_path}: {e!s}. Please ensure the provider is supported by doing AnyLLM.get_supported_providers()"
            raise ImportError(msg) from e

        provider_class: type[AnyLLM] = getattr(module, provider_class_name)
        return provider_class

    @classmethod
    def get_supported_providers(cls) -> list[str]:
        """Get a list of supported provider keys."""
        return [provider.value for provider in LLMProvider]

    @classmethod
    def get_all_provider_metadata(cls) -> list[ProviderMetadata]:
        """Get metadata for all supported providers.

        Returns:
            List of dictionaries containing provider metadata

        """
        providers: list[ProviderMetadata] = []
        for provider_key in cls.get_supported_providers():
            provider_class = cls.get_provider_class(provider_key)
            metadata = provider_class.get_provider_metadata()
            providers.append(metadata)

        # Sort providers by name
        providers.sort(key=lambda x: x.name)
        return providers

    @classmethod
    def get_provider_enum(cls, provider_key: str) -> LLMProvider:
        """Convert a string provider key to a ProviderName enum."""
        try:
            return LLMProvider(provider_key)
        except ValueError as e:
            supported = [provider.value for provider in LLMProvider]
            raise UnsupportedProviderError(provider_key, supported) from e

    @classmethod
    def split_model_provider(cls, model: str) -> tuple[LLMProvider, str]:
        """Extract the provider key from the model identifier.

        Supports both new format 'provider:model' (e.g., 'mistral:mistral-small')
        and legacy format 'provider/model' (e.g., 'mistral/mistral-small').

        The legacy format will be deprecated in version 1.0.
        """
        colon_index = model.find(":")
        slash_index = model.find("/")

        # Determine which delimiter comes first
        if colon_index != -1 and (slash_index == -1 or colon_index < slash_index):
            # The colon came first, so it's using the new syntax.
            provider, model_name = model.split(":", 1)
        elif slash_index != -1:
            # Slash comes first, so it's the legacy syntax
            warnings.warn(
                f"Model format 'provider/model' is deprecated and will be removed in version 1.0. "
                f"Please use 'provider:model' format instead. Got: '{model}'",
                DeprecationWarning,
                stacklevel=3,
            )
            provider, model_name = model.split("/", 1)
        else:
            msg = f"Invalid model format. Expected 'provider:model' or 'provider/model', got '{model}'"
            raise ValueError(msg)

        if not provider or not model_name:
            msg = f"Invalid model format. Expected 'provider:model' or 'provider/model', got '{model}'"
            raise ValueError(msg)
        return cls.get_provider_enum(provider), model_name

    @staticmethod
    @abstractmethod
    def _convert_completion_params(params: CompletionParams, **kwargs: Any) -> dict[str, Any]:
        msg = "Subclasses must implement this method"
        raise NotImplementedError(msg)

    @staticmethod
    @abstractmethod
    def _convert_completion_response(response: Any) -> ChatCompletion:
        msg = "Subclasses must implement this method"
        raise NotImplementedError(msg)

    @staticmethod
    @abstractmethod
    def _convert_completion_chunk_response(response: Any, **kwargs: Any) -> ChatCompletionChunk:
        msg = "Subclasses must implement this method"
        raise NotImplementedError(msg)

    @staticmethod
    @abstractmethod
    def _convert_embedding_params(params: Any, **kwargs: Any) -> dict[str, Any]:
        msg = "Subclasses must implement this method"
        raise NotImplementedError(msg)

    @staticmethod
    @abstractmethod
    def _convert_embedding_response(response: Any) -> CreateEmbeddingResponse:
        msg = "Subclasses must implement this method"
        raise NotImplementedError(msg)

    @staticmethod
    @abstractmethod
    def _convert_list_models_response(response: Any) -> Sequence[Model]:
        msg = "Subclasses must implement this method"
        raise NotImplementedError(msg)

    @staticmethod
    def _convert_rerank_params(model: str, query: str, documents: list[str], **kwargs: Any) -> dict[str, Any]:
        msg = "Subclasses must implement this method"
        raise NotImplementedError(msg)

    @staticmethod
    def _convert_rerank_response(response: Any) -> RerankResponse:
        msg = "Subclasses must implement this method"
        raise NotImplementedError(msg)

    @classmethod
    def get_provider_metadata(cls) -> ProviderMetadata:
        """Get provider metadata without requiring instantiation.

        Returns:
            Dictionary containing provider metadata including name, environment variable,
            documentation URL, and class name.

        """
        return ProviderMetadata(
            name=cls.PROVIDER_NAME,
            env_key=cls.ENV_API_KEY_NAME,
            env_api_base=cls.ENV_API_BASE_NAME,
            doc_url=cls.PROVIDER_DOCUMENTATION_URL,
            streaming=cls.SUPPORTS_COMPLETION_STREAMING,
            reasoning=cls.SUPPORTS_COMPLETION_REASONING,
            completion=cls.SUPPORTS_COMPLETION,
            image=cls.SUPPORTS_COMPLETION_IMAGE,
            pdf=cls.SUPPORTS_COMPLETION_PDF,
            embedding=cls.SUPPORTS_EMBEDDING,
            moderation=cls.SUPPORTS_MODERATION,
            responses=cls.SUPPORTS_RESPONSES,
            list_models=cls.SUPPORTS_LIST_MODELS,
            batch_completion=cls.SUPPORTS_BATCH,
            image_generation=cls.SUPPORTS_IMAGE_GENERATION,
            audio_transcription=cls.SUPPORTS_AUDIO_TRANSCRIPTION,
            audio_speech=cls.SUPPORTS_AUDIO_SPEECH,
            rerank=cls.SUPPORTS_RERANK,
            messages=cls.SUPPORTS_MESSAGES,
            class_name=cls.__name__,
        )

    @abstractmethod
    def _init_client(self, api_key: str | None = None, api_base: str | None = None, **kwargs: Any) -> None:
        msg = "Subclasses must implement this method"
        raise NotImplementedError(msg)

    # Overloads let type checkers narrow the return type based on response_format and stream.
    # The implementation only declares these discriminating params; everything else
    # passes through via **kwargs to acompletion().
    @overload
    def completion(
        self,
        model: str,
        messages: list[dict[str, Any] | ChatCompletionMessage],
        *,
        response_format: type[ResponseFormatT],
        stream: Literal[False] | None = ...,
        **kwargs: Any,
    ) -> ParsedChatCompletion[ResponseFormatT]: ...

    @overload
    def completion(
        self,
        model: str,
        messages: list[dict[str, Any] | ChatCompletionMessage],
        *,
        stream: Literal[True],
        **kwargs: Any,
    ) -> Iterator[ChatCompletionChunk]: ...

    @overload
    def completion(
        self,
        model: str,
        messages: list[dict[str, Any] | ChatCompletionMessage],
        *,
        response_format: dict[str, Any] | None = ...,
        stream: Literal[False] | None = ...,
        **kwargs: Any,
    ) -> ChatCompletion: ...

    @overload
    def completion(
        self,
        model: str,
        messages: list[dict[str, Any] | ChatCompletionMessage],
        *,
        response_format: dict[str, Any] | type | None = ...,
        stream: bool | None = ...,
        **kwargs: Any,
    ) -> ChatCompletion | Iterator[ChatCompletionChunk] | ParsedChatCompletion[Any]: ...

    def completion(
        self,
        model: str,
        messages: list[dict[str, Any] | ChatCompletionMessage],
        *,
        response_format: dict[str, Any] | type | None = None,
        stream: bool | None = None,
        allow_running_loop: bool | None = None,
        **kwargs: Any,
    ) -> ChatCompletion | Iterator[ChatCompletionChunk] | ParsedChatCompletion[Any]:
        """Create a chat completion synchronously.

        See [AnyLLM.acompletion][any_llm.any_llm.AnyLLM.acompletion]
        """
        if allow_running_loop is None:
            allow_running_loop = INSIDE_NOTEBOOK
        if stream:
            return async_coro_to_sync_iter(
                self.acompletion(
                    model=model,
                    messages=messages,
                    response_format=response_format,
                    stream=stream,
                    **kwargs,
                ),
                allow_running_loop=allow_running_loop,
            )

        response = run_async_in_sync(
            self.acompletion(model=model, messages=messages, response_format=response_format, stream=stream, **kwargs),
            allow_running_loop=allow_running_loop,
        )
        if isinstance(response, ChatCompletion):
            return response

        return async_iter_to_sync_iter(response, allow_running_loop=allow_running_loop)

    # Overloads let type checkers narrow the return type based on response_format and stream.
    @overload
    async def acompletion(
        self,
        model: str,
        messages: list[dict[str, Any] | ChatCompletionMessage],
        *,
        response_format: type[ResponseFormatT],
        stream: Literal[False] | None = ...,
        **kwargs: Any,
    ) -> ParsedChatCompletion[ResponseFormatT]: ...

    @overload
    async def acompletion(
        self,
        model: str,
        messages: list[dict[str, Any] | ChatCompletionMessage],
        *,
        stream: Literal[True],
        **kwargs: Any,
    ) -> AsyncIterator[ChatCompletionChunk]: ...

    @overload
    async def acompletion(
        self,
        model: str,
        messages: list[dict[str, Any] | ChatCompletionMessage],
        *,
        response_format: dict[str, Any] | None = ...,
        stream: Literal[False] | None = ...,
        **kwargs: Any,
    ) -> ChatCompletion: ...

    @overload
    async def acompletion(
        self,
        model: str,
        messages: list[dict[str, Any] | ChatCompletionMessage],
        *,
        response_format: dict[str, Any] | type | None = ...,
        stream: bool | None = ...,
        **kwargs: Any,
    ) -> ChatCompletion | AsyncIterator[ChatCompletionChunk] | ParsedChatCompletion[Any]: ...

    @handle_exceptions(wrap_streaming=True)
    async def acompletion(
        self,
        model: str,
        messages: list[dict[str, Any] | ChatCompletionMessage],
        *,
        tools: list[dict[str, Any] | Callable[..., Any]] | Any | None = None,
        tool_choice: str | dict[str, Any] | None = None,
        temperature: float | None = None,
        top_p: float | None = None,
        max_tokens: int | None = None,
        response_format: dict[str, Any] | type | None = None,
        stream: bool | None = None,
        n: int | None = None,
        stop: str | list[str] | None = None,
        presence_penalty: float | None = None,
        frequency_penalty: float | None = None,
        seed: int | None = None,
        user: str | None = None,
        session_label: str | None = None,
        parallel_tool_calls: bool | None = None,
        logprobs: bool | None = None,
        top_logprobs: int | None = None,
        logit_bias: dict[str, float] | None = None,
        stream_options: dict[str, Any] | None = None,
        max_completion_tokens: int | None = None,
        reasoning_effort: ReasoningEffort | None = "auto",
        **kwargs: Any,
    ) -> ChatCompletion | AsyncIterator[ChatCompletionChunk] | ParsedChatCompletion[Any]:
        """Create a chat completion asynchronously.

        Args:
            model: Model identifier for the chosen provider (e.g., model='gpt-4.1-mini' for LLMProvider.OPENAI).
            messages: List of messages for the conversation
            tools: List of tools for tool calling. Can be Python callables or OpenAI tool format dicts
            tool_choice: Controls which tools the model can call
            temperature: Controls randomness in the response (0.0 to 2.0)
            top_p: Controls diversity via nucleus sampling (0.0 to 1.0)
            max_tokens: Maximum number of tokens to generate
            response_format: Format specification for the response
            stream: Whether to stream the response
            n: Number of completions to generate
            stop: Stop sequences for generation
            presence_penalty: Penalize new tokens based on presence in text
            frequency_penalty: Penalize new tokens based on frequency in text
            seed: Random seed for reproducible results
            user: Unique identifier for the end user
            session_label: Deprecated, no longer used. Previously used for platform traces.
            parallel_tool_calls: Whether to allow parallel tool calls
            logprobs: Include token-level log probabilities in the response
            top_logprobs: Number of alternatives to return when logprobs are requested
            logit_bias: Bias the likelihood of specified tokens during generation
            stream_options: Additional options controlling streaming behavior
            max_completion_tokens: Maximum number of tokens for the completion
            reasoning_effort: Reasoning effort level for models that support it. "auto" will map to each provider's default.
            **kwargs: Additional provider-specific arguments that will be passed to the provider's API call.

        Returns:
            The completion response from the provider

        """
        prepared_tools = None
        if tools:
            prepared_tools = prepare_tools(tools, built_in_tools=self.BUILT_IN_TOOLS)

        processed_messages: list[dict[str, Any]] = []
        for message in messages:
            if isinstance(message, ChatCompletionMessage):
                # Dump the message but exclude the extra field that we extend from OpenAI Spec
                processed_messages.append(message.model_dump(exclude_none=True, exclude={"reasoning"}))
            else:
                processed_messages.append(message)

        params = CompletionParams(
            model_id=model,
            messages=processed_messages,
            tools=prepared_tools,
            tool_choice=tool_choice,
            temperature=temperature,
            top_p=top_p,
            max_tokens=max_tokens,
            response_format=response_format,
            stream=stream,
            n=n,
            stop=stop,
            presence_penalty=presence_penalty,
            frequency_penalty=frequency_penalty,
            seed=seed,
            user=user,
            parallel_tool_calls=parallel_tool_calls,
            logprobs=logprobs,
            top_logprobs=top_logprobs,
            logit_bias=logit_bias,
            stream_options=stream_options,
            max_completion_tokens=max_completion_tokens,
            reasoning_effort=reasoning_effort,
        )

        result = await self._acompletion(params, **kwargs)

        if is_structured_output_type(response_format):
            if isinstance(result, ParsedChatCompletion):
                parsed_completion = result
            elif isinstance(result, ChatCompletion):
                parsed_completion = ParsedChatCompletion.model_validate(result, from_attributes=True)
            else:
                return result

            for choice in parsed_completion.choices:
                if choice.message.parsed is not None:
                    continue
                if choice.finish_reason == "length":
                    raise LengthFinishReasonError(completion=parsed_completion)
                if choice.finish_reason == "content_filter":
                    raise ContentFilterFinishReasonError(completion=parsed_completion)
                if choice.message.content and not choice.message.refusal:
                    choice.message.parsed = parse_json_content(response_format, choice.message.content)
            return parsed_completion

        return result

    async def _acompletion(
        self, params: CompletionParams, **kwargs: Any
    ) -> ChatCompletion | AsyncIterator[ChatCompletionChunk]:
        if not self.SUPPORTS_COMPLETION:
            msg = "Provider doesn't support completion."
            raise NotImplementedError(msg)
        msg = "Subclasses must implement _acompletion method"
        raise NotImplementedError(msg)

    def messages(
        self,
        *,
        allow_running_loop: bool | None = None,
        **kwargs: Any,
    ) -> MessageResponse | ParsedMessage[Any] | Iterator[MessageStreamEvent]:
        """Create a message using the Anthropic Messages API synchronously.

        See [AnyLLM.amessages][any_llm.any_llm.AnyLLM.amessages]
        """
        if allow_running_loop is None:
            allow_running_loop = INSIDE_NOTEBOOK
        response = run_async_in_sync(self.amessages(**kwargs), allow_running_loop=allow_running_loop)
        if isinstance(response, (MessageResponse, ParsedMessage)):
            return response
        return async_iter_to_sync_iter(response)

    @handle_exceptions(wrap_streaming=True)
    async def amessages(
        self,
        model: str,
        messages: list[dict[str, Any]],
        max_tokens: int,
        *,
        system: str | list[dict[str, Any]] | None = None,
        temperature: float | None = None,
        top_p: float | None = None,
        top_k: int | None = None,
        stream: bool | None = None,
        stop_sequences: list[str] | None = None,
        tools: list[dict[str, Any]] | None = None,
        tool_choice: dict[str, Any] | None = None,
        metadata: dict[str, Any] | None = None,
        thinking: dict[str, Any] | None = None,
        cache_control: dict[str, Any] | None = None,
        output_format: type | dict[str, Any] | None = None,
        **kwargs: Any,
    ) -> MessageResponse | ParsedMessage[Any] | AsyncIterator[MessageStreamEvent]:
        """Create a message using the Anthropic Messages API asynchronously.

        All providers support this via automatic conversion to/from Chat Completions.
        The Anthropic provider uses a native pass-through for efficiency.

        Args:
            model: Model identifier for the chosen provider.
            messages: List of messages in Anthropic format.
            max_tokens: Maximum number of tokens to generate.
            system: System prompt (string or list of content blocks with optional cache_control).
            temperature: Controls randomness (0.0 to 1.0).
            top_p: Controls diversity via nucleus sampling.
            top_k: Only sample from the top K options.
            stream: Whether to stream the response.
            stop_sequences: Custom stop sequences.
            tools: List of tools in Anthropic format.
            tool_choice: Controls which tool the model uses.
            metadata: Request metadata.
            thinking: Thinking/reasoning configuration.
            cache_control: Cache control configuration for prompt caching.
            output_format: Structured output, mirroring Anthropic's ``messages.parse``/
                ``output_config``. Either a Pydantic ``BaseModel``/dataclass **type** (typed
                ``parsed_output``) or a raw Anthropic ``output_config`` **dict** for non-Pydantic
                JSON schemas (``parsed_output`` holds the parsed JSON). The call returns
                Anthropic's ``ParsedMessage``. Not supported with ``stream=True``.
            **kwargs: Additional provider-specific arguments.

        Returns:
            MessageResponse (or ParsedMessage when `output_format` is given), or an async
            iterator of MessageStreamEvent (if streaming).

        Raises:
            ValueError: If `output_format` is combined with `stream=True`.

        """
        if output_format is not None and stream:
            msg = "stream is not supported for output_format"
            raise ValueError(msg)

        params = MessagesParams(
            model=model,
            messages=messages,
            max_tokens=max_tokens,
            system=system,
            temperature=temperature,
            top_p=top_p,
            top_k=top_k,
            stream=stream,
            stop_sequences=stop_sequences,
            tools=tools,
            tool_choice=tool_choice,
            metadata=metadata,
            thinking=thinking,
            cache_control=cache_control,
            output_format=output_format,
        )
        result = await self._amessages(params, **kwargs)

        # The Anthropic provider already returns a ParsedMessage via native messages.parse (typed
        # case); for the raw-dict case and for all bridged providers it returns a MessageResponse,
        # so build the same ParsedMessage shape from the response's JSON text here.
        if output_format is not None and isinstance(result, MessageResponse):
            return build_parsed_message(result, output_format)

        return result

    async def _amessages(
        self, params: MessagesParams, **kwargs: Any
    ) -> MessageResponse | ParsedMessage[Any] | AsyncIterator[MessageStreamEvent]:
        """Default implementation: converts Messages ↔ Completions format.

        Providers with native Messages API support (e.g., Anthropic) override this
        for direct pass-through.
        """
        from any_llm.types.completion import CompletionParams
        from any_llm.utils.messages_compat import (
            StreamingState,
            chat_completion_chunk_to_message_stream_events,
            chat_completion_to_message_response,
            messages_params_to_completion_params,
        )

        completion_kwargs = messages_params_to_completion_params(params)
        completion_params = CompletionParams(**completion_kwargs)
        result = await self._acompletion(completion_params, **kwargs)

        if isinstance(result, ChatCompletion):
            return chat_completion_to_message_response(result)

        async def convert_stream() -> AsyncIterator[MessageStreamEvent]:
            state = StreamingState()

            def usage_delta(stop_reason: StopReason | None) -> MessageDeltaEvent:
                return MessageDeltaEvent(
                    type="message_delta",
                    delta=MessageDelta(stop_reason=stop_reason),
                    usage=MessageDeltaUsage(
                        output_tokens=state.output_tokens,
                        input_tokens=state.input_tokens,
                        cache_read_input_tokens=state.cache_read_input_tokens or None,
                    ),
                )

            try:
                async for chunk in result:
                    for event in chat_completion_chunk_to_message_stream_events(chunk, state):
                        yield event
            except Exception:
                # Flush the usage accumulated so far before re-raising, so a mid-stream failure still reports tokens.
                if state.started:
                    yield usage_delta(state.stop_reason)
                raise
            # Emit the closing events after the full stream is consumed so trailing-chunk usage is included.
            if state.started:
                if state.current_block_type is not None:
                    yield ContentBlockStopEvent(
                        type="content_block_stop",
                        index=state.current_block_index,
                    )
                yield usage_delta(state.stop_reason or "end_turn")
                yield MessageStopEvent(type="message_stop")

        return convert_stream()

    # Overloads let type checkers narrow the return type based on response_format and stream.
    @overload
    def responses(
        self,
        model: str,
        input_data: str | ResponseInputParam,
        *,
        response_format: type[ResponseFormatT],
        stream: Literal[False] | None = ...,
        **kwargs: Any,
    ) -> ParsedResponse[ResponseFormatT]: ...

    @overload
    def responses(
        self,
        model: str,
        input_data: str | ResponseInputParam,
        *,
        stream: Literal[True],
        **kwargs: Any,
    ) -> Iterator[ResponseStreamEvent]: ...

    @overload
    def responses(
        self,
        model: str,
        input_data: str | ResponseInputParam,
        *,
        response_format: dict[str, Any] | None = ...,
        stream: Literal[False] | None = ...,
        **kwargs: Any,
    ) -> ResponseResource | Response: ...

    @overload
    def responses(
        self,
        model: str,
        input_data: str | ResponseInputParam,
        *,
        response_format: dict[str, Any] | type | None = ...,
        stream: bool | None = ...,
        **kwargs: Any,
    ) -> ResponseResource | Response | Iterator[ResponseStreamEvent] | ParsedResponse[Any]: ...

    def responses(
        self, model: str, input_data: str | ResponseInputParam, **kwargs: Any
    ) -> ResponseResource | Response | Iterator[ResponseStreamEvent] | ParsedResponse[Any]:
        """Create a response synchronously.

        See [AnyLLM.aresponses][any_llm.any_llm.AnyLLM.aresponses]
        """
        allow_running_loop = kwargs.pop("allow_running_loop", INSIDE_NOTEBOOK)
        if kwargs.get("stream"):
            return async_coro_to_sync_iter(
                cast(
                    "Coroutine[Any, Any, AsyncIterator[ResponseStreamEvent]]",
                    self.aresponses(model, input_data, **kwargs),
                ),
                allow_running_loop=allow_running_loop,
            )

        response = run_async_in_sync(
            self.aresponses(model, input_data, **kwargs), allow_running_loop=allow_running_loop
        )
        # ParsedResponse (structured output) is a subclass of Response, so it is covered here.
        if isinstance(response, (ResponseResource, Response)):
            return response
        return async_iter_to_sync_iter(response, allow_running_loop=allow_running_loop)

    # Overloads let type checkers narrow the return type based on response_format and stream.
    @overload
    async def aresponses(
        self,
        model: str,
        input_data: str | ResponseInputParam,
        *,
        response_format: type[ResponseFormatT],
        stream: Literal[False] | None = ...,
        **kwargs: Any,
    ) -> ParsedResponse[ResponseFormatT]: ...

    @overload
    async def aresponses(
        self,
        model: str,
        input_data: str | ResponseInputParam,
        *,
        stream: Literal[True],
        **kwargs: Any,
    ) -> AsyncIterator[ResponseStreamEvent]: ...

    @overload
    async def aresponses(
        self,
        model: str,
        input_data: str | ResponseInputParam,
        *,
        response_format: dict[str, Any] | None = ...,
        stream: Literal[False] | None = ...,
        **kwargs: Any,
    ) -> ResponseResource | Response: ...

    @overload
    async def aresponses(
        self,
        model: str,
        input_data: str | ResponseInputParam,
        *,
        response_format: dict[str, Any] | type | None = ...,
        stream: bool | None = ...,
        **kwargs: Any,
    ) -> ResponseResource | Response | AsyncIterator[ResponseStreamEvent] | ParsedResponse[Any]: ...

    @handle_exceptions(wrap_streaming=True)
    async def aresponses(
        self,
        model: str,
        input_data: str | ResponseInputParam,
        *,
        tools: list[dict[str, Any] | Callable[..., Any]] | Any | None = None,
        tool_choice: str | dict[str, Any] | None = None,
        max_output_tokens: int | None = None,
        temperature: float | None = None,
        top_p: float | None = None,
        stream: bool | None = None,
        instructions: str | None = None,
        max_tool_calls: int | None = None,
        parallel_tool_calls: bool | None = None,
        reasoning: Any | None = None,
        text: Any | None = None,
        response_format: dict[str, Any] | type | None = None,
        presence_penalty: float | None = None,
        frequency_penalty: float | None = None,
        truncation: str | None = None,
        store: bool | None = None,
        service_tier: str | None = None,
        user: str | None = None,
        metadata: dict[str, str] | None = None,
        previous_response_id: str | None = None,
        include: list[str] | None = None,
        background: bool | None = None,
        safety_identifier: str | None = None,
        prompt_cache_key: str | None = None,
        prompt_cache_retention: str | None = None,
        conversation: str | dict[str, Any] | None = None,
        **kwargs: Any,
    ) -> ResponseResource | Response | AsyncIterator[ResponseStreamEvent] | ParsedResponse[Any]:
        """Create a response using the OpenResponses API.

        This implements the OpenResponses specification and returns either
        `openresponses_types.ResponseResource` (for OpenResponses-compliant providers)
        or `openai.types.responses.Response` (for providers using OpenAI's native API).
        If `stream=True`, an iterator of streaming event dicts is returned.

        Args:
            model: Model identifier for the chosen provider (e.g., model='gpt-4.1-mini' for LLMProvider.OPENAI).
            input_data: The input payload accepted by provider's Responses API.
                For OpenAI-compatible providers, this is typically a list mixing
                text, images, and tool instructions, or a dict per OpenAI spec.
            tools: Optional tools for tool calling (Python callables or OpenAI tool dicts)
            tool_choice: Controls which tools the model can call
            max_output_tokens: Maximum number of output tokens to generate
            temperature: Controls randomness in the response (0.0 to 2.0)
            top_p: Controls diversity via nucleus sampling (0.0 to 1.0)
            stream: Whether to stream response events
            instructions: A system (or developer) message inserted into the model's context.
            max_tool_calls: The maximum number of total calls to built-in tools that can be processed in a response. This maximum number applies across all built-in tool calls, not per individual tool. Any further attempts to call a tool by the model will be ignored.
            parallel_tool_calls: Whether to allow the model to run tool calls in parallel.
            reasoning: Configuration options for reasoning models.
            text: Configuration options for a text response from the model. Can be plain text or structured JSON data.
            response_format: Structured-output type. When a Pydantic ``BaseModel`` or dataclass is passed, the
                response is parsed and returned as a ``ParsedResponse`` whose ``output_parsed`` holds the typed
                object (the Responses-API analogue of ``client.responses.parse``). A raw OpenAI ``text.format``
                dict is also accepted and passed through unparsed.
            presence_penalty: Penalizes new tokens based on whether they appear in the text so far.
            frequency_penalty: Penalizes new tokens based on their frequency in the text so far.
            truncation: Controls how the service truncates input when it exceeds the model context window.
            store: Whether to store the response so it can be retrieved later.
            service_tier: The service tier to use for this request.
            user: A unique identifier representing your end user.
            metadata: Key-value pairs for custom metadata (up to 16 pairs).
            previous_response_id: The ID of the response to use as the prior turn for this request.
            include: Items to include in the response (e.g., 'reasoning.encrypted_content').
            background: Whether to run the request in the background and return immediately.
            safety_identifier: A stable identifier used for safety monitoring and abuse detection.
            prompt_cache_key: A key to use when reading from or writing to the prompt cache.
            prompt_cache_retention: How long to retain a prompt cache entry created by this request.
            conversation: The conversation to associate this response with (ID string or ConversationParam object).
            **kwargs: Additional provider-specific arguments that will be passed to the provider's API call.

        Returns:
            Either a `ResponseResource` object (OpenResponses-compliant providers),
            a `Response` object (non-compliant providers), a `ParsedResponse` (when a
            structured `response_format` type is given), or an iterator of
            `ResponseStreamEvent` (streaming).

        Raises:
            NotImplementedError: If the selected provider does not support the Responses API.
            ValueError: If a structured `response_format` is combined with `stream=True`.

        """
        if is_structured_output_type(response_format) and stream:
            msg = "stream is not supported for response_format"
            raise ValueError(msg)

        prepared_tools = None
        if tools:
            prepared_tools = prepare_tools(tools, built_in_tools=self.BUILT_IN_TOOLS)

        params = ResponsesParams(
            model=model,
            input=input_data,
            tools=prepared_tools,
            tool_choice=tool_choice,
            max_output_tokens=max_output_tokens,
            temperature=temperature,
            top_p=top_p,
            stream=stream,
            instructions=instructions,
            max_tool_calls=max_tool_calls,
            parallel_tool_calls=parallel_tool_calls,
            reasoning=reasoning,
            text=text,
            response_format=response_format,
            presence_penalty=presence_penalty,
            frequency_penalty=frequency_penalty,
            truncation=truncation,
            store=store,
            service_tier=service_tier,
            user=user,
            metadata=metadata,
            previous_response_id=previous_response_id,
            include=include,
            background=background,
            safety_identifier=safety_identifier,
            prompt_cache_key=prompt_cache_key,
            prompt_cache_retention=prompt_cache_retention,
            conversation=conversation,
            **kwargs,
        )

        result = await self._aresponses(params)

        if is_structured_output_type(response_format):
            # OpenAI-SDK providers return a ParsedResponse directly (via responses.parse);
            # other providers return a raw Response/ResponseResource that we parse here.
            if isinstance(result, ParsedResponse):
                return result
            if isinstance(result, (Response, ResponseResource)):
                parsed = parse_responses_output(result, response_format)
                if parsed is not None:
                    return parsed

        return result

    async def _aresponses(
        self, params: ResponsesParams, **kwargs: Any
    ) -> ResponseResource | Response | ParsedResponse[Any] | AsyncIterator[ResponseStreamEvent]:
        if not self.SUPPORTS_RESPONSES:
            msg = "Provider doesn't support responses."
            raise NotImplementedError(msg)
        msg = "Subclasses must implement _aresponses method"
        raise NotImplementedError(msg)

    def _embedding(self, model: str, inputs: str | list[str], **kwargs: Any) -> CreateEmbeddingResponse:
        allow_running_loop = kwargs.pop("allow_running_loop", INSIDE_NOTEBOOK)
        return run_async_in_sync(self.aembedding(model, inputs, **kwargs), allow_running_loop=allow_running_loop)

    @handle_exceptions()
    async def aembedding(self, model: str, inputs: str | list[str], **kwargs: Any) -> CreateEmbeddingResponse:
        return await self._aembedding(model, inputs, **kwargs)

    async def _aembedding(self, model: str, inputs: str | list[str], **kwargs: Any) -> CreateEmbeddingResponse:
        if not self.SUPPORTS_EMBEDDING:
            msg = "Provider doesn't support embedding."
            raise NotImplementedError(msg)
        msg = "Subclasses must implement _aembedding method"
        raise NotImplementedError(msg)

    def _image_generation(self, model: str, prompt: str, **kwargs: Any) -> ImagesResponse:
        allow_running_loop = kwargs.pop("allow_running_loop", INSIDE_NOTEBOOK)
        return run_async_in_sync(self.aimage_generation(model, prompt, **kwargs), allow_running_loop=allow_running_loop)

    @handle_exceptions()
    async def aimage_generation(self, model: str, prompt: str, **kwargs: Any) -> ImagesResponse:
        """Create an image asynchronously.

        Args:
            model: Model identifier for the chosen provider (e.g., model='dall-e-3' for LLMProvider.OPENAI).
            prompt: A text description of the desired image(s).
            **kwargs: Additional parameters (n, size, quality, style, response_format, user).

        Returns:
            The image generation response from the provider.

        """
        params = ImageGenerationParams(model_id=model, prompt=prompt, **kwargs)
        return await self._aimage_generation(params)

    async def _aimage_generation(self, params: ImageGenerationParams, **kwargs: Any) -> ImagesResponse:
        if not self.SUPPORTS_IMAGE_GENERATION:
            msg = "Provider doesn't support image generation."
            raise NotImplementedError(msg)
        msg = "Subclasses must implement _aimage_generation method"
        raise NotImplementedError(msg)

    def _transcription(self, model: str, file: bytes | IO[bytes], **kwargs: Any) -> Transcription:
        """Transcribe audio synchronously.

        See [AnyLLM.atranscription][any_llm.any_llm.AnyLLM.atranscription]
        """
        allow_running_loop = kwargs.pop("allow_running_loop", INSIDE_NOTEBOOK)
        return run_async_in_sync(self.atranscription(model, file, **kwargs), allow_running_loop=allow_running_loop)

    @handle_exceptions()
    async def atranscription(self, model: str, file: bytes | IO[bytes], **kwargs: Any) -> Transcription:
        """Transcribe audio asynchronously.

        Args:
            model: Model identifier for the chosen provider (e.g., model='whisper-1' for LLMProvider.OPENAI).
            file: Audio file content as bytes or file-like object.
                Supported formats: flac, mp3, mp4, mpeg, mpga, m4a, ogg, wav, webm.
            **kwargs: Additional parameters (language, prompt, response_format, temperature, timestamp_granularities).

        Returns:
            The transcription response from the provider.

        """
        params = AudioTranscriptionParams(model_id=model, file=file, **kwargs)
        return await self._atranscription(params)

    async def _atranscription(self, params: AudioTranscriptionParams, **kwargs: Any) -> Transcription:
        if not self.SUPPORTS_AUDIO_TRANSCRIPTION:
            msg = "Provider doesn't support audio transcription."
            raise NotImplementedError(msg)
        msg = "Subclasses must implement _atranscription method"
        raise NotImplementedError(msg)

    def _speech(
        self,
        model: str,
        input: str,  # noqa: A002
        voice: str,
        **kwargs: Any,
    ) -> bytes:
        """Generate speech from text synchronously.

        See [AnyLLM.aspeech][any_llm.any_llm.AnyLLM.aspeech]
        """
        allow_running_loop = kwargs.pop("allow_running_loop", INSIDE_NOTEBOOK)
        return run_async_in_sync(self.aspeech(model, input, voice, **kwargs), allow_running_loop=allow_running_loop)

    @handle_exceptions()
    async def aspeech(self, model: str, input: str, voice: str, **kwargs: Any) -> bytes:  # noqa: A002
        """Generate speech from text asynchronously.

        Args:
            model: Model identifier for the chosen provider (e.g., model='tts-1' for LLMProvider.OPENAI).
            input: The text to generate audio for. Maximum 4096 characters.
            voice: The voice to use for generation (e.g., 'alloy', 'echo', 'shimmer').
            **kwargs: Additional parameters (instructions, response_format, speed).

        Returns:
            The generated audio content as bytes.

        """
        params = AudioSpeechParams(model_id=model, input=input, voice=voice, **kwargs)
        return await self._aspeech(params)

    async def _aspeech(self, params: AudioSpeechParams, **kwargs: Any) -> bytes:
        if not self.SUPPORTS_AUDIO_SPEECH:
            msg = "Provider doesn't support audio speech."
            raise NotImplementedError(msg)
        msg = "Subclasses must implement _aspeech method"
        raise NotImplementedError(msg)

    def _moderation(
        self,
        model: str,
        input: str | list[str] | list[dict[str, Any]],  # noqa: A002
        **kwargs: Any,
    ) -> ModerationResponse:
        """Run a moderation check synchronously.

        See [AnyLLM.amoderation][any_llm.any_llm.AnyLLM.amoderation]
        """
        allow_running_loop = kwargs.pop("allow_running_loop", INSIDE_NOTEBOOK)
        return run_async_in_sync(
            self.amoderation(model, input, **kwargs),
            allow_running_loop=allow_running_loop,
        )

    @handle_exceptions()
    async def amoderation(
        self,
        model: str,
        input: str | list[str] | list[dict[str, Any]],  # noqa: A002
        **kwargs: Any,
    ) -> ModerationResponse:
        """Run a moderation check asynchronously.

        Args:
            model: Provider-specific moderation model identifier.
            input: A string, list of strings, or list of OpenAI-style content-part
                dicts (for multimodal moderation).
            **kwargs: Additional provider-specific arguments. Pass
                ``include_raw=True`` to populate ``ModerationResult.provider_raw``.

        Returns:
            A ModerationResponse with one ``ModerationResult`` per input item.

        Raises:
            NotImplementedError: If the provider does not support moderation.
        """
        return await self._amoderation(model, input, **kwargs)

    async def _amoderation(
        self,
        model: str,
        input: str | list[str] | list[dict[str, Any]],  # noqa: A002
        **kwargs: Any,
    ) -> ModerationResponse:
        if not self.SUPPORTS_MODERATION:
            msg = f"Provider {self.PROVIDER_NAME} does not support moderation"
            raise NotImplementedError(msg)
        msg = "Subclasses must implement _amoderation method"
        raise NotImplementedError(msg)

    def _rerank(self, model: str, query: str, documents: list[str], **kwargs: Any) -> RerankResponse:
        allow_running_loop = kwargs.pop("allow_running_loop", INSIDE_NOTEBOOK)
        return run_async_in_sync(self.arerank(model, query, documents, **kwargs), allow_running_loop=allow_running_loop)

    @handle_exceptions()
    async def arerank(self, model: str, query: str, documents: list[str], **kwargs: Any) -> RerankResponse:
        """Rerank documents by relevance to a query.

        Args:
            model: Model identifier for the rerank model.
            query: The search query string.
            documents: List of document strings to rerank.
            **kwargs: Additional provider-specific arguments.

        Returns:
            RerankResponse with results sorted by relevance_score descending.

        Raises:
            NotImplementedError: If the provider does not support reranking.

        """
        return await self._arerank(model, query, documents, **kwargs)

    async def _arerank(self, model: str, query: str, documents: list[str], **kwargs: Any) -> RerankResponse:
        if not self.SUPPORTS_RERANK:
            msg = "Provider doesn't support rerank."
            raise NotImplementedError(msg)
        msg = "Subclasses must implement _arerank method"
        raise NotImplementedError(msg)

    def list_models(self, **kwargs: Any) -> Sequence[Model]:
        allow_running_loop = kwargs.pop("allow_running_loop", INSIDE_NOTEBOOK)
        return run_async_in_sync(self.alist_models(**kwargs), allow_running_loop=allow_running_loop)

    @handle_exceptions()
    async def alist_models(self, **kwargs: Any) -> Sequence[Model]:
        return await self._alist_models(**kwargs)

    async def _alist_models(self, **kwargs: Any) -> Sequence[Model]:
        if not self.SUPPORTS_LIST_MODELS:
            msg = "Provider doesn't support listing models."
            raise NotImplementedError(msg)
        msg = "Subclasses must implement _alist_models method"
        raise NotImplementedError(msg)

    def create_batch(self, **kwargs: Any) -> Batch:
        """Create a batch synchronously.

        See [AnyLLM.acreate_batch][any_llm.any_llm.AnyLLM.acreate_batch]
        """
        allow_running_loop = kwargs.pop("allow_running_loop", INSIDE_NOTEBOOK)
        return run_async_in_sync(self.acreate_batch(**kwargs), allow_running_loop=allow_running_loop)

    @handle_exceptions()
    async def acreate_batch(
        self,
        input_file_path: str,
        endpoint: str,
        completion_window: str = "24h",
        metadata: dict[str, str] | None = None,
        **kwargs: Any,
    ) -> Batch:
        """Create a batch job asynchronously.

        Args:
            input_file_path: Path to a local file containing batch requests in JSONL format.
            endpoint: The endpoint to be used for all requests (e.g., '/v1/chat/completions')
            completion_window: The time frame within which the batch should be processed (default: '24h')
            metadata: Optional custom metadata for the batch
            **kwargs: Additional provider-specific arguments

        Returns:
            The created batch object

        """
        return await self._acreate_batch(
            input_file_path=input_file_path,
            endpoint=endpoint,
            completion_window=completion_window,
            metadata=metadata,
            **kwargs,
        )

    async def _acreate_batch(
        self,
        input_file_path: str,
        endpoint: str,
        completion_window: str = "24h",
        metadata: dict[str, str] | None = None,
        **kwargs: Any,
    ) -> Batch:
        if not self.SUPPORTS_BATCH:
            msg = "Provider doesn't support batch completions."
            raise NotImplementedError(msg)
        msg = "Subclasses must implement _acreate_batch method"
        raise NotImplementedError(msg)

    def retrieve_batch(self, batch_id: str, **kwargs: Any) -> Batch:
        """Retrieve a batch synchronously.

        See [AnyLLM.aretrieve_batch][any_llm.any_llm.AnyLLM.aretrieve_batch]
        """
        allow_running_loop = kwargs.pop("allow_running_loop", INSIDE_NOTEBOOK)
        return run_async_in_sync(self.aretrieve_batch(batch_id, **kwargs), allow_running_loop=allow_running_loop)

    @handle_exceptions()
    async def aretrieve_batch(self, batch_id: str, **kwargs: Any) -> Batch:
        """Retrieve a batch job asynchronously.

        Args:
            batch_id: The ID of the batch to retrieve
            **kwargs: Additional provider-specific arguments

        Returns:
            The batch object

        """
        return await self._aretrieve_batch(batch_id, **kwargs)

    async def _aretrieve_batch(self, batch_id: str, **kwargs: Any) -> Batch:
        if not self.SUPPORTS_BATCH:
            msg = "Provider doesn't support batch completions."
            raise NotImplementedError(msg)
        msg = "Subclasses must implement _aretrieve_batch method"
        raise NotImplementedError(msg)

    def cancel_batch(self, batch_id: str, **kwargs: Any) -> Batch:
        """Cancel a batch synchronously.

        See [AnyLLM.acancel_batch][any_llm.any_llm.AnyLLM.acancel_batch]
        """
        allow_running_loop = kwargs.pop("allow_running_loop", INSIDE_NOTEBOOK)
        return run_async_in_sync(self.acancel_batch(batch_id, **kwargs), allow_running_loop=allow_running_loop)

    @handle_exceptions()
    async def acancel_batch(self, batch_id: str, **kwargs: Any) -> Batch:
        """Cancel a batch job asynchronously.

        Args:
            batch_id: The ID of the batch to cancel
            **kwargs: Additional provider-specific arguments

        Returns:
            The cancelled batch object

        """
        return await self._acancel_batch(batch_id, **kwargs)

    async def _acancel_batch(self, batch_id: str, **kwargs: Any) -> Batch:
        if not self.SUPPORTS_BATCH:
            msg = "Provider doesn't support batch completions."
            raise NotImplementedError(msg)
        msg = "Subclasses must implement _acancel_batch method"
        raise NotImplementedError(msg)

    def list_batches(
        self,
        after: str | None = None,
        limit: int | None = None,
        **kwargs: Any,
    ) -> Sequence[Batch]:
        """List batches synchronously.

        See [AnyLLM.alist_batches][any_llm.any_llm.AnyLLM.alist_batches]
        """
        allow_running_loop = kwargs.pop("allow_running_loop", INSIDE_NOTEBOOK)
        return run_async_in_sync(
            self.alist_batches(after=after, limit=limit, **kwargs), allow_running_loop=allow_running_loop
        )

    @handle_exceptions()
    async def alist_batches(
        self,
        after: str | None = None,
        limit: int | None = None,
        **kwargs: Any,
    ) -> Sequence[Batch]:
        """List batch jobs asynchronously.

        Args:
            after: A cursor for pagination. Returns batches after this batch ID.
            limit: Maximum number of batches to return (default: 20)
            **kwargs: Additional provider-specific arguments

        Returns:
            A list of batch objects

        """
        return await self._alist_batches(after=after, limit=limit, **kwargs)

    async def _alist_batches(
        self,
        after: str | None = None,
        limit: int | None = None,
        **kwargs: Any,
    ) -> Sequence[Batch]:
        if not self.SUPPORTS_BATCH:
            msg = "Provider doesn't support batch completions."
            raise NotImplementedError(msg)
        msg = "Subclasses must implement _alist_batches method"
        raise NotImplementedError(msg)

    def retrieve_batch_results(self, batch_id: str, **kwargs: Any) -> BatchResult:
        """Retrieve batch results synchronously.

        See [AnyLLM.aretrieve_batch_results][any_llm.any_llm.AnyLLM.aretrieve_batch_results]
        """
        allow_running_loop = kwargs.pop("allow_running_loop", INSIDE_NOTEBOOK)
        return run_async_in_sync(
            self.aretrieve_batch_results(batch_id, **kwargs),
            allow_running_loop=allow_running_loop,
        )

    @handle_exceptions()
    async def aretrieve_batch_results(self, batch_id: str, **kwargs: Any) -> BatchResult:
        """Retrieve the results of a completed batch job asynchronously.

        Args:
            batch_id: The ID of the batch to retrieve results for.
            **kwargs: Additional provider-specific arguments.

        Returns:
            The batch results containing per-request outcomes.

        Raises:
            BatchNotCompleteError: If the batch status is not 'completed'.

        """
        return await self._aretrieve_batch_results(batch_id, **kwargs)

    async def _aretrieve_batch_results(self, batch_id: str, **kwargs: Any) -> BatchResult:
        if not self.SUPPORTS_BATCH:
            msg = "Provider doesn't support batch completions."
            raise NotImplementedError(msg)
        msg = "Subclasses must implement _aretrieve_batch_results method"
        raise NotImplementedError(msg)