feat(lint): add LIT013 flagging *-ok suppressions that suppress nothing and remove the 240 stale ones (#42793)

litellm/_logging.py

@@ -631,9 +631,9 @@ class LevelRoutingStreamHandler(logging.StreamHandler):
         )
         preferred: Final = sys.stdout if is_stdout_record else sys.stderr
         if preferred is None or getattr(preferred, "closed", False):
-            self.stream = sys.stderr  # rebind-ok: fall back to the pre-fix stream rather than raising per record
+            self.stream = sys.stderr
         else:
-            self.stream = preferred  # rebind-ok: StreamHandler.emit writes self.stream under the handler lock
+            self.stream = preferred
         super().emit(record)
 
 

litellm/completion_extras/litellm_responses_transformation/transformation.py

@@ -191,8 +191,6 @@ def _as_chat_reasoning_items(
 ) -> list[ChatCompletionReasoningItem] | None:
     if not reasoning_items:
         return None
-    # cast-ok: _BuiltReasoningItem is the structural shape ChatCompletionReasoningItem
-    # describes, and TypedDict invariance is what stops the two from unifying here.
     return cast(list[ChatCompletionReasoningItem], list(reasoning_items))
 
 
@@ -1370,7 +1368,7 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
         if tool_call_index_map is None:
             return output_index
         if output_index not in tool_call_index_map:
-            tool_call_index_map[output_index] = len(tool_call_index_map)  # mutable-ok: per-stream accumulator state
+            tool_call_index_map[output_index] = len(tool_call_index_map)
         return tool_call_index_map[output_index]
 
     @staticmethod

litellm/experimental_mcp_client/client.py

@@ -764,7 +764,7 @@ class MCPClient:
                 follow_redirects=True,
                 event_hooks=MappingProxyType(
                     {"response": [capture_upstream_error_response], "request": [guard] if guard else []}
-                ),  # mutable-ok: httpx types require lists of hooks
+                ),
             )
 
         return factory
@@ -921,9 +921,7 @@ class MCPClient:
         with anyio.fail_after(max(self.timeout, MCP_TOOL_LISTING_TIMEOUT)):
             for page_index in range(MCP_TOOL_LISTING_MAX_PAGES):
                 try:
-                    page = await fetch_page(  # rebind-ok: each SDK page replaces the previous one
-                        None if cursor is None else PaginatedRequestParams(cursor=cursor)
-                    )
+                    page = await fetch_page(None if cursor is None else PaginatedRequestParams(cursor=cursor))
                 except MCPError as error:
                     if page_index > 0 and error.error.code == METHOD_NOT_FOUND:
                         raise RuntimeError("MCP list operation became unavailable during pagination") from error

litellm/integrations/custom_guardrail.py

@@ -1641,5 +1641,5 @@ def log_guardrail_information(func):
             return async_wrapper(*args, **kwargs)
         return sync_wrapper(*args, **kwargs)
 
-    vars(wrapper)[LOGS_GUARDRAIL_INFORMATION_MARKER] = True  # rebind-ok: stamps the wrapper this call just built
+    vars(wrapper)[LOGS_GUARDRAIL_INFORMATION_MARKER] = True
     return wrapper

litellm/integrations/newrelic/newrelic_metrics.py

@@ -366,9 +366,7 @@ class NewRelicMetricsLogger(CustomBatchLogger):
         error to keep the client-error path (drop) distinct from 5xx (retry)."""
         payload: Final = build_metric_payload(records=batch, window_start=window_start, now=time.time())
         try:
-            status = (
-                await self.async_send_compressed_data(payload)
-            ).status_code  # rebind-ok: reassigned from the raised HTTPStatusError below
+            status = (await self.async_send_compressed_data(payload)).status_code
         except HTTPStatusError as e:
             status = e.response.status_code
         except Exception as e:  # noqa: BLE001  # transport/network failure re-queues the batch

litellm/integrations/otel/plumbing/providers.py

@@ -353,7 +353,7 @@ class _DrainPool:
 
     def _drain_until_closed(self) -> None:
         while True:
-            processor: SpanProcessor | None = self._pending.get()  # rebind-ok: loop variable
+            processor: SpanProcessor | None = self._pending.get()
             if processor is None:
                 return
             _shutdown_quietly(processor)
@@ -572,7 +572,7 @@ class TenantFanOutSpanProcessor(SpanProcessor):
                 span, destination.span_scope
             ):
                 continue
-            processor = self._acquire(destination)  # rebind-ok: loop variable; pyright forbids Final in a loop
+            processor = self._acquire(destination)
             if processor is None:
                 continue
             try:

litellm/integrations/otel/presets/destinations.py

@@ -151,7 +151,7 @@ def destination_for(
     endpoint, protocol = resolved
     return OtelDestination(
         endpoint=endpoint,
-        headers=MappingProxyType(dict(headers)),  # mutable-ok: MappingProxyType needs a concrete mapping to wrap
+        headers=MappingProxyType(dict(headers)),
         resource_attributes=MappingProxyType({"service.name": service_name}) if service_name else _NO_ATTRS,
         callback_name=callback_name,
         protocol=protocol,

litellm/integrations/prometheus.py

@@ -131,7 +131,7 @@ def _paginated_table(repository: BaseRepository[_TableRowT]) -> _PaginatedPrisma
     """View a repository's prisma table through the pagination surface budget metrics need."""
     return cast(
         _PaginatedPrismaTable[_TableRowT],
-        repository.table,  # cast-ok: prisma rows carry the budget columns the domain model declares
+        repository.table,
     )
 
 

litellm/integrations/shadow_eval_logger.py

@@ -873,7 +873,6 @@ class ShadowEvalLogger(CustomLogger):
                 await prisma.db.litellm_shadowevalattempt.group_by(
                     by=["job_id"],
                     count=True,
-                    # mutable-ok: Prisma aggregate spec
                     sum={"judge_cost": True, "shadow_cost": True, "shadow_classifier_cost": True},
                     where={"job_id": {"in": [str(record.id) for record in records]}},  # mutable-ok: Prisma filter
                 )
@@ -901,7 +900,7 @@ class ShadowEvalLogger(CustomLogger):
                 {target: tuple(job for _, job in group) for target, group in groupby(by_target, key=itemgetter(0))}
             )
             await self._jobs_cache.async_set_cache(_JOBS_CACHE_KEY, jobs)
-            self._job_starts = {}  # rebind-ok: new generation, counts absorbed into the fill
+            self._job_starts = {}
             return jobs
         except Exception as e:  # noqa: BLE001  # a DB blip must never break request logging
             verbose_logger.debug("shadow_eval: active-job read failed: %s", e)
@@ -1033,7 +1032,7 @@ class ShadowEvalLogger(CustomLogger):
                         real_cache_hit=real_cache_hit,
                         control_tier=control_tier,
                         shadow_params=shadow_params,
-                        parent_metadata=MappingProxyType(dict(request_metadata)),  # mutable-ok: frozen snapshot
+                        parent_metadata=MappingProxyType(dict(request_metadata)),
                     )
                 ).add_done_callback(self._release_shadow_slot)
         except Exception as e:  # noqa: BLE001  # logging hooks must never fail the request
@@ -1347,7 +1346,7 @@ class ShadowEvalLogger(CustomLogger):
             {
                 "role": "user",
                 "content": _judge_user_prompt(conversation, response_a, response_b, _tool_definitions_text(tools)),
-            },  # mutable-ok: SDK message
+            },
         ]
         try:
             response: Final = await judge_acompletion(

litellm/interactions/background_cost_polling.py

@@ -79,9 +79,7 @@ async def _fetch_interaction(context: BackgroundInteractionPollContext) -> Inter
         custom_llm_provider=context.custom_llm_provider,
         api_key=context.api_key,
         api_base=context.api_base,
-        **{
-            "no-log": True
-        },  # mutable-ok: "no-log" is not a valid identifier, so it can only be passed through a mapping
+        **{"no-log": True},
     )
 
 

litellm/litellm_core_utils/core_helpers.py

@@ -764,4 +764,4 @@ def set_response_cost_in_hidden_params(response: _CarriesHiddenParams, cost: flo
         **(additional_headers if isinstance(additional_headers, Mapping) else _NO_HEADERS),
         RESPONSE_COST_HEADER: cost,
     }
-    hidden_params["additional_headers"] = merged  # rebind-ok: the caller's record is the point
+    hidden_params["additional_headers"] = merged

litellm/litellm_core_utils/get_supported_openai_params.py

