scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-09 22:31:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 133

fix: skip one-shot Claude Code cache injection (#40175)

Browse source
This commit is contained in:
tin-berri 2026-09-07 18:03:43 -07:00 • committed by GitHub
parent 26d589cd28
commit 7da6fe54b5
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
7 changed files with 492 additions and 11 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

82
litellm/integrations/anthropic_cache_control_hook.py
View file

@ -16,6 +16,8 @@ from collections.abc import Iterable, Mapping, Sequence
from typing import TYPE_CHECKING, Any, Final, cast
from urllib.parse import urlparse
from pydantic import TypeAdapter, ValidationError
from litellm._logging import verbose_logger
from litellm.integrations.custom_logger import CustomLogger
from litellm.integrations.custom_prompt_management import CustomPromptManagement
@ -23,6 +25,7 @@ from litellm.integrations.prompt_management_base import PromptManagementClient
from litellm.litellm_core_utils.prompt_templates.common_utils import (
with_prompt_cache_breakpoint,
)
from litellm.llms.anthropic.common_utils import is_claude_code_one_shot_subagent_request
from litellm.types.integrations.anthropic_cache_control_hook import (
GATEWAY_INJECTED_CACHE_METADATA_KEY,
GATEWAY_INJECTED_FOR_EVERY_DEPLOYMENT,
@ -62,10 +65,26 @@ OPENAI_PROMPT_CACHE_BREAKPOINT_BLOCK_TYPES: Final = frozenset(
)
OPENAI_API_HOST: Final = "api.openai.com"
OPENAI_API_BASE_ENV_VARS: Final = ("OPENAI_BASE_URL", "OPENAI_API_BASE")
_OBJECT_MAPPING_ADAPTER: Final = TypeAdapter(dict[object, object])
_OBJECT_LIST_ADAPTER: Final = TypeAdapter(list[object])
AllToolParamValues = ChatCompletionToolParam | AllAnthropicToolsValues
def _validated_object_mapping(value: object) -> dict[object, object] | None:
try:
return _OBJECT_MAPPING_ADAPTER.validate_python(value)
except ValidationError:
return None
def _validated_object_list(value: object) -> list[object] | None:
try:
return _OBJECT_LIST_ADAPTER.validate_python(value)
except ValidationError:
return None
def supports_openai_prompt_cache_breakpoint(model: str) -> bool:
model_map_flag: Final = _model_map_prompt_cache_breakpoint_flag(model)
if model_map_flag is not None:
@ -114,6 +133,36 @@ CARRY_UNMATCHED_MESSAGE_POINTS: Final = "_litellm_carry_unmatched_cache_control_
class AnthropicCacheControlHook(CustomPromptManagement):
@staticmethod
def _request_value(request_kwargs: object, key: str) -> object:
request_mapping: Final = _validated_object_mapping(request_kwargs)
if request_mapping is None:
return None
return request_mapping.get(key)
@staticmethod
def _request_user_agent(request_kwargs: object) -> str | None:
proxy_server_request: Final = AnthropicCacheControlHook._request_value(request_kwargs, "proxy_server_request")
proxy_server_request_mapping: Final = _validated_object_mapping(proxy_server_request)
if proxy_server_request_mapping is None:
return None
headers: Final = proxy_server_request_mapping.get("headers")
headers_mapping: Final = _validated_object_mapping(headers)
if headers_mapping is None:
return None
user_agent: Final = next(
(value for key, value in headers_mapping.items() if isinstance(key, str) and key.lower() == "user-agent"),
None,
)
return user_agent if isinstance(user_agent, str) else None
@staticmethod
def _request_system(request_kwargs: object) -> str | list[object] | None:
system: Final = AnthropicCacheControlHook._request_value(request_kwargs, "system")
if isinstance(system, str):
return system
return _validated_object_list(system)
def get_chat_completion_prompt(
self,
model: str,
@ -520,12 +569,13 @@ class AnthropicCacheControlHook(CustomPromptManagement):
points: Sequence[CacheControlInjectionPoint],
messages: list[AllMessageValues],
tools: list[object] | None,
cache_control: object,
model: str,
custom_llm_provider: str | None,
api_base: object,
prompt_cache_options: object,
) -> Sequence[Mapping[str, object]] | None:
if AnthropicCacheControlHook._should_stand_down(points, messages, None, tools):
if AnthropicCacheControlHook._should_stand_down(points, messages, None, tools, cache_control):
return None
return AnthropicCacheControlHook._stamped_with_dialect(
points, model, custom_llm_provider, api_base, prompt_cache_options
@ -561,6 +611,7 @@ class AnthropicCacheControlHook(CustomPromptManagement):
messages: list[AllMessageValues],
system: str | list | None,
tools: list | None,
cache_control: object = None,
) -> bool:
"""Whether configured injection points must yield to client-set cache_control.
@ -573,13 +624,14 @@ class AnthropicCacheControlHook(CustomPromptManagement):
"""
if all(point.get("_litellm_judged") for point in points):
return False
return AnthropicCacheControlHook._request_has_cache_control(messages, system, tools)
return AnthropicCacheControlHook._request_has_cache_control(messages, system, tools, cache_control)
@staticmethod
def _request_has_cache_control(
messages: list[AllMessageValues],
system: str | list | None,
tools: list | None = None,
cache_control: object = None,
) -> bool:
"""Return True if the request already carries any client-supplied cache_control.
@ -591,6 +643,8 @@ class AnthropicCacheControlHook(CustomPromptManagement):
carry the mark either at the top level (Anthropic shape) or nested under
``function`` (OpenAI shape); the Anthropic chat transform accepts both.
"""
if cache_control is not None:
return True
if AnthropicCacheControlHook.count_request_cache_breakpoints(messages, system) > 0:
return True
if tools is not None:
@ -612,6 +666,8 @@ class AnthropicCacheControlHook(CustomPromptManagement):
custom_llm_provider: str | None,
tools: list | None = None,
enable_prompt_caching: bool | None = None,
cache_control: object = None,
request_kwargs: object = None,
) -> list[CacheControlInjectionPoint]:
"""Default breakpoints when ``litellm.enable_anthropic_prompt_caching`` is on.
@ -649,7 +705,12 @@ class AnthropicCacheControlHook(CustomPromptManagement):
if not supports_prompt_caching(model=model, custom_llm_provider=provider):
return []
if AnthropicCacheControlHook._request_has_cache_control(messages, system, tools):
if AnthropicCacheControlHook._request_has_cache_control(messages, system, tools, cache_control):
return []
if is_claude_code_one_shot_subagent_request(
messages, system, tools, AnthropicCacheControlHook._request_user_agent(request_kwargs)
):
return []
control: Final = AnthropicCacheControlHook._default_control()
@ -665,6 +726,7 @@ class AnthropicCacheControlHook(CustomPromptManagement):
models: Iterable[str],
tools: list[AllToolParamValues] | None = None,
enable_prompt_caching: bool | None = None,
request_kwargs: object = None,
) -> list[AllMessageValues]:
"""Return the messages auto prompt caching will send, default breakpoints included.
@ -681,11 +743,13 @@ class AnthropicCacheControlHook(CustomPromptManagement):
for candidate in (
AnthropicCacheControlHook.get_default_injection_points(
messages=messages,
system=None,
model=model,
custom_llm_provider=None,
tools=tools,
enable_prompt_caching=enable_prompt_caching,
system=AnthropicCacheControlHook._request_system(request_kwargs),
cache_control=AnthropicCacheControlHook._request_value(request_kwargs, "cache_control"),
request_kwargs=request_kwargs,
)
for model in models
)
@ -730,6 +794,7 @@ class AnthropicCacheControlHook(CustomPromptManagement):
non_default_params["cache_control_injection_points"],
messages,
tools,
non_default_params.get("cache_control"),
model,
custom_llm_provider,
api_base,
@ -747,6 +812,8 @@ class AnthropicCacheControlHook(CustomPromptManagement):
custom_llm_provider=custom_llm_provider,
tools=tools,
enable_prompt_caching=enable_prompt_caching,
cache_control=non_default_params.get("cache_control"),
request_kwargs=non_default_params,
)
if points:
non_default_params["cache_control_injection_points"] = points
@ -853,10 +920,13 @@ class AnthropicCacheControlHook(CustomPromptManagement):
enable_prompt_caching: Final = cast( # cast-ok: kwargs is untyped; key stamped as bool by the proxy
bool | None, kwargs.pop("enable_prompt_caching", None)
)
cache_control: Final = kwargs.get("cache_control")
configured: Final = cast( # cast-ok: kwargs is untyped; this key only holds the documented injection-point list
list[CacheControlInjectionPoint] | None, kwargs.pop("cache_control_injection_points", None)
)
if configured and AnthropicCacheControlHook._should_stand_down(configured, typed_messages, system, tools):
if configured and AnthropicCacheControlHook._should_stand_down(
configured, typed_messages, system, tools, cache_control
):
return messages, system
injection_points: list[CacheControlInjectionPoint] = configured or []
if not injection_points and model is not None:
@ -867,6 +937,8 @@ class AnthropicCacheControlHook(CustomPromptManagement):
model=model,
custom_llm_provider=custom_llm_provider,
enable_prompt_caching=enable_prompt_caching,
cache_control=cache_control,
request_kwargs=kwargs,
)
if not injection_points:
return messages, system

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

