From 036eec9e06365bc7f97f035ed596fa7fa1033ec8 Mon Sep 17 00:00:00 2001 From: Tan Nguyen Date: Fri, 9 Oct 2026 02:31:27 +0700 Subject: [PATCH] fix(anthropic): type the block and role reads so the basedpyright budget stays flat on main --- .../anthropic_cache_control_hook.py | 22 ++++++++++++++----- 1 file changed, 16 insertions(+), 6 deletions(-) diff --git a/litellm/integrations/anthropic_cache_control_hook.py b/litellm/integrations/anthropic_cache_control_hook.py index 7ddcd796072..c2c46c2195b 100644 --- a/litellm/integrations/anthropic_cache_control_hook.py +++ b/litellm/integrations/anthropic_cache_control_hook.py @@ -230,7 +230,7 @@ def _accepts_prompt_cache_breakpoint(block: object) -> bool: return isinstance(block, dict) and block.get("type") in OPENAI_PROMPT_CACHE_BREAKPOINT_BLOCK_TYPES -def _index_of_block_accepting_cache_control(content: list[object], on_a_tool_message: bool) -> int | None: +def _index_of_block_accepting_cache_control(content: Sequence[object], on_a_tool_message: bool) -> int | None: """Last block a marker written here reaches the provider on, searched from the end. An empty text block is replaced by a placeholder unless it sits on a tool message, @@ -240,14 +240,22 @@ def _index_of_block_accepting_cache_control(content: list[object], on_a_tool_mes block = content[index] if not isinstance(block, dict): continue - if block.get("type") in ANTHROPIC_BLOCK_TYPES_WITHOUT_CACHE_CONTROL: + fields = cast(dict[str, object], block) # cast-ok: a runtime dict whose value types are not known here + if fields.get("type") in ANTHROPIC_BLOCK_TYPES_WITHOUT_CACHE_CONTROL: continue - if block.get("type") == "text" and not block.get("text") and not on_a_tool_message: + if fields.get("type") == "text" and not fields.get("text") and not on_a_tool_message: continue return index return None +def _role_of(message: object) -> object: + """The ``role`` of a message dict, or None for anything else.""" + if not isinstance(message, dict): + return None + return cast(dict[str, object], message).get("role") # cast-ok: a runtime dict whose value types are not known here + + def _message_accepts_cache_control(message: object) -> bool: """Whether a marker written on this message reaches the provider.""" if not isinstance(message, dict): @@ -258,7 +266,8 @@ def _message_accepts_cache_control(message: object) -> bool: if isinstance(content, str): return content != "" or on_a_tool_message if isinstance(content, list): - return _index_of_block_accepting_cache_control(content, on_a_tool_message) is not None + blocks: Final = cast(Sequence[object], content) # cast-ok: a runtime list whose block types are not known here + return _index_of_block_accepting_cache_control(blocks, on_a_tool_message) is not None return False @@ -580,7 +589,7 @@ class AnthropicCacheControlHook(CustomPromptManagement): candidates: Final = tuple( index for index, message in enumerate(messages) - if targetted_role is None or message.get("role") == targetted_role + if targetted_role is None or _role_of(message) == targetted_role ) free: Final = frozenset( index @@ -650,8 +659,9 @@ class AnthropicCacheControlHook(CustomPromptManagement): message["cache_control"] = control # 2. list of objects - the last block that accepts a marker, per Anthropic spec elif isinstance(message_content, list): + blocks: Final = cast(Sequence[object], message_content) # cast-ok: TypedDict blocks read as objects target_index: Final = _index_of_block_accepting_cache_control( - message_content, on_a_tool_message=message.get("role") == "tool" + blocks, on_a_tool_message=_role_of(message) == "tool" ) if target_index is not None: message_content[target_index]["cache_control"] = control # pyright: ignore[reportGeneralTypeIssues] # loose runtime dict