scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-16 04:29:31 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 138

fix(anthropic): count a replayed server tool call's cache mark the way modify_params sends it (#46591)

Browse source 
* fix(anthropic): count a replayed server tool call's cache mark the way modify_params sends it

With modify_params on, a trailing tool search turn whose call has a kept result
got a placeholder answer from the factory and went out as a marked tool_use,
but the cache control hook counted marks before that placeholder existed and
skipped the mark, so a request at the cap reached Anthropic with 5 marks and
failed with a 400. The hook now finds client answers on the sanitized messages,
the same list the factory reads.

Also make the replay checks cheaper (prefix check before validation, early
return without provider_specific_fields, and look up client answers only when
a tool call carries cache_control), drop the two audit cells that cover old
gaps outside the tool search fix, and derive the pass-through test timeout
from the graceful stop budget.

* fix(anthropic): keep the duplicate tool result warning out of the cache control hook

The hook finds client answers with only the step of the modify_params sanitize
pass that pairs tool calls with their results, so a request with a duplicate
tool result logs the warning once again instead of once per mark count.

Also restore the Snowflake non-stream and cached-stream tool search cells as
known bugs with pytest.skip instead of deleting them, and put the Anthropic
pass-through test timeout back to 240 seconds.

* fix(anthropic): count cache marks next to a message the pairing step cannot read

With modify_params on and a marked tool call, the cache control hook paired tool
calls with their results before counting. A message without a get method (a
stray string or null in extra_body.messages, or an openai SDK
ChatCompletionMessage) made that pairing raise AttributeError before any
provider call, where main counted the marks and let the request through.

The hook now pairs only when every message can be read the way the pairing
step reads it, and otherwise counts on the messages as sent, like main did.

* fix(anthropic): count cache marks after placing a later system message the way the request builder does

* test(anthropic): cover a leading system prompt and a system message before the answer in the cache mark census

* test(anthropic): align the tool search replay cells with main and cover a system message after the search turn

The malformed kept result cells now expect the unanswered search call to be dropped, as main does since #45882.
The stream cut cell checks streamed tool call names, since the kept search result names get_weather as a tool
reference. The streamed response cache hit cell runs again because #46104 fixed it. The modify_params cap cell
expects the four client marks, since the placeholder answer sends the search call as a plain tool_use with its
mark, and a new cell covers a system message after the trailing search turn in both stream modes

* fix(anthropic): count marks on the raw messages when they cannot be placed, and find client answers once per request

* test(anthropic): cover marked tool calls next to a malformed tool id and a later system message

* perf(anthropic): look up client tool answers only when a marked tool call is a server tool call

Answered ids only decide whether a marked srvtoolu_ call is rebuilt as server_tool_use, so a request whose marked
tool calls are all client calls no longer places system messages or runs the modify_params pairing step to count
marks

* test(anthropic): reach the client answer lookup with a marked search call in the malformed tool id cell

* perf(anthropic): place mid-conversation system messages in linear time for models that keep them

* fix(anthropic): count cache marks with the model's own system message placement and pair tool results in linear time

* test(anthropic): cover modify_params dropping a tool answer that answers no call of the latest assistant turn

* fix(anthropic): count marks with in-place system placement when the model lookup cannot confirm the flag

* test(anthropic): cover a bare model with custom_llm_provider filling the cache mark budget through litellm.completion and Router

* fix(anthropic): give unmapped models the request builder's converting placement and survive any model lookup failure

* test(anthropic): cover an unmapped model name counting cache marks with the request builder's converting placement

* fix(anthropic): count server tool call cache marks the request builder's way only on routes that build Anthropic Messages requests

Databricks, OpenRouter and Bedrock Converse send every marked tool call as is, so the hook keeps the census main uses there. The seeding step carries custom_llm_provider on the injection points when a server tool call is marked, so a bare model name still resolves its route.

* test(anthropic): cover two answered marked server tool calls before a system message on Databricks, OpenRouter and Bedrock Converse

* fix(anthropic): keep main's cache mark count on GitHub Copilot and ChatGPT routes without resolving them, so a sync call with no stored login never starts the device flow

* test(anthropic): cover a marked server tool call reaching the GitHub Copilot and ChatGPT login hosts like an unmarked one
This commit is contained in:
Mateo Wang 2026-10-15 14:46:43 -07:00 • committed by GitHub
parent b3c9d4c1f7
commit 62af39aaf8
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
12 changed files with 1292 additions and 165 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

134
litellm/integrations/anthropic_cache_control_hook.py
View file

