"""
Anthropic Message Handler for Unified Guardrails

This module provides a class-based handler for Anthropic-format messages.
The class methods can be overridden for custom behavior.

Pattern Overview:
-----------------
1. Extract text content from messages/responses (both string and list formats)
2. Create async tasks to apply guardrails to each text segment
3. Track mappings to know where each response belongs
4. Apply guardrail responses back to the original structure
"""

import json
from collections.abc import Mapping, Sequence
from copy import deepcopy
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any, Final, Protocol, cast, overload, runtime_checkable

from typing_extensions import ReadOnly, TypedDict, assert_never

from litellm._logging import verbose_proxy_logger
from litellm.llms.anthropic.chat.transformation import AnthropicConfig
from litellm.llms.anthropic.experimental_pass_through.adapters.transformation import (
    LiteLLMAnthropicMessagesAdapter,
    is_provider_native_tool_dict,
)
from litellm.llms.base_llm.guardrail_translation.base_translation import BaseTranslation
from litellm.llms.base_llm.guardrail_translation.utils import (
    anthropic_tool_name,
    anthropic_tool_names,
    effective_scan_only_tool_results_for_guardrail,
    effective_skip_system_message_for_guardrail,
    effective_skip_tool_message_for_guardrail,
    merge_guardrailed_scoped_messages,
    merge_returned_tools_into_request_tools,
    scoped_structured_message_indices,
)
from litellm.proxy.pass_through_endpoints.llm_provider_handlers.anthropic_passthrough_logging_handler import (
    AnthropicPassthroughLoggingHandler,
)
from litellm.types.llms.anthropic import (
    AllAnthropicToolsValues,
    AnthropicMessagesRequest,
)
from litellm.types.llms.openai import (
    AllMessageValues,
    ChatCompletionRequest,
    ChatCompletionToolCallChunk,
    ChatCompletionToolParam,
)
from litellm.types.utils import (
    ChatCompletionMessageToolCall,
    Choices,
    GenericGuardrailAPIInputs,
    ModelResponse,
)

if TYPE_CHECKING:
    from fastapi import HTTPException

    from litellm.integrations.custom_guardrail import (
        CustomGuardrail,
        ModifyResponseException,
    )
    from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
    from litellm.proxy._types import UserAPIKeyAuth
    from litellm.types.llms.anthropic_messages.anthropic_response import (
        AnthropicMessagesResponse,
    )


@dataclass(frozen=True, slots=True)
class MessageContentTarget:
    msg_idx: int


@dataclass(frozen=True, slots=True)
class ContentBlockTextTarget:
    msg_idx: int
    content_idx: int


@dataclass(frozen=True, slots=True)
class ToolResultStringTarget:
    msg_idx: int
    content_idx: int


@dataclass(frozen=True, slots=True)
class ToolResultBlockTextTarget:
    msg_idx: int
    content_idx: int
    block_idx: int


InputWriteBackTarget = (
    MessageContentTarget | ContentBlockTextTarget | ToolResultStringTarget | ToolResultBlockTextTarget
)


def _as_str_mapping(value: Mapping[str, object]) -> Mapping[str, object]:
    return value


def _content_block_at(blocks: Sequence[object], index: int) -> object:
    return blocks[index]


@runtime_checkable
class _ModelDumpBlock(Protocol):
    def model_dump(self) -> Mapping[str, object]: ...


@runtime_checkable
class _TextAttrBlock(Protocol):
    text: str


class _WritableMessage(Protocol):
    @overload
    def get(self, key: str, /) -> object | None: ...

    @overload
    def get(self, key: str, default: object, /) -> object: ...

    def __setitem__(self, key: str, value: object, /) -> None: ...


def _as_writable(value: _WritableMessage) -> _WritableMessage:
    return value


@dataclass(frozen=True, slots=True)
class ScannedText:
    text: str
    target: InputWriteBackTarget


@dataclass(frozen=True, slots=True)
class ExtractedInput:
    scanned: tuple[ScannedText, ...]
    images: tuple[str, ...]


EMPTY_EXTRACTED_INPUT: Final = ExtractedInput(scanned=(), images=())


class _AnthropicSSEDelta(TypedDict, total=False):
    type: ReadOnly[str]
    text: ReadOnly[str]
    stop_reason: ReadOnly[str | None]


class _AnthropicSSEEvent(TypedDict, total=False):
    delta: ReadOnly[_AnthropicSSEDelta]