@ -67,6 +67,96 @@ _BEDROCK_VERSION_SUFFIX_RE: Final = re.compile(r"-v\d+(?::\d+)?$")
_INFERENCE_PROFILE_MINOR_RE: Final = re.compile(r":\d+$")
_DATED_RELEASE_SUFFIX_RE: Final = re.compile(r"-\d{8}$")
_DOTTED_VERSION_RE: Final = re.compile(r"(\d)\.(\d)")
_CLAUDE_CODE_BILLING_HEADER_PREFIX: Final = "x-anthropic-billing-header:"
_CLAUDE_CODE_OBJECT_MAPPING_ADAPTER: Final = TypeAdapter(dict[object, object])
_CLAUDE_CODE_OBJECT_LIST_ADAPTER: Final = TypeAdapter(list[object])
def is_claude_code_user_agent(user_agent: str) -> bool:
return user_agent.startswith("claude-cli/")
def _validated_claude_code_mapping(value: object) -> dict[object, object] | None:
try:
return _CLAUDE_CODE_OBJECT_MAPPING_ADAPTER.validate_python(value)
except ValidationError:
return None
def _validated_claude_code_list(value: object) -> list[object] | None:
try:
return _CLAUDE_CODE_OBJECT_LIST_ADAPTER.validate_python(value)
except ValidationError:
return None
def _claude_code_billing_fields(text: str) -> tuple[tuple[str, str], ...] | None:
stripped: Final = text.strip()
if "\n" in stripped or "\r" in stripped or not stripped.startswith(_CLAUDE_CODE_BILLING_HEADER_PREFIX):
return None
fields: Final = tuple(
field
for raw_field in stripped.removeprefix(_CLAUDE_CODE_BILLING_HEADER_PREFIX).split(";")
if (field := raw_field.strip())
)
if not fields or any("=" not in field for field in fields):
return None
parsed_fields: Final = tuple(
(parts[0].strip(), parts[1].strip()) for field in fields for parts in (field.split("=", 1),)
)
if any(not key or not value for key, value in parsed_fields):
return None
return parsed_fields
def _claude_code_billing_texts(system: object) -> tuple[str, ...] | None:
if isinstance(system, str):
return (system,)
blocks: Final = _validated_claude_code_list(system)
if blocks is None:
return None
block_mappings: Final = tuple(_validated_claude_code_mapping(block) for block in blocks)
if any(block is None for block in block_mappings):
return None
text_values: Final = tuple(
block.get("text") for block in block_mappings if block is not None and block.get("type") == "text"
)
if len(text_values) != len(blocks) or any(not isinstance(text, str) for text in text_values):
return None
meaningful_text: Final = tuple(text for text in text_values if isinstance(text, str) and text.strip())
return meaningful_text or None
def _is_claude_code_subagent_billing_system(system: object) -> bool:
billing_texts: Final = _claude_code_billing_texts(system)
if billing_texts is None:
return False
billing_fields: Final = tuple(
fields for text in billing_texts if (fields := _claude_code_billing_fields(text)) is not None
)
if len(billing_fields) != len(billing_texts):
return False
subagent_values: Final = tuple(
value for fields in billing_fields for key, value in fields if key == "cc_is_subagent"
)
return subagent_values == ("true",)
def is_claude_code_one_shot_subagent_request(
messages: list[AllMessageValues],
system: object,
tools: object,
user_agent: str | None,
) -> bool:
only_message: Final = _validated_claude_code_mapping(messages[0]) if len(messages) == 1 else None
return (
user_agent is not None
and is_claude_code_user_agent(user_agent)
and not tools
and only_message is not None
and only_message.get("role") == "user"
and _is_claude_code_subagent_billing_system(system)
)
def _strip_bedrock_id_suffixes(model: str) -> str:

8
litellm/proxy/litellm_pre_call_utils.py
View file

@ -789,12 +789,6 @@ def apply_missing_session_id_policy(
)
def is_claude_code_user_agent(user_agent: str) -> bool:
"""Claude Code identifies itself as ``claude-cli/<version> ...``; the IDE
extensions and the Agent SDK run through the same CLI and share that prefix."""
return user_agent.startswith("claude-cli/")
def is_codex_user_agent(user_agent: str) -> bool:
"""Codex builds its user agent as ``<originator>/<version> ...`` and ships
several first-party originators: ``codex-tui``, ``codex_cli_rs``,
@ -811,6 +805,8 @@ def should_auto_drop_params_for_agentic_cli(user_agent: str, data: dict, proxy_c
requests routed to providers that reject them. An explicit drop_params
from the caller or in the operator's ``litellm_settings`` always wins
over this default."""
from litellm.llms.anthropic.common_utils import is_claude_code_user_agent
if not (is_claude_code_user_agent(user_agent) or is_codex_user_agent(user_agent)):
return False
if "drop_params" in data:

1
litellm/router_utils/pre_call_checks/prompt_caching_deployment_check.py
View file

@ -90,6 +90,7 @@ class PromptCachingDeploymentCheck(CustomLogger):
enable_prompt_caching=(
request_kwargs.get("enable_prompt_caching") is True if request_kwargs is not None else None
),
request_kwargs=request_kwargs,
)
model_id_dict: Final = await prompt_cache.async_get_model_id(

237
tests/test_litellm/integrations/test_anthropic_cache_control_hook.py
View file

@ -1777,6 +1777,224 @@ class TestEnableAnthropicPromptCaching:
assert messages == before
class TestClaudeCodeOneShotAutoCaching:
BILLING_TEXT = "x-anthropic-billing-header: cc_version=2.1.263; cc_entrypoint=cli; cc_is_subagent=true;"
BILLING_SYSTEM = [{"type": "text", "text": BILLING_TEXT}]
MESSAGES = [{"role": "user", "content": [{"type": "text", "text": "unique fetched document"}]}]
@staticmethod
def _kwargs(configured=None):
kwargs = {
"litellm_metadata": {},
"proxy_server_request": {
"headers": {
"user-agent": "claude-cli/2.1.263 (external, cli)",
"x-app": "cli-bg",
}
},
}
if configured is not None:
kwargs["cache_control_injection_points"] = configured
return kwargs
@pytest.mark.parametrize(
"system",
[
BILLING_TEXT,
BILLING_SYSTEM,
[*BILLING_SYSTEM, {"type": "text", "text": " "}],
[
*BILLING_SYSTEM,
{"type": "text", "text": "x-anthropic-billing-header: cc_version=2.1.263; cc_entrypoint=cli;"},
],
],
ids=["string", "text_block", "whitespace_block", "multiple_billing_blocks"],
)
@pytest.mark.parametrize("tools", [None, []], ids=["absent_tools", "empty_tools"])
def test_skips_defaults_and_attribution_for_one_shot_subagent(self, monkeypatch, system, tools):
monkeypatch.setattr(litellm, "enable_anthropic_prompt_caching", True)
messages = copy.deepcopy(self.MESSAGES)
kwargs = self._kwargs()
result_messages, result_system = AnthropicCacheControlHook.maybe_inject_cache_control(
messages,
copy.deepcopy(system),
kwargs,
model="claude-sonnet-4-5",
custom_llm_provider="anthropic",
tools=tools,
)
assert result_messages == self.MESSAGES
assert result_system == system
assert "litellm_gateway_injected_cache" not in kwargs["litellm_metadata"]
def test_user_agent_header_lookup_is_case_insensitive(self, monkeypatch):
monkeypatch.setattr(litellm, "enable_anthropic_prompt_caching", True)
kwargs = self._kwargs()
user_agent = kwargs["proxy_server_request"]["headers"].pop("user-agent")
kwargs["proxy_server_request"]["headers"]["User-Agent"] = user_agent
result_messages, result_system = AnthropicCacheControlHook.maybe_inject_cache_control(
copy.deepcopy(self.MESSAGES),
copy.deepcopy(self.BILLING_SYSTEM),
kwargs,
model="claude-sonnet-4-5",
custom_llm_provider="anthropic",
)
assert result_messages == self.MESSAGES
assert result_system == self.BILLING_SYSTEM
def test_router_affinity_skips_string_billing_system(self, monkeypatch):
monkeypatch.setattr(litellm, "enable_anthropic_prompt_caching", True)
messages = copy.deepcopy(self.MESSAGES)
kwargs = self._kwargs()
kwargs["system"] = self.BILLING_TEXT
result = AnthropicCacheControlHook.messages_with_default_injections(
messages=messages,
models=("claude-sonnet-4-5",),
request_kwargs=kwargs,
)
assert result == messages
@pytest.mark.parametrize(
"headers,system",
[
("not-a-mapping", BILLING_SYSTEM),
(
{"user-agent": "claude-cli/2.1.263 (external, cli)"},
[{"type": "text", "text": "x-anthropic-billing-header: malformed"}],
),
({"user-agent": "claude-cli/2.1.263 (external, cli)"}, None),
({"user-agent": "claude-cli/2.1.263 (external, cli)"}, ["not-a-mapping"]),
(
{"user-agent": "claude-cli/2.1.263 (external, cli)"},
[{"type": "image", "text": BILLING_TEXT}],
),
],
ids=["malformed_headers", "malformed_billing", "missing_system", "malformed_block", "non_text_block"],
)
def test_malformed_untrusted_context_keeps_defaults(self, monkeypatch, headers, system):
monkeypatch.setattr(litellm, "enable_anthropic_prompt_caching", True)
points = AnthropicCacheControlHook.get_default_injection_points(
messages=copy.deepcopy(self.MESSAGES),
system=copy.deepcopy(system),
model="claude-sonnet-4-5",
custom_llm_provider="anthropic",
request_kwargs={"proxy_server_request": {"headers": headers}},
)
assert len(points) == 2
def test_message_without_role_keeps_defaults(self, monkeypatch):
monkeypatch.setattr(litellm, "enable_anthropic_prompt_caching", True)
points = AnthropicCacheControlHook.get_default_injection_points(
messages=[{"content": "missing role"}],
system=copy.deepcopy(self.BILLING_SYSTEM),
model="claude-sonnet-4-5",
custom_llm_provider="anthropic",
request_kwargs=self._kwargs(),
)
assert len(points) == 2
@pytest.mark.parametrize(
"messages,system,tools",
[
(
MESSAGES,
BILLING_SYSTEM,
[{"name": "WebFetch", "description": "fetch", "input_schema": {"type": "object"}}],
),
(MESSAGES, [*BILLING_SYSTEM, {"type": "text", "text": "Explore the repository"}], None),
(
[
{"role": "user", "content": "first turn"},
{"role": "assistant", "content": "reply"},
*MESSAGES,
],
BILLING_SYSTEM,
None,
),
],
ids=["tools", "real_system", "history"],
)
def test_keeps_defaults_for_reusable_subagents(self, monkeypatch, messages, system, tools):
monkeypatch.setattr(litellm, "enable_anthropic_prompt_caching", True)
kwargs = self._kwargs()
result_messages, result_system = AnthropicCacheControlHook.maybe_inject_cache_control(
copy.deepcopy(messages),
copy.deepcopy(system),
kwargs,
model="claude-sonnet-4-5",
custom_llm_provider="anthropic",
tools=copy.deepcopy(tools),
)
assert AnthropicCacheControlHook.count_request_cache_breakpoints(result_messages, result_system) == 2
assert kwargs["litellm_metadata"]["litellm_gateway_injected_cache"] == ""
@pytest.mark.parametrize(
"user_agent,system",
[
("anthropic-sdk-python/0.75.0", BILLING_SYSTEM),
(
"claude-cli/2.1.263 (external, cli)",
[
{
"type": "text",
"text": f"{BILLING_TEXT}\nadditional system instructions",
}
],
),
(
"claude-cli/2.1.263 (external, cli)",
[
{
"type": "text",
"text": "x-anthropic-billing-header: cc_version=2.1.263; cc_is_subagent=false;",
}
],
),
],
ids=["different_client", "appended_instructions", "not_a_subagent"],
)
def test_ambiguous_or_unmatched_signals_fail_open(self, monkeypatch, user_agent, system):
monkeypatch.setattr(litellm, "enable_anthropic_prompt_caching", True)
kwargs = self._kwargs()
kwargs["proxy_server_request"]["headers"]["user-agent"] = user_agent
result_messages, result_system = AnthropicCacheControlHook.maybe_inject_cache_control(
copy.deepcopy(self.MESSAGES),
copy.deepcopy(system),
kwargs,
model="claude-sonnet-4-5",
custom_llm_provider="anthropic",
)
assert AnthropicCacheControlHook.count_request_cache_breakpoints(result_messages, result_system) == 2
def test_explicit_injection_points_remain_authoritative(self, monkeypatch):
monkeypatch.setattr(litellm, "enable_anthropic_prompt_caching", True)
kwargs = self._kwargs([{"location": "message", "role": "user"}])
result_messages, _ = AnthropicCacheControlHook.maybe_inject_cache_control(
copy.deepcopy(self.MESSAGES),
copy.deepcopy(self.BILLING_SYSTEM),
kwargs,
model="claude-sonnet-4-5",
custom_llm_provider="anthropic",
)
assert result_messages[0]["content"][-1]["cache_control"] == {"type": "ephemeral"}
class TestPerKeyEnablePromptCaching:
"""Per-request enable_prompt_caching override (stamped from key metadata) with the global flag off."""
@ -1977,6 +2195,25 @@ class TestConfiguredInjectionPointsStandDown:
_, result_sys = self._inject(copy.deepcopy(self.V1_MESSAGES), kwargs)
assert result_sys == [{"type": "text", "text": "sys", "cache_control": {"type": "ephemeral"}}]
@pytest.mark.parametrize(
"configured",
[None, CONFIGURED],
ids=["automatic_defaults", "configured_points"],
)
def test_v1_messages_stands_down_for_root_cache_control(self, monkeypatch, configured):
monkeypatch.setattr(litellm, "enable_anthropic_prompt_caching", True)
root_cache_control = {"type": "ephemeral"}
kwargs = {"cache_control": root_cache_control, "litellm_metadata": {}}
if configured is not None:
kwargs["cache_control_injection_points"] = copy.deepcopy(configured)
result_messages, result_system = self._inject(copy.deepcopy(self.V1_MESSAGES), kwargs)
assert result_messages == self.V1_MESSAGES
assert result_system == "sys"
assert kwargs["cache_control"] is root_cache_control
assert "litellm_gateway_injected_cache" not in kwargs["litellm_metadata"]
def test_v1_messages_reentry_flow_preserves_tool_config_remainder(self):
"""The advisor interceptor re-enters anthropic_messages() with the outer
request's kwargs and post-injection messages. The first pass applies the

24
tests/test_litellm/llms/anthropic/test_anthropic_common_utils.py
View file

@ -26,6 +26,30 @@ FAKE_REGULAR_KEY = "sk-ant-api03-regular-key-for-testing-123456789"
FAKE_AUTH_TOKEN = "sk-ant-aut01-fake-auth-token-for-testing-123456789"
@pytest.mark.parametrize(
"messages,system,expected",
[
([{"role": "user", "content": "hi"}], "x-anthropic-billing-header: cc_is_subagent=true;", True),
([{"role": "user", "content": "hi"}], "x-anthropic-billing-header: =junk; cc_is_subagent=true;", False),
([{"role": "user", "content": "hi"}], "x-anthropic-billing-header: cc_version=; cc_is_subagent=true;", False),
([{"role": "user", "content": "hi"}], "x-anthropic-billing-header: malformed", False),
([{"content": "missing role"}], "x-anthropic-billing-header: cc_is_subagent=true;", False),
(["not-a-mapping"], "x-anthropic-billing-header: cc_is_subagent=true;", False),
([{"role": "user", "content": "hi"}], ["not-a-mapping"], False),
([{"role": "user", "content": "hi"}], None, False),
],
)
def test_is_claude_code_one_shot_subagent_request(messages, system, expected):
from litellm.llms.anthropic.common_utils import is_claude_code_one_shot_subagent_request
assert is_claude_code_one_shot_subagent_request(
messages=messages,
system=system,
tools=None,
user_agent="claude-cli/2.1.263 (external, cli)",
) is expected
class TestOptionallyHandleAnthropicOAuth:
"""Tests for optionally_handle_anthropic_oauth function."""

61
tests/test_litellm/router_utils/pre_call_checks/test_prompt_caching_deployment_check.py
View file

@ -332,6 +332,67 @@ async def test_per_request_enable_prompt_caching_reaches_the_affinity_key(monkey
assert filtered == [deployments[1]]
@pytest.mark.asyncio
async def test_claude_code_one_shot_subagent_does_not_reuse_an_auto_injected_affinity_key(monkeypatch):
monkeypatch.setattr(litellm, "enable_anthropic_prompt_caching", True)
cache = DualCache()
check = PromptCachingDeploymentCheck(cache=cache)
deployments = _deployments(AUTO_CACHING_MODEL, AUTO_CACHING_MODEL)
messages = cast(List[AllMessageValues], [{"role": "user", "content": "unique " * 3000}])
request_kwargs = {
"system": [
{
"type": "text",
"text": "x-anthropic-billing-header: cc_version=2.1.263; cc_is_subagent=true;",
}
],
"proxy_server_request": {"headers": {"user-agent": "claude-cli/2.1.263 (external, cli)"}},
}
auto_injected_messages = AnthropicCacheControlHook.messages_with_default_injections(
messages=messages,
models=(AUTO_CACHING_MODEL,),
)
assert auto_injected_messages != messages
await PromptCachingCache(cache=cache).async_add_model_id(
model_id="dep-2", messages=auto_injected_messages, tools=None
)
filtered = await check.async_filter_deployments(
model=MODEL_GROUP_ALIAS,
healthy_deployments=deployments,
messages=messages,
request_kwargs=request_kwargs,
)
assert filtered == deployments
@pytest.mark.asyncio
async def test_root_cache_control_does_not_reuse_an_auto_injected_affinity_key(monkeypatch):
monkeypatch.setattr(litellm, "enable_anthropic_prompt_caching", True)
cache = DualCache()
check = PromptCachingDeploymentCheck(cache=cache)
deployments = _deployments(AUTO_CACHING_MODEL, AUTO_CACHING_MODEL)
messages = _auto_caching_messages()
auto_injected_messages = AnthropicCacheControlHook.messages_with_default_injections(
messages=messages,
models=(AUTO_CACHING_MODEL,),
)
assert auto_injected_messages != messages
await PromptCachingCache(cache=cache).async_add_model_id(
model_id="dep-2", messages=auto_injected_messages, tools=None
)
filtered = await check.async_filter_deployments(
model=MODEL_GROUP_ALIAS,
healthy_deployments=deployments,
messages=messages,
request_kwargs={"cache_control": {"type": "ephemeral"}},
)
assert filtered == deployments
@pytest.mark.asyncio
async def test_tool_marked_cache_control_keeps_routing_off_another_requests_prefix(monkeypatch, local_model_cost_map):
"""