from collections.abc import AsyncIterator
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, cast

import httpx

import litellm
from litellm.anthropic_beta_headers_manager import filter_and_transform_beta_headers
from litellm.constants import (
    BEDROCK_MIN_THINKING_BUDGET_TOKENS,
    DEFAULT_REASONING_EFFORT_HIGH_THINKING_BUDGET,
    DEFAULT_REASONING_EFFORT_MEDIUM_THINKING_BUDGET,
    DEFAULT_REASONING_EFFORT_XHIGH_THINKING_BUDGET,
)
from litellm.litellm_core_utils.litellm_logging import verbose_logger
from litellm.llms.anthropic.chat.transformation import (
    DROP_UNSUPPORTED_OUTPUT_CONFIG_WARNING,
    AnthropicConfig,
)
from litellm.llms.anthropic.common_utils import AnthropicModelInfo
from litellm.llms.anthropic.experimental_pass_through.messages.transformation import (
    AnthropicMessagesConfig,
)
from litellm.llms.base_llm.anthropic_messages.transformation import (
    BaseAnthropicMessagesConfig,
)
from litellm.llms.bedrock.chat.invoke_handler import AWSEventStreamDecoder
from litellm.llms.bedrock.chat.invoke_transformations.base_invoke_transformation import (
    AmazonInvokeConfig,
)
from litellm.llms.bedrock.common_utils import (
    apply_bedrock_invoke_structured_output,
    ensure_bedrock_anthropic_messages_tool_names,
    get_anthropic_beta_from_headers,
    is_claude_4_5_on_bedrock,
    normalize_bedrock_opus_output_config_effort,
    normalize_custom_field_on_tools,
    normalize_tool_input_schema_types_for_bedrock_invoke,
    strip_unsupported_bedrock_invoke_output_config_keys,
)
from litellm.llms.bedrock.request_metadata import (
    bedrock_request_metadata_headers,
    merge_bedrock_invoke_headers,
)
from litellm.types.llms.anthropic import (
    ANTHROPIC_BETA_HEADER_VALUES,
    ANTHROPIC_TOOL_SEARCH_BETA_HEADER,
)
from litellm.types.llms.bedrock import BedrockInvokeAnthropicMessagesRequest
from litellm.types.llms.openai import AllMessageValues
from litellm.types.router import GenericLiteLLMParams
from litellm.types.utils import GenericStreamingChunk, ModelResponseStream
from litellm.types.utils import GenericStreamingChunk as GChunk

if TYPE_CHECKING:
    from litellm.litellm_core_utils.litellm_logging import Logging as _LiteLLMLoggingObj

    LiteLLMLoggingObj = _LiteLLMLoggingObj
else:
    LiteLLMLoggingObj = Any


