fix(anthropic-messages): apply cache_control_injection_points on /v1/messages path (#30341)

cache_control_injection_points was only consumed by the chat/completions
prompt-management hook; on the native Anthropic /v1/messages path it was
forwarded unused, so deployment-level cache injection was silently dropped
(cache_creation_input_tokens stayed 0 for Anthropic-native clients).

Add AnthropicCacheControlHook.apply_to_anthropic_messages_request to inject
cache_control at block level for system / tools / message locations (the only
forms /v1/messages accepts), wire it into the native anthropic_messages
handler, and pop the param so it does not leak upstream as an unknown field.
A {location: message, role: system} config is redirected to the top-level
system prompt so the same YAML works on both endpoints.

Injection respects Anthropic's 4-block cache_control limit shared across
system, tools, and messages: client-supplied markers count toward the cap and
are never overwritten, a slot is reserved per Bedrock tool_config point, and
injection stops once the budget is exhausted. Locations this path cannot
represent (tool_config) are forwarded downstream instead of being silently
consumed, mirroring get_chat_completion_prompt's remaining_points pass-through.

Built on litellm_internal_staging. Refs BerriAI/litellm#30293

litellm/integrations/anthropic_cache_control_hook.py

@@ -31,6 +31,14 @@ else:
 # breakpoints: "A maximum of 4 blocks with cache_control may be provided."
 MAX_CACHE_CONTROL_BLOCKS = 4
 
+# Tool types Anthropic rejects `cache_control` on. Mirrors the exclusion in
+# litellm/llms/anthropic/chat/transformation.py so the native /v1/messages path
+# never marks a tool the API would 400 on.
+NON_CACHEABLE_TOOL_TYPES = (
+    "tool_search_tool_regex_20251119",
+    "tool_search_tool_bm25_20251119",
+)
+
 
 class AnthropicCacheControlHook(CustomPromptManagement):
     def get_chat_completion_prompt(
@@ -242,6 +250,330 @@ class AnthropicCacheControlHook(CustomPromptManagement):
                 message_content[-1]["cache_control"] = control  # type: ignore
         return message
 
+    @staticmethod
+    def apply_to_anthropic_messages_request(
+        messages: List[Dict],
+        system: Optional[Union[str, List[Dict]]],
+        tools: Optional[List[Dict]],
+        non_default_params: Dict,
+    ) -> Tuple[List[Dict], Optional[Union[str, List[Dict]]], Optional[List[Dict]]]:
+        """Apply ``cache_control_injection_points`` to a native Anthropic
+        ``/v1/messages`` request.
+
+        The OpenAI chat/completions path injects cache control via
+        ``get_chat_completion_prompt`` (message-level ``cache_control``). The
+        native Anthropic Messages endpoint differs in two ways that the OpenAI
+        helper does not handle, so this method exists alongside it:
+
+        1. The system prompt is a top-level ``system`` parameter, not a
+           ``role: system`` entry in ``messages``.
+        2. Anthropic rejects message-level ``cache_control``; the marker must
+           live on a content *block* (or directly on a ``system`` / tool block).
+
+        Injection is capped at ``MAX_CACHE_CONTROL_BLOCKS`` markers shared across
+        system, tools, and messages (client-supplied markers count toward the cap
+        and are never overwritten), so the outbound payload never exceeds the
+        Anthropic/Bedrock limit.
+
+        Args:
+            messages: Anthropic-format messages (each ``content`` is a string or
+                a list of content blocks).
+            system: Top-level system prompt (string or list of system blocks).
+            tools: Tool definitions sent to the model.
+            non_default_params: Request params; ``cache_control_injection_points``
+                is read and popped from here so it is not forwarded upstream as
+                an unknown field. Any injection points this method does not
+                handle (e.g. Bedrock ``tool_config``) are written back so
+                downstream provider transforms still receive them.
+
+        Returns:
+            A tuple of (messages, system, tools) with cache control applied. The
+            inputs are deep-copied, so the caller's originals are never mutated.
+        """
+        injection_points: List[CacheControlInjectionPoint] = non_default_params.pop(
+            "cache_control_injection_points", []
+        )
+        if not injection_points:
+            return messages, system, tools
+
+        processed_messages = copy.deepcopy(messages)
+        processed_system = copy.deepcopy(system)
+        processed_tools = copy.deepcopy(tools)
+
+        # Injection points this native path cannot represent in the payload
+        # (e.g. Bedrock `tool_config`). Forwarded rather than dropped so the
+        # param is never silently swallowed. Note these are inert on the native
+        # `/v1/messages` path: it uses the Anthropic Invoke transform, while only
+        # the Converse transform on the chat/completions path consumes
+        # `tool_config`. No block budget is reserved for them here (unlike
+        # get_chat_completion_prompt, which reserves a slot because Converse does
+        # append a tool_config cachePoint downstream).
+        remaining_points: List[CacheControlInjectionPoint] = []
+
+        # Anthropic/Bedrock reject requests with more than MAX_CACHE_CONTROL_BLOCKS
+        # cache_control breakpoints. Client-supplied markers count toward that
+        # limit, so inject in config order, never overwrite a client's existing
+        # marker, and stop once the budget is exhausted.
+        max_blocks = MAX_CACHE_CONTROL_BLOCKS
+        used_blocks = AnthropicCacheControlHook._count_request_cache_control_blocks(
+            system=processed_system, tools=processed_tools, messages=processed_messages
+        )
+        limit_reached = False
+
+        for point in injection_points:
+            location = point.get("location")
+
+            if location not in ("system", "tools", "message"):
+                # Unhandled location (tool_config / future types): forward it.
+                remaining_points.append(point)
+                continue
+
+            if used_blocks >= max_blocks:
+                # Out of payload-level budget; keep scanning so any remaining
+                # tool_config points are still forwarded downstream.
+                limit_reached = True
+                continue
+
+            control = AnthropicCacheControlHook._control_from_point(point)
+
+            if location == "system":
+                processed_system, added = (
+                    AnthropicCacheControlHook._insert_cache_control_in_system(
+                        system=processed_system, control=control
+                    )
+                )
+                used_blocks += int(added)
+            elif location == "tools":
+                processed_tools, added = (
+                    AnthropicCacheControlHook._insert_cache_control_in_tools(
+                        tools=processed_tools, control=control
+                    )
+                )
+                used_blocks += int(added)
+            else:  # message
+                message_point = cast(CacheControlMessageInjectionPoint, point)
+                # `role: system` has no message-level analogue on /v1/messages
+                # (the system prompt is the top-level `system` param), so
+                # redirect it to the system prompt when one exists. This keeps
+                # the same YAML config working across both endpoints.
+                if (
+                    message_point.get("role") == "system"
+                    and message_point.get("index") is None
+                    and not any(m.get("role") == "system" for m in processed_messages)
+                    and processed_system is not None
+                ):
+                    processed_system, added = (
+                        AnthropicCacheControlHook._insert_cache_control_in_system(
+                            system=processed_system, control=control
+                        )
+                    )
+                    used_blocks += int(added)
+                else:
+                    processed_messages, added_count, hit_limit = (
+                        AnthropicCacheControlHook._process_anthropic_message_injection(
+                            point=message_point,
+                            messages=processed_messages,
+                            used_blocks=used_blocks,
+                            max_blocks=max_blocks,
+                        )
+                    )
+                    used_blocks += added_count
+                    limit_reached = limit_reached or hit_limit
+
+        if limit_reached:
+            verbose_logger.warning(
+                f"AnthropicCacheControlHook: Reached the Anthropic limit of "
+                f"{MAX_CACHE_CONTROL_BLOCKS} cache_control blocks on the "
+                f"/v1/messages path. Skipping further injection."
+            )
+
+        if remaining_points:
+            non_default_params["cache_control_injection_points"] = remaining_points
+
+        return processed_messages, processed_system, processed_tools
+
+    @staticmethod
+    def _control_from_point(
+        point: CacheControlInjectionPoint,
+    ) -> ChatCompletionCachedContent:
+        """Resolve the ``cache_control`` value for an injection point.
+
+        Defaults to ``{"type": "ephemeral"}`` when no explicit control is given.
+        ``control`` is not declared on every member of the
+        ``CacheControlInjectionPoint`` union (e.g. ``tool_config`` has none), so
+        it is read via a structural dict view rather than a misleading
+        ``Optional`` cast on a value that is immediately defaulted to non-None.
+        """
+        control = cast(Dict[str, Any], point).get("control")
+        return control or ChatCompletionCachedContent(type="ephemeral")
+
+    @staticmethod
+    def _count_request_cache_control_blocks(
+        system: Optional[Union[str, List[Dict]]],
+        tools: Optional[List[Dict]],
+        messages: List[Dict],
+    ) -> int:
+        """Count cache_control markers already present across a /v1/messages
+        payload (system blocks, tool definitions, and message content blocks).
+
+        Client-supplied markers count toward Anthropic's limit, so they must be
+        tallied before injecting more. A string ``system`` carries no marker.
+        """
+        count = 0
+        if isinstance(system, list):
+            count += sum(
+                1
+                for block in system
+                if isinstance(block, dict) and block.get("cache_control") is not None
+            )
+        if isinstance(tools, list):
+            count += sum(
+                1
+                for tool in tools
+                if isinstance(tool, dict) and tool.get("cache_control") is not None
+            )
+        for message in messages:
+            count += AnthropicCacheControlHook._count_cache_control_blocks(
+                cast(AllMessageValues, message)
+            )
+            # `_count_cache_control_blocks` only sees message-level and top-level
+            # content-block markers. Anthropic `tool_result` blocks nest their
+            # own content list, whose items may carry their own cache_control;
+            # count those too so the cap reflects the real marker total and we
+            # never push a near-full request over the limit.
+            content = message.get("content")
+            if isinstance(content, list):
+                for block in content:
+                    if not isinstance(block, dict):
+                        continue
+                    nested = block.get("content")
+                    if isinstance(nested, list):
+                        count += sum(
+                            1
+                            for item in nested
+                            if isinstance(item, dict)
+                            and item.get("cache_control") is not None
+                        )
+        return count
+
+    @staticmethod
+    def _insert_cache_control_in_system(
+        system: Optional[Union[str, List[Dict]]],
+        control: ChatCompletionCachedContent,
+    ) -> Tuple[Optional[Union[str, List[Dict]]], bool]:
+        """Insert cache control on the last block of an Anthropic system prompt.
+
+        A string system prompt is promoted to a single-block list so the marker
+        can live at block level (Anthropic rejects a bare string + cache marker).
+        No-op when there is no system prompt or the last block already carries a
+        marker (a client's marker is preserved).
+
+        Returns ``(system, added)`` where ``added`` is True iff a new marker was
+        written, so the caller can track the cache_control block budget.
+        """
+        if system is None:
+            return system, False
+        if isinstance(system, str):
+            if system == "":
+                return system, False
+            return [{"type": "text", "text": system, "cache_control": control}], True
+        if isinstance(system, list) and len(system) > 0:
+            last_block = system[-1]
+            if isinstance(last_block, dict) and last_block.get("cache_control") is None:
+                last_block["cache_control"] = control
+                return system, True
+        return system, False
+
+    @staticmethod
+    def _insert_cache_control_in_tools(
+        tools: Optional[List[Dict]],
+        control: ChatCompletionCachedContent,
+    ) -> Tuple[Optional[List[Dict]], bool]:
+        """Insert cache control on the last *cacheable* tool definition.
+
+        Marking a tool caches the tool-list prefix up to and including it
+        (Anthropic caches the prefix). Tool-search tools (see
+        ``NON_CACHEABLE_TOOL_TYPES``) reject ``cache_control`` and are skipped,
+        so the marker lands on the last tool that supports it — otherwise a valid
+        request would 400. No-op when there are no cacheable tools or that tool
+        already carries a marker.
+
+        Returns ``(tools, added)`` where ``added`` is True iff a new marker was
+        written, so the caller can track the cache_control block budget.
+        """
+        if not tools:
+            return tools, False
+        for tool in reversed(tools):
+            if not isinstance(tool, dict):
+                continue
+            if tool.get("type") in NON_CACHEABLE_TOOL_TYPES:
+                continue
+            if tool.get("cache_control") is None:
+                tool["cache_control"] = control
+                return tools, True
+            return tools, False  # last cacheable tool already marked
+        return tools, False
+
+    @staticmethod
+    def _process_anthropic_message_injection(
+        point: CacheControlMessageInjectionPoint,
+        messages: List[Dict],
+        used_blocks: int,
+        max_blocks: int,
+    ) -> Tuple[List[Dict], int, bool]:
+        """Apply block-level cache control to targeted messages within budget.
+
+        Reuses :meth:`_resolve_target_indices` for index / role targeting
+        (negative index support, out-of-bounds warning) but writes the marker on
+        a content *block* rather than the message object, because Anthropic's
+        ``/v1/messages`` endpoint rejects message-level ``cache_control``.
+        Messages already carrying a marker are left untouched (client TTL
+        preserved) and injection stops once the shared block budget is exhausted.
+
+        Returns ``(messages, added, limit_reached)``.
+        """
+        control = AnthropicCacheControlHook._control_from_point(point)
+        target_indices = AnthropicCacheControlHook._resolve_target_indices(
+            point=point, messages=cast(List[AllMessageValues], messages)
+        )
+
+        added = 0
+        limit_reached = False
+        for target_index in target_indices:
+            if used_blocks + added >= max_blocks:
+                limit_reached = True
+                break
+            if AnthropicCacheControlHook._message_has_cache_control(
+                cast(AllMessageValues, messages[target_index])
+            ):
+                # Client already marked this message; don't overwrite it.
+                continue
+            AnthropicCacheControlHook._insert_cache_control_in_message_block(
+                messages[target_index], control
+            )
+            added += 1
+        return messages, added, limit_reached
+
+    @staticmethod
+    def _insert_cache_control_in_message_block(
+        message: Dict, control: ChatCompletionCachedContent
+    ) -> Dict:
+        """Insert cache control on the last content block of one message.
+
+        A string ``content`` is promoted to a single text block so the marker
+        can live at block level, as required by ``/v1/messages``.
+        """
+        message_content = message.get("content", None)
+        if isinstance(message_content, str):
+            message["content"] = [
+                {"type": "text", "text": message_content, "cache_control": control}
+            ]
+        elif isinstance(message_content, list) and len(message_content) > 0:
+            last_block = message_content[-1]
+            if isinstance(last_block, dict):
+                last_block["cache_control"] = control
+        return message
+
     @property
     def integration_name(self) -> str:
         """Return the integration name for this hook."""

litellm/llms/anthropic/experimental_pass_through/messages/handler.py

@@ -179,6 +179,49 @@ async def _try_websearch_short_circuit(
     return None
 
 
+def _supports_native_anthropic_messages(
+    model: str,
+    custom_llm_provider: Optional[str],
+    api_base: Optional[str] = None,
+    api_key: Optional[str] = None,
+) -> bool:
+    """Return True when the resolved provider serves ``/v1/messages`` natively.
+
+    Mirrors the native-vs-adapter predicate in ``anthropic_messages_handler``
+    (a non-None ``get_provider_anthropic_messages_config``). Fallback providers
+    routed through the chat/completions or Responses adapters return False — an
+    Anthropic block-level ``cache_control`` rewrite is only correct for the
+    native path; on the fallback path the downstream OpenAI hook owns caching.
+
+    ``api_base`` / ``api_key`` are forwarded to ``get_llm_provider`` so a native
+    provider selected purely by endpoint (e.g. an ``/anthropic`` ``api_base``)
+    resolves the same way it will in the handler, instead of being misread as a
+    fallback and having its cache injection skipped.
+    """
+    from litellm.types.utils import LlmProviders
+
+    try:
+        resolved_model, resolved_provider, _, _ = litellm.get_llm_provider(
+            model=model,
+            custom_llm_provider=custom_llm_provider,
+            api_base=api_base,
+            api_key=api_key,
+        )
+    except Exception:
+        return False
+    if not resolved_provider or resolved_provider not in [
+        provider.value for provider in LlmProviders
+    ]:
+        return False
+    return (
+        ProviderConfigManager.get_provider_anthropic_messages_config(
+            model=resolved_model,
+            provider=LlmProviders(resolved_provider),
+        )
+        is not None
+    )
+
+
 @client
 async def anthropic_messages(
     max_tokens: int,
@@ -187,7 +230,7 @@ async def anthropic_messages(
     metadata: Optional[Dict] = None,
     stop_sequences: Optional[List[str]] = None,
     stream: Optional[bool] = False,
-    system: Optional[str] = None,
+    system: Optional[Union[str, List[Dict]]] = None,
     temperature: Optional[float] = None,
     thinking: Optional[Dict] = None,
     tool_choice: Optional[Dict] = None,
@@ -260,6 +303,43 @@ async def anthropic_messages(
     # Merge back any other modifications
     kwargs.update(request_kwargs)
 
+    # Apply deployment-level `cache_control_injection_points` to the native
+    # Anthropic Messages payload. The OpenAI chat/completions path injects this
+    # via `litellm_logging_obj.async_get_chat_completion_prompt`, but that hook
+    # never runs on the native `/v1/messages` path, so config-level cache
+    # injection was silently dropped here (cache_creation_input_tokens stayed
+    # 0). Rewrite `messages` / `system` / `tools` at block level (the only form
+    # `/v1/messages` accepts) and pop the param so it does not leak upstream as
+    # an unknown field; locations not representable here (e.g. Bedrock
+    # tool_config) are forwarded but inert on this native path. See
+    # BerriAI/litellm#30293.
+    #
+    # Gate on native support: providers without a native Anthropic Messages
+    # config fall back to the chat/completions (or Responses) adapter, which
+    # applies caching via the OpenAI-path hook in the format those providers
+    # understand. Rewriting to Anthropic block form there would consume the
+    # param and silently break caching for e.g. Gemini, so leave it untouched.
+    if kwargs.get(
+        "cache_control_injection_points"
+    ) and _supports_native_anthropic_messages(
+        model=model,
+        custom_llm_provider=custom_llm_provider,
+        api_base=api_base,
+        api_key=api_key,
+    ):
+        from litellm.integrations.anthropic_cache_control_hook import (
+            AnthropicCacheControlHook,
+        )
+
+        messages, system, tools = (
+            AnthropicCacheControlHook.apply_to_anthropic_messages_request(
+                messages=messages,
+                system=system,
+                tools=tools,
+                non_default_params=kwargs,
+            )
+        )
+
     # Short-circuit web-search-only requests: detect the pattern, execute
     # search directly via Tavily/Perplexity, and return a synthetic response
     # without ever touching the backend LLM or the adapter path.
@@ -360,7 +440,7 @@ def anthropic_messages_handler(
     metadata: Optional[Dict] = None,
     stop_sequences: Optional[List[str]] = None,
     stream: Optional[bool] = False,
-    system: Optional[str] = None,
+    system: Optional[Union[str, List[Dict]]] = None,
     temperature: Optional[float] = None,
     thinking: Optional[Dict] = None,
     tool_choice: Optional[Dict] = None,

litellm/types/integrations/anthropic_cache_control_hook.py

@@ -22,7 +22,46 @@ class CacheControlToolConfigInjectionPoint(TypedDict):
     location: Literal["tool_config"]
 
 
+class _CacheControlSystemInjectionPointBase(TypedDict):
+    location: Literal["system"]
+
+
+class CacheControlSystemInjectionPoint(
+    _CacheControlSystemInjectionPointBase, total=False
+):
+    """Type for system-prompt injection points (Anthropic ``/v1/messages``).
+
+    On the native Anthropic Messages endpoint the system prompt is a top-level
+    ``system`` parameter rather than a ``role: system`` entry inside
+    ``messages`` (as it is for OpenAI chat/completions), so it needs its own
+    targetable location. ``location`` is required (it lives on the base class);
+    only ``control`` is optional.
+    """
+
+    control: Optional[ChatCompletionCachedContent]
+
+
+class _CacheControlToolsInjectionPointBase(TypedDict):
+    location: Literal["tools"]
+
+
+class CacheControlToolsInjectionPoint(
+    _CacheControlToolsInjectionPointBase, total=False
+):
+    """Type for tool-list injection points (Anthropic ``/v1/messages``).
+
+    Caches the (typically long, static) tool list sent by Anthropic-native
+    clients such as Claude Code. Distinct from the Bedrock-only ``tool_config``
+    location, which is consumed by the Converse transform. ``location`` is
+    required (it lives on the base class); only ``control`` is optional.
+    """
+
+    control: Optional[ChatCompletionCachedContent]
+
+
 CacheControlInjectionPoint = Union[
     CacheControlMessageInjectionPoint,
     CacheControlToolConfigInjectionPoint,
+    CacheControlSystemInjectionPoint,
+    CacheControlToolsInjectionPoint,
 ]

tests/test_litellm/integrations/test_anthropic_cache_control_hook_messages.py

@@ -0,0 +1,534 @@
+"""Unit + wiring tests for cache_control_injection_points on the native
+Anthropic ``/v1/messages`` path. Regression coverage for
+BerriAI/litellm#30293, where deployment-level cache injection was silently
+dropped on ``/v1/messages`` (worked only on ``/chat/completions``).
+"""
+
+import os
+import sys
+
+import pytest
+
+sys.path.insert(0, os.path.abspath("../../.."))
+
+from litellm.integrations.anthropic_cache_control_hook import AnthropicCacheControlHook
+
+EPHEMERAL = {"type": "ephemeral"}
+
+
+def _apply(messages, system, tools, injection_points):
+    """Helper: run the hook entrypoint with the given injection points."""
+    non_default_params = {"cache_control_injection_points": injection_points}
+    return AnthropicCacheControlHook.apply_to_anthropic_messages_request(
+        messages=messages,
+        system=system,
+        tools=tools,
+        non_default_params=non_default_params,
+    )
+
+
+class TestApplyToAnthropicMessagesRequest:
+    def test_noop_when_no_injection_points(self):
+        messages = [{"role": "user", "content": "hi"}]
+        out_messages, out_system, out_tools = _apply(messages, "sys", None, [])
+        assert out_messages == messages
+        assert out_system == "sys"
+        assert out_tools is None
+
+    def test_pops_injection_points_from_params(self):
+        non_default_params = {
+            "cache_control_injection_points": [{"location": "system"}],
+            "other": 1,
+        }
+        AnthropicCacheControlHook.apply_to_anthropic_messages_request(
+            messages=[{"role": "user", "content": "hi"}],
+            system="sys",
+            tools=None,
+            non_default_params=non_default_params,
+        )
+        assert "cache_control_injection_points" not in non_default_params
+        assert non_default_params["other"] == 1
+
+    def test_string_system_promoted_to_block_with_cache_control(self):
+        _, out_system, _ = _apply(
+            [{"role": "user", "content": "hi"}],
+            "long system prompt",
+            None,
+            [{"location": "system"}],
+        )
+        assert out_system == [
+            {
+                "type": "text",
+                "text": "long system prompt",
+                "cache_control": EPHEMERAL,
+            }
+        ]
+
+    def test_list_system_marks_last_block(self):
+        system = [
+            {"type": "text", "text": "block 1"},
+            {"type": "text", "text": "block 2"},
+        ]
+        _, out_system, _ = _apply(
+            [{"role": "user", "content": "hi"}], system, None, [{"location": "system"}]
+        )
+        assert "cache_control" not in out_system[0]
+        assert out_system[1]["cache_control"] == EPHEMERAL
+
+    def test_tools_marks_last_tool(self):
+        tools = [{"name": "a"}, {"name": "b"}]
+        _, _, out_tools = _apply(
+            [{"role": "user", "content": "hi"}], None, tools, [{"location": "tools"}]
+        )
+        assert "cache_control" not in out_tools[0]
+        assert out_tools[1]["cache_control"] == EPHEMERAL
+
+    def test_tools_skips_non_cacheable_tool_search_tool(self):
+        # Anthropic rejects cache_control on tool-search tools; the marker must
+        # land on the last *cacheable* tool, not the trailing search tool.
+        tools = [
+            {"name": "real_tool"},
+            {"type": "tool_search_tool_regex_20251119", "name": "search"},
+        ]
+        _, _, out_tools = _apply(
+            [{"role": "user", "content": "hi"}], None, tools, [{"location": "tools"}]
+        )
+        assert out_tools[0]["cache_control"] == EPHEMERAL
+        assert "cache_control" not in out_tools[1]
+
+    def test_tools_all_non_cacheable_is_noop(self):
+        tools = [{"type": "tool_search_tool_bm25_20251119", "name": "s"}]
+        _, _, out_tools = _apply(
+            [{"role": "user", "content": "hi"}], None, tools, [{"location": "tools"}]
+        )
+        assert "cache_control" not in out_tools[0]
+
+    def test_message_string_content_promoted_to_block(self):
+        messages = [{"role": "user", "content": "hello"}]
+        out_messages, _, _ = _apply(
+            messages, None, None, [{"location": "message", "index": -1}]
+        )
+        assert out_messages[0]["content"] == [
+            {"type": "text", "text": "hello", "cache_control": EPHEMERAL}
+        ]
+
+    def test_message_list_content_marks_last_block(self):
+        messages = [
+            {
+                "role": "user",
+                "content": [
+                    {"type": "text", "text": "a"},
+                    {"type": "text", "text": "b"},
+                ],
+            }
+        ]
+        out_messages, _, _ = _apply(
+            messages, None, None, [{"location": "message", "index": -1}]
+        )
+        content = out_messages[0]["content"]
+        assert "cache_control" not in content[0]
+        assert content[1]["cache_control"] == EPHEMERAL
+
+    def test_message_targets_by_role(self):
+        messages = [
+            {"role": "user", "content": "first"},
+            {"role": "assistant", "content": "reply"},
+            {"role": "user", "content": "second"},
+        ]
+        out_messages, _, _ = _apply(
+            messages, None, None, [{"location": "message", "role": "user"}]
+        )
+        assert out_messages[0]["content"][0]["cache_control"] == EPHEMERAL
+        assert out_messages[2]["content"][0]["cache_control"] == EPHEMERAL
+        # assistant message untouched (still a string)
+        assert out_messages[1]["content"] == "reply"
+
+    def test_role_system_auto_translates_to_system_prompt(self):
+        # The customer's exact /chat-completions config: {message, role: system}.
+        # On /v1/messages there is no role:system message, so it must target the
+        # top-level system prompt instead of silently matching nothing.
+        messages = [{"role": "user", "content": "hi"}]
+        _, out_system, _ = _apply(
+            messages, "system prompt", None, [{"location": "message", "role": "system"}]
+        )
+        assert out_system == [
+            {"type": "text", "text": "system prompt", "cache_control": EPHEMERAL}
+        ]
+
+    def test_role_system_not_translated_when_system_message_present(self):
+        # If an explicit role:system message exists, target it (don't redirect).
+        messages = [
+            {"role": "system", "content": "in-band system"},
+            {"role": "user", "content": "hi"},
+        ]
+        out_messages, out_system, _ = _apply(
+            messages,
+            "top-level system",
+            None,
+            [{"location": "message", "role": "system"}],
+        )
+        assert out_messages[0]["content"][0]["cache_control"] == EPHEMERAL
+        # top-level system prompt left alone
+        assert out_system == "top-level system"
+
+    def test_role_system_noop_when_no_system_at_all(self):
+        messages = [{"role": "user", "content": "hi"}]
+        out_messages, out_system, _ = _apply(
+            messages, None, None, [{"location": "message", "role": "system"}]
+        )
+        assert out_messages[0]["content"] == "hi"
+        assert out_system is None
+
+    def test_explicit_control_value_respected(self):
+        control = {"type": "ephemeral", "ttl": "1h"}
+        _, out_system, _ = _apply(
+            [{"role": "user", "content": "hi"}],
+            "sys",
+            None,
+            [{"location": "system", "control": control}],
+        )
+        assert out_system[0]["cache_control"] == control
+
+    def test_multiple_injection_points(self):
+        messages = [{"role": "user", "content": "hi"}]
+        out_messages, out_system, out_tools = _apply(
+            messages,
+            "sys",
+            [{"name": "t"}],
+            [
+                {"location": "system"},
+                {"location": "tools"},
+                {"location": "message", "index": -1},
+            ],
+        )
+        assert out_system[0]["cache_control"] == EPHEMERAL
+        assert out_tools[0]["cache_control"] == EPHEMERAL
+        assert out_messages[0]["content"][0]["cache_control"] == EPHEMERAL
+
+    def test_out_of_bounds_index_is_noop(self):
+        messages = [{"role": "user", "content": "hi"}]
+        out_messages, _, _ = _apply(
+            messages, None, None, [{"location": "message", "index": 5}]
+        )
+        assert out_messages[0]["content"] == "hi"
+
+    def test_inputs_not_mutated(self):
+        messages = [{"role": "user", "content": "hi"}]
+        system = "sys"
+        tools = [{"name": "t"}]
+        _apply(
+            messages,
+            system,
+            tools,
+            [
+                {"location": "system"},
+                {"location": "tools"},
+                {"location": "message", "index": -1},
+            ],
+        )
+        # originals untouched (deep copy)
+        assert messages == [{"role": "user", "content": "hi"}]
+        assert system == "sys"
+        assert tools == [{"name": "t"}]
+
+    def test_tool_config_injection_point_forwarded_downstream(self):
+        # `tool_config` (Bedrock) is not representable in the /v1/messages
+        # payload here; it must be forwarded downstream for the provider
+        # transform (e.g. Bedrock Converse) to consume, NOT silently dropped.
+        # Regression guard for the original /v1/messages bypass: before, an
+        # unrecognised location was popped and lost.
+        non_default_params = {
+            "cache_control_injection_points": [{"location": "tool_config"}],
+        }
+        out_messages, out_system, out_tools = (
+            AnthropicCacheControlHook.apply_to_anthropic_messages_request(
+                messages=[{"role": "user", "content": "hi"}],
+                system="sys",
+                tools=[{"name": "t"}],
+                non_default_params=non_default_params,
+            )
+        )
+        # forwarded for downstream handling, not consumed
+        assert non_default_params["cache_control_injection_points"] == [
+            {"location": "tool_config"}
+        ]
+        # payload left untouched for the location we don't handle here
+        assert out_system == "sys"
+        assert out_tools == [{"name": "t"}]
+        assert out_messages == [{"role": "user", "content": "hi"}]
+
+    def _count_all_blocks(self, messages, system, tools):
+        """Count cache_control markers across system + tools + messages."""
+        total = 0
+        if isinstance(system, list):
+            total += sum(1 for b in system if b.get("cache_control"))
+        if isinstance(tools, list):
+            total += sum(1 for t in tools if t.get("cache_control"))
+        for m in messages:
+            content = m.get("content")
+            if isinstance(content, list):
+                total += sum(
+                    1 for b in content if isinstance(b, dict) and b.get("cache_control")
+                )
+        return total
+
+    def test_respects_max_cache_control_blocks_limit(self):
+        # role:user matches 6 messages, but Anthropic allows at most 4 markers.
+        messages = [{"role": "user", "content": f"m{i}"} for i in range(6)]
+        out_messages, _, _ = _apply(
+            messages, None, None, [{"location": "message", "role": "user"}]
+        )
+        marked = sum(
+            1
+            for m in out_messages
+            if isinstance(m["content"], list) and m["content"][-1].get("cache_control")
+        )
+        assert marked == 4  # capped at MAX_CACHE_CONTROL_BLOCKS
+
+    def test_existing_cache_control_counts_toward_limit(self):
+        # Client already marked 3 message blocks; only 1 slot remains.
+        marked_block = [{"type": "text", "text": "x", "cache_control": EPHEMERAL}]
+        messages = [
+            {"role": "user", "content": list(marked_block)},
+            {"role": "user", "content": list(marked_block)},
+            {"role": "user", "content": list(marked_block)},
+            {"role": "user", "content": "d"},
+            {"role": "user", "content": "e"},
+        ]
+        out_messages, out_system, out_tools = _apply(
+            messages, None, None, [{"location": "message", "role": "user"}]
+        )
+        # 3 client markers preserved + exactly 1 injected = 4 total, not 5.
+        assert self._count_all_blocks(out_messages, out_system, out_tools) == 4
+
+    def test_system_tools_messages_share_block_budget(self):
+        # system + tools + 5 role-matched messages would be 7 markers; cap at 4.
+        messages = [{"role": "user", "content": f"m{i}"} for i in range(5)]
+        out_messages, out_system, out_tools = _apply(
+            messages,
+            "sys",
+            [{"name": "t"}],
+            [
+                {"location": "system"},
+                {"location": "tools"},
+                {"location": "message", "role": "user"},
+            ],
+        )
+        assert self._count_all_blocks(out_messages, out_system, out_tools) == 4
+
+    def test_tool_config_does_not_reserve_budget_on_native_path(self):
+        # tool_config is inert on the native /v1/messages path (only the
+        # chat/completions Converse transform consumes it), so it must NOT
+        # reserve a block slot here: all 4 markers go to the message points,
+        # and tool_config is still forwarded for completeness.
+        messages = [{"role": "user", "content": f"m{i}"} for i in range(6)]
+        non_default_params = {
+            "cache_control_injection_points": [
+                {"location": "message", "role": "user"},
+                {"location": "tool_config"},
+            ],
+        }
+        out_messages, out_system, out_tools = (
+            AnthropicCacheControlHook.apply_to_anthropic_messages_request(
+                messages=messages,
+                system=None,
+                tools=None,
+                non_default_params=non_default_params,
+            )
+        )
+        assert self._count_all_blocks(out_messages, out_system, out_tools) == 4
+        assert non_default_params["cache_control_injection_points"] == [
+            {"location": "tool_config"}
+        ]
+
+    def test_nested_tool_result_markers_count_toward_limit(self):
+        # 4 markers already present inside tool_result.content nested blocks =
+        # cap reached. A further system injection would be the 5th and must be
+        # skipped (Anthropic rejects >4 cache_control blocks).
+        messages = [
+            {
+                "role": "user",
+                "content": [
+                    {
+                        "type": "tool_result",
+                        "tool_use_id": f"t{i}",
+                        "content": [
+                            {
+                                "type": "text",
+                                "text": f"r{i}",
+                                "cache_control": EPHEMERAL,
+                            }
+                        ],
+                    }
+                    for i in range(4)
+                ],
+            }
+        ]
+        _, out_system, _ = _apply(messages, "sys", None, [{"location": "system"}])
+        # budget already full from the 4 nested markers -> system left untouched
+        assert out_system == "sys"
+
+    def test_does_not_overwrite_existing_message_marker(self):
+        # A message the client already marked is left untouched.
+        messages = [
+            {
+                "role": "user",
+                "content": [
+                    {
+                        "type": "text",
+                        "text": "keep",
+                        "cache_control": {"type": "ephemeral", "ttl": "1h"},
+                    }
+                ],
+            }
+        ]
+        out_messages, _, _ = _apply(
+            messages, None, None, [{"location": "message", "role": "user"}]
+        )
+        # original 1h TTL preserved, not overwritten with default ephemeral
+        assert out_messages[0]["content"][0]["cache_control"] == {
+            "type": "ephemeral",
+            "ttl": "1h",
+        }
+
+    def test_mixed_known_and_unknown_locations(self):
+        # Known locations are applied to the payload; unknown ones (tool_config)
+        # are forwarded. The two must not interfere.
+        non_default_params = {
+            "cache_control_injection_points": [
+                {"location": "system"},
+                {"location": "tool_config"},
+            ],
+        }
+        _, out_system, _ = (
+            AnthropicCacheControlHook.apply_to_anthropic_messages_request(
+                messages=[{"role": "user", "content": "hi"}],
+                system="sys",
+                tools=None,
+                non_default_params=non_default_params,
+            )
+        )
+        # system applied inline
+        assert out_system[0]["cache_control"] == EPHEMERAL
+        # only the unhandled tool_config point survives for downstream
+        assert non_default_params["cache_control_injection_points"] == [
+            {"location": "tool_config"}
+        ]
+
+
+class TestWiringIntoAnthropicMessagesHandler:
+    """End-to-end: deployment-level cache_control_injection_points must reach
+    the outbound /v1/messages payload via the native handler.
+    """
+
+    @pytest.mark.asyncio
+    async def test_injection_applied_on_native_messages_path(self, monkeypatch):
+        from litellm.llms.anthropic.experimental_pass_through.messages import handler
+
+        captured = {}
+
+        def fake_handler(*args, **kwargs):
+            # The native path forwards system/tools inside
+            # `anthropic_messages_optional_request_params` and the remaining
+            # caller kwargs inside `kwargs`.
+            captured["optional_params"] = kwargs.get(
+                "anthropic_messages_optional_request_params"
+            )
+            captured["inner_kwargs"] = kwargs.get("kwargs") or {}
+            return {"id": "msg_test", "type": "message", "content": []}
+
+        monkeypatch.setattr(
+            handler.base_llm_http_handler,
+            "anthropic_messages_handler",
+            fake_handler,
+        )
+
+        await handler.anthropic_messages(
+            max_tokens=100,
+            messages=[{"role": "user", "content": "hello"}],
+            model="anthropic/claude-sonnet-4-5",
+            system="a long cached system prompt",
+            custom_llm_provider="anthropic",
+            cache_control_injection_points=[
+                {"location": "message", "role": "system"},
+            ],
+            api_key="fake-key",
+        )
+
+        # system prompt promoted to a cached block on the outbound payload
+        assert captured["optional_params"]["system"] == [
+            {
+                "type": "text",
+                "text": "a long cached system prompt",
+                "cache_control": EPHEMERAL,
+            }
+        ]
+        # injection param popped, not forwarded upstream as an unknown field
+        assert "cache_control_injection_points" not in captured["inner_kwargs"]
+
+    @pytest.mark.asyncio
+    async def test_noop_when_no_injection_points(self, monkeypatch):
+        from litellm.llms.anthropic.experimental_pass_through.messages import handler
+
+        captured = {}
+
+        def fake_handler(*args, **kwargs):
+            captured["optional_params"] = kwargs.get(
+                "anthropic_messages_optional_request_params"
+            )
+            return {"id": "msg_test", "type": "message", "content": []}
+
+        monkeypatch.setattr(
+            handler.base_llm_http_handler,
+            "anthropic_messages_handler",
+            fake_handler,
+        )
+
+        await handler.anthropic_messages(
+            max_tokens=100,
+            messages=[{"role": "user", "content": "hello"}],
+            model="anthropic/claude-sonnet-4-5",
+            system="plain system prompt",
+            custom_llm_provider="anthropic",
+            api_key="fake-key",
+        )
+
+        # unchanged when no injection points configured
+        assert captured["optional_params"]["system"] == "plain system prompt"
+
+    @pytest.mark.asyncio
+    async def test_injection_param_preserved_for_fallback_provider(self, monkeypatch):
+        # A non-native provider (Gemini) falls back to the chat/completions
+        # adapter. The native Anthropic block rewrite must NOT run here, and the
+        # param must survive so the downstream OpenAI-path hook applies caching
+        # in the form Gemini understands. Regression guard for the fallback path.
+        from litellm.llms.anthropic.experimental_pass_through.messages import handler
+
+        captured = {}
+
+        def fake_adapter(*args, **kwargs):
+            captured["kwargs"] = kwargs
+            captured["messages"] = kwargs.get("messages")
+            return {"id": "msg_test", "type": "message", "content": []}
+
+        monkeypatch.setattr(
+            handler.LiteLLMMessagesToCompletionTransformationHandler,
+            "anthropic_messages_handler",
+            fake_adapter,
+        )
+
+        await handler.anthropic_messages(
+            max_tokens=100,
+            messages=[{"role": "user", "content": "hello"}],
+            model="gemini/gemini-2.0-flash",
+            custom_llm_provider="gemini",
+            cache_control_injection_points=[{"location": "message", "role": "user"}],
+            api_key="fake-key",
+        )
+
+        # param left untouched for the fallback adapter (not consumed here)
+        assert captured["kwargs"].get("cache_control_injection_points") == [
+            {"location": "message", "role": "user"}
+        ]
+        # messages not rewritten into Anthropic cached-block form by our hook
+        assert captured["messages"] == [{"role": "user", "content": "hello"}]