diff --git a/litellm/integrations/anthropic_cache_control_hook.py b/litellm/integrations/anthropic_cache_control_hook.py index 296bfb6fc85..705793c4c47 100644 --- a/litellm/integrations/anthropic_cache_control_hook.py +++ b/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.""" diff --git a/litellm/llms/anthropic/experimental_pass_through/messages/handler.py b/litellm/llms/anthropic/experimental_pass_through/messages/handler.py index a3ac465c463..cf2b3abf5fa 100644 --- a/litellm/llms/anthropic/experimental_pass_through/messages/handler.py +++ b/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, diff --git a/litellm/types/integrations/anthropic_cache_control_hook.py b/litellm/types/integrations/anthropic_cache_control_hook.py index 83e5a9e7f01..a3967beb9c4 100644 --- a/litellm/types/integrations/anthropic_cache_control_hook.py +++ b/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, ] diff --git a/tests/test_litellm/integrations/test_anthropic_cache_control_hook_messages.py b/tests/test_litellm/integrations/test_anthropic_cache_control_hook_messages.py new file mode 100644 index 00000000000..d551aa512d6 --- /dev/null +++ b/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"}]