class AmazonAnthropicClaudeMessagesConfig(
    AnthropicMessagesConfig,
    AmazonInvokeConfig,
):
    """
    Call Claude model family in the /v1/messages API spec
    Supports anthropic_beta parameter for beta features.
    """

    DEFAULT_BEDROCK_ANTHROPIC_API_VERSION = "bedrock-2023-05-31"

    WEBSEARCH_INTERCEPTION_DOCS_URL = "https://docs.litellm.ai/docs/integrations/websearch_interception"

    @property
    def custom_llm_provider(self) -> str | None:
        return "bedrock"

    BEDROCK_INVOKE_ALLOWED_TOP_LEVEL_FIELDS = frozenset(BedrockInvokeAnthropicMessagesRequest.__annotations__.keys())

    def __init__(self, **kwargs):
        BaseAnthropicMessagesConfig.__init__(self, **kwargs)
        AmazonInvokeConfig.__init__(self, **kwargs)

    def validate_anthropic_messages_environment(
        self,
        headers: dict,
        model: str,
        messages: list[Any],
        optional_params: dict,
        litellm_params: dict,
        api_key: str | None = None,
        api_base: str | None = None,
    ) -> tuple[dict, str | None]:
        owned_names, metadata_headers = bedrock_request_metadata_headers(litellm_params)
        return merge_bedrock_invoke_headers(headers, (), metadata_headers, owned_names), api_base

    def sign_request(
        self,
        headers: dict,
        optional_params: dict,
        request_data: dict,
        api_base: str,
        api_key: str | None = None,
        model: str | None = None,
        stream: bool | None = None,
        fake_stream: bool | None = None,
    ) -> tuple[dict, bytes | None]:
        return AmazonInvokeConfig.sign_request(
            self=self,
            headers=headers,
            optional_params=optional_params,
            request_data=request_data,
            api_base=api_base,
            api_key=api_key,
            model=model,
            stream=stream,
            fake_stream=fake_stream,
        )

    def get_complete_url(
        self,
        api_base: str | None,
        api_key: str | None,
        model: str,
        optional_params: dict,
        litellm_params: dict,
        stream: bool | None = None,
    ) -> str:
        return AmazonInvokeConfig.get_complete_url(
            self=self,
            api_base=api_base,
            api_key=api_key,
            model=model,
            optional_params=optional_params,
            litellm_params=litellm_params,
            stream=stream,
        )

    def _remove_ttl_from_cache_control(self, anthropic_messages_request: dict, model: str | None = None) -> None:
        """
        Remove unsupported fields from cache_control for Bedrock.

        Bedrock only supports `type` and `ttl` in cache_control. It does NOT support:
        - `scope` (e.g., "global") - always removed
        - `ttl` - removed for older models; Claude 4.5+ supports "5m" and "1h"

        Processes `tools`, `system`, and `messages` content blocks.

        Args:
            anthropic_messages_request: The request dictionary to modify in-place
            model: The model name to check if it supports ttl
        """
        is_claude_4_5 = False
        if model:
            is_claude_4_5 = self._is_claude_4_5_on_bedrock(model)

        def _sanitize_cache_control(cache_control: dict) -> None:
            if not isinstance(cache_control, dict):
                return
            # Bedrock doesn't support scope (e.g., "global" for cross-request caching)
            cache_control.pop("scope", None)
            # Remove ttl for models that don't support it
            if "ttl" in cache_control:
                ttl: Final = cache_control["ttl"]
                if is_claude_4_5 and ttl in ["5m", "1h"]:
                    return
                cache_control.pop("ttl", None)

        def _process_content_list(content: list) -> None:
            for item in content:
                if isinstance(item, dict) and "cache_control" in item:
                    _sanitize_cache_control(item["cache_control"])

        # Process tools
        if "tools" in anthropic_messages_request:
            for tool in anthropic_messages_request["tools"]:
                if isinstance(tool, dict) and "cache_control" in tool:
                    _sanitize_cache_control(tool["cache_control"])

        # Process system (list of content blocks)
        if "system" in anthropic_messages_request:
            system: Final = anthropic_messages_request["system"]
            if isinstance(system, list):
                _process_content_list(system)

        # Process messages
        if "messages" in anthropic_messages_request:
            for message in anthropic_messages_request["messages"]:
                if isinstance(message, dict) and "content" in message:
                    content = message["content"]
                    if isinstance(content, list):
                        _process_content_list(content)

    def _supports_extended_thinking_on_bedrock(self, model: str) -> bool:
        """
        Check if the model supports extended thinking beta headers on Bedrock.

        On 3rd-party platforms (e.g., Amazon Bedrock), extended thinking is supported
        on the adaptive-thinking models (sourced from the cost map) plus the legacy
        non-adaptive set: Claude Opus 4.5, Claude Opus 4.1, Opus 4, or Sonnet 4.

        Ref: https://docs.anthropic.com/en/docs/build-with-claude/extended-thinking

        Args:
            model: The model name

        Returns:
            True if the model supports extended thinking on Bedrock
        """
        if AnthropicModelInfo._is_adaptive_thinking_model(model, "bedrock"):
            return True

        model_lower: Final = model.lower()
        non_adaptive_patterns: Final = [
            "opus-4.5",
            "opus_4.5",
            "opus-4-5",
            "opus_4_5",  # Opus 4.5
            "opus-4.1",
            "opus_4.1",
            "opus-4-1",
            "opus_4_1",  # Opus 4.1
            "opus-4",
            "opus_4",  # Opus 4
            "sonnet-4",
            "sonnet_4",  # Sonnet 4
        ]

        return any(pattern in model_lower for pattern in non_adaptive_patterns)

    def _ensure_thinking_for_clear_thinking_context_management(
        self,
        anthropic_messages_request: dict,
        model: str,
    ) -> bool:
        """
        Bedrock rejects ``clear_thinking_20251015`` context-management edits unless
        extended thinking is ``enabled`` or ``adaptive``. Claude Code often sends
        context management without a top-level ``thinking`` field.

        When we detect that edit type on a model that supports extended thinking on
        Bedrock, inject a minimal ``thinking`` config so the request succeeds.

        Returns:
            True if ``thinking`` was added or upgraded for this fix (caller may
            need to add the interleaved-thinking beta header).
        """
        cm: Final = anthropic_messages_request.get("context_management")
        if not isinstance(cm, dict):
            return False
        edits: Final = cm.get("edits")
        if not isinstance(edits, list):
            return False
        needs_thinking: Final = any(isinstance(e, dict) and e.get("type") == "clear_thinking_20251015" for e in edits)
        if not needs_thinking:
            return False
        if not self._supports_extended_thinking_on_bedrock(model):
            return False

        is_adaptive_thinking_model: Final = AnthropicModelInfo._is_adaptive_thinking_model(model, "bedrock")

        thinking: Final = anthropic_messages_request.get("thinking")
        if isinstance(thinking, dict):
            t: Final = thinking.get("type")
            if t == "adaptive":
                return False
            if t == "enabled" and not is_adaptive_thinking_model:
                return False
            if t == "enabled":
                budget_tokens: Final = self._resolve_clear_thinking_budget_tokens(thinking.get("budget_tokens"))
                self._inject_adaptive_thinking_for_clear_thinking(anthropic_messages_request, budget_tokens, model)
                return True
            verbose_logger.debug(
                "Bedrock clear_thinking_20251015: replacing thinking=%s with minimal thinking config",
                thinking,
            )

        max_tokens: Final = anthropic_messages_request.get("max_tokens")
        budget: Final = BEDROCK_MIN_THINKING_BUDGET_TOKENS
        if isinstance(max_tokens, int) and max_tokens <= budget:
            verbose_logger.warning(
                "Bedrock clear_thinking_20251015: max_tokens=%s is not greater than "
                "minimum thinking budget (%s); cannot inject thinking safely",
                max_tokens,
                budget,
            )
            return False

        if is_adaptive_thinking_model:
            self._inject_adaptive_thinking_for_clear_thinking(anthropic_messages_request, budget, model)
            return True

        anthropic_messages_request["thinking"] = {
            "type": "enabled",
            "budget_tokens": budget,
        }
        verbose_logger.debug(
            "Bedrock clear_thinking_20251015: injected thinking with budget_tokens=%s",
            budget,
        )
        return True

    @staticmethod
    def _resolve_clear_thinking_budget_tokens(budget_tokens: int | None) -> int:
        """Honor an explicit ``budget_tokens`` (including ``0``); only fall back to
        the Bedrock minimum when the caller omitted it. A truthiness check would
        wrongly treat an explicit ``0`` as missing."""
        if budget_tokens is None:
            return BEDROCK_MIN_THINKING_BUDGET_TOKENS
        return int(budget_tokens)

    @staticmethod
    def _effort_from_thinking_budget(budget_tokens: int) -> str:
        if budget_tokens >= DEFAULT_REASONING_EFFORT_XHIGH_THINKING_BUDGET:
            return "xhigh"
        if budget_tokens >= DEFAULT_REASONING_EFFORT_HIGH_THINKING_BUDGET:
            return "high"
        if budget_tokens >= DEFAULT_REASONING_EFFORT_MEDIUM_THINKING_BUDGET:
            return "medium"
        return "low"

    def _inject_adaptive_thinking_for_clear_thinking(
        self, anthropic_messages_request: dict, budget_tokens: int, model: str
    ) -> None:
        """Adaptive-thinking models (Opus 4.7/4.8, Fable 5) reject
        ``thinking.type=enabled`` on Bedrock. Use ``thinking.type=adaptive`` plus
        an ``output_config.effort`` derived from the budget so ``clear_thinking``
        stays valid without the legacy shape."""
        output_config = anthropic_messages_request.get("output_config")
        if not isinstance(output_config, dict):
            output_config = {}
        output_config.setdefault("effort", self._effort_from_thinking_budget(budget_tokens))
        anthropic_messages_request["output_config"] = output_config
        anthropic_messages_request["thinking"] = {"type": "adaptive"}
        verbose_logger.debug(
            "Bedrock clear_thinking_20251015: injected adaptive thinking with effort=%s for model=%s",
            output_config.get("effort"),
            model,
        )

    def _is_claude_opus_4_5(self, model: str) -> bool:
        """
        Check if the model is Claude Opus 4.5.

        Args:
            model: The model name

        Returns:
            True if the model is Claude Opus 4.5
        """
        model_lower: Final = model.lower()
        opus_4_5_patterns: Final = [
            "opus-4.5",
            "opus_4.5",
            "opus-4-5",
            "opus_4_5",
        ]
        return any(pattern in model_lower for pattern in opus_4_5_patterns)

    def _is_claude_4_5_on_bedrock(self, model: str) -> bool:
        """
        Check if the model is Claude 4.5 on Bedrock.

        Claude Sonnet 4.5, Haiku 4.5, and Opus 4.5 support 1-hour prompt caching.

        Args:
            model: The model name

        Returns:
            True if the model is Claude 4.5
        """
        return is_claude_4_5_on_bedrock(model)

    def _supports_tool_search_on_bedrock(self, model: str) -> bool:
        """
        Check if the model supports tool search on Bedrock.

        The model map's ``supports_tool_search`` flag is authoritative when
        ``model`` resolves to an entry that sets it; the name patterns below
        cover ids the map cannot resolve (ARNs, unlisted regional variants).

        Ref: https://platform.claude.com/docs/en/agents-and-tools/tool-use/tool-search-tool

        Args:
            model: The model name

        Returns:
            True if the model supports tool search on Bedrock
        """
        catalog: Final = AnthropicModelInfo._get_provider_resolved_capability(model, "supports_tool_search", "bedrock")
        if catalog is not None:
            return catalog

        model_lower: Final = model.lower()

        supported_patterns: Final = [
            # Opus 4.5
            "opus-4.5",
            "opus_4.5",
            "opus-4-5",
            "opus_4_5",
            # Sonnet 4.5
            "sonnet-4.5",
            "sonnet_4.5",
            "sonnet-4-5",
            "sonnet_4_5",
            # Opus 4.6
            "opus-4.6",
            "opus_4.6",
            "opus-4-6",
            "opus_4_6",
            # sonnet 4.6
            "sonnet-4.6",
            "sonnet_4.6",
            "sonnet-4-6",
            "sonnet_4_6",
            # Opus 4.7
            "opus-4.7",
            "opus_4.7",
            "opus-4-7",
            "opus_4_7",
            # Haiku 4.5
            "haiku-4.5",
            "haiku_4.5",
            "haiku-4-5",
            "haiku_4_5",
        ]

        return any(pattern in model_lower for pattern in supported_patterns)

    def _get_tool_search_beta_header_for_bedrock(
        self,
        model: str,
        tool_search_used: bool,
        programmatic_tool_calling_used: bool,
        input_examples_used: bool,
        beta_set: set,
    ) -> None:
        """
        Adjust tool search beta header for Bedrock.

        Bedrock requires a different beta header for tool search than the
        Anthropic API when tool search is used without programmatic tool
        calling or input examples: `tool-search-tool-2025-10-19`, and only on
        the models listed in `_supports_tool_search_on_bedrock`.

        Ref: https://platform.claude.com/docs/en/agents-and-tools/tool-use/tool-search-tool

        Args:
            model: The model name
            tool_search_used: Whether tool search is used
            programmatic_tool_calling_used: Whether programmatic tool calling is used
            input_examples_used: Whether input examples are used
            beta_set: The set of beta headers to modify in-place
        """
        if tool_search_used and not (programmatic_tool_calling_used or input_examples_used):
            beta_set.discard(ANTHROPIC_TOOL_SEARCH_BETA_HEADER)
            if self._supports_tool_search_on_bedrock(model):
                beta_set.add("tool-search-tool-2025-10-19")

    # Bedrock-InvokeModel-supported ``context_management.edits`` types and the
    # ``anthropic-beta`` header that each one requires. ``clear_thinking_20251015``
    # is intentionally absent — it is LiteLLM-internal, consumed via
    # ``_ensure_thinking_for_clear_thinking_context_management``, and forwarding
    # the raw edit trips Bedrock's
    # ``"context_management: Extra inputs are not permitted"`` 400.
    #
    # Bedrock InvokeModel DOES support ``clear_tool_uses_20250919`` under the
    # ``context-management-2025-06-27`` beta. AWS docs:
    # https://docs.aws.amazon.com/bedrock/latest/userguide/model-parameters-anthropic-claude-messages-tool-use.md
    _BEDROCK_INVOKE_SUPPORTED_CONTEXT_MANAGEMENT_EDITS: dict[str, str] = {
        "compact_20260112": ANTHROPIC_BETA_HEADER_VALUES.COMPACT_2026_01_12.value,
        "clear_tool_uses_20250919": ANTHROPIC_BETA_HEADER_VALUES.CONTEXT_MANAGEMENT_2025_06_27.value,
    }

    @staticmethod
    def _filter_context_management_for_bedrock_invoke(
        anthropic_messages_request: dict,
        beta_set: set,
    ) -> None:
        """
        Filter ``context_management.edits`` to the subset that Bedrock InvokeModel
        accepts and add the matching ``anthropic-beta`` header for each surviving
        edit type.

        - ``compact_20260112`` -> ``compact-2026-01-12``
        - ``clear_tool_uses_20250919`` -> ``context-management-2025-06-27``

        Other edit types (notably ``clear_thinking_20251015``, which Claude Code
        sends on every request) are LiteLLM-internal: thinking is injected
        separately via ``_ensure_thinking_for_clear_thinking_context_management``,
        and forwarding the raw edit would trip Bedrock's
        ``"context_management: Extra inputs are not permitted"`` 400.

        Refs:
          * https://github.com/BerriAI/litellm/issues/27532
          * https://docs.aws.amazon.com/bedrock/latest/userguide/model-parameters-anthropic-claude-messages-tool-use.md
        """
        cm: Final = anthropic_messages_request.get("context_management")
        if not isinstance(cm, dict):
            return
        edits: Final = cm.get("edits")
        if not isinstance(edits, list):
            anthropic_messages_request.pop("context_management", None)
            return

        supported: Final = AmazonAnthropicClaudeMessagesConfig._BEDROCK_INVOKE_SUPPORTED_CONTEXT_MANAGEMENT_EDITS
        retained_edits: Final = [e for e in edits if isinstance(e, dict) and e.get("type") in supported]
        if not retained_edits:
            anthropic_messages_request.pop("context_management", None)
            return

        beta_set.update(supported[e["type"]] for e in retained_edits)
        anthropic_messages_request["context_management"] = {
            **cm,
            "edits": retained_edits,
        }

    def _get_bedrock_invoke_anthropic_beta_headers(
        self,
        model: str,
        messages: list[dict],
        anthropic_messages_optional_request_params: dict,
        headers: dict,
        anthropic_messages_request: dict,
        injected_thinking_for_clear_thinking: bool,
    ) -> list[str]:
        anthropic_model_info: Final = AnthropicModelInfo()
        tools: Final = anthropic_messages_optional_request_params.get("tools")
        messages_typed: Final = cast(list[AllMessageValues], messages)
        tool_search_used: Final = anthropic_model_info.is_tool_search_used(tools)
        programmatic_tool_calling_used: Final = anthropic_model_info.is_programmatic_tool_calling_used(tools)
        input_examples_used: Final = anthropic_model_info.is_input_examples_used(tools)

        user_beta_set: Final = set(get_anthropic_beta_from_headers(headers))
        beta_set: Final = set(user_beta_set)
        auto_betas: Final = anthropic_model_info.get_anthropic_beta_list(
            model=model,
            optional_params=anthropic_messages_optional_request_params,
            computer_tool_used=anthropic_model_info.is_computer_tool_used(tools),
            prompt_caching_set=False,
            file_id_used=anthropic_model_info.is_file_id_used(messages_typed),
            mcp_server_used=anthropic_model_info.is_mcp_server_used(
                anthropic_messages_optional_request_params.get("mcp_servers")
            ),
            custom_llm_provider="bedrock",
        )
        beta_set.update(auto_betas)

        if injected_thinking_for_clear_thinking:
            beta_set.add("interleaved-thinking-2025-05-14")

        self._filter_context_management_for_bedrock_invoke(
            anthropic_messages_request=anthropic_messages_request,
            beta_set=beta_set,
        )

        self._get_tool_search_beta_header_for_bedrock(
            model=model,
            tool_search_used=tool_search_used,
            programmatic_tool_calling_used=programmatic_tool_calling_used,
            input_examples_used=input_examples_used,
            beta_set=beta_set,
        )

        if "tool-search-tool-2025-10-19" in beta_set:
            beta_set.add("tool-examples-2025-10-29")

        filtered_betas: Final = sorted(
            filter_and_transform_beta_headers(
                beta_headers=list(beta_set),
                provider="bedrock",
            )
        )

        dropped_user_betas: Final = sorted(
            b for b in user_beta_set if not filter_and_transform_beta_headers([b], provider="bedrock")
        )
        if dropped_user_betas:
            verbose_logger.warning(
                "Bedrock Invoke: dropping unsupported anthropic-beta values "
                "from client headers: %s. Bedrock has no mapping entry for "
                "these; forwarding them would cause a 400.",
                dropped_user_betas,
            )

        return filtered_betas

    @staticmethod
    def _reject_unsupported_web_search_tools(anthropic_messages_request: dict[str, object], model: str) -> None:
        """
        Bedrock's Anthropic endpoints cannot execute Anthropic's server-side
        ``web_search_*`` tool; forwarding it returns an opaque
        "The provided request is not valid" 400 from Bedrock. Fail fast with an
        error that names the problem and the fix instead.

        When web search interception is enabled
        (``litellm_settings.callbacks: ["websearch_interception"]``), the tool
        is converted to a regular function tool before this transform runs, so
        this guard never fires.
        """
        from litellm.integrations.websearch_interception.tools import (
            is_anthropic_native_web_search_tool,
        )

        tools: Final = anthropic_messages_request.get("tools")
        if not isinstance(tools, list):
            return
        web_search_tool: Final = next(
            (t for t in tools if isinstance(t, dict) and is_anthropic_native_web_search_tool(t)),
            None,
        )
        if web_search_tool is None:
            return
        raise litellm.BadRequestError(
            message=(
                f"Bedrock does not support Anthropic's server-side web search tool "
                f"(tool type '{web_search_tool.get('type')}', model '{model}'). "
                "To use web search with this model, enable LiteLLM's web search interception "
                "so the proxy executes the search instead: "
                f"{AmazonAnthropicClaudeMessagesConfig.WEBSEARCH_INTERCEPTION_DOCS_URL}. "
                "Alternatively, remove the web_search tool from the request."
            ),
            model=model,
            llm_provider="bedrock",
        )

    def _strip_unsupported_bedrock_invoke_fields(
        self,
        anthropic_messages_request: dict,
    ) -> dict:
        allowed: Final = self.BEDROCK_INVOKE_ALLOWED_TOP_LEVEL_FIELDS
        stripped: Final = sorted(k for k in anthropic_messages_request if k not in allowed)
        if stripped:
            verbose_logger.debug(
                "Bedrock Invoke: stripping unsupported top-level request fields: %s",
                stripped,
            )
        return {k: v for k, v in anthropic_messages_request.items() if k in allowed}

    @staticmethod
    def _clamp_adaptive_reasoning_effort_for_bedrock(model: str, optional_params: dict) -> None:
        """Lower ``reasoning_effort`` to the Bedrock effort ceiling before validation.

        The shared ``/v1/messages`` effort gate rejects tiers a model does not
        natively support (e.g. ``xhigh`` on Opus 4.6). Bedrock's chat paths instead
        clamp the tier to the model's ``bedrock_output_config_effort_ceiling`` so
        Claude Code "goal mode" keeps working; mirror that here so the messages
        path degrades ``xhigh`` -> ``max`` rather than 400-ing. Non-adaptive models
        and models without a ceiling are left untouched.
        """
        if not AnthropicModelInfo._is_adaptive_thinking_model(model, "bedrock"):
            return
        effort: Final = optional_params.get("reasoning_effort")
        if not isinstance(effort, str):
            return
        clamped: Final = {"effort": effort}
        normalize_bedrock_opus_output_config_effort(model=model, output_config=clamped)
        optional_params["reasoning_effort"] = clamped["effort"]

    def transform_anthropic_messages_request(
        self,
        model: str,
        messages: list[dict],
        anthropic_messages_optional_request_params: dict,
        litellm_params: GenericLiteLLMParams,
        headers: dict,
    ) -> dict:
        self._clamp_adaptive_reasoning_effort_for_bedrock(
            model=model,
            optional_params=anthropic_messages_optional_request_params,
        )
        anthropic_messages_request = AnthropicMessagesConfig.transform_anthropic_messages_request(
            self=self,
            model=model,
            messages=messages,
            anthropic_messages_optional_request_params=anthropic_messages_optional_request_params,
            litellm_params=litellm_params,
            headers=headers,
        )
        self._normalize_system_role_messages(anthropic_messages_request, model=model)
        #########################################################
        ############## BEDROCK Invoke SPECIFIC TRANSFORMATION ###
        #########################################################

        self._reject_unsupported_web_search_tools(anthropic_messages_request=anthropic_messages_request, model=model)

        # 1. anthropic_version is required for all claude models
        if "anthropic_version" not in anthropic_messages_request:
            anthropic_messages_request["anthropic_version"] = self.DEFAULT_BEDROCK_ANTHROPIC_API_VERSION

        # 2. `stream` is not allowed in request body for bedrock invoke
        if "stream" in anthropic_messages_request:
            anthropic_messages_request.pop("stream", None)

        # 3. `model` is not allowed in request body for bedrock invoke
        if "model" in anthropic_messages_request:
            anthropic_messages_request.pop("model", None)

        injected_thinking_for_clear_thinking: Final = self._ensure_thinking_for_clear_thinking_context_management(
            anthropic_messages_request=anthropic_messages_request,
            model=model,
        )

        # 4. Remove `ttl` field from cache_control in messages (Bedrock doesn't support it for older models)
        self._remove_ttl_from_cache_control(anthropic_messages_request=anthropic_messages_request, model=model)

        # 5. Route structured-output params (`output_format` /
        # `output_config.format`) to native enforcement or the inline-schema
        # fallback, then strip `output_config` keys the model does not accept.
        # Ref: https://github.com/BerriAI/litellm/issues/22797
        existing_output_config: Final = anthropic_messages_request.get("output_config")
        if isinstance(existing_output_config, dict):
            anthropic_messages_request["output_config"] = dict(existing_output_config)
        apply_bedrock_invoke_structured_output(
            model=model,
            request_body=anthropic_messages_request,
        )
        normalize_bedrock_opus_output_config_effort(
            model=model,
            output_config=anthropic_messages_request.get("output_config"),
        )
        strip_unsupported_bedrock_invoke_output_config_keys(
            model=model,
            request_body=anthropic_messages_request,
        )

        # 5b. Hoist `custom.defer_loading` then drop `custom` (Bedrock doesn't support it)
        # Ref: https://github.com/BerriAI/litellm/issues/22847
        normalize_custom_field_on_tools(anthropic_messages_request)
        normalize_tool_input_schema_types_for_bedrock_invoke(anthropic_messages_request)
        ensure_bedrock_anthropic_messages_tool_names(anthropic_messages_request)

        # 6. AUTO-INJECT beta headers based on features used
        filtered_betas: Final = self._get_bedrock_invoke_anthropic_beta_headers(
            model=model,
            messages=messages,
            anthropic_messages_optional_request_params=anthropic_messages_optional_request_params,
            headers=headers,
            anthropic_messages_request=anthropic_messages_request,
            injected_thinking_for_clear_thinking=injected_thinking_for_clear_thinking,
        )

        if filtered_betas:
            anthropic_messages_request["anthropic_beta"] = filtered_betas

        remaining_output_config: Final = anthropic_messages_request.get("output_config")
        if (
            litellm.drop_params is True
            and isinstance(remaining_output_config, dict)
            and any(key != "format" for key in remaining_output_config)
            and not AnthropicConfig._model_supports_effort_param(model, "bedrock")
        ):
            verbose_logger.warning(
                DROP_UNSUPPORTED_OUTPUT_CONFIG_WARNING,
                model,
            )
            anthropic_messages_request.pop("output_config", None)

        # 7. Final safety net: filter top-level fields to the Bedrock Invoke allowlist.
        # Catches Anthropic-only extensions (output_config, speed, mcp_servers, ...)
        # and any future additions Claude Code may start sending. ``context_management``
        # has already been pre-filtered to its Bedrock-supported subset above.
        anthropic_messages_request = self._strip_unsupported_bedrock_invoke_fields(anthropic_messages_request)

        return anthropic_messages_request

    def get_async_streaming_response_iterator(
        self,
        model: str,
        httpx_response: httpx.Response,
        request_body: dict,
        litellm_logging_obj: LiteLLMLoggingObj,
    ) -> AsyncIterator:
        aws_decoder: Final = AmazonAnthropicClaudeMessagesStreamDecoder(
            model=model,
        )
        completion_stream: Final = aws_decoder.aiter_bytes(
            httpx_response.aiter_bytes(chunk_size=aws_decoder.DEFAULT_CHUNK_SIZE)
        )
        # Convert decoded Bedrock events to Server-Sent Events expected by Anthropic clients.
        return self.bedrock_sse_wrapper(
            completion_stream=completion_stream,
            litellm_logging_obj=litellm_logging_obj,
            request_body=request_body,
        )

    async def bedrock_sse_wrapper(
        self,
        completion_stream: AsyncIterator[bytes | GenericStreamingChunk | ModelResponseStream | dict],
        litellm_logging_obj: LiteLLMLoggingObj,
        request_body: dict,
    ):
        """
        Bedrock invoke does not return SSE formatted data. This function is a wrapper to ensure litellm chunks are SSE formatted.

        Bedrock's Anthropic-compatible streaming usually puts cache usage fields
        (cache_creation_input_tokens, cache_read_input_tokens) on message_stop.
        Some deployments (including GovCloud) emit the cache breakdown only on
        ``message_start.message.usage``; ``message_delta`` / ``message_stop`` then
        repeat uncached ``input_tokens`` only. We promote cache fields from
        ``message_stop`` onto ``message_delta``, and when those are absent we
        merge them from ``message_start`` so logging/cost sees a consistent usage
        object (fixes negative input costs: LIT-2411).
        """
        from litellm.llms.anthropic.experimental_pass_through.messages.streaming_iterator import (
            BaseAnthropicMessagesStreamingIterator,
        )

        handler: Final = BaseAnthropicMessagesStreamingIterator(
            litellm_logging_obj=litellm_logging_obj,
            request_body=request_body,
        )

        patched_stream: Final = self._promote_message_stop_usage(completion_stream)

        sse_stream: Final = handler.async_sse_wrapper(patched_stream)
        try:
            async for chunk in sse_stream:
                yield chunk
        finally:
            # Close the inner generator deterministically so a client disconnect
            # (GeneratorExit here) reaches async_sse_wrapper's partial-spend logging
            # now instead of at garbage collection. See LIT-5839.
            await sse_stream.aclose()

    @staticmethod
    def _merge_message_start_cache_into_delta_usage(
        delta_usage: dict[str, Any],
        start_usage: dict[str, Any] | None,
    ) -> None:
        """
        Copy cache breakdown from message_start onto message_delta usage when
        those keys are missing on the delta (GovCloud / some Bedrock streams).
        """
        if not start_usage:
            return
        for field in ("cache_creation_input_tokens", "cache_read_input_tokens"):
            if field not in delta_usage:
                val = start_usage.get(field)
                if val is not None:
                    delta_usage[field] = val
        if "cache_creation" not in delta_usage:
            cc: Final = start_usage.get("cache_creation")
            if cc is not None:
                delta_usage["cache_creation"] = cc

    @staticmethod
    async def _promote_message_stop_usage(
        completion_stream: AsyncIterator[bytes | GenericStreamingChunk | ModelResponseStream | dict],
    ) -> AsyncIterator[bytes | GenericStreamingChunk | ModelResponseStream | dict]:
        """
        Promote cache usage fields onto message_delta from message_stop (and,
        when stop lacks them, from message_start).  Ensures the final usage
        chunk that logging/cost sees is always self-consistent.
        """
        _CACHE_FIELDS: Final = ("cache_creation_input_tokens", "cache_read_input_tokens")
        pending_delta: dict[str, Any] | None = None
        start_usage_snapshot: dict[str, Any] | None = None

        async for chunk in completion_stream:
            if not isinstance(chunk, dict):
                if pending_delta is not None:
                    yield pending_delta
                    pending_delta = None
                yield chunk
                continue

            chunk_type = chunk.get("type")

            if chunk_type == "message_start":
                msg: dict[str, Any] = cast(dict[str, Any], chunk.get("message") or {})
                u = msg.get("usage")
                if isinstance(u, dict):
                    start_usage_snapshot = dict(u)
                if pending_delta is not None:
                    yield pending_delta
                    pending_delta = None
                yield chunk
                continue

            if chunk_type == "message_delta":
                pending_delta = cast(dict[str, Any], chunk)
                continue

            if chunk_type == "message_stop" and pending_delta is not None:
                stop_usage = dict(chunk.get("usage") or {})
                delta_usage = dict(pending_delta.get("usage") or {})

                for field in _CACHE_FIELDS:
                    if field in stop_usage:
                        delta_usage[field] = stop_usage[field]

                raw_input = stop_usage.get("input_tokens")
                if raw_input is not None:
                    delta_usage["input_tokens"] = raw_input if isinstance(raw_input, int) else 0

                AmazonAnthropicClaudeMessagesConfig._merge_message_start_cache_into_delta_usage(
                    delta_usage, start_usage_snapshot
                )

                if delta_usage:
                    pending_delta["usage"] = delta_usage

                yield pending_delta
                pending_delta = None
                yield chunk
                continue

            if pending_delta is not None:
                yield pending_delta
                pending_delta = None

            yield chunk

        if pending_delta is not None:
            delta_usage = dict(pending_delta.get("usage") or {})
            AmazonAnthropicClaudeMessagesConfig._merge_message_start_cache_into_delta_usage(
                delta_usage, start_usage_snapshot
            )
            if delta_usage:
                pending_delta["usage"] = delta_usage
            yield pending_delta