@ -10,6 +10,7 @@ Supported for both `v1/chat/completions` (via the prompt-management hook) and
"""
import copy
import itertools
import os
import re
from collections.abc import Callable, Container, Iterable, Mapping, Sequence
@ -26,7 +27,10 @@ from litellm.litellm_core_utils.prompt_templates.common_utils import (
cache_control_as_prompt_cache_breakpoint,
with_prompt_cache_breakpoint,
)
from litellm.litellm_core_utils.prompt_templates.factory import tool_call_ids_answered_by_client
from litellm.litellm_core_utils.prompt_templates.factory import (
tool_call_ids_answered_by_client,
tool_call_ids_answered_in,
)
from litellm.llms.anthropic.common_utils import (
is_claude_code_one_shot_subagent_request,
supports_anthropic_cache_control,
@ -39,6 +43,7 @@ from litellm.types.integrations.anthropic_cache_control_hook import (
CacheControlMessageInjectionPoint,
)
from litellm.types.llms.anthropic import (
ANTHROPIC_SERVER_TOOL_USE_ID_PREFIX,
ANTHROPIC_TOOL_SEARCH_TOOL_TYPES,
AllAnthropicToolsValues,
AnthropicSystemMessageContent,
@ -238,6 +243,77 @@ def _tool_call_carries_cache_breakpoint(tool_call: object, message: object, answ
)
def _tool_call_ids_answered_by_client_when_a_server_tool_call_is_marked(
messages: Sequence[object], model: str | None = None, custom_llm_provider: str | None = None
) -> frozenset[str]:
if not _marks_a_server_tool_call(messages):
return frozenset()
if model is not None and not _builds_anthropic_messages(model, custom_llm_provider):
return tool_call_ids_answered_in(messages)
return tool_call_ids_answered_by_client(messages, _keeps_mid_conversation_system(model, custom_llm_provider))
def _marks_a_server_tool_call(messages: Sequence[object]) -> bool:
tool_calls: Final = itertools.chain.from_iterable(
_as_object_list(_attribute_or_key(message, "tool_calls")) or () for message in messages
)
return any(_is_marked_server_tool_call(tool_call) for tool_call in tool_calls)
def _builds_anthropic_messages(model: str, custom_llm_provider: str | None) -> bool:
from litellm.exceptions import BadRequestError
from litellm.litellm_core_utils.get_llm_provider_logic import declared_authenticating_provider, get_llm_provider
from litellm.llms.anthropic.chat.transformation import AnthropicConfig
from litellm.llms.openai.chat.gpt_transformation import OpenAIGPTConfig
from litellm.types.utils import LlmProviders
from litellm.utils import ProviderConfigManager
if declared_authenticating_provider(model, custom_llm_provider) is not None:
return False
try:
served_model, provider, _, _ = get_llm_provider(model=model, custom_llm_provider=custom_llm_provider)
chat_config: Final = ProviderConfigManager.get_provider_chat_config(
model=served_model, provider=LlmProviders(provider)
)
except (BadRequestError, ValueError):
return True
return isinstance(chat_config, AnthropicConfig) and not isinstance(chat_config, OpenAIGPTConfig)
def _custom_llm_provider(non_default_params: Mapping[str, object], point: Mapping[str, object]) -> str | None:
provider: Final = non_default_params.get("custom_llm_provider") or point.get(CUSTOM_LLM_PROVIDER_STAMP)
return provider if isinstance(provider, str) else None
def _keeps_mid_conversation_system(model: str | None, custom_llm_provider: str | None) -> bool:
if model is None:
return True
from litellm.utils import supports_mid_conversation_system
if supports_mid_conversation_system(model=model, custom_llm_provider=custom_llm_provider):
return True
return _model_info_flags_mid_conversation_system(model, custom_llm_provider)
def _model_info_flags_mid_conversation_system(model: str, custom_llm_provider: str | None) -> bool:
from litellm.utils import get_model_info
try:
model_info: Final = get_model_info(model=model, custom_llm_provider=custom_llm_provider)
except Exception:
return False
return model_info.get("supports_mid_conversation_system") is True
def _is_marked_server_tool_call(tool_call: object) -> bool:
tool_call_id: Final = _attribute_or_key(tool_call, "id")
return (
isinstance(tool_call_id, str)
and tool_call_id.startswith(ANTHROPIC_SERVER_TOOL_USE_ID_PREFIX)
and _attribute_or_key(tool_call, "cache_control") is not None
)
def _tool_carries_cache_breakpoint(tool: object) -> bool:
return _carries_cache_breakpoint(tool) or (
isinstance(tool, dict) and _carries_cache_breakpoint(tool.get("function"))
@ -263,6 +339,7 @@ def _may_carry_prompt_cache_breakpoint(message: Mapping[str, object]) -> bool:
CARRY_UNMATCHED_MESSAGE_POINTS: Final = "_litellm_carry_unmatched_cache_control_points"
EXTERNAL_BREAKPOINTS_STAMP: Final = "_litellm_external_breakpoints"
CUSTOM_LLM_PROVIDER_STAMP: Final = "_litellm_custom_llm_provider"
class AnthropicCacheControlHook(CustomPromptManagement):
@ -336,13 +413,14 @@ class AnthropicCacheControlHook(CustomPromptManagement):
)
remaining_points: Final = tuple(point for point in injection_points if point.get("location") != "message")
custom_llm_provider: Final = _custom_llm_provider(non_default_params, injection_points[0])
stamped_dialect: Final = injection_points[0].get("_litellm_openai_dialect")
openai_dialect: Final = (
stamped_dialect
if isinstance(stamped_dialect, bool)
else AnthropicCacheControlHook.targets_openai_prompt_cache_breakpoint(
model,
non_default_params.get("custom_llm_provider"),
custom_llm_provider,
non_default_params.get("api_base") or non_default_params.get("base_url"),
non_default_params.get("prompt_cache_options"),
)
@ -365,16 +443,25 @@ class AnthropicCacheControlHook(CustomPromptManagement):
reserved_blocks: Final = AnthropicCacheControlHook._blocks_reserved_outside_messages(
remaining_points, external_breakpoints, openai_dialect
)
breakpoints_before: Final = AnthropicCacheControlHook.count_request_cache_breakpoints(processed_messages)
answered_tool_ids: Final = _tool_call_ids_answered_by_client_when_a_server_tool_call_is_marked(
processed_messages, model, custom_llm_provider
)
breakpoints_before: Final = AnthropicCacheControlHook.count_request_cache_breakpoints(
processed_messages, answered_tool_ids=answered_tool_ids
)
processed_messages = self._apply_message_injections(
points=applied_message_points,
messages=processed_messages,
max_blocks=MAX_CACHE_CONTROL_BLOCKS - reserved_blocks,
openai_dialect=openai_dialect,
answered_tool_ids=answered_tool_ids,
)
if (
openai_dialect
and AnthropicCacheControlHook.count_request_cache_breakpoints(processed_messages) > breakpoints_before
and AnthropicCacheControlHook.count_request_cache_breakpoints(
processed_messages, answered_tool_ids=answered_tool_ids
)
> breakpoints_before
and non_default_params.get("prompt_cache_options") is None
):
non_default_params["prompt_cache_options"] = PromptCacheOptions(mode="implicit")
@ -387,7 +474,10 @@ class AnthropicCacheControlHook(CustomPromptManagement):
carried_points: Final[Sequence[CacheControlInjectionPoint]] = (
*AnthropicCacheControlHook._points_with_a_slot_left(
remaining_points,
AnthropicCacheControlHook.count_request_cache_breakpoints(processed_messages) + external_breakpoints,
AnthropicCacheControlHook.count_request_cache_breakpoints(
processed_messages, answered_tool_ids=answered_tool_ids
)
+ external_breakpoints,
openai_dialect,
),
*carried_message_points,
@ -422,14 +512,20 @@ class AnthropicCacheControlHook(CustomPromptManagement):
return _served_provider_for_model(model)
@staticmethod
def count_request_cache_breakpoints(messages: Iterable[object], system: object = None) -> int:
def count_request_cache_breakpoints(
messages: Iterable[object], system: object = None, answered_tool_ids: frozenset[str] | None = None
) -> int:
system_blocks: Final = (
sum(1 for block in system if _carries_cache_breakpoint(block)) if isinstance(system, list) else 0
)
message_list: Final = tuple(messages)
answered_tool_ids: Final = tool_call_ids_answered_by_client(message_list)
client_answers: Final = (
_tool_call_ids_answered_by_client_when_a_server_tool_call_is_marked(message_list)
if answered_tool_ids is None
else answered_tool_ids
)
return system_blocks + sum(
AnthropicCacheControlHook._count_cache_control_blocks(msg, answered_tool_ids) for msg in message_list
AnthropicCacheControlHook._count_cache_control_blocks(msg, client_answers) for msg in message_list
)
@staticmethod
@ -504,6 +600,7 @@ class AnthropicCacheControlHook(CustomPromptManagement):
messages: list[AllMessageValues],
max_blocks: int,
openai_dialect: bool = False,
answered_tool_ids: frozenset[str] | None = None,
) -> list[AllMessageValues]:
"""Apply message-level cache control injection points in order.
@ -514,8 +611,14 @@ class AnthropicCacheControlHook(CustomPromptManagement):
``max_blocks`` is reached. Injection points are honored in config order,
so earlier points win when slots are scarce.
"""
used_blocks = AnthropicCacheControlHook.count_request_cache_breakpoints(messages)
answered_tool_ids: Final = tool_call_ids_answered_by_client(messages)
client_answers: Final = (
_tool_call_ids_answered_by_client_when_a_server_tool_call_is_marked(messages)
if answered_tool_ids is None
else answered_tool_ids
)
used_blocks = AnthropicCacheControlHook.count_request_cache_breakpoints(
messages, answered_tool_ids=client_answers
)
limit_reached = False
for point in points:
@ -532,14 +635,14 @@ class AnthropicCacheControlHook(CustomPromptManagement):
limit_reached = True
break
if AnthropicCacheControlHook._message_has_cache_control(messages[target_index], answered_tool_ids):
if AnthropicCacheControlHook._message_has_cache_control(messages[target_index], client_answers):
# Client already marked this message; don't overwrite it.
continue
messages[target_index] = AnthropicCacheControlHook._safe_insert_cache_control_in_message(
messages[target_index], control, openai_dialect
)
if AnthropicCacheControlHook._message_has_cache_control(messages[target_index], answered_tool_ids):
if AnthropicCacheControlHook._message_has_cache_control(messages[target_index], client_answers):
used_blocks += 1
if limit_reached:
@ -1008,7 +1111,7 @@ class AnthropicCacheControlHook(CustomPromptManagement):
tools_keeping_marks: Final = tuple(
tool for tool in tools or () if not _chat_transform_drops_tool_cache_control(tool)
)
non_default_params["cache_control_injection_points"] = AnthropicCacheControlHook._stamped_for_prompt_hook(
stamped: Final = AnthropicCacheControlHook._stamped_for_prompt_hook(
configured,
AnthropicCacheControlHook.count_external_cache_breakpoints(
tools_keeping_marks, non_default_params.get("cache_control"), non_default_params
@ -1018,6 +1121,11 @@ class AnthropicCacheControlHook(CustomPromptManagement):
api_base,
non_default_params.get("prompt_cache_options"),
)
non_default_params["cache_control_injection_points"] = (
AnthropicCacheControlHook._stamped(stamped, CUSTOM_LLM_PROVIDER_STAMP, custom_llm_provider)
if custom_llm_provider is not None and _marks_a_server_tool_call(messages)
else stamped
)
return
points: Final = AnthropicCacheControlHook.get_default_injection_points(
messages=messages,

148
litellm/litellm_core_utils/prompt_templates/factory.py
View file

@ -26,6 +26,8 @@ from litellm.litellm_core_utils.prompt_templates.mid_conversation_system import
is_anthropic_tool_search_use,
message_field,
parts_of,
place_mid_conversation_system,
split_leading_system_run,
)
from litellm.litellm_core_utils.url_utils import async_safe_get, safe_get
from litellm.llms.custom_httpx.http_handler import HTTPHandler, get_async_httpx_client, header_value
@ -1883,6 +1885,8 @@ def _tool_call_ids_with_a_kept_server_result(message: object) -> frozenset[str]:
from litellm.llms.anthropic.common_utils import tool_call_is_rebuilt_as_server_tool_use
provider_specific_fields: Final = message_field(message, "provider_specific_fields")
if provider_specific_fields is None:
return frozenset()
return frozenset(
tool_call_id
for tool_call in parts_of(message_field(message, "tool_calls"))
@ -1914,10 +1918,38 @@ def server_tool_answers_replaced_by_their_result(messages: Sequence[object]) ->
)
def tool_call_ids_answered_by_client(messages: Sequence[object]) -> frozenset[str]:
def tool_call_ids_answered_by_client(
messages: Sequence[object], supports_mid_conversation_system: bool = True
) -> frozenset[str]:
return tool_call_ids_answered_in(_as_the_request_builder_sends_them(messages, supports_mid_conversation_system))
def tool_call_ids_answered_in(messages: Sequence[object]) -> frozenset[str]:
return _tool_call_ids_in_tool_messages(messages) - server_tool_answers_replaced_by_their_result(messages)
def _as_the_request_builder_sends_them(
messages: Sequence[object], supports_mid_conversation_system: bool
) -> Sequence[object]:
if not all(callable(getattr(message, "get", None)) for message in messages):
return messages
chat_messages: Final = cast(Sequence[AllMessageValues], messages) # cast-ok: placement reads messages via get
try:
return _placed_and_paired(chat_messages, supports_mid_conversation_system)
except TypeError:
return messages
def _placed_and_paired(
messages: Sequence[AllMessageValues], supports_mid_conversation_system: bool
) -> Sequence[AllMessageValues]:
_, conversation = split_leading_system_run(messages)
placed: Final = place_mid_conversation_system(
conversation, supports_mid_conversation_system=supports_mid_conversation_system
)
return _pair_tool_calls_with_results(placed) if litellm.modify_params else placed
_AnthropicToolInvokeItem: TypeAlias = AnthropicMessagesToolUseParam | dict[str, Any]
@ -2225,7 +2257,7 @@ def _sanitize_empty_text_content(
def _add_missing_tool_results(
current_message: AllMessageValues,
messages: list[AllMessageValues],
messages: Sequence[AllMessageValues],
current_index: int,
) -> tuple[list[AllMessageValues], int]:
"""
@ -2328,7 +2360,7 @@ def _add_missing_tool_results(
def _is_orphaned_tool_result(
current_message: AllMessageValues,
sanitized_messages: list[AllMessageValues],
latest_assistant_tool_call_ids: frozenset[object],
) -> bool:
"""
Case B: Orphaned tool_result (unexpected result)
@ -2346,32 +2378,26 @@ def _is_orphaned_tool_result(
if not tool_call_id:
return False
# Look back to find the most recent assistant message with tool_calls
found_matching_tool_call = False
if latest_assistant_tool_call_ids and tool_call_id in latest_assistant_tool_call_ids:
return False
for j in range(len(sanitized_messages) - 1, -1, -1):
prev_msg = sanitized_messages[j]
if prev_msg.get("role") == "assistant":
tool_calls = prev_msg.get("tool_calls")
if tool_calls:
for tool_call in cast(list, tool_calls):
tc_id = None
if isinstance(tool_call, dict):
tc_id = tool_call.get("id")
else:
tc_id = getattr(tool_call, "id", None)
verbose_logger.debug("_is_orphaned_tool_result: Found orphaned tool result with redacted tool_call_id")
return True
if tc_id == tool_call_id:
found_matching_tool_call = True
break
break
def _tool_call_id(tool_call: object) -> object:
return tool_call.get("id") if isinstance(tool_call, dict) else getattr(tool_call, "id", None)
if not found_matching_tool_call:
verbose_logger.debug("_is_orphaned_tool_result: Found orphaned tool result with redacted tool_call_id")
return True
return False
def _tool_call_ids_of_the_latest_assistant(
latest_ids: frozenset[object], message: AllMessageValues
) -> frozenset[object]:
if message.get("role") != "assistant":
return latest_ids
tool_calls: Final = message.get("tool_calls") or ()
return frozenset(
tool_call_id for tool_call_id in map(_tool_call_id, cast(Iterable[object], tool_calls)) if tool_call_id
)
def _declared_tool_call_ids(message: Mapping[str, object]) -> frozenset[str]:
@ -2444,35 +2470,7 @@ def sanitize_messages_for_tool_calling(
"""
if not litellm.modify_params:
return messages
sanitized_messages: list[AllMessageValues] = []
i = 0
while i < len(messages):
current_message = messages[i]
# Case C: Sanitize empty text content
current_message = _sanitize_empty_text_content(current_message)
# Case A: Check if assistant message has tool_calls without following tool results
if current_message.get("role") == "assistant":
result_messages, messages_consumed = _add_missing_tool_results(current_message, messages, i)
# If dummy tool results were added, extend sanitized_messages and skip consumed messages
if len(result_messages) > 1:
sanitized_messages.extend(result_messages)
# Skip the assistant message and any actual tool results that were included
i += 1 + messages_consumed
continue
# Case B: Check for orphaned tool results
if _is_orphaned_tool_result(current_message, sanitized_messages):
i += 1
continue # Skip this orphaned tool result
# Add the message to sanitized list
sanitized_messages.append(current_message)
i += 1
sanitized_messages: Final = _pair_tool_calls_with_results(messages)
# Case D: Deduplicate tool results with the same tool_call_id.
# Anthropic requires exactly one tool_result per tool_use. Session history
@ -2510,10 +2508,43 @@ def sanitize_messages_for_tool_calling(
# a turn boundary and would mask real within-block duplicates.
seen_in_block = {}
if duplicates_to_remove:
sanitized_messages = [msg for idx, msg in enumerate(sanitized_messages) if idx not in duplicates_to_remove]
return [msg for idx, msg in enumerate(sanitized_messages) if idx not in duplicates_to_remove]
return sanitized_messages
def _pair_tool_calls_with_results(messages: Sequence[AllMessageValues]) -> tuple[AllMessageValues, ...]:
latest_assistant_tool_call_ids: Final = tuple(
itertools.accumulate(messages, _tool_call_ids_of_the_latest_assistant, initial=frozenset[object]())
)
sanitized_messages: list[AllMessageValues] = []
i = 0
while i < len(messages):
current_message = messages[i]
# Case C: Sanitize empty text content
current_message = _sanitize_empty_text_content(current_message)
# Case A: Check if assistant message has tool_calls without following tool results
if current_message.get("role") == "assistant":
result_messages, messages_consumed = _add_missing_tool_results(current_message, messages, i)
# If dummy tool results were added, extend sanitized_messages and skip consumed messages
if len(result_messages) > 1:
sanitized_messages.extend(result_messages)
# Skip the assistant message and any actual tool results that were included
i += 1 + messages_consumed
continue
# Case B: Check for orphaned tool results
if _is_orphaned_tool_result(current_message, latest_assistant_tool_call_ids[i]):
i += 1
continue # Skip this orphaned tool result
# Add the message to sanitized list
sanitized_messages.append(current_message)
i += 1
return tuple(sanitized_messages)
def _replayable_thinking_blocks(
@ -3499,11 +3530,6 @@ def gemini_text_image_pt(messages: list):
]
}
"""
try:
pass
except Exception:
raise Exception("Importing google.generativeai failed, please run 'pip install -q google-generativeai")
prompt = ""
images: Final = []
for message in messages:

77
litellm/litellm_core_utils/prompt_templates/mid_conversation_system.py
View file

@ -30,7 +30,8 @@ Anthropic wire shape is built later by ``anthropic_messages_pt``.
"""
from collections.abc import Iterable, Iterator, Mapping, Sequence
from itertools import chain, dropwhile, groupby, takewhile
from itertools import accumulate, chain, dropwhile, groupby, takewhile
from operator import itemgetter
from types import MappingProxyType
from typing import Final, Literal, TypeAlias, TypeVar
@ -259,8 +260,9 @@ def _system_runs(messages: Sequence[AllMessageValues]) -> tuple[tuple[int, ...],
)
def _block_containing(message_index: int, blocks: Sequence[tuple[bool, tuple[int, ...]]]) -> int:
return next(index for index, (_, indices) in enumerate(blocks) if message_index in indices)
def _block_index_by_message(blocks: Sequence[tuple[bool, tuple[int, ...]]]) -> Mapping[int, int]:
pairs: Final = (((index, block_index) for index in indices) for block_index, (_, indices) in enumerate(blocks))
return MappingProxyType(dict(chain.from_iterable(pairs)))
def _thinking_block_renders(block: object) -> bool:
@ -433,37 +435,30 @@ def _rendered_indices(messages: Sequence[object]) -> frozenset[int]:
return frozenset(index for index, message in enumerate(messages) if _renders(message, answered.get(index, _NO_IDS)))
def _rendered_block(
message_index: int,
rendered: frozenset[int],
blocks: Sequence[tuple[bool, tuple[int, ...]]],
) -> int | None:
block_index: Final = _block_containing(message_index, blocks)
_, indices = blocks[block_index]
return block_index if not rendered.isdisjoint(indices) else None
def _anchorable_blocks(rendered: frozenset[int], blocks: Sequence[tuple[bool, tuple[int, ...]]]) -> tuple[bool, ...]:
"""Whether a system message behind each block lands on the wire after a user turn and before an assistant turn.
def _system_may_follow(
block_index: int,
rendered: frozenset[int],
blocks: Sequence[tuple[bool, tuple[int, ...]]],
) -> bool:
"""Whether a system message behind this block precedes an assistant turn or ends the array on the wire.
Blocks alternate between user-type and assistant, so the check is whether the
first later block that puts anything on the wire is an assistant block.
The block has to put something on the wire. Blocks alternate between user-type
and assistant, so the first later block that puts anything on the wire has to be
an assistant block, or there is none and the system message ends the array.
"""
return next(
(not is_user for is_user, indices in blocks[block_index + 1 :] if not rendered.isdisjoint(indices)),
True,
def may_follow_from(may_follow_later: bool, block: tuple[bool, tuple[int, ...]]) -> bool:
is_user, indices = block
return may_follow_later if rendered.isdisjoint(indices) else not is_user
may_follow_from_block: Final = tuple(reversed(tuple(accumulate(reversed(blocks), may_follow_from, initial=True))))
return tuple(
not rendered.isdisjoint(indices) and may_follow_from_block[block_index + 1]
for block_index, (_, indices) in enumerate(blocks)
)
def _anchor_block(
run: Sequence[int],
messages: Sequence[AllMessageValues],
rendered: frozenset[int],
blocks: Sequence[tuple[bool, tuple[int, ...]]],
block_index_by_message: Mapping[int, int],
anchorable: Sequence[bool],
) -> int | None:
"""The user-type block a system run must follow, or ``None`` when it converts in place.
@ -480,10 +475,17 @@ def _anchor_block(
neighbour: Final = previous if _is_user_type(messages[previous]) else run[-1] + 1
if neighbour >= len(messages) or not _is_user_type(messages[neighbour]):
return None
block_index: Final = _rendered_block(neighbour, rendered, blocks)
if block_index is None or not _system_may_follow(block_index, rendered, blocks):
return None
return block_index
block_index: Final = block_index_by_message[neighbour]
return block_index if anchorable[block_index] else None
def _runs_by_anchor(
anchors: Sequence[tuple[tuple[int, ...], int | None]],
) -> Mapping[int, tuple[tuple[int, ...], ...]]:
anchored: Final = tuple(sorted(((anchor, run) for run, anchor in anchors if anchor is not None), key=itemgetter(0)))
return MappingProxyType(
{anchor: tuple(run for _, run in group) for anchor, group in groupby(anchored, key=itemgetter(0))}
)
def _placed_for_flagged_model(messages: Sequence[AllMessageValues]) -> tuple[AllMessageValues, ...]:
@ -496,19 +498,22 @@ def _placed_for_flagged_model(messages: Sequence[AllMessageValues]) -> tuple[All
move it. Runs that share a user turn merge into one system message.
"""
blocks: Final = _user_type_blocks(messages)
rendered: Final = _rendered_indices(messages)
anchors: Final = tuple((run, _anchor_block(run, messages, rendered, blocks)) for run in _system_runs(messages))
block_index_by_message: Final = _block_index_by_message(blocks)
anchorable: Final = _anchorable_blocks(_rendered_indices(messages), blocks)
anchors: Final = tuple(
(run, _anchor_block(run, messages, block_index_by_message, anchorable)) for run in _system_runs(messages)
)
runs_by_anchor: Final = _runs_by_anchor(anchors)
converted_run_after: Final = MappingProxyType({run[0] - 1: run for run, anchor in anchors if anchor is None})
def messages_of(run: tuple[int, ...]) -> tuple[AllMessageValues, ...]:
return tuple(messages[index] for index in run)
def anchored_to(block_index: int) -> tuple[AllMessageValues, ...]:
anchored_runs: Final = tuple(run for run, anchor in anchors if anchor == block_index)
return tuple(chain.from_iterable(map(messages_of, anchored_runs)))
return tuple(chain.from_iterable(map(messages_of, runs_by_anchor.get(block_index, ()))))
def converted_after(message_index: int) -> tuple[ChatCompletionUserMessage, ...]:
following_runs: Final = tuple(run for run, anchor in anchors if anchor is None and run[0] == message_index + 1)
return tuple(chain.from_iterable(_converted_user_turns(messages_of(run)) for run in following_runs))
return _converted_user_turns(messages_of(converted_run_after.get(message_index, ())))
def emit(block_index: int) -> Iterator[AllMessageValues]:
is_user, indices = blocks[block_index]

10
litellm/llms/anthropic/common_utils.py
View file

@ -48,6 +48,7 @@ from litellm.types.llms.anthropic import (
ANTHROPIC_MID_CONVERSATION_TOOL_CHANGES_BETA_HEADER,
ANTHROPIC_OAUTH_BETA_HEADER,
ANTHROPIC_OAUTH_TOKEN_PREFIX,
ANTHROPIC_SERVER_TOOL_USE_ID_PREFIX,
ANTHROPIC_THINKING_DISPLAY_UPDATES_BETA_HEADER,
AllAnthropicToolsValues,
AnthropicMcpServerTool,
@ -1932,8 +1933,15 @@ def _replayed_server_tool_use(block: object) -> _ReplayedServerToolUse | None:
def tool_call_is_rebuilt_as_server_tool_use(
tool_call_id: object, provider_specific_fields: object, answered_tool_ids: Container[str] = frozenset()
) -> bool:
if (
not isinstance(tool_call_id, str)
or not tool_call_id.startswith(ANTHROPIC_SERVER_TOOL_USE_ID_PREFIX)
or tool_call_id in answered_tool_ids
or provider_specific_fields is None
):
return False
fields: Final = validated_object_mapping(provider_specific_fields)
if not isinstance(tool_call_id, str) or tool_call_id in answered_tool_ids or fields is None:
if fields is None:
return False
return (
find_anthropic_server_tool_result(

2
litellm/types/integrations/anthropic_cache_control_hook.py
View file

@ -19,6 +19,7 @@ class CacheControlMessageInjectionPoint(TypedDict):
control: ChatCompletionCachedContent | None
_litellm_openai_dialect: NotRequired[ReadOnly[bool]]
_litellm_external_breakpoints: NotRequired[ReadOnly[int]]
_litellm_custom_llm_provider: NotRequired[ReadOnly[str]]
class CacheControlToolConfigInjectionPoint(TypedDict):
@ -28,6 +29,7 @@ class CacheControlToolConfigInjectionPoint(TypedDict):
control: ChatCompletionCachedContent | None
_litellm_openai_dialect: NotRequired[ReadOnly[bool]]
_litellm_external_breakpoints: NotRequired[ReadOnly[int]]
_litellm_custom_llm_provider: NotRequired[ReadOnly[str]]
CacheControlInjectionPoint = CacheControlMessageInjectionPoint | CacheControlToolConfigInjectionPoint

15
tests/integration/providers/test_anthropic_tool_search_result_matrix_wire.py
View file

@ -7,7 +7,7 @@ import httpx
import litellm
import openai
import pytest
from integration._support.anthropic_thinking import JSON_LIST, JSON_OBJECT
from integration._support.anthropic_thinking import JSON_LIST, JSON_OBJECT, chunks_of, deltas_of
from integration._support.client import Gateway
from integration._support.wire import Reply, Request, wire_server
from integration.providers._cache_control_marks_support import (
@ -146,6 +146,8 @@ def test_tool_search_result_round_trips_on_every_anthropic_wire_provider(
@_STREAM_MODES
def test_snowflake_claude_keeps_the_tool_search_result_on_the_response(gateway: Gateway, stream: bool) -> None:
if not stream:
pytest.skip("BUG: a non-streamed Snowflake Claude response drops the tool_search_tool_result block")
marker: Final = uuid.uuid4().hex
with wire_server(scripted_peer()) as wire, gateway.scenario() as scenario:
model: Final = deployment(scenario, "snowflake", wire.url, gateway.upstream_url)
@ -375,13 +377,13 @@ def test_search_call_without_a_kept_result_replays_plainly_with_both_answers(
pytest.param({"tool_results": [search_result("srvtoolu_01Other")]}, id="result-for-another-call"),
),
)
def test_malformed_kept_results_replay_the_search_call_plainly(gateway: Gateway, fields: JsonValue) -> None:
def test_malformed_kept_results_drop_the_unanswered_search_call(gateway: Gateway, fields: JsonValue) -> None:
marker: Final = uuid.uuid4().hex
assistant: Final = hand_built_assistant((SEARCH_CALL, WEATHER_CALL), fields)
response, received = _replayed(gateway, replay_messages(marker, assistant))
assert response.status_code == 200, response.text
assert len(received) == 1, received
assert wire_assistant(received[0]) == [_PLAIN_SEARCH_USE, WEATHER_USE], received[0].body
assert wire_assistant(received[0]) == [WEATHER_USE], received[0].body
assert answered_ids(received[0]) == [WEATHER_ID], received[0].body
@ -531,6 +533,11 @@ def _cut_stream_text(gateway: Gateway, body: Event) -> str:
return repr(error)
def _streamed_tool_names(text: str) -> list[JsonValue]:
calls: Final = JSON_LIST.validate_python(folded_assistant(deltas_of(chunks_of(text)))["tool_calls"])
return [JSON_OBJECT.validate_python(JSON_OBJECT.validate_python(call)["function"])["name"] for call in calls]
def test_stream_cut_right_after_the_search_result_leaves_the_proxy_serving(gateway: Gateway) -> None:
marker: Final = uuid.uuid4().hex
@ -542,7 +549,7 @@ def test_stream_cut_right_after_the_search_result_leaves_the_proxy_serving(gatew
with wire_server(respond) as wire, gateway.scenario() as scenario:
model: Final = deployment(scenario, "anthropic", wire.url, gateway.upstream_url)
text: Final = _cut_stream_text(gateway, chat_body(model, first_messages(marker), stream=True))
assert "get_weather" not in text, text
assert "get_weather" not in _streamed_tool_names(text), text
assert gateway.client.get("/health/liveliness").status_code == 200
follow_up: Final = uuid.uuid4().hex
recovered: Final = chat(gateway, chat_body(model, first_messages(follow_up), stream=True))

388
tests/integration/providers/test_anthropic_tool_search_result_modify_params_wire.py
View file

@ -1,37 +1,49 @@
import json
import uuid
from collections.abc import Iterator
from itertools import chain
from pathlib import Path
from typing import Final
import pytest
import yaml
from integration._support.client import Gateway, gateway_from_environment, object_value
from integration._support.openai_wire import answering_model_discovery, chat_reply
from integration._support.process import graceful_stop_seconds, owned_proxy
from integration._support.wire import wire_server
from integration._support.wire import Reply, Request, wire_server
from integration.providers._cache_control_marks_support import (
ANTHROPIC_CACHE_CONTROL_CAP,
ASK,
ASK_LABEL,
EPHEMERAL,
POINTS,
PROVIDER_KEY,
SYSTEM,
SYSTEM_LABEL,
TOOL,
anthropic_labels,
anthropic_peer,
ask,
bedrock_labels,
bedrock_peer,
call_id,
final_text,
marker_of,
post_chat,
tool_call,
tool_use_label,
)
from integration.providers._tool_search_support import (
KEPT_FIELDS,
MODEL,
SEARCH_CALL,
SEARCH_ID,
SEARCH_RESULT,
SEARCH_TURN,
SEARCH_USE,
SEARCH_RESULT,
WEATHER_CALL,
WEATHER_ID,
Event,
answer,
answer_text,
answered_ids,
@ -45,12 +57,13 @@ from integration.providers._tool_search_support import (
scripted_peer,
wire_assistant,
)
from pydantic import JsonValue
from pydantic import JsonValue, TypeAdapter
pytestmark = pytest.mark.timeout(2 * graceful_stop_seconds() + 120)
_LANE_CONFIG: Final = Path(__file__).resolve().parents[1] / "proxy_config.yaml"
_CITIES: Final = ("London", "Paris")
_SYSTEM_NOTE: Final = "Answer in one sentence."
def _modify_params_config(directory: Path) -> Path:
@ -107,25 +120,26 @@ def test_modify_params_placeholder_for_the_kept_search_is_dropped_when_no_answer
assert SEARCH_ID not in answered_ids(received[0]), received[0].body
def _marked_search_call(search_id: str) -> dict[str, JsonValue]:
return {
"id": search_id,
"type": "function",
"function": {"name": "tool_search_tool_bm25", "arguments": '{"query": "weather"}'},
"cache_control": EPHEMERAL,
}
def _marked_search_turn() -> dict[str, JsonValue]:
return {
"role": "assistant",
"content": "",
"tool_calls": [
{
"id": SEARCH_ID,
"type": "function",
"function": {"name": "tool_search_tool_bm25", "arguments": '{"query": "weather"}'},
"cache_control": EPHEMERAL,
}
],
"tool_calls": [_marked_search_call(SEARCH_ID)],
"provider_specific_fields": KEPT_FIELDS,
}
def test_modify_params_trailing_search_turn_stays_within_the_cache_mark_cap(modify_params_gateway: Gateway) -> None:
marker: Final = uuid.uuid4().hex
messages: Final[list[JsonValue]] = [
def _trailing_search_conversation(marker: str) -> list[JsonValue]:
return [
{"role": "system", "content": SYSTEM},
ask(marked=True),
{
@ -137,6 +151,10 @@ def test_modify_params_trailing_search_turn_stays_within_the_cache_mark_cap(modi
{"role": "user", "content": final_text(marker)},
_marked_search_turn(),
]
def test_modify_params_trailing_search_turn_stays_within_the_cache_mark_cap(modify_params_gateway: Gateway) -> None:
messages: Final = _trailing_search_conversation(uuid.uuid4().hex)
with wire_server(anthropic_peer) as wire, modify_params_gateway.scenario() as scenario:
model: Final = deployment(scenario, "anthropic", wire.url, modify_params_gateway.upstream_url)
status, _, text = post_chat(
@ -146,5 +164,343 @@ def test_modify_params_trailing_search_turn_stays_within_the_cache_mark_cap(modi
assert status == 200, text
assert len(received) == 1, received
labels: Final = anthropic_labels(received[0])
assert len(labels) <= 4, labels
assert labels[:4] == [SYSTEM_LABEL, ASK_LABEL, *(tool_use_label(city) for city in _CITIES)], labels
assert labels == [ASK_LABEL, *(tool_use_label(city) for city in _CITIES), f"assistant:tool_use:{SEARCH_ID}"], labels
assert SEARCH_ID in answered_ids(received[0]), received[0].body
@pytest.mark.parametrize("stream", [False, True], ids=["non-stream", "stream"])
def test_modify_params_trailing_search_turn_before_a_system_message_keeps_the_system_mark_within_the_cap(
modify_params_gateway: Gateway, stream: bool
) -> None:
messages: Final = [*_trailing_search_conversation(uuid.uuid4().hex), {"role": "system", "content": _SYSTEM_NOTE}]
with wire_server(anthropic_peer) as wire, modify_params_gateway.scenario() as scenario:
model: Final = deployment(scenario, "anthropic", wire.url, modify_params_gateway.upstream_url)
response: Final = chat(
modify_params_gateway, chat_body(model, messages, stream, cache_control_injection_points=POINTS)
)
received: Final = wire.drain()
assert response.status_code == 200, response.text
assert answer_text(response, stream) == "sunny", response.text
assert len(received) == 1, received
labels: Final = anthropic_labels(received[0])
assert labels == [SYSTEM_LABEL, ASK_LABEL, *(tool_use_label(city) for city in _CITIES)], labels
assert wire_assistant(received[0], 3)[-2:] == [SEARCH_USE, SEARCH_RESULT], received[0].body
assert SEARCH_ID not in answered_ids(received[0]), received[0].body
_SECOND_SEARCH_ID: Final = "srvtoolu_02ScriptedSearch"
@pytest.mark.parametrize(
("model", "expected_labels"),
[
pytest.param(
f"anthropic/{MODEL}",
[SYSTEM_LABEL, ASK_LABEL, f"assistant:tool_use:{SEARCH_ID}", f"assistant:tool_use:{_SECOND_SEARCH_ID}"],
id="keeps-mid-conversation-system",
),
pytest.param(
"anthropic/claude-sonnet-4-6",
[SYSTEM_LABEL, ASK_LABEL, f"user:text:{_SYSTEM_NOTE}"],
id="converts-mid-conversation-system",
),
],
)
def test_modify_params_client_answer_to_one_of_two_marked_search_calls_fills_the_cache_mark_budget(
modify_params_gateway: Gateway, model: str, expected_labels: list[str]
) -> None:
second_result: Final[Event] = {**SEARCH_RESULT, "tool_use_id": _SECOND_SEARCH_ID}
messages: Final[list[JsonValue]] = [
{"role": "system", "content": SYSTEM},
ask(marked=True),
{
"role": "assistant",
"content": "",
"tool_calls": [_marked_search_call(SEARCH_ID), _marked_search_call(_SECOND_SEARCH_ID)],
"provider_specific_fields": {"tool_results": [SEARCH_RESULT, second_result]},
},
{"role": "tool", "tool_call_id": SEARCH_ID, "content": "found get_weather"},
{"role": "system", "content": _SYSTEM_NOTE},
]
with wire_server(anthropic_peer) as wire, modify_params_gateway.scenario() as scenario:
deployed: Final = scenario.model(model=model, api_base=wire.url, api_key="scripted-anthropic-key")
status, _, text = post_chat(
modify_params_gateway, chat_body(deployed, messages, stream=False, cache_control_injection_points=POINTS)
)
received: Final = wire.drain()
assert status == 200, text
assert len(received) == 1, received
assert anthropic_labels(received[0]) == expected_labels, received[0].body
_JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue])
def _chat_message_mark_labels(message: JsonValue) -> list[str]:
fields: Final = object_value(message)
role: Final = fields.get("role")
content: Final = fields.get("content")
calls: Final = fields.get("tool_calls")
blocks: Final = [object_value(block) for block in (content if isinstance(content, list) else [])]
marked_calls: Final = [object_value(call) for call in (calls if isinstance(calls, list) else [])]
return [
*([f"{role}:message"] if fields.get("cache_control") is not None else []),
*(f"{role}:text:{block.get('text')}" for block in blocks if block.get("cache_control") is not None),
*(f"{role}:tool_call:{call.get('id')}" for call in marked_calls if call.get("cache_control") is not None),
]
def _route_mark_labels(request: Request) -> list[str]:
if "/converse" in request.target:
return bedrock_labels(request)
messages: Final = _JSON_OBJECT.validate_json(request.body).get("messages")
return list(chain.from_iterable(map(_chat_message_mark_labels, messages if isinstance(messages, list) else [])))
def _cache_capped_route_peer(request: Request) -> Reply:
if "/converse" in request.target:
return bedrock_peer(request)
found: Final = len(_route_mark_labels(request))
if found > ANTHROPIC_CACHE_CONTROL_CAP:
message: Final = (
f"A maximum of {ANTHROPIC_CACHE_CONTROL_CAP} blocks with cache_control may be provided. Found {found}."
)
return Reply(
status=400, body=json.dumps({"error": {"message": message, "type": "invalid_request_error"}}).encode()
)
return chat_reply(f"chatcmpl-{marker_of(request)}", "scripted", "sunny", stream=False)
def _route_fields(model: str, wire_url: str) -> dict[str, JsonValue]:
if model.startswith("bedrock/"):
return {"aws_region_name": "us-east-1", "aws_bedrock_runtime_endpoint": wire_url}
return {}
def _two_answered_marked_searches_before_a_system_note(marker: str) -> list[JsonValue]:
second_result: Final[Event] = {**SEARCH_RESULT, "tool_use_id": _SECOND_SEARCH_ID}
return [
{"role": "system", "content": SYSTEM},
ask(marked=True),
{
"role": "assistant",
"content": "",
"tool_calls": [_marked_search_call(SEARCH_ID), _marked_search_call(_SECOND_SEARCH_ID)],
"provider_specific_fields": {"tool_results": [SEARCH_RESULT, second_result]},
},
{"role": "tool", "tool_call_id": SEARCH_ID, "content": "found get_weather"},
{"role": "tool", "tool_call_id": _SECOND_SEARCH_ID, "content": "found get_weather"},
{"role": "system", "content": f"{_SYSTEM_NOTE} marker-{marker}"},
]
_ROUTE_SYSTEM_LABEL: Final = f"system:text:{SYSTEM}"
_CHAT_ROUTE_LABELS: Final = [
_ROUTE_SYSTEM_LABEL,
ASK_LABEL,
f"assistant:tool_call:{SEARCH_ID}",
f"assistant:tool_call:{_SECOND_SEARCH_ID}",
]
_CONVERSE_LABELS: Final = [
_ROUTE_SYSTEM_LABEL,
ASK_LABEL,
f"assistant:toolUse:{SEARCH_ID}",
f"assistant:toolUse:{_SECOND_SEARCH_ID}",
]
@pytest.mark.parametrize(
("model", "target", "expected_labels"),
[
pytest.param(
"databricks/databricks-claude-opus-5-5", "/chat/completions", _CHAT_ROUTE_LABELS, id="databricks-converts"
),
pytest.param(
"databricks/databricks-claude-opus-5", "/chat/completions", _CHAT_ROUTE_LABELS, id="databricks-keeps"
),
pytest.param(
"openrouter/anthropic/claude-sonnet-4.6", "/chat/completions", _CHAT_ROUTE_LABELS, id="openrouter-converts"
),
pytest.param(
"openrouter/anthropic/claude-opus-5.5", "/chat/completions", _CHAT_ROUTE_LABELS, id="openrouter-keeps"
),
pytest.param(
"bedrock/converse/us.anthropic.claude-opus-4-7",
"/model/us.anthropic.claude-opus-4-7/converse",
_CONVERSE_LABELS,
id="converse-converts",
),
pytest.param(
"bedrock/converse/us.anthropic.claude-opus-5-5",
"/model/us.anthropic.claude-opus-5-5/converse",
_CONVERSE_LABELS,
id="converse-keeps",
),
],
)
@pytest.mark.parametrize("modify_params", [False, True], ids=["modify-params-off", "modify-params-on"])
def test_two_answered_marked_search_calls_before_a_system_message_stay_within_the_cache_mark_cap_off_anthropic(
gateway: Gateway,
modify_params_gateway: Gateway,
modify_params: bool,
model: str,
target: str,
expected_labels: list[str],
) -> None:
proxy: Final = modify_params_gateway if modify_params else gateway
marker: Final = uuid.uuid4().hex
messages: Final = _two_answered_marked_searches_before_a_system_note(marker)
with wire_server(answering_model_discovery(_cache_capped_route_peer)) as wire, proxy.scenario() as scenario:
deployed: Final = scenario.model(
model=model, api_base=wire.url, api_key=PROVIDER_KEY, **_route_fields(model, wire.url)
)
status, _, text = post_chat(
proxy, chat_body(deployed, messages, stream=False, tools=[TOOL], cache_control_injection_points=POINTS)
)
posted: Final = [request for request in wire.drain() if request.method == "POST"]
assert status == 200, text
assert [(request.target, marker_of(request)) for request in posted] == [(target, marker)], posted
assert _route_mark_labels(posted[0]) == expected_labels, posted[0].body
_STRAY_ANSWER: Final = "cloudy."
_STRAY_TOOL_ANSWERS: Final = (
pytest.param(call_id("London"), False, id="answers-an-earlier-turn"),
pytest.param("call_never_issued", False, id="unknown-id"),
pytest.param(7, False, id="integer-id"),
pytest.param(f"call_{'x' * 5120}", False, id="five-kilobyte-id"),
pytest.param(call_id("Paris"), False, id="answers-the-latest-turn-twice"),
pytest.param("", True, id="empty-id"),
)
def _two_tool_turns(marker: str, stray_id: JsonValue) -> list[JsonValue]:
return [
{"role": "system", "content": SYSTEM},
ask(marked=False),
{"role": "assistant", "content": "", "tool_calls": [tool_call("London")]},
{"role": "tool", "tool_call_id": call_id("London"), "content": "sunny."},
{"role": "user", "content": "And Paris?"},
{"role": "assistant", "content": "", "tool_calls": [tool_call("Paris")]},
{"role": "tool", "tool_call_id": stray_id, "content": _STRAY_ANSWER},
{"role": "tool", "tool_call_id": call_id("Paris"), "content": "sunny."},
{"role": "user", "content": final_text(marker)},
]
def _function_call(city: str) -> dict[str, JsonValue]:
return {
"type": "function_call",
"call_id": call_id(city),
"name": "lookup_weather",
"arguments": json.dumps({"city": city}),
}
def _two_tool_turns_as_responses_input(marker: str, stray_id: JsonValue) -> list[JsonValue]:
return [
{"role": "user", "content": ASK},
_function_call("London"),
{"type": "function_call_output", "call_id": call_id("London"), "output": "sunny."},
{"role": "user", "content": "And Paris?"},
_function_call("Paris"),
{"type": "function_call_output", "call_id": stray_id, "output": _STRAY_ANSWER},
{"type": "function_call_output", "call_id": call_id("Paris"), "output": "sunny."},
{"role": "user", "content": final_text(marker)},
]
def _assert_only_the_latest_turn_answers_reach_the_upstream(request: Request, marker: str, stray_kept: bool) -> None:
ids: Final = answered_ids(request)
assert marker_of(request) == marker, request.body
assert ids.count(call_id("London")) == 1, request.body
assert ids.count(call_id("Paris")) == 1, request.body
assert len(ids) == 2 + stray_kept, request.body
assert (_STRAY_ANSWER.encode() in request.body) is stray_kept, request.body
@pytest.mark.parametrize("stream", [False, True], ids=["non-stream", "stream"])
@pytest.mark.parametrize(("stray_id", "stray_kept"), _STRAY_TOOL_ANSWERS)
def test_modify_params_drops_a_tool_answer_that_answers_no_call_of_the_latest_assistant_turn(
modify_params_gateway: Gateway, stray_id: JsonValue, stray_kept: bool, stream: bool
) -> None:
marker: Final = uuid.uuid4().hex
with wire_server(anthropic_peer) as wire, modify_params_gateway.scenario() as scenario:
model: Final = deployment(scenario, "anthropic", wire.url, modify_params_gateway.upstream_url)
response: Final = chat(modify_params_gateway, chat_body(model, _two_tool_turns(marker, stray_id), stream))
received: Final = wire.drain()
assert response.status_code == 200, response.text
assert len(received) == 1, received
_assert_only_the_latest_turn_answers_reach_the_upstream(received[0], marker, stray_kept)
def test_modify_params_responses_input_drops_an_output_for_an_earlier_function_call(
modify_params_gateway: Gateway,
) -> None:
marker: Final = uuid.uuid4().hex
with wire_server(anthropic_peer) as wire, modify_params_gateway.scenario() as scenario:
model: Final = deployment(scenario, "anthropic", wire.url, modify_params_gateway.upstream_url)
response: Final = modify_params_gateway.request(
"POST",
"/v1/responses",
{
"model": model,
"max_output_tokens": 64,
"tools": [{"type": "function", "name": "lookup_weather", "parameters": {"type": "object"}}],
"input": _two_tool_turns_as_responses_input(marker, call_id("London")),
},
)
received: Final = wire.drain()
assert response.status_code == 200, response.text
assert len(received) == 1, received
_assert_only_the_latest_turn_answers_reach_the_upstream(received[0], marker, False)
def _with_tokyo_call(identity: JsonValue) -> dict[str, JsonValue]:
return {
"role": "assistant",
"content": "",
"tool_calls": [tool_call("Tokyo", cache_control=EPHEMERAL, id=identity)],
}
_MALFORMED_MESSAGES: Final = (
pytest.param(_with_tokyo_call({"id": call_id("Tokyo")}), id="tool-call-id-object"),
pytest.param(_with_tokyo_call([call_id("Tokyo")]), id="tool-call-id-list"),
)
@pytest.mark.parametrize("malformed", _MALFORMED_MESSAGES)
@pytest.mark.parametrize("modify_params", [False, True], ids=["modify-params-off", "modify-params-on"])
def test_marked_search_call_next_to_a_malformed_tool_id_and_a_later_system_message_answers_400(
gateway: Gateway,
modify_params_gateway: Gateway,
modify_params: bool,
malformed: dict[str, JsonValue],
) -> None:
proxy: Final = modify_params_gateway if modify_params else gateway
marker: Final = uuid.uuid4().hex
messages: Final[list[JsonValue]] = [
{"role": "system", "content": SYSTEM},
ask(marked=False),
{
"role": "assistant",
"content": "",
"tool_calls": [tool_call(city, cache_control=EPHEMERAL) for city in _CITIES],
},
*({"role": "tool", "tool_call_id": call_id(city), "content": "sunny."} for city in _CITIES),
_marked_search_turn(),
malformed,
{"role": "user", "content": final_text(marker)},
{"role": "system", "content": _SYSTEM_NOTE},
]
with wire_server(anthropic_peer) as wire, proxy.scenario() as scenario:
model: Final = deployment(scenario, "anthropic", wire.url, proxy.upstream_url)
status, _, text = post_chat(
proxy, chat_body(model, messages, stream=False, cache_control_injection_points=POINTS)
)
received: Final = wire.drain()
assert status == 400, text
assert '"error"' in text, text
assert received == (), [request.target for request in received]
assert proxy.client.get("/health/liveliness").status_code == 200

112
tests/integration/sdk/test_cache_control_census_pydantic_messages_sdk.py
View file

@ -7,11 +7,13 @@ from integration._support.openai_wire import responses_reply
from integration._support.wire import Reply, Request, Wire, wire_server
from integration.providers._cache_control_marks_support import (
ANTHROPIC_MODEL,
ASK_LABEL,
CITIES,
EPHEMERAL,
POINTS,
PROVIDER_KEY,
SYSTEM,
SYSTEM_LABEL,
TOOL,
UNMAPPED_BEDROCK_ARN,
anthropic_labels,
@ -21,13 +23,16 @@ from integration.providers._cache_control_marks_support import (
call_id,
client_marked,
final_text,
marker_of,
new_marker,
sent_without_cache_points,
)
from integration.providers._tool_search_support import SEARCH_ID, SEARCH_RESULT
from openai.types.responses import FunctionTool
from pydantic import JsonValue
import litellm
from litellm import Router
from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper
from litellm.types.utils import ChatCompletionMessageToolCall, Message, ModelResponse
@ -159,3 +164,110 @@ def test_each_input_shape_reads_the_same_whether_or_not_the_route_keeps_item_mar
stripped_sent: Final = sent_without_cache_points(wire.drain(), stripped_marker)
assert stripped == kept
assert stripped_sent == kept_sent
_SECOND_SEARCH_ID: Final = "srvtoolu_02ScriptedSearch"
_BARE_MODEL: Final = "bare-bedrock-invoke"
def _marked_search_call(search_id: str) -> dict[str, JsonValue]:
return {
"id": search_id,
"type": "function",
"function": {"name": "tool_search_tool_bm25", "arguments": '{"query": "weather"}'},
"cache_control": EPHEMERAL,
}
def _system_note(marker: str) -> str:
return f"Answer in one sentence. marker-{marker}"
def _client_answer_to_one_of_two_marked_searches(marker: str) -> list[JsonValue]:
return [
{"role": "system", "content": SYSTEM},
ask(marked=True),
{
"role": "assistant",
"content": "",
"tool_calls": [_marked_search_call(SEARCH_ID), _marked_search_call(_SECOND_SEARCH_ID)],
"provider_specific_fields": {
"tool_results": [SEARCH_RESULT, {**SEARCH_RESULT, "tool_use_id": _SECOND_SEARCH_ID}]
},
},
{"role": "tool", "tool_call_id": SEARCH_ID, "content": "found get_weather"},
{"role": "system", "content": _system_note(marker)},
]
def _bare_bedrock_invoke(model: str, wire: Wire) -> dict[str, object]:
return {
"model": f"invoke/{model}",
"custom_llm_provider": "bedrock",
"api_base": wire.url,
"api_key": PROVIDER_KEY,
"aws_region_name": "us-east-1",
"aws_bedrock_runtime_endpoint": wire.url,
"num_retries": 0,
}
def _bare_anthropic(model: str, wire: Wire) -> dict[str, object]:
return {
"model": model,
"custom_llm_provider": "anthropic",
"api_base": wire.url,
"api_key": PROVIDER_KEY,
"num_retries": 0,
}
def _litellm_completion(deployment: dict[str, object], messages: list[JsonValue]) -> object:
return litellm.completion(**deployment, messages=messages, max_tokens=64, cache_control_injection_points=POINTS) # pyright: ignore[reportArgumentType] # the dict shapes a caller passes, unvalidated
def _router_completion(deployment: dict[str, object], messages: list[JsonValue]) -> object:
router: Final = Router(model_list=[{"model_name": _BARE_MODEL, "litellm_params": deployment}], num_retries=0)
return router.completion(model=_BARE_MODEL, messages=messages, max_tokens=64, cache_control_injection_points=POINTS) # pyright: ignore[reportUnknownMemberType, reportArgumentType] # Router.completion leaves **kwargs untyped and names a narrower message shape than callers pass
def _expected_marks(keeps_mid_conversation_system: bool, marker: str) -> list[str]:
if keeps_mid_conversation_system:
return [SYSTEM_LABEL, ASK_LABEL, f"assistant:tool_use:{SEARCH_ID}", f"assistant:tool_use:{_SECOND_SEARCH_ID}"]
return [SYSTEM_LABEL, ASK_LABEL, f"user:text:{_system_note(marker)}"]
@pytest.mark.parametrize(
"complete",
[
pytest.param(_litellm_completion, id="litellm-completion"),
pytest.param(_router_completion, id="router-completion"),
],
)
@pytest.mark.parametrize(
("deployment", "model", "keeps_mid_conversation_system"),
[
pytest.param(_bare_bedrock_invoke, "us.anthropic.claude-opus-5-5", True, id="keeps-mid-conversation-system"),
pytest.param(
_bare_bedrock_invoke, "us.anthropic.claude-sonnet-4-6", False, id="converts-mid-conversation-system"
),
pytest.param(_bare_anthropic, "unmapped-claude-alias", False, id="unmapped-model"),
],
)
def test_bare_model_with_custom_llm_provider_fills_the_cache_mark_budget_like_the_request_builder(
monkeypatch: pytest.MonkeyPatch,
complete: Callable[[dict[str, object], list[JsonValue]], object],
deployment: Callable[[str, Wire], dict[str, object]],
model: str,
keeps_mid_conversation_system: bool,
) -> None:
monkeypatch.setattr(litellm, "modify_params", True)
marker: Final = new_marker()
with wire_server(anthropic_peer) as wire:
response: Final = complete(deployment(model, wire), _client_answer_to_one_of_two_marked_searches(marker))
received: Final = wire.drain()
assert isinstance(response, ModelResponse), type(response)
assert response.choices[0].message.content == "sunny", response # pyright: ignore[reportAttributeAccessIssue] # choices are Choices for a non-stream response
assert [(marker_of(request), anthropic_labels(request)) for request in received] == [
(marker, _expected_marks(keeps_mid_conversation_system, marker))
]

39
tests/integration/sdk/test_device_code_login_guard_sdk.py
View file

@ -44,9 +44,28 @@ _SCRIPT: Final = textwrap.dedent(
def sync_completion():
return litellm.completion(model=model, messages=messages, num_retries=0)
def server_tool_call_turns(marked):
call = {"id": "srvtoolu_01DeviceLogin", "type": "function", "function": {"name": "tool_search_tool_bm25", "arguments": "{}"}}
mark = {"cache_control": {"type": "ephemeral"}} if marked else {}
result = {"type": "tool_search_tool_result", "tool_use_id": call["id"], "content": {"type": "tool_search_tool_search_result", "tool_references": []}}
return [
{"role": "system", "content": "Answer in one sentence."},
*messages,
{"role": "assistant", "content": "", "tool_calls": [{**call, **mark}], "provider_specific_fields": {"tool_results": [result]}},
{"role": "tool", "tool_call_id": call["id"], "content": "found get_weather"},
{"role": "user", "content": "go on"},
]
def run():
if variant == "completion":
return sync_completion()
if variant in ("marked-server-tool-call", "unmarked-server-tool-call"):
return litellm.completion(
model=model,
messages=server_tool_call_turns(variant == "marked-server-tool-call"),
num_retries=0,
cache_control_injection_points=[{"location": "message", "role": "system"}, {"location": "message", "index": -1}],
)
if variant == "responses":
return litellm.responses(model=model, input=f"Reply to marker-{marker}", num_retries=0)
if variant == "acompletion":
@ -228,3 +247,23 @@ def test_sync_call_on_the_main_thread_fails_without_hanging_when_the_auth_host_d
assert f"{dl.CHATGPT_AUTH_HOST if provider == 'chatgpt' else dl.GITHUB_HOST}:443" in clean.peers.auth_connections()
assert clean.peers.api.drain() == ()
assert clean.token_files() == ()
def _auth_requests(peers: dl.Peers) -> list[tuple[str, str, str]]:
return [(request.headers.get("host", ""), request.method, request.target) for request in peers.auth.drain()]
@pytest.mark.parametrize("provider", ("chatgpt", "copilot"))
def test_a_marked_server_tool_call_reaches_the_auth_host_like_an_unmarked_one_when_the_login_is_denied(
clean: _Bench, provider: Provider
) -> None:
unmarked: Final = clean.run("unmarked-server-tool-call", provider, uuid.uuid4().hex)
unmarked_auth: Final = _auth_requests(clean.peers)
marked: Final = clean.run("marked-server-tool-call", provider, uuid.uuid4().hex)
marked_auth: Final = _auth_requests(clean.peers)
assert unmarked.verdict()["ok"] is False, (unmarked.verdict(), unmarked.stderr[-3000:])
assert marked.verdict()["ok"] is False, (marked.verdict(), marked.stderr[-3000:])
assert unmarked_auth, unmarked.stderr[-3000:]
assert marked_auth == unmarked_auth
assert clean.peers.api.drain() == ()
assert clean.token_files() == ()

426
tests/unit/integrations/test_anthropic_cache_control_hook.py
View file

@ -1,6 +1,9 @@
import copy
import itertools
import json
import logging
import os
import pathlib
import subprocess
import sys
import textwrap
@ -9,9 +12,11 @@ from typing import Final, List, Optional, Tuple, cast
from unittest.mock import MagicMock, patch
import pytest
from openai.types.chat import ChatCompletionMessage
from pydantic import BaseModel, ConfigDict
import litellm
from litellm._logging import verbose_logger
from litellm.integrations.anthropic_cache_control_hook import (
CARRY_UNMATCHED_MESSAGE_POINTS,
EXTERNAL_BREAKPOINTS_STAMP,
@ -23,9 +28,11 @@ from litellm.litellm_core_utils.prompt_templates.factory import (
_convert_to_bedrock_tool_call_invoke,
convert_to_anthropic_tool_invoke,
)
from litellm.llms.anthropic.chat.transformation import AnthropicConfig
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
from litellm.types.llms.openai import AllMessageValues, ChatCompletionAssistantToolCall
from litellm.types.utils import ChatCompletionMessageToolCall, Message
from litellm.types.utils import ChatCompletionMessageToolCall, LlmProviders, Message
from litellm.utils import ProviderConfigManager
@pytest.fixture(autouse=True)
@ -1166,46 +1173,303 @@ def test_cache_control_hook_counts_tool_call_marks_except_answered_server_tool_c
assert AnthropicCacheControlHook.count_request_cache_breakpoints([message]) == 2
@pytest.mark.parametrize(
("follow_ups", "expected_breakpoints"),
[
pytest.param((), 0, id="rebuilt_from_its_kept_result"),
pytest.param(
({"role": "tool", "tool_call_id": "srvtoolu_01Search", "content": "found get_weather"},),
1,
id="replayed_with_the_client_answer",
),
pytest.param(
(
{"role": "tool", "tool_call_id": "srvtoolu_01Search", "content": "found get_weather"},
{"role": "user", "content": "Go on"},
),
0,
id="client_answer_replaced_by_the_kept_result",
),
],
)
def test_cache_control_hook_counts_a_server_tool_call_mark_only_when_the_call_replays_as_tool_use(
follow_ups: tuple[AllMessageValues, ...], expected_breakpoints: int
):
search_call: Final = cast(
_SEARCH_ANSWER: Final = {"role": "tool", "tool_call_id": "srvtoolu_01Search", "content": "found get_weather"}
_GO_ON: Final = {"role": "user", "content": "Go on"}
_SYSTEM_NOTE: Final = {"role": "system", "content": "Answer in one sentence."}
_SEARCH_CALL: Final = {
"id": "srvtoolu_01Search",
"type": "function",
"function": {"name": "tool_search_tool_bm25", "arguments": "{}"},
"cache_control": {"type": "ephemeral"},
}
_WEATHER_CALL: Final = {
"id": "toolu_01Weather",
"type": "function",
"function": {"name": "get_weather", "arguments": '{"city": "Paris"}'},
"cache_control": {"type": "ephemeral"},
}
def _wire_cache_marks(messages: list[AllMessageValues], model: str = "claude-opus-5-5") -> int:
request: Final = AnthropicConfig().transform_request(
model=model,
messages=copy.deepcopy(messages),
optional_params={"max_tokens": 64},
litellm_params={},
headers={},
)
message_blocks: Final = itertools.chain.from_iterable(
message["content"] for message in request["messages"] if isinstance(message["content"], list)
)
blocks: Final = itertools.chain(request.get("system") or (), message_blocks)
return sum(1 for block in blocks if isinstance(block, dict) and "cache_control" in block)
def _replayed_search_call(*tool_calls: dict[str, object]) -> AllMessageValues:
return cast(
AllMessageValues,
{
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "srvtoolu_01Search",
"type": "function",
"function": {"name": "tool_search_tool_bm25", "arguments": "{}"},
"cache_control": {"type": "ephemeral"},
}
],
"provider_specific_fields": {"tool_results": [{"tool_use_id": "srvtoolu_01Search"}]},
"tool_calls": list(tool_calls),
"provider_specific_fields": {
"tool_results": [
{
"type": "tool_search_tool_result",
"tool_use_id": "srvtoolu_01Search",
"content": {"type": "tool_search_tool_search_result", "tool_references": []},
}
]
},
},
)
assert AnthropicCacheControlHook.count_request_cache_breakpoints([search_call, *follow_ups]) == expected_breakpoints
@pytest.mark.parametrize(
("tool_calls", "follow_ups", "modify_params", "expected_breakpoints"),
[
pytest.param((_SEARCH_CALL,), (), False, 0, id="trailing_call_rebuilt_from_its_kept_result"),
pytest.param((_SEARCH_CALL,), (), True, 1, id="trailing_call_replayed_with_the_modify_params_placeholder"),
pytest.param((_SEARCH_CALL,), (_SEARCH_ANSWER,), False, 1, id="replayed_with_the_client_answer"),
pytest.param(
(_SEARCH_CALL,), (_SEARCH_ANSWER,), True, 1, id="replayed_with_the_client_answer_under_modify_params"
),
pytest.param(
(_SEARCH_CALL,), (_SEARCH_ANSWER, _GO_ON), False, 0, id="client_answer_replaced_by_the_kept_result"
),
pytest.param((_SEARCH_CALL,), (_GO_ON,), True, 0, id="placeholder_replaced_by_the_kept_result"),
pytest.param(
(_SEARCH_CALL, _WEATHER_CALL),
(_SEARCH_ANSWER,),
True,
1,
id="client_answer_replaced_when_modify_params_answers_the_other_call",
),
pytest.param(
(_SEARCH_CALL,), (_SYSTEM_NOTE,), True, 0, id="placeholder_replaced_when_a_later_system_message_is_sent"
),
pytest.param((_SEARCH_CALL,), (_SYSTEM_NOTE,), False, 0, id="trailing_call_rebuilt_before_a_system_message"),
],
)
@pytest.mark.parametrize("model", ["claude-opus-5-5", "claude-sonnet-4-6"])
def test_cache_control_hook_counts_a_server_tool_call_mark_only_when_anthropic_receives_it(
monkeypatch: pytest.MonkeyPatch,
tool_calls: tuple[dict[str, object], ...],
follow_ups: tuple[AllMessageValues, ...],
modify_params: bool,
expected_breakpoints: int,
model: str,
):
monkeypatch.setattr(litellm, "modify_params", modify_params)
messages: Final = [
cast(AllMessageValues, {"role": "user", "content": "Weather in Paris?"}),
_replayed_search_call(*tool_calls),
*follow_ups,
]
census: Final = AnthropicCacheControlHook.count_request_cache_breakpoints(messages)
assert (census, _wire_cache_marks(messages, model)) == (expected_breakpoints, expected_breakpoints)
@pytest.mark.parametrize(
"after_the_call",
[(_SEARCH_ANSWER, _SYSTEM_NOTE), (_SYSTEM_NOTE, _SEARCH_ANSWER)],
ids=["system_message_after_the_answer", "system_message_before_the_answer"],
)
@pytest.mark.parametrize("modify_params", [False, True])
@pytest.mark.parametrize("model", ["claude-opus-5-5", "claude-sonnet-4-6"])
def test_cache_control_hook_never_counts_fewer_marks_than_anthropic_receives_after_a_system_message(
monkeypatch: pytest.MonkeyPatch, modify_params: bool, model: str, after_the_call: tuple[dict[str, object], ...]
):
monkeypatch.setattr(litellm, "modify_params", modify_params)
messages: Final = [
cast(AllMessageValues, {"role": "system", "content": "You are a weather assistant."}),
cast(AllMessageValues, {"role": "user", "content": "Weather in Paris?"}),
_replayed_search_call(_SEARCH_CALL),
*(cast(AllMessageValues, message) for message in after_the_call),
]
census: Final = AnthropicCacheControlHook.count_request_cache_breakpoints(messages)
assert census >= _wire_cache_marks(messages, model)
@pytest.mark.parametrize(
("hook_model", "wire_model"),
[
("anthropic/claude-opus-5-5", "claude-opus-5-5"),
("anthropic/claude-sonnet-4-6", "claude-sonnet-4-6"),
pytest.param("invoke/us.anthropic.claude-opus-5-5", "claude-opus-5-5", id="provider-left-out-of-the-params"),
pytest.param("anthropic/my-claude-alias", "my-claude-alias", id="model-the-cost-map-does-not-know"),
pytest.param(
"anthropic/my-unreadable-claude-alias", "my-unreadable-claude-alias", id="cost-map-entry-the-lookup-cannot-read"
),
],
)
def test_cache_control_hook_fills_the_mark_budget_when_the_client_answers_one_of_two_marked_search_calls(
monkeypatch: pytest.MonkeyPatch, hook_model: str, wire_model: str
):
monkeypatch.setattr(litellm, "modify_params", True)
monkeypatch.setitem(litellm.model_cost, "my-unreadable-claude-alias", [])
marked_text: Final = {"type": "text", "text": "Context: a trip.", "cache_control": {"type": "ephemeral"}}
two_kept_searches: Final = {
"role": "assistant",
"content": None,
"tool_calls": [_SEARCH_CALL, {**_SEARCH_CALL, "id": "srvtoolu_02Search"}],
"provider_specific_fields": {
"tool_results": [
{
"type": "tool_search_tool_result",
"tool_use_id": tool_use_id,
"content": {"type": "tool_search_tool_search_result", "tool_references": []},
}
for tool_use_id in ("srvtoolu_01Search", "srvtoolu_02Search")
]
},
}
messages: Final = [
{"role": "system", "content": "You are a weather assistant."},
{"role": "user", "content": [marked_text, {**marked_text, "text": "Find a weather tool."}]},
two_kept_searches,
_SEARCH_ANSWER,
_SYSTEM_NOTE,
]
_, processed, _ = AnthropicCacheControlHook().get_chat_completion_prompt(
model=hook_model,
messages=cast(list[AllMessageValues], messages),
non_default_params={
"cache_control_injection_points": [
{"location": "message", "role": "system"},
{"location": "message", "index": -1},
]
},
prompt_id=None,
prompt_variables=None,
dynamic_callback_params={},
)
assert _wire_cache_marks(processed, wire_model) == 4
def _marks_in(value: object) -> int:
if isinstance(value, Mapping):
mapping: Final = cast(Mapping[str, object], value)
own_mark: Final = int("cache_control" in mapping or "cachePoint" in mapping)
return own_mark + sum(_marks_in(item) for item in mapping.values())
if isinstance(value, list):
return sum(_marks_in(item) for item in cast(list[object], value))
return 0
def _marks_the_route_sends(model: str, custom_llm_provider: str | None, messages: list[AllMessageValues]) -> int:
served_model, provider, _, _ = (
(model, custom_llm_provider, None, None) if custom_llm_provider else litellm.get_llm_provider(model=model)
)
route_config: Final = ProviderConfigManager.get_provider_chat_config(
model=served_model, provider=LlmProviders(provider)
)
assert route_config is not None
weather_tool: Final = {
"type": "function",
"function": {"name": "get_weather", "parameters": {"type": "object", "properties": {}}},
}
return _marks_in(
route_config.transform_request(
model=served_model,
messages=copy.deepcopy(messages),
optional_params={"max_tokens": 64, "tools": [weather_tool]},
litellm_params={},
headers={},
)
)
_SECOND_SEARCH_CALL: Final = {**_SEARCH_CALL, "id": "srvtoolu_02Search"}
@pytest.mark.parametrize(
("model", "custom_llm_provider"),
[
("anthropic/claude-sonnet-4-6", None),
("bedrock/invoke/us.anthropic.claude-opus-5-5", None),
("vertex_ai/claude-opus-5-5@default", None),
("databricks/databricks-claude-sonnet-4-6", None),
pytest.param("databricks-claude-sonnet-4-6", "databricks", id="provider-passed-beside-a-bare-model"),
("openrouter/anthropic/claude-sonnet-4.6", None),
("bedrock/converse/us.anthropic.claude-sonnet-4-6", None),
pytest.param("claude-sonnet-4.6", "github_copilot", id="copilot-with-no-stored-login"),
],
)
@pytest.mark.parametrize(
("tool_calls", "follow_ups"),
[
pytest.param(
(_SEARCH_CALL, _SECOND_SEARCH_CALL),
(_SEARCH_ANSWER, {**_SEARCH_ANSWER, "tool_call_id": "srvtoolu_02Search"}, _SYSTEM_NOTE),
id="both-search-calls-answered-before-a-system-message",
),
pytest.param(
(_SEARCH_CALL, _WEATHER_CALL), (_SEARCH_ANSWER,), id="function-call-left-for-modify-params-to-answer"
),
],
)
def test_cache_control_hook_fills_the_mark_budget_every_claude_route_sends_after_marked_search_calls(
monkeypatch: pytest.MonkeyPatch,
tmp_path: pathlib.Path,
model: str,
custom_llm_provider: str | None,
tool_calls: tuple[dict[str, object], ...],
follow_ups: tuple[dict[str, object], ...],
):
monkeypatch.setattr(litellm, "modify_params", True)
monkeypatch.setenv("GITHUB_COPILOT_TOKEN_DIR", str(tmp_path))
monkeypatch.setenv("GITHUB_COPILOT_DEVICE_CODE_URL", "http://127.0.0.1:9/login/device/code")
marked_text: Final = {"type": "text", "text": "Context: a trip.", "cache_control": {"type": "ephemeral"}}
kept_searches: Final = {
"role": "assistant",
"content": None,
"tool_calls": list(tool_calls),
"provider_specific_fields": {
"tool_results": [
{
"type": "tool_search_tool_result",
"tool_use_id": tool_call["id"],
"content": {"type": "tool_search_tool_search_result", "tool_references": []},
}
for tool_call in tool_calls
if str(tool_call["id"]).startswith("srvtoolu_")
]
},
}
messages: Final = cast(
list[AllMessageValues],
[
{"role": "system", "content": "You are a weather assistant."},
{"role": "user", "content": [marked_text, {**marked_text, "text": "Find a weather tool."}]},
kept_searches,
*follow_ups,
],
)
params: Final[dict[str, object]] = {
"cache_control_injection_points": [
{"location": "message", "role": "system"},
{"location": "message", "index": -1},
]
}
AnthropicCacheControlHook.maybe_seed_default_injection_points(params, messages, model, custom_llm_provider)
_, processed, _ = AnthropicCacheControlHook().get_chat_completion_prompt(
model=model,
messages=messages,
non_default_params=params,
prompt_id=None,
prompt_variables=None,
dynamic_callback_params={},
)
assert _marks_the_route_sends(model, custom_llm_provider, processed) == 4
@pytest.mark.parametrize("tool_call_type", ["function", "custom", None])
@ -4572,3 +4836,97 @@ class TestMalformedInjectionPointsAreIgnored:
kwargs: dict = {"cache_control_injection_points": copy.deepcopy(value)}
_stamp_injection_points_for_the_hook(kwargs, "gpt-5.6", "openai", None, None)
assert kwargs == {"cache_control_injection_points": value}
def test_cache_control_hook_leaves_the_duplicate_tool_result_warning_to_the_request_builder(
monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
):
monkeypatch.setattr(litellm, "modify_params", True)
weather_answer: Final = {"role": "tool", "tool_call_id": "toolu_01Weather", "content": "18C"}
messages: Final = [
cast(AllMessageValues, {"role": "user", "content": "Weather in Paris?"}),
cast(AllMessageValues, {"role": "assistant", "content": None, "tool_calls": [_SEARCH_CALL, _WEATHER_CALL]}),
cast(AllMessageValues, _SEARCH_ANSWER),
cast(AllMessageValues, weather_answer),
cast(AllMessageValues, weather_answer),
]
monkeypatch.setattr(verbose_logger, "propagate", True)
caplog.set_level(logging.WARNING, logger=verbose_logger.name)
census: Final = AnthropicCacheControlHook.count_request_cache_breakpoints(messages)
warnings_from_the_hook: Final = len(caplog.records)
wire_marks: Final = _wire_cache_marks(messages)
assert (census, wire_marks) == (2, 2)
assert (warnings_from_the_hook, len(caplog.records)) == (0, 1)
@pytest.mark.parametrize(
"stray_message",
[
pytest.param("junk", id="string"),
pytest.param(None, id="null"),
pytest.param(ChatCompletionMessage(role="assistant", content="Checking the weather"), id="openai_sdk_message"),
],
)
def test_cache_control_hook_counts_marked_tool_calls_next_to_a_message_it_cannot_pair(
monkeypatch: pytest.MonkeyPatch, stray_message: object
):
monkeypatch.setattr(litellm, "modify_params", True)
messages: Final = [
stray_message,
{"role": "assistant", "content": None, "tool_calls": [_SEARCH_CALL, _WEATHER_CALL]},
_SEARCH_ANSWER,
{"role": "tool", "tool_call_id": "toolu_01Weather", "content": "18C"},
]
census: Final = AnthropicCacheControlHook.count_request_cache_breakpoints(messages)
envelope_census: Final = AnthropicCacheControlHook.count_external_cache_breakpoints(
None, request_kwargs={"extra_body": {"messages": messages}}
)
assert (census, envelope_census) == (2, 2)
@pytest.mark.parametrize("modify_params", [False, True])
@pytest.mark.parametrize(
"malformed",
[
pytest.param(
{"role": "assistant", "content": None, "tool_calls": [{**_WEATHER_CALL, "id": {"id": "toolu_02"}}]},
id="tool_call_id_is_a_dict",
),
pytest.param(
{"role": "assistant", "content": None, "tool_calls": [{**_WEATHER_CALL, "id": ["toolu_02"]}]},
id="tool_call_id_is_a_list",
),
pytest.param({"role": "tool", "tool_call_id": {"id": "toolu_02"}, "content": "18C"}, id="answer_id_is_a_dict"),
pytest.param({"role": ["assistant"], "content": "Checking"}, id="role_is_a_list"),
],
)
def test_cache_control_hook_injects_next_to_a_malformed_message_instead_of_raising(
monkeypatch: pytest.MonkeyPatch, modify_params: bool, malformed: dict[str, object]
):
monkeypatch.setattr(litellm, "modify_params", modify_params)
messages: Final = [
{"role": "system", "content": "You are a weather assistant."},
{"role": "user", "content": "Weather in Paris?"},
{"role": "assistant", "content": None, "tool_calls": [_SEARCH_CALL]},
_SEARCH_ANSWER,
malformed,
_SYSTEM_NOTE,
]
_, processed, _ = AnthropicCacheControlHook().get_chat_completion_prompt(
model="anthropic/claude-opus-5-5",
messages=cast(list[AllMessageValues], messages),
non_default_params={"cache_control_injection_points": [{"location": "message", "role": "system"}]},
prompt_id=None,
prompt_variables=None,
dynamic_callback_params={},
)
assert [message.get("cache_control") for message in processed if message.get("role") == "system"] == [
{"type": "ephemeral"},
{"type": "ephemeral"},
]

62
tests/unit/litellm_core_utils/prompt_templates/test_litellm_core_utils_prompt_templates_factory.py
View file

@ -39,6 +39,7 @@ from litellm.litellm_core_utils.prompt_templates.factory import (
THOUGHT_SIGNATURE_SEPARATOR,
)
from litellm.types.llms.openai import ChatCompletionAssistantMessage, ChatCompletionToolMessage
from litellm.types.utils import ChatCompletionMessageToolCall, Function
from litellm.utils import (
_invalidate_model_cost_lowercase_map,
function_setup,
@ -3521,6 +3522,67 @@ def test_sanitize_messages_combined_case_a_and_case_d():
litellm.modify_params = original
def test_sanitize_messages_drops_tool_results_the_latest_assistant_turn_did_not_ask_for(
monkeypatch: pytest.MonkeyPatch,
):
monkeypatch.setattr(litellm, "modify_params", True)
weather_call: Final = {"id": "call_a", "type": "function", "function": {"name": "get_weather", "arguments": "{}"}}
time_call: Final = ChatCompletionMessageToolCall(
id="call_b", type="function", function=Function(name="get_time", arguments="{}")
)
weather_answer: Final = {"role": "tool", "tool_call_id": "call_a", "content": "18C"}
time_answer: Final = {"role": "tool", "tool_call_id": "call_b", "content": "noon"}
asked_for_time: Final = {"role": "assistant", "content": None, "tool_calls": [time_call]}
done: Final = {"role": "assistant", "content": "Done"}
messages: Final = (
{"role": "user", "content": "Weather?"},
{"role": "assistant", "content": None, "tool_calls": [weather_call]},
weather_answer,
{"role": "user", "content": "Time?"},
asked_for_time,
time_answer,
weather_answer,
done,
time_answer,
)
sanitized: Final = sanitize_messages_for_tool_calling(list(messages))
assert tuple(sanitized) == (*messages[:6], done)
def _sanitize_seconds(answers: int) -> float:
tool_calls: Final = [
{"id": f"call_{index}", "type": "function", "function": {"name": "get_weather", "arguments": "{}"}}
for index in range(answers)
]
messages: Final = (
{"role": "user", "content": "Weather everywhere?"},
{"role": "assistant", "content": None, "tool_calls": tool_calls},
*({"role": "tool", "tool_call_id": f"call_{index}", "content": "18C"} for index in range(answers)),
{"role": "user", "content": "And now?"},
)
def once() -> float:
started: Final = time.perf_counter()
sanitized: Final = sanitize_messages_for_tool_calling(list(messages))
elapsed: Final = time.perf_counter() - started
assert tuple(sanitized) == messages
return elapsed
return min(once() for _ in range(3))
def test_sanitize_messages_cost_grows_linearly_with_the_number_of_answered_tool_calls(
monkeypatch: pytest.MonkeyPatch,
):
monkeypatch.setattr(litellm, "modify_params", True)
growth: Final = _sanitize_seconds(4000) / _sanitize_seconds(1000)
assert growth < 11.0, f"cost grew {growth:.1f}x for 4x the tool answers"
def test_anthropic_messages_pt_file_block_preserves_cache_control():
"""
Test that cache_control is preserved on file-type content blocks

44
tests/unit/litellm_core_utils/prompt_templates/test_litellm_core_utils_prompt_templates_mid_conversation_system.py
View file

@ -338,6 +338,50 @@ def test_flagged_placement_handles_a_long_run_of_unanswered_tool_search_uses_in_
assert elapsed < 1.0
def test_flagged_placement_merges_every_system_run_that_shares_a_user_turn():
placed: Final = place_mid_conversation_system(
[
{"role": "user", "content": "q1"},
{"role": "system", "content": "first"},
{"role": "user", "content": "q2"},
{"role": "system", "content": "second"},
{"role": "assistant", "content": "a"},
],
supports_mid_conversation_system=True,
)
assert _roles(placed) == ["user", "user", "system", "assistant"]
assert _texts(placed[2]) == ["first", "second"]
def _flagged_placement_seconds(turns: int) -> float:
messages: Final = tuple(
chain.from_iterable(
(
{"role": "user", "content": f"q{index}"},
{"role": "system", "content": f"reminder {index}"},
{"role": "assistant", "content": f"a{index}"},
)
for index in range(turns)
)
)
def once() -> float:
started: Final = time.perf_counter()
placed: Final = place_mid_conversation_system(messages, supports_mid_conversation_system=True)
elapsed: Final = time.perf_counter() - started
assert _roles(placed) == ["user", "system", "assistant"] * turns
return elapsed
return min(once() for _ in range(3))
def test_flagged_placement_cost_grows_linearly_with_the_number_of_mid_conversation_system_messages():
growth: Final = _flagged_placement_seconds(4000) / _flagged_placement_seconds(1000)
assert growth < 11.0, f"cost grew {growth:.1f}x for 4x the messages"
def test_flagged_placement_keeps_a_system_before_a_tool_search_use_the_caller_answered():
placed = place_mid_conversation_system(
[