diff --git a/litellm/llms/anthropic/chat/guardrail_translation/handler.py b/litellm/llms/anthropic/chat/guardrail_translation/handler.py index 18ae2530ed5..c781531f563 100644 --- a/litellm/llms/anthropic/chat/guardrail_translation/handler.py +++ b/litellm/llms/anthropic/chat/guardrail_translation/handler.py @@ -661,12 +661,11 @@ class AnthropicMessagesHandler(BaseTranslation): ] ) if not scan_only_tool_results: - retained_names: Final = frozenset( - name - for t in (data.get("tools") or ()) - for name in anthropic_tool_names(t) # comprehension-ok: flattens tool names across tools + data_dict: Final[dict[str, object]] = cast(dict[str, object], data) # cast-ok: request dict + retained_names: Final = frozenset( # comprehension-ok: flattens tool names across tools + name for t in list_field(data_dict, "tools") for name in anthropic_tool_names(t) ) - self._strip_disallowed_tool_additions(data, retained_names) + self._strip_disallowed_tool_additions(data_dict, retained_names) guardrailed_structured_messages: Final = guardrailed_inputs.get("structured_messages") if ( @@ -945,65 +944,64 @@ class AnthropicMessagesHandler(BaseTranslation): top_names: Final = [ # comprehension-ok: flattens tool names across tools name for tool in list_field(data, "tools") for name in anthropic_tool_names(tool) ] - messages: Final = data.get("messages") - if not isinstance(messages, Sequence): + messages: Final = list_field(data, "messages") + if not messages: return top_names extra_names: Final[list[str]] = [] # mutable-ok: accumulator for m in messages: - if isinstance(m, Mapping): - content: Final = m.get("content") - if isinstance(content, Sequence) and not isinstance(content, (str, bytes)): - for block in content: - if isinstance(block, Mapping) and block.get("type") == "tool_addition": - tool: Final = extract_tool_from_addition_block(block) - if tool is not None: - for name in anthropic_tool_names(tool): - if name not in top_names and name not in extra_names: - extra_names.append(name) + m_map = as_str_mapping(m) + if m_map is not None: + for block in list_field(m_map, "content"): + block_map = as_str_mapping(block) + if block_map is not None and block_map.get("type") == "tool_addition": + tool = extract_tool_from_addition_block(block_map) + if tool is not None: + for name in anthropic_tool_names(tool): + if name not in top_names and name not in extra_names: + extra_names.append(name) return [*top_names, *extra_names] + @staticmethod + def _is_disallowed_addition_block(block: object, retained_names: frozenset[str]) -> bool: + block_map: Final = as_str_mapping(block) + if block_map is None or block_map.get("type") != "tool_addition": + return False + tool: Final = extract_tool_from_addition_block(block_map) + if tool is None: + return True + tool_names: Final = anthropic_tool_names(tool) + if not tool_names: + return True + return any(name not in retained_names for name in tool_names) + @classmethod def _strip_disallowed_tool_additions(cls, data: Mapping[str, object], retained_names: frozenset[str]) -> None: - messages: Final = data.get("messages") - if not isinstance(messages, list): + if not isinstance(data, dict): return + messages_val: Final = data.get("messages") + if not isinstance(messages_val, list): + return + messages: Final = cast(Sequence[object], messages_val) # cast-ok: messages list cleaned_messages: Final[list[object]] = [] # mutable-ok: accumulator for m in messages: - if not isinstance(m, dict): + m_map = as_str_mapping(m) + if m_map is None: cleaned_messages.append(m) continue - content: Final = m.get("content") - if not isinstance(content, list): + content_val = m_map.get("content") + if not isinstance(content_val, list): cleaned_messages.append(m) continue - had_disallowed_addition: Final = any( - isinstance(block, dict) - and block.get("type") == "tool_addition" - and ( - not (tool := extract_tool_from_addition_block(block)) - or not (tool_names := anthropic_tool_names(tool)) - or not all(name in retained_names for name in tool_names) - ) - for block in content - ) + content: Sequence[object] = cast(Sequence[object], content_val) # cast-ok: content list + had_disallowed_addition = any(cls._is_disallowed_addition_block(block, retained_names) for block in content) if not had_disallowed_addition: cleaned_messages.append(m) continue - remaining_blocks: Final[list[object]] = [ # mutable-ok: accumulator - block - for block in content - if not ( - isinstance(block, dict) - and block.get("type") == "tool_addition" - and ( - not (tool := extract_tool_from_addition_block(block)) - or not (tool_names := anthropic_tool_names(tool)) - or not all(name in retained_names for name in tool_names) - ) - ) + remaining_blocks: list[object] = [ # mutable-ok: accumulator + block for block in content if not cls._is_disallowed_addition_block(block, retained_names) ] if remaining_blocks: - new_m: Final[dict[str, object]] = dict(m) # mutable-ok: message clone + new_m: dict[str, object] = dict(m_map) # mutable-ok: message clone new_m["content"] = remaining_blocks cleaned_messages.append(new_m) data["messages"] = cleaned_messages # rebind-ok: update request in place diff --git a/litellm/llms/anthropic/pass_through/adapters/transformation.py b/litellm/llms/anthropic/pass_through/adapters/transformation.py index 867bc734877..328ba31d59e 100644 --- a/litellm/llms/anthropic/pass_through/adapters/transformation.py +++ b/litellm/llms/anthropic/pass_through/adapters/transformation.py @@ -8,7 +8,7 @@ from typing import TYPE_CHECKING, Any, Final, Literal, TypeAlias, TypeVar, cast from pydantic import JsonValue, TypeAdapter import litellm -from litellm.litellm_core_utils.core_helpers import normalize_drop_params +from litellm.litellm_core_utils.core_helpers import as_str_mapping, normalize_drop_params from litellm.litellm_core_utils.hidden_params import get_hidden_params from litellm.llms.anthropic.pass_through.convertible_blocks import tool_result_text_part from litellm.llms.anthropic.pass_through.utils import ( @@ -16,7 +16,10 @@ from litellm.llms.anthropic.pass_through.utils import ( json_copy_or_original, prompt_cache_key_from_user_id, ) -from litellm.llms.base_llm.guardrail_translation.utils import extract_tool_from_addition_block +from litellm.llms.base_llm.guardrail_translation.utils import ( + extract_tool_from_addition_block, + list_field, +) from litellm.responses.utils import auto_reasoning_summary # OpenAI has a 64-character limit for function/tool names @@ -476,7 +479,9 @@ class LiteLLMAnthropicMessagesAdapter: ) -> list[AllMessageValues]: new_messages: Final[list[AllMessageValues]] = [] _, cleaned_input_messages = self._extract_and_apply_midturn_tool_changes(messages) - replayable_messages: Final = strip_encrypted_reasoning_blocks_from_anthropic_messages(cleaned_input_messages) + replayable_messages: Final = strip_encrypted_reasoning_blocks_from_anthropic_messages( + cast(Sequence[dict[str, object]], cleaned_input_messages) # cast-ok: anthropic messages shape + ) leading_count: Final = next( (i for i, m in enumerate(replayable_messages) if not is_system_role_message(m)), len(replayable_messages), @@ -1232,10 +1237,19 @@ class LiteLLMAnthropicMessagesAdapter: if k not in translatable_params: # pass remaining params as is new_kwargs[k] = v + @staticmethod + def _adapter_tool_name(tool: object) -> str | None: + tool_map: Final = as_str_mapping(tool) + if tool_map is not None: + name_val: Final = tool_map.get("name") + if isinstance(name_val, str) and name_val: + return name_val + return None + def _extract_and_apply_midturn_tool_changes( self, messages: Sequence[Mapping[str, object] | AllAnthropicPassThroughMessageValues], - initial_tools: Sequence[AllAnthropicToolsValues] | None = None, + initial_tools: Sequence[object] | None = None, ) -> tuple[tuple[AllAnthropicToolsValues, ...] | None, tuple[AllAnthropicPassThroughMessageValues, ...]]: """ Extract and apply mid-conversation tool changes (tool_addition, tool_removal) @@ -1248,57 +1262,67 @@ class LiteLLMAnthropicMessagesAdapter: - cleaned_messages: Messages tuple with tool_addition and tool_removal blocks stripped, and carrier messages left without content dropped. """ - active_tools: tuple[AllAnthropicToolsValues, ...] = tuple(initial_tools or ()) # rebind-ok: tool updates + active_tools: tuple[AllAnthropicToolsValues, ...] = tuple( # rebind-ok: tool updates + cast( # cast-ok: initial tools + AllAnthropicToolsValues, t + ) + for t in (initial_tools or ()) + ) had_tool_changes: bool = False # rebind-ok: state flag cleaned_messages: Final[list[AllAnthropicPassThroughMessageValues]] = [] # mutable-ok: accumulator for m in messages: - if not isinstance(m, Mapping): - cleaned_messages.append(m) # pyright: ignore[reportArgumentType] # non-mapping passthrough + m_map = as_str_mapping(m) + if m_map is None: + cleaned_messages.append(cast(AllAnthropicPassThroughMessageValues, m)) # cast-ok: msg continue - content: Final = m.get("content") + content = m_map.get("content") if not isinstance(content, Sequence) or isinstance(content, (str, bytes)): - msg_val: Final = cast(AllAnthropicPassThroughMessageValues, m) # cast-ok: msg + msg_val = cast(AllAnthropicPassThroughMessageValues, m) # cast-ok: msg cleaned_messages.append(msg_val) continue - remaining_blocks: Final[list[object]] = [] # mutable-ok: accumulator - for block in content: - if not isinstance(block, Mapping): + content_seq: Sequence[object] = cast(Sequence[object], content) # cast-ok: content blocks + remaining_blocks: list[object] = [] # mutable-ok: accumulator + for block in content_seq: + block_map = as_str_mapping(block) + if block_map is None: remaining_blocks.append(block) continue - btype: Final = block.get("type") + btype = block_map.get("type") if btype == "tool_addition": had_tool_changes = True - raw_tool: Final = extract_tool_from_addition_block(block) - if isinstance(raw_tool, Mapping) and raw_tool.get("name"): - tool_name: Final = str(raw_tool["name"]) - if not any(isinstance(t, Mapping) and t.get("name") == tool_name for t in active_tools): - added_tool: Final = cast(AllAnthropicToolsValues, dict(raw_tool)) # cast-ok: tool + raw_tool = extract_tool_from_addition_block(block_map) + if raw_tool is not None: + tool_name = self._adapter_tool_name(raw_tool) + if tool_name is not None and not any( + self._adapter_tool_name(t) == tool_name for t in active_tools + ): + added_tool = cast(AllAnthropicToolsValues, dict(raw_tool)) # cast-ok: tool active_tools = (*active_tools, added_tool) elif btype == "tool_removal": had_tool_changes = True - rem_name: Final = ( - block.get("tool_name") - or block.get("name") - or (block.get("tool", {}).get("name") if isinstance(block.get("tool"), Mapping) else None) - or (block.get("tool") if isinstance(block.get("tool"), str) else None) + tool_obj = block_map.get("tool") + tool_map = as_str_mapping(tool_obj) + rem_name = ( + block_map.get("tool_name") + or block_map.get("name") + or (tool_map.get("name") if tool_map is not None else None) + or (tool_obj if isinstance(tool_obj, str) else None) ) if rem_name: - rem_name_str: Final = str(rem_name) - active_tools = tuple( - t for t in active_tools if not (isinstance(t, Mapping) and t.get("name") == rem_name_str) - ) + rem_name_str = str(rem_name) + active_tools = tuple(t for t in active_tools if self._adapter_tool_name(t) != rem_name_str) else: remaining_blocks.append(block) if remaining_blocks: - new_m: Final[dict[str, object]] = dict(m) # mutable-ok: message clone + new_m: dict[str, object] = dict(m_map) # mutable-ok: message clone new_m["content"] = remaining_blocks - new_msg: Final = cast(AllAnthropicPassThroughMessageValues, new_m) # cast-ok: msg + new_msg = cast(AllAnthropicPassThroughMessageValues, new_m) # cast-ok: msg cleaned_messages.append(new_msg) elif not had_tool_changes: - clean_msg: Final = cast(AllAnthropicPassThroughMessageValues, m) # cast-ok: msg + clean_msg = cast(AllAnthropicPassThroughMessageValues, m) # cast-ok: msg cleaned_messages.append(clean_msg) if not had_tool_changes and initial_tools is None: @@ -1332,25 +1356,27 @@ class LiteLLMAnthropicMessagesAdapter: list[AllAnthropicPassThroughMessageValues], anthropic_message_request["messages"], ) + request_dict: Final[dict[str, object]] = dict(anthropic_message_request) # mutable-ok: clone + initial_tools_val: Final = list_field(request_dict, "tools") if "tools" in request_dict else None ( active_tools, cleaned_messages, ) = self._extract_and_apply_midturn_tool_changes( messages=messages_list, - initial_tools=anthropic_message_request.get("tools"), - ) - request_for_translation: Final[AnthropicMessagesRequest] = cast( # cast-ok: anthropic request type - AnthropicMessagesRequest, dict(anthropic_message_request) + initial_tools=initial_tools_val, ) if active_tools is not None: if active_tools: - request_for_translation["tools"] = list(active_tools) + request_dict["tools"] = list(active_tools) else: - request_for_translation.pop("tools", None) + request_dict.pop("tools", None) + request_for_translation: Final[AnthropicMessagesRequest] = cast( # cast-ok: anthropic request type + AnthropicMessagesRequest, request_dict + ) new_messages: Final = self.translate_anthropic_messages_to_openai( messages=list(cleaned_messages), - model=request_for_translation.get("model"), + model=anthropic_message_request.get("model"), custom_llm_provider=custom_llm_provider, preserve_midturn_system=preserve_midturn_system, ) diff --git a/litellm/llms/base_llm/guardrail_translation/utils.py b/litellm/llms/base_llm/guardrail_translation/utils.py index a4e587c36f3..7fcac135f45 100644 --- a/litellm/llms/base_llm/guardrail_translation/utils.py +++ b/litellm/llms/base_llm/guardrail_translation/utils.py @@ -6,6 +6,7 @@ from typing import Final, TypeVar, cast # noqa: TID251 # a rebuilt chat row ha from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError +from litellm.litellm_core_utils.core_helpers import as_str_mapping from litellm.types.llms.anthropic_messages.anthropic_response import AnthropicUsage from litellm.types.llms.openai import AllMessageValues, ResponseAPIUsage @@ -321,21 +322,22 @@ def anthropic_tool_name(tool: object) -> str | None: def extract_tool_from_addition_block(block: object) -> Mapping[str, object] | None: """Extract raw tool dict from an Anthropic tool_addition block.""" - if not isinstance(block, Mapping): + mapping: Final = as_str_mapping(block) + if mapping is None: return None - tool_data: Final = block.get("tool") - if isinstance(tool_data, Mapping): + tool_data: Final = as_str_mapping(mapping.get("tool")) + if tool_data is not None: if ( tool_data.get("type") == "tool_definition" and "definition" in tool_data - and isinstance(tool_data["definition"], Mapping) + and (definition := as_str_mapping(tool_data.get("definition"))) is not None ): - return tool_data["definition"] + return definition return tool_data - if "definition" in block and isinstance(block["definition"], Mapping): - return block["definition"] - if isinstance(block.get("tool_definition"), Mapping): - return block["tool_definition"] + if "definition" in mapping and (definition := as_str_mapping(mapping.get("definition"))) is not None: + return definition + if (tool_def := as_str_mapping(mapping.get("tool_definition"))) is not None: + return tool_def return None