class AmazonAnthropicClaudeMessagesStreamDecoder(AWSEventStreamDecoder):
    def __init__(
        self,
        model: str,
    ) -> None:
        """
        Iterator to return Bedrock invoke response in anthropic /messages format
        """
        super().__init__(model=model)
        self.DEFAULT_CHUNK_SIZE = 1024

    def _chunk_parser(self, chunk_data: dict) -> GChunk | ModelResponseStream | dict:
        """
        Parse the chunk data into anthropic /messages format

        Bedrock returns usage metrics using camelCase keys. Convert these to
        the Anthropic `/v1/messages` specification so callers receive a
        consistent response shape when streaming.

        Token counts already present in the chunk's own Anthropic usage block
        win over the invocationMetrics-derived ones, and cache token fields
        (``cache_read_input_tokens`` / ``cache_creation_input_tokens`` on
        ``message_stop.usage``, or ``cacheReadInputTokenCount`` /
        ``cacheWriteInputTokenCount`` inside the invocation metrics) are
        preserved: ``invocationMetrics.inputTokenCount`` excludes cache reads
        and writes, so replacing the whole usage block with input/output counts
        alone drops the cache breakdown, ``_promote_message_stop_usage`` has
        nothing left to promote, and cache tokens end up billed at $0.
        """
        amazon_bedrock_invocation_metrics: Final = chunk_data.pop("amazon-bedrock-invocationMetrics", {})
        if amazon_bedrock_invocation_metrics:
            existing_usage: Final = chunk_data.get("usage")
            preserved_usage: Final = existing_usage if isinstance(existing_usage, dict) else MappingProxyType({})
            metrics_usage: Final = MappingProxyType(
                {
                    anthropic_key: amazon_bedrock_invocation_metrics[metrics_key]
                    for anthropic_key, metrics_key in (
                        ("input_tokens", "inputTokenCount"),
                        ("output_tokens", "outputTokenCount"),
                        ("cache_read_input_tokens", "cacheReadInputTokenCount"),
                        ("cache_creation_input_tokens", "cacheWriteInputTokenCount"),
                    )
                    if metrics_key in amazon_bedrock_invocation_metrics
                }
            )
            chunk_data["usage"] = {**metrics_usage, **preserved_usage}
        return chunk_data
