"""
Translating between OpenAI's `/chat/completion` format and Amazon's `/converse` format
"""

import copy
import json
import time
import types
from collections.abc import Mapping
from typing import TYPE_CHECKING, Final, Literal, cast, overload

import httpx

import litellm
from litellm._logging import verbose_logger
from litellm.constants import (
    BEDROCK_MIN_THINKING_BUDGET_TOKENS,
    RESPONSE_FORMAT_TOOL_NAME,
)
from litellm.litellm_core_utils.core_helpers import (
    filter_exceptions_from_params,
    filter_internal_params,
    map_finish_reason,
    safe_deep_copy,
)
from litellm.litellm_core_utils.litellm_logging import Logging
from litellm.litellm_core_utils.prompt_templates.common_utils import (
    _parse_content_for_reasoning,
)
from litellm.litellm_core_utils.prompt_templates.factory import (
    BedrockConverseMessagesProcessor,
    _bedrock_converse_messages_pt,
    _bedrock_tools_pt,
    make_valid_bedrock_tool_name,
)
from litellm.llms.anthropic.chat.transformation import (
    DROP_UNSUPPORTED_ADAPTIVE_THINKING_WARNING,
    DROP_UNSUPPORTED_OUTPUT_CONFIG_WARNING,
    REASONING_EFFORT_TO_OUTPUT_CONFIG_EFFORT,
    AnthropicConfig,
)
from litellm.llms.anthropic.common_utils import AnthropicModelInfo
from litellm.llms.base_llm.chat.transformation import BaseConfig, BaseLLMException
from litellm.llms.bedrock.request_metadata import (
    bedrock_request_metadata_headers,
    bedrock_request_metadata_is_owned,
    merge_bedrock_invoke_headers,
    resolve_bedrock_request_metadata,
)
from litellm.types.llms.bedrock import *
from litellm.types.llms.openai import (
    AllMessageValues,
    ChatCompletionAnnotation,
    ChatCompletionAssistantMessage,
    ChatCompletionRedactedThinkingBlock,
    ChatCompletionResponseMessage,
    ChatCompletionSystemMessage,
    ChatCompletionThinkingBlock,
    ChatCompletionToolCallChunk,
    ChatCompletionToolCallFunctionChunk,
    ChatCompletionToolParam,
    ChatCompletionToolParamFunctionChunk,
    ChatCompletionUserMessage,
    OpenAIChatCompletionToolParam,
    OpenAIMessageContentListBlock,
)
from litellm.types.utils import (
    CacheCreationTokenDetails,
    ChatCompletionMessageToolCall,
    CompletionTokensDetailsWrapper,
    Function,
    Message,
    ModelResponse,
    PromptTokensDetailsWrapper,
    Usage,
)
from litellm.utils import (
    add_dummy_tool,
    any_assistant_message_has_thinking_blocks,
    has_tool_call_blocks,
    last_assistant_with_tool_calls_has_no_thinking_blocks,
    supports_reasoning,
    token_counter,
)

from ..common_utils import (
    BedrockError,
    BedrockModelInfo,
    bedrock_converse_supports_parallel_tool_use_config,
    bedrock_model_accepts_cache_points,
    get_anthropic_beta_from_headers,
    get_bedrock_tool_name,
    is_bedrock_application_inference_profile_arn,
    is_claude_4_5_on_bedrock,
    normalize_bedrock_opus_output_config_effort,
)

if TYPE_CHECKING:
    import tiktoken

# Computer use tool prefixes supported by Bedrock
BEDROCK_COMPUTER_USE_TOOLS: Final = [
    "computer_use_preview",
    "computer_",
    "bash_",
    "text_editor_",
]

# Beta header patterns that are not supported by Bedrock Converse API
# These will be filtered out to prevent errors
UNSUPPORTED_BEDROCK_CONVERSE_BETA_PATTERNS: Final = [
    "advanced-tool-use",  # Bedrock Converse doesn't support advanced-tool-use beta headers
    "prompt-caching",  # Prompt caching not supported in Converse API
    "compact-2026-01-12",  # The compact beta feature is not currently supported on the Converse and ConverseStream APIs
]


class AmazonConverseConfig(BaseConfig):
    """
    Reference - https://docs.aws.amazon.com/bedrock/latest/APIReference/API_runtime_Converse.html
    #2 - https://docs.aws.amazon.com/bedrock/latest/userguide/conversation-inference.html#conversation-inference-supported-models-features
    """

    maxTokens: int | None
    stopSequences: list[str] | None
    temperature: int | None
    topP: int | None
    topK: int | None

    def __init__(
        self,
        maxTokens: int | None = None,
        stopSequences: list[str] | None = None,
        temperature: int | None = None,
        topP: int | None = None,
        topK: int | None = None,
    ) -> None:
        locals_: Final = locals().copy()
        for key, value in locals_.items():
            if key != "self" and value is not None:
                setattr(self.__class__, key, value)

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

    @classmethod
    def get_config_blocks(cls) -> dict:
        return {
            "guardrailConfig": GuardrailConfigBlock,
            "performanceConfig": PerformanceConfigBlock,
            "serviceTier": ServiceTierBlock,
        }

    @staticmethod
    def _convert_consecutive_user_messages_to_guarded_text(
        messages: list[AllMessageValues], optional_params: dict
    ) -> list[AllMessageValues]:
        """
        Convert consecutive user messages at the end to guarded_text type if guardrailConfig is present
        and no guarded_text is already present in those messages.
        """
        # Check if guardrailConfig is present
        if "guardrailConfig" not in optional_params:
            return messages

        # Find all consecutive user messages at the end
        consecutive_user_message_indices: Final = []
        for i in range(len(messages) - 1, -1, -1):
            if messages[i].get("role") == "user":
                consecutive_user_message_indices.append(i)
            else:
                break

        if not consecutive_user_message_indices:
            return messages

        # Process each consecutive user message
        messages_copy: Final = copy.deepcopy(messages)
        for user_message_index in consecutive_user_message_indices:
            user_message = messages_copy[user_message_index]
            content = user_message.get("content", [])

            if isinstance(content, list):
                has_guarded_text = any(
                    isinstance(item, dict) and item.get("type") == "guarded_text" for item in content
                )
                if has_guarded_text:
                    continue  # Skip this message if it already has guarded_text

                # Convert text elements to guarded_text
                new_content = []
                for item in content:
                    if isinstance(item, dict) and item.get("type") == "text":
                        new_item = {"type": "guarded_text", "text": item["text"]}
                        new_content.append(new_item)
                    else:
                        new_content.append(item)

                messages_copy[user_message_index]["content"] = new_content
            elif isinstance(content, str):
                # If content is a string, convert it to guarded_text
                messages_copy[user_message_index]["content"] = [{"type": "guarded_text", "text": content}]

        return messages_copy

    @classmethod
    def get_config(cls):
        return {
            k: v
            for k, v in cls.__dict__.items()
            if not k.startswith("_")
            and not isinstance(
                v,
                (
                    types.FunctionType,
                    types.BuiltinFunctionType,
                    classmethod,
                    staticmethod,
                ),
            )
            and v is not None
        }

    def _validate_request_metadata(self, metadata: dict) -> None:
        """
        Validate requestMetadata according to AWS Bedrock Converse API constraints.

        Constraints:
        - Maximum of 16 items
        - Keys: 1-256 characters, pattern [a-zA-Z0-9\\s:_@$#=/+,-.]{1,256}
        - Values: 0-256 characters, pattern [a-zA-Z0-9\\s:_@$#=/+,-.]{0,256}
        """
        import re

        if not isinstance(metadata, dict):
            raise litellm.exceptions.BadRequestError(
                message="requestMetadata must be a dictionary",
                model="bedrock",
                llm_provider="bedrock",
            )

        if len(metadata) > 16:
            raise litellm.exceptions.BadRequestError(
                message="requestMetadata can contain a maximum of 16 items",
                model="bedrock",
                llm_provider="bedrock",
            )

        key_pattern: Final = re.compile(r"^[a-zA-Z0-9\s:_@$#=/+,.-]{1,256}$")
        value_pattern: Final = re.compile(r"^[a-zA-Z0-9\s:_@$#=/+,.-]{0,256}$")

        for key, value in metadata.items():
            if not isinstance(key, str):
                raise litellm.exceptions.BadRequestError(
                    message="requestMetadata keys must be strings",
                    model="bedrock",
                    llm_provider="bedrock",
                )

            if not isinstance(value, str):
                raise litellm.exceptions.BadRequestError(
                    message="requestMetadata values must be strings",
                    model="bedrock",
                    llm_provider="bedrock",
                )

            if len(key) == 0 or len(key) > 256:
                raise litellm.exceptions.BadRequestError(
                    message="requestMetadata key length must be 1-256 characters",
                    model="bedrock",
                    llm_provider="bedrock",
                )

            if len(value) > 256:
                raise litellm.exceptions.BadRequestError(
                    message="requestMetadata value length must be 0-256 characters",
                    model="bedrock",
                    llm_provider="bedrock",
                )

            if not key_pattern.match(key):
                raise litellm.exceptions.BadRequestError(
                    message=f"requestMetadata key '{key}' contains invalid characters. Allowed: [a-zA-Z0-9\\s:_@$#=/+,.-]",
                    model="bedrock",
                    llm_provider="bedrock",
                )

            if not value_pattern.match(value):
                raise litellm.exceptions.BadRequestError(
                    message=f"requestMetadata value '{value}' contains invalid characters. Allowed: [a-zA-Z0-9\\s:_@$#=/+,.-]",
                    model="bedrock",
                    llm_provider="bedrock",
                )

    def _is_nova_2_model(self, model: str) -> bool:
        """
        Check if the model is a Nova 2 model that supports reasoningConfig.

        Nova 2 models use a different reasoning configuration structure compared to
        Anthropic's thinking parameter and GPT-OSS's reasoning_effort parameter.

        Supported models:
        - amazon.nova-2-lite-v1:0
        - amazon.nova-2-pro-preview-20251202-v1:0
        - us.amazon.nova-2-lite-v1:0
        - eu.amazon.nova-2-lite-v1:0
        - apac.amazon.nova-2-lite-v1:0
        - (and other regional variants)

        Args:
            model: The model identifier

        Returns:
            True if the model is a Nova 2 model, False otherwise

        Examples:
            >>> config = AmazonConverseConfig()
            >>> config._is_nova_2_model("amazon.nova-2-lite-v1:0")
            True
            >>> config._is_nova_2_model("us.amazon.nova-2-lite-v1:0")
            True
            >>> config._is_nova_2_model("us.amazon.nova-2-pro-preview-20251202-v1:0")
            True
            >>> config._is_nova_2_model("amazon.nova-pro-1-5-v1:0")
            False
            >>> config._is_nova_2_model("amazon.nova-pro-v1:0")
            False
        """
        # Remove provider routing prefix if present (bedrock/converse/, bedrock/, converse/)
        model_without_region = model
        for routing_prefix in ["bedrock/converse/", "bedrock/", "converse/"]:
            if model_without_region.startswith(routing_prefix):
                model_without_region = model_without_region[len(routing_prefix) :]
                break

        # Remove regional prefix if present (us., eu., apac.)
        for prefix in ["us.", "eu.", "apac."]:
            if model_without_region.startswith(prefix):
                model_without_region = model_without_region[len(prefix) :]
                break

        # Check if the model is a Nova 2 model (matches nova-2-lite, nova-2-pro, etc.)
        # Also check for nova-2/ spec prefix for imported models
        return model_without_region.startswith("amazon.nova-2-") or model_without_region.startswith("nova-2/")

    def _map_web_search_options(self, web_search_options: dict, model: str) -> BedrockToolBlock | None:
        """
        Map web_search_options to Nova grounding systemTool.

        Nova grounding (web search) is only supported on Amazon Nova models.
        Returns None for non-Nova models.

        Args:
            web_search_options: The web_search_options dict from the request
            model: The model identifier string

        Returns:
            BedrockToolBlock with systemTool for Nova models, None otherwise

        Reference: https://docs.aws.amazon.com/nova/latest/userguide/grounding.html
        """
        # Only Nova models support nova_grounding
        # Model strings can be like: "amazon.nova-pro-v1:0", "us.amazon.nova-pro-v1:0", etc.
        if "nova" not in model.lower():
            verbose_logger.debug(
                "web_search_options passed but model %s is not a Nova model. Nova grounding is only supported on Amazon Nova models.",
                model,
            )
            return None

        # Nova doesn't support search_context_size or user_location params
        # (unlike Anthropic), so we just enable grounding with no options
        return BedrockToolBlock(systemTool={"name": "nova_grounding"})

    def _transform_reasoning_effort_to_reasoning_config(self, reasoning_effort: str) -> dict:
        """
        Transform reasoning_effort parameter to Nova 2 reasoningConfig structure.

        Nova 2 models use a reasoningConfig structure in additionalModelRequestFields
        that differs from both Anthropic's thinking parameter and GPT-OSS's reasoning_effort.

        Args:
            reasoning_effort: The reasoning effort level, must be "low" or "high"

        Returns:
            dict: A dictionary containing the reasoningConfig structure:
                {
                    "reasoningConfig": {
                        "type": "enabled",
                        "maxReasoningEffort": "low" | "medium" |"high"
                    }
                }

        Raises:
            BadRequestError: If reasoning_effort is not "low", "medium" or "high"

        Examples:
            >>> config = AmazonConverseConfig()
            >>> config._transform_reasoning_effort_to_reasoning_config("high")
            {'reasoningConfig': {'type': 'enabled', 'maxReasoningEffort': 'high'}}
            >>> config._transform_reasoning_effort_to_reasoning_config("low")
            {'reasoningConfig': {'type': 'enabled', 'maxReasoningEffort': 'low'}}
        """
        valid_values: Final = ["low", "medium", "high"]
        if reasoning_effort not in valid_values:
            raise litellm.exceptions.BadRequestError(
                message=f"Invalid reasoning_effort value '{reasoning_effort}' for Nova 2 models. "
                f"Supported values: {valid_values}",
                model="amazon.nova-2-lite-v1:0",
                llm_provider="bedrock_converse",
            )

        return {
            "reasoningConfig": {
                "type": "enabled",
                "maxReasoningEffort": reasoning_effort,
            }
        }

    def _handle_reasoning_effort_parameter(self, model: str, reasoning_effort: str, optional_params: dict) -> None:
        """
        Handle the reasoning_effort parameter based on the model type.

        - GPT-OSS models: passed through unchanged via additionalModelRequestFields.
        - OpenAI GPT-5.x models: mapped to ``reasoning.effort`` via additionalModelRequestFields.
        - Nova 2 models: transformed to reasoningConfig.
        - Anthropic models: mapped to ``thinking`` (and ``output_config.effort`` on
          adaptive Claude 4.6 / 4.7).
        """
        if "gpt-oss" in model:
            optional_params["reasoning_effort"] = reasoning_effort
        elif "openai.gpt-5" in model:
            reasoning: Final[BedrockConverseGptReasoningEffortBlock] = {"effort": reasoning_effort}
            optional_params["reasoning"] = reasoning
        elif self._is_nova_2_model(model):
            reasoning_config: Final = self._transform_reasoning_effort_to_reasoning_config(reasoning_effort)
            optional_params.update(reasoning_config)
        else:
            mapped_thinking: Final = AnthropicConfig._map_reasoning_effort(
                reasoning_effort=reasoning_effort,
                model=model,
                custom_llm_provider="bedrock",
                llm_provider="bedrock_converse",
            )
            if mapped_thinking is None:
                optional_params.pop("thinking", None)
                optional_params.pop("output_config", None)
            else:
                optional_params["thinking"] = mapped_thinking
                if AnthropicConfig._is_adaptive_thinking_model(model, "bedrock"):
                    mapped_effort = REASONING_EFFORT_TO_OUTPUT_CONFIG_EFFORT.get(reasoning_effort)
                    if mapped_effort is None:
                        AnthropicConfig._raise_invalid_reasoning_effort(
                            model=model,
                            value=reasoning_effort,
                            llm_provider="bedrock_converse",
                        )
                    existing_output_config = optional_params.get("output_config")
                    if not isinstance(existing_output_config, dict):
                        existing_output_config = {}
                    existing_output_config.setdefault("effort", mapped_effort)
                    normalize_bedrock_opus_output_config_effort(
                        model=model,
                        output_config=existing_output_config,
                    )
                    mapped_effort = existing_output_config["effort"]
                    self._validate_anthropic_adaptive_effort(model=model, effort=mapped_effort)
                    optional_params["output_config"] = existing_output_config
                    optional_params["_output_config_normalized"] = True

    @staticmethod
    def _validate_anthropic_adaptive_effort(model: str, effort: str) -> None:
        """Validate ``output_config.effort`` for adaptive-thinking Claude 4.6/4.7."""
        valid_efforts: Final = {"high", "medium", "low", "xhigh", "max"}
        if effort not in valid_efforts:
            raise litellm.exceptions.BadRequestError(
                message=(
                    f"Invalid reasoning_effort/output_config.effort value: "
                    f"{effort!r}. Must be one of: 'low', 'medium', 'high', "
                    f"'xhigh', or 'max'."
                ),
                model=model,
                llm_provider="bedrock_converse",
            )
        error = AnthropicConfig._validate_effort_for_model(model=model, effort=effort, custom_llm_provider="bedrock")
        if error is not None:
            raise litellm.exceptions.BadRequestError(
                message=error,
                model=model,
                llm_provider="bedrock_converse",
            )

    @staticmethod
    def _clamp_thinking_budget_tokens(optional_params: dict) -> None:
        """
        Clamp thinking.budget_tokens to the Bedrock minimum (1024).

        Bedrock returns a 400 error if budget_tokens < 1024.
        """
        thinking: Final = optional_params.get("thinking")
        if isinstance(thinking, dict):
            budget: Final = thinking.get("budget_tokens")
            if isinstance(budget, int) and budget < BEDROCK_MIN_THINKING_BUDGET_TOKENS:
                verbose_logger.debug(
                    "Bedrock requires thinking.budget_tokens >= %d, got %d. Clamping to minimum.",
                    BEDROCK_MIN_THINKING_BUDGET_TOKENS,
                    budget,
                )
                thinking["budget_tokens"] = BEDROCK_MIN_THINKING_BUDGET_TOKENS

    def get_supported_openai_params(self, model: str) -> list[str]:
        from litellm.utils import supports_function_calling

        supported_params: Final = [
            "max_tokens",
            "max_completion_tokens",
            "stream",
            "stream_options",
            "stop",
            "temperature",
            "top_p",
            "extra_headers",
            "response_format",
            "requestMetadata",
            "service_tier",
            "parallel_tool_calls",
        ]

        if "arn" in model:  # we can't infer the model from the arn, so just add all params
            supported_params.append("tools")
            supported_params.append("tool_choice")
            supported_params.append("thinking")
            supported_params.append("reasoning_effort")
            supported_params.append("output_config")
            # For nova imported models, also add web_search_options
            if "nova" in model.lower():
                supported_params.append("web_search_options")
            return supported_params

        ## Filter out 'cross-region' from model name
        base_model: Final = BedrockModelInfo.get_base_model(model)

        if (
            base_model.startswith("anthropic")
            or base_model.startswith("mistral")
            or base_model.startswith("cohere")
            or base_model.startswith("meta.llama3-1")
            or base_model.startswith("meta.llama3-2")
            or base_model.startswith("meta.llama3-3")
            or base_model.startswith("meta.llama4")
            or base_model.startswith("amazon.nova")
            or supports_function_calling(model=model, custom_llm_provider=self.custom_llm_provider)
        ):
            supported_params.append("tools")

        # Nova models support web_search_options (mapped to nova_grounding systemTool)
        if base_model.startswith("amazon.nova"):
            supported_params.append("web_search_options")

        if litellm.utils.supports_tool_choice(
            model=model, custom_llm_provider=self.custom_llm_provider
        ) or litellm.utils.supports_tool_choice(model=base_model, custom_llm_provider=self.custom_llm_provider):
            # only anthropic and mistral support tool choice config. otherwise (E.g. cohere) will fail the call - https://docs.aws.amazon.com/bedrock/latest/APIReference/API_runtime_ToolChoice.html
            supported_params.append("tool_choice")

        if "gpt-oss" in model or "openai.gpt-5" in model or "openai.gpt-5" in base_model:
            supported_params.append("reasoning_effort")
        elif self._is_nova_2_model(model):
            # Nova 2 models support reasoning_effort (transformed to reasoningConfig)
            # These models use a different reasoning structure than Anthropic's thinking parameter
            supported_params.append("reasoning_effort")
        elif (
            "claude-3-7" in model
            or "claude-sonnet-4" in model
            or "claude-opus-4" in model
            or "deepseek.r1" in model
            or supports_reasoning(
                model=model,
                custom_llm_provider=self.custom_llm_provider,
            )
            or supports_reasoning(model=base_model, custom_llm_provider=self.custom_llm_provider)
        ):
            supported_params.append("thinking")
            supported_params.append("reasoning_effort")
            supported_params.append("output_config")

        if base_model.startswith("anthropic"):
            supported_params.append("context_management")
        return supported_params

    @staticmethod
    def _auto_tool_choice() -> ToolChoiceValuesBlock:
        return ToolChoiceValuesBlock(auto={})

    def map_tool_choice_values(
        self, model: str, tool_choice: str | dict, drop_params: bool
    ) -> ToolChoiceValuesBlock | None:
        if tool_choice == "none":
            if litellm.drop_params is True or drop_params is True:
                return None
            else:
                raise litellm.utils.UnsupportedParamsError(
                    message=f"Bedrock doesn't support tool_choice={tool_choice}. To drop it from the call, set `litellm.drop_params = True.",
                    status_code=400,
                )
        elif tool_choice == "required":
            if AnthropicModelInfo.forced_tool_use_downgraded(model, drop_params):
                return self._auto_tool_choice()
            return ToolChoiceValuesBlock(any={})
        elif tool_choice == "auto":
            return self._auto_tool_choice()
        elif isinstance(tool_choice, dict):
            if AnthropicModelInfo.forced_tool_use_downgraded(model, drop_params):
                return self._auto_tool_choice()
            # only supported for anthropic + mistral models - https://docs.aws.amazon.com/bedrock/latest/APIReference/API_runtime_ToolChoice.html
            specific_tool: Final = SpecificToolChoiceBlock(
                name=make_valid_bedrock_tool_name(tool_choice.get("function", {}).get("name", ""))
            )
            return ToolChoiceValuesBlock(tool=specific_tool)
        else:
            raise litellm.utils.UnsupportedParamsError(
                message=f"Bedrock doesn't support tool_choice={tool_choice}. Supported tool_choice values=['auto', 'required', json object]. To drop it from the call, set `litellm.drop_params = True.",
                status_code=400,
            )

    def get_supported_image_types(self) -> list[str]:
        return ["png", "jpeg", "gif", "webp"]

    def get_supported_document_types(self) -> list[str]:
        return ["pdf", "csv", "doc", "docx", "xls", "xlsx", "html", "txt", "md"]

    def get_supported_video_types(self) -> list[str]:
        return ["mp4", "mov", "mkv", "webm", "flv", "mpeg", "mpg", "wmv", "3gp"]

    def get_all_supported_content_types(self) -> list[str]:
        return self.get_supported_image_types() + self.get_supported_document_types() + self.get_supported_video_types()

    def is_computer_use_tool_used(self, tools: list[OpenAIChatCompletionToolParam] | None, model: str) -> bool:
        """Check if computer use tools are being used in the request."""
        if tools is None:
            return False

        for tool in tools:
            if "type" in tool:
                tool_type = tool["type"]
                for computer_use_prefix in BEDROCK_COMPUTER_USE_TOOLS:
                    if tool_type.startswith(computer_use_prefix):
                        return True
        return False

    def _transform_computer_use_tools(self, computer_use_tools: list[OpenAIChatCompletionToolParam]) -> list[dict]:
        """Transform computer use tools to Bedrock format."""
        transformed_tools: Final[list[dict]] = []

        for tool in computer_use_tools:
            tool_type = tool.get("type", "")

            # Check if this is a computer use tool with the startswith method
            is_computer_use_tool = False
            for computer_use_prefix in BEDROCK_COMPUTER_USE_TOOLS:
                if tool_type.startswith(computer_use_prefix):
                    is_computer_use_tool = True
                    break

            transformed_tool: dict = {}
            if is_computer_use_tool:
                if tool_type.startswith("computer_") and "function" in tool:
                    # Computer use tool with function format
                    func = tool["function"]
                    transformed_tool = {
                        "type": tool_type,
                        "name": func.get("name", "computer"),
                        **func.get("parameters", {}),
                    }
                else:
                    # Direct tools - just need to ensure name is present
                    transformed_tool = dict(tool)
                    if "name" not in transformed_tool:
                        if tool_type.startswith("bash_"):
                            transformed_tool["name"] = "bash"
                        elif tool_type.startswith("text_editor_"):
                            transformed_tool["name"] = "str_replace_editor"
            else:
                # Pass through other tools as-is
                transformed_tool = dict(tool)

            transformed_tools.append(transformed_tool)

        return transformed_tools

    def _separate_computer_use_tools(
        self, tools: list[OpenAIChatCompletionToolParam], model: str
    ) -> tuple[list[OpenAIChatCompletionToolParam], list[OpenAIChatCompletionToolParam]]:
        """
        Separate computer use tools from regular function tools.

        Args:
            tools: List of tools to separate
            model: The model name to check if it supports computer use

        Returns:
            Tuple of (computer_use_tools, regular_tools)
        """
        computer_use_tools: Final = []
        regular_tools: Final = []

        for tool in tools:
            if "type" in tool:
                tool_type = tool["type"]
                is_computer_use_tool = False
                for computer_use_prefix in BEDROCK_COMPUTER_USE_TOOLS:
                    if tool_type.startswith(computer_use_prefix):
                        is_computer_use_tool = True
                        break
                if is_computer_use_tool:
                    computer_use_tools.append(tool)
                else:
                    regular_tools.append(tool)
            else:
                regular_tools.append(tool)

        return computer_use_tools, regular_tools

    def _create_json_tool_call_for_response_format(
        self,
        json_schema: dict | None = None,
        description: str | None = None,
    ) -> ChatCompletionToolParam:
        """
        Handles creating a tool call for getting responses in JSON format.

        Args:
            json_schema (Optional[dict]): The JSON schema the response should be in

        Returns:
            AnthropicMessagesTool: The tool call to send to Anthropic API to get responses in JSON format
        """

        if json_schema is None:
            # Anthropic raises a 400 BadRequest error if properties is passed as None
            # see usage with additionalProperties (Example 5) https://github.com/anthropics/anthropic-cookbook/blob/main/tool_use/extracting_structured_json.ipynb
            _input_schema = {
                "type": "object",
                "additionalProperties": True,
                "properties": {},
            }
        else:
            # Use the schema as-is for Bedrock
            # Bedrock requires the tool schema to be of type "object" and doesn't need unwrapping
            _input_schema = json_schema

        tool_param_function_chunk: Final = ChatCompletionToolParamFunctionChunk(
            name=RESPONSE_FORMAT_TOOL_NAME, parameters=_input_schema
        )
        if description:
            tool_param_function_chunk["description"] = description

        _tool: Final = ChatCompletionToolParam(
            type="function",
            function=tool_param_function_chunk,
        )
        return _tool

    @staticmethod
    def _supports_native_structured_outputs(model: str, custom_llm_provider: str | None = None) -> bool:
        """Check if the Bedrock model supports native structured outputs (outputConfig.textFormat).

        Delegates to the standard ``supports_native_structured_output`` utility
        which looks up the flag in ``litellm.model_cost`` via
        ``_get_model_info_helper``.
        Ref: https://docs.aws.amazon.com/bedrock/latest/userguide/structured-output.html
        """
        from litellm.utils import supports_native_structured_output

        return supports_native_structured_output(model=model, custom_llm_provider=custom_llm_provider)

    @staticmethod
    def _add_additional_properties_to_schema(schema: dict) -> dict:
        """
        Recursively ensure all object types in a JSON schema have
        ``"additionalProperties": false``.

        Bedrock's native structured-outputs API requires this field to be
        explicitly set on every object node, otherwise it returns a
        validation error.
        """
        if not isinstance(schema, dict):
            return schema

        result: Final = dict(schema)

        if result.get("type") == "object" and "additionalProperties" not in result:
            result["additionalProperties"] = False

        # Recurse into nested schemas
        if "properties" in result and isinstance(result["properties"], dict):
            result["properties"] = {
                k: AmazonConverseConfig._add_additional_properties_to_schema(v) for k, v in result["properties"].items()
            }
        if "items" in result and isinstance(result["items"], dict):
            result["items"] = AmazonConverseConfig._add_additional_properties_to_schema(result["items"])
        for defs_key in ("$defs", "definitions"):
            if defs_key in result and isinstance(result[defs_key], dict):
                result[defs_key] = {
                    k: AmazonConverseConfig._add_additional_properties_to_schema(v) for k, v in result[defs_key].items()
                }
        for key in ("anyOf", "allOf", "oneOf"):
            if key in result and isinstance(result[key], list):
                result[key] = [AmazonConverseConfig._add_additional_properties_to_schema(item) for item in result[key]]

        return result

    @staticmethod
    def _create_output_config_for_response_format(
        json_schema: dict | None = None,
        name: str | None = None,
        description: str | None = None,
    ) -> "OutputConfigBlock":
        """
        Build an outputConfig block for Bedrock's native structured outputs API.

        The Converse API expects:
        {
            "outputConfig": {
                "textFormat": {
                    "type": "json_schema",
                    "structure": {
                        "jsonSchema": {
                            "schema": "<json-string>",
                            "name": "optional",
                            "description": "optional"
                        }
                    }
                }
            }
        }
        """
        if json_schema is not None:
            json_schema = AmazonConverseConfig._add_additional_properties_to_schema(json_schema)
        schema_str: Final = json.dumps(json_schema) if json_schema is not None else "{}"
        json_schema_def: Final[JsonSchemaDefinition] = {"schema": schema_str}
        if name is not None:
            json_schema_def["name"] = name
        if description is not None:
            json_schema_def["description"] = description

        return OutputConfigBlock(
            textFormat=OutputFormat(
                type="json_schema",
                structure=OutputFormatStructure(jsonSchema=json_schema_def),
            )
        )

    def _apply_tool_call_transformation(
        self,
        tools: list[OpenAIChatCompletionToolParam],
        model: str,
        non_default_params: dict,
        optional_params: dict,
    ):
        optional_params = self._add_tools_to_optional_params(optional_params=optional_params, tools=tools)

        if "meta.llama3-3-70b-instruct-v1:0" in model and non_default_params.get("stream", False) is True:
            optional_params["fake_stream"] = True

    def map_openai_params(
        self,
        non_default_params: dict,
        optional_params: dict,
        model: str,
        drop_params: bool,
    ) -> dict:
        is_thinking_enabled: Final = self.is_thinking_enabled(non_default_params)

        for param, value in non_default_params.items():
            if param == "response_format" and isinstance(value, dict):
                optional_params = self._translate_response_format_param(
                    value=value,
                    model=model,
                    optional_params=optional_params,
                    non_default_params=non_default_params,
                    is_thinking_enabled=is_thinking_enabled,
                )
            if param == "max_tokens" or param == "max_completion_tokens":
                optional_params["maxTokens"] = value
            if param == "stream":
                optional_params["stream"] = value
            if param == "stop":
                if isinstance(value, str):
                    if len(value) == 0:  # converse raises error for empty strings
                        continue
                    value = [value]
                optional_params["stopSequences"] = value
            if param == "temperature" or param == "top_p":
                AnthropicConfig._apply_sampling_param(
                    optional_params=optional_params,
                    model=model,
                    param=param,
                    value=value,
                    drop_params=drop_params,
                    output_key="topP" if param == "top_p" else param,
                )
            if param == "tools" and isinstance(value, list):
                self._apply_tool_call_transformation(
                    tools=cast(list[OpenAIChatCompletionToolParam], value),
                    model=model,
                    non_default_params=non_default_params,
                    optional_params=optional_params,
                )
            if param == "tool_choice":
                _tool_choice_value = self.map_tool_choice_values(
                    model=model,
                    tool_choice=value,
                    drop_params=drop_params,
                )
                if _tool_choice_value is not None:
                    optional_params["tool_choice"] = _tool_choice_value
            if param == "parallel_tool_calls":
                optional_params["_parallel_tool_use_config"] = {
                    "tool_choice": {"type": "auto", "disable_parallel_tool_use": not value}
                }
            if param == "thinking" and "openai.gpt-5" not in model:
                if (
                    isinstance(value, dict)
                    and value.get("type") == "adaptive"
                    and not AnthropicConfig._is_adaptive_thinking_model(model, "bedrock")
                ):
                    max_tokens = non_default_params.get("max_completion_tokens") or non_default_params.get("max_tokens")
                    legacy_thinking = AnthropicConfig._map_reasoning_effort(
                        reasoning_effort="medium",
                        model=model,
                        custom_llm_provider="bedrock",
                    )
                    capped = (
                        AnthropicConfig.cap_thinking_budget_to_max_tokens(legacy_thinking, max_tokens)
                        if legacy_thinking is not None
                        else None
                    )
                    if capped is not None:
                        optional_params["thinking"] = capped
                    else:
                        litellm.verbose_logger.warning(DROP_UNSUPPORTED_ADAPTIVE_THINKING_WARNING, model)
                else:
                    optional_params["thinking"] = value
            elif param == "reasoning_effort" and isinstance(value, str):
                self._handle_reasoning_effort_parameter(
                    model=model, reasoning_effort=value, optional_params=optional_params
                )
            elif param == "output_config" and isinstance(value, dict):
                mapped_output_config = dict(value)
                normalize_bedrock_opus_output_config_effort(model=model, output_config=mapped_output_config)
                optional_params["output_config"] = mapped_output_config  # rebind-ok: out-param store like siblings
            elif param == "context_management" and isinstance(value, (dict, list)):
                self._map_context_management_param(value, optional_params)
            if param == "requestMetadata":
                self._map_request_metadata_param(value, optional_params)
            if param == "service_tier" and isinstance(value, str):
                self._map_service_tier_param(value, optional_params)

            if param == "web_search_options" and isinstance(value, dict):
                # Note: we use `isinstance(value, dict)` instead of `value and isinstance(value, dict)`
                # because empty dict {} is falsy but is a valid way to enable Nova grounding
                grounding_tool = self._map_web_search_options(value, model)
                if grounding_tool is not None:
                    optional_params = self._add_tools_to_optional_params(
                        optional_params=optional_params, tools=[grounding_tool]
                    )

        # Only update thinking tokens for non-GPT-OSS models and non-Nova-Lite-2 models
        # Nova 2 handles token budgeting differently through reasoningConfig
        if "gpt-oss" not in model and not self._is_nova_2_model(model):
            self.update_optional_params_with_thinking_tokens(
                non_default_params=non_default_params, optional_params=optional_params
            )

        final_is_thinking_enabled: Final = self.is_thinking_enabled(optional_params)
        if final_is_thinking_enabled and "tool_choice" in optional_params:
            tool_choice_block: Final = optional_params["tool_choice"]
            if isinstance(tool_choice_block, dict):
                if "any" in tool_choice_block or "tool" in tool_choice_block:
                    verbose_logger.info(
                        "%s does not support forced tool use (tool_choice='required' or specific tool) when reasoning is enabled. Changing tool_choice to 'auto'.",
                        model,
                    )
                    optional_params["tool_choice"] = ToolChoiceValuesBlock(auto={})

        return optional_params

    def _map_request_metadata_param(self, value: Any, optional_params: dict) -> None:
        if value is not None and isinstance(value, dict):
            self._validate_request_metadata(value)
            optional_params["requestMetadata"] = value

    def _map_context_management_param(self, value: dict | list, optional_params: dict) -> None:
        # Match the dispatcher's ``_normalize_spec`` behavior: only run the
        # OpenAI→Anthropic mapper for list inputs. Dict inputs are already in
        # Anthropic-native shape (``{"edits": [...]}``) and should pass
        # through unchanged so an Anthropic-format ``context_management``
        # value isn't silently dropped when the mapper can't classify it.
        if isinstance(value, list):
            mapped = AnthropicConfig.map_openai_context_management_to_anthropic(cast(dict | list, value))
        else:
            mapped = value
        # Skip when the mapper returned None for malformed input — leaving the
        # key out is safer than passing `context_management: null` downstream,
        # which Bedrock would reject and which can confuse intermediate checks
        # before the final _filter_context_management_for_bedrock_converse step.
        if mapped is not None:
            optional_params["context_management"] = mapped

    def _map_service_tier_param(self, value: str, optional_params: dict) -> None:
        """Map OpenAI service_tier (string) to Bedrock serviceTier (object).

        OpenAI values: "auto", "default", "flex", "priority"
        Bedrock values: "default", "flex", "priority" (no "auto")
        """
        bedrock_tier = value
        if value == "auto":
            bedrock_tier = "default"  # Bedrock doesn't support "auto"
        if bedrock_tier in ("default", "flex", "priority"):
            optional_params["serviceTier"] = {"type": bedrock_tier}

    def _translate_response_format_param(
        self,
        value: dict,
        model: str,
        optional_params: dict,
        non_default_params: dict,
        is_thinking_enabled: bool,
    ) -> dict:
        """
        Handles translation of response_format parameter to Bedrock format.

        Returns `optional_params` with the translated response_format parameter.
        """
        ignore_response_format_types: Final = ["text"]
        if value["type"] in ignore_response_format_types:  # value is a no-op
            return optional_params

        json_schema: dict | None = None
        name: str | None = None
        description: str | None = None
        if "response_schema" in value:
            json_schema = value["response_schema"]
        elif "json_schema" in value:
            json_schema = value["json_schema"]["schema"]
            name = value["json_schema"].get("name")
            description = value["json_schema"].get("description")

        if "type" in value and value["type"] == "text":
            return optional_params

        if self._supports_native_structured_outputs(model, self.custom_llm_provider) and json_schema is not None:
            # Use Bedrock's native structured outputs API (outputConfig.textFormat)
            # No synthetic tool injection, no fake_stream needed.
            # Requires an explicit schema — json_object with no schema falls through
            # to the tool-call path below.
            output_config: Final = self._create_output_config_for_response_format(
                json_schema=json_schema,
                name=name,
                description=description,
            )
            optional_params["outputConfig"] = output_config
        elif json_schema is not None:
            # Fallback: translate to a synthetic tool call
            # https://docs.anthropic.com/en/docs/build-with-claude/tool-use#json-mode
            _tool: Final = self._create_json_tool_call_for_response_format(
                json_schema=json_schema,
                description=description,
            )
            optional_params = self._add_tools_to_optional_params(optional_params=optional_params, tools=[_tool])

            if (
                litellm.utils.supports_tool_choice(model=model, custom_llm_provider=self.custom_llm_provider)
                and not is_thinking_enabled
                and not AnthropicModelInfo.forced_tool_use_unsupported(model)
            ):
                optional_params["tool_choice"] = ToolChoiceValuesBlock(
                    tool=SpecificToolChoiceBlock(name=RESPONSE_FORMAT_TOOL_NAME)
                )
            if non_default_params.get("stream", False) is True:
                optional_params["fake_stream"] = True
        # else: response_format=json_object with no schema.
        # Don't inject the synthetic json_tool_call tool here. When no
        # schema is given, _create_json_tool_call_for_response_format
        # produces an empty schema (properties: {}), and the model
        # returns {} instead of the requested JSON. The model already
        # returns JSON when the prompt asks for it.

        optional_params["json_mode"] = True
        return optional_params

    def update_optional_params_with_thinking_tokens(self, non_default_params: dict, optional_params: dict):
        """
        Handles scenario where max tokens is not specified. For anthropic models (anthropic api/bedrock/vertex ai), this requires having the max tokens being set and being greater than the thinking token budget.

        Checks 'non_default_params' for 'thinking' and 'max_tokens'

        if 'thinking' is enabled and 'max_tokens' is not specified, set 'max_tokens' to the thinking token budget + DEFAULT_MAX_TOKENS

        Also clamps thinking.budget_tokens to the Bedrock minimum (1024) to
        prevent 400 errors from the Bedrock API.
        """
        from litellm.constants import DEFAULT_MAX_TOKENS

        self._clamp_thinking_budget_tokens(optional_params)

        is_thinking_enabled: Final = self.is_thinking_enabled(optional_params)
        is_max_tokens_in_request: Final = self.is_max_tokens_in_request(non_default_params)
        if is_thinking_enabled and not is_max_tokens_in_request:
            thinking_value: Final = optional_params.get("thinking")
            thinking_token_budget: Final = (
                thinking_value.get("budget_tokens") if isinstance(thinking_value, dict) else None
            )
            if thinking_token_budget is not None:
                optional_params["maxTokens"] = thinking_token_budget + DEFAULT_MAX_TOKENS

    @overload
    def get_cache_point_block(
        self,
        message_block: OpenAIMessageContentListBlock
        | ChatCompletionUserMessage
        | ChatCompletionSystemMessage
        | ChatCompletionAssistantMessage,
        block_type: Literal["system"],
        model: str | None = None,
    ) -> SystemContentBlock | None:
        pass

    @overload
    def get_cache_point_block(
        self,
        message_block: OpenAIMessageContentListBlock
        | ChatCompletionUserMessage
        | ChatCompletionSystemMessage
        | ChatCompletionAssistantMessage,
        block_type: Literal["content_block"],
        model: str | None = None,
    ) -> ContentBlock | None:
        pass

    def get_cache_point_block(
        self,
        message_block: OpenAIMessageContentListBlock
        | ChatCompletionUserMessage
        | ChatCompletionSystemMessage
        | ChatCompletionAssistantMessage,
        block_type: Literal["system", "content_block"],
        model: str | None = None,
    ) -> SystemContentBlock | ContentBlock | None:
        cache_control: Final = message_block.get("cache_control", None)
        if cache_control is None or not bedrock_model_accepts_cache_points(model):
            return None

        cache_point: Final = self._build_cache_point_block(cache_control, model)

        if block_type == "system":
            return SystemContentBlock(cachePoint=cache_point)
        else:
            return ContentBlock(cachePoint=cache_point)

    @staticmethod
    def _build_cache_point_block(control: dict | None, model: str | None = None) -> CachePointBlock:
        """Build a Bedrock ``cachePoint`` block from an OpenAI-style ``cache_control``/``control`` dict.

        ``type`` is always ``"default"`` (the only value Bedrock's Converse API
        accepts). ``ttl`` is only honored for models that support extended TTL
        caching (Claude 4.5 family on Bedrock).
        """
        cache_point: Final = CachePointBlock(type="default")
        if isinstance(control, dict) and "ttl" in control:
            ttl: Final = control["ttl"]
            if ttl in ["5m", "1h"] and model is not None and is_claude_4_5_on_bedrock(model):
                cache_point["ttl"] = ttl
        return cache_point

    def _transform_system_message(
        self, messages: list[AllMessageValues], model: str | None = None
    ) -> tuple[list[AllMessageValues], list[SystemContentBlock]]:
        system_prompt_indices: Final = []
        system_content_blocks: Final[list[SystemContentBlock]] = []
        for idx, message in enumerate(messages):
            if message["role"] == "system":
                system_prompt_indices.append(idx)
                if isinstance(message["content"], str) and message["content"]:
                    system_content_blocks.append(SystemContentBlock(text=message["content"]))
                    cache_block = self.get_cache_point_block(message, block_type="system", model=model)
                    if cache_block:
                        system_content_blocks.append(cache_block)
                elif isinstance(message["content"], list):
                    for m in message["content"]:
                        if m.get("type") == "text" and m.get("text"):
                            system_content_blocks.append(SystemContentBlock(text=m["text"]))
                            cache_block = self.get_cache_point_block(m, block_type="system", model=model)
                            if cache_block:
                                system_content_blocks.append(cache_block)
        if len(system_prompt_indices) > 0:
            for idx in reversed(system_prompt_indices):
                messages.pop(idx)
        return messages, system_content_blocks

    def _transform_inference_params(self, inference_params: dict) -> InferenceConfig:
        if "top_k" in inference_params:
            inference_params["topK"] = inference_params.pop("top_k")
        return InferenceConfig(**inference_params)

    def _handle_top_k_value(self, model: str, inference_params: dict, drop_params: bool = False) -> dict:
        base_model: Final = BedrockModelInfo.get_base_model(model)

        val_top_k = None
        if "topK" in inference_params:
            val_top_k = inference_params.pop("topK")
        elif "top_k" in inference_params:
            val_top_k = inference_params.pop("top_k")

        if val_top_k is not None:
            if base_model.startswith("anthropic"):
                top_k_params: Final[dict] = {}
                AnthropicConfig._apply_sampling_param(
                    optional_params=top_k_params,
                    model=model,
                    param="top_k",
                    value=val_top_k,
                    drop_params=drop_params,
                    output_key="top_k",
                )
                return top_k_params
            if base_model.startswith("amazon.nova"):
                return {"inferenceConfig": {"topK": val_top_k}}

        return {}

    @staticmethod
    def _merge_parallel_tool_use_config(additional_request_params: dict, parallel_tool_use_config: dict) -> dict:
        merged_entries: Final = {
            key: (
                {
                    **value,
                    **additional_request_params[key],
                    **{k: v for k, v in value.items() if k != "type"},
                }
                if isinstance(additional_request_params.get(key), dict) and isinstance(value, dict)
                else value
            )
            for key, value in parallel_tool_use_config.items()
        }
        return {**additional_request_params, **merged_entries}

    @staticmethod
    def _drop_tool_choice_type_conflicting_with_tool_config(additional_request_params: dict) -> None:
        """Drop ``tool_choice.type`` from the Anthropic passthrough fields.

        Converse rejects a request carrying both ``toolConfig.toolChoice`` and an
        ``additionalModelRequestFields.tool_choice.type``, so once the caller asked for a
        tool choice the type has to come from ``toolChoice`` alone. Sibling keys such as
        ``disable_parallel_tool_use`` have no ``toolConfig`` equivalent and are accepted
        alongside ``toolChoice``, so they stay.
        """
        tool_choice = additional_request_params.get("tool_choice")
        if isinstance(tool_choice, dict):
            tool_choice.pop("type", None)

    def _prepare_request_params(
        self, optional_params: dict, model: str, drop_params: bool = False
    ) -> tuple[dict, dict, dict, OutputConfigBlock | None]:
        """Prepare and separate request parameters."""
        # Consume the internal ``_output_config_normalized`` marker set by
        # ``_handle_reasoning_effort_parameter`` so it does not linger on the
        # caller's ``optional_params`` after the transformation returns.
        anthropic_output_config_already_normalized = bool(optional_params.pop("_output_config_normalized", False))
        # Filter out exception objects before deepcopy to prevent deepcopy failures
        # Exceptions should not be stored in optional_params (this is a defensive fix)
        cleaned_params: Final = filter_exceptions_from_params(optional_params)
        inference_params = safe_deep_copy(cleaned_params)
        supported_converse_params: Final = list(AmazonConverseConfig.__annotations__.keys()) + ["top_k"]
        supported_tool_call_params: Final = ["tools", "tool_choice"]
        supported_config_params: Final = list(self.get_config_blocks().keys())
        total_supported_params: Final = supported_converse_params + supported_tool_call_params + supported_config_params
        inference_params.pop("json_mode", None)  # used for handling json_schema

        # Anthropic-only ``output_config`` (snake_case) — re-attached to
        # ``additionalModelRequestFields`` for Anthropic models below. The
        # structured-output ``format`` subfield is consumed into Bedrock's
        # native ``outputConfig`` (camelCase), which is handled separately.
        anthropic_output_config = inference_params.pop("output_config", None)
        output_config_format = None
        if isinstance(anthropic_output_config, dict):
            anthropic_output_config = dict(anthropic_output_config)
            candidate_output_config_format: Final = anthropic_output_config.pop("format", None)
            if isinstance(candidate_output_config_format, dict):
                output_config_format = candidate_output_config_format
            if not anthropic_output_config:
                anthropic_output_config = None

        # Extract requestMetadata before processing other parameters
        request_metadata: Final = inference_params.pop("requestMetadata", None)
        if request_metadata is not None:
            self._validate_request_metadata(request_metadata)

        output_config: OutputConfigBlock | None = inference_params.pop("outputConfig", None)
        base_model: Final = BedrockModelInfo.get_base_model(model)
        if (
            output_config is None
            and output_config_format is not None
            and output_config_format.get("type") == "json_schema"
            and base_model.startswith("anthropic")
            and self._supports_native_structured_outputs(model, self.custom_llm_provider)
        ):
            output_config = self._create_output_config_for_response_format(
                json_schema=output_config_format.get("schema"),
                name=output_config_format.get("name"),
                description=output_config_format.get("description"),
            )
        elif output_config is None and output_config_format is not None:
            litellm.verbose_logger.warning(
                "Bedrock Converse: dropping `output_config.format` for model=%s — "
                "model does not advertise `supports_native_structured_output` in "
                "model_prices_and_context_window.json. The schema will not be "
                "enforced; pass `response_format` to use the synthetic tool-call "
                "fallback.",
                model,
            )

        # keep supported params in 'inference_params', and set all model-specific params in 'additional_request_params'
        additional_request_params = {k: v for k, v in inference_params.items() if k not in total_supported_params}
        inference_params = {k: v for k, v in inference_params.items() if k in total_supported_params}

        # Handle parallel_tool_calls configuration
        parallel_tool_use_config: Final = additional_request_params.pop("_parallel_tool_use_config", None)
        if parallel_tool_use_config is not None and bedrock_converse_supports_parallel_tool_use_config(model):
            additional_request_params = self._merge_parallel_tool_use_config(
                additional_request_params, parallel_tool_use_config
            )

        additional_request_params.pop("parallel_tool_calls", None)

        # Only set the topK value in for models that support it
        additional_request_params.update(self._handle_top_k_value(model, inference_params, drop_params))

        # Filter out internal/MCP-related parameters that shouldn't be sent to the API
        # These are LiteLLM internal parameters, not API parameters
        additional_request_params = filter_internal_params(additional_request_params)

        # Filter out non-serializable objects (exceptions, callables, logging objects, etc.)
        # from additional_request_params to prevent JSON serialization errors
        # This filters: Exception objects, callable objects (functions), Logging objects, etc.
        additional_request_params = filter_exceptions_from_params(additional_request_params)

        if anthropic_output_config is not None and isinstance(anthropic_output_config, dict):
            # Application inference profile ARNs hide the underlying model, so the
            # effort ceiling and capability gates below cannot run; forward
            # verbatim (like ``thinking``) and let Bedrock enforce.
            if is_bedrock_application_inference_profile_arn(model):
                additional_request_params["output_config"] = anthropic_output_config
            elif base_model.startswith("anthropic"):
                if litellm.drop_params is True and not AnthropicConfig._model_supports_effort_param(model, "bedrock"):
                    litellm.verbose_logger.warning(
                        DROP_UNSUPPORTED_OUTPUT_CONFIG_WARNING,
                        model,
                    )
                else:
                    if not anthropic_output_config_already_normalized:
                        normalize_bedrock_opus_output_config_effort(
                            model=model,
                            output_config=anthropic_output_config,
                        )
                    effort: Final = anthropic_output_config.get("effort")
                    if effort is not None:
                        self._validate_anthropic_adaptive_effort(model=model, effort=effort)
                    additional_request_params["output_config"] = anthropic_output_config

        return (
            inference_params,
            additional_request_params,
            request_metadata,
            output_config,
        )

    def _process_tools_and_beta(
        self,
        original_tools: list,
        model: str,
        headers: dict | None,
        additional_request_params: dict,
    ) -> tuple[list[ToolBlock], list]:
        """Process tools and collect anthropic_beta values."""
        bedrock_tools: list[ToolBlock] = []

        # Collect anthropic_beta values from user headers
        anthropic_beta_list: Final = []
        if headers:
            user_betas: Final = get_anthropic_beta_from_headers(headers)
            anthropic_beta_list.extend(user_betas)

        # Separate pre-formatted Bedrock tools (e.g. systemTool from web_search_options)
        # from OpenAI-format tools that need transformation via _bedrock_tools_pt
        filtered_tools: Final = []
        pre_formatted_tools: Final[list[ToolBlock]] = []
        if original_tools:
            for tool in original_tools:
                # Already-formatted Bedrock tools (e.g. systemTool for Nova grounding)
                if "systemTool" in tool:
                    pre_formatted_tools.append(tool)
                    continue
                tool_type = tool.get("type", "")
                if tool_type in (
                    "tool_search_tool_regex_20251119",
                    "tool_search_tool_bm25_20251119",
                ):
                    # Tool search not supported in Converse API - skip it
                    continue
                filtered_tools.append(tool)

        # Only separate tools if computer use tools are actually present
        if filtered_tools and self.is_computer_use_tool_used(filtered_tools, model):
            # Separate computer use tools from regular function tools
            computer_use_tools, regular_tools = self._separate_computer_use_tools(filtered_tools, model)

            # Process regular function tools using existing logic
            bedrock_tools = _bedrock_tools_pt(regular_tools, model=model)

            # Add computer use tools and anthropic_beta if needed (only when computer use tools are present)
            if computer_use_tools:
                # Determine the correct computer-use beta header based on model
                # "computer-use-2025-11-24" for Claude Opus 4.7, Opus 4.6, and Opus 4.5
                # "computer-use-2025-01-24" for Claude Sonnet 4.5, Haiku 4.5, Opus 4.1, Sonnet 4, Opus 4, and Sonnet 3.7
                # "computer-use-2024-10-22" for older models
                model_lower: Final = model.lower()
                if (
                    "opus-4.7" in model_lower
                    or "opus_4.7" in model_lower
                    or "opus-4-7" in model_lower
                    or "opus_4_7" in model_lower
                    or "opus-4.6" in model_lower
                    or "opus_4.6" in model_lower
                    or "opus-4-6" in model_lower
                    or "opus_4_6" in model_lower
                    or "sonnet-4.6" in model_lower
                    or "sonnet_4.6" in model_lower
                    or "sonnet-4-6" in model_lower
                    or "sonnet_4_6" in model_lower
                ) or (
                    "opus-4.5" in model_lower
                    or "opus_4.5" in model_lower
                    or "opus-4-5" in model_lower
                    or "opus_4_5" in model_lower
                ):
                    computer_use_header = "computer-use-2025-11-24"
                elif any(
                    pattern in model_lower
                    for pattern in [
                        "sonnet-4.5",
                        "sonnet_4.5",
                        "sonnet-4-5",
                        "sonnet_4_5",
                        "haiku-4.5",
                        "haiku_4.5",
                        "haiku-4-5",
                        "haiku_4_5",
                        "opus-4.1",
                        "opus_4.1",
                        "opus-4-1",
                        "opus_4_1",
                        "sonnet-4",
                        "sonnet_4",
                        "opus-4",
                        "opus_4",
                        "sonnet-3.7",
                        "sonnet_3.7",
                        "sonnet-3-7",
                        "sonnet_3_7",
                    ]
                ):
                    computer_use_header = "computer-use-2025-01-24"
                else:
                    computer_use_header = "computer-use-2024-10-22"

                anthropic_beta_list.append(computer_use_header)
                # Transform computer use tools to proper Bedrock format
                transformed_computer_tools: Final = self._transform_computer_use_tools(computer_use_tools)
                additional_request_params["tools"] = transformed_computer_tools
        else:
            # No computer use tools, process all tools as regular tools
            bedrock_tools = _bedrock_tools_pt(filtered_tools, model=model)

        # Append pre-formatted tools (systemTool etc.) after transformation
        bedrock_tools.extend(pre_formatted_tools)

        # Opus 4.5 gates ``output_config.effort`` behind a beta header;
        # Claude 4.6/4.7 accept it without one.
        base_model: Final = BedrockModelInfo.get_base_model(model)
        if base_model.startswith("anthropic"):
            output_config: Final = additional_request_params.get("output_config")
            if (
                isinstance(output_config, dict)
                and output_config.get("effort") is not None
                and not AnthropicConfig._is_adaptive_thinking_model(model, "bedrock")
            ):
                from litellm.types.llms.anthropic import (
                    ANTHROPIC_EFFORT_BETA_HEADER,
                )

                if ANTHROPIC_EFFORT_BETA_HEADER not in anthropic_beta_list:
                    anthropic_beta_list.append(ANTHROPIC_EFFORT_BETA_HEADER)

            # Bedrock Converse: compact_20260112 edits only (+ beta header).
            AmazonConverseConfig._filter_context_management_for_bedrock_converse(
                additional_request_params, anthropic_beta_list
            )

        # Set anthropic_beta in additional_request_params if we have any beta features
        # ONLY apply to Anthropic/Claude models - other models (e.g., Qwen, Llama) don't support this field
        if anthropic_beta_list and base_model.startswith("anthropic"):
            additional_request_params["anthropic_beta"] = anthropic_beta_list

        return bedrock_tools, anthropic_beta_list

    @staticmethod
    def _filter_context_management_for_bedrock_converse(
        additional_request_params: dict,
        anthropic_beta_list: list,
    ) -> None:
        """Keep only compact_20260112 edits for Bedrock; add beta header or drop field."""
        from litellm.llms.anthropic.experimental_pass_through.context_management.constants import (
            COMPACT_EDIT_TYPE,
        )
        from litellm.types.llms.anthropic import ANTHROPIC_BETA_HEADER_VALUES

        cm: Final = additional_request_params.get("context_management")
        if not isinstance(cm, dict):
            additional_request_params.pop("context_management", None)
            return
        edits: Final = cm.get("edits")
        if not isinstance(edits, list):
            additional_request_params.pop("context_management", None)
            return

        compact_edits: Final = [e for e in edits if isinstance(e, dict) and e.get("type") == COMPACT_EDIT_TYPE]
        if compact_edits:
            compact_beta: Final = ANTHROPIC_BETA_HEADER_VALUES.COMPACT_2026_01_12.value
            if compact_beta not in anthropic_beta_list:
                anthropic_beta_list.append(compact_beta)
            additional_request_params["context_management"] = {
                **cm,
                "edits": compact_edits,
            }
        else:
            additional_request_params.pop("context_management", None)

    def _transform_request_helper(
        self,
        model: str,
        system_content_blocks: list[SystemContentBlock],
        optional_params: dict,
        messages: list[AllMessageValues] | None = None,
        headers: dict | None = None,
        drop_params: bool = False,
        litellm_params: Mapping[str, object] | None = None,
    ) -> CommonRequestObject:
        ## VALIDATE REQUEST
        """
        Bedrock doesn't support tool calling without `tools=` param specified.
        """
        if "tools" not in optional_params and messages is not None and has_tool_call_blocks(messages):
            if litellm.modify_params:
                optional_params["tools"] = add_dummy_tool(custom_llm_provider="bedrock_converse")
            else:
                raise litellm.UnsupportedParamsError(
                    message="Bedrock doesn't support tool calling without `tools=` param specified. Pass `tools=` param OR set `litellm.modify_params = True` // `litellm_settings::modify_params: True` to add dummy tool to the request.",
                    model="",
                    llm_provider="bedrock",
                )

        # Drop thinking param if thinking is enabled but thinking_blocks are missing
        # This prevents the error: "Expected thinking or redacted_thinking, but found tool_use"
        #
        # IMPORTANT: Only drop thinking if NO assistant messages have thinking_blocks.
        # If any message has thinking_blocks, we must keep thinking enabled, otherwise
        # Related issues: https://github.com/BerriAI/litellm/issues/14194
        if (
            optional_params.get("thinking") is not None
            and messages is not None
            and last_assistant_with_tool_calls_has_no_thinking_blocks(messages)
            and not any_assistant_message_has_thinking_blocks(messages)
        ):
            if litellm.modify_params:
                optional_params.pop("thinking", None)
                litellm.verbose_logger.warning(
                    "Dropping 'thinking' param because the last assistant message with tool_calls "
                    "has no thinking_blocks. The model won't use extended thinking for this turn."
                )

        AnthropicModelInfo.maybe_drop_disabled_thinking(
            model=model,
            optional_params=optional_params,
            custom_llm_provider="bedrock",
        )

        # Prepare and separate parameters
        (
            inference_params,
            additional_request_params,
            request_metadata,
            output_config,
        ) = self._prepare_request_params(optional_params, model, drop_params)

        original_tools: Final = inference_params.pop("tools", [])

        # Process tools and collect beta values
        bedrock_tools, anthropic_beta_list = self._process_tools_and_beta(
            original_tools, model, headers, additional_request_params
        )

        # Append cachePoint to tools if cache_control_injection_points has tool_config
        cache_injection_points: Final = additional_request_params.pop("cache_control_injection_points", None)
        if cache_injection_points and len(bedrock_tools) > 0 and bedrock_model_accepts_cache_points(model):
            for point in cache_injection_points:
                if point.get("location") == "tool_config":
                    cache_point = self._build_cache_point_block(point.get("control"), model)
                    bedrock_tools.append(ToolBlock(cachePoint=cache_point))
                    # Spend attribution credits the gateway only for breakpoints it placed, and
                    # this is the one place a tool_config point becomes one. The hook that reads
                    # the configuration cannot record it: whether a cachePoint lands depends on
                    # this provider and on the request carrying tools, neither of which it sees.
                    if litellm_params is not None:
                        from litellm.integrations.anthropic_cache_control_hook import (
                            AnthropicCacheControlHook,
                        )

                        AnthropicCacheControlHook.record_gateway_injection(litellm_params, 1)
                    break

        bedrock_tool_config: ToolConfigBlock | None = None
        if len(bedrock_tools) > 0:
            tool_choice_values: Final[ToolChoiceValuesBlock] = inference_params.pop("tool_choice", None)
            bedrock_tool_config = ToolConfigBlock(
                tools=bedrock_tools,
            )
            if tool_choice_values is not None:
                bedrock_tool_config["toolChoice"] = tool_choice_values
                self._drop_tool_choice_type_conflicting_with_tool_config(additional_request_params)

        config_block_entries: Final = tuple(
            (config_name, config_class, inference_params.pop(config_name, None))
            for config_name, config_class in self.get_config_blocks().items()
        )

        data: Final[CommonRequestObject] = {
            "inferenceConfig": self._transform_inference_params(inference_params=inference_params),
        }
        if additional_request_params:
            data["additionalModelRequestFields"] = additional_request_params
            if "thinking" in additional_request_params:
                data["additionalModelResponseFieldPaths"] = ("/usage/output_tokens_details",)
        if system_content_blocks:
            data["system"] = system_content_blocks

        for config_name, config_class, config_value in config_block_entries:
            if config_value is not None:
                data[config_name] = config_class(**config_value)

        # Tool Config
        if bedrock_tool_config is not None:
            data["toolConfig"] = bedrock_tool_config

        # Request Metadata (top-level field)
        if request_metadata is not None:
            data["requestMetadata"] = request_metadata

        if output_config is not None:
            data["outputConfig"] = output_config

        return data

    async def _async_transform_request(
        self,
        model: str,
        messages: list[AllMessageValues],
        optional_params: dict,
        litellm_params: dict,
        headers: dict | None = None,
    ) -> RequestObject:
        messages, system_content_blocks = self._transform_system_message(messages, model=model)

        # Convert last user message to guarded_text if guardrailConfig is present
        messages = self._convert_consecutive_user_messages_to_guarded_text(messages, optional_params)
        ## TRANSFORMATION ##

        _data: Final[CommonRequestObject] = self._transform_request_helper(
            model=model,
            system_content_blocks=system_content_blocks,
            optional_params=optional_params,
            messages=messages,
            headers=headers,
            drop_params=litellm_params.get("drop_params") is True,
            litellm_params=litellm_params,
        )

        bedrock_messages: Final = await BedrockConverseMessagesProcessor._bedrock_converse_messages_pt_async(
            messages=messages,
            model=model,
            llm_provider="bedrock_converse",
            user_continue_message=litellm_params.pop("user_continue_message", None),
        )

        request_metadata: Final = resolve_bedrock_request_metadata(
            litellm_params=litellm_params, caller_metadata=_data.get("requestMetadata")
        )
        if bedrock_request_metadata_is_owned():
            _data.pop("requestMetadata", None)
            if request_metadata is not None:
                _data["requestMetadata"] = request_metadata
        data: Final[RequestObject] = {"messages": bedrock_messages, **_data}

        return data

    def transform_request(
        self,
        model: str,
        messages: list[AllMessageValues],
        optional_params: dict,
        litellm_params: dict,
        headers: dict,
    ) -> dict:
        return cast(
            dict,
            self._transform_request(
                model=model,
                messages=messages,
                optional_params=optional_params,
                litellm_params=litellm_params,
                headers=headers,
            ),
        )

    def _transform_request(
        self,
        model: str,
        messages: list[AllMessageValues],
        optional_params: dict,
        litellm_params: dict,
        headers: dict | None = None,
    ) -> RequestObject:
        messages, system_content_blocks = self._transform_system_message(messages, model=model)

        # Convert last user message to guarded_text if guardrailConfig is present
        messages = self._convert_consecutive_user_messages_to_guarded_text(messages, optional_params)

        _data: Final[CommonRequestObject] = self._transform_request_helper(
            model=model,
            system_content_blocks=system_content_blocks,
            optional_params=optional_params,
            messages=messages,
            headers=headers,
            drop_params=litellm_params.get("drop_params") is True,
            litellm_params=litellm_params,
        )

        ## TRANSFORMATION ##
        bedrock_messages: Final[list[MessageBlock]] = _bedrock_converse_messages_pt(
            messages=messages,
            model=model,
            llm_provider="bedrock_converse",
            user_continue_message=litellm_params.pop("user_continue_message", None),
        )

        request_metadata: Final = resolve_bedrock_request_metadata(
            litellm_params=litellm_params, caller_metadata=_data.get("requestMetadata")
        )
        if bedrock_request_metadata_is_owned():
            _data.pop("requestMetadata", None)
            if request_metadata is not None:
                _data["requestMetadata"] = request_metadata
        data: Final[RequestObject] = {"messages": bedrock_messages, **_data}

        return data

    def transform_response(
        self,
        model: str,
        raw_response: httpx.Response,
        model_response: ModelResponse,
        logging_obj: Logging,
        request_data: dict,
        messages: list[AllMessageValues],
        optional_params: dict,
        litellm_params: dict,
        encoding: "tiktoken.Encoding | None",
        api_key: str | None = None,
        json_mode: bool | None = None,
    ) -> ModelResponse:
        return self._transform_response(
            model=model,
            response=raw_response,
            model_response=model_response,
            stream=optional_params.get("stream", False),
            logging_obj=logging_obj,
            optional_params=optional_params,
            api_key=api_key,
            data=request_data,
            messages=messages,
            encoding=encoding,
        )

    def _transform_reasoning_content(self, reasoning_content_blocks: list[BedrockConverseReasoningContentBlock]) -> str:
        """
        Extract the reasoning text from the reasoning content blocks

        Ensures deepseek reasoning content compatible output.
        """
        reasoning_content_str = ""
        for block in reasoning_content_blocks:
            if "reasoningText" in block:
                reasoning_content_str += block["reasoningText"]["text"]
        return reasoning_content_str

    def _transform_thinking_blocks(
        self, thinking_blocks: list[BedrockConverseReasoningContentBlock]
    ) -> list[ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock]:
        """Return a consistent format for thinking blocks between Anthropic and Bedrock."""
        thinking_blocks_list: Final[list[ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock]] = []
        for block in thinking_blocks:
            if "reasoningText" in block:
                _thinking_block = ChatCompletionThinkingBlock(type="thinking")
                _text = block["reasoningText"].get("text")
                _signature = block["reasoningText"].get("signature")
                if _text is not None:
                    _thinking_block["thinking"] = _text
                if _signature is not None:
                    _thinking_block["signature"] = _signature
                thinking_blocks_list.append(_thinking_block)
            elif "redactedContent" in block:
                _redacted_block = ChatCompletionRedactedThinkingBlock(
                    type="redacted_thinking", data=block["redactedContent"]
                )
                thinking_blocks_list.append(_redacted_block)
        return thinking_blocks_list

    @staticmethod
    def _parse_cache_details(usage: ConverseTokenUsageBlock) -> "CacheCreationTokenDetails | None":
        """Split ``cacheDetails`` into 5m/1h buckets, or ``None`` unless the split fully
        accounts for ``cacheWriteInputTokens``, since a partial or unrecognized-ttl
        breakdown would understate the cache-write cost.

        https://docs.aws.amazon.com/bedrock/latest/APIReference/API_runtime_CacheDetail.html
        """
        cache_details: Final = usage.get("cacheDetails")
        if not cache_details:
            return None
        tokens_5m: Final = sum(d["inputTokens"] for d in cache_details if d.get("ttl") == "5m")
        tokens_1h: Final = sum(d["inputTokens"] for d in cache_details if d.get("ttl") == "1h")
        if tokens_5m + tokens_1h != usage.get("cacheWriteInputTokens", 0):
            return None
        return CacheCreationTokenDetails(
            ephemeral_5m_input_tokens=tokens_5m,
            ephemeral_1h_input_tokens=tokens_1h,
        )

    @staticmethod
    def thinking_tokens_from_additional_fields(additional_fields: object) -> int | None:
        """Converse omits thinking tokens from its usage block; they only arrive under
        ``additionalModelResponseFields`` when ``/usage/output_tokens_details`` is requested."""
        if not isinstance(additional_fields, Mapping):
            return None
        usage: Final = additional_fields.get("usage")
        if not isinstance(usage, Mapping):
            return None
        return AnthropicConfig.thinking_tokens_from_usage(usage)

    @staticmethod
    def is_converse_usage_shape(usage_object: Mapping[str, object]) -> bool:
        """Converse-family models report camelCase token counts, not Anthropic's snake_case."""
        return "inputTokens" in usage_object or "outputTokens" in usage_object

    @staticmethod
    def _usage_count(usage_object: Mapping[str, object], *keys: str) -> int:
        for key in keys:
            value = usage_object.get(key)
            if isinstance(value, (int, float)) and not isinstance(value, bool):
                return int(value)
        return 0

    def usage_from_batch_output(self, usage_object: Mapping[str, object]) -> Usage:
        """Read a Converse-shaped usage block out of a batch output line.

        Batch output omits fields the live API always sends, so the block is
        completed before going through the same transform, keeping a batch and an
        equivalent non-batch call in agreement on tokens.
        """
        input_tokens: Final = self._usage_count(usage_object, "inputTokens")
        output_tokens: Final = self._usage_count(usage_object, "outputTokens")
        cache_read: Final = self._usage_count(usage_object, "cacheReadInputTokens", "cacheReadInputTokenCount")
        cache_write: Final = self._usage_count(usage_object, "cacheWriteInputTokens", "cacheWriteInputTokenCount")
        return self.transform_usage(
            ConverseTokenUsageBlock(
                inputTokens=input_tokens,
                outputTokens=output_tokens,
                totalTokens=self._usage_count(usage_object, "totalTokens") or input_tokens + output_tokens,
                cacheReadInputTokenCount=cache_read,
                cacheReadInputTokens=cache_read,
                cacheWriteInputTokenCount=cache_write,
                cacheWriteInputTokens=cache_write,
            )
        )

    def transform_usage(
        self,
        usage: ConverseTokenUsageBlock,
        reasoning_content: str | None = None,
        thinking_ran: bool = False,
        provider_reasoning_tokens: int | None = None,
    ) -> Usage:
        input_tokens = usage["inputTokens"]
        output_tokens: Final = usage["outputTokens"]
        total_tokens: Final = usage["totalTokens"]
        cache_creation_input_tokens: int = 0
        cache_read_input_tokens: int = 0

        raw_input_tokens: Final = input_tokens  # capture before inflation
        if "cacheReadInputTokens" in usage:
            cache_read_input_tokens = usage["cacheReadInputTokens"]
            input_tokens += cache_read_input_tokens
        if "cacheWriteInputTokens" in usage:
            cache_creation_input_tokens = usage["cacheWriteInputTokens"]
            input_tokens += cache_creation_input_tokens

        prompt_tokens_details: Final = PromptTokensDetailsWrapper(
            cached_tokens=cache_read_input_tokens,
            cache_creation_tokens=cache_creation_input_tokens,
            cache_creation_token_details=self._parse_cache_details(usage),
            text_tokens=raw_input_tokens,
        )
        estimated_reasoning_tokens: Final = (
            token_counter(text=reasoning_content, count_response_tokens=True) if reasoning_content else 0
        )
        reasoning_tokens: Final = (
            min(max(0, provider_reasoning_tokens), output_tokens)
            if provider_reasoning_tokens is not None
            else estimated_reasoning_tokens
        )
        completion_tokens_details: Final = (
            CompletionTokensDetailsWrapper(
                reasoning_tokens=reasoning_tokens,
                text_tokens=output_tokens - reasoning_tokens,
            )
            if reasoning_tokens > 0
            else CompletionTokensDetailsWrapper(
                reasoning_tokens=None if thinking_ran else 0,
                text_tokens=None if thinking_ran else output_tokens,
            )
        )
        openai_usage: Final = Usage(
            prompt_tokens=input_tokens,
            completion_tokens=output_tokens,
            total_tokens=total_tokens,
            prompt_tokens_details=prompt_tokens_details,
            cache_creation_input_tokens=cache_creation_input_tokens,
            cache_read_input_tokens=cache_read_input_tokens,
            completion_tokens_details=completion_tokens_details,
        )
        return openai_usage

    def get_tool_call_names(
        self,
        tools: list[ToolBlock] | list[OpenAIChatCompletionToolParam] | None = None,
    ) -> list[str]:
        if tools is None:
            return []
        tool_set: Final[set[str]] = set()
        for tool in tools:
            tool_spec = tool.get("toolSpec")
            function = tool.get("function")
            if tool_spec is not None:
                _name = cast(dict, tool_spec).get("name")
                if _name is not None and isinstance(_name, str):
                    tool_set.add(_name)
            if function is not None:
                _name = cast(dict, function).get("name")
                if _name is not None and isinstance(_name, str):
                    tool_set.add(_name)
        return list(tool_set)

    def apply_tool_call_transformation_if_needed(
        self,
        message: Message,
        tools: list[ToolBlock] | None = None,
        initial_finish_reason: str | None = None,
    ) -> tuple[Message, str | None]:
        """
        Apply tool call transformation to a message.

        LLM providers (e.g. Bedrock, Vertex AI) sometimes return tool call in the response content.

        If the response content is a JSON object, we can parse it and return the tool call in the tool_calls field.
        """
        returned_finish_reason = initial_finish_reason
        if tools is None:
            return message, returned_finish_reason

        if message.content is not None:
            try:
                tool_call_names: Final = self.get_tool_call_names(tools)
                json_content: Final = json.loads(message.content)
                if json_content.get("type") == "function" and json_content.get("name") in tool_call_names:
                    tool_calls: Final = [ChatCompletionMessageToolCall(function=Function(**json_content))]

                    message.tool_calls = tool_calls
                    message.content = None
                    returned_finish_reason = "tool_calls"
            except Exception:
                pass

        return message, returned_finish_reason

    def _translate_message_content(
        self, content_blocks: list[ContentBlock]
    ) -> tuple[
        str,
        list[ChatCompletionToolCallChunk],
        list[BedrockConverseReasoningContentBlock] | None,
        list[CitationsContentBlock] | None,
    ]:
        """
        Translate the message content to a string and a list of tool calls, reasoning content blocks, and citations.

        Returns:
            content_str: str
            tools: List[ChatCompletionToolCallChunk]
            reasoningContentBlocks: Optional[List[BedrockConverseReasoningContentBlock]]
            citationsContentBlocks: Optional[List[CitationsContentBlock]] - Citations from Nova grounding
        """
        content_str = ""
        tools: Final[list[ChatCompletionToolCallChunk]] = []
        reasoningContentBlocks: list[BedrockConverseReasoningContentBlock] | None = None
        citationsContentBlocks: list[CitationsContentBlock] | None = None
        for idx, content in enumerate(content_blocks):
            """
            - Content is either a tool response or text
            """
            extracted_reasoning_content_str: str | None = None
            if "text" in content:
                (
                    extracted_reasoning_content_str,
                    _content_str,
                ) = _parse_content_for_reasoning(content["text"])
                if _content_str is not None:
                    content_str += _content_str
            if "toolUse" in content:
                ## check tool name was formatted by litellm
                _response_tool_name = content["toolUse"]["name"]
                response_tool_name = get_bedrock_tool_name(response_tool_name=_response_tool_name)
                _function_chunk = ChatCompletionToolCallFunctionChunk(
                    name=response_tool_name,
                    arguments=json.dumps(content["toolUse"]["input"]),
                )

                _tool_response_chunk = ChatCompletionToolCallChunk(
                    id=content["toolUse"]["toolUseId"],
                    type="function",
                    function=_function_chunk,
                    index=idx,
                )
                tools.append(_tool_response_chunk)
            if extracted_reasoning_content_str is not None:
                if reasoningContentBlocks is None:
                    reasoningContentBlocks = []
                reasoningContentBlocks.append(
                    BedrockConverseReasoningContentBlock(
                        reasoningText=BedrockConverseReasoningTextBlock(
                            text=extracted_reasoning_content_str,
                        )
                    )
                )
            if "reasoningContent" in content:
                if reasoningContentBlocks is None:
                    reasoningContentBlocks = []
                reasoningContentBlocks.append(content["reasoningContent"])
            # Handle Nova grounding citations content
            if "citationsContent" in content:
                if citationsContentBlocks is None:
                    citationsContentBlocks = []
                citationsContentBlocks.append(content["citationsContent"])

        return content_str, tools, reasoningContentBlocks, citationsContentBlocks

    @staticmethod
    def _transform_citations_to_annotations(
        citations_content_blocks: list[CitationsContentBlock] | None,
    ) -> tuple[str | None, list[ChatCompletionAnnotation] | None]:
        """
        Convert Bedrock citationsContent blocks into OpenAI-style annotations.

        Returns:
            citations_text: concatenated text from citationsContent.content
            annotations: OpenAI URL citation annotations
        """
        if not citations_content_blocks:
            return None, None

        annotations: Final[list[ChatCompletionAnnotation]] = []
        citations_text_parts: Final[list[str]] = []
        content_offset = 0

        for citations_block in citations_content_blocks:
            block_text = ""
            raw_content = citations_block.get("content")
            if isinstance(raw_content, list):
                for content_part in raw_content:
                    if isinstance(content_part, dict):
                        _text = content_part.get("text")
                        if isinstance(_text, str):
                            block_text += _text

            block_offset = content_offset
            if block_text:
                citations_text_parts.append(block_text)
                content_offset += len(block_text)

            raw_citations = citations_block.get("citations")
            if not isinstance(raw_citations, list):
                continue

            for citation in raw_citations:
                if not isinstance(citation, dict):
                    continue

                location = citation.get("location")
                if not isinstance(location, dict):
                    continue

                search_location = location.get("searchResultLocation")
                if not isinstance(search_location, dict):
                    continue

                start = search_location.get("start")
                end = search_location.get("end")
                if not isinstance(start, int) or not isinstance(end, int):
                    continue

                annotations.append(
                    ChatCompletionAnnotation(
                        type="url_citation",
                        url_citation={
                            "start_index": block_offset + start,
                            "end_index": block_offset + end,
                            "title": str(citation.get("title") or ""),
                            "url": str(citation.get("source") or ""),
                        },
                    )
                )

        citations_text: Final = "".join(citations_text_parts) if citations_text_parts else None
        return citations_text, annotations or None

    @staticmethod
    def _unwrap_bedrock_properties(json_str: str) -> str:
        """
        Unwrap Bedrock's response_format JSON structure.

        If the JSON has a single "properties" key, extract its value.
        Otherwise, return the original string.

        Args:
            json_str: JSON string to unwrap

        Returns:
            Unwrapped JSON string or original if unwrapping not needed
        """
        try:
            response_data = json.loads(json_str)
            if isinstance(response_data, dict) and "properties" in response_data and len(response_data) == 1:
                response_data = response_data["properties"]
                return json.dumps(response_data)
        except json.JSONDecodeError:
            pass
        return json_str

    @staticmethod
    def _filter_json_mode_tools(
        json_mode: bool | None,
        tools: list[ChatCompletionToolCallChunk],
        chat_completion_message: ChatCompletionResponseMessage,
    ) -> list[ChatCompletionToolCallChunk] | None:
        """
        When json_mode is True, Bedrock may return the internal `json_tool_call`
        tool alongside real user-defined tools. This method handles 3 scenarios:

        1. Only json_tool_call present  -> convert to text content, return None
        2. Mixed json_tool_call + real  -> filter out json_tool_call, return real tools
        3. No json_tool_call / no json_mode -> return tools as-is
        """
        if not json_mode or not tools:
            return tools if tools else None

        json_tool_indices = [i for i, t in enumerate(tools) if t["function"].get("name") == RESPONSE_FORMAT_TOOL_NAME]

        if not json_tool_indices:
            # No json_tool_call found, return tools unchanged
            return tools

        if len(json_tool_indices) == len(tools):
            # All tools are json_tool_call — convert first one to content
            verbose_logger.debug("Processing JSON tool call response for response_format")
            json_mode_content_str: str | None = tools[0]["function"].get("arguments")
            if json_mode_content_str is not None:
                json_mode_content_str = AmazonConverseConfig._unwrap_bedrock_properties(json_mode_content_str)
                chat_completion_message["content"] = json_mode_content_str
            return None

        # Mixed: filter out json_tool_call, keep real tools.
        # Preserve the json_tool_call content as message text so the structured
        # output from response_format is not silently lost.
        first_idx: Final = json_tool_indices[0]
        json_mode_args = tools[first_idx]["function"].get("arguments")
        if json_mode_args is not None:
            json_mode_args = AmazonConverseConfig._unwrap_bedrock_properties(json_mode_args)
            existing: Final = chat_completion_message.get("content") or ""
            chat_completion_message["content"] = existing + json_mode_args if existing else json_mode_args

        real_tools: Final = [t for i, t in enumerate(tools) if i not in json_tool_indices]
        return real_tools if real_tools else None

    def _transform_response(
        self,
        model: str,
        response: httpx.Response,
        model_response: ModelResponse,
        stream: bool,
        logging_obj: Logging | None,
        optional_params: dict,
        api_key: str | None,
        data: dict | str,
        messages: list,
        encoding,
    ) -> ModelResponse:
        ## LOGGING
        if logging_obj is not None:
            logging_obj.post_call(
                input=messages,
                api_key=api_key,
                original_response=response.text,
                additional_args={"complete_input_dict": data},
            )

        json_mode: Final[bool | None] = optional_params.get("json_mode", None)
        ## RESPONSE OBJECT
        try:
            completion_response: Final = ConverseResponseBlock(**response.json())
        except Exception as e:
            raise BedrockError(
                message=f"Error converting to valid response block={e}. File an issue if litellm error - https://github.com/BerriAI/litellm/issues",
                status_code=422,
            )

        """
        Bedrock Response Object has optional message block

        completion_response["output"].get("message", None)

        A message block looks like this (Example 1):
        "output": {
            "message": {
                "role": "assistant",
                "content": [
                    {
                        "text": "Is there anything else you'd like to talk about? Perhaps I can help with some economic questions or provide some information about economic concepts?"
                    }
                ]
            }
        },
        (Example 2):
        "output": {
            "message": {
                "role": "assistant",
                "content": [
                    {
                        "toolUse": {
                            "toolUseId": "tooluse_hbTgdi0CSLq_hM4P8csZJA",
                            "name": "top_song",
                            "input": {
                                "sign": "WZPZ"
                            }
                        }
                    }
                ]
            }
        }

        """
        message: Final[MessageBlock | None] = completion_response["output"]["message"]
        chat_completion_message: Final[ChatCompletionResponseMessage] = {"role": "assistant"}
        content_str = ""
        tools: list[ChatCompletionToolCallChunk] = []
        reasoningContentBlocks: list[BedrockConverseReasoningContentBlock] | None = None
        citationsContentBlocks: list[CitationsContentBlock] | None = None

        if message is not None:
            (
                content_str,
                tools,
                reasoningContentBlocks,
                citationsContentBlocks,
            ) = self._translate_message_content(message["content"])

        # Initialize provider_specific_fields if we have any special content blocks
        provider_specific_fields: Final[dict] = {}
        if reasoningContentBlocks is not None:
            provider_specific_fields["reasoningContentBlocks"] = reasoningContentBlocks
        if citationsContentBlocks is not None:
            provider_specific_fields["citationsContent"] = citationsContentBlocks

        if provider_specific_fields:
            chat_completion_message["provider_specific_fields"] = provider_specific_fields

        citations_text, annotations = self._transform_citations_to_annotations(citationsContentBlocks)
        citations_included_in_content = False
        if citations_text:
            stripped_content: Final = content_str.strip()
            if not stripped_content:
                content_str = citations_text
                citations_included_in_content = True
            elif not any(char.isalnum() for char in stripped_content):
                # Bedrock may emit the cited sentence in citationsContent and only
                # punctuation in the text blocks; stitch citations_text in front so
                # its annotation span indices stay aligned with the final content.
                content_str = citations_text + content_str
                citations_included_in_content = True
        if annotations and citations_included_in_content:
            chat_completion_message["annotations"] = annotations

        if reasoningContentBlocks is not None:
            chat_completion_message["reasoning_content"] = self._transform_reasoning_content(reasoningContentBlocks)
            chat_completion_message["thinking_blocks"] = self._transform_thinking_blocks(reasoningContentBlocks)
        chat_completion_message["content"] = content_str
        filtered_tools: Final = self._filter_json_mode_tools(
            json_mode=json_mode,
            tools=tools,
            chat_completion_message=chat_completion_message,
        )
        if filtered_tools:
            chat_completion_message["tool_calls"] = filtered_tools

        ## CALCULATING USAGE - bedrock returns usage in the headers
        usage: Final = self.transform_usage(
            completion_response["usage"],
            reasoning_content=chat_completion_message.get("reasoning_content"),
            thinking_ran=reasoningContentBlocks is not None,
            provider_reasoning_tokens=self.thinking_tokens_from_additional_fields(
                completion_response.get("additionalModelResponseFields")
            ),
        )

        ## HANDLE TOOL CALLS
        _message: Final = Message(**chat_completion_message)
        initial_finish_reason = map_finish_reason(completion_response["stopReason"])

        # When json_mode filtered out all synthetic tool calls the response
        # is plain content, not a pending tool invocation. Fix finish_reason
        # so callers (e.g. OpenAI SDK) don't misinterpret it.
        if json_mode and not filtered_tools and tools:
            initial_finish_reason = "stop"

        (
            returned_message,
            returned_finish_reason,
        ) = self.apply_tool_call_transformation_if_needed(
            message=_message,
            tools=optional_params.get("tools"),
            initial_finish_reason=initial_finish_reason,
        )
        model_response.choices = [
            litellm.Choices(
                finish_reason=returned_finish_reason,
                index=0,
                message=returned_message,
            )
        ]
        model_response.created = int(time.time())
        model_response.model = model

        setattr(model_response, "usage", usage)

        # Add "trace" from Bedrock guardrails - if user has opted in to returning it
        if "trace" in completion_response:
            setattr(model_response, "trace", completion_response["trace"])

        # Add service_tier if present in Bedrock response
        # Map Bedrock serviceTier (object) to OpenAI service_tier (string)
        if "serviceTier" in completion_response:
            service_tier_block: Final = completion_response["serviceTier"]
            if isinstance(service_tier_block, dict) and "type" in service_tier_block:
                setattr(model_response, "service_tier", service_tier_block["type"])

        return model_response

    def get_error_class(self, error_message: str, status_code: int, headers: dict | httpx.Headers) -> BaseLLMException:
        return BedrockError(
            message=error_message,
            status_code=status_code,
            headers=headers,
        )

    def validate_environment(
        self,
        headers: dict,
        model: str,
        messages: list[AllMessageValues],
        optional_params: dict,
        litellm_params: dict,
        api_key: str | None = None,
        api_base: str | None = None,
    ) -> dict:
        if api_key:
            headers["Authorization"] = f"Bearer {api_key}"
        owned_names, metadata_headers = bedrock_request_metadata_headers(litellm_params)
        return merge_bedrock_invoke_headers(headers, (), metadata_headers, owned_names)

    def should_fake_stream(
        self,
        model: str | None,
        stream: bool | None,
        custom_llm_provider: str | None = None,
        fake_stream: bool | None = None,
    ) -> bool:
        """
        Returns True if the model/provider should fake stream
        """
        ###################################################################
        # If an upstream method already set fake_stream to True, return True
        ###################################################################
        if fake_stream is True:
            return True

        ###################################################################
        # Bedrock Converse Specific Logic
        ###################################################################
        if stream is True:
            if model is not None:
                ###################################################################
                # AI21 models do not support streaming
                ###################################################################
                if "ai21" in model:
                    return True
        return False