class AnthropicMessagesHandler(BaseTranslation):
    """Process Anthropic messages with guardrails.

    In-sequence system entries are untrusted client input. This handler scans and preserves
    them through guardrail rewrites; downstream provider handling is out of scope.
    """

    def __init__(self):
        super().__init__()
        self.adapter = LiteLLMAnthropicMessagesAdapter()

    @staticmethod
    def _build_streaming_usage_response(
        responses_so_far: Sequence[object],
        request_data: dict | None,
    ) -> ModelResponse | None:
        chunks: Final = tuple(response for response in responses_so_far if isinstance(response, (str, bytes)))
        if not chunks:
            return None
        try:
            return AnthropicPassthroughLoggingHandler._build_usage_only_response_from_chunks(
                all_chunks=chunks,
                model=str((request_data or {}).get("model") or ""),
            )
        except (AttributeError, TypeError, ValueError):
            return None

    def build_block_sse_chunks(
        self,
        exc: "ModifyResponseException",
        stream_started: bool = False,
        responses_so_far: Sequence[object] | None = None,
    ) -> list[bytes]:
        """
        Build an Anthropic SSE sequence delivering the guardrail block message
        and terminating the stream cleanly.

        - ``stream_started`` False (buffered / pre-stream): nothing has been
          sent, so emit a complete standalone message (message_start ->
          content_block_* -> message_delta -> message_stop) via
          FakeAnthropicMessagesStreamIterator, the same converter the
          /v1/messages pre-stream block handler uses.
        - ``stream_started`` True (sampling / detect-only end-of-stream): real
          chunks were already sent, so *continue* the in-progress message --
          close the open content block, append the block message as a new text
          block, then end the message. Emitting a second ``message_start`` here
          would make Anthropic clients reject the stream.
        """
        if stream_started:
            return list(self._block_continuation_chunks(exc, responses_so_far or []))
        return self._standalone_block_chunks(exc)

    def build_stream_error_items(
        self,
        exc: "HTTPException",
        responses_so_far: Sequence[Any] | None = None,
    ) -> Sequence[Any] | None:
        from litellm.proxy.common_request_processing import (
            serialize_http_exception_detail,
        )
        from litellm.proxy.guardrails.anthropic_sse import anthropic_sse_error_frames

        message, _ = serialize_http_exception_detail(exc.detail)
        return tuple(anthropic_sse_error_frames(message))

    def _standalone_block_chunks(self, exc: "ModifyResponseException") -> list[bytes]:
        import uuid

        from litellm.llms.anthropic.experimental_pass_through.messages.fake_stream_iterator import (
            FakeAnthropicMessagesStreamIterator,
        )
        from litellm.llms.base_llm.guardrail_translation.utils import (
            blocked_response_usage,
        )
        from litellm.types.utils import AnthropicMessagesResponse

        block_response: Final = AnthropicMessagesResponse(
            id=f"msg_{uuid.uuid4()}",
            type="message",
            role="assistant",
            content=[{"type": "text", "text": exc.message}],
            model=exc.model,
            stop_reason="end_turn",
            usage=blocked_response_usage(getattr(exc, "original_response", None)),
        )
        return list(FakeAnthropicMessagesStreamIterator(response=block_response))

    def _block_continuation_chunks(
        self, exc: "ModifyResponseException", responses_so_far: Sequence[object]
    ) -> Sequence[bytes]:
        """Continue an already-started message: close the open content block,
        append the block message as a new text block, then end the message --
        without a second message_start."""

        from litellm.llms.base_llm.guardrail_translation.utils import (
            blocked_response_usage,
        )

        def _sse(event_type: str, payload: dict) -> bytes:
            return f"event: {event_type}\ndata: {json.dumps(payload)}\n\n".encode()

        output_tokens: Final = blocked_response_usage(getattr(exc, "original_response", None)).get("output_tokens", 0)
        open_index, max_index = self._content_block_state(responses_so_far)
        new_index: Final = (max_index + 1) if max_index is not None else 0
        chunks: list[bytes] = []
        if open_index is not None:
            chunks.append(_sse("content_block_stop", {"type": "content_block_stop", "index": open_index}))
        chunks += [
            _sse(
                "content_block_start",
                {
                    "type": "content_block_start",
                    "index": new_index,
                    "content_block": {"type": "text", "text": ""},
                },
            ),
            _sse(
                "content_block_delta",
                {
                    "type": "content_block_delta",
                    "index": new_index,
                    "delta": {"type": "text_delta", "text": exc.message},
                },
            ),
            _sse("content_block_stop", {"type": "content_block_stop", "index": new_index}),
            _sse(
                "message_delta",
                {
                    "type": "message_delta",
                    "delta": {"stop_reason": "end_turn", "stop_sequence": None},
                    "usage": {"output_tokens": output_tokens},
                },
            ),
            _sse("message_stop", {"type": "message_stop"}),
        ]
        return chunks

    @staticmethod
    def _content_block_state(
        responses_so_far: Sequence[object],
    ) -> tuple[int | None, int | None]:
        """From the SSE chunks already sent to the client, return (open
        content-block index or None, highest content-block index seen or None).

        A single streamed item may bundle multiple SSE events (raw bytes) or be
        an already-parsed event dict, so every event across every item is
        considered -- matching how ``get_streaming_string_so_far`` reads the
        same stream."""
        open_indices: Final[set[int]] = set()
        max_index: int | None = None
        for item in responses_so_far:
            for data in AnthropicMessagesHandler._iter_sse_events(item):
                event_type = data.get("type")
                index = data.get("index")
                if not isinstance(index, int):
                    continue
                if event_type == "content_block_start":
                    open_indices.add(index)
                    max_index = index if max_index is None else max(max_index, index)
                elif event_type == "content_block_stop":
                    open_indices.discard(index)
        open_index: Final = max(open_indices) if open_indices else None
        return open_index, max_index

    @staticmethod
    def _parse_sse_data_line(raw_line: str) -> tuple[Mapping[str, object], ...]:
        line: Final = raw_line.strip()
        if not line.startswith("data:"):
            return ()
        try:
            parsed: Final[object] = json.loads(line[len("data:") :].strip())
        except json.JSONDecodeError:
            return ()
        if not isinstance(parsed, dict):
            return ()
        return (_as_str_mapping(parsed),)

    @staticmethod
    def _iter_sse_events(item: object) -> Sequence[Mapping[str, object]]:
        """Yield the event-data dicts in one stream chunk.

        Handles both formats this stream can carry (see
        ``get_streaming_string_so_far``): raw SSE ``bytes`` -- which may bundle
        several events separated by a blank line -- and an already-parsed event
        ``dict``."""
        if isinstance(item, dict):
            return (_as_str_mapping(item),)
        if not isinstance(item, (bytes, bytearray)):
            return ()
        return tuple(
            event
            for block in item.decode("utf-8", errors="replace").split("\n\n")
            for line in block.split("\n")
            for event in AnthropicMessagesHandler._parse_sse_data_line(line)
        )

    def _translate_to_openai(self, data: dict) -> ChatCompletionRequest:
        """Translate Anthropic request to OpenAI chat completion format."""
        (
            chat_completion_compatible_request,
            _tool_name_mapping,
        ) = LiteLLMAnthropicMessagesAdapter().translate_anthropic_to_openai(
            anthropic_message_request=cast(AnthropicMessagesRequest, data.copy())
        )
        return chat_completion_compatible_request

    def get_structured_messages(self, data: dict) -> list[AllMessageValues] | None:
        """
        Convert Anthropic messages request data to OpenAI-spec structured messages.

        Uses the Anthropic-to-OpenAI adapter to translate message format.
        """
        messages: Final = data.get("messages")
        if messages is None:
            return None
        chat_completion_compatible_request: Final = self._translate_to_openai(data)
        result: Final = cast(
            list[AllMessageValues],
            chat_completion_compatible_request.get("messages", []),
        )
        return result if result else None

    async def process_input_messages(
        self,
        data: dict,
        guardrail_to_apply: "CustomGuardrail",
        litellm_logging_obj: "LiteLLMLoggingObj | None" = None,
    ) -> Mapping[str, object]:
        """
        Process input messages by applying guardrails to text content.
        """
        messages: Final = data.get("messages")
        if messages is None:
            return data

        skip_system: Final = effective_skip_system_message_for_guardrail(guardrail_to_apply)
        skip_tool: Final = effective_skip_tool_message_for_guardrail(guardrail_to_apply)
        scan_only_tool_results: Final = effective_scan_only_tool_results_for_guardrail(guardrail_to_apply)

        # Exclude only the trusted top-level prompt. In-sequence system entries are untrusted
        # and must stay aligned with texts_to_check for positional masking. When the top-level
        # prompt is included, the pre-existing count mismatch disables positional masking.
        translation_source: Final = {  # mutable-ok: API message payload
            key: value for key, value in data.items() if key != "system"
        }
        chat_completion_compatible_request: Final = self._translate_to_openai(translation_source)

        full_structured_messages: Final = cast(
            list[AllMessageValues],
            chat_completion_compatible_request.get("messages", []),
        )
        has_midturn_system_message: Final = any(
            str(message.get("role") or "").lower() == "system" for message in full_structured_messages
        )
        hoisted_system_message: Final = None if skip_system else self._hoisted_top_level_system_message(data)
        if hoisted_system_message is not None:
            full_structured_messages.insert(0, hoisted_system_message)
        # skip_system already excluded the trusted top-level prompt (it is simply not hoisted);
        # in-sequence system entries are untrusted and always stay in scope.
        scoped_message_indices: Final = scoped_structured_message_indices(
            full_structured_messages,
            scan_only_tool_results=scan_only_tool_results,
            skip_system=False,
            skip_tool=skip_tool,
        )
        structured_messages: Final = [full_structured_messages[index] for index in scoped_message_indices]

        tools_to_check: Final[list[ChatCompletionToolParam]] = (
            []
            if scan_only_tool_results
            else [
                tool
                for tool in chat_completion_compatible_request.get("tools", [])
                if not is_provider_native_tool_dict(tool)
            ]
        )

        # Step 1: Extract all text content and images
        extracted: Final = tuple(
            self._extract_input_text_and_images(
                message=message,
                msg_idx=msg_idx,
                skip_system_message=skip_system,
                skip_tool_message=skip_tool,
                scan_only_tool_results=scan_only_tool_results,
            )
            for msg_idx, message in enumerate(messages)
        )
        scanned: Final = tuple(item for one_message in extracted for item in one_message.scanned)
        texts_to_check: Final = [item.text for item in scanned]  # mutable-ok: GenericGuardrailAPIInputs takes list[str]
        images_to_check: Final = [
            image for one_message in extracted for image in one_message.images
        ]  # mutable-ok: GenericGuardrailAPIInputs takes list[str]

        # Step 2: Apply guardrail to all texts in batch
        if texts_to_check:
            inputs: Final = GenericGuardrailAPIInputs(texts=texts_to_check)
            if images_to_check:
                inputs["images"] = images_to_check
            if tools_to_check:
                inputs["tools"] = tools_to_check
            original_structured_messages: Final = structured_messages
            if structured_messages:
                inputs["structured_messages"] = structured_messages
            # Include model information if available
            model: Final = data.get("model")
            if model:
                inputs["model"] = model
            guardrailed_inputs: Final = await guardrail_to_apply.apply_guardrail(
                inputs=inputs,
                request_data=data,
                input_type="request",
                logging_obj=litellm_logging_obj,
            )

            guardrailed_texts: Final = guardrailed_inputs.get("texts", [])
            guardrailed_tools: Final = guardrailed_inputs.get("tools")
            if guardrailed_tools is not None:
                # Convert tools back from OpenAI format to Anthropic format
                anthropic_config: Final = AnthropicConfig()
                anthropic_tools: Final[list[AllAnthropicToolsValues]] = []
                for tool in guardrailed_tools:
                    converted_tool, mcp_server = anthropic_config._map_tool_helper(tool)
                    if converted_tool is not None:
                        anthropic_tools.append(converted_tool)
                    # Note: MCP servers are handled separately in the main transformation
                data["tools"] = (
                    merge_returned_tools_into_request_tools(
                        request_tools=data.get("tools"),
                        returned_tools=anthropic_tools,
                        tool_name=anthropic_tool_name,
                    )
                    if scan_only_tool_results
                    else [
                        *(tool for tool in data.get("tools") or [] if is_provider_native_tool_dict(tool)),
                        *anthropic_tools,
                    ]
                )

            guardrailed_structured_messages: Final = guardrailed_inputs.get("structured_messages")
            if (
                guardrailed_structured_messages is not None
                and guardrailed_structured_messages is not original_structured_messages
            ):
                self._write_back_structured_messages(
                    data,
                    guardrailed_structured_messages
                    if guardrail_to_apply.structured_messages_cover_full_request()
                    else merge_guardrailed_scoped_messages(
                        full_messages=full_structured_messages,
                        scoped_indices=scoped_message_indices,
                        guardrailed_scoped=guardrailed_structured_messages,
                    ),
                    hoisted_system_message=hoisted_system_message,
                    preserve_system_messages=has_midturn_system_message,
                )
            else:
                # Step 3: Map guardrail responses back to original message structure
                await self._apply_guardrail_responses_to_input(
                    messages=messages,
                    responses=guardrailed_texts,
                    scanned=scanned,
                )

        verbose_proxy_logger.debug("Anthropic Messages: Processed input messages: %s", messages)

        return data

    def _hoisted_top_level_system_message(
        self, data: dict
    ) -> AllMessageValues | None:  # mutable-ok: API message payload
        """Return the system message produced by translating the top-level prompt."""
        system: Final = data.get("system")
        if not system:
            return None
        probe: Final = self._translate_to_openai(
            {  # mutable-ok: API message payload
                "model": data.get("model") or "",
                "messages": [],  # mutable-ok: API message payload
                "system": system,
            }
        )
        hoisted: Final = probe.get("messages") or []  # mutable-ok: API message payload
        return hoisted[0] if hoisted else None

    @staticmethod
    def _openai_system_message_to_anthropic(
        message: Mapping[str, object],
    ) -> dict[str, object] | None:  # mutable-ok: API message payload
        """Convert an OpenAI system message to the client's Anthropic-shaped entry."""
        content: Final = message.get("content")
        if isinstance(content, str):
            return (
                {"role": "system", "content": content} if content else None  # mutable-ok: API message payload
            )  # mutable-ok: API message payload
        if not isinstance(content, list):
            return None
        blocks: Final[list[dict[str, object]]] = []  # mutable-ok: API message payload
        for block in content:
            if not isinstance(block, dict) or block.get("type") != "text":
                continue
            text = block.get("text")
            if not isinstance(text, str) or not text:
                continue
            anthropic_block: dict[str, object] = {  # mutable-ok: API message payload
                "type": "text",
                "text": text,
            }  # mutable-ok: API message payload
            cache_control = block.get("cache_control")
            if cache_control:
                anthropic_block["cache_control"] = deepcopy(cache_control)
            blocks.append(anthropic_block)
        return (
            {"role": "system", "content": blocks} if blocks else None  # mutable-ok: API message payload
        )  # mutable-ok: API message payload

    @staticmethod
    def _fold_leading_systems_into_top_level(
        data: dict[str, object],  # mutable-ok: API message payload
        leading_systems: Sequence[object],
        include_existing_system: bool,
    ) -> None:
        """Deliver leading system rows through Anthropic's top-level system param, which rejects them in messages."""
        existing: Final = data.get("system") if include_existing_system else None
        existing_blocks: Final[list[object]] = (  # mutable-ok: API message payload
            [{"type": "text", "text": existing}]
            if isinstance(existing, str) and existing
            else list(existing)
            if isinstance(existing, list)
            else []
        )
        converted_rows: Final = tuple(
            AnthropicMessagesHandler._openai_system_message_to_anthropic(message)
            for message in leading_systems
            if isinstance(message, dict)
        )
        folded: Final[list[object]] = existing_blocks + [  # mutable-ok: API message payload
            block
            for row in converted_rows
            if row is not None
            for block in (
                [{"type": "text", "text": row["content"]}] if isinstance(row["content"], str) else row["content"]
            )
        ]
        if folded:
            data["system"] = folded  # rebind-ok: write-back mutates the request payload in place
        else:
            data.pop("system", None)

    @staticmethod
    def _is_hoisted_top_level_system(message: object, hoisted_system_message: object) -> bool:
        """Match the hoisted prompt by identity, or by value after serialization."""
        if hoisted_system_message is None:
            return False
        if message is hoisted_system_message:
            return True
        return (
            isinstance(message, dict) and isinstance(hoisted_system_message, dict) and message == hoisted_system_message
        )

    @staticmethod
    def _is_system(message: object) -> bool:
        """Whether the row is an in-sequence system message."""
        return isinstance(message, dict) and str(message.get("role") or "").lower() == "system"

    @staticmethod
    def _defer_systems_inside_tool_exchanges(
        structured_messages: Sequence[Mapping[str, object]],
    ) -> list:
        """Hold a system row until the tool exchange around it completes so the call/result pair converts together."""
        from litellm.litellm_core_utils.prompt_templates.factory import group_tool_exchanges

        non_system_positions: Final[list[int]] = [
            index
            for index, message in enumerate(structured_messages)
            if not AnthropicMessagesHandler._is_system(message)
        ]
        exchange_end_for_start: Final[dict[int, int]] = {
            non_system_positions[group[0]]: non_system_positions[group[-1]]
            for group in group_tool_exchanges([structured_messages[index] for index in non_system_positions])
            if len(group) > 1
        }
        ordered: Final[list] = []  # mutable-ok: API message payload
        deferred_systems: Final[list] = []  # mutable-ok: API message payload
        open_exchange_end = -1  # rebind-ok: advances to the enclosing exchange's last index
        for index, message in enumerate(structured_messages):
            if AnthropicMessagesHandler._is_system(message) and index < open_exchange_end:
                deferred_systems.append(message)
                continue
            open_exchange_end = exchange_end_for_start.get(index, open_exchange_end)
            ordered.append(message)
            if index >= open_exchange_end and deferred_systems:
                ordered.extend(deferred_systems)
                deferred_systems.clear()
        ordered.extend(deferred_systems)
        return ordered

    @staticmethod
    def _write_back_structured_messages(
        data: dict,  # mutable-ok: API message payload
        structured_messages: list,  # mutable-ok: API message payload
        hoisted_system_message: object = None,
        preserve_system_messages: bool = False,
    ) -> None:
        """Write a guardrail's structured-message rewrite back without losing corrections."""
        from litellm.litellm_core_utils.prompt_templates.factory import (
            anthropic_messages_pt,
            group_tool_exchanges,
        )

        _is_system: Final = AnthropicMessagesHandler._is_system
        model: Final = str(data.get("model") or "")
        converted: Final[list] = []  # mutable-ok: API message payload

        def _convert_run(run: list) -> None:  # mutable-ok: API message payload
            for group in group_tool_exchanges(run):
                converted.extend(
                    anthropic_messages_pt(
                        messages=[run[index] for index in group],  # mutable-ok: API message payload
                        model=model,
                        llm_provider="anthropic",
                    )
                )

        ordered: Final = AnthropicMessagesHandler._defer_systems_inside_tool_exchanges(structured_messages)
        leading_count: Final = next(
            (index for index, message in enumerate(ordered) if not _is_system(message)),
            len(ordered),
        )
        leading_systems: Final = ordered[:leading_count]
        hoisted_in_leading: Final = any(
            AnthropicMessagesHandler._is_hoisted_top_level_system(message, hoisted_system_message)
            for message in leading_systems
        )
        if leading_systems and not (leading_count == 1 and hoisted_in_leading):
            AnthropicMessagesHandler._fold_leading_systems_into_top_level(
                data,
                leading_systems,
                include_existing_system=hoisted_system_message is None,
            )
        run: Final[list] = []  # mutable-ok: API message payload
        hoisted_dropped = hoisted_in_leading  # rebind-ok: flips once the hoisted prompt is dropped
        for message in ordered[leading_count:]:
            if not _is_system(message):
                run.append(message)
                continue
            _convert_run(run)
            run.clear()
            if not hoisted_dropped and AnthropicMessagesHandler._is_hoisted_top_level_system(
                message, hoisted_system_message
            ):
                hoisted_dropped = True
                continue
            if preserve_system_messages:
                anthropic_system = AnthropicMessagesHandler._openai_system_message_to_anthropic(message)
                if anthropic_system is not None:
                    converted.append(anthropic_system)
        _convert_run(run)
        if not any(not _is_system(message) for message in converted):
            converted.extend(anthropic_messages_pt(messages=[], model=model, llm_provider="anthropic"))
        for msg in converted:
            content = msg.get("content")
            if isinstance(content, list):
                for block in content:
                    if isinstance(block, dict) and block.get("type") == "thinking":
                        block.pop("cache_control", None)
        data["messages"] = converted

    @staticmethod
    def _extract_midturn_system_text(
        message: Mapping[str, object],
        msg_idx: int,
    ) -> ExtractedInput:
        """Match the adapter's filtering so positional guardrail write-back stays aligned."""
        content: Final = message.get("content")
        if isinstance(content, str):
            if not content:
                return EMPTY_EXTRACTED_INPUT
            return ExtractedInput(scanned=(ScannedText(content, MessageContentTarget(msg_idx)),), images=())
        if not isinstance(content, list):
            return EMPTY_EXTRACTED_INPUT
        return ExtractedInput(
            scanned=tuple(
                ScannedText(text_str, ContentBlockTextTarget(msg_idx, content_idx))
                for content_idx, content_item in enumerate(content)
                if isinstance(content_item, dict)
                and content_item.get("type") == "text"
                and isinstance(text_str := content_item.get("text"), str)
                and text_str
            ),
            images=(),
        )

    def extract_request_tool_names(self, data: dict) -> list[str]:
        """Extract every tool name in an Anthropic messages request: tools[].name, plus
        tools[].function.name for OpenAI-format tools the bridge forwards verbatim."""
        return [name for tool in data.get("tools") or [] for name in anthropic_tool_names(tool)]

    @classmethod
    def _extract_input_text_and_images(
        cls,
        message: Mapping[str, object],
        msg_idx: int,
        skip_system_message: bool = False,
        skip_tool_message: bool = False,
        scan_only_tool_results: bool = False,
    ) -> ExtractedInput:
        """Extract text content and images from a message.

        In-sequence system entries are scanned even when ``skip_system_message`` is set:
        that flag covers only the trusted top-level prompt, which never appears here.
        """
        role: Final = str(message.get("role") or "")
        if role == "system":
            if scan_only_tool_results:
                return EMPTY_EXTRACTED_INPUT
            return cls._extract_midturn_system_text(message=message, msg_idx=msg_idx)
        if skip_tool_message and role.lower() == "tool":
            return EMPTY_EXTRACTED_INPUT

        content: Final = message.get("content", None)
        if isinstance(content, str):
            if scan_only_tool_results:
                return EMPTY_EXTRACTED_INPUT
            return ExtractedInput(scanned=(ScannedText(content, MessageContentTarget(msg_idx)),), images=())
        if not isinstance(content, list):
            return EMPTY_EXTRACTED_INPUT

        blocks: Final = tuple(
            cls._extract_content_block(
                content_item=content_item,
                msg_idx=msg_idx,
                content_idx=content_idx,
                skip_tool_message=skip_tool_message,
                scan_only_tool_results=scan_only_tool_results,
            )
            for content_idx, content_item in enumerate(content)
            if isinstance(content_item, dict)
        )
        return ExtractedInput(
            scanned=tuple(item for block in blocks for item in block.scanned),
            images=tuple(image for block in blocks for image in block.images),
        )

    @classmethod
    def _extract_content_block(
        cls,
        content_item: Mapping[str, Any],
        msg_idx: int,
        content_idx: int,
        skip_tool_message: bool,
        scan_only_tool_results: bool = False,
    ) -> ExtractedInput:
        if content_item.get("type") == "tool_result":
            if skip_tool_message:
                return EMPTY_EXTRACTED_INPUT
            return cls._extract_tool_result(content_item=content_item, msg_idx=msg_idx, content_idx=content_idx)

        if scan_only_tool_results:
            return EMPTY_EXTRACTED_INPUT

        text_str: Final[str | None] = content_item.get("text")
        return ExtractedInput(
            scanned=(
                () if text_str is None else (ScannedText(text_str, ContentBlockTextTarget(msg_idx, content_idx)),)
            ),
            images=cls._image_sources(content_item) if content_item.get("type") == "image" else (),
        )

    @classmethod
    def _extract_tool_result(
        cls,
        content_item: Mapping[str, object],
        msg_idx: int,
        content_idx: int,
    ) -> ExtractedInput:
        tool_result_content: Final = content_item.get("content")

        if isinstance(tool_result_content, str):
            return ExtractedInput(
                scanned=(ScannedText(tool_result_content, ToolResultStringTarget(msg_idx, content_idx)),),
                images=(),
            )
        if not isinstance(tool_result_content, list):
            return EMPTY_EXTRACTED_INPUT

        blocks: Final = tuple(
            (block_idx, block) for block_idx, block in enumerate(tool_result_content) if isinstance(block, dict)
        )
        return ExtractedInput(
            scanned=tuple(
                ScannedText(block["text"], ToolResultBlockTextTarget(msg_idx, content_idx, block_idx))
                for block_idx, block in blocks
                if isinstance(block.get("text"), str)
            ),
            images=tuple(
                image for _, block in blocks if block.get("type") == "image" for image in cls._image_sources(block)
            ),
        )

    @staticmethod
    def _image_sources(block: Mapping[str, object]) -> tuple[str, ...]:
        """Normalize an Anthropic image block into strings a guardrail can read.

        base64 becomes a data URI so the format travels with the payload, which is what
        the OpenAI path already puts in this field. A file source yields nothing: those
        bytes live behind the Files API and this extractor has no client to fetch them.
        """
        source: Final = block.get("source")
        if not isinstance(source, Mapping):
            return ()

        source_type: Final = source.get("type")
        if source_type == "url":
            url: Final = source.get("url")
            return (url,) if isinstance(url, str) and url else ()

        data: Final = source.get("data")
        if not isinstance(data, str) or not data:
            return ()
        media_type: Final = source.get("media_type")
        if isinstance(media_type, str) and media_type:
            return (f"data:{media_type};base64,{data}",)
        return (data,)

    async def _apply_guardrail_responses_to_input(
        self,
        messages: Sequence[_WritableMessage],
        responses: list[str],
        scanned: tuple[ScannedText, ...],
    ) -> None:
        """
        Apply guardrail responses back to input messages.
        """
        for item, guardrail_response in zip(scanned, responses):
            target = item.target
            message = messages[target.msg_idx]
            content = message.get("content", None)
            if content is None:
                continue

            match target:
                case MessageContentTarget():
                    if isinstance(content, str):
                        message["content"] = (
                            guardrail_response  # mutable-ok: guardrails rewrite the caller's request payload in place
                        )
                case ContentBlockTextTarget(content_idx=content_idx):
                    if isinstance(content, list):
                        content[content_idx]["text"] = (
                            guardrail_response  # mutable-ok: guardrails rewrite the caller's request payload in place
                        )
                case ToolResultStringTarget(content_idx=content_idx):
                    if isinstance(content, list):
                        content[content_idx]["content"] = (
                            guardrail_response  # mutable-ok: guardrails rewrite the caller's request payload in place
                        )
                case ToolResultBlockTextTarget(content_idx=content_idx, block_idx=block_idx):
                    if isinstance(content, list):
                        content[content_idx]["content"][block_idx]["text"] = (
                            guardrail_response  # mutable-ok: guardrails rewrite the caller's request payload in place
                        )
                case _:
                    assert_never(target)

    async def process_output_response(
        self,
        response: "AnthropicMessagesResponse",
        guardrail_to_apply: "CustomGuardrail",
        litellm_logging_obj: "LiteLLMLoggingObj | None" = None,
        user_api_key_dict: "UserAPIKeyAuth | None" = None,
        request_data: dict | None = None,
    ) -> "AnthropicMessagesResponse":
        """
        Process output response by applying guardrails to text content and tool calls.

        Args:
            response: Anthropic MessagesResponse object
            guardrail_to_apply: The guardrail instance to apply
            litellm_logging_obj: Optional logging object
            user_api_key_dict: User API key metadata to pass to guardrails

        Returns:
            Modified response with guardrail applied to content

        Response Format Support:
            - List content: response.content = [
                {"type": "text", "text": "text here"},
                {"type": "tool_use", "id": "...", "name": "...", "input": {...}},
                ...
            ]
        """
        texts_to_check: Final[list[str]] = []
        images_to_check: Final[list[str]] = []
        tool_calls_to_check: Final[list[ChatCompletionToolCallChunk]] = []
        task_mappings: Final[list[tuple[int, int | None]]] = []

        response_content: Final = self._get_response_content(response)
        if not response_content:
            return response

        # Step 1: Extract all text content and tool calls from response
        self._extract_from_content_blocks(
            response_content,
            texts_to_check,
            images_to_check,
            task_mappings,
            tool_calls_to_check,
        )

        # Step 2: Apply guardrail to all texts in batch
        if texts_to_check or tool_calls_to_check:
            request_data = self._prepare_request_data(
                request_data,
                response,
                user_api_key_dict,
                key="response",
            )

            inputs: Final = self._build_guardrail_inputs(
                texts_to_check,
                images_to_check,
                tool_calls_to_check,
                response,
            )

            guardrailed_inputs: Final = await guardrail_to_apply.apply_guardrail(
                inputs=inputs,
                request_data=request_data,
                input_type="response",
                logging_obj=litellm_logging_obj,
            )

            guardrailed_texts: Final = guardrailed_inputs.get("texts", [])

            # Step 3: Map guardrail responses back to original response structure
            await self._apply_guardrail_responses_to_output(
                response=response,
                responses=guardrailed_texts,
                task_mappings=task_mappings,
            )

        verbose_proxy_logger.debug("Anthropic Messages: Processed output response: %s", response)

        return response

    async def process_output_streaming_response(
        self,
        responses_so_far: list[Any],
        guardrail_to_apply: "CustomGuardrail",
        litellm_logging_obj: "LiteLLMLoggingObj | None" = None,
        user_api_key_dict: "UserAPIKeyAuth | None" = None,
        request_data: dict | None = None,
    ) -> Sequence[object]:
        """
        Process output streaming response by applying guardrails to text content.

        Get the string so far, check the apply guardrail to the string so far, and return the list of responses so far.
        """
        from litellm.integrations.custom_guardrail import ModifyResponseException

        has_ended: Final = self._check_streaming_has_ended(responses_so_far)
        if has_ended:
            # build the model response from the responses_so_far
            built_response: Final = AnthropicPassthroughLoggingHandler._build_complete_streaming_response(
                all_chunks=responses_so_far,
                litellm_logging_obj=cast("LiteLLMLoggingObj", litellm_logging_obj),
                model="",
            )

            # Check if model_response is valid and has choices before accessing
            if built_response is not None and hasattr(built_response, "choices") and built_response.choices:
                model_response: Final = cast(ModelResponse, built_response)
                first_choice: Final = cast(Choices, model_response.choices[0])
                tool_calls_list: Final = cast(
                    list[ChatCompletionMessageToolCall] | None,
                    first_choice.message.tool_calls,
                )
                string_so_far = first_choice.message.content
                guardrail_inputs: Final = GenericGuardrailAPIInputs()
                if string_so_far:
                    guardrail_inputs["texts"] = [string_so_far]
                if tool_calls_list:
                    guardrail_inputs["tool_calls"] = tool_calls_list

                try:
                    prepared_request_data = self._prepare_request_data(
                        request_data,
                        model_response,
                        user_api_key_dict,
                        key="response",
                    )
                    _guardrailed_inputs = await guardrail_to_apply.apply_guardrail(
                        inputs=guardrail_inputs,
                        request_data=prepared_request_data,
                        input_type="response",
                        logging_obj=litellm_logging_obj,
                    )
                except ModifyResponseException as e:
                    if e.original_response is None:
                        e.original_response = built_response or self._build_streaming_usage_response(
                            responses_so_far, request_data
                        )
                    raise
            else:
                verbose_proxy_logger.debug("Skipping output guardrail - model response has no choices")
            return responses_so_far

        string_so_far = self.get_streaming_string_so_far(responses_so_far)
        try:
            prepared_request_data = self._prepare_request_data(
                request_data,
                responses_so_far,
                user_api_key_dict,
                key="responses",
            )
            _guardrailed_inputs = await guardrail_to_apply.apply_guardrail(
                inputs={"texts": [string_so_far]},
                request_data=prepared_request_data,
                input_type="response",
                logging_obj=litellm_logging_obj,
            )
        except ModifyResponseException as e:
            if e.original_response is None:
                e.original_response = self._build_streaming_usage_response(responses_so_far, request_data)
            raise
        return responses_so_far

    def _prepare_request_data(
        self,
        request_data: dict | None,
        response: object,
        user_api_key_dict: "UserAPIKeyAuth | None",
        key: str,
    ) -> dict:
        """Ensure request_data has the response/responses_so_far key and metadata."""
        if request_data is None:
            request_data = {key: response}
        else:
            if key not in request_data:
                request_data[key] = response

        if "litellm_metadata" not in request_data:
            user_metadata: Final = self.transform_user_api_key_dict_to_metadata(user_api_key_dict)
            if user_metadata:
                request_data["litellm_metadata"] = user_metadata
        return request_data

    @staticmethod
    def _get_response_content(response: object) -> Sequence[object]:
        """Extract content list from a dict or object response."""
        if isinstance(response, dict):
            return response.get("content", []) or []
        elif hasattr(response, "content"):
            return getattr(response, "content", None) or []
        return []

    def _extract_from_content_blocks(
        self,
        response_content: Sequence[object],
        texts_to_check: list[str],
        images_to_check: list[str],
        task_mappings: list[tuple[int, int | None]],
        tool_calls_to_check: list["ChatCompletionToolCallChunk"],
    ) -> None:
        """Extract text, images, and tool calls from content blocks."""
        for content_idx, content_block in enumerate(response_content):
            fields = self._output_block_fields(content_block)
            if fields is None:
                continue
            block_type, block_dict = fields

            if block_type in ["text", "tool_use"]:
                self._extract_output_text_and_images(
                    content_block=block_dict,
                    content_idx=content_idx,
                    texts_to_check=texts_to_check,
                    images_to_check=images_to_check,
                    task_mappings=task_mappings,
                    tool_calls_to_check=tool_calls_to_check,
                )

    @staticmethod
    def _output_block_fields(content_block: object) -> "tuple[object, Mapping[str, object]] | None":
        if isinstance(content_block, dict):
            block_dict: Final = _as_str_mapping(content_block)
            return block_dict.get("type"), block_dict
        if not hasattr(content_block, "type"):
            return None
        block_type: Final = getattr(content_block, "type", None)
        if isinstance(content_block, _ModelDumpBlock):
            return block_type, content_block.model_dump()
        return block_type, {
            "type": block_type,
            "text": getattr(content_block, "text", None),
        }

    @staticmethod
    def _build_guardrail_inputs(
        texts_to_check: list[str],
        images_to_check: list[str],
        tool_calls_to_check: list["ChatCompletionToolCallChunk"],
        response: object,
    ) -> "GenericGuardrailAPIInputs":
        """Build GenericGuardrailAPIInputs with optional images, tool calls, model."""
        inputs: Final = GenericGuardrailAPIInputs(texts=texts_to_check)
        if images_to_check:
            inputs["images"] = images_to_check
        if tool_calls_to_check:
            inputs["tool_calls"] = tool_calls_to_check
        response_model = None
        if isinstance(response, dict):
            response_model = response.get("model")
        elif hasattr(response, "model"):
            response_model = getattr(response, "model", None)
        if response_model:
            inputs["model"] = response_model
        return inputs

    def get_streaming_string_so_far(self, responses_so_far: Sequence[object]) -> str:
        """
        Parse streaming responses and extract accumulated text content.

        Handles two formats:
        1. Raw bytes in SSE (Server-Sent Events) format from Anthropic API
        2. Parsed dict objects (for backwards compatibility)

        SSE format example:
            b'event: content_block_delta\\ndata: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":" curious"}}\\n\\n'

        Dict format example:
            {
                "type": "content_block_delta",
                "index": 0,
                "delta": {
                    "type": "text_delta",
                    "text": " curious"
                }
            }
        """
        text_so_far = ""
        for response in responses_so_far:
            # Handle raw bytes in SSE format
            if isinstance(response, bytes):
                text_so_far += self._extract_text_from_sse(response)
            # Handle already-parsed dict format
            elif isinstance(response, dict):
                delta = response.get("delta") if response.get("delta") else None
                if delta and delta.get("type") == "text_delta":
                    text = delta.get("text", "")
                    if text:
                        text_so_far += text
        return text_so_far

    def _extract_text_from_sse(self, sse_bytes: bytes) -> str:
        """
        Extract text content from Server-Sent Events (SSE) format.

        Args:
            sse_bytes: Raw bytes in SSE format

        Returns:
            Accumulated text from all content_block_delta events
        """
        text = ""
        try:
            # Decode bytes to string
            sse_string: Final = sse_bytes.decode("utf-8")

            # Split by double newline to get individual events
            events: Final = sse_string.split("\n\n")

            for event in events:
                if not event.strip():
                    continue

                # Parse event lines
                lines = event.strip().split("\n")
                event_type = None
                data_line = None

                for line in lines:
                    if line.startswith("event:"):
                        event_type = line[6:].strip()
                    elif line.startswith("data:"):
                        data_line = line[5:].strip()

                # Only process content_block_delta events
                if event_type == "content_block_delta" and data_line:
                    try:
                        data: _AnthropicSSEEvent = json.loads(data_line)
                        delta: _AnthropicSSEDelta = data.get("delta", {})
                        if delta.get("type") == "text_delta":
                            text += delta.get("text", "")
                    except json.JSONDecodeError:
                        verbose_proxy_logger.warning("Failed to parse JSON from SSE data: %s", data_line)

        except Exception as e:
            verbose_proxy_logger.error("Error extracting text from SSE: %s", e)

        return text

    def _check_streaming_has_ended(self, responses_so_far: Sequence[object]) -> bool:
        """
        Check if streaming response has ended by looking for non-null stop_reason.

        Handles two formats:
        1. Raw bytes in SSE (Server-Sent Events) format from Anthropic API
        2. Parsed dict objects (for backwards compatibility)

        SSE format example:
            b'event: message_delta\\ndata: {"type":"message_delta","delta":{"stop_reason":"tool_use","stop_sequence":null},...}\\n\\n'

        Dict format example:
            {
                "type": "message_delta",
                "delta": {
                    "stop_reason": "tool_use",
                    "stop_sequence": null
                }
            }

        Returns:
            True if stop_reason is set to a non-null value, indicating stream has ended
        """
        for response in responses_so_far:
            # Handle raw bytes in SSE format
            if isinstance(response, bytes):
                try:
                    # Decode bytes to string
                    sse_string = response.decode("utf-8")

                    # Split by double newline to get individual events
                    events = sse_string.split("\n\n")

                    for event in events:
                        if not event.strip():
                            continue

                        # Parse event lines
                        lines = event.strip().split("\n")
                        event_type = None
                        data_line = None

                        for line in lines:
                            if line.startswith("event:"):
                                event_type = line[6:].strip()
                            elif line.startswith("data:"):
                                data_line = line[5:].strip()

                        # Check for message_delta event with stop_reason
                        if event_type == "message_delta" and data_line:
                            try:
                                data: _AnthropicSSEEvent = json.loads(data_line)
                                delta: _AnthropicSSEDelta = data.get("delta", {})
                                stop_reason: str | None = delta.get("stop_reason")
                                if stop_reason is not None:
                                    return True
                            except json.JSONDecodeError:
                                verbose_proxy_logger.warning("Failed to parse JSON from SSE data: %s", data_line)

                except Exception as e:
                    verbose_proxy_logger.error("Error checking streaming end in SSE: %s", e)

            # Handle already-parsed dict format
            elif isinstance(response, dict):
                if response.get("type") == "message_delta":
                    delta = response.get("delta", {})
                    stop_reason = delta.get("stop_reason")
                    if stop_reason is not None:
                        return True

        return False

    def _has_text_content(self, response: "AnthropicMessagesResponse") -> bool:
        """
        Check if response has any text content to process.

        Override this method to customize text content detection.
        """
        if isinstance(response, dict):
            response_content = response.get("content", [])
        else:
            response_content = getattr(response, "content", None) or []

        if not response_content:
            return False
        for content_block in response_content:
            # Check if this is a text block by checking the 'type' field
            if isinstance(content_block, dict) and content_block.get("type") == "text":
                content_text = content_block.get("text")
                if content_text and isinstance(content_text, str):
                    return True
        return False

    def _extract_output_text_and_images(
        self,
        content_block: Mapping[str, object],
        content_idx: int,
        texts_to_check: list[str],
        images_to_check: list[str],
        task_mappings: list[tuple[int, int | None]],
        tool_calls_to_check: list[ChatCompletionToolCallChunk] | None = None,
    ) -> None:
        """
        Extract text content, images, and tool calls from a response content block.

        Override this method to customize text/image/tool extraction logic.
        """
        content_type: Final = content_block.get("type")

        # Extract text content
        if content_type == "text":
            content_text: Final = content_block.get("text")
            if content_text and isinstance(content_text, str):
                # Simple string content
                texts_to_check.append(content_text)
                task_mappings.append((content_idx, None))

        # Extract tool calls
        elif content_type == "tool_use" and isinstance(content_block, dict):
            tool_call: Final = AnthropicConfig.convert_tool_use_to_openai_format(
                anthropic_tool_content=content_block,
                index=content_idx,
            )
            if tool_calls_to_check is None:
                tool_calls_to_check = []
            tool_calls_to_check.append(tool_call)

    async def _apply_guardrail_responses_to_output(
        self,
        response: "AnthropicMessagesResponse",
        responses: list[str],
        task_mappings: list[tuple[int, int | None]],
    ) -> None:
        """
        Apply guardrail responses back to output response.

        Override this method to customize how responses are applied.
        """
        for task_idx, guardrail_response in enumerate(responses):
            mapping = task_mappings[task_idx]
            content_idx = cast(int, mapping[0])

            # Handle both dict and object responses
            response_content: Sequence[object] = []
            if isinstance(response, dict):
                response_content = response.get("content", []) or []
            elif hasattr(response, "content"):
                content = getattr(response, "content", None)
                response_content = content or []
            else:
                continue

            if not response_content:
                continue

            # Get the content block at the index
            if content_idx >= len(response_content):
                continue

            content_block = _content_block_at(response_content, content_idx)

            # Verify it's a text block and update the text field
            # Handle both dict and Pydantic object content blocks
            if isinstance(content_block, dict):
                block = _as_writable(content_block)
                if block.get("type") == "text":
                    block["text"] = guardrail_response
            elif hasattr(content_block, "type") and getattr(content_block, "type", None) == "text":
                # Update Pydantic object's text attribute
                if isinstance(content_block, _TextAttrBlock):
                    content_block.text = guardrail_response