@@ -32,9 +32,7 @@ def get_supported_openai_params(
     - None if unmapped
     """
     if not custom_llm_provider:
-        custom_llm_provider = declared_authenticating_provider(
-            model
-        )  # rebind-ok: resolving would run the provider's OAuth flow
+        custom_llm_provider = declared_authenticating_provider(model)
     if not custom_llm_provider:
         try:
             custom_llm_provider = litellm.get_llm_provider(model=model)[1]

litellm/litellm_core_utils/json_fragment_accumulator.py

@@ -21,20 +21,18 @@ class JSONFragmentAccumulator:
 
     def __init__(self) -> None:
         self._chunks: list[str] = []  # mutable-ok: O(1) append; string concat would copy the buffer each time
-        self._buffer: str = (
-            ""  # mutable-ok: lazily materialized join of _chunks, rebuilt only when _chunks is non-empty
-        )
-        self._offset: int = 0  # mutable-ok: cursor past already-consumed values; avoids re-slicing on every pop
-        self._could_close: bool = False  # mutable-ok: cached heuristic; rescanning past fragments was itself O(n^2)
+        self._buffer: str = ""
+        self._offset: int = 0
+        self._could_close: bool = False
 
     def __bool__(self) -> bool:
         return bool(self._chunks) or self._offset < len(self._buffer)
 
     def append(self, fragment: str) -> None:
-        self._chunks.append(fragment)  # mutable-ok: see __init__
+        self._chunks.append(fragment)
         stripped: Final = fragment.rstrip()
         if stripped:
-            self._could_close = stripped[-1] in ("}", "]")  # mutable-ok: see __init__
+            self._could_close = stripped[-1] in ("}", "]")
 
     def could_close_json(self) -> bool:
         """
@@ -50,8 +48,8 @@ class JSONFragmentAccumulator:
         if not self._chunks:
             return
         unconsumed: Final = self._buffer[self._offset :]
-        self._buffer = unconsumed + "".join(self._chunks)  # mutable-ok: merge pending fragments, once per append batch
-        self._offset = 0  # mutable-ok: see __init__
+        self._buffer = unconsumed + "".join(self._chunks)
+        self._offset = 0
         self._chunks = []  # mutable-ok: see __init__
 
     def pop_next_value(self) -> tuple[bool, object]:
@@ -69,7 +67,7 @@ class JSONFragmentAccumulator:
         while start < length and self._buffer[start].isspace():
             start += 1
         if start >= length:
-            self._offset = start  # mutable-ok: see __init__
+            self._offset = start
             return False, None
         decoder: Final = json.JSONDecoder()
         try:
@@ -77,11 +75,11 @@ class JSONFragmentAccumulator:
         except json.JSONDecodeError:
             return False, None
         decoded, end_index = cast("tuple[object, int]", raw_value)  # cast-ok: raw_decode returns tuple[Any, int]
-        self._offset = end_index  # mutable-ok: see __init__
+        self._offset = end_index
         if self._offset >= len(self._buffer):
-            self._buffer = ""  # mutable-ok: see __init__
-            self._offset = 0  # mutable-ok: see __init__
-            self._could_close = False  # mutable-ok: buffer is empty, nothing can close
+            self._buffer = ""
+            self._offset = 0
+            self._could_close = False
         return True, decoded
 
     def snapshot(self) -> str:
@@ -91,7 +89,7 @@ class JSONFragmentAccumulator:
     def set(self, value: str) -> None:
         """Replace the buffer's contents with a single fragment."""
         self._chunks = []  # mutable-ok: see __init__
-        self._buffer = value  # mutable-ok: see __init__
-        self._offset = 0  # mutable-ok: see __init__
+        self._buffer = value
+        self._offset = 0
         stripped: Final = value.rstrip()
-        self._could_close = bool(stripped) and stripped[-1] in ("}", "]")  # mutable-ok: see __init__
+        self._could_close = bool(stripped) and stripped[-1] in ("}", "]")

litellm/litellm_core_utils/litellm_logging.py

@@ -6667,7 +6667,7 @@ def get_standard_logging_object_payload(
                     "version": 3,
                     "status": "unknown",
                     "reason": "pending_projection",
-                }  # mutable-ok: spend-log JSON serialization requires plain mappings
+                }
                 if captured_baseline is not None
                 else (
                     {  # mutable-ok: spend-log JSON serialization requires plain mappings

litellm/litellm_core_utils/llm_response_utils/convert_dict_to_response.py

@@ -372,9 +372,7 @@ from collections import defaultdict
 
 
 def _handle_invalid_parallel_tool_calls(
-    tool_calls: list[
-        ChatCompletionMessageToolCall | ChatCompletionMessageCustomToolCall
-    ],  # mutable-ok: patched in place via slice assignment
+    tool_calls: list[ChatCompletionMessageToolCall | ChatCompletionMessageCustomToolCall],
 ):
     """
     Handle hallucinated parallel tool call from openai - https://community.openai.com/t/model-tries-to-call-unknown-function-multi-tool-use-parallel/490653

litellm/litellm_core_utils/prompt_templates/common_utils.py

@@ -208,7 +208,7 @@ def _content_parts_contain_image(parts: Sequence[object]) -> bool:
     for _ in range(_IMAGE_SCAN_MAX_DEPTH):
         if any(isinstance(part, Mapping) and part.get("type") in _IMAGE_CONTENT_PART_TYPES for part in frontier):
             return True
-        frontier = tuple(  # rebind-ok: depth-bounded frontier walk
+        frontier = tuple(
             nested
             for part in frontier
             if isinstance(part, Mapping)
@@ -2020,7 +2020,7 @@ def _anthropic_content_lists(messages: Sequence[object]) -> Iterator[object]:
 def _strip_encrypted_reasoning_from_blocks(content: object) -> None:
     blocks: Final = cast(list[object], content)  # cast-ok: narrowed by the caller's isinstance
     kept: Final = tuple(block for block in blocks if not is_encrypted_reasoning_block(block))
-    blocks[:] = kept  # rebind-ok: shared with fallback snapshot
+    blocks[:] = kept
 
 
 def _reasoning_replay_group_key(indexed_block: tuple[int, Mapping[str, object]]) -> str:

litellm/litellm_core_utils/provider_affinity.py

@@ -83,7 +83,7 @@ def get_stable_session_id(litellm_params: object | None) -> str | None:
     return None
 
 
-def add_provider_affinity_header(  # mutable-ok: downstream handlers add auth and signing headers
+def add_provider_affinity_header(
     headers: Mapping[str, object], litellm_params: object | None
 ) -> dict[str, object]:  # mutable-ok: downstream handlers add auth and signing headers
     header_name: Final = _get_provider_affinity_header_name(litellm_params)

litellm/litellm_core_utils/streaming_chunk_builder_utils.py

@@ -475,9 +475,7 @@ class ChunkProcessor:
 
     def get_combined_tool_content(
         self, tool_call_chunks: Sequence["_ToolCallChunk"]
-    ) -> list[
-        ChatCompletionMessageToolCall | ChatCompletionMessageCustomToolCall
-    ]:  # mutable-ok: assigned verbatim to Message.tool_calls, a list field
+    ) -> list[ChatCompletionMessageToolCall | ChatCompletionMessageCustomToolCall]:
         tool_calls_list: list[
             ChatCompletionMessageToolCall | ChatCompletionMessageCustomToolCall
         ] = []  # mutable-ok: see return type

litellm/llms/anthropic/chat/guardrail_translation/handler.py

@@ -199,9 +199,7 @@ def _write_back_system_block(system: object, block_idx: int, response: str) -> N
         return
     text_blocks: Final = tuple(block for block in system if isinstance(block, dict) and block.get("type") == "text")
     if block_idx < len(text_blocks):
-        text_blocks[block_idx]["text"] = (
-            response  # mutable-ok: guardrails rewrite the caller's request payload in place
-        )
+        text_blocks[block_idx]["text"] = response
 
 
 def _write_back_message_text(message: _WritableMessage, target: MessageTextTarget, response: str) -> None:
@@ -211,22 +209,16 @@ def _write_back_message_text(message: _WritableMessage, target: MessageTextTarge
     match target:
         case MessageContentTarget():
             if isinstance(content, str):
-                message["content"] = response  # mutable-ok: guardrails rewrite the caller's request payload in place
+                message["content"] = response
         case ContentBlockTextTarget(content_idx=content_idx):
             if isinstance(content, list):
-                content[content_idx]["text"] = (
-                    response  # mutable-ok: guardrails rewrite the caller's request payload in place
-                )
+                content[content_idx]["text"] = response
         case ToolResultStringTarget(content_idx=content_idx):
             if isinstance(content, list):
-                content[content_idx]["content"] = (
-                    response  # mutable-ok: guardrails rewrite the caller's request payload in place
-                )
+                content[content_idx]["content"] = response
         case ToolResultBlockTextTarget(content_idx=content_idx, block_idx=block_idx):
             if isinstance(content, list):
-                content[content_idx]["content"][block_idx]["text"] = (
-                    response  # mutable-ok: guardrails rewrite the caller's request payload in place
-                )
+                content[content_idx]["content"][block_idx]["text"] = response
         case _:
             assert_never(target)
 
@@ -248,9 +240,9 @@ def _write_back_tool_use(
     block: Final = content[target.content_idx] if isinstance(content, list) else None
     if not isinstance(block, dict):
         return
-    block["input"] = rewritten_input  # mutable-ok: guardrails rewrite the caller's request payload in place
+    block["input"] = rewritten_input
     if shape.name is not None and shape.name != block.get("name"):
-        block["name"] = shape.name  # mutable-ok: guardrails rewrite the caller's request payload in place
+        block["name"] = shape.name
 
 
 @dataclass(frozen=True, slots=True)
@@ -603,13 +595,9 @@ class AnthropicMessagesHandler(BaseTranslation):
             *(item for one_message in extracted for item in one_message.scanned),
         )
         texts_to_check: Final = [item.text for item in scanned]  # mutable-ok: GenericGuardrailAPIInputs takes list[str]
-        images_to_check: Final = [
-            image for one_message in extracted for image in one_message.images
-        ]  # mutable-ok: GenericGuardrailAPIInputs takes list[str]
+        images_to_check: Final = [image for one_message in extracted for image in one_message.images]
         scanned_tool_calls: Final = tuple(item for one_message in extracted for item in one_message.tool_calls)
-        tool_calls_to_check: Final = [
-            item.tool_call for item in scanned_tool_calls
-        ]  # mutable-ok: GenericGuardrailAPIInputs takes list[ChatCompletionToolCallChunk]
+        tool_calls_to_check: Final = [item.tool_call for item in scanned_tool_calls]
         pre_guardrail_tool_calls: Final = _tool_call_shapes(tool_calls_to_check)
 
         # Step 2: Apply guardrail to all texts and tool calls in batch
@@ -697,9 +685,7 @@ class AnthropicMessagesHandler(BaseTranslation):
 
         return data
 
-    def _hoisted_top_level_system_message(
-        self, data: dict
-    ) -> AllMessageValues | None:  # mutable-ok: API message payload
+    def _hoisted_top_level_system_message(self, data: dict) -> AllMessageValues | None:
         """Return the system message produced by translating the top-level prompt."""
         system: Final = data.get("system")
         if not system:
@@ -736,7 +722,7 @@ class AnthropicMessagesHandler(BaseTranslation):
         if isinstance(content, str):
             return (
                 {"role": "system", "content": content} if content else None  # mutable-ok: API message payload
-            )  # mutable-ok: API message payload
+            )
         if not isinstance(content, list):
             return None
         blocks: Final[list[dict[str, object]]] = []  # mutable-ok: API message payload
@@ -749,14 +735,14 @@ class AnthropicMessagesHandler(BaseTranslation):
             anthropic_block: dict[str, object] = {  # mutable-ok: API message payload
                 "type": "text",
                 "text": text,
-            }  # mutable-ok: API message payload
+            }
             cache_control = block.get("cache_control")
             if cache_control:
                 anthropic_block["cache_control"] = deepcopy(cache_control)
             blocks.append(anthropic_block)
         return (
             {"role": "system", "content": blocks} if blocks else None  # mutable-ok: API message payload
-        )  # mutable-ok: API message payload
+        )
 
     @staticmethod
     def _fold_leading_systems_into_top_level(
@@ -1098,9 +1084,7 @@ class AnthropicMessagesHandler(BaseTranslation):
             match item.target:
                 case SystemStringTarget():
                     if isinstance(data.get("system"), str):
-                        data["system"] = (
-                            guardrail_response  # mutable-ok: guardrails rewrite the caller's request payload in place
-                        )
+                        data["system"] = guardrail_response
                 case SystemBlockTextTarget(block_idx=block_idx):
                     _write_back_system_block(data.get("system"), block_idx, guardrail_response)
                 case (

litellm/llms/anthropic/common_utils.py

@@ -1591,7 +1591,7 @@ def _flatten_web_search_results_in_message(message: object) -> object:
     return {**message, "content": [b for b in rewritten if b is not None]}  # mutable-ok: JSON wire format
 
 
-def flatten_unencrypted_web_search_results_in_anthropic_messages(  # mutable-ok: as sibling sanitizers
+def flatten_unencrypted_web_search_results_in_anthropic_messages(
     messages: list[Any],
 ) -> list[Any]:
     """

litellm/llms/anthropic/experimental_pass_through/messages/response_cache.py

@@ -88,9 +88,7 @@ class AnthropicMessagesStreamCacheWriter:
 
         try:
             events: Final = _split_sse_events(collected_stream.decode("utf-8"))
-            cached_payload: Final = {
-                CACHED_STREAM_EVENTS_KEY: events
-            }  # mutable-ok: cache backends serialize plain dicts
+            cached_payload: Final = {CACHED_STREAM_EVENTS_KEY: events}
             await litellm.cache.async_add_cache(
                 cached_payload,
                 dynamic_cache_object=self.caching_handler.dual_cache,

litellm/llms/anthropic/experimental_pass_through/messages/streaming_iterator.py

@@ -186,7 +186,7 @@ def _sse_event(event_type: str, payload: Mapping[str, object]) -> bytes:
 
 
 def _incomplete_stream_error_sse_event() -> bytes:
-    return _sse_event(  # mutable-ok: one-shot JSON payload, never mutated after construction
+    return _sse_event(
         "error",
         {"type": "error", "error": {"type": "api_error", "message": INCOMPLETE_STREAM_ERROR_MESSAGE}},
     )

litellm/llms/anthropic/experimental_pass_through/responses_adapters/transformation.py

@@ -148,13 +148,11 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
         if isinstance(content, str):
             return (
                 [{"type": "input_text", "text": content}] if content else []  # mutable-ok: API message payload
-            )  # mutable-ok: API message payload
+            )
         if not isinstance(content, list):
             return []  # mutable-ok: API message payload
         return [  # mutable-ok: API message payload
-            with_prompt_cache_breakpoint(
-                {"type": "input_text", "text": text}, block.get("prompt_cache_breakpoint")
-            )  # mutable-ok: API message payload
+            with_prompt_cache_breakpoint({"type": "input_text", "text": text}, block.get("prompt_cache_breakpoint"))
             for block in content
             if isinstance(block, dict) and block.get("type") == "text" and (text := block.get("text"))  # pyright: ignore[reportUnnecessaryIsInstance]  # untrusted client payload
         ]

litellm/llms/azure/passthrough/transformation.py

@@ -59,9 +59,7 @@ def logged_responses_stream(all_chunks: Sequence[str], logging_obj: Logging) ->
     terminal_event: Final = OpenAIResponsesAPIConfig.parse_terminal_event_from_stream_chunks(all_chunks=all_chunks)
     if terminal_event is None:
         return None
-    logging_obj.call_type = (
-        RESPONSES_RELAY_SHAPE.call_type.value
-    )  # rebind-ok: routes cost calculation to the relayed shape's pricing path
+    logging_obj.call_type = RESPONSES_RELAY_SHAPE.call_type.value
     return terminal_event
 
 

litellm/llms/azure_ai/image_generation/flux_transformation.py

@@ -73,9 +73,7 @@ class AzureFoundryFluxImageGenerationConfig(GPTImageGenerationConfig):
         normalized_model: Final = model.lower().replace(".", "-").replace("_", "-")
         return "flux-2-flex" if "flux-2-flex" in normalized_model else "flux-2-pro"
 
-    def get_supported_openai_params(  # mutable-ok: inherited config contract returns a list
-        self, model: str
-    ) -> list[OpenAIImageGenerationOptionalParams]:
+    def get_supported_openai_params(self, model: str) -> list[OpenAIImageGenerationOptionalParams]:
         if not self.is_flux2_model(model):
             return super().get_supported_openai_params(model)
         return [  # mutable-ok: BaseImageGenerationConfig requires a list

litellm/llms/base_llm/passthrough/transformation.py

@@ -95,9 +95,7 @@ def logged_relay_shape(
         parsed: Final = shape.parse(body)
     except ValidationError:
         return None
-    logging_obj.call_type = (
-        shape.call_type.value
-    )  # rebind-ok: routes cost calculation to the relayed shape's pricing path
+    logging_obj.call_type = shape.call_type.value
     return parsed
 
 

litellm/llms/bedrock/messages/mantle_transformation.py

@@ -45,7 +45,7 @@ def _move_betas_into_header(request: Mapping[str, object], headers: dict[str, st
     if betas:
         headers["anthropic-beta"] = ",".join(betas)  # rebind-ok: the handler signs and sends this same dict
         return
-    headers.pop("anthropic-beta", None)  # rebind-ok: a caller header Mantle rejects in full must not reach it
+    headers.pop("anthropic-beta", None)
 
 
 class AmazonMantleMessagesConfig(AmazonAnthropicClaudeMessagesConfig):

litellm/llms/bedrock/realtime/handler.py

@@ -489,9 +489,7 @@ class BedrockRealtime(BaseAWSLLM):
                 parsed_client_message = _parse_client_message(message)
                 is_session_update = _json_str(parsed_client_message.get("type")) == "session.update"
                 if is_session_update:
-                    client_ws.scope[BEDROCK_REALTIME_PENDING_SESSION_UPDATE_SCOPE_KEY] = (
-                        message  # rebind-ok: scope outlives the attempt
-                    )
+                    client_ws.scope[BEDROCK_REALTIME_PENDING_SESSION_UPDATE_SCOPE_KEY] = message
 
                 transformed_messages = transformation_config.transform_realtime_request(
                     message=message,

litellm/llms/fal_ai/image_edit/flux_lora_depth_transformation.py

@@ -27,7 +27,7 @@ class FalAIFluxLoraDepthEditConfig(FalAIImageEditConfig):
     def get_supported_openai_params(self, model: str) -> list:  # mutable-ok: base class contract returns a list
         return list(SUPPORTED_OPENAI_PARAMS)  # mutable-ok: base class contract returns a list
 
-    def map_openai_params(  # mutable-ok: base class contract returns a dict
+    def map_openai_params(
         self,
         image_edit_optional_params: ImageEditOptionalRequestParams,
         model: str,
@@ -63,9 +63,7 @@ class FalAIFluxLoraDepthEditConfig(FalAIImageEditConfig):
         if len(images) > 1:
             raise ValueError(f"{FLUX_LORA_DEPTH_ENDPOINT} accepts exactly one control image")
         provider_params: Final[Mapping[str, object]] = MappingProxyType(
-            {
-                key: value for key, value in image_edit_optional_request_params.items() if key != "mask"
-            }  # mutable-ok: frozen by MappingProxyType
+            {key: value for key, value in image_edit_optional_request_params.items() if key != "mask"}
         )
         request_body: Final[dict[str, object]] = {  # mutable-ok: base class contract returns a dict
             "prompt": prompt,

litellm/llms/fal_ai/image_edit/transformation.py

@@ -84,7 +84,7 @@ class FalAIImageEditConfig(BaseImageEditConfig):
     def get_supported_openai_params(self, model: str) -> list:  # mutable-ok: base class contract returns a list
         return list(SUPPORTED_OPENAI_PARAMS)  # mutable-ok: base class contract returns a list
 
-    def map_openai_params(  # mutable-ok: base class contract returns a dict
+    def map_openai_params(
         self,
         image_edit_optional_params: ImageEditOptionalRequestParams,
         model: str,
@@ -146,9 +146,7 @@ class FalAIImageEditConfig(BaseImageEditConfig):
             MappingProxyType({"mask_url": to_data_url(mask)}) if mask is not None else MappingProxyType({})
         )
         provider_params: Final[Mapping[str, object]] = MappingProxyType(
-            {
-                key: value for key, value in image_edit_optional_request_params.items() if key != "mask"
-            }  # mutable-ok: frozen by MappingProxyType
+            {key: value for key, value in image_edit_optional_request_params.items() if key != "mask"}
         )
         request_body: Final[dict[str, object]] = {  # mutable-ok: base class contract returns a dict
             "prompt": prompt,

litellm/llms/fal_ai/image_generation/gpt_image_2_transformation.py

@@ -101,12 +101,10 @@ class FalAIGPTImage2Config(FalAIBaseConfig):
         endpoint: Final[str] = model if model.startswith(self.MODEL_PREFIX) else f"{self.MODEL_PREFIX}{model}"
         return f"{base_url}/{endpoint}"
 
-    def get_supported_openai_params(  # mutable-ok: base class contract returns a list
-        self, model: str
-    ) -> list[OpenAIImageGenerationOptionalParams]:
+    def get_supported_openai_params(self, model: str) -> list[OpenAIImageGenerationOptionalParams]:
         return list(SUPPORTED_OPENAI_PARAMS)  # mutable-ok: base class contract returns a list
 
-    def map_openai_params(  # mutable-ok: base class contract returns a dict
+    def map_openai_params(
         self,
         non_default_params: Mapping[str, object],
         optional_params: Mapping[str, object],
@@ -138,7 +136,7 @@ class FalAIGPTImage2Config(FalAIBaseConfig):
             return map_gpt_image_quality(value, model)
         return value
 
-    def transform_image_generation_request(  # mutable-ok: base class contract returns a dict
+    def transform_image_generation_request(
         self,
         model: str,
         prompt: str,

litellm/llms/gigachat/authenticator.py

@@ -243,7 +243,7 @@ def _parse_token_response(response: httpx.Response) -> tuple[str, int]:
         )
 
     # expires_at is in milliseconds
-    expires_at: int  # rebind-ok: conditionally assigned from str or int
+    expires_at: int
     if isinstance(expires_at_raw, str):
         expires_at = int(expires_at_raw)  # rebind-ok: conditionally assigned from str or int
     else:

litellm/llms/gigachat/chat/streaming.py

@@ -30,7 +30,7 @@ class GigaChatModelResponseIterator:
 
     def chunk_parser(self, chunk: Mapping[str, object]) -> GenericStreamingChunk:
         """Parse a single streaming chunk from GigaChat."""
-        choices: Sequence = chunk.get("choices") or ()  # mutable-ok: tuple literal as default
+        choices: Sequence = chunk.get("choices") or ()
         if not choices:
             return GenericStreamingChunk(
                 text="",
@@ -56,7 +56,7 @@ class GigaChatModelResponseIterator:
         if chunk_finish_reason == "function_call" and isinstance(raw_function_call, Mapping) and raw_function_call:
             func_call: Final[Mapping[str, object]] = raw_function_call
             args_raw: Final[object] = func_call.get("arguments") or {}
-            args_str: str  # rebind-ok: conditionally assigned from dict or str
+            args_str: str
             if isinstance(args_raw, dict):
                 args_str = json.dumps(args_raw, ensure_ascii=False)  # rebind-ok: build from dict
             else:
@@ -80,10 +80,10 @@ class GigaChatModelResponseIterator:
             usage = convert_usage(validated_usage)
             _prompt_details: dict | None = (
                 usage.prompt_tokens_details.model_dump() if usage.prompt_tokens_details else None
-            )  # rebind-ok: conditional
+            )
             _completion_details: dict | None = (
                 usage.completion_tokens_details.model_dump() if usage.completion_tokens_details else None
-            )  # rebind-ok: conditional
+            )
             usage_block = ChatCompletionUsageBlock(  # pyright: ignore[reportCallIssue]  # TypedDict kwarg constructor
                 prompt_tokens=usage.prompt_tokens,
                 completion_tokens=usage.completion_tokens,

litellm/llms/mistral/batches/transformation.py

@@ -33,7 +33,7 @@ OpenAIBatchStatus: TypeAlias = Literal[
     "validating", "failed", "in_progress", "finalizing", "completed", "expired", "cancelling", "cancelled"
 ]
 
-_NO_HEADERS: Final[Mapping[str, str]] = MappingProxyType({})  # mutable-ok: frozen at module scope
+_NO_HEADERS: Final[Mapping[str, str]] = MappingProxyType({})
 _STATUS_MAP: Final[MappingProxyType[MistralBatchStatus, OpenAIBatchStatus]] = MappingProxyType(
     {
         "QUEUED": "validating",

litellm/llms/mongodb/vector_stores/transformation.py

@@ -197,7 +197,7 @@ class MongoDBVectorStoreConfig(BaseQueryEmbeddingVectorStoreConfig):
             **headers,
             "Authorization": f"Bearer {api_key}",
             "Content-Type": "application/json",
-        }  # mutable-ok: writable HTTP headers
+        }
 
     def get_complete_url(self, api_base: str | None, litellm_params: Mapping[str, object]) -> str:
         if not api_base:

litellm/llms/nvidia_nim/passthrough/transformation.py

@@ -110,7 +110,7 @@ class NvidiaNimPassthroughConfig(BasePassthroughConfig):
         return {
             **headers,
             "Authorization": f"Bearer {api_key}",
-        }  # mutable-ok: base class contract returns dict for httpx
+        }
 
     @staticmethod
     def get_api_base(api_base: str | None = None) -> str | None:

litellm/llms/nvidia_nim/rerank/transformation.py

@@ -215,7 +215,7 @@ class NvidiaNimRerankConfig(BaseRerankConfig):
             elif isinstance(doc, dict):
                 # Preserve only the structured passage fields supported by the
                 # selected rerank route.
-                supported_fields: NvidiaNimPassageObject = {}  # mutable-ok: assembling a request TypedDict
+                supported_fields: NvidiaNimPassageObject = {}
                 if "text" in self.SUPPORTED_PASSAGE_FIELDS and "text" in doc:
                     supported_fields["text"] = doc["text"]
                 if "image" in self.SUPPORTED_PASSAGE_FIELDS and "image" in doc:

litellm/llms/openai/chat/gpt_transformation.py

@@ -596,9 +596,7 @@ class OpenAIGPTConfig(BaseLLMModelInfo, BaseConfig):
         for choice in choices:
             ## HANDLE JSON MODE - anthropic returns single function call]
             tool_calls = choice["message"].get("tool_calls", None)
-            new_tool_calls: list[ChatCompletionMessageToolCall | ChatCompletionMessageCustomToolCall] | None = (
-                None  # mutable-ok: holds _handle_invalid_parallel_tool_calls' list; Message.__init__ expects list
-            )
+            new_tool_calls: list[ChatCompletionMessageToolCall | ChatCompletionMessageCustomToolCall] | None = None
             message_content = choice["message"].get("content", None)
             if tool_calls is not None:
                 _openai_tool_calls = []

litellm/llms/openai/openai.py

@@ -1427,9 +1427,7 @@ class OpenAIChatCompletion(BaseLLM, BaseOpenAILLM):
                 },
             )
 
-            request_data: Final = (  # mutable-ok: the OpenAI SDK takes the request body as a dict
-                {**data, "extra_headers": headers} if headers else data
-            )
+            request_data: Final = {**data, "extra_headers": headers} if headers else data
             response = await openai_aclient.images.generate(**request_data, timeout=timeout)
             stringified_response: Final = response.model_dump()
             ## LOGGING
@@ -1513,9 +1511,7 @@ class OpenAIChatCompletion(BaseLLM, BaseOpenAILLM):
             )
 
             ## COMPLETION CALL
-            request_data: Final = (  # mutable-ok: the OpenAI SDK takes the request body as a dict
-                {**data, "extra_headers": headers} if headers else data
-            )
+            request_data: Final = {**data, "extra_headers": headers} if headers else data
             _response: Final = openai_client.images.generate(**request_data, timeout=timeout)
 
             response: Final = _response.model_dump()

litellm/llms/snowflake/chat/transformation.py

@@ -118,7 +118,7 @@ def _convert_image_url_to_anthropic(block: Mapping[str, object]) -> object:
         anthropic_process_openai_file_message({"type": "file", "file": {"file_data": url}})
         if select_anthropic_content_block_type_for_file(_data_uri_media_type(url)) == "document"
         else create_anthropic_image_param(
-            image_url if isinstance(image_url, dict) else url,  # mutable-ok: caller's JSON block
+            image_url if isinstance(image_url, dict) else url,
             format=_image_url_field(image_url, "format"),
             is_bedrock_invoke=True,
         )
@@ -191,12 +191,8 @@ def _signed_thinking_blocks(msg: object) -> list[dict[str, object]]:  # mutable-
     ]
 
 
-def _clean_input_schema(schema: object) -> object:  # mutable-ok: JSON schema copy
-    return (
-        {key: value for key, value in schema.items() if key != "$schema"}
-        if isinstance(schema, Mapping)
-        else schema  # mutable-ok: JSON schema copy
-    )  # mutable-ok: JSON schema copy
+def _clean_input_schema(schema: object) -> object:
+    return {key: value for key, value in schema.items() if key != "$schema"} if isinstance(schema, Mapping) else schema
 
 
 class SnowflakeConfig(SnowflakeBaseConfig, OpenAIGPTConfig):
@@ -299,9 +295,7 @@ class SnowflakeConfig(SnowflakeBaseConfig, OpenAIGPTConfig):
                 )
         return anthropic_tools
 
-    def _extract_system_and_messages(  # mutable-ok: JSON wire messages
-        self, messages: list[AllMessageValues]
-    ) -> tuple[list[dict] | None, list[dict]]:
+    def _extract_system_and_messages(self, messages: list[AllMessageValues]) -> tuple[list[dict] | None, list[dict]]:
         """
         Split messages into system prompt and conversation turns for Anthropic format.
 
@@ -330,9 +324,7 @@ class SnowflakeConfig(SnowflakeBaseConfig, OpenAIGPTConfig):
                         {  # mutable-ok: JSON wire system block
                             "type": "text",
                             "text": block.get("text", ""),
-                            **(
-                                {"cache_control": block["cache_control"]} if "cache_control" in block else {}
-                            ),  # mutable-ok: JSON wire block
+                            **({"cache_control": block["cache_control"]} if "cache_control" in block else {}),
                         }
                         for block in content
                         if isinstance(block, Mapping) and block.get("type") == "text"
@@ -372,7 +364,7 @@ class SnowflakeConfig(SnowflakeBaseConfig, OpenAIGPTConfig):
                         ]
                         if isinstance(content, list)
                         else [*thinking_blocks, *([{"type": "text", "text": content}] if content else [])]
-                    )  # rebind-ok: loop-local normalized content
+                    )
                     conversation.append({"role": "assistant", "content": thinking_content})
                 else:
                     conversation.append({"role": "assistant", "content": content})
@@ -380,9 +372,7 @@ class SnowflakeConfig(SnowflakeBaseConfig, OpenAIGPTConfig):
                 tool_call_id_value = (
                     msg.get("tool_call_id", "") if isinstance(msg, dict) else getattr(msg, "tool_call_id", "")
                 )
-                tool_call_id = (
-                    tool_call_id_value if isinstance(tool_call_id_value, str) else ""
-                )  # rebind-ok: normalized loop value
+                tool_call_id = tool_call_id_value if isinstance(tool_call_id_value, str) else ""
                 tool_result_block = _convert_tool_result_to_anthropic(content, tool_call_id, msg_cache_control)
                 if (
                     conversation
@@ -395,13 +385,13 @@ class SnowflakeConfig(SnowflakeBaseConfig, OpenAIGPTConfig):
                 else:
                     conversation.append(
                         {"role": "user", "content": [tool_result_block]}  # mutable-ok: JSON wire message
-                    )  # mutable-ok: JSON wire message
+                    )
             else:
-                conversation.append(  # mutable-ok: JSON wire message
+                conversation.append(
                     {  # mutable-ok: JSON wire message
                         "role": role,
                         "content": _convert_image_url_blocks_to_anthropic(content),
-                    }  # mutable-ok: JSON wire message
+                    }
                 )
 
         system: Final[list[dict] | None] = system_parts if system_parts else None  # mutable-ok: JSON wire messages
@@ -516,11 +506,11 @@ class SnowflakeConfig(SnowflakeBaseConfig, OpenAIGPTConfig):
                 "messages": conversation,
                 "stream": stream,
                 **optional_params,
-                **extra_body,  # mutable-ok: JSON wire body
+                **extra_body,
             }
         )
         if system is not None:
-            body["system"] = normalize_cache_control_in_anthropic_payload(  # mutable-ok: JSON wire payload
+            body["system"] = normalize_cache_control_in_anthropic_payload(
                 {"system": system}  # mutable-ok: JSON wire payload
             )["system"]
 

litellm/llms/vertex_ai/text_to_speech/transformation.py

@@ -43,9 +43,7 @@ else:
     LiteLLMLoggingObj = Any
     HttpxBinaryResponseContent = Any
 
-_LyriaVoice: TypeAlias = (
-    str | dict | None
-)  # mutable-ok: inherited interface supports structured provider voice dictionaries
+_LyriaVoice: TypeAlias = str | dict | None
 
 
 class VertexAITextToSpeechConfig(BaseTextToSpeechConfig, VertexBase):
@@ -664,21 +662,15 @@ class VertexAILyriaTextToSpeechConfig(VertexAITextToSpeechConfig):
         if model_info["vertex_ai_audio_api"] == "lyria_predict":
             predictions: Final = response_json.get("predictions") or ()
             if predictions:
-                audio_data = predictions[0].get("audioContent") or predictions[0].get(
-                    "bytesBase64Encoded"
-                )  # rebind-ok: predict response supplies the generated audio value
+                audio_data = predictions[0].get("audioContent") or predictions[0].get("bytesBase64Encoded")
                 mime_type = predictions[0].get("mimeType")  # rebind-ok: predict response supplies its audio MIME type
         else:
             for step in response_json.get("steps") or response_json.get("outputs") or ():
                 content_items = step.get("content") or () if step.get("type") == "model_output" else (step,)
                 for content in content_items:
                     if content.get("type") == "audio" and content.get("data"):
-                        audio_data = content[
-                            "data"
-                        ]  # rebind-ok: interactions response supplies the generated audio value
-                        mime_type = content.get(
-                            "mime_type"
-                        )  # rebind-ok: interactions response supplies its audio MIME type
+                        audio_data = content["data"]
+                        mime_type = content.get("mime_type")
         if audio_data is None:
             raise ValueError(f"No generated audio found in Vertex AI {base_model} response")
         binary_data: Final = base64.b64decode(audio_data)

litellm/llms/xai/audio_transcription/transformation.py

@@ -168,9 +168,7 @@ class XAIAudioTranscriptionConfig(BaseAudioTranscriptionConfig):
                 for word in payload.words
             ]
 
-        hidden_params: Final[dict[str, object]] = dict(
-            payload.model_dump(mode="json")
-        )  # mutable-ok: TranscriptionResponse._hidden_params is a dict
+        hidden_params: Final[dict[str, object]] = dict(payload.model_dump(mode="json"))
         if payload.duration is not None:
             hidden_params["audio_transcription_duration"] = payload.duration
         response._hidden_params = hidden_params  # pyright: ignore[reportPrivateUsage]  # TranscriptionResponse exposes no public hidden-params setter

litellm/ocr/main.py

@@ -173,9 +173,7 @@ def _prepare_ocr_request(
         custom_llm_provider=custom_llm_provider,
         extra_headers=extra_headers,
         provider_config=ocr_provider_config,
-        optional_params=cast(
-            dict[str, object], optional_params
-        ),  # cast-ok: provider configs return heterogeneous OCR options
+        optional_params=cast(dict[str, object], optional_params),
         litellm_params=dict(litellm_params),
         effective_timeout=effective_timeout,
         litellm_logging_obj=litellm_logging_obj,

litellm/passthrough/main.py

@@ -428,9 +428,7 @@ def llm_passthrough_route(
 
     _is_async: Final = bool(kwargs.get("allm_passthrough_route", False))
 
-    litellm_logging_obj: Final = cast(
-        LiteLLMLoggingObj, kwargs.get("litellm_logging_obj")
-    )  # cast-ok: logging obj is constructed upstream; tests inject mocks
+    litellm_logging_obj: Final = cast(LiteLLMLoggingObj, kwargs.get("litellm_logging_obj"))
 
     model, custom_llm_provider, api_key, api_base = get_llm_provider(
         model=model,
@@ -516,9 +514,7 @@ def llm_passthrough_route(
         forward_headers=False,
     )
 
-    _request_data: dict | None = (
-        data if isinstance(data, dict) else (json if isinstance(json, dict) else None)
-    )  # rebind-ok: conditional
+    _request_data: dict | None = data if isinstance(data, dict) else (json if isinstance(json, dict) else None)
     headers, signed_json_body = provider_config.sign_request(
         headers=headers,
         litellm_params=litellm_params_dict,
@@ -544,9 +540,7 @@ def llm_passthrough_route(
     )
 
     ## IS STREAMING REQUEST
-    _streaming_request_data: dict = (
-        data if isinstance(data, dict) else (json if isinstance(json, dict) else {})
-    )  # rebind-ok: conditional
+    _streaming_request_data: dict = data if isinstance(data, dict) else (json if isinstance(json, dict) else {})
     is_streaming_request: Final = provider_config.is_streaming_request(
         endpoint=endpoint,
         request_data=_streaming_request_data,

litellm/proxy/_experimental/mcp_server/contracts.py

@@ -57,24 +57,20 @@ class OperationContext:
     ) -> tuple[
         UserAPIKeyAuth | None,
         str | None,
-        list[str] | None,  # mutable-ok: detached legacy server-list payload
-        dict[str, dict[str, str]] | None,  # mutable-ok: legacy auth dispatch requires concrete dict headers
-        dict[str, str] | None,  # mutable-ok: detached legacy header payload
-        dict[str, str] | None,  # mutable-ok: detached legacy header payload
+        list[str] | None,
+        dict[str, dict[str, str]] | None,
+        dict[str, str] | None,
+        dict[str, str] | None,
         str | None,
     ]:
         return (
             self.user_api_key_auth,
             self.mcp_auth_header,
             list(self.mcp_servers) if self.mcp_servers is not None else None,  # mutable-ok: legacy policy list input
-            {
-                key: dict(value) for key, value in self.mcp_server_auth_headers.items()
-            }  # mutable-ok: legacy auth dispatch checks concrete dict headers
+            {key: dict(value) for key, value in self.mcp_server_auth_headers.items()}
             if self.mcp_server_auth_headers is not None
             else None,
-            dict(self.oauth2_headers)
-            if self.oauth2_headers is not None
-            else None,  # mutable-ok: legacy OAuth header input
+            dict(self.oauth2_headers) if self.oauth2_headers is not None else None,
             dict(self.raw_headers) if self.raw_headers is not None else None,  # mutable-ok: legacy request header input
             self.client_ip,
         )

litellm/proxy/_experimental/mcp_server/db.py

@@ -636,8 +636,6 @@ async def get_all_mcp_servers(
     where: Final[prisma_db_types.LiteLLM_MCPServerTableWhereInput] = (
         {"approval_status": approval_status}
         if approval_status is not None
-        # mutable-ok: prisma where-inputs must be plain dicts, and both `NOT` and `not` drop
-        # NULL rows (measured), so the OR is the only NULL-preserving way to exclude drafts
         else {"OR": [{"approval_status": None}, {"approval_status": {"not": MCPApprovalStatus.draft}}]}
     )
     mcp_servers: Final = await _db_find_mcp_server_rows(prisma_client, where)

litellm/proxy/_experimental/mcp_server/legacy_callbacks.py

@@ -55,9 +55,7 @@ def create_sampling_callback(
             params=params,
             default_model=getattr(litellm, "default_mcp_sampling_model", None),
             user_api_key_auth=captured.user_api_key_auth,
-            raw_headers=dict(captured.raw_headers)
-            if captured.raw_headers is not None
-            else None,  # mutable-ok: handler consumes an owned request header dict
+            raw_headers=dict(captured.raw_headers) if captured.raw_headers is not None else None,
             client_ip=captured.client_ip,
         )
 

litellm/proxy/_experimental/mcp_server/mcp_debug.py

@@ -174,9 +174,7 @@ class MCPAuthDiagnostics:
             {
                 "x-mcp-debug-auth-resolution": AuthResolution.multiple.value,
                 "x-mcp-debug-auth-resolutions": json.dumps(
-                    {
-                        server_id: source.value for server_id, source in self._outcomes[:32]
-                    },  # mutable-ok: JSON encoder requires a concrete dict
+                    {server_id: source.value for server_id, source in self._outcomes[:32]},
                     separators=(",", ":"),
                     ensure_ascii=True,
                 ),
@@ -597,9 +595,7 @@ async def capture_upstream_error_response(response: httpx.Response | httpx2.Resp
         )
     except (asyncio.TimeoutError, httpx.HTTPError, httpx.StreamError, httpx2.HTTPError, httpx2.StreamError):
         response._content = b""  # pyright: ignore[reportPrivateUsage]  # rebind-ok: httpx auth retries must survive diagnostic read failures
-        response.extensions[_CAPTURE_EXTENSION] = (
-            "(unavailable: error body read failed)"  # rebind-ok: httpx response hooks communicate through extensions
-        )
+        response.extensions[_CAPTURE_EXTENSION] = "(unavailable: error body read failed)"
         return
     response.extensions[_CAPTURE_EXTENSION] = preview  # rebind-ok: httpx response hooks communicate through extensions
 

litellm/proxy/_experimental/mcp_server/operations.py

@@ -3103,9 +3103,7 @@ class GatewayOperations:
                 return await _execute_mcp_tool(
                     name=operation.name,
                     arguments=dict(operation.arguments),  # mutable-ok: existing tool hooks own mutable argument data
-                    allowed_mcp_servers=list(
-                        operation.allowed_mcp_servers
-                    ),  # mutable-ok: legacy dispatch list contract
+                    allowed_mcp_servers=list(operation.allowed_mcp_servers),
                     start_time=operation.start_time,
                     user_api_key_auth=auth,
                     mcp_auth_header=token,

litellm/proxy/_experimental/mcp_server/tool_search.py

@@ -103,7 +103,7 @@ def _tool_result(tool: Tool) -> ToolSearchResult:
         "name": tool.name,
         "description": tool.description or "",
         "inputSchema": tool.input_schema,
-    }  # mutable-ok: wire schema payload
+    }
 
 
 def _scored_result(tool: Tool, score: float) -> ToolSearchResult:
@@ -112,7 +112,7 @@ def _scored_result(tool: Tool, score: float) -> ToolSearchResult:
         "description": tool.description or "",
         "inputSchema": tool.input_schema,
         "score": score,
-    }  # mutable-ok: wire schema payload
+    }
 
 
 _MCP_PROXY_IDENTITY_META_KEY: Final[str] = "litellm.ai/proxy_tool_identity"
@@ -120,7 +120,7 @@ _MCP_PROXY_IDENTITY_META_KEY: Final[str] = "litellm.ai/proxy_tool_identity"
 
 def with_mcp_proxy_identity(tool: Tool, server_id: str) -> Tool:
     identity: Final[MCPProxyToolIdentity] = {"server_id": server_id, "tool_name": tool.name}
-    return tool.model_copy(  # mutable-ok: Pydantic requires mutable update and metadata mappings
+    return tool.model_copy(
         update={  # mutable-ok: Pydantic update payload
             "meta": {**(tool.meta or {}), _MCP_PROXY_IDENTITY_META_KEY: identity}  # mutable-ok: metadata mapping
         }

litellm/proxy/agent_endpoints/agent_registry.py

@@ -135,9 +135,7 @@ def _dump_agent_params(raw: Mapping[str, object]) -> dict[str, object]:
 
 _AGENT_PARAMS_MASKER: Final = SensitiveDataMasker()
 _REDACT_AGENT_PARAMS_MAX_DEPTH: Final = 10
-_AGENT_PARAMS_ADAPTER: Final[TypeAdapter[dict[str, object]]] = TypeAdapter(
-    dict[str, object]
-)  # mutable-ok: safe_dumps() and AgentResponse.litellm_params both require a real dict, not a Mapping
+_AGENT_PARAMS_ADAPTER: Final[TypeAdapter[dict[str, object]]] = TypeAdapter(dict[str, object])
 _AGENT_PARAMS_SEQUENCE_ADAPTER: Final[TypeAdapter[tuple[object, ...]]] = TypeAdapter(tuple[object, ...])
 _EMPTY_LITELLM_PARAMS: Final[Mapping[str, object]] = MappingProxyType({})
 
@@ -189,7 +187,7 @@ def _redact_agent_params_tree(value: object, _depth: int) -> object:
             else _redact_agent_params_tree(nested_value, _depth + 1)
         )
         for key, nested_value in typed_params.items()
-    }  # mutable-ok: consumed by json.dumps()/AgentResponse.litellm_params, both of which require a real dict
+    }
 
 
 def parse_agent_litellm_params(value: object) -> Mapping[str, object]:
@@ -318,7 +316,7 @@ def _restore_redacted_litellm_params(
         key: value
         for key in all_keys
         if (value := _resolved_agent_param_value(key, incoming, existing, _depth)) is not _MISSING_AGENT_PARAM
-    }  # mutable-ok: fed to safe_dumps() for JSON-column storage, which requires a real dict
+    }
 
 
 class GrantMigrationResult(NamedTuple):

litellm/proxy/auth/auth_utils.py

@@ -1410,7 +1410,7 @@ def log_once_if_budget_reservation_disabled(
         "Set disable_budget_reservation to False or remove it to restore "
         "hard per-request budget enforcement."
     )
-    constants.budget_reservation_disabled_info_emitted = True  # rebind-ok: process-wide one-shot sentinel
+    constants.budget_reservation_disabled_info_emitted = True
 
 
 def is_pass_through_provider_route(route: str) -> bool:

litellm/proxy/client/cli/commands/agents.py

@@ -241,9 +241,7 @@ def prepare_codex(
 
 _Preparer: TypeAlias = Callable[[str, str, Mapping[str, str]], Sequence[str]]
 
-_PREPARERS: Final[Mapping[str, _Preparer]] = MappingProxyType(
-    {"pi": prepare_pi, "codex": prepare_codex}  # mutable-ok: MappingProxyType freezes the provider registry
-)
+_PREPARERS: Final[Mapping[str, _Preparer]] = MappingProxyType({"pi": prepare_pi, "codex": prepare_codex})
 
 
 def agent_launch_args(command: str, base_url: str) -> list[str]:

litellm/proxy/client/cli/commands/claude_settings.py

@@ -621,7 +621,7 @@ def unconfigure_claude_settings(
     )
     target: Final = _write_target(settings_path)
     file_removed: Final = not settings and not (receipt.file_existed and target.exists())
-    kept_receipt: Final = (  # mutable-ok: pydantic serializes the update as given and rejects a mappingproxy
+    kept_receipt: Final = (
         receipt.model_copy(update={"written": {item.key: _fingerprint(absent) for item in withheld}})
         if withheld
         else None

litellm/proxy/client/cli/commands/codex_settings.py

@@ -106,7 +106,6 @@ def _with(document: TOMLDocument, path: str, snapshot: str | None) -> TOMLDocume
     if section and section not in document and snapshot is not None:
         contents: Final = tomlkit.parse(tomlkit.dumps(MappingProxyType({key: tomlkit.parse(snapshot).item("value")})))
         return tomlkit.parse(document.as_string() + "\n" + tomlkit.dumps(MappingProxyType({section: contents})))
-    # mutable-ok: TOMLKit editing requires private node mutation to preserve comments and order
     updated: Final = tomlkit.parse(document.as_string())
     parent: Final = _table(_mapping(updated).get(section)) if section else updated
     if parent is None:

litellm/proxy/client/cli/commands/pi.py

@@ -175,7 +175,7 @@ def _model_entry(
     )
     output: Final[dict[str, JsonValue]] = (  # mutable-ok: JSON field
         {"maxTokens": limit.max_tokens} if limit and limit.max_tokens else {}
-    )  # mutable-ok: JSON field
+    )
     return {"id": model_id, **context, **output}  # mutable-ok: JSON serialization requires a mutable object
 
 
@@ -208,9 +208,7 @@ def sync_models_json(
 ) -> PiSyncError | None:
     """Replace only the litellm provider entry, leaving the rest of the file intact."""
     try:
-        current: Final = (  # mutable-ok: JSON object default
-            _MODELS_FILE_ADAPTER.validate_json(path.read_text()) if path.exists() else {}
-        )
+        current: Final = _MODELS_FILE_ADAPTER.validate_json(path.read_text()) if path.exists() else {}
     except (OSError, ValidationError) as e:
         return PiSyncError(f"Could not read {path} as a JSON object: {e}. Fix or move the file, then retry.")
     existing_providers: Final = current.get("providers", {})  # mutable-ok: JSON object default

litellm/proxy/common_utils/auth_cache_invalidation_pubsub.py

@@ -184,17 +184,17 @@ class AuthCacheInvalidationSubscriber:
         backoff_seconds = _BACKOFF_INITIAL_SECONDS  # rebind-ok: exponential backoff accumulator across reconnects
         while True:
             try:
-                client = _pubsub_capable_client(self._redis_cache)  # rebind-ok: re-resolved on every reconnect
+                client = _pubsub_capable_client(self._redis_cache)
                 if client is None:
                     verbose_proxy_logger.warning(
                         "auth cache invalidation subscriber disabled: cluster redis client has no pub/sub support; "
                         "cross-worker eviction falls back to the local cache TTL"
                     )
                     return
-                pubsub = client.pubsub()  # rebind-ok: fresh pubsub per reconnect
+                pubsub = client.pubsub()
                 try:
                     await pubsub.subscribe(auth_cache_invalidation_channel(self._redis_cache))
-                    backoff_seconds = _BACKOFF_INITIAL_SECONDS  # rebind-ok: reset after successful subscribe
+                    backoff_seconds = _BACKOFF_INITIAL_SECONDS
                     await self._consume(pubsub)
                 finally:
                     await self._close_pubsub(pubsub)
@@ -207,7 +207,7 @@ class AuthCacheInvalidationSubscriber:
                     backoff_seconds,
                 )
                 await asyncio.sleep(backoff_seconds)
-                backoff_seconds = min(backoff_seconds * 2, _BACKOFF_MAX_SECONDS)  # rebind-ok: backoff accumulator
+                backoff_seconds = min(backoff_seconds * 2, _BACKOFF_MAX_SECONDS)
 
     async def _consume(self, pubsub: _ConfigSyncPubSub) -> None:
         while True:

litellm/proxy/common_utils/reset_budget_job.py

@@ -240,12 +240,8 @@ def _queue_budget_linked_resets(
     one transaction, so the reverse order lets the zero re-match a row the
     decrement just moved into the (0, cap] range and erase its carried spend."""
     for budget_id, cap in cascade.rollover_caps.items():
-        writes.queue_spend_zero(
-            where={"budget_id": budget_id, **extra, "spend": {"gt": 0, "lte": cap}}
-        )  # mutable-ok: prisma where filter must be a dict
-        writes.queue_spend_decrement(
-            where={"budget_id": budget_id, **extra, "spend": {"gt": cap}}, amount=cap
-        )  # mutable-ok: prisma where filter must be a dict
+        writes.queue_spend_zero(where={"budget_id": budget_id, **extra, "spend": {"gt": 0, "lte": cap}})
+        writes.queue_spend_decrement(where={"budget_id": budget_id, **extra, "spend": {"gt": cap}}, amount=cap)
     plain_ids: Final = tuple(bid for bid in cascade.budget_ids if bid not in cascade.rollover_caps)
     if plain_ids:
         writes.queue_spend_zero(where=_budget_link_where(plain_ids, extra))
@@ -267,16 +263,10 @@ def _queue_enduser_resets(writes: LinkedSpendResetWrites, cascade: "_BudgetCasca
         return
     cap: Final = cascade.rollover_caps.get(default_budget_id)
     if cap is None:
-        writes.queue_spend_zero(
-            where={"budget_id": None, **_SPENT_ROWS_WHERE}
-        )  # mutable-ok: prisma where filter must be a dict
+        writes.queue_spend_zero(where={"budget_id": None, **_SPENT_ROWS_WHERE})
         return
-    writes.queue_spend_zero(
-        where={"budget_id": None, "spend": {"gt": 0, "lte": cap}}
-    )  # mutable-ok: prisma where filter must be a dict
-    writes.queue_spend_decrement(
-        where={"budget_id": None, "spend": {"gt": cap}}, amount=cap
-    )  # mutable-ok: prisma where filter must be a dict
+    writes.queue_spend_zero(where={"budget_id": None, "spend": {"gt": 0, "lte": cap}})
+    writes.queue_spend_decrement(where={"budget_id": None, "spend": {"gt": cap}}, amount=cap)
 
 
 @dataclass(frozen=True, slots=True)

litellm/proxy/common_utils/sse_keepalive.py

@@ -65,9 +65,7 @@ async def _keepalive_ping_stream(
     ping_interval_seconds: float,
     ping_chunk: str,
 ) -> AsyncGenerator[str, None]:
-    pending = asyncio.ensure_future(
-        stream.__anext__()
-    )  # rebind-ok: re-armed with the next __anext__ after each delivered chunk
+    pending = asyncio.ensure_future(stream.__anext__())
     try:
         while True:
             await asyncio.wait({pending}, timeout=ping_interval_seconds)
@@ -125,9 +123,7 @@ async def _keepalive_ping_byte_stream(
     stream: AsyncGenerator[bytes, None],
     ping_interval_seconds: float,
 ) -> AsyncGenerator[bytes, None]:
-    pending = asyncio.ensure_future(
-        stream.__anext__()
-    )  # rebind-ok: re-armed with the next __anext__ after each delivered chunk
+    pending = asyncio.ensure_future(stream.__anext__())
     # The tail of the bytes relayed so far, long enough to hold any delimiter.
     # Seeded as a delimiter because a stream starts at a frame boundary, and kept
     # across chunks because a delimiter can be split between two transport reads,

litellm/proxy/db/baseline_accounting.py

@@ -491,7 +491,7 @@ class BaselineAccountingStore:
             tuple(await db.query_raw(_READ_PAGE, scope, after_revision, cursor, _PAGE_TIMESTAMPS, withdraw_from))
         ):
             yield page
-            cursor = page[-1].started_at  # rebind-ok: keyset pagination advances after each complete timestamp group
+            cursor = page[-1].started_at
 
     async def _withdraw(self, db: SupportsRawQueries, scope: str, started_at: float) -> None:
         async for page in self._pages(db, scope, 0, withdraw_from=started_at):
@@ -623,9 +623,7 @@ async def flush_baseline_accounting(client: PrismaClient) -> None:
     store: Final = BaselineAccountingStore.for_client(client)
     async with client.baseline_accounting_lock:
         batch: Final = tuple(client.baseline_accounting_transactions[:32])
-        client.baseline_accounting_transactions = client.baseline_accounting_transactions[
-            32:
-        ]  # rebind-ok: drain under lock
+        client.baseline_accounting_transactions = client.baseline_accounting_transactions[32:]
         more_queued: Final = bool(client.baseline_accounting_transactions)
     try:
         remaining: Final = await asyncio.wait_for(_flush_records(store, batch), timeout=5)

litellm/proxy/db/shadow_eval_funnel.py

@@ -40,7 +40,7 @@ def pending_shadow_eval_funnel_events() -> int:
 def record_shadow_eval_funnel_event(job_id: str, stage: ShadowEvalFunnelStage) -> None:
     """Count one skipped request for one job leg; synchronous so the hook's read-modify-
     write cannot interleave with the flush's snapshot on the shared event loop."""
-    counters: Final = _pending.setdefault(job_id, dict.fromkeys(FUNNEL_STAGES, 0))  # mutable-ok: queue entry
+    counters: Final = _pending.setdefault(job_id, dict.fromkeys(FUNNEL_STAGES, 0))
     counters[stage] += 1
 
 

litellm/proxy/guardrails/guardrail_hooks/alice/alice.py

@@ -286,7 +286,7 @@ class AliceGuardrail(CustomGuardrail):
             text = replacement.get("text")
             if not (isinstance(index, int) and isinstance(text, str) and 0 <= index < len(texts)):
                 raise self._mask_rejected(verdict)
-            texts[index] = text  # mutable-ok: item assignment into the local working copy above
+            texts[index] = text
 
         inputs["texts"] = texts
 

litellm/proxy/guardrails/guardrail_hooks/bedrock_guardrails.py

@@ -1218,7 +1218,7 @@ class BedrockGuardrail(CustomGuardrail, BaseAWSLLM):
         bedrock_request_data: Final = {  # mutable-ok: outbound JSON request body
             **base_request_data,
             "content": content,
-        }  # mutable-ok: outbound JSON request body
+        }
         prepared_request: Final = await run_aws_signing(
             self._prepare_request,
             credentials=credentials,
@@ -1266,9 +1266,7 @@ class BedrockGuardrail(CustomGuardrail, BaseAWSLLM):
                 )
             response_usage: Final = bedrock_guardrail_response.get("usage")
             if isinstance(response_usage, dict):
-                completed_chunk_usages.append(
-                    response_usage
-                )  # rebind-ok: accumulator threaded from make_bedrock_api_request, recording this billed call
+                completed_chunk_usages.append(response_usage)
             return bedrock_guardrail_response
 
         status_code, detail_message = self._parse_bedrock_guardrail_error_response(httpx_response)
@@ -2860,9 +2858,9 @@ class BedrockGuardrail(CustomGuardrail, BaseAWSLLM):
                 return
             except ModifyResponseException as e:
                 if raw_sse:
-                    e.model = _pre_block_response.model or e.model  # rebind-ok: exc.model defaults to the guardrail
+                    e.model = _pre_block_response.model or e.model
                     if e.original_response is None:
-                        e.original_response = _pre_block_response  # rebind-ok: the block builder reads usage off this
+                        e.original_response = _pre_block_response
                     for block_chunk in AnthropicMessagesHandler().build_block_sse_chunks(e, stream_started=False):
                         yield block_chunk
                     return

litellm/proxy/guardrails/guardrail_hooks/presidio.py

@@ -168,7 +168,7 @@ class _OPTIONAL_PresidioPIIMasking(CustomGuardrail):
 
         # Per-loop semaphores bounding chunked-analyze fan-out across ALL
         # concurrent oversized blocks/requests on this instance, not per call
-        self._loop_chunk_semaphores: _LoopSemaphores = {}  # mutable-ok: per-loop semaphore cache
+        self._loop_chunk_semaphores: _LoopSemaphores = {}
 
         if mock_testing is True:  # for testing purposes only
             return

litellm/proxy/hooks/autorouter_baseline_cache.py

@@ -230,12 +230,10 @@ class AutoRouterBaselineCache(CustomLogger):
 async def invalidate_baseline_cache(logging_obj: Logging, reason: str, *, completed: bool = False) -> None:
     context: Final = logging_obj.baseline_cache_context
     if context is not None:
-        logging_obj.baseline_cache_context = replace(
-            context, invalidated=reason
-        )  # rebind-ok: request-owned retry marker
+        logging_obj.baseline_cache_context = replace(context, invalidated=reason)
         logging_obj.baseline_observation = context.capture.model_copy(
             update=MappingProxyType(
-                {  # rebind-ok: capture uncertainty for failure logging
+                {
                     "observation": context.capture.observation.model_copy(
                         update=MappingProxyType(
                             {

litellm/proxy/hooks/batch_rate_limiter.py

@@ -114,9 +114,7 @@ class BatchFileUsage(BaseModel):
     # each target a different model, so the project's per-model ITPM/OTPM
     # quota for a row's actual model must be charged with that row's own
     # tokens -- see `_create_project_io_descriptors_for_models`.
-    per_model_usage: dict[str, dict[str, int]] = Field(
-        default_factory=dict
-    )  # mutable-ok: accumulated incrementally per row while parsing the batch file
+    per_model_usage: dict[str, dict[str, int]] = Field(default_factory=dict)
 
 
 class _PROXY_BatchRateLimiter(CustomLogger):
@@ -465,7 +463,7 @@ class _PROXY_BatchRateLimiter(CustomLogger):
         body: Final[Mapping[str, object]] = (
             MappingProxyType(_BATCH_BODY_ADAPTER.validate_python(raw_body))
             if isinstance(raw_body, Mapping)
-            else MappingProxyType({})  # mutable-ok: immediately frozen empty fallback
+            else MappingProxyType({})
         )
         # `max_tokens`/`max_completion_tokens` cap chat completions; `/v1/responses`
         # rows cap output with `max_output_tokens` instead -- omitting it here

litellm/proxy/hooks/parallel_request_limiter_v3.py

@@ -3210,7 +3210,7 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
             filtered_content = [  # mutable-ok: token_counter requires list content blocks
                 block for block in content if not (isinstance(block, dict) and block.get("type") == "input_audio")
             ]
-            sanitized.append(  # mutable-ok: token_counter requires mutable message dicts
+            sanitized.append(
                 {**message, "content": filtered_content}  # mutable-ok: token_counter requires message dicts
             )
         return sanitized
@@ -3572,7 +3572,7 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
                 try:
                     await asyncio.shield(cleanup)
                 except asyncio.CancelledError as exc:
-                    cancellation = exc  # rebind-ok: retain the latest cancellation without interrupting slot release
+                    cancellation = exc
             cleanup.result()
             if cancellation is not None:
                 raise cancellation

litellm/proxy/litellm_pre_call_utils.py

@@ -137,7 +137,7 @@ def add_otel_trace_id_to_request(
         return
     data["litellm_trace_id"] = trace_id  # rebind-ok: data is an out-param
     if isinstance(metadata, dict):
-        metadata["trace_id"] = trace_id  # rebind-ok: metadata is the request's own out-param dict
+        metadata["trace_id"] = trace_id
 
 
 def _session_id_from_baggage(baggage: str) -> str | None:
@@ -3142,11 +3142,7 @@ async def move_guardrails_to_metadata(
     - Moves include_guardrail_response into request metadata before provider dispatch
     """
     if "include_guardrail_response" in data:
-        data[_metadata_variable_name][
-            "include_guardrail_response"
-        ] = (  # rebind-ok: pre-call hooks mutate the shared request dict in place
-            data.pop("include_guardrail_response") is True
-        )
+        data[_metadata_variable_name]["include_guardrail_response"] = data.pop("include_guardrail_response") is True
 
     # Early-out: skip all guardrails processing when nothing is configured
     key_metadata: Final = user_api_key_dict.metadata

litellm/proxy/management_endpoints/auto_router_endpoints.py

@@ -1448,7 +1448,7 @@ def _target_labels(
     """Display labels by (target_type, target_id): a key's (alias, masked name), a
     team's (alias, None), a user's (email, None)."""
     return MappingProxyType(
-        {  # mutable-ok: MappingProxyType needs a dict to wrap
+        {
             key: value
             for key, value in chain(
                 ((("key", row.token), (row.key_alias, row.key_name)) for row in key_rows),
@@ -1548,7 +1548,7 @@ async def _shadow_eval_results(
         await _query_raw(prisma_client, _ATTEMPT_AGG_BY_LEG_SQL, leg_ids) or ()
     )
     verdicts_by_target: Final[Mapping[tuple[str, str], ShadowEvalSlice]] = MappingProxyType(
-        {  # mutable-ok: MappingProxyType needs a dict to wrap
+        {
             target_by_leg[slice.group]: slice.model_copy(
                 update={"group": target_by_leg[slice.group][1]}  # mutable-ok: pydantic update payload
             )
@@ -1760,7 +1760,7 @@ async def start_shadow_eval(
                     "id": leg_id,
                     "target_type": target_type,
                     "target_id": target_id,
-                }  # mutable-ok: Prisma payload
+                }
                 for leg_id, (target_type, target_id) in zip(leg_ids, requested_targets)
             ]
         )

litellm/proxy/management_endpoints/config_override_endpoints.py

@@ -796,9 +796,7 @@ async def get_cyberark_config(
 
     field_schema: Final = _build_field_schema(CyberArkConfig)
 
-    db_record: Final = await _config_overrides_table(prisma_client).find_unique(
-        where={"config_type": "cyberark"}
-    )  # mutable-ok: prisma where clause
+    db_record: Final = await _config_overrides_table(prisma_client).find_unique(where={"config_type": "cyberark"})
 
     if db_record is not None and db_record.config_value is not None:
         config_data: Final = _parse_config_value(db_record.config_value)
@@ -860,9 +858,7 @@ async def delete_cyberark_config(
 
     deleted = False  # rebind-ok: set true once the DB row is removed
     try:
-        await _config_overrides_table(prisma_client).delete(
-            where={"config_type": "cyberark"}
-        )  # mutable-ok: prisma where clause
+        await _config_overrides_table(prisma_client).delete(where={"config_type": "cyberark"})
         deleted = True  # rebind-ok: set true once the DB row is removed
     except RecordNotFoundError:
         verbose_proxy_logger.debug("No existing CyberArk config record to delete")

litellm/proxy/management_endpoints/management_v1/budgets.py

@@ -115,7 +115,7 @@ def _scope(caller: UserAPIKeyAuth) -> Scope:
 # budget_duration is deliberately absent from `sortable`: the column holds strings
 # like "7d" and "30d", so a lexicographic ORDER BY puts "30d" ahead of "7d".
 BUDGET_FILTERS: Final[Mapping[str, FilterSpec]] = MappingProxyType(
-    {  # mutable-ok: an immutable mapping has no literal form; MappingProxyType freezes this one and it never escapes
+    {
         "budget_duration": FilterSpec(type=str, ops=frozenset(("in", "is_null"))),
         "max_budget": FilterSpec(type=float, ops=frozenset(("gte", "lte", "is_null"))),
         "created_at": FilterSpec(type=datetime, ops=frozenset(("gte", "lte"))),

litellm/proxy/management_endpoints/mcp_management_endpoints.py

@@ -706,9 +706,7 @@ if MCP_AVAILABLE:
         if not caller_user_id:
             raise HTTPException(
                 status_code=status.HTTP_400_BAD_REQUEST,
-                detail={
-                    "error": "User ID not found in token"
-                },  # mutable-ok: FastAPI HTTPException detail requires a plain dict
+                detail={"error": "User ID not found in token"},
             )
         return caller_user_id
 
@@ -1865,9 +1863,7 @@ if MCP_AVAILABLE:
 
         classified: Final = tuple(_classify(index, conversion) for index, conversion in enumerate(conversions))
         outcomes: Final = tuple(
-            [
-                await _create(entry) if isinstance(entry, ConvertedConnector) else entry for entry in classified
-            ]  # mutable-ok: await is illegal in a generator expression here
+            [await _create(entry) if isinstance(entry, ConvertedConnector) else entry for entry in classified]
         )
 
         imported: Final = tuple(entry for entry in outcomes if isinstance(entry, MCPConnectorImportResult))

litellm/proxy/management_endpoints/scim/scim_v2.py

@@ -582,7 +582,6 @@ async def _users_named_by_member_value(
     subject: Final = value.strip()
     email: Final[_CaseInsensitiveMatch] = {"equals": subject, "mode": "insensitive"}
     rows: Final = await _table(UserRepository(prisma_client)).find_many(
-        # mutable-ok: the Prisma serializer requires concrete dicts and a concrete list
         where={"OR": [{"sso_user_id": subject}, {"user_email": email}]},
         take=take,
     )

litellm/proxy/management_endpoints/team_endpoints.py

@@ -2996,7 +2996,7 @@ async def _update_team_members_list(
     # extend() consumes the generator as it appends, so a member already added by this
     # same call is seen by the next _member_already_in_team check - the batch dedupes
     # against itself exactly as the append-one-at-a-time loop this replaced did.
-    complete_team_data.members_with_roles.extend(  # rebind-ok: this helper's contract is to grow the caller's roster in place
+    complete_team_data.members_with_roles.extend(
         m for m in resolved_members if not _member_already_in_team(m, complete_team_data)
     )
 
@@ -4137,9 +4137,7 @@ async def reset_team_member_budget_fn(
 
     team_default_budget_id: Final = await _existing_team_default_budget_id(team_obj, prisma_client)
     budget_link: Final = (
-        {
-            "connect": {"budget_id": team_default_budget_id}
-        }  # mutable-ok: prisma client requires a plain dict data= argument
+        {"connect": {"budget_id": team_default_budget_id}}
         if team_default_budget_id is not None
         else {"disconnect": True}  # mutable-ok: same prisma data= argument
     )

litellm/proxy/management_helpers/bulk_user_creation.py

@@ -543,7 +543,7 @@ async def _write_team_roster(
             already_present: Final = frozenset(member.user_id for member in roster if member.user_id)
             new_members: Final = tuple(member for member in members if member.user_id not in already_present)
             budget_ids: Final = tuple(
-                [  # mutable-ok: budgets are created one at a time on the transaction's single connection
+                [
                     await _resolve_member_budget_id(
                         prisma_client=prisma_client,
                         user_api_key_dict=user_api_key_dict,

litellm/proxy/openai_files_endpoints/batch_guardrails.py

@@ -355,9 +355,7 @@ def build_scan_metadata(request_metadata: Mapping[str, object]) -> Mapping[str,
     Passing the whole thing through would carry values that cannot be copied, such as the parent
     OTel span, and would hand every record proxy state it has no business seeing.
     """
-    return MappingProxyType(
-        {key: value for key, value in request_metadata.items() if key in _SCAN_METADATA_KEYS}
-    )  # mutable-ok: MappingProxyType freezes the comprehension
+    return MappingProxyType({key: value for key, value in request_metadata.items() if key in _SCAN_METADATA_KEYS})
 
 
 async def _scan_record(
@@ -546,7 +544,7 @@ def rewrite_batch_input_file(file_source: BinaryIO, result: BatchScanResult) ->
     """
     redacted: Final = MappingProxyType(
         {change.line_number: change for change in result.changes if isinstance(change, RecordRedacted)}
-    )  # mutable-ok: MappingProxyType freezes the lookup table
+    )
     dropped: Final = frozenset(change.line_number for change in result.changes if isinstance(change, RecordDropped))
 
     output: Final = tempfile.SpooledTemporaryFile(  # noqa: SIM115  # the caller uploads this handle

litellm/proxy/pass_through_endpoints/llm_passthrough_endpoints.py

@@ -468,9 +468,7 @@ async def fal_ai_proxy_route(
     endpoint_func: Final = create_pass_through_route(
         endpoint=endpoint,
         target=str(updated_url),
-        custom_headers={
-            "Authorization": f"Key {fal_ai_api_key}"
-        },  # mutable-ok: pass-through request headers require a mutable mapping
+        custom_headers={"Authorization": f"Key {fal_ai_api_key}"},
         custom_llm_provider="fal_ai",
         is_streaming_request=False,
     )
@@ -3801,13 +3799,9 @@ async def gigachat_proxy_route(
     raw_model: Final = request_body.get("model")
     model: Final = raw_model if isinstance(raw_model, str) else None
     if model:
-        is_router_model = is_passthrough_request_using_router_model(
-            request_body, llm_router
-        )  # rebind-ok: conditionally set to True
+        is_router_model = is_passthrough_request_using_router_model(request_body, llm_router)
     elif any(word in endpoint for word in ("completions", "embeddings")):
-        raise HTTPException(
-            status_code=400, detail={"error": "Model is required in request body"}
-        )  # mutable-ok: HTTPException detail dict
+        raise HTTPException(status_code=400, detail={"error": "Model is required in request body"})
 
     # If router model, use dedicated router passthrough handler
     # This uses the same common processing path as non-router models
@@ -3908,9 +3902,7 @@ async def handle_gigachat_passthrough_router_model(
 
     is_streaming: Final = request_body.get("stream", False)  # pyright: ignore[reportUnknownVariableType]  # request_body is dict[Unknown, Unknown]
 
-    data: dict[str, Any] = await _read_request_body(
-        request=request
-    )  # mutable-ok: mutated in place by proxy pipeline; pyright: ignore[reportExplicitAny]  # Any needed for proxy pipeline
+    data: dict[str, Any] = await _read_request_body(request=request)  # Any needed for proxy pipeline
     if user_api_key_dict is not None:
         auth_metadata: Final = {
             metadata_key: value

litellm/proxy/pass_through_endpoints/llm_provider_handlers/vertex_passthrough_logging_handler.py

@@ -447,9 +447,7 @@ class VertexPassthroughLoggingHandler:
         kwargs["model"] = model  # rebind-ok: callback metadata records the resolved model
         kwargs["custom_llm_provider"] = "vertex_ai"  # rebind-ok: callback metadata records the resolved provider
 
-        standard_pass_through_response_object: Final[
-            StandardPassThroughResponseObject
-        ] = {  # mutable-ok: callback contract requires a concrete response dictionary
+        standard_pass_through_response_object: Final[StandardPassThroughResponseObject] = {
             "response": json_response,
         }
         return {  # mutable-ok: passthrough logging contract requires a concrete result dictionary

litellm/proxy/pass_through_endpoints/managed_id_rewriter.py

@@ -81,9 +81,7 @@ if TYPE_CHECKING:
     from litellm.integrations.custom_logger import CustomLogger
     from litellm.proxy.utils import PrismaClient
 
-_RowT = TypeVar(
-    "_RowT", bound=ManagedResourceRow
-)  # rebind-ok: TypeVar declarations must stay bare assignments for pyright
+_RowT = TypeVar("_RowT", bound=ManagedResourceRow)
 
 # ---------------------------------------------------------------------------
 # Field map
@@ -998,9 +996,7 @@ async def _build_list_where_with_cursor(
     params: Final = query_params or {}
     after_id: Final[str | None] = params.get("after")
     before_id: Final[str | None] = params.get("before")
-    where: PrismaWhere = dict(
-        owner_filter
-    )  # rebind-ok: narrowed with the cursor boundary when a valid cursor row exists
+    where: PrismaWhere = dict(owner_filter)
     fetch_order: SortOrder = "desc"  # rebind-ok: flipped to asc when paging backwards from a before cursor
 
     cursor_id: Final = after_id or before_id

litellm/proxy/pass_through_endpoints/pass_through_endpoints.py

@@ -819,9 +819,7 @@ def _resolve_team_callback_wiring(
             user_api_key_dict=user_api_key_dict, proxy_config=proxy_config
         )
         if callback_settings_obj and callback_settings_obj.callback_vars:
-            for (
-                item
-            ) in callback_settings_obj.callback_vars.items():  # rebind-ok: dict.items iteration for env-ref validation
+            for item in callback_settings_obj.callback_vars.items():
                 validate_no_callback_env_reference(item[0], item[1], source="key/team callback metadata")
     except Exception:  # noqa: BLE001 - a broken logging config must never fail the passthrough request
         verbose_proxy_logger.exception(

litellm/proxy/pass_through_endpoints/streaming_handler.py

@@ -217,9 +217,7 @@ class PassThroughStreamingHandler:
                 async for chunk in response.aiter_bytes():
                     raw_bytes.append(chunk)
                     PassThroughStreamingHandler._stamp_first_chunk_if_needed(litellm_logging_obj)
-                    complete_frames, pending = split_complete_sse_frames(
-                        pending + chunk
-                    )  # rebind-ok: SSE frame reassembly buffer across transport chunks
+                    complete_frames, pending = split_complete_sse_frames(pending + chunk)
                     if complete_frames:
                         yield ProxyBaseLLMRequestProcessing._process_chunk_with_cost_injection(
                             complete_frames, resolved_model_name, litellm_logging_obj

litellm/proxy/policy_engine/pipeline_executor.py

@@ -108,7 +108,7 @@ _GuardrailMethodT = TypeVar("_GuardrailMethodT", bound=Callable[..., object])
 
 
 def _logged_by_inner_guardrail(method: _GuardrailMethodT) -> _GuardrailMethodT:
-    vars(method)[LOGS_GUARDRAIL_INFORMATION_MARKER] = True  # rebind-ok: stamps the method the class body just defined
+    vars(method)[LOGS_GUARDRAIL_INFORMATION_MARKER] = True
     return method
 
 
@@ -278,9 +278,7 @@ def _prepare_hook_input(
     guardrail loops do this."""
     if "metadata" not in data:
         data["metadata"] = {}  # mutable-ok: request metadata bucket, hooks mutate it
-    data["metadata"]["guardrails"] = [
-        step.guardrail
-    ]  # mutable-ok: guardrails list is part of the request-payload shape
+    data["metadata"]["guardrails"] = [step.guardrail]
 
     scans_raw_request: Final = callback.scan_raw_request
     hook_input: Final[dict] = (  # mutable-ok: same request-payload shape as data
@@ -456,7 +454,7 @@ class PipelineExecutor:
         observer: Final = _StreamRewriteObserver(scanner)
         deliver_rewrites: Final = type(endpoint_translation).delivers_ended_stream_rewrites
         originals: Final = copy.deepcopy(streaming_chunks)
-        hook_input.pop("response", None)  # rebind-ok: an earlier step's stored response goes so this step's is stored
+        hook_input.pop("response", None)
         try:
             if deliver_rewrites:
                 await endpoint_translation.process_output_streaming_response(
@@ -582,7 +580,7 @@ class PipelineExecutor:
                     {"response": response},
                     None,
                     None,
-                )  # mutable-ok: modified-data contract is a plain dict
+                )
             return ("pass", response if isinstance(response, dict) else None, None, None)
 
         except Exception as e:

litellm/proxy/proxy_server.py

@@ -5261,9 +5261,7 @@ class ProxyConfig:
             return
 
         with open(f"{user_config_file_path}", "w") as config_file:
-            yaml.dump(
-                dict(new_config), config_file, default_flow_style=False
-            )  # mutable-ok: YAML must serialize a plain dict
+            yaml.dump(dict(new_config), config_file, default_flow_style=False)
 
     async def _save_changed_config_section(
         self,
@@ -10137,7 +10135,7 @@ class ProxyStartupEvent:
                         str(identity): str(fingerprint)
                         for identity, fingerprint in (decoded.items() if isinstance(decoded, Mapping) else ())
                     }
-                )  # mutable-ok: MappingProxyType owns the completed immutable baseline
+                )
             snapshot: Final = snapshot_tuning_baselines(deployments)
             try:
                 await config_table.create(
@@ -10163,7 +10161,7 @@ class ProxyStartupEvent:
                             competing_decoded.items() if isinstance(competing_decoded, Mapping) else ()
                         )
                     }
-                )  # mutable-ok: MappingProxyType owns the completed immutable baseline
+                )
         except Exception as e:  # noqa: BLE001  # enforcement is skipped for this boot; refusing every tuned router on a DB blip is the one outcome the gate forbids
             verbose_proxy_logger.warning("Heuristic-v1 tuning baseline unavailable, gate not enforced this boot: %s", e)
             return None
@@ -10199,7 +10197,7 @@ class ProxyStartupEvent:
         proxy_logging_obj: ProxyLogging,
     ) -> ProxyWorkerHeartbeat:
         """Initializes scheduled background jobs"""
-        global heuristic_v1_tuning_baselines, store_model_in_db, scheduler, scheduler_executor  # rebind-ok: startup publishes the one read-only baseline snapshot
+        global heuristic_v1_tuning_baselines, store_model_in_db, scheduler, scheduler_executor
 
         # MEMORY LEAK FIX: Configure scheduler with optimized settings
         # Memray analysis showed APScheduler's normalize() and _apply_jitter() causing

litellm/proxy/rag_endpoints/endpoints.py

@@ -824,7 +824,7 @@ async def rag_query(
         merged_retrieval_config: Final = {
             **retrieval_config,
             **store_data,
-        }  # mutable-ok: litellm.aquery requires a plain dict payload
+        }
 
         # Add litellm data
         request_data: dict[str, object] = {}

litellm/proxy/response_api_endpoints/endpoints.py

@@ -97,11 +97,7 @@ def _normalize_tool_dialect(
     tools: Final = data.get("tools")
     tool_choice: Final = data.get("tool_choice")
     normalized_tools: Final = (
-        [
-            _convert_tool_envelope(tool, to_chat=to_chat) for tool in tools
-        ]  # mutable-ok: body's tools stays a plain JSON list
-        if isinstance(tools, list)
-        else tools
+        [_convert_tool_envelope(tool, to_chat=to_chat) for tool in tools] if isinstance(tools, list) else tools
     )
     normalized_choice: Final = _convert_tool_envelope(tool_choice, to_chat=to_chat)
     if normalized_tools == tools and normalized_choice == tool_choice:

litellm/proxy/spend_tracking/carried_budget_state.py

@@ -36,9 +36,7 @@ def carry_team_and_user_budget_state(
 
 def carry_organization_budget_state(valid_token: UserAPIKeyAuth, org_table: LiteLLM_OrganizationTable) -> None:
     budget_table: Final = org_table.litellm_budget_table
-    valid_token.organization_alias = (
-        org_table.organization_alias
-    )  # rebind-ok: the request credential is pinned in place
+    valid_token.organization_alias = org_table.organization_alias
     valid_token.org_budget_snapshot = OrgBudgetSnapshot(  # rebind-ok: same object the caller keeps using
         spend=org_table.spend,
         max_budget=budget_table.max_budget if budget_table is not None else None,

litellm/proxy/utils.py

@@ -602,15 +602,11 @@ def _partition_post_call_callbacks() -> tuple[tuple[CustomGuardrail, ...], tuple
     return (guardrails, others)
 
 
-def _merge_pipeline_metadata_bucket(
-    data: dict, bucket_key: str, modified_bucket_value: object
-) -> None:  # mutable-ok: request payload dict, written in place
+def _merge_pipeline_metadata_bucket(data: dict, bucket_key: str, modified_bucket_value: object) -> None:
     if not isinstance(modified_bucket_value, dict):
         return
     modified_bucket: Final = cast("dict[str, object]", modified_bucket_value)  # cast-ok: metadata buckets are str-keyed
-    surviving_writes: Final = {
-        key: value for key, value in modified_bucket.items() if key != "guardrails"
-    }  # mutable-ok: merged into the live request metadata bucket in place
+    surviving_writes: Final = {key: value for key, value in modified_bucket.items() if key != "guardrails"}
     existing_bucket: Final = data.get(bucket_key)
     if isinstance(existing_bucket, dict):
         cast("dict[str, object]", existing_bucket).update(surviving_writes)  # cast-ok: metadata buckets are str-keyed
@@ -618,9 +614,7 @@ def _merge_pipeline_metadata_bucket(
         data[bucket_key] = surviving_writes
 
 
-def _merge_pipeline_metadata_writes(
-    data: dict, modified_data: Mapping[str, object]
-) -> None:  # mutable-ok: request payload dict, written in place
+def _merge_pipeline_metadata_writes(data: dict, modified_data: Mapping[str, object]) -> None:
     """
     Copy metadata-bucket writes from a pipeline's working copy back onto the request.
 
@@ -1052,7 +1046,6 @@ def _deployment_attribution_for_model_group(model_group: object, team_id: str |
     )
     return MappingProxyType(
         {
-            # mutable-ok: frozen immediately by the outer MappingProxyType
             **({"custom_llm_provider": shared_provider} if shared_provider is not None else {}),
             **(
                 {  # mutable-ok: frozen immediately by the outer MappingProxyType
@@ -1976,9 +1969,7 @@ class ProxyLogging:
         """
         scans_raw_request: Final = callback.scan_raw_request
         should_use_raw_snapshot: Final = scans_raw_request and raw_request_snapshot is not None
-        input_data: Final = (  # mutable-ok: same request-payload shape as data
-            independent_snapshot(raw_request_snapshot) if should_use_raw_snapshot else data
-        )
+        input_data: Final = independent_snapshot(raw_request_snapshot) if should_use_raw_snapshot else data
         # _process_guardrail_callback always calls mark_pre_call_hook_ran on a
         # successful run, which unconditionally stamps bookkeeping metadata onto
         # the dict regardless of whether the guardrail's own hook mutated
@@ -2169,9 +2160,7 @@ class ProxyLogging:
             if pipeline.mode != event_hook:
                 continue
 
-            step_input: dict = (
-                {**data, "response": current_response} if current_response is not None else data
-            )  # mutable-ok: same request-payload shape as data
+            step_input: dict = {**data, "response": current_response} if current_response is not None else data
 
             result: PipelineExecutionResult = await PipelineExecutor.execute_steps(
                 steps=pipeline.steps,

litellm/rerank_api/main.py

@@ -44,9 +44,7 @@ async def arerank(
     """
     Async: Reranks a list of documents based on their relevance to the query
     """
-    _custom_llm_provider: str | None = (
-        None  # rebind-ok: set by the declared-provider guard or the get_llm_provider unpack; read in the except
-    )
+    _custom_llm_provider: str | None = None
     try:
         loop: Final = asyncio.get_event_loop()
         kwargs["arerank"] = True

litellm/responses/additional_tools.py

@@ -37,12 +37,7 @@ def _tools_of_item(item: object) -> tuple[ALL_RESPONSES_API_TOOL_PARAMS, ...]:
         parsed: Final = _AdditionalToolsItem.model_validate(item)
     except ValidationError:
         return ()
-    return tuple(
-        cast(
-            "ALL_RESPONSES_API_TOOL_PARAMS", tool
-        )  # cast-ok: nested tools carry the same raw tool JSON as top-level tools
-        for tool in parsed.tools
-    )
+    return tuple(cast("ALL_RESPONSES_API_TOOL_PARAMS", tool) for tool in parsed.tools)
 
 
 def hoist_additional_tools(

litellm/responses/litellm_completion_transformation/transformation.py

@@ -866,14 +866,14 @@ class LiteLLMCompletionResponsesConfig:
             elif pending:
                 # Not followed by an assistant message — keep the reasoning
                 # standalone instead of dropping it.
-                merged.extend(  # mutable-ok: append reasoning messages
+                merged.extend(
                     [_standalone(text, blocks) for text, blocks in pending]  # mutable-ok: append reasoning messages
                 )
                 pending = []  # mutable-ok: reset accumulator
 
             merged.append(msg)
 
-        merged.extend(  # mutable-ok: append trailing reasoning
+        merged.extend(
             [_standalone(text, blocks) for text, blocks in pending]  # mutable-ok: append trailing reasoning
         )
 

litellm/responses/streaming_iterator.py

@@ -170,7 +170,7 @@ def _log_background_task_failure(task: asyncio.Task[object], *, task_name: str)
 
 
 _ERROR_CODE_HTTP_STATUS: Final[Mapping[str, int]] = MappingProxyType(
-    {  # mutable-ok: immediately frozen by MappingProxyType
+    {
         "server_error": 500,
         "rate_limit_exceeded": 429,
         "insufficient_quota": 429,
@@ -1633,9 +1633,7 @@ def _extract_frame_quota_estimate_inputs(msg_obj: Mapping[str, object]) -> tuple
     params: Final[Mapping[str, object]] = (
         nested
         if _is_json_object(nested) and nested
-        else MappingProxyType(  # mutable-ok: immediately frozen filtered frame
-            {k: v for k, v in msg_obj.items() if k != "type"}
-        )
+        else MappingProxyType({k: v for k, v in msg_obj.items() if k != "type"})
     )
     text_parts: Final[list[str]] = []  # mutable-ok: local accumulator built in one pass, not shared
     pending: Final[list[object]] = [  # mutable-ok: explicit worklist avoids recursion
@@ -2297,7 +2295,7 @@ class ResponsesWebSocketStreaming:
         except RateLimitError as e:
             try:
                 await self.websocket.send_text(
-                    json.dumps(  # mutable-ok: WebSocket wire payload requires JSON objects
+                    json.dumps(
                         {  # mutable-ok: WebSocket wire payload requires JSON objects
                             "type": "error",
                             "error": {  # mutable-ok: nested WebSocket error object
@@ -2743,9 +2741,7 @@ class ManagedResponsesWebSocketHandler:
         directly (before serialization) to avoid a redundant JSON round-trip on
         every chunk.  Returns the completed event dict, or ``None``.
         """
-        completed_event: _MutableJsonObject | None = (
-            None  # rebind-ok: captures the completed event once the stream yields it
-        )
+        completed_event: _MutableJsonObject | None = None
         stream_response: Final = await litellm.aresponses(model=model, **call_kwargs)
         async for chunk in stream_response:
             if chunk is None:

litellm/responses/utils.py

@@ -566,7 +566,7 @@ class ResponsesAPIRequestUtils:
             return
         items: Final = cast(list[object], request_input)  # cast-ok: untyped client json
         stripped: Final = tuple(ResponsesAPIRequestUtils._without_encrypted_reasoning(item) for item in items)
-        items[:] = (item for item in stripped if item is not None)  # rebind-ok: list shared with fallback snapshot
+        items[:] = (item for item in stripped if item is not None)
 
     @staticmethod
     def _without_encrypted_reasoning(item: object) -> object | None:

litellm/router.py

@@ -675,7 +675,7 @@ class RoutingArgs(enum.Enum):
 # entries their deployments own. Weak so a router nothing references any more, such
 # as the per-request one built from a caller-supplied user_config, drops out on its
 # own rather than leaving entries behind that nothing can withdraw.
-_live_routers: Final["weakref.WeakSet[Router]"] = weakref.WeakSet()  # mutable-ok: identity set of live routers
+_live_routers: Final["weakref.WeakSet[Router]"] = weakref.WeakSet()
 
 
 def _replay_live_router_model_cost() -> None:
@@ -2954,10 +2954,10 @@ class Router:
                         fallback_headers_are_settled = False
                         async for fallback_item in fallback_response:
                             if not fallback_headers_are_settled:
-                                fallback_headers_are_settled = True  # rebind-ok: one-shot latch
+                                fallback_headers_are_settled = True
                                 # a fallback that failed over again only repoints itself once it yields
-                                prepared_fallback_hidden_params = (  # rebind-ok: re-read once the fallback yields
-                                    Router._adopt_fallback_response_headers(wrapper_ref, fallback_response)
+                                prepared_fallback_hidden_params = Router._adopt_fallback_response_headers(
+                                    wrapper_ref, fallback_response
                                 )
                             Router._apply_fallback_hidden_params_to_item(fallback_item, prepared_fallback_hidden_params)
                             if (
@@ -3513,10 +3513,10 @@ class Router:
                         fallback_headers_are_settled = False
                         for fallback_item in fallback_response:
                             if not fallback_headers_are_settled:
-                                fallback_headers_are_settled = True  # rebind-ok: one-shot latch
+                                fallback_headers_are_settled = True
                                 # a fallback that failed over again only repoints itself once it yields
-                                prepared_fallback_hidden_params = (  # rebind-ok: re-read once the fallback yields
-                                    Router._adopt_fallback_response_headers(wrapper_ref, fallback_response)
+                                prepared_fallback_hidden_params = Router._adopt_fallback_response_headers(
+                                    wrapper_ref, fallback_response
                                 )
                             Router._apply_fallback_hidden_params_to_item(fallback_item, prepared_fallback_hidden_params)
                             if (
@@ -5459,23 +5459,23 @@ class Router:
                     if _anthropic_stream_should_drop_pre_content_ping(chunk, has_generated_content):
                         continue
                     if _anthropic_stream_commits_now(chunk, has_generated_content, len(buffered_lifecycle_chunks)):
-                        has_generated_content = True  # rebind-ok: real content seen, or the buffer cap was hit
+                        has_generated_content = True
                     # A transport can split one SSE data line across byte chunks, so pre-content
                     # detection parses the accumulated buffer plus the current chunk, never the
                     # chunk alone; the buffer is already capped, which bounds this window too.
-                    parse_window = (  # rebind-ok: freshly computed each iteration, never carried over
+                    parse_window = (
                         b"".join(c for c in (*buffered_lifecycle_chunks, chunk) if isinstance(c, (bytes, bytearray)))  # pyright: ignore[reportUnnecessaryIsInstance]  # bridge-path chunks are not always bytes at runtime
                         if not has_generated_content and isinstance(chunk, (bytes, bytearray))  # pyright: ignore[reportUnnecessaryIsInstance]  # bridge-path chunks are not always bytes at runtime
                         else chunk
                     )
                     error_event = parse_anthropic_error_event(parse_window)
-                    retriable_pending_error = (  # rebind-ok: freshly computed each iteration, never carried over
+                    retriable_pending_error = (
                         not has_generated_content
                         and error_event is not None
                         and _is_retriable_anthropic_status(error_event[2])
                         and not _anthropic_stream_error_is_gateway_verdict(chunk)
                     )
-                    refusal_stop_details = (  # rebind-ok: freshly computed each iteration, never carried over
+                    refusal_stop_details = (
                         parse_anthropic_refusal_stop_details(parse_window)
                         if not has_generated_content and error_event is None
                         else None
@@ -5493,7 +5493,7 @@ class Router:
                         buffered_lifecycle_chunks = (*buffered_lifecycle_chunks, chunk)
                         continue
                     if retriable_pending_error:
-                        assert error_event is not None  # guard-ok: retriable_pending_error implies this
+                        assert error_event is not None
                         _error_type, message, status_code = error_event
                         raise MidStreamFallbackError(
                             message=message,
@@ -10951,10 +10951,8 @@ class Router:
             model_group_info.supports_fast_mode = model_group_info.supports_fast_mode and (
                 AnthropicModelInfo.supports_fast_mode(litellm_model, llm_provider)
             )
-            deployment_reasoning_efforts = (
-                resolve_supported_reasoning_efforts(  # rebind-ok: recalculated per deployment
-                    model_info, deployment_is_mapped=deployment_is_mapped
-                )
+            deployment_reasoning_efforts = resolve_supported_reasoning_efforts(
+                model_info, deployment_is_mapped=deployment_is_mapped
             )
             if deployment_reasoning_efforts is None:
                 reasoning_efforts_unknown = True

litellm/router_strategy/complexity_router/complexity_router.py

@@ -1093,7 +1093,7 @@ def _with_classifier_forecast(
     if forecast is None:
         return decision
     verdict: Final = forecast.verdict
-    enriched: Final[StandardLoggingRoutingDecision] = {  # mutable-ok: routing decisions are JSON TypedDict records
+    enriched: Final[StandardLoggingRoutingDecision] = {
         **decision,
         "classifier_crux": verdict.crux,
         "classifier_primary_rule": verdict.primary_rule,
@@ -2484,7 +2484,7 @@ class ComplexityRouter(CustomLogger):
             {"role": "user", "content": opening_task},  # mutable-ok: SDK messages are dict-shaped
         ]
         if latest_follow_up is not None:
-            task_messages.append(  # mutable-ok: the provider SDK requires a concrete message list
+            task_messages.append(
                 {"role": "user", "content": latest_follow_up}  # mutable-ok: SDK messages are dict-shaped
             )
 

litellm/router_strategy/tag_based_routing.py

@@ -217,9 +217,7 @@ def _strip_routing_prefix(tags: Sequence[str], prefix: str) -> tuple[tuple[str,
 
 def _split_tags(tags: Sequence[str]) -> tuple[tuple[str, ...], list[str], tuple[str, ...]]:
     required: Final = tuple(tag[1:] for tag in tags if tag.startswith("&") and len(tag) > 1)
-    positive: Final = [
-        t for t in tags if not t.startswith("!") and not t.startswith("&")
-    ]  # mutable-ok: feeds _match_deployment's existing list[str]-typed request_tags param
+    positive: Final = [t for t in tags if not t.startswith("!") and not t.startswith("&")]
     excluded: Final = tuple(tag[1:] for tag in tags if tag.startswith("!") and len(tag) > 1)
     return required, positive, excluded
 

litellm/router_utils/auto_router_tuning_baseline.py

@@ -120,7 +120,7 @@ def snapshot_tuning_baselines(deployments: Iterable[Mapping[str, object]]) -> Ma
             if (pair := heuristic_v1_router_fingerprint(deployment)) is not None
             for identity, fingerprint in (pair,)
         }
-    )  # mutable-ok: MappingProxyType owns the completed immutable snapshot
+    )
 
 
 def is_mutable_tuned_candidate(candidate: Mapping[str, object], baselines: Mapping[str, str]) -> bool:

litellm/router_utils/fallback_event_handlers.py

@@ -655,7 +655,7 @@ async def run_async_fallback(
             # LOGGING
             kwargs = litellm_router.log_retry(kwargs=kwargs, e=original_exception)
             verbose_router_logger.info("Falling back to model_group = %s", mask_sensitive_structure(mg))
-            kwargs.pop("_target_order", None)  # rebind-ok: next hop must not inherit the previous order target
+            kwargs.pop("_target_order", None)
             if isinstance(mg, str):
                 kwargs["model"] = mg
             elif isinstance(mg, dict):

litellm/rust_bridge/lifecycle.py

@@ -46,7 +46,7 @@ class StreamClosed(Exception):
 async def _settle(execution: Execution, step: Step) -> Settled:
     while isinstance(step, Await):
         try:
-            value = await step.awaitable  # rebind-ok: each selected await produces the next protocol input
+            value = await step.awaitable
         except GeneratorExit:
             raise
         except BaseException as error:

litellm/rust_bridge/logger.py

@@ -53,7 +53,7 @@ def emit(
             extra={
                 "rust_target": target,
                 "rust_fields": dict(fields),
-            },  # mutable-ok: LogRecord requires JSON dict extras
+            },
         )
         _REDACTION.filter(record)
         _CORRELATION.filter(record)

litellm/types/management_endpoints/auto_router_endpoints.py

@@ -156,7 +156,7 @@ class AutoRouterRoutingTestRequest(BaseModel):
         the serving path.
         """
         return MappingProxyType(
-            {  # mutable-ok: MappingProxyType needs a dict to wrap
+            {
                 key: value
                 for key, value in (("messages", self.messages), ("system", self.system), ("tools", self.tools))
                 if value is not None