chore(lint): keep the leftover mutable-ok markers for a follow-up

Restore the ~1.4k `# mutable-ok` markers stripped in the previous commit
so this PR only touches the checker, its tests, the budget, and docs.
Those markers no longer suppress anything, so `# mutable-ok` is exempt
from LIT013 until a follow-up strips them.

enterprise/litellm_enterprise/enterprise_callbacks/secret_detection.py

@@ -616,7 +616,7 @@ class _ENTERPRISE_SecretDetection(CustomGuardrail):
             data["prompt"] = self.redact_text(prompt, source="prompt")
             return 1
         if isinstance(prompt, list):
-            data["prompt"] = [
+            data["prompt"] = [  # mutable-ok: data["prompt"] is a list on the wire
                 self.redact_text(item, source="prompt")
                 if isinstance(item, str) and item
                 else item

litellm/__init__.py

@@ -663,7 +663,7 @@ azure_anthropic_models: Set = set()
 azure_text_models: Set = set()
 anyscale_models: Set = set()
 cerebras_models: Set = set()
-nadir_models: Set = set()
+nadir_models: Set = set()  # mutable-ok: provider registry, filled from model_cost at import like every sibling provider
 galadriel_models: Set = set()
 nvidia_nim_models: Set = set()
 nvidia_riva_models: Set = set()
@@ -697,7 +697,7 @@ recraft_models: Set = set()
 cometapi_models: Set = set()
 oci_models: Set = set()
 vercel_ai_gateway_models: Set = set()
-edenai_models: Set = set()
+edenai_models: Set = set()  # mutable-ok: filled from the price map at import, like the sibling provider sets
 volcengine_models: Set = set()
 wandb_models: Set = set(WANDB_MODELS)
 ovhcloud_models: Set = set()

litellm/_logging.py

@@ -352,9 +352,13 @@ def _replace_string_leaves(value: object, values: Iterator[str]) -> object:
     if isinstance(value, str):
         return next(values)
     if isinstance(value, dict):
-        return {key: _replace_string_leaves(child, values) for key, child in value.items()}
+        return {  # mutable-ok: LogRecord extras must keep JSON dict shape for handlers
+            key: _replace_string_leaves(child, values) for key, child in value.items()
+        }
     if isinstance(value, list):
-        return [_replace_string_leaves(child, values) for child in value]
+        return [  # mutable-ok: LogRecord extras must keep JSON list shape for handlers
+            _replace_string_leaves(child, values) for child in value
+        ]
     if isinstance(value, tuple):
         return tuple(_replace_string_leaves(child, values) for child in value)
     return value
@@ -364,9 +368,13 @@ def _sort_processed_sets(original: object, processed: object) -> object:
     if isinstance(original, set) and isinstance(processed, list):
         return sorted(processed)
     if isinstance(original, dict) and isinstance(processed, dict):
-        return {key: _sort_processed_sets(original.get(key), value) for key, value in processed.items()}
+        return {  # mutable-ok: sorting nested sets must preserve the surrounding JSON dict
+            key: _sort_processed_sets(original.get(key), value) for key, value in processed.items()
+        }
     if isinstance(original, list) and isinstance(processed, list):
-        return [_sort_processed_sets(before, after) for before, after in zip(original, processed)]
+        return [  # mutable-ok: sorting nested sets must preserve the surrounding JSON list
+            _sort_processed_sets(before, after) for before, after in zip(original, processed)
+        ]
     if isinstance(original, tuple) and isinstance(processed, tuple):
         return tuple(_sort_processed_sets(before, after) for before, after in zip(original, processed))
     return processed

litellm/_redis.py

@@ -233,7 +233,7 @@ def _coerce_redis_kwargs_types(
             "socket_keepalive": bool,
         }
     )
-    result: Final = dict(redis_kwargs)
+    result: Final = dict(redis_kwargs)  # mutable-ok: per-key try/except coercion below needs to drop individual keys
     for key, value in redis_kwargs.items():
         if not isinstance(value, str):
             continue
@@ -803,7 +803,7 @@ def _credential_provider_auth_kwargs(redis_kwargs: dict) -> dict:
 
     superseded: Final = frozenset({"redis_connect_func", "username", "password"})
     kept: Final = ((k, v) for k, v in redis_kwargs.items() if k not in superseded)
-    return dict(kept, credential_provider=credential_provider)
+    return dict(kept, credential_provider=credential_provider)  # mutable-ok: the branches below mutate these kwargs
 
 
 def get_redis_client(**env_overrides):

litellm/a2a_protocol/main.py

@@ -145,7 +145,7 @@ def _a2a_cost_params(litellm_params: Mapping[str, object] | None) -> Mapping[str
 
 
 def _card_http_kwargs(extra_headers: dict[str, str] | None) -> dict[str, object] | None:
-    return {"headers": extra_headers} if extra_headers else None
+    return {"headers": extra_headers} if extra_headers else None  # mutable-ok: a2a-sdk's get_agent_card takes a dict
 
 
 def _agent_card_path(litellm_params: Mapping[str, object]) -> str | None:
@@ -612,7 +612,7 @@ def _build_streaming_logging_obj(
         logging_obj.model_call_details["agent_id"] = agent_id
 
     _request_context: Final = (("metadata", metadata), ("proxy_server_request", proxy_server_request))
-    _litellm_params: Final = dict(
+    _litellm_params: Final = dict(  # mutable-ok: Logging.litellm_params is declared as a dict
         (*_a2a_cost_params(litellm_params).items(), *((key, value) for key, value in _request_context if value))
     )
 

litellm/batches/batch_utils.py

@@ -127,7 +127,7 @@ async def _handle_completed_batch(
         return BatchCostUsageResult(
             cost=0.0,
             usage=Usage(prompt_tokens=0, completion_tokens=0, total_tokens=0),
-            models=[],
+            models=[],  # mutable-ok: no output file means no model was ever priced; BatchCostUsageResult.models requires list[str]
             successful_requests=0,
             failed_requests=await count_error_file_failed_requests(
                 batch, custom_llm_provider=custom_llm_provider, litellm_params=litellm_params

litellm/caching/affinity_cache.py

@@ -103,10 +103,10 @@ async def claim_affinity_pin(
         try:
             claim_script: Final = redis_cache.async_register_script(_CLAIM_PIN_SCRIPT)
             args: Final = (
-                json.dumps(dict(pin_value)),
+                json.dumps(dict(pin_value)),  # mutable-ok: JSON serialization requires dict, not a generic Mapping
                 int(ttl_seconds),
                 *(
-                    (json.dumps(tuple(dict(value) for value in eligible_values)),)
+                    (json.dumps(tuple(dict(value) for value in eligible_values)),)  # mutable-ok: JSON requires dict
                     if eligible_values is not None
                     else ()
                 ),

litellm/caching/caching_handler.py

@@ -702,14 +702,14 @@ class LLMCachingHandler:
         )
         merged: Final = EmbeddingResponse(
             model=cached.model,
-            data=[
+            data=[  # mutable-ok: EmbeddingResponse.data is a pydantic list field
                 item
                 if item is not None
                 else Embedding(embedding=next(fresh_items)["embedding"], index=position, object="embedding")
                 for position, item in enumerate(cached.data)
             ],
             usage=merged_usage,
-            hidden_params={
+            hidden_params={  # mutable-ok: EmbeddingResponse._hidden_params is a mutable dict field
                 **cached._hidden_params,
                 "cache_hit": True,
             },

litellm/caching/dual_cache.py

@@ -252,7 +252,9 @@ class DualCache(BaseCache):
                     if value is not None:
                         self.in_memory_cache.set_cache(key, value, **self._backfill_kwargs(kwargs))
 
-            return list(redis_result.get(key) if value is None else value for key, value in zip(keys, result))
+            return list(  # mutable-ok: public list contract
+                redis_result.get(key) if value is None else value for key, value in zip(keys, result)
+            )
         except Exception as e:
             log_redis_failure(
                 verbose_logger, logging.ERROR, "LiteLLM Cache: exception in batch_get_cache", e, with_traceback=True
@@ -327,8 +329,8 @@ class DualCache(BaseCache):
     def reserve_redis_batch_reads(self, keys: Sequence[str]) -> tuple[list[str], dict[str, float | None]]:
         """Reserve the memory-missed keys whose throttled Redis reads are due, as a batch read would."""
         if self.redis_cache is None:
-            return [], {}
-        key_list: Final = list(keys)
+            return [], {}  # mutable-ok: API contract returns an empty list and dictionary
+        key_list: Final = list(keys)  # mutable-ok: batch_get_cache takes a list
         memory: Final = self.in_memory_cache
         in_memory_result: Final = (
             None
@@ -384,7 +386,7 @@ class DualCache(BaseCache):
 
     async def declare_batch_get(self, keys: Sequence[str], batch: RedisBatch) -> DeclaredBatchRead:
         pending: Final = await self._prepare_batch_get(
-            list(keys),
+            list(keys),  # mutable-ok: the shared batch read takes a list
             local_only=False,
             throttle_redis=False,
         )
@@ -631,7 +633,7 @@ class DualCache(BaseCache):
         parent_otel_span: Span | None = None,
     ) -> None:
         batch: Final = None if self.redis_cache is None else active_post_call_redis_batch(self.redis_cache)
-        operations: Final = list(increment_list)
+        operations: Final = list(increment_list)  # mutable-ok: both increment pipelines take a list
         if batch is None:
             await self.async_increment_cache_pipeline(operations, parent_otel_span=parent_otel_span)
             return

litellm/caching/evicted_client_closer.py

@@ -238,7 +238,7 @@ class EvictedClientCloser:
         the front rather than having to be searched for.
         """
         with self._queue_lock:
-            bucket: Final = self._buckets.setdefault(_bucket_key(pending), deque())
+            bucket: Final = self._buckets.setdefault(_bucket_key(pending), deque())  # mutable-ok: FIFO by design
             while bucket and bucket[0].client_ref() is None:
                 bucket.popleft()
                 self._pending_count -= 1

litellm/caching/redis_batch.py

@@ -33,7 +33,7 @@ from litellm.types.services import ServiceTypes
 
 _T = TypeVar("_T")
 _ScriptArg = str | bytes | int | float
-SettledHook = Callable[[asyncio.Future[_T]], Awaitable[None] | None]
+SettledHook = Callable[[asyncio.Future[_T]], Awaitable[None] | None]  # mutable-ok: Callable params
 POST_CALL_FLUSH_DEADLINE_SECONDS: Final = 1.0
 
 
@@ -139,7 +139,7 @@ class _MGet(_Op[Mapping[str, object]]):
         )
 
     async def run_alone(self) -> Mapping[str, object]:
-        found: Mapping[str, object] = await self._redis_cache.async_batch_get_cache(key_list=list(self._keys))  # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType]  # untyped cache API
+        found: Mapping[str, object] = await self._redis_cache.async_batch_get_cache(key_list=list(self._keys))  # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType]  # untyped cache API  # mutable-ok: the cache API takes a list
         if any(key not in found for key in self._keys):
             raise ConnectionError("batch get did not return every key")
         return found

litellm/completion_extras/litellm_responses_transformation/transformation.py

@@ -123,13 +123,13 @@ def _reasoning_input_items(msg: "AllMessageValues") -> list[dict[str, object]]:
     blocks are the fallback for turns that arrived over another API surface.
     """
     items: Final = _get_reasoning_items(msg)
-    stored: Final = [_reasoning_item_to_response_input(item) for item in items]
+    stored: Final = [_reasoning_item_to_response_input(item) for item in items]  # mutable-ok: API message payload
     if stored:
         return stored
     raw_blocks: Final = msg.get("thinking_blocks") or ()
     blocks: Final = cast("Iterable[ChatCompletionThinkingBlock]", raw_blocks)  # cast-ok: untyped client json
     replayed: Final = responses_reasoning_items_from_thinking_blocks(blocks)
-    return [dict(item) for item in replayed]
+    return [dict(item) for item in replayed]  # mutable-ok: API message payload
 
 
 def _build_reasoning_item(
@@ -441,7 +441,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
                 input_items.extend(_reasoning_input_items(msg))
                 if content:
                     input_items.append(
-                        {
+                        {  # mutable-ok: API message payload
                             "type": "message",
                             "role": "assistant",
                             "content": self._convert_content_to_responses_format(content, "assistant"),
@@ -475,7 +475,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
                 if role == "assistant":
                     input_items.extend(_reasoning_input_items(msg))
                 input_items.append(
-                    {
+                    {  # mutable-ok: API message payload
                         "type": "message",
                         "role": role,
                         "content": self._convert_content_to_responses_format(content, cast(str, role)),
@@ -531,11 +531,11 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
     ) -> "ResponseText":
         existing: Final = cast(  # cast-ok: text field is a ResponseText | dict[str, Any] | None union
             "dict[str, object]",
-            dict(responses_api_request).get("text") or {},
+            dict(responses_api_request).get("text") or {},  # mutable-ok: one-shot merge seed
         )
         return cast(  # cast-ok: merged mapping is a valid ResponseText shape
             "ResponseText",
-            {**existing, **update},
+            {**existing, **update},  # mutable-ok: one-shot merged payload
         )
 
     def _build_sanitized_litellm_params(self, litellm_params: dict) -> dict[str, object]:
@@ -1506,7 +1506,7 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
                     # tool call; per-stream callers already received it via
                     # output_item.added and the argument delta events
                     return ModelResponseStream(
-                        choices=[
+                        choices=[  # mutable-ok: ModelResponseStream coerces only list choices
                             StreamingChoices(
                                 index=0,
                                 delta=Delta(
@@ -1612,7 +1612,7 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
                     )
                 ],
                 usage=usage,
-                provider_specific_fields=dict(provider_metadata) or None,
+                provider_specific_fields=dict(provider_metadata) or None,  # mutable-ok: field is typed dict
                 **(
                     MappingProxyType({"service_tier": served_service_tier})
                     if isinstance(served_service_tier, str)

litellm/cost_calculator.py

@@ -2874,7 +2874,9 @@ class ResponsesWebSocketTokenUsageProcessor(BaseTokenUsageProcessor):
         collected_usage_objects: Final = ResponsesWebSocketTokenUsageProcessor.collect_usage_from_responses_ws_results(
             results
         )
-        return ResponsesWebSocketTokenUsageProcessor.combine_usage_objects(list(collected_usage_objects))
+        return ResponsesWebSocketTokenUsageProcessor.combine_usage_objects(
+            list(collected_usage_objects)  # mutable-ok: combine_usage_objects requires a list parameter
+        )
 
 
 _TRANSCRIPTION_COMPLETED_EVENT_TYPE: Final = "conversation.item.input_audio_transcription.completed"

litellm/experimental_mcp_client/client.py

@@ -1136,7 +1136,7 @@ class MCPClient:
         async def _list_resource_templates_operation(session: ClientSession) -> ListResourceTemplatesResult:
             capabilities: Final = session.server_capabilities
             if capabilities is not None and capabilities.resources is None:
-                return ListResourceTemplatesResult(resource_templates=[])
+                return ListResourceTemplatesResult(resource_templates=[])  # mutable-ok: MCP result payload
             try:
                 return ListResourceTemplatesResult(
                     resource_templates=await self._list_optional_pages(
@@ -1150,7 +1150,7 @@ class MCPClient:
                 verbose_logger.debug(
                     "MCP client list_resource_templates is unsupported by %s: %s", self.server_url or "stdio", error
                 )
-                return ListResourceTemplatesResult(resource_templates=[])
+                return ListResourceTemplatesResult(resource_templates=[])  # mutable-ok: MCP result payload
 
         try:
             result: Final = await self.run_with_session(_list_resource_templates_operation)

litellm/experimental_mcp_client/tools.py

@@ -171,7 +171,9 @@ async def load_mcp_tools(
     """
     tools: Final = await list_tools_with_pagination(session)
     if format == "openai":
-        return [transform_mcp_tool_to_openai_tool(mcp_tool=tool) for tool in tools]
+        return [  # mutable-ok: public API returns a list
+            transform_mcp_tool_to_openai_tool(mcp_tool=tool) for tool in tools
+        ]
     return tools
 
 

litellm/integrations/SlackAlerting/slack_alerting.py

@@ -1131,7 +1131,7 @@ Model Info:
             message=message,
             level=level,
             alert_type=AlertType.model_deprecation_warnings,
-            alerting_metadata={
+            alerting_metadata={  # mutable-ok: send_alert takes a dict payload
                 "deprecated_count": len(snapshot.deprecated),
                 "imminent_count": len(snapshot.imminent),
                 "upcoming_count": len(snapshot.upcoming),
@@ -1245,8 +1245,8 @@ Model Info:
         try:
             existing_invitations: Final = TypeAdapter(list[InvitationModel]).validate_python(
                 await InvitationLinkRepository(prisma_client).table.find_many(  # pyright: ignore[reportAny]  # untyped prisma boundary (any-ok), result validated by TypeAdapter
-                    where={"user_id": recipient_user_id},
-                    order={"created_at": "desc"},
+                    where={"user_id": recipient_user_id},  # mutable-ok: prisma find_many requires a dict where filter
+                    order={"created_at": "desc"},  # mutable-ok: prisma find_many requires a dict order arg
                 ),
                 from_attributes=True,
             )
@@ -2011,7 +2011,7 @@ Model Info:
                     message="\n\n".join(event.message for event in typed_events),
                     level="High",
                     alert_type=alert_type,
-                    alerting_metadata={},
+                    alerting_metadata={},  # mutable-ok: send_alert takes a dict payload
                 )
                 for event in typed_events:
                     await self.internal_usage_cache.async_set_cache(

litellm/integrations/azure_sentinel/azure_sentinel.py

@@ -337,7 +337,7 @@ class AzureSentinelLogger(CustomBatchLogger):
             Raises a NON Blocking verbose_logger.exception if an error occurs
         """
         batch_to_send: Final = tuple(self.log_queue)
-        self.log_queue = []
+        self.log_queue = []  # mutable-ok: queue ownership is detached before the async send
         try:
             undelivered: Final = await self._async_send_batch_to_api(
                 log_queue=batch_to_send,
@@ -360,7 +360,7 @@ class AzureSentinelLogger(CustomBatchLogger):
         Sends the batch of audit logs to Azure Monitor Logs Ingestion API
         """
         batch_to_send: Final = tuple(self.audit_log_queue)
-        self.audit_log_queue = []
+        self.audit_log_queue = []  # mutable-ok: queue ownership is detached before the async send
         try:
             undelivered: Final = await self._async_send_batch_to_api(
                 log_queue=batch_to_send,
@@ -384,7 +384,7 @@ class AzureSentinelLogger(CustomBatchLogger):
         queue: list[_QueuedPayload],
         log_type: str,
     ) -> list[_QueuedPayload]:
-        merged: Final = [*undelivered, *queue]
+        merged: Final = [*undelivered, *queue]  # mutable-ok: queue trimming returns a mutable logger queue
         overflow: Final = len(merged) - self.max_queue_size
         if overflow <= 0:
             return merged

litellm/integrations/clickhouse/clickhouse_spend_logger.py

@@ -58,7 +58,7 @@ def _json(value: object) -> str:
 
 
 def _json_mapping(value: Mapping[str, Any]) -> str:
-    return _json(dict(value))
+    return _json(dict(value))  # mutable-ok: [LIT002] JSON serialization requires a dict
 
 
 def _find_traceparent(metadata: Mapping[str, Any], kwargs: Mapping[str, Any]) -> tuple[str, str]:
@@ -88,8 +88,8 @@ def _cache_tokens(usage: Mapping[str, Any]) -> tuple[int, int]:
 
 def _request_tags(value: object) -> list[str]:
     if not isinstance(value, list):
-        return []
-    return [str(tag) for tag in value]
+        return []  # mutable-ok: [LIT002] empty spend-log tag payload
+    return [str(tag) for tag in value]  # mutable-ok: [LIT002] SpendLogRecord schema
 
 
 def _session_id(payload: StandardLoggingPayload, kwargs: Mapping[str, Any]) -> str:
@@ -165,6 +165,6 @@ class ClickHouseSpendLogger(ClickHouseBatchLogger):
             if payload is None or _is_trace_ingest(payload):
                 return
             row: Final = spend_log_row_from_payload(payload, kwargs)
-            self.enqueue([dict(row)])
+            self.enqueue([dict(row)])  # mutable-ok: [LIT002] batch logger API
         except Exception as e:
             verbose_logger.exception("ClickHouseSpendLogger: failed to log request: %s", e)

litellm/integrations/custom_guardrail.py

@@ -363,7 +363,7 @@ class CustomGuardrail(CustomLogger):
             land and degrade to blocking instead of silently letting the
             flagged request through unmodified.
         """
-        advisory_message: Final = {"role": "system", "content": message}
+        advisory_message: Final = {"role": "system", "content": message}  # mutable-ok: plain dict for live request
         existing_messages: Final = data.get("messages")
         existing_input: Final = data.get("input")
         existing_instructions: Final = data.get("instructions")
@@ -374,7 +374,7 @@ class CustomGuardrail(CustomLogger):
             # model to disregard a trailing warning. Prefer it over "input"
             # whenever present.
             if isinstance(existing_messages, list):
-                messages_with_instructions_note: Final = [
+                messages_with_instructions_note: Final = [  # mutable-ok: fresh list
                     *existing_messages,
                     advisory_message,
                 ]
@@ -386,7 +386,7 @@ class CustomGuardrail(CustomLogger):
             # real, read field (e.g. a chat-completions call carrying a stray
             # "input"), so write to both when both are present.
             if isinstance(existing_messages, list):
-                messages_with_input_note: Final = [*existing_messages, advisory_message]
+                messages_with_input_note: Final = [*existing_messages, advisory_message]  # mutable-ok: fresh list
                 data["messages"] = messages_with_input_note  # rebind-ok: mutates caller's dict by design
             # The Responses API reads "input", not "messages" -- appending only to
             # "messages" would leave the advisory unreachable for that endpoint.
@@ -400,10 +400,10 @@ class CustomGuardrail(CustomLogger):
             # non-delivery so the caller degrades to blocking.
             return False
         if isinstance(existing_messages, list):
-            messages_without_input_note: Final = [*existing_messages, advisory_message]
+            messages_without_input_note: Final = [*existing_messages, advisory_message]  # mutable-ok: fresh list
             data["messages"] = messages_without_input_note  # rebind-ok: mutates caller's dict by design
             return True
-        sole_message: Final = [advisory_message]
+        sole_message: Final = [advisory_message]  # mutable-ok: plain list for the live JSON request
         data["messages"] = sole_message  # rebind-ok: mutates caller's dict by design
         return True
 

litellm/integrations/datadog/datadog.py

@@ -397,7 +397,7 @@ class DataDogLogger(
                 verbose_logger.debug("[DATADOG MOCK] Batch of %s events successfully mocked", len(batch_to_send))
 
         except BatchSendCancelled as cancelled:
-            self.log_queue = list(cancelled.undelivered) + self.log_queue
+            self.log_queue = list(cancelled.undelivered) + self.log_queue  # mutable-ok: logger queue remains appendable
             raise asyncio.CancelledError() from cancelled
         except Exception as e:
             self.log_queue = batch_to_send + self.log_queue
@@ -425,7 +425,7 @@ class DataDogLogger(
             drop_error_message=DD_ERRORS.DATADOG_413_ERROR.value,
             non_success_handler=requeue_after_http_error,
         )
-        return list(undelivered)
+        return list(undelivered)  # mutable-ok: caller prepends records to the logger queue
 
     @staticmethod
     def _exceeds_intake_limits(chunk: Sequence[DatadogPayload]) -> bool:

litellm/integrations/datadog/datadog_llm_obs.py

@@ -131,7 +131,7 @@ def _guardrail_entry_without_prompt_carriers(entry: Mapping[str, object]) -> Map
     Built as an allow-list rather than a deny-list: a key neither set classifies is dropped, so a
     guardrail that records its own extra detail cannot put the caller's prompt on a redacted span.
     """
-    return {
+    return {  # mutable-ok: a fresh record built per entry, handed straight to the span serializer
         field: REDACTED_BY_LITELM_STRING if field in PROMPT_CARRYING_GUARDRAIL_FIELDS else value
         for field, value in entry.items()
         if field in _CLASSIFIED_GUARDRAIL_FIELDS

litellm/integrations/langfuse/langfuse.py

@@ -868,8 +868,10 @@ class LangFuseLogger:
                 "id": clean_metadata.pop("generation_id", generation_id),
                 "input": masked_input if not mask_input else "redacted-by-litellm",
                 "output": masked_output if not mask_output else "redacted-by-litellm",
-                "cost_details": {"total": cost} if usage is not None and isinstance(cost, (int, float)) else None,
-                "metadata": {
+                "cost_details": {"total": cost}  # mutable-ok: langfuse serializes this payload
+                if usage is not None and isinstance(cost, (int, float))
+                else None,
+                "metadata": {  # mutable-ok: langfuse serializes this payload, a proxy is not json-encodable
                     **log_requester_metadata(redact_user_api_key_info(metadata=allowlisted_metadata)),  # pyright: ignore[reportArgumentType]  # TypedDict in, plain metadata dict out
                     **enrichments,
                     **_lookup_ids(litellm_call_id, response_obj),

litellm/integrations/newrelic/newrelic_metrics.py

@@ -109,7 +109,7 @@ def _metric_record_from_payload(standard_logging_object: StandardLoggingPayload)
 
 def _bucket_metrics(bucket_records: tuple[NewRelicMetricRecord, ...]) -> tuple[NewRelicMetric, ...]:
     first: Final = bucket_records[0]
-    attributes: Final[Mapping[str, str]] = {
+    attributes: Final[Mapping[str, str]] = {  # mutable-ok: JSON leaf; safe_dumps stringifies MappingProxyType
         key: value[:NEWRELIC_METRIC_ATTRIBUTE_MAX_LEN]
         for key, value in (
             ("team_id", first.team_id),
@@ -150,7 +150,7 @@ def _team_budget_gauges(record: NewRelicMetricRecord) -> tuple[NewRelicMetric, .
     team_max_budget: Final = record.team_max_budget
     if team_max_budget is None:
         return ()
-    attributes: Final[Mapping[str, str]] = {
+    attributes: Final[Mapping[str, str]] = {  # mutable-ok: JSON leaf; safe_dumps stringifies MappingProxyType
         key: value[:NEWRELIC_METRIC_ATTRIBUTE_MAX_LEN]
         for key, value in (("team_id", record.team_id), ("team_alias", record.team_alias))
         if value
@@ -265,7 +265,7 @@ class NewRelicMetricsLogger(CustomBatchLogger):
                         dropped,
                         NEWRELIC_METRICS_MAX_DRAIN_PASSES,
                     )
-                self.log_queue[:] = list(survivors)
+                self.log_queue[:] = list(survivors)  # mutable-ok: leave late arrivals for the next serialized drain
 
     async def _drain_flush_once(self) -> None:
         """Attempt every queued record once, in ``batch_size`` chunks, without

litellm/integrations/otel/model/metadata.py

@@ -384,7 +384,7 @@ def metadata_from_request_data(data: object) -> Mapping[str, object] | None:
 
 def flatten_metadata(raw: Mapping[str, object]) -> Iterator[tuple[str, str]]:
     """Scalar leaves of a nested metadata mapping, keyed by their dotted path."""
-    stack: Final = list(tuple(raw.items())[::-1])
+    stack: Final = list(tuple(raw.items())[::-1])  # mutable-ok: iterative worklist keeps the walk off the call stack
     while stack:
         key, value = stack.pop()
         if (nested := as_str_mapping(value)) is not None:

litellm/integrations/otel/model/payloads.py

@@ -803,7 +803,7 @@ def _joined_choice(parts: tuple[str, ...]) -> tuple[_Choice, ...]:
 def _text_completion_choice(choice: Mapping[str, object], text: str) -> Mapping[str, object]:
     synthesized: Final = _text_choice(text, as_str(choice.get("finish_reason")))
     merged: Final = (*choice.items(), *synthesized.items())
-    return {k: v for k, v in merged if k != "text"}
+    return {k: v for k, v in merged if k != "text"}  # mutable-ok: mappers json.dumps and isinstance(dict) it
 
 
 def _completion_choices(response: Mapping[str, object]) -> tuple[Mapping[str, object], ...]:

litellm/integrations/otel/model/request_io.py

@@ -79,7 +79,7 @@ def stream_output(chunks: Sequence[object], data: Mapping[str, object]) -> str |
 def _assembled_chat_stream(chunks: Sequence[object], data: Mapping[str, object]) -> object:
     try:
         return litellm.stream_chunk_builder(  # pyright: ignore[reportUnknownMemberType]  # upstream types chunks as a bare list
-            chunks=list(chunks),
+            chunks=list(chunks),  # mutable-ok: stream_chunk_builder takes a list
             messages=_MESSAGES.validate_python(data.get("messages")),
         )
     except (litellm.APIError, ValidationError):

litellm/integrations/otel/plumbing/context.py

@@ -380,8 +380,10 @@ def inject_trace_context(headers: Mapping[str, str], parent_span: object = None)
     """
     context: Final = _outgoing_trace_context(parent_span)
     if context is None:
-        return dict(headers)
-    carrier: Final = {key: value for key, value in headers.items() if key.lower() not in _W3C_TRACE_HEADERS}
+        return dict(headers)  # mutable-ok: OpenTelemetry propagator requires a mutable carrier
+    carrier: Final = {  # mutable-ok: OpenTelemetry propagator requires a mutable carrier
+        key: value for key, value in headers.items() if key.lower() not in _W3C_TRACE_HEADERS
+    }
     _PROPAGATOR.inject(carrier, context=_propagated_context(headers, context))
     return carrier
 

litellm/integrations/otel/plumbing/providers.py

@@ -636,7 +636,9 @@ class TenantFanOutSpanProcessor(SpanProcessor):
             live: Final = tuple((id(p), p) for p in (*self._processors.values(), *self._retired.values()))
             closing: Final = tuple(p for ident, p in live if ident not in self._exporting)
             self._processors.clear()
-            self._retired = OrderedDict((ident, p) for ident, p in live if ident in self._exporting)
+            self._retired = OrderedDict(  # mutable-ok: the same bounded map, keeping only what is still exporting
+                (ident, p) for ident, p in live if ident in self._exporting
+            )
         for processor in closing:
             self._drain.submit(processor)
         self._drain.close(timeout=max(0.0, deadline - time.monotonic()))

litellm/integrations/otel/plumbing/routing.py

@@ -171,7 +171,9 @@ class TenantTracerCache:
         # thread-pool workers concurrently with the event loop, so cache
         # updates, span counts, and retirement must be atomic.
         self._lock: Final = threading.Lock()
-        self._providers: OrderedDict[_RouteKey, TracerProvider] = OrderedDict()
+        self._providers: OrderedDict[_RouteKey, TracerProvider] = (
+            OrderedDict()  # mutable-ok: bounded LRU; eviction needs in-place ordered mutation
+        )
         self._open_span_counts: dict[TracerProvider, int] = {}  # mutable-ok: live refcount state
         # Oldest-first so an overflow of draining providers sheds the stalest.
         self._retired: OrderedDict[TracerProvider, None] = OrderedDict()  # mutable-ok: draining evicted providers
@@ -391,7 +393,7 @@ class TenantTracerCache:
             if project_headers and kind not in _GRPC_KINDS
             else base
         )
-        update: Final = {
+        update: Final = {  # mutable-ok: model_copy(update=...) requires a plain dict
             field: value
             for field, value in (("headers", routed), ("endpoint", endpoint))
             if (field == "headers" and routed != spec.headers)

litellm/integrations/otel/presets/langfuse.py

@@ -30,7 +30,7 @@ def langfuse_preset(
         if not allow_missing_credentials:
             raise
         return base.model_copy(
-            update={
+            update={  # mutable-ok: pydantic model_copy takes a plain update mapping
                 "exporters": credential_gated_exporters(base.exporters, ExporterOwner.LANGFUSE_OTEL),
                 "mapper_names": mappers,
             }

litellm/integrations/otel/presets/signoz.py

@@ -91,5 +91,5 @@ def signoz_dynamic_headers(
 ) -> dict[str, str]:  # mutable-ok: DYNAMIC_HEADERS_BY_CALLBACK returns a dict
     key: Final = params.get("signoz_ingestion_key")
     if _tenant_endpoint_is_unusable(params) or not key:
-        return {}
-    return {"signoz-ingestion-key": key}
+        return {}  # mutable-ok: same registry contract
+    return {"signoz-ingestion-key": key}  # mutable-ok: same registry contract

litellm/integrations/otel/presets/weave.py

@@ -31,7 +31,7 @@ def weave_preset(
         if not allow_missing_credentials:
             raise
         return base.model_copy(
-            update={
+            update={  # mutable-ok: pydantic model_copy takes a plain update mapping
                 "exporters": credential_gated_exporters(base.exporters, ExporterOwner.WEAVE_OTEL),
                 "mapper_names": mappers,
             }

litellm/integrations/pointfive/logger.py

@@ -207,7 +207,9 @@ class PointFiveLogger(CustomBatchLogger):
         the excluded-field list and this callback's own setting are applied here, then the
         global, per-request and header settings that only the framework's predicate knows.
         """
-        details: Final = self.redact_standard_logging_payload_from_model_call_details(dict(kwargs))
+        details: Final = self.redact_standard_logging_payload_from_model_call_details(
+            dict(kwargs)  # mutable-ok: both framework helpers take the call details as a dict
+        )
         payload: Final = details.get("standard_logging_object")
         if not isinstance(payload, dict):
             return None

litellm/integrations/pointfive/upload_client.py

@@ -147,7 +147,7 @@ class PointFiveUploadClient:
             response: Final = await self.http_client.post(
                 self.api_url + path,
                 json=request.model_dump(by_alias=True),
-                headers={
+                headers={  # mutable-ok: AsyncHTTPHandler.post types headers as dict
                     "Authorization": f"Bearer {self.api_key}",
                     "Content-Type": "application/json",
                 },
@@ -172,7 +172,7 @@ class PointFiveUploadClient:
         if isinstance(destination, PointFiveUploadFailure):
             return destination
         url, host = destination
-        headers: Final = dict(PUT_HEADERS, Host=host) if host else dict(PUT_HEADERS)
+        headers: Final = dict(PUT_HEADERS, Host=host) if host else dict(PUT_HEADERS)  # mutable-ok: put wants dict
         try:
             await self.http_client.put(url, data=body, headers=headers, follow_redirects=False)
         except httpx.HTTPStatusError as e:

litellm/integrations/prometheus.py

@@ -1195,7 +1195,7 @@ class PrometheusLogger(CustomLogger):
                 return metric_class(*args, **kwargs)
 
             kept: Final = tuple(name for name in original_labelnames if name not in self.exclude_labels)
-            kept_kwargs: Final = {**kwargs, "labelnames": kept}
+            kept_kwargs: Final = {**kwargs, "labelnames": kept}  # mutable-ok: ** needs a mapping to override labelnames
             real_metric: Final = metric_class(*args, **kept_kwargs)
             return _ExcludedLabelMetric(real_metric, original_labelnames, self.exclude_labels)
 

litellm/integrations/s3_v2.py

@@ -628,7 +628,7 @@ class S3Logger(CustomBatchLogger, BaseAWSLLM):
         #########################################################
         uploads: Final = self._batch_file_elements(batch) if self._batch_file_mode_active() else batch
         self._flush_retries = 0
-        self._flush_dropped = {}
+        self._flush_dropped = {}  # mutable-ok: per-flush drop marks read back by _upload_bounded
         stale: Final = min(self._requeued_count, len(uploads)) if len(uploads) == len(batch) else 0
         order: Final = (*range(stale, len(uploads)), *range(stale))
         ordered: Final = await asyncio.gather(*(self._upload_outcome(uploads[i]) for i in order))
@@ -680,7 +680,7 @@ class S3Logger(CustomBatchLogger, BaseAWSLLM):
                 self.max_queue_size,
                 overflow,
             )
-        self.log_queue = [
+        self.log_queue = [  # mutable-ok: log_queue is the flush buffer shared with custom_batch_logger
             *requeued,
             *arrivals,
         ][overflow:]

litellm/integrations/shadow_eval_logger.py

@@ -357,7 +357,11 @@ class GuardrailRequestSnapshot:
         if fingerprint is None:
             return None
         return GuardrailRequestSnapshot(
-            body=MappingProxyType(_CHAT_REQUEST_ADAPTER.validate_python(independent_snapshot(dict(body)))),
+            body=MappingProxyType(
+                _CHAT_REQUEST_ADAPTER.validate_python(
+                    independent_snapshot(dict(body))  # mutable-ok: snapshot helper requires a plain dictionary
+                )
+            ),
             fingerprint=fingerprint,
         )
 
@@ -809,7 +813,7 @@ def _as_active_job(record: object, attempts: int, spend: float) -> ActiveShadowE
     except ValidationError as e:
         verbose_logger.debug("shadow_eval: skipping unsamplable job row: %s", e)
         return None
-    return job.model_copy(update={"attempts": attempts, "spend": spend})
+    return job.model_copy(update={"attempts": attempts, "spend": spend})  # mutable-ok: pydantic update payload
 
 
 _jobs_cache: Final = InMemoryCache(max_size_in_memory=4, default_ttl=_JOBS_CACHE_TTL_SECONDS)
@@ -860,9 +864,9 @@ class ShadowEvalLogger(CustomLogger):
             return _EMPTY_JOBS
         try:
             records: Final = await prisma.db.litellm_shadowevaljob.find_many(
-                where={
+                where={  # mutable-ok: Prisma filter
                     "stopped_at": None,
-                    "ends_at": {"gt": datetime.now(timezone.utc)},
+                    "ends_at": {"gt": datetime.now(timezone.utc)},  # mutable-ok: Prisma filter
                 },
             )
             grouped: Final = (
@@ -870,12 +874,12 @@ class ShadowEvalLogger(CustomLogger):
                     by=["job_id"],
                     count=True,
                     sum={"judge_cost": True, "shadow_cost": True, "shadow_classifier_cost": True},
-                    where={"job_id": {"in": [str(record.id) for record in records]}},
+                    where={"job_id": {"in": [str(record.id) for record in records]}},  # mutable-ok: Prisma filter
                 )
                 if records
                 else ()
             )
-            attempt_stats: Final = {
+            attempt_stats: Final = {  # mutable-ok: frozen snapshot of the grouped read
                 str(row["job_id"]): (
                     int(row["_count"]["_all"]),
                     _leg_eval_spend(row["_sum"] or _EMPTY_METADATA),
@@ -948,13 +952,13 @@ class ShadowEvalLogger(CustomLogger):
             payload: Final[StandardLoggingPayload | None] = kwargs.get("standard_logging_object")  # pyright: ignore[reportAssignmentType]  # untyped callback kwargs
             if payload is None:
                 return
-            raw_meta: Final = get_litellm_metadata_from_kwargs(dict(kwargs))
+            raw_meta: Final = get_litellm_metadata_from_kwargs(dict(kwargs))  # mutable-ok: helper needs dict
             request_metadata: Final = raw_meta if isinstance(raw_meta, Mapping) else _EMPTY_METADATA
             if request_metadata.get(INTERNAL_CALL_ORIGIN_METADATA_KEY):
                 return  # internal sub-call (our own shadow/judge, a classifier), not user traffic
             # redaction rewrites logged content before callbacks run, so this hook
             # only ever sees placeholders for a redacted request
-            if should_redact_message_logging(dict(kwargs)):
+            if should_redact_message_logging(dict(kwargs)):  # mutable-ok: predicate takes a plain dict
                 return
             metadata: Final = payload.get("metadata") or _EMPTY_METADATA
             # Each identity the request resolved to is a candidate target; JWT-auth
@@ -995,7 +999,7 @@ class ShadowEvalLogger(CustomLogger):
             sample: Final = _judgeable_sample(
                 ops,
                 sample_kwargs,
-                MappingProxyType(dict(payload.get("model_parameters") or {})),
+                MappingProxyType(dict(payload.get("model_parameters") or {})),  # mutable-ok: frozen snapshot
                 response_obj,
             )
             if sample is None:
@@ -1242,7 +1246,7 @@ class ShadowEvalLogger(CustomLogger):
             return
         try:
             await prisma.db.litellm_shadowevalattempt.create(
-                data={
+                data={  # mutable-ok: Prisma payload
                     "job_id": job.id,
                     "request_id": request_id,
                     "router_name": router_name,
@@ -1283,10 +1287,12 @@ class ShadowEvalLogger(CustomLogger):
         try:
             response: Final = await router.acompletion(
                 model=target_model,
-                messages=[dict(m) for m in messages],  # pyright: ignore[reportArgumentType]  # snapshot of the SDK's own message dicts
+                messages=[  # mutable-ok: provider transforms rewrite messages in place, so the router gets its own copy
+                    dict(m) for m in messages
+                ],  # pyright: ignore[reportArgumentType]  # snapshot of the SDK's own message dicts
                 metadata=shadow_metadata,
                 num_retries=0,
-                fallbacks=[],
+                fallbacks=[],  # mutable-ok: SDK kwarg; a failed shadow is a recorded error, never a spend multiplier
                 **shadow_params,
             )
         except Exception as e:  # noqa: BLE001  # provider errors become error rows, not crashes
@@ -1335,8 +1341,8 @@ class ShadowEvalLogger(CustomLogger):
             if m.get("content") is not None
         )
         judge_metadata: Final = sanitized_forwardable_call_metadata(parent_metadata, SHADOW_EVAL_JUDGE_CALL_ORIGIN)
-        judge_messages: Final = [
-            {"role": "system", "content": PAIRWISE_JUDGE_SYSTEM_PROMPT},
+        judge_messages: Final = [  # mutable-ok: SDK takes a list
+            {"role": "system", "content": PAIRWISE_JUDGE_SYSTEM_PROMPT},  # mutable-ok: SDK message
             {
                 "role": "user",
                 "content": _judge_user_prompt(conversation, response_a, response_b, _tool_definitions_text(tools)),

litellm/integrations/websearch_interception/handler.py

@@ -1683,7 +1683,7 @@ class WebSearchInterceptionLogger(CustomLogger):
         user_api_key_metadata: Final[StandardLoggingUserAPIKeyMetadata] = (
             LiteLLMProxyRequestSetup.get_sanitized_user_information_from_key(user_api_key_dict=user_api_key_auth)
         )
-        return {
+        return {  # mutable-ok: litellm's metadata channel is a plain dict its logging path reads and enriches
             **user_api_key_metadata,
             **parent_correlation.as_search_metadata(),
             "model_group": search_tool_name,

litellm/integrations/zerobus/logger.py

@@ -188,7 +188,9 @@ class ZerobusLogger(CustomBatchLogger):
 
     def _payload_for(self, kwargs: Mapping[str, object]) -> Mapping[str, object] | None:
         """The payload to buffer, redacted the way the framework redacts the success path."""
-        details: Final = self.redact_standard_logging_payload_from_model_call_details(dict(kwargs))
+        details: Final = self.redact_standard_logging_payload_from_model_call_details(
+            dict(kwargs)  # mutable-ok: both framework helpers take the call details as a dict
+        )
         payload: Final = details.get("standard_logging_object")
         if not isinstance(payload, dict):
             return None

litellm/litellm_core_utils/agentic_followup_kwargs.py

@@ -15,7 +15,7 @@ def build_agentic_followup_kwargs(
     fingerprint: str,
 ) -> Mapping[str, object]:
     """Kwargs for an agentic follow-up call: the request's kwargs overlaid by the plan's, never repeating a key already sent as a request param"""
-    seen: Final = [*fingerprints, fingerprint]
+    seen: Final = [*fingerprints, fingerprint]  # mutable-ok: the chat loop's settings reader only accepts a list
     return MappingProxyType(
         {
             key: value

litellm/litellm_core_utils/chat_completion_agentic_loop.py

@@ -125,7 +125,7 @@ def _with_agentic_loop_metadata(kwargs_for_followup: Mapping[str, object]) -> Ma
     return MappingProxyType(
         {
             **kwargs_for_followup,
-            "litellm_metadata": dict(
+            "litellm_metadata": dict(  # mutable-ok: the follow-up call's logging and proxy hooks write into litellm_metadata in place
                 chain(
                     metadata.items() if isinstance(metadata, dict) else (),
                     (

litellm/litellm_core_utils/core_helpers.py

@@ -579,7 +579,7 @@ def independent_snapshot(
     """
     sanitized: Final = {
         key: (
-            {
+            {  # mutable-ok: same request-payload shape as data
                 inner_key: ("placeholder" if inner_key == "litellm_parent_otel_span" else inner_value)
                 for inner_key, inner_value in value.items()
             }
@@ -601,13 +601,15 @@ def independent_snapshot(
             and isinstance(original_value, dict)
             and "litellm_parent_otel_span" in original_value
         ):
-            return {
+            return {  # mutable-ok: same request-payload shape as data
                 **copied_value,
                 "litellm_parent_otel_span": original_value["litellm_parent_otel_span"],
             }
         return copied_value
 
-    return {key: _copied_value(key, value) for key, value in sanitized.items()}
+    return {  # mutable-ok: same request-payload shape as data
+        key: _copied_value(key, value) for key, value in sanitized.items()
+    }
 
 
 def filter_exceptions_from_params(data: object, max_depth: int = 20) -> Any:

litellm/litellm_core_utils/get_litellm_params.py

@@ -99,7 +99,9 @@ class InvalidControlOption:
 
 
 def parse_control_options(kwargs: Mapping[str, object]) -> ControlOptions | InvalidControlOption:
-    given: Final = {name: kwargs[name] for name in _CONTROL_OPTION_NAMES if name in kwargs}
+    given: Final = {  # mutable-ok: TypeAdapter.validate_python takes a dict
+        name: kwargs[name] for name in _CONTROL_OPTION_NAMES if name in kwargs
+    }
     try:
         return _CONTROL_OPTIONS.validate_python(given)
     except ValidationError as e:
@@ -116,8 +118,8 @@ def stored_control_options(litellm_params: Mapping[str, object]) -> ControlOptio
 
 def with_control_options(litellm_params: Mapping[str, object], control: ControlOptions) -> dict[str, object]:
     if control == ControlOptions():
-        return dict(litellm_params)
-    return {**litellm_params, CONTROL_OPTIONS_KEY: control}
+        return dict(litellm_params)  # mutable-ok: completion() hands litellm_params to provider code typed as dict
+    return {**litellm_params, CONTROL_OPTIONS_KEY: control}  # mutable-ok: same dict contract as above
 
 
 def _get_base_model_from_litellm_call_metadata(

litellm/litellm_core_utils/get_model_cost_map.py

@@ -656,7 +656,7 @@ def get_model_cost_map(
     if isinstance(outcome, _FetchAttemptRetryable) and max_attempts > 1:
         threading.Thread(
             target=_retry_remote_fetch_in_background,
-            kwargs={
+            kwargs={  # mutable-ok: threading requires a mutable keyword-arguments mapping
                 "url": url,
                 "timeout": timeout,
                 "max_attempts": max_attempts,

litellm/litellm_core_utils/internal_call_metadata.py

@@ -112,16 +112,16 @@ def sanitize_user_api_key_auth(auth: object) -> object:
     """Copy of the auth object with its budget reservation removed; the cost callback
     falls back to reading the reservation from inside the auth object."""
     if isinstance(auth, dict):
-        return {k: v for k, v in auth.items() if k != "budget_reservation"}
+        return {k: v for k, v in auth.items() if k != "budget_reservation"}  # mutable-ok: SDK metadata value
     reservation: Final[object] = getattr(auth, "budget_reservation", None)
     model_copy: Final[object] = getattr(auth, "model_copy", None)
     if reservation is not None and callable(model_copy):
-        return model_copy(update={"budget_reservation": None})
+        return model_copy(update={"budget_reservation": None})  # mutable-ok: pydantic update payload
     return auth
 
 
 def _sanitized(parent_metadata: Mapping[str, object]) -> dict[str, object]:  # mutable-ok: SDK metadata kwarg
-    return {
+    return {  # mutable-ok: SDK metadata kwarg
         k: sanitize_user_api_key_auth(v) if k == _USER_API_KEY_AUTH_KEY else v
         for k, v in parent_metadata.items()
         if k not in BUDGET_RESERVATION_METADATA_KEYS
@@ -138,8 +138,10 @@ def forwarded_internal_call_metadata(
     parent's full context still describes the call being made.
     """
     if not parent_metadata:
-        return {}
-    return _sanitized(parent_metadata) | {INTERNAL_CALL_ORIGIN_METADATA_KEY: call_origin}
+        return {}  # mutable-ok: SDK metadata kwarg
+    return _sanitized(parent_metadata) | {  # mutable-ok: SDK metadata kwarg
+        INTERNAL_CALL_ORIGIN_METADATA_KEY: call_origin
+    }
 
 
 def parent_session_kwargs(request_kwargs: Mapping[str, object] | None) -> Mapping[str, str]:
@@ -165,4 +167,4 @@ def sanitized_forwardable_call_metadata(
     must not inherit per-request state such as its routing decision or logging payload.
     """
     identity: Final = {k: v for k, v in parent_metadata.items() if k in FORWARDABLE_IDENTITY_METADATA_KEYS}
-    return _sanitized(identity) | {INTERNAL_CALL_ORIGIN_METADATA_KEY: call_origin}
+    return _sanitized(identity) | {INTERNAL_CALL_ORIGIN_METADATA_KEY: call_origin}  # mutable-ok: SDK metadata kwarg

litellm/litellm_core_utils/json_fragment_accumulator.py

@@ -50,7 +50,7 @@ class JSONFragmentAccumulator:
         unconsumed: Final = self._buffer[self._offset :]
         self._buffer = unconsumed + "".join(self._chunks)
         self._offset = 0
-        self._chunks = []
+        self._chunks = []  # mutable-ok: see __init__
 
     def pop_next_value(self) -> tuple[bool, object]:
         """
@@ -88,7 +88,7 @@ class JSONFragmentAccumulator:
 
     def set(self, value: str) -> None:
         """Replace the buffer's contents with a single fragment."""
-        self._chunks = []
+        self._chunks = []  # mutable-ok: see __init__
         self._buffer = value
         self._offset = 0
         stripped: Final = value.rstrip()

litellm/litellm_core_utils/litellm_logging.py

@@ -692,7 +692,7 @@ class Logging(LiteLLMLoggingBaseClass):
         self.caching_details: CachingDetails | None = None
         # Timing for results that cannot carry ``_hidden_params`` (plain-dict /v1/messages
         # responses and the bridge stream wrappers); see ``update_response_metadata``.
-        self.response_timing_metrics: Mapping[str, float] = {}
+        self.response_timing_metrics: Mapping[str, float] = {}  # mutable-ok: kept deep-copyable
 
         # Passthrough endpoint guardrails config for field targeting
         self.passthrough_guardrails_config: dict[str, object] | None = None
@@ -716,7 +716,7 @@ class Logging(LiteLLMLoggingBaseClass):
 
     def set_response_timing_metrics(self, timing_metrics: Mapping[str, float]) -> None:
         """Keep ``_response_ms`` / ``litellm_overhead_time_ms`` for a result that has no ``_hidden_params``."""
-        self.response_timing_metrics = dict(timing_metrics)
+        self.response_timing_metrics = dict(timing_metrics)  # mutable-ok: kept deep-copyable
 
     def add_dynamic_callback(self, callback: CustomLogger) -> None:
         self.dynamic_input_callbacks = self._with_dynamic_callback(self.dynamic_input_callbacks, callback)
@@ -4139,7 +4139,7 @@ class Logging(LiteLLMLoggingBaseClass):
         if result.status == "completed":
             return InteractionsAPIResponse.model_validate(
                 result.model_dump(
-                    exclude={
+                    exclude={  # mutable-ok: pydantic types exclude as set[str], which a frozenset does not satisfy
                         "event_type",
                         "delta",
                         "index",
@@ -5242,7 +5242,9 @@ def _has_operator_exporter(config: "OpenTelemetryV2Config") -> bool:
 
 
 def _only_the_gated_exporter(config: "OpenTelemetryV2Config") -> "OpenTelemetryV2Config":
-    return config.model_copy(update={"exporters": [spec for spec in config.exporters if _is_gated(spec)]})
+    return config.model_copy(
+        update={"exporters": [spec for spec in config.exporters if _is_gated(spec)]}  # mutable-ok: model_copy update
+    )
 
 
 def _is_gated(spec: "ExporterSpec") -> bool:
@@ -5714,7 +5716,7 @@ class StandardLoggingPayloadSetup:
                 if key not in user_metadata
             }
         )
-        return {**user_metadata, **model_metadata}
+        return {**user_metadata, **model_metadata}  # mutable-ok: function contract returns a plain dict
 
     @staticmethod
     def get_standard_logging_metadata(
@@ -6562,7 +6564,9 @@ def get_standard_logging_object_payload(
         if clean_hidden_params["litellm_overhead_time_ms"] is None and status == "success":
             # /v1/messages dict results and the bridge stream wrappers keep it on the logging object;
             # failure payloads stay None like every response type that carries its own _hidden_params
-            timing_metrics: Final = getattr(logging_obj, "response_timing_metrics", None) or {}
+            timing_metrics: Final = (
+                getattr(logging_obj, "response_timing_metrics", None) or {}  # mutable-ok: empty fallback
+            )
             clean_hidden_params["litellm_overhead_time_ms"] = timing_metrics.get("litellm_overhead_time_ms")
 
         model_cost_information: Final = StandardLoggingPayloadSetup.get_model_cost_information(
@@ -6677,14 +6681,14 @@ def get_standard_logging_object_payload(
             cost_breakdown=request_cost_breakdown,
             autorouter_savings=autorouter_savings,
             autorouter_savings_estimate=(
-                {
+                {  # mutable-ok: spend-log JSON serialization requires plain mappings
                     "version": 3,
                     "status": "unknown",
                     "reason": "pending_projection",
                 }
                 if captured_baseline is not None
                 else (
-                    {
+                    {  # mutable-ok: spend-log JSON serialization requires plain mappings
                         "version": 1,
                         "status": "estimated" if autorouter_savings is not None else "unknown",
                         "reason": "uncached_usage" if autorouter_savings is not None else "baseline_unavailable",

litellm/litellm_core_utils/llm_cost_calc/guardrail_cost.py

@@ -79,7 +79,7 @@ def bedrock_guardrail_cost_by_unit(
     pricing: Final = _bedrock_guardrail_pricing(aws_region_name)
     if pricing is None:
         return None
-    return {
+    return {  # mutable-ok: stamped into guardrail_information, which safe_dumps only serializes as a plain dict
         counter: _priced_units(units, pricing.guardrail_cost_per_unit.get(counter))
         for counter, units in usage_units.items()
     }

litellm/litellm_core_utils/llm_judge.py

@@ -44,7 +44,7 @@ def parse_json_verdict(raw: str) -> dict[str, object]:  # mutable-ok: plain pars
         parsed = json.loads(text[start : end + 1])
     if not isinstance(parsed, dict):
         raise ValueError("judge response is not a JSON object")
-    return {str(k): v for k, v in parsed.items()}
+    return {str(k): v for k, v in parsed.items()}  # mutable-ok: plain parsed-JSON payload
 
 
 def extract_text_from_content(content: object) -> str:

litellm/litellm_core_utils/llm_request_utils.py

@@ -16,7 +16,9 @@ def _form_field_value(value: object) -> str:
 def _flatten_form_field(key: str, value: object) -> tuple[tuple[str, str], ...]:
     pending_fields: Final[  # mutable-ok: depth-capped stack walks nested JSON into multipart names
         list[tuple[str, object, int]]
-    ] = [(key, value, 0)]
+    ] = [  # mutable-ok: depth-capped stack walks nested JSON into multipart names
+        (key, value, 0)
+    ]
     flat_fields: Final[list[tuple[str, str]]] = []  # mutable-ok: local accumulator
     while pending_fields:
         current_key, current_value, depth = pending_fields.pop()
@@ -46,7 +48,9 @@ def _is_form_scalar(value: object) -> bool:
 def _flatten_form_data_field(key: str, value: object) -> tuple[tuple[str, str | tuple[str, ...]], ...]:
     pending_fields: Final[  # mutable-ok: depth-capped stack walks nested JSON into multipart names
         list[tuple[str, object, int]]
-    ] = [(key, value, 0)]
+    ] = [  # mutable-ok: depth-capped stack walks nested JSON into multipart names
+        (key, value, 0)
+    ]
     flat_fields: Final[list[tuple[str, str | tuple[str, ...]]]] = []  # mutable-ok: local accumulator
     while pending_fields:
         current_key, current_value, depth = pending_fields.pop()

litellm/litellm_core_utils/llm_response_utils/response_metadata.py

@@ -71,7 +71,7 @@ def response_timing_metrics(
     receive_anchored: Final = timing_window[1]
     total_response_time_ms: Final = (end_time.timestamp() - window_start.timestamp()) * 1000
     if not include_overhead:
-        return {"_response_ms": total_response_time_ms}
+        return {"_response_ms": total_response_time_ms}  # mutable-ok: read-only timing result
     caching_details: Final = logging_obj.caching_details
     cache_duration_ms: Final = (
         caching_details.get("cache_duration_ms")

litellm/litellm_core_utils/logging_utils.py

@@ -312,7 +312,7 @@ def _set_duration_in_model_call_details(
 def speech_request_body(model: str, voice: str, optional_params: Mapping[str, object]) -> Mapping[str, object]:
     """Speech request body for telemetry, without the caller headers the provider SDKs
     take as request kwargs rather than body fields."""
-    return {
+    return {  # mutable-ok: loggers isinstance-check the request body as a dict
         "model": model,
         "voice": voice,
         **{key: value for key, value in optional_params.items() if key != "extra_headers"},

litellm/litellm_core_utils/prompt_templates/common_utils.py

@@ -1274,7 +1274,7 @@ def _flatten_schema_against_root(
     if not is_object_schema:
         return schema
 
-    merged_properties: Final = {
+    merged_properties: Final = {  # mutable-ok: tool parameters are JSON dicts
         name: value for source in (*reversed(branches), schema) for name, value in _schema_properties(source).items()
     }
     required_names: Final = _schema_required_names(schema).union(
@@ -1282,7 +1282,7 @@ def _flatten_schema_against_root(
     )
     kept: Final = MappingProxyType({key: value for key, value in schema.items() if key not in dropped})
     required_update: Final = MappingProxyType({"required": sorted(required_names)}) if required_names else _EMPTY_SCHEMA
-    return {
+    return {  # mutable-ok: tool parameters are JSON dicts
         **kept,
         "type": "object",
         "properties": merged_properties,
@@ -1309,7 +1309,7 @@ def flatten_top_level_schema_combinators(schema: Mapping[str, object]) -> Mappin
     OpenAI's own validation still applies. Non-object schemas pass through
     unchanged and the input is never mutated.
     """
-    return _flatten_schema_against_root(schema, schema, frozenset(), 0, {})
+    return _flatten_schema_against_root(schema, schema, frozenset(), 0, {})  # mutable-ok: fresh per-call $ref memo
 
 
 _SUBSCHEMA_KEYWORDS: Final = frozenset(
@@ -1384,7 +1384,7 @@ def _subschemas(node: Mapping[str, object]) -> Iterator[Mapping[str, object]]:
 def _node_without_non_python_regex(
     node: Mapping[str, object], rebuilt: Mapping[int, Mapping[str, object]]
 ) -> Mapping[str, object]:
-    kept: Final = {
+    kept: Final = {  # mutable-ok: tool parameters are JSON dicts
         key: _keyword_value_rebuilt(key, value, rebuilt)
         for key, value in node.items()
         if key != "pattern" or not isinstance(value, str) or _is_python_regex(value)
@@ -1394,14 +1394,14 @@ def _node_without_non_python_regex(
 
 def _keyword_value_rebuilt(key: str, value: object, rebuilt: Mapping[int, Mapping[str, object]]) -> object:
     if key in _SUBSCHEMA_MAP_KEYWORDS and isinstance(value, dict):
-        kept: Final = {
+        kept: Final = {  # mutable-ok: tool parameters are JSON dicts
             name: rebuilt.get(id(sub), sub)
             for name, sub in value.items()
             if key != "patternProperties" or not isinstance(name, str) or _is_python_regex(name)
         }
         return value if len(kept) == len(value) and all(kept[name] is value[name] for name in kept) else kept
     if key in _SUBSCHEMA_LIST_KEYWORDS and isinstance(value, list):
-        items: Final = [rebuilt.get(id(sub), sub) for sub in value]
+        items: Final = [rebuilt.get(id(sub), sub) for sub in value]  # mutable-ok: tool parameters are JSON lists
         return value if all(new is old for new, old in zip(items, value, strict=True)) else items
     if key in _SUBSCHEMA_KEYWORDS and isinstance(value, dict):
         return rebuilt.get(id(value), value)
@@ -1433,7 +1433,7 @@ def tool_with_sanitized_parameters(
     sanitized: Final = sanitize(parameters)
     if sanitized is parameters:
         return tool
-    return {**tool, "function": {**function, "parameters": sanitized}}
+    return {**tool, "function": {**function, "parameters": sanitized}}  # mutable-ok: request tools are JSON dicts
 
 
 def _get_image_mime_type_from_url(url: str) -> str | None:
@@ -1689,7 +1689,7 @@ _MarkedT: Final = TypeVar("_MarkedT", bound=Mapping[str, object])
 def with_prompt_cache_breakpoint(target: _MarkedT, marker: object) -> _MarkedT:
     if marker is None:
         return target
-    marked: Final = {**target, "prompt_cache_breakpoint": marker}
+    marked: Final = {**target, "prompt_cache_breakpoint": marker}  # mutable-ok: API message payload
     return cast(_MarkedT, marked)  # cast-ok: same block shape as the input plus the marker key
 
 
@@ -1703,7 +1703,9 @@ def strip_litellm_internal_message_fields(message: AllMessageValues) -> AllMessa
         return message
     return cast(  # cast-ok: same TypedDict minus internal keys
         AllMessageValues,
-        {key: value for key, value in message.items() if key not in LITELLM_INTERNAL_MESSAGE_FIELDS},
+        {  # mutable-ok: provider transforms mutate message dicts in place downstream
+            key: value for key, value in message.items() if key not in LITELLM_INTERNAL_MESSAGE_FIELDS
+        },
     )
 
 
@@ -2192,9 +2194,11 @@ def _split_images_from_tool_message(
     )
     if not image_parts:
         return message, ()
-    remaining_parts = [part for part in content if not _is_image_url_part(part)]
+    remaining_parts = [  # mutable-ok: tool message content must stay a json list
+        part for part in content if not _is_image_url_part(part)
+    ]
     new_content = remaining_parts if remaining_parts else TOOL_RESULT_IMAGE_PLACEHOLDER
-    rewritten = {**message, "content": new_content}
+    rewritten = {**message, "content": new_content}  # mutable-ok: chat messages are plain json dicts
     return cast(AllMessageValues, rewritten), image_parts  # cast-ok: dict spread keeps keys like cache_control
 
 
@@ -2202,12 +2206,14 @@ def _hoist_images_in_tool_message_run(
     run: Iterable[AllMessageValues],
 ) -> list[AllMessageValues]:  # mutable-ok: message pipelines type messages as mutable lists
     split_results = tuple(_split_images_from_tool_message(message) for message in run)
-    hoisted_images = [image for _, images in split_results for image in images]
-    rewritten_messages = [message for message, _ in split_results]
+    hoisted_images = [  # mutable-ok: user message content must be a json list
+        image for _, images in split_results for image in images
+    ]
+    rewritten_messages = [message for message, _ in split_results]  # mutable-ok: pipelines mutate message lists
     if not hoisted_images:
         return rewritten_messages
     boundary_part = ChatCompletionTextObject(type="text", text=TOOL_RESULT_IMAGE_BOUNDARY)
-    hoisted_content = [boundary_part, *hoisted_images]
+    hoisted_content = [boundary_part, *hoisted_images]  # mutable-ok: user message content must be a json list
     rewritten_messages.append(ChatCompletionUserMessage(role="user", content=hoisted_content))
     return rewritten_messages
 
@@ -2231,7 +2237,7 @@ def hoist_images_from_tool_messages(
     """
     if not any(_tool_message_carries_image(message) for message in messages):
         return messages
-    return [
+    return [  # mutable-ok: pipelines mutate message lists
         rewritten_message
         for is_tool_run, run in groupby(messages, key=lambda message: message.get("role") == "tool")
         for rewritten_message in (_hoist_images_in_tool_message_run(run) if is_tool_run else run)
@@ -2253,9 +2259,11 @@ def _drop_tool_reference_parts(message: AllMessageValues) -> AllMessageValues:
     if not _tool_message_carries_tool_reference(message):
         return message
     content = cast(list, message.get("content"))  # cast-ok: shape checked by _tool_message_carries_tool_reference
-    remaining_parts = [part for part in content if not _is_tool_reference_part(part)]
+    remaining_parts = [  # mutable-ok: tool message content must stay a json list
+        part for part in content if not _is_tool_reference_part(part)
+    ]
     new_content = remaining_parts if remaining_parts else ""
-    rewritten = {**message, "content": new_content}
+    rewritten = {**message, "content": new_content}  # mutable-ok: chat messages are plain json dicts
     return cast(AllMessageValues, rewritten)  # cast-ok: dict spread keeps keys like cache_control
 
 
@@ -2273,7 +2281,7 @@ def drop_tool_reference_parts_from_tool_messages(
     """
     if not any(_tool_message_carries_tool_reference(message) for message in messages):
         return messages
-    return [_drop_tool_reference_parts(message) for message in messages]
+    return [_drop_tool_reference_parts(message) for message in messages]  # mutable-ok: pipelines mutate message lists
 
 
 INSTRUCTION_MESSAGE_ROLES: Final = frozenset({"system", "developer"})
@@ -2286,7 +2294,7 @@ def _is_instruction_message(message: AllMessageValues) -> bool:
 def system_messages_first(
     messages: list[AllMessageValues],  # mutable-ok: message pipelines type messages as mutable lists
 ) -> list[AllMessageValues]:  # mutable-ok: message pipelines type messages as mutable lists
-    return [
+    return [  # mutable-ok: pipelines mutate message lists
         *(message for message in messages if _is_instruction_message(message)),
         *(message for message in messages if not _is_instruction_message(message)),
     ]
@@ -2307,14 +2315,16 @@ def _merge_system_message_run(run: Sequence[AllMessageValues]) -> AllMessageValu
     if all(isinstance(content, str) for content in contents):
         joined_text: Final = "\n\n".join(cast(tuple[str, ...], contents))  # cast-ok: every content is a str
         return cast(AllMessageValues, {**run[0], "content": joined_text})  # cast-ok: dict spread keeps message shape
-    merged_parts: Final = [part for content in contents for part in _system_content_as_text_parts(content)]
+    merged_parts: Final = [  # mutable-ok: chat message content must stay a json list
+        part for content in contents for part in _system_content_as_text_parts(content)
+    ]
     return cast(AllMessageValues, {**run[0], "content": merged_parts})  # cast-ok: dict spread keeps message shape
 
 
 def merge_consecutive_system_messages(
     messages: list[AllMessageValues],  # mutable-ok: message pipelines type messages as mutable lists
 ) -> list[AllMessageValues]:  # mutable-ok: message pipelines type messages as mutable lists
-    return [
+    return [  # mutable-ok: pipelines mutate message lists
         merged
         for is_system_run, run in groupby(messages, key=lambda message: message.get("role") == "system")
         for merged in ((_merge_system_message_run(tuple(run)),) if is_system_run else run)

litellm/litellm_core_utils/prompt_templates/factory.py

@@ -2376,7 +2376,7 @@ def anthropic_messages_pt(
     # add role=tool support to allow function call result/error submission
     user_message_types: Final = {"user", "tool", "function"}
     # reformat messages to ensure user/assistant are alternating, if there's either 2 consecutive 'user' messages or 2 consecutive 'assistant' message, merge them.
-    new_messages: Final[_AnthropicMessageList] = []
+    new_messages: Final[_AnthropicMessageList] = []  # mutable-ok: accumulator behind the mutable return contract
 
     if len(messages) == 0:
         if not litellm.modify_params:

litellm/litellm_core_utils/prompt_templates/image_handling.py

@@ -243,28 +243,28 @@ def _inferred_format(file: Mapping[str, object], url: str) -> Mapping[str, str]:
 
 
 def _inlined_image_url(image_url: Mapping[str, object] | None, data_url: str) -> Mapping[str, object] | str:
-    return {**image_url, "url": data_url} if image_url is not None else data_url
+    return {**image_url, "url": data_url} if image_url is not None else data_url  # mutable-ok: json-serialized part
 
 
 def _inlined_file(file: Mapping[str, object], url: str, data_url: str) -> Mapping[str, object]:
-    kept: Final = {k: v for k, v in file.items() if k != "file_id"}
-    return {**kept, **_inferred_format(file, url), "file_data": data_url}
+    kept: Final = {k: v for k, v in file.items() if k != "file_id"}  # mutable-ok: json-serialized message part
+    return {**kept, **_inferred_format(file, url), "file_data": data_url}  # mutable-ok: json-serialized part
 
 
 def _base64_source(url: str, data_url: str) -> Mapping[str, str]:
     fetched_media_type, data = data_url.removeprefix("data:").split(";base64,", 1)
     media_type: Final = "application/pdf" if url.lower().endswith(".pdf") else fetched_media_type
-    return {"type": "base64", "media_type": media_type, "data": data}
+    return {"type": "base64", "media_type": media_type, "data": data}  # mutable-ok: json-serialized message part
 
 
 def _inline(remote: _RemoteImage | _RemoteFile | _RemoteSource, data_url: str) -> Mapping[str, object]:
     match remote:
         case _RemoteImage(part, image_url, _):
-            return {**part, "image_url": _inlined_image_url(image_url, data_url)}
+            return {**part, "image_url": _inlined_image_url(image_url, data_url)}  # mutable-ok: json-serialized part
         case _RemoteFile(part, file, url):
-            return {**part, "file": _inlined_file(file, url, data_url)}
+            return {**part, "file": _inlined_file(file, url, data_url)}  # mutable-ok: json-serialized message part
         case _RemoteSource(part, _, url):
-            return {**part, "source": _base64_source(url, data_url)}
+            return {**part, "source": _base64_source(url, data_url)}  # mutable-ok: json-serialized message part
 
 
 def _content_parts(message: Mapping[str, object]) -> tuple[object, ...]:
@@ -286,8 +286,10 @@ def _inline_message(
     parts: Final = _content_parts(message)
     if not parts:
         return message
-    inlined_parts: Final = [_inline_part(part, data_urls, should_inline) for part in parts]
-    inlined_message: Final = {**message, "content": inlined_parts}
+    inlined_parts: Final = [  # mutable-ok: content must stay a list for the transforms' isinstance checks
+        _inline_part(part, data_urls, should_inline) for part in parts
+    ]
+    inlined_message: Final = {**message, "content": inlined_parts}  # mutable-ok: json-serialized message
     return inlined_message  # pyright: ignore[reportReturnType]  # the same message with its remote parts inlined
 
 
@@ -324,4 +326,6 @@ async def async_inline_remote_media(
         return messages
     data_urls: Final = await _fetch_data_urls(remote_urls)
     inlined: Final = MappingProxyType(dict(zip(remote_urls, data_urls, strict=True)))
-    return [_inline_message(message, inlined, should_inline) for message in messages]
+    return [  # mutable-ok: transform_request takes a list
+        _inline_message(message, inlined, should_inline) for message in messages
+    ]

litellm/litellm_core_utils/prompt_templates/mid_conversation_system.py

@@ -167,7 +167,7 @@ def anthropic_system_messages(message: object) -> tuple[AnthropicMessagesSystemM
         return ()
     wire: Final[AnthropicMessagesSystemMessageParam] = {
         "role": "system",
-        "content": list(blocks),
+        "content": list(blocks),  # mutable-ok: wire payload; cache_control hooks edit content blocks in place
     }
     return (wire,)
 

litellm/litellm_core_utils/provider_affinity.py

@@ -88,11 +88,11 @@ def add_provider_affinity_header(
 ) -> dict[str, object]:  # mutable-ok: downstream handlers add auth and signing headers
     header_name: Final = _get_provider_affinity_header_name(litellm_params)
     if header_name is None or any(key.lower() == header_name.lower() for key in headers):
-        return dict(headers)
+        return dict(headers)  # mutable-ok: downstream handlers add auth and signing headers
 
     session_id: Final = get_stable_session_id(litellm_params)
     if session_id is None:
-        return dict(headers)
+        return dict(headers)  # mutable-ok: downstream handlers add auth and signing headers
     if any(character in session_id for character in ("\r", "\n", "\0")):
         raise ValueError("session_id cannot contain HTTP header control characters")
-    return {**headers, header_name: session_id}
+    return {**headers, header_name: session_id}  # mutable-ok: downstream handlers add auth and signing headers

litellm/litellm_core_utils/sentry_scrubbing.py

@@ -109,12 +109,12 @@ def scrub_json_strings(value: JsonValue, scrub: Callable[[str], str], path: Json
         return scrub(value)
     if isinstance(value, dict):
         unscrubbed_keys: Final = SOURCE_CONTEXT_KEYS if path in STACK_FRAME_PATHS else frozenset[str]()
-        return {
+        return {  # mutable-ok: JSON object
             key: item if key in unscrubbed_keys else scrub_json_strings(item, scrub, (*path, key))
             for key, item in value.items()
         }
     if isinstance(value, list):
-        return [scrub_json_strings(item, scrub, (*path, "*")) for item in value]
+        return [scrub_json_strings(item, scrub, (*path, "*")) for item in value]  # mutable-ok: JSON array
     return value
 
 
@@ -141,8 +141,8 @@ def build_sentry_init_options(env: Mapping[str, str]) -> SentryInitOptions:
         sample_rate=float(env.get("SENTRY_API_SAMPLE_RATE") or "1.0"),
         send_default_pii=send_default_pii,
         event_scrubber=EventScrubber(
-            denylist=list(SECRET_FIELD_NAMES),
-            pii_denylist=list(PII_FIELD_NAMES),
+            denylist=list(SECRET_FIELD_NAMES),  # mutable-ok: EventScrubber appends pii_denylist onto denylist in place
+            pii_denylist=list(PII_FIELD_NAMES),  # mutable-ok: EventScrubber takes List[str]
             recursive=True,
             send_default_pii=send_default_pii,
         ),

litellm/litellm_core_utils/streaming_chunk_builder_utils.py

@@ -490,7 +490,9 @@ class ChunkProcessor:
     def get_combined_tool_content(
         self, tool_call_chunks: Sequence["_ToolCallChunk"]
     ) -> list[ChatCompletionMessageToolCall | ChatCompletionMessageCustomToolCall]:
-        tool_calls_list: list[ChatCompletionMessageToolCall | ChatCompletionMessageCustomToolCall] = []
+        tool_calls_list: list[
+            ChatCompletionMessageToolCall | ChatCompletionMessageCustomToolCall
+        ] = []  # mutable-ok: see return type
         tool_call_map: Final[dict[_ToolCallKey, dict[str, Any]]] = {}
 
         for chunk in tool_call_chunks:

litellm/litellm_core_utils/streaming_handler.py

@@ -189,7 +189,9 @@ def _provider_hidden_params(
     hidden: Final[object] = getattr(chunk, "_hidden_params", None)
     parsed: Final = _parsed_provider_hidden_params(hidden)
     provider_specific_fields: Final[object | None] = (
-        dict(parsed.provider_specific_fields) if parsed is not None and parsed.provider_specific_fields else None
+        dict(parsed.provider_specific_fields)  # mutable-ok: stream assembly merges provider metadata into this dict
+        if parsed is not None and parsed.provider_specific_fields
+        else None
     )
     params: Final[Mapping[str, object]] = MappingProxyType(
         {

litellm/litellm_core_utils/tokenizer.py

@@ -72,7 +72,7 @@ class OpenAIEncoding:
         return self._special_tokens["<|endoftext|>"]
 
     @property
-    def special_tokens_set(self) -> set[str]:  # mutable-ok: [LIT001] SDK return type
+    def special_tokens_set(self) -> set[str]:  # mutable-ok: [LIT001, LIT002] SDK return type
         return set(self._special_tokens)
 
     def is_special_token(self, token: int) -> bool:
@@ -80,7 +80,7 @@ class OpenAIEncoding:
 
     # ---- encoding -------------------------------------------------------------------------
 
-    def encode_ordinary(self, text: str) -> list[int]:  # mutable-ok: [LIT001] SDK return type
+    def encode_ordinary(self, text: str) -> list[int]:  # mutable-ok: [LIT001, LIT002] SDK return type
         return self._native.encode(text)
 
     def encode(
@@ -89,7 +89,7 @@ class OpenAIEncoding:
         *,
         allowed_special: AllowedSpecial = frozenset(),
         disallowed_special: SpecialTokens = "all",
-    ) -> list[int]:  # mutable-ok: [LIT001] SDK return type
+    ) -> list[int]:  # mutable-ok: [LIT001, LIT002] SDK return type
         allowed: Final = self._allowed(text, allowed_special, disallowed_special)
         if not allowed:
             return self.encode_ordinary(text)
@@ -111,9 +111,11 @@ class OpenAIEncoding:
 
     def encode_ordinary_batch(
         self, text: Sequence[str], *, num_threads: int = 8
-    ) -> list[list[int]]:  # mutable-ok: [LIT001] SDK return type
+    ) -> list[list[int]]:  # mutable-ok: [LIT001, LIT002] SDK return type
         with ThreadPoolExecutor(num_threads) as executor:
-            return list(executor.map(self.encode_ordinary, text))
+            return list(  # mutable-ok: [LIT002] SDK returns a list
+                executor.map(self.encode_ordinary, text)
+            )
 
     def encode_batch(
         self,
@@ -122,10 +124,12 @@ class OpenAIEncoding:
         num_threads: int = 8,
         allowed_special: AllowedSpecial = frozenset(),
         disallowed_special: SpecialTokens = "all",
-    ) -> list[list[int]]:  # mutable-ok: [LIT001] SDK return type
+    ) -> list[list[int]]:  # mutable-ok: [LIT001, LIT002] SDK return type
         encode: Final = partial(self.encode, allowed_special=allowed_special, disallowed_special=disallowed_special)
         with ThreadPoolExecutor(num_threads) as executor:
-            return list(executor.map(encode, text))
+            return list(  # mutable-ok: [LIT002] SDK returns a list
+                executor.map(encode, text)
+            )
 
     def encode_with_unstable(
         self,
@@ -133,7 +137,7 @@ class OpenAIEncoding:
         *,
         allowed_special: AllowedSpecial = frozenset(),
         disallowed_special: SpecialTokens = "all",
-    ) -> tuple[list[int], list[list[int]]]:  # mutable-ok: [LIT001] SDK return type
+    ) -> tuple[list[int], list[list[int]]]:  # mutable-ok: [LIT001, LIT002] SDK return type
         """The stable tokens of `text` and every completion its unstable tail could become.
 
         Completions come back sorted; tiktoken returns them in hash order."""
@@ -160,12 +164,14 @@ class OpenAIEncoding:
     def decode_single_token_bytes(self, token: int) -> bytes:
         return self.decode_bytes((token,))
 
-    def decode_tokens_bytes(self, tokens: Sequence[int]) -> list[bytes]:  # mutable-ok: [LIT001] SDK return type
-        return [self.decode_single_token_bytes(token) for token in tokens]
+    def decode_tokens_bytes(self, tokens: Sequence[int]) -> list[bytes]:  # mutable-ok: [LIT001, LIT002] SDK return type
+        return [  # mutable-ok: [LIT002] SDK returns a list
+            self.decode_single_token_bytes(token) for token in tokens
+        ]
 
     def decode_with_offsets(
         self, tokens: Sequence[int]
-    ) -> tuple[str, list[int]]:  # mutable-ok: [LIT001] SDK return type
+    ) -> tuple[str, list[int]]:  # mutable-ok: [LIT001, LIT002] SDK return type
         """The decoded text and, per token, the index of the first character holding its bytes.
 
         Like tiktoken, raises `UnicodeDecodeError` when the tokens do not decode to valid UTF-8."""
@@ -179,17 +185,21 @@ class OpenAIEncoding:
 
     def decode_batch(
         self, batch: Sequence[Sequence[int]], *, errors: str = "replace", num_threads: int = 8
-    ) -> list[str]:  # mutable-ok: [LIT001] SDK return type
+    ) -> list[str]:  # mutable-ok: [LIT001, LIT002] SDK return type
         with ThreadPoolExecutor(num_threads) as executor:
-            return list(executor.map(partial(self.decode, errors=errors), batch))
+            return list(  # mutable-ok: [LIT002] SDK returns a list
+                executor.map(partial(self.decode, errors=errors), batch)
+            )
 
     def decode_bytes_batch(
         self, batch: Sequence[Sequence[int]], *, num_threads: int = 8
-    ) -> list[bytes]:  # mutable-ok: [LIT001] SDK return type
+    ) -> list[bytes]:  # mutable-ok: [LIT001, LIT002] SDK return type
         with ThreadPoolExecutor(num_threads) as executor:
-            return list(executor.map(self.decode_bytes, batch))
+            return list(  # mutable-ok: [LIT002] SDK returns a list
+                executor.map(self.decode_bytes, batch)
+            )
 
-    def token_byte_values(self) -> list[bytes]:  # mutable-ok: [LIT001] SDK return type
+    def token_byte_values(self) -> list[bytes]:  # mutable-ok: [LIT001, LIT002] SDK return type
         return self._native.token_byte_values()
 
     def __reduce__(self) -> tuple[Callable[[str], OpenAIEncoding], tuple[str]]:
@@ -263,14 +273,16 @@ class HuggingFaceTokenizer:
     def id_to_token(self, id: int) -> str | None:
         return self._native.id_to_token(id)
 
-    def get_vocab(self, with_added_tokens: bool = True) -> dict[str, int]:  # mutable-ok: [LIT001] SDK return type
+    def get_vocab(
+        self, with_added_tokens: bool = True
+    ) -> dict[str, int]:  # mutable-ok: [LIT001, LIT002] SDK return type
         return self._native.get_vocab(with_added_tokens)
 
     def get_vocab_size(self, with_added_tokens: bool = True) -> int:
         return self._native.get_vocab_size(with_added_tokens)
 
-    def get_added_tokens_decoder(self) -> dict[int, AddedToken]:  # mutable-ok: [LIT001] SDK return type
-        return {
+    def get_added_tokens_decoder(self) -> dict[int, AddedToken]:  # mutable-ok: [LIT001, LIT002] SDK return type
+        return {  # mutable-ok: [LIT002] SDK returns a dict
             token_id: AddedToken(
                 content, single_word=single_word, lstrip=lstrip, rstrip=rstrip, normalized=normalized, special=special
             )
@@ -288,11 +300,11 @@ class HuggingFaceTokenizer:
         return self._native.num_special_tokens_to_add(is_pair)
 
     @property
-    def padding(self) -> dict[str, object] | None:  # mutable-ok: [LIT001] SDK return type
+    def padding(self) -> dict[str, object] | None:  # mutable-ok: [LIT001, LIT002] SDK return type
         return self._native.padding()
 
     @property
-    def truncation(self) -> dict[str, object] | None:  # mutable-ok: [LIT001] SDK return type
+    def truncation(self) -> dict[str, object] | None:  # mutable-ok: [LIT001, LIT002] SDK return type
         return self._native.truncation()
 
     @property
@@ -315,7 +327,7 @@ class HuggingFaceTokenizer:
         input: Sequence[HuggingFaceBatchInput],
         is_pretokenized: bool = False,
         add_special_tokens: bool = True,
-    ) -> list[HuggingFaceEncoding]:  # mutable-ok: [LIT001] SDK return type
+    ) -> list[HuggingFaceEncoding]:  # mutable-ok: [LIT001, LIT002] SDK return type
         return self._encode_batch(input, is_pretokenized, add_special_tokens, fast=False)
 
     def encode_batch_fast(
@@ -323,12 +335,12 @@ class HuggingFaceTokenizer:
         input: Sequence[HuggingFaceBatchInput],
         is_pretokenized: bool = False,
         add_special_tokens: bool = True,
-    ) -> list[HuggingFaceEncoding]:  # mutable-ok: [LIT001] SDK return type
+    ) -> list[HuggingFaceEncoding]:  # mutable-ok: [LIT001, LIT002] SDK return type
         return self._encode_batch(input, is_pretokenized, add_special_tokens, fast=True)
 
     def _encode_batch(
         self, input: Sequence[HuggingFaceBatchInput], is_pretokenized: bool, add_special_tokens: bool, fast: bool
-    ) -> list[HuggingFaceEncoding]:  # mutable-ok: [LIT001] SDK return type
+    ) -> list[HuggingFaceEncoding]:  # mutable-ok: [LIT001, LIT002] SDK return type
         sequences: Final = tuple(_batch_input(item, is_pretokenized) for item in input)
         return self._native.encode_batch_huggingface(sequences, is_pretokenized, add_special_tokens, fast)
 
@@ -341,8 +353,10 @@ class HuggingFaceTokenizer:
 
     def decode_batch(
         self, sequences: Sequence[Sequence[int]], skip_special_tokens: bool = True
-    ) -> list[str]:  # mutable-ok: [LIT001] SDK return type
-        return [self.decode(ids, skip_special_tokens=skip_special_tokens) for ids in sequences]
+    ) -> list[str]:  # mutable-ok: [LIT001, LIT002] SDK return type
+        return [  # mutable-ok: [LIT002] SDK returns a list
+            self.decode(ids, skip_special_tokens=skip_special_tokens) for ids in sequences
+        ]
 
     def __reduce__(self) -> tuple[Callable[[str], HuggingFaceTokenizer], tuple[str]]:
         return (HuggingFaceTokenizer.from_str, (self.to_str(),))

litellm/llms/a2a/chat/transformation.py

@@ -53,7 +53,7 @@ def _registry_headers(agent_litellm_params: Mapping[str, object]) -> dict[str, o
     if not isinstance(stored_headers, Mapping):
         return None
     entra_owns_authorization: Final = _agent_authenticates_with_entra(agent_litellm_params)
-    return {
+    return {  # mutable-ok: completion() and httpx take the request headers as a dict
         name: value
         for name, value in stored_headers.items()
         if not (entra_owns_authorization and str(name).lower() == "authorization")

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

@@ -263,7 +263,7 @@ def _rewritten_event(event: Mapping[str, object], rewrite_event: _SSEEventRewrit
     section: Final = None if rewrite is None else event.get(rewrite.section)
     if rewrite is None or not isinstance(section, Mapping):
         return event
-    return {**event, rewrite.section: {**section, rewrite.field: rewrite.value}}
+    return {**event, rewrite.section: {**section, rewrite.field: rewrite.value}}  # mutable-ok: json.dumps needs a dict
 
 
 def _tool_call_shapes(tool_calls: Sequence[object]) -> tuple[_ToolCallShape, ...]:
@@ -539,7 +539,9 @@ class AnthropicMessagesHandler(BaseTranslation):
 
         # The top-level prompt is translated on its own below so it can be hoisted in front of
         # any mid-turn system entries and scanned first, aligned with that structured position.
-        translation_source: Final = {key: value for key, value in data.items() if key != "system"}
+        translation_source: Final = {  # mutable-ok: API message payload
+            key: value for key, value in data.items() if key != "system"
+        }
         chat_completion_compatible_request: Final = self._translate_to_openai(translation_source)
 
         full_structured_messages: Final = cast(
@@ -592,7 +594,7 @@ class AnthropicMessagesHandler(BaseTranslation):
             *top_level_system_scanned,
             *(item for one_message in extracted for item in one_message.scanned),
         )
-        texts_to_check: Final = [item.text for item in 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]
         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]
@@ -689,13 +691,13 @@ class AnthropicMessagesHandler(BaseTranslation):
         if not system:
             return None
         probe: Final = self._translate_to_openai(
-            {
+            {  # mutable-ok: API message payload
                 "model": data.get("model") or "",
-                "messages": [],
+                "messages": [],  # mutable-ok: API message payload
                 "system": system,
             }
         )
-        hoisted: Final = probe.get("messages") or []
+        hoisted: Final = probe.get("messages") or []  # mutable-ok: API message payload
         return hoisted[0] if hoisted else None
 
     @staticmethod
@@ -718,7 +720,9 @@ class AnthropicMessagesHandler(BaseTranslation):
         """Convert an OpenAI system message to the client's Anthropic-shaped entry."""
         content: Final = message.get("content")
         if isinstance(content, str):
-            return {"role": "system", "content": content} if content else None
+            return (
+                {"role": "system", "content": content} if content else None  # mutable-ok: API message payload
+            )
         if not isinstance(content, list):
             return None
         blocks: Final[list[dict[str, object]]] = []  # mutable-ok: API message payload
@@ -736,7 +740,9 @@ class AnthropicMessagesHandler(BaseTranslation):
             if cache_control:
                 anthropic_block["cache_control"] = deepcopy(cache_control)
             blocks.append(anthropic_block)
-        return {"role": "system", "content": blocks} if blocks else None
+        return (
+            {"role": "system", "content": blocks} if blocks else None  # mutable-ok: API message payload
+        )
 
     @staticmethod
     def _fold_leading_systems_into_top_level(
@@ -840,7 +846,7 @@ class AnthropicMessagesHandler(BaseTranslation):
             for group in group_tool_exchanges(run):
                 converted.extend(
                     anthropic_messages_pt(
-                        messages=[run[index] for index in group],
+                        messages=[run[index] for index in group],  # mutable-ok: API message payload
                         model=model,
                         llm_provider="anthropic",
                     )

litellm/llms/anthropic/chat/handler.py

@@ -763,7 +763,7 @@ class ModelResponseIterator:
         return content_block_start
 
     def _web_search_call_snapshot(self) -> dict[str, object]:
-        return dict(self._web_search_calls)
+        return dict(self._web_search_calls)  # mutable-ok: stream payload snapshot
 
     def _complete_web_search_call(self, result: dict[str, object]) -> None:
         tool_use_id: Final = result.get("tool_use_id")
@@ -771,7 +771,7 @@ class ModelResponseIterator:
             return
         self._web_search_calls[tool_use_id] = build_web_search_call(
             tool_id=tool_use_id,
-            tool_input=self._server_tool_inputs.get(tool_use_id, {}),
+            tool_input=self._server_tool_inputs.get(tool_use_id, {}),  # mutable-ok: empty provider input
             result=result,
         )
 
@@ -880,7 +880,7 @@ class ModelResponseIterator:
                             self._web_search_calls[self._current_server_tool_id] = build_web_search_call(
                                 self._current_server_tool_id,
                                 tool_input,
-                                {"content": []},
+                                {"content": []},  # mutable-ok: no provider result yet
                                 status="in_progress",
                             )
                             provider_specific_fields["web_search_calls"] = self._web_search_call_snapshot()

litellm/llms/anthropic/chat/transformation.py

@@ -1978,7 +1978,9 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
         # system message stays in the conversation: hoisting it rewrites the cached
         # prefix and re-bills the whole history at cache-write pricing (#36559).
         leading_system_run, later_messages = split_leading_system_run(messages)
-        anthropic_system_message_list: Final = self.translate_system_message(messages=list(leading_system_run))
+        anthropic_system_message_list: Final = self.translate_system_message(
+            messages=list(leading_system_run)  # mutable-ok: translate_system_message pops from the list it is given
+        )
         # Handling anthropic API Prompt Caching
         if len(anthropic_system_message_list) > 0:
             optional_params["system"] = anthropic_system_message_list
@@ -1992,7 +1994,7 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
         try:
             anthropic_messages = anthropic_messages_pt(
                 model=model,
-                messages=list(conversation),
+                messages=list(conversation),  # mutable-ok: anthropic_messages_pt rewrites entries in place
                 llm_provider=self._resolved_provider,
             )
         except Exception as e:
@@ -2106,7 +2108,7 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
                 optional_params.pop("output_config", None)
                 data.pop("output_config", None)
                 return
-            format_only: Final = {"format": preserved_format}
+            format_only: Final = {"format": preserved_format}  # mutable-ok: json body
             optional_params["output_config"] = format_only  # rebind-ok: out-param store
             data["output_config"] = format_only  # rebind-ok: out-param store
             return
@@ -2513,7 +2515,7 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
     ) -> list[object]:
         content: Final = completion_response.get("content")
         blocks: Final = content if isinstance(content, Sequence) else ()
-        inputs: Final = {
+        inputs: Final = {  # mutable-ok: indexes provider server inputs
             call_id: tool_input
             for block in blocks
             if isinstance(block, Mapping)
@@ -2522,10 +2524,10 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
             and isinstance((call_id := block.get("id")), str)
             and isinstance((tool_input := block.get("input")), Mapping)
         }
-        return [
+        return [  # mutable-ok: provider-neutral response items
             build_web_search_call(
                 tool_id=tool_use_id,
-                tool_input=inputs.get(tool_use_id, {}),
+                tool_input=inputs.get(tool_use_id, {}),  # mutable-ok: empty provider input
                 result=result,
             )
             for result in web_search_results

litellm/llms/anthropic/common_utils.py

@@ -1300,12 +1300,12 @@ def _without_encrypted_reasoning_blocks(message: dict) -> dict | None:  # mutabl
     content: Final = message.get("content")
     if not isinstance(content, list):
         return message
-    kept: Final = [b for b in content if not is_encrypted_reasoning_block(b)]
+    kept: Final = [b for b in content if not is_encrypted_reasoning_block(b)]  # mutable-ok: API message payload
     if len(kept) == len(content):
         return message
     if not kept:
         return None
-    return {**message, "content": kept}
+    return {**message, "content": kept}  # mutable-ok: API message payload
 
 
 def strip_encrypted_reasoning_blocks_from_anthropic_messages(
@@ -1317,7 +1317,7 @@ def strip_encrypted_reasoning_blocks_from_anthropic_messages(
     Anthropic, which cannot verify them. Anthropic's own signed blocks are kept.
     """
     stripped: Final = (_without_encrypted_reasoning_blocks(m) for m in messages)
-    return [m for m in stripped if m is not None]
+    return [m for m in stripped if m is not None]  # mutable-ok: API message payload
 
 
 def strip_thinking_blocks_from_anthropic_messages_request_dict(
@@ -1605,7 +1605,7 @@ def _flatten_web_search_results_in_message(message: object) -> object:
         }
     )
     rewritten: Final = tuple(_rewrite_replayed_web_search_block(block, flattenable, queries) for block in content)
-    return {**message, "content": [b for b in rewritten if b is not None]}
+    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(
@@ -1623,47 +1623,49 @@ def flatten_unencrypted_web_search_results_in_anthropic_messages(
     evidence in the conversation instead of 400ing the follow-up turn, and leaves
     genuine Anthropic-issued blocks untouched.
     """
-    return [_flatten_web_search_results_in_message(m) for m in messages]
+    return [_flatten_web_search_results_in_message(m) for m in messages]  # mutable-ok: JSON wire format
 
 
 def _without_provider_specific_fields(block: object) -> object:
     if not isinstance(block, dict) or "provider_specific_fields" not in block:
         return block
-    return {k: v for k, v in block.items() if k != "provider_specific_fields"}
+    return {k: v for k, v in block.items() if k != "provider_specific_fields"}  # mutable-ok: JSON wire format
 
 
 def _strip_provider_specific_fields_in_message(message: object) -> object:
     if not isinstance(message, dict) or not isinstance(message.get("content"), list):
         return message
-    content: Final = [_without_provider_specific_fields(b) for b in message["content"]]
-    return {**message, "content": content}
+    content: Final = [_without_provider_specific_fields(b) for b in message["content"]]  # mutable-ok: JSON wire format
+    return {**message, "content": content}  # mutable-ok: JSON wire format
 
 
 def strip_provider_specific_fields_from_anthropic_messages(
     messages: Sequence[object],
 ) -> Sequence[object]:
-    return [_strip_provider_specific_fields_in_message(m) for m in messages]
+    return [_strip_provider_specific_fields_in_message(m) for m in messages]  # mutable-ok: JSON wire format
 
 
 def _normalized_cache_control(cache_control: object) -> dict[str, str] | None:  # mutable-ok: JSON wire format
     if not isinstance(cache_control, Mapping):
         return None
     cache_type: Final = cache_control.get("type")
-    return {"type": cache_type if isinstance(cache_type, str) else "ephemeral"}
+    return {"type": cache_type if isinstance(cache_type, str) else "ephemeral"}  # mutable-ok: JSON wire format
 
 
 def _with_portable_cache_control(block: Mapping[str, object]) -> dict[str, object]:  # mutable-ok: JSON wire format
     if "cache_control" not in block:
-        return dict(block)
+        return dict(block)  # mutable-ok: JSON wire format
     normalized: Final = _normalized_cache_control(block["cache_control"])
-    rest: Final = {key: value for key, value in block.items() if key != "cache_control"}
-    return rest if normalized is None else {**rest, "cache_control": normalized}
+    rest: Final = {key: value for key, value in block.items() if key != "cache_control"}  # mutable-ok: JSON wire format
+    return rest if normalized is None else {**rest, "cache_control": normalized}  # mutable-ok: JSON wire format
 
 
 def _with_portable_cache_control_in_blocks(blocks: object) -> object:
     if isinstance(blocks, str) or not isinstance(blocks, Sequence):
         return blocks
-    return [_with_portable_cache_control(block) if isinstance(block, Mapping) else block for block in blocks]
+    return [  # mutable-ok: JSON wire format
+        _with_portable_cache_control(block) if isinstance(block, Mapping) else block for block in blocks
+    ]
 
 
 def _with_portable_cache_control_in_content_block(block: object) -> object:
@@ -1672,7 +1674,7 @@ def _with_portable_cache_control_in_content_block(block: object) -> object:
     portable: Final = _with_portable_cache_control(block)
     if portable.get("type") != "tool_result" or "content" not in portable:
         return portable
-    return {
+    return {  # mutable-ok: JSON wire format
         **portable,
         "content": _with_portable_cache_control_in_blocks(portable["content"]),
     }
@@ -1684,16 +1686,20 @@ def _with_portable_cache_control_in_message(message: object) -> object:
     content: Final = message["content"]
     if isinstance(content, str) or not isinstance(content, Sequence):
         return message
-    return {
+    return {  # mutable-ok: JSON wire format
         **message,
-        "content": [_with_portable_cache_control_in_content_block(block) for block in content],
+        "content": [  # mutable-ok: JSON wire format
+            _with_portable_cache_control_in_content_block(block) for block in content
+        ],
     }
 
 
 def _with_portable_cache_control_in_messages(messages: object) -> object:
     if isinstance(messages, str) or not isinstance(messages, Sequence):
         return messages
-    return [_with_portable_cache_control_in_message(message) for message in messages]
+    return [  # mutable-ok: JSON wire format
+        _with_portable_cache_control_in_message(message) for message in messages
+    ]
 
 
 def _with_portable_cache_control_in_scoped_value(key: str, value: object) -> object:
@@ -1725,7 +1731,9 @@ def normalize_cache_control_in_anthropic_payload(
     dropped entirely. The caller's payload is never mutated.
     """
     portable: Final = _with_portable_cache_control(payload)
-    return {key: _with_portable_cache_control_in_scoped_value(key, value) for key, value in portable.items()}
+    return {  # mutable-ok: JSON wire format
+        key: _with_portable_cache_control_in_scoped_value(key, value) for key, value in portable.items()
+    }
 
 
 def process_anthropic_headers(headers: httpx.Headers | dict) -> dict:
@@ -1752,7 +1760,7 @@ def _anthropic_model_entry(
     source: Final[Mapping[str, object]] = (
         MappingProxyType({"source_model": model["id"]}) if listed_id is not None else MappingProxyType({})
     )
-    return {
+    return {  # mutable-ok: JSON response body, serialized by the route and never mutated
         "type": "model",
         "id": listed_id or model["id"],
         **source,
@@ -1783,8 +1791,10 @@ def create_anthropic_model_list_response(
     created_at: Final = (
         datetime.fromtimestamp(DEFAULT_MODEL_CREATED_AT_TIME, tz=timezone.utc).isoformat().replace("+00:00", "Z")
     )
-    data: Final = [_anthropic_model_entry(model, created_at, display_names, listed_ids) for model in models]
-    return {
+    data: Final = [  # mutable-ok: JSON response body, serialized by the route and never mutated
+        _anthropic_model_entry(model, created_at, display_names, listed_ids) for model in models
+    ]
+    return {  # mutable-ok: JSON response body, serialized by the route and never mutated
         "data": data,
         "has_more": False,
         "first_id": data[0]["id"] if data else None,

litellm/llms/anthropic/pass_through/adapters/streaming_iterator.py

@@ -1212,7 +1212,9 @@ class AnthropicSSEStream(AsyncIterator[bytes]):
     def __init__(self, anthropic_wrapper: AnthropicStreamWrapper) -> None:
         self._anthropic_wrapper = anthropic_wrapper
         self._byte_stream: Final[AsyncIterator[bytes]] = anthropic_wrapper.async_anthropic_sse_wrapper()
-        self._hidden_params: dict[str, object] = {}
+        self._hidden_params: dict[
+            str, object
+        ] = {}  # mutable-ok: the proxy merges provider headers onto _hidden_params in place
 
     @property
     def chunks(self) -> "list[ModelResponseStream] | None":

litellm/llms/anthropic/pass_through/adapters/transformation.py

@@ -1314,7 +1314,7 @@ class LiteLLMAnthropicMessagesAdapter:
             case ({"type": "text", "text": str(text)},):
                 return text
             case _:
-                return list(parts)
+                return list(parts)  # mutable-ok: content must be a json list
 
     def _tool_result_part(self, item: object) -> ToolMessageContentPart | None:
         if isinstance(item, str):

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

@@ -132,7 +132,7 @@ class CachedAnthropicMessagesStreamIterator(BaseAnthropicMessagesStreamingIterat
         litellm_logging_obj: "LiteLLMLoggingObj",
         request_body: Mapping[str, object],
     ) -> None:
-        body: Final = dict(request_body)
+        body: Final = dict(request_body)  # mutable-ok: the base iterator takes a plain dict
         super().__init__(litellm_logging_obj=litellm_logging_obj, request_body=body)
         self.chunks: Final[tuple[bytes, ...]] = tuple(event.encode("utf-8") for event in events)
         self.current_index = 0
@@ -147,7 +147,7 @@ class CachedAnthropicMessagesStreamIterator(BaseAnthropicMessagesStreamingIterat
         if self.current_index >= len(self.chunks):
             if not self.logged:
                 self.logged = True
-                chunks: Final = list(self.chunks)
+                chunks: Final = list(self.chunks)  # mutable-ok: the logging handler takes a list
                 await self._handle_streaming_logging(chunks)
             raise StopAsyncIteration
         chunk: Final = self.chunks[self.current_index]

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

@@ -212,15 +212,15 @@ def _anthropic_content_block_start_and_deltas(
     match block.get("type"):
         case "tool_use":
             return (
-                {
+                {  # mutable-ok: one-shot payload
                     "id": block.get("id"),
                     "name": block.get("name"),
-                    "input": {},
+                    "input": {},  # mutable-ok: one-shot payload
                     "type": "tool_use",
                 },
                 (
-                    {
-                        "partial_json": json.dumps(block.get("input") or {}),
+                    {  # mutable-ok: one-shot payload
+                        "partial_json": json.dumps(block.get("input") or {}),  # mutable-ok: one-shot payload
                         "type": "input_json_delta",
                     },
                 ),
@@ -228,23 +228,23 @@ def _anthropic_content_block_start_and_deltas(
         case "thinking":
             signature: Final = block.get("signature")
             signature_deltas: Final = (
-                ({"signature": signature, "type": "signature_delta"},)
+                ({"signature": signature, "type": "signature_delta"},)  # mutable-ok: one-shot payload
                 if isinstance(signature, str) and signature
                 else ()
             )
             return (
-                {"thinking": "", "signature": "", "type": "thinking"},
+                {"thinking": "", "signature": "", "type": "thinking"},  # mutable-ok: one-shot payload
                 (
-                    {"thinking": block.get("thinking") or "", "type": "thinking_delta"},
+                    {"thinking": block.get("thinking") or "", "type": "thinking_delta"},  # mutable-ok: one-shot payload
                     *signature_deltas,
                 ),
             )
         case "redacted_thinking":
-            return ({"type": "redacted_thinking", "data": block.get("data")}, ())
+            return ({"type": "redacted_thinking", "data": block.get("data")}, ())  # mutable-ok: one-shot JSON payload
         case _:
             return (
-                {"type": "text", "text": ""},
-                ({"type": "text_delta", "text": block.get("text") or ""},),
+                {"type": "text", "text": ""},  # mutable-ok: one-shot JSON payload
+                ({"type": "text_delta", "text": block.get("text") or ""},),  # mutable-ok: one-shot JSON payload
             )
 
 
@@ -268,51 +268,51 @@ def anthropic_messages_response_as_sse_events(response: AnthropicMessagesRespons
     # a zero output_tokens - those are only known once generation finishes, so
     # copying the completed response's final values here would let a client
     # treat the message as already finished, or double-count output tokens.
-    message_start_usage: Final = {
+    message_start_usage: Final = {  # mutable-ok: one-shot JSON payload
         **(response.get("usage") or {}),
         "output_tokens": 0,
     }
-    message_start_payload: Final = {
+    message_start_payload: Final = {  # mutable-ok: one-shot JSON payload, never mutated after construction
         "type": "message_start",
-        "message": {
+        "message": {  # mutable-ok: one-shot JSON payload
             **response,
-            "content": [],
+            "content": [],  # mutable-ok: one-shot JSON payload
             "stop_reason": None,
             "stop_sequence": None,
             "usage": message_start_usage,
         },
     }
-    message_delta_payload: Final = {
+    message_delta_payload: Final = {  # mutable-ok: one-shot JSON payload, never mutated after construction
         "type": "message_delta",
-        "delta": {
+        "delta": {  # mutable-ok: one-shot JSON payload
             "stop_reason": response.get("stop_reason"),
             "stop_sequence": response.get("stop_sequence"),
         },
-        "usage": response.get("usage") or {},
+        "usage": response.get("usage") or {},  # mutable-ok: one-shot JSON payload
     }
     return (
         _sse_event("message_start", message_start_payload),
         *content_events,
         _sse_event("message_delta", message_delta_payload),
-        _sse_event("message_stop", {"type": "message_stop"}),
+        _sse_event("message_stop", {"type": "message_stop"}),  # mutable-ok: one-shot JSON payload
     )
 
 
 def _anthropic_content_block_events(index: int, block: Mapping[str, object]) -> tuple[bytes, ...]:
     start_block, deltas = _anthropic_content_block_start_and_deltas(block)
-    start_payload: Final = {
+    start_payload: Final = {  # mutable-ok: one-shot payload
         "type": "content_block_start",
         "index": index,
         "content_block": start_block,
     }
-    stop_payload: Final = {
+    stop_payload: Final = {  # mutable-ok: one-shot payload
         "type": "content_block_stop",
         "index": index,
     }
     delta_events: Final = tuple(
         _sse_event(
             "content_block_delta",
-            {"type": "content_block_delta", "index": index, "delta": delta},
+            {"type": "content_block_delta", "index": index, "delta": delta},  # mutable-ok: one-shot payload
         )
         for delta in deltas
     )

litellm/llms/anthropic/pass_through/responses_adapters/streaming_iterator.py

@@ -191,20 +191,20 @@ class AnthropicResponsesStreamWrapper:
         if block_idx < 0:
             redacted_idx: Final = self._open_block(
                 item_id,
-                {"type": "redacted_thinking", "data": signature},
+                {"type": "redacted_thinking", "data": signature},  # mutable-ok: API message payload
             )
-            stop: Final = {"type": "content_block_stop", "index": redacted_idx}
+            stop: Final = {"type": "content_block_stop", "index": redacted_idx}  # mutable-ok: API message payload
             self._chunk_queue.append(stop)
             return
         if signature is not None:
             self._chunk_queue.append(
-                {
+                {  # mutable-ok: API message payload
                     "type": "content_block_delta",
                     "index": block_idx,
-                    "delta": {"type": "signature_delta", "signature": signature},
+                    "delta": {"type": "signature_delta", "signature": signature},  # mutable-ok: API message payload
                 }
             )
-        self._chunk_queue.append({"type": "content_block_stop", "index": block_idx})
+        self._chunk_queue.append({"type": "content_block_stop", "index": block_idx})  # mutable-ok: API message payload
 
     def _process_event(self, event: object) -> None:
         """Convert one Responses API event into zero or more Anthropic chunks queued for emission."""
@@ -296,10 +296,10 @@ class AnthropicResponsesStreamWrapper:
             if part_block_idx < 0 or not isinstance(summary_index, int) or summary_index == 0:
                 return
             self._chunk_queue.append(
-                {
+                {  # mutable-ok: API message payload
                     "type": "content_block_delta",
                     "index": part_block_idx,
-                    "delta": {
+                    "delta": {  # mutable-ok: API message payload
                         "type": "thinking_delta",
                         "thinking": REASONING_SUMMARY_PART_SEPARATOR,
                     },
@@ -317,7 +317,7 @@ class AnthropicResponsesStreamWrapper:
                     return
                 block_idx = self._open_block(
                     item_id,
-                    {"type": "thinking", "thinking": "", "signature": ""},
+                    {"type": "thinking", "thinking": "", "signature": ""},  # mutable-ok: API message payload
                 )
             self._chunk_queue.append(
                 {
@@ -413,10 +413,16 @@ class AnthropicResponsesStreamWrapper:
                 else AnthropicUsage(input_tokens=0, output_tokens=0)
             )
 
-            message_delta_payload: Final = {
+            message_delta_payload: Final = {  # mutable-ok: fresh message_delta payload built per chunk
                 "stop_reason": stop_reason,
                 "stop_sequence": None,
-                **({"stop_details": refusal_stop_details(refusal_text)} if stop_reason == "refusal" else {}),
+                **(
+                    {  # mutable-ok: fresh message_delta stop_details entry built per chunk
+                        "stop_details": refusal_stop_details(refusal_text)
+                    }
+                    if stop_reason == "refusal"
+                    else {}  # mutable-ok: empty spread placeholder for non-refusal stop
+                ),
             }
 
             self._chunk_queue.append(

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

@@ -115,7 +115,7 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
             )
             raw_title: Final = block.get("title")
             filename: Final = raw_title if isinstance(raw_title, str) and raw_title else "document.pdf"
-            return {
+            return {  # mutable-ok: API message payload
                 "type": "input_file",
                 "filename": filename,
                 "file_data": f"data:{media_type};base64,{data}",
@@ -124,7 +124,7 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
             url: Final = source.get("url")
             if not isinstance(url, str) or not url:
                 return None
-            return {"type": "input_file", "file_url": url}
+            return {"type": "input_file", "file_url": url}  # mutable-ok: API message payload
         return None
 
     @staticmethod
@@ -135,8 +135,10 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
         """Plain string output, or a part list when document file parts are present."""
         if not file_parts:
             return output_text
-        text_parts: Final = [{"type": "input_text", "text": output_text}] if output_text else []
-        return [*text_parts, *file_parts]
+        text_parts: Final = (
+            [{"type": "input_text", "text": output_text}] if output_text else []  # mutable-ok: API message payload
+        )
+        return [*text_parts, *file_parts]  # mutable-ok: API message payload
 
     @staticmethod
     def _translate_midturn_system_content_to_responses(
@@ -144,10 +146,12 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
     ) -> list[dict[str, object]]:  # mutable-ok: API message payload
         """Convert in-sequence system content to Responses input-text parts."""
         if isinstance(content, str):
-            return [{"type": "input_text", "text": content}] if content else []
+            return (
+                [{"type": "input_text", "text": content}] if content else []  # mutable-ok: API message payload
+            )
         if not isinstance(content, list):
-            return []
-        return [
+            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"))
             for block in content
             if isinstance(block, dict) and block.get("type") == "text" and (text := block.get("text"))  # pyright: ignore[reportUnnecessaryIsInstance]  # untrusted client payload
@@ -199,14 +203,14 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
         btype: Final = first.get("type")
         if btype in ("thinking", "redacted_thinking"):
             replayed: Final = responses_reasoning_items_from_thinking_blocks(group)
-            return tuple(dict(item) for item in replayed)
+            return tuple(dict(item) for item in replayed)  # mutable-ok: API message payload
         if btype == "tool_use":
             return (
-                {
+                {  # mutable-ok: API message payload
                     "type": "function_call",
                     "call_id": first.get("id", ""),
                     "name": first.get("name", ""),
-                    "arguments": json.dumps(first.get("input", {})),
+                    "arguments": json.dumps(first.get("input", {})),  # mutable-ok: API message payload
                 },
             )
         return ()
@@ -235,7 +239,7 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
                 system_parts = self._translate_midturn_system_content_to_responses(m.get("content"))
                 if system_parts:
                     input_items.append(
-                        {
+                        {  # mutable-ok: API message payload
                             "type": "message",
                             "role": "system",
                             "content": system_parts,
@@ -318,7 +322,8 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
                                         else TOOL_RESULT_IMAGE_PLACEHOLDER
                                     )
                                     tool_image_parts.extend(
-                                        {"type": "input_image", "image_url": url} for url in image_urls
+                                        {"type": "input_image", "image_url": url}  # mutable-ok: json content part
+                                        for url in image_urls
                                     )
                             else:
                                 output_text = str(inner)
@@ -331,15 +336,15 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
                                 }
                             )
                     if tool_image_parts:
-                        boundary_part = {
+                        boundary_part = {  # mutable-ok: json content part
                             "type": "input_text",
                             "text": TOOL_RESULT_IMAGE_BOUNDARY,
                         }
                         input_items.append(
-                            {
+                            {  # mutable-ok: json input item
                                 "type": "message",
                                 "role": "user",
-                                "content": [boundary_part, *tool_image_parts],
+                                "content": [boundary_part, *tool_image_parts],  # mutable-ok: json content list
                             }
                         )
                     if user_parts:
@@ -368,7 +373,7 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
                         for item in self._assistant_group_to_input_items(tuple(block for _, block in group))
                     )
                     asst_parts: list[dict[str, Any]] = [  # mutable-ok: API message payload
-                        {"type": "output_text", "text": block.get("text", "")}
+                        {"type": "output_text", "text": block.get("text", "")}  # mutable-ok: API message payload
                         for block in blocks
                         if block.get("type") == "text"
                     ]
@@ -526,7 +531,7 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
         if developer_parts:
             input_items.insert(
                 0,
-                {
+                {  # mutable-ok: API message payload
                     "type": "message",
                     "role": "developer",
                     "content": developer_parts,
@@ -538,7 +543,7 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
             "input": input_items,
         }
         if include_encrypted_reasoning:
-            responses_kwargs["include"] = [RESPONSES_INCLUDE_ENCRYPTED_REASONING]
+            responses_kwargs["include"] = [RESPONSES_INCLUDE_ENCRYPTED_REASONING]  # mutable-ok: API request payload
 
         if system and not developer_parts:
             if isinstance(system, str):

litellm/llms/anthropic/prompt_cache_prediction.py

@@ -505,7 +505,7 @@ class TokenCounter(Protocol):
 def _count_objects(
     values: Sequence[Mapping[str, JsonValue]],
 ) -> list[dict[str, JsonValue]]:  # mutable-ok: the existing provider count API requires JSON lists/dicts
-    return [dict(value) for value in values]
+    return [dict(value) for value in values]  # mutable-ok: serialize read-only inputs at the provider API boundary
 
 
 def _messages_url(model: str, api_key: str, api_base: str | None) -> str:

litellm/llms/azure/azure.py

@@ -1279,7 +1279,7 @@ class AzureChatCompletion(BaseAzureLLM, BaseLLM):
                 api_base=api_base,
                 is_async=False,
             )
-            request_headers: Final = dict(
+            request_headers: Final = dict(  # mutable-ok: the httpx request helpers take a dict
                 get_azure_request_auth_headers(headers=headers, azure_client_params=azure_client_params)
             )
             if aimg_generation is True:
@@ -1411,7 +1411,7 @@ class AzureChatCompletion(BaseAzureLLM, BaseLLM):
         logging_obj.pre_call(
             input=input,
             api_key=api_key,
-            additional_args={
+            additional_args={  # mutable-ok: loggers isinstance-check this payload as a dict
                 "complete_input_dict": speech_request_body(model, voice, optional_params),
                 "api_base": str(azure_client.base_url),
             },
@@ -1455,7 +1455,7 @@ class AzureChatCompletion(BaseAzureLLM, BaseLLM):
         logging_obj.pre_call(
             input=input,
             api_key=api_key,
-            additional_args={
+            additional_args={  # mutable-ok: loggers isinstance-check this payload as a dict
                 "complete_input_dict": speech_request_body(model, voice, optional_params),
                 "api_base": str(azure_client.base_url),
             },

litellm/llms/azure/chat/gpt_transformation.py

@@ -44,7 +44,7 @@ def sanitized_tools_update(optional_params: Mapping[str, object]) -> Mapping[str
     tools: Final = optional_params.get("tools")
     if not isinstance(tools, list):
         return _NO_TOOLS_UPDATE
-    sanitized: Final = [
+    sanitized: Final = [  # mutable-ok: request tools are a JSON list
         tool_with_sanitized_parameters(tool, flatten_combinators_and_drop_non_python_regex_patterns)
         if isinstance(tool, dict)
         else tool

litellm/llms/azure/chat/o_series_transformation.py

@@ -109,7 +109,7 @@ class AzureOpenAIO1Config(OpenAIOSeriesConfig):
         headers: dict,
     ) -> dict:
         model = model.replace("o_series/", "")  # handle o_series/my-random-deployment-name
-        flattened_params: Final = {
+        flattened_params: Final = {  # mutable-ok: transform_request's contract takes a plain JSON params dict
             **optional_params,
             **sanitized_tools_update(optional_params),
         }

litellm/llms/azure/common_utils.py

@@ -440,7 +440,7 @@ def get_azure_request_auth_headers(
 
 
 def redact_azure_auth_headers(headers: Mapping[str, str]) -> Mapping[str, str]:
-    return {
+    return {  # mutable-ok: logging callbacks JSON-serialize this copy
         name: (_REDACTED_AZURE_HEADER_VALUE if name.lower() in _AZURE_AUTH_HEADER_NAMES else value)
         for name, value in headers.items()
     }

litellm/llms/azure/search/transformation.py

@@ -289,7 +289,7 @@ class BingGroundingSearchConfig(BaseSearchConfig):
         Returns a new dict rather than mutating ``headers``: the http handler calls this
         a second time after ``litellm/search/main.py`` already did, so it has to be idempotent.
         """
-        return {
+        return {  # mutable-ok: httpx requires a plain dict of headers
             **headers,
             **self._auth_header(api_key, api_base),
             "Content-Type": "application/json",
@@ -387,7 +387,7 @@ class BingGroundingSearchConfig(BaseSearchConfig):
             raise self.get_error_class(
                 error_message=f"response does not match the Foundry Responses API schema: {e}",
                 status_code=raw_response.status_code,
-                headers=dict(raw_response.headers),
+                headers=dict(raw_response.headers),  # mutable-ok: BaseSearchConfig.get_error_class signature
             )
         if parsed.status == "failed":
             detail: Final = (
@@ -408,7 +408,7 @@ class BingGroundingSearchConfig(BaseSearchConfig):
         return self.get_error_class(
             error_message=detail,
             status_code=_UPSTREAM_ERROR_STATUS,
-            headers=dict(raw_response.headers),
+            headers=dict(raw_response.headers),  # mutable-ok: BaseSearchConfig.get_error_class signature
         )
 
     def _priced(self, results: tuple[SearchResult, ...]) -> SearchResponse:
@@ -416,12 +416,16 @@ class BingGroundingSearchConfig(BaseSearchConfig):
         inherit the connection-mode ``bing_grounding/search`` price; zero its per-query
         cost while leaving connection mode to the cost map."""
         response: Final = SearchResponse(
-            results=list(results),
+            results=list(results),  # mutable-ok: SearchResponse.results is list[SearchResult]
             object="search",
         )
         if get_secret_str(CONNECTION_ID_ENV):
             return response
-        response._hidden_params["additional_headers"] = {_RESPONSE_COST_HEADER: 0.0}
+        response._hidden_params[
+            "additional_headers"
+        ] = {  # mutable-ok: response_cost_calculator writes into _hidden_params
+            _RESPONSE_COST_HEADER: 0.0
+        }
         return response
 
     def get_error_class(

litellm/llms/azure_ai/azure_model_router/transformation.py

@@ -102,7 +102,7 @@ class AzureModelRouterConfig(AzureAIStudioConfig):
         if selected_model:
             # Rebuilt rather than mutated in place: ModelResponseBase declares _hidden_params as a
             # class-level dict, so an in-place write can bleed into unrelated responses.
-            transformed_response._hidden_params = {  # pyright: ignore[reportPrivateUsage]  # ModelResponse exposes no public hidden-params setter
+            transformed_response._hidden_params = {  # pyright: ignore[reportPrivateUsage]  # ModelResponse exposes no public hidden-params setter  # mutable-ok: ModelResponse requires _hidden_params to be a plain dict
                 **get_hidden_params_dict(transformed_response),
                 AZURE_MODEL_ROUTER_SELECTED_MODEL_KEY: selected_model,
             }

litellm/llms/azure_ai/image_generation/flux_transformation.py

@@ -76,7 +76,7 @@ class AzureFoundryFluxImageGenerationConfig(GPTImageGenerationConfig):
     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 [
+        return [  # mutable-ok: BaseImageGenerationConfig requires a list
             "n",
             "size",
             "output_format",
@@ -151,4 +151,4 @@ class AzureFoundryFluxImageGenerationConfig(GPTImageGenerationConfig):
                 for mapped_name, mapped_value in self._map_parameter(name, value, model)
             }
         )
-        return {**optional_params, **mapped_params}
+        return {**optional_params, **mapped_params}  # mutable-ok: inherited config contract returns a dict

litellm/llms/azure_ai/passthrough/transformation.py

@@ -134,7 +134,7 @@ class AzureAIPassthroughConfig(AzureFoundryModelInfo, BasePassthroughConfig):
             litellm_params=litellm_params,
             api_key_header=api_key_header_for_base(api_base),
         )
-        return {**headers, **auth_headers}
+        return {**headers, **auth_headers}  # mutable-ok: base class contract returns dict for httpx
 
     def logging_non_streaming_response(
         self,
@@ -151,7 +151,7 @@ class AzureAIPassthroughConfig(AzureFoundryModelInfo, BasePassthroughConfig):
             model=model,
             custom_llm_provider=custom_llm_provider,
             httpx_response=httpx_response,
-            request_data=dict(request_data),
+            request_data=dict(request_data),  # mutable-ok: AzurePassthroughConfig wants a dict
             logging_obj=logging_obj,
             endpoint=endpoint,
         )

litellm/llms/azure_ai/responses/transformation.py

@@ -34,7 +34,7 @@ class AzureAIResponsesAPIConfig(AzureOpenAIResponsesAPIConfig):
             litellm_params=params.model_dump(),
             api_key_header=api_key_header_for_base(AzureFoundryModelInfo.get_api_base(params.api_base)),
         )
-        return {
+        return {  # mutable-ok: the handler updates the returned headers in place per the dict contract
             **headers,
             **auth_headers,
             "Content-Type": "application/json",

litellm/llms/base_llm/guardrail_translation/utils.py

@@ -389,12 +389,12 @@ def message_text_slot_count(message: AllMessageValues) -> int:
 def _part_with_text(part: object, text: str) -> object:
     if not isinstance(part, Mapping):
         return part
-    return {**part, "text": text}
+    return {**part, "text": text}  # mutable-ok: content parts stay JSON-plain dicts
 
 
 def _content_with_slot_texts(content: Sequence[object], texts: Sequence[str]) -> Sequence[object]:
     remaining_texts: Final = iter(texts)
-    return [
+    return [  # mutable-ok: message content stays a JSON list
         _part_with_text(part, next(remaining_texts)) if _content_part_text(part) is not None else part
         for part in content
     ]
@@ -413,7 +413,7 @@ def message_with_slot_texts(message: AllMessageValues, texts: Sequence[str]) ->
     if not isinstance(content, (str, list)):
         return message
     rewritten_content: Final = texts[0] if isinstance(content, str) else _content_with_slot_texts(content, texts)
-    rewritten: Final = {**message, "content": rewritten_content}
+    rewritten: Final = {**message, "content": rewritten_content}  # mutable-ok: chat rows stay JSON-plain dicts
     return cast("AllMessageValues", rewritten)  # cast-ok: the same row with only its text slots swapped
 
 

litellm/llms/base_llm/responses/codex_compat.py

@@ -129,7 +129,7 @@ def normalize_codex_input_items(
         return input, ()
     normalized: Final = tuple(_normalize_input_item(item) for item in input)
     rewritten_types: Final = tuple(sorted(frozenset(item_type for _, item_type in normalized if item_type is not None)))
-    kept: Final = [i for i, _ in normalized if i is not None]
+    kept: Final = [i for i, _ in normalized if i is not None]  # mutable-ok: downstream narrows on isinstance(list)
     # Codex passthrough items sit outside the OpenAI input union.
     return kept, rewritten_types  # pyright: ignore[reportReturnType]  # see above
 

litellm/llms/base_llm/search/transformation.py

@@ -280,7 +280,7 @@ class BaseSearchConfig:
         return self.get_error_class(
             error_message=error.response.text,
             status_code=error.response.status_code,
-            headers=dict(error.response.headers),
+            headers=dict(error.response.headers),  # mutable-ok: provider error factories require dict headers
         )
 
     def get_error_class(

litellm/llms/base_llm/vector_store/transformation.py

@@ -48,8 +48,8 @@ class LiteLLMVectorStoreEmbeddingExecutor:
 
         return litellm.embedding(  # pyright: ignore[reportCallIssue, reportUnknownMemberType, reportUnknownVariableType]  # provider kwargs are intentionally dynamic
             model=model,
-            input=[query],
-            **dict(configuration),  # pyright: ignore[reportArgumentType]  # provider-specific embedding config is validated downstream
+            input=[query],  # mutable-ok: LiteLLM embedding requires a mutable input list
+            **dict(configuration),  # pyright: ignore[reportArgumentType]  # provider-specific embedding config is validated downstream  # mutable-ok: kwargs require a concrete dict
         )
 
     async def aembed(self, model: str, query: str, configuration: Mapping[str, object]) -> EmbeddingResponse:
@@ -57,8 +57,8 @@ class LiteLLMVectorStoreEmbeddingExecutor:
 
         return await litellm.aembedding(  # pyright: ignore[reportUnknownMemberType]  # provider kwargs are intentionally dynamic
             model=model,
-            input=[query],
-            **dict(configuration),  # pyright: ignore[reportArgumentType]  # provider-specific embedding config is validated downstream
+            input=[query],  # mutable-ok: LiteLLM embedding requires a mutable input list
+            **dict(configuration),  # pyright: ignore[reportArgumentType]  # provider-specific embedding config is validated downstream  # mutable-ok: kwargs require a concrete dict
         )
 
 
@@ -105,7 +105,7 @@ class RouterVectorStoreEmbeddingExecutor:
             return LiteLLMVectorStoreEmbeddingExecutor().embed(model, query, embedding_kwargs)
         return self.router.embedding(  # pyright: ignore[reportUnknownMemberType]  # Router embedding input retains a legacy untyped list
             model=model,
-            input=[query],
+            input=[query],  # mutable-ok: Router embedding requires a mutable input list
             **embedding_kwargs,  # pyright: ignore[reportArgumentType]  # provider kwargs are intentionally dynamic
         )
 
@@ -115,7 +115,7 @@ class RouterVectorStoreEmbeddingExecutor:
             return await LiteLLMVectorStoreEmbeddingExecutor().aembed(model, query, embedding_kwargs)
         return await self.router.aembedding(  # pyright: ignore[reportUnknownMemberType]  # Router embedding input retains a legacy untyped list
             model=model,
-            input=[query],
+            input=[query],  # mutable-ok: Router embedding requires a mutable input list
             **embedding_kwargs,  # pyright: ignore[reportArgumentType]  # provider kwargs are intentionally dynamic
         )
 
@@ -429,4 +429,4 @@ class BaseDirectVectorStoreConfig(BaseVectorStoreConfig):
         return BaseVectorStoreAuthCredentials()
 
     def get_vector_store_endpoints_by_type(self) -> VectorStoreIndexEndpoints:
-        return VectorStoreIndexEndpoints(read=[], write=[])
+        return VectorStoreIndexEndpoints(read=[], write=[])  # mutable-ok: the TypedDict declares list fields

litellm/llms/bedrock/chat/converse_transformation.py

@@ -1434,7 +1434,7 @@ class AmazonConverseConfig(BaseConfig):
         if not text_blocks:
             return None
         note: Final = ChatCompletionTextObject(type="text", text=CONVERTED_SYSTEM_NOTE)
-        body: Final = [
+        body: Final = [  # mutable-ok: _bedrock_converse_messages_pt narrows content with isinstance(list)
             note,
             *text_blocks,
         ]
@@ -1501,7 +1501,7 @@ class AmazonConverseConfig(BaseConfig):
             )
         )
         converted: Final = tuple(self._converted_or_kept(message) for message in reordered)
-        kept: Final = [message for message in converted if message is not None]
+        kept: Final = [message for message in converted if message is not None]  # mutable-ok: converse pt takes a list
         return kept, system_content_blocks
 
     def _transform_inference_params(self, inference_params: dict) -> InferenceConfig:

litellm/llms/bedrock/common_utils.py

@@ -100,7 +100,7 @@ def merge_bedrock_aws_request_params(
     server. Requests may still provide AWS credentials when the deployment has
     no static credentials configured.
     """
-    request_params: Final = {**optional_params, **litellm_params}
+    request_params: Final = {**optional_params, **litellm_params}  # mutable-ok: AWS helpers require a plain dict
     has_static_deployment_credentials: Final = all(
         isinstance(litellm_params.get(key), str) and bool(litellm_params.get(key))
         for key in ("aws_access_key_id", "aws_secret_access_key", "aws_region_name")
@@ -258,7 +258,7 @@ def apply_bedrock_invoke_structured_output(
         if isinstance(existing_output_config, dict):
             existing_output_config["format"] = schema_format
         else:
-            request_body["output_config"] = {"format": schema_format}  # rebind-ok: out-param
+            request_body["output_config"] = {"format": schema_format}  # rebind-ok: out-param  # mutable-ok: json
         return
 
     verbose_logger.warning(
@@ -311,7 +311,7 @@ def strip_unsupported_bedrock_invoke_output_config_keys(
     if preserved_format is None:
         request_body.pop("output_config", None)
     else:
-        request_body["output_config"] = {"format": preserved_format}  # rebind-ok: out-param
+        request_body["output_config"] = {"format": preserved_format}  # rebind-ok: out-param  # mutable-ok: json
 
 
 def normalize_custom_field_on_tools(request_body: dict) -> None:

litellm/llms/bedrock/files/transformation.py

@@ -1384,7 +1384,7 @@ class BedrockFilesConfig(BaseAWSLLM, BaseFilesConfig):
             _listed_managed_file(entry, bucket_name, configured_bucket_name, allow_legacy_cloud_file_ids)
             for entry in listing.iterfind("{*}Contents")
         )
-        return [
+        return [  # mutable-ok: the base files contract returns a list
             listed_file
             for listed_file in listed_files
             if listed_file is not None and (purpose is None or listed_file.purpose == purpose)
@@ -1429,7 +1429,7 @@ class BedrockFilesConfig(BaseAWSLLM, BaseFilesConfig):
             request_params=target.request_params,
         )
         litellm_params[S3_SIGNED_REQUEST_HEADERS_PARAM] = signed_headers  # rebind-ok: handed to validate_environment
-        return url, {}
+        return url, {}  # mutable-ok: the base files contract returns the query as a dict
 
     def _s3_request_target(
         self,
@@ -1446,7 +1446,7 @@ class BedrockFilesConfig(BaseAWSLLM, BaseFilesConfig):
         )
         region_preference: Final = request_params.s3_region_name or request_params.aws_region_name
         aws_region_name: Final = self._get_aws_region_name(
-            optional_params={"aws_region_name": region_preference},
+            optional_params={"aws_region_name": region_preference},  # mutable-ok: BaseAWSLLM takes a dict
             model="",
         )
         endpoint_url: Final = (
@@ -1481,7 +1481,7 @@ class BedrockFilesConfig(BaseAWSLLM, BaseFilesConfig):
         aws_request: Final = AWSRequest(  # any-ok: botocore AWSRequest is untyped
             method=method,
             url=api_base,
-            headers={"x-amz-content-sha256": empty_body_hash},
+            headers={"x-amz-content-sha256": empty_body_hash},  # mutable-ok: botocore AWSRequest takes a dict
         )
         auth: Final = S3SigV4Auth(credentials, "s3", aws_region_name)  # any-ok: botocore untyped
         auth.add_auth(aws_request)  # any-ok: botocore request mutation is untyped

litellm/llms/bedrock/messages/mantle_transformation.py

@@ -110,7 +110,7 @@ class AmazonMantleMessagesConfig(AmazonAnthropicClaudeMessagesConfig):
                 if value
             }
         )
-        return {
+        return {  # mutable-ok: the base class contract returns a dict the handler signs into in place
             **merged_headers,
             **mantle_headers,
         }, resolved_api_base
@@ -141,7 +141,7 @@ class AmazonMantleMessagesConfig(AmazonAnthropicClaudeMessagesConfig):
         mantle_fields: Final = MappingProxyType(
             {key: value for key, value in (("model", model_id), ("stream", streaming)) if value}
         )
-        return {
+        return {  # mutable-ok: the base class contract returns the dict the handler serializes as the body
             **body,
             **mantle_fields,
         }

litellm/llms/bedrock/realtime/handler.py

@@ -395,7 +395,7 @@ class BedrockRealtime(BaseAWSLLM):
         if logged_events:
             GLOBAL_LOGGING_WORKER.ensure_initialized_and_enqueue(
                 logging_obj.dispatch_success_handlers(
-                    list(logged_events),
+                    list(logged_events),  # mutable-ok: realtime spend logging requires a list result
                     prefer_async_handlers=True,
                 )
             )

litellm/llms/bedrock/realtime/transformation.py

@@ -887,7 +887,7 @@ class BedrockRealtimeConfig(BaseRealtimeConfig):
                 id=f"resp_{uuid.uuid4()}",
                 status="completed",
                 conversation_id=f"conv_{uuid.uuid4()}",
-                usage=dict(usage),
+                usage=dict(usage),  # mutable-ok: OpenAIRealtimeResponseDoneObject types usage as plain dict
             ),
         )
         return (leftover_done,)

litellm/llms/bedrock/responses/transformation.py

@@ -119,24 +119,24 @@ def _inline_block(block: object, inlined: "Mapping[str, str]") -> object:
     url: Final = _remote_image_url(block)
     if url is None or not isinstance(block, dict):
         return block
-    return {**block, "image_url": inlined[url]}
+    return {**block, "image_url": inlined[url]}  # mutable-ok: outgoing JSON request item
 
 
 def _inline_value(value: object, inlined: "Mapping[str, str]") -> object:
     if isinstance(value, list):
-        return [_inline_block(block, inlined) for block in value]
+        return [_inline_block(block, inlined) for block in value]  # mutable-ok: outgoing JSON request item
     return _inline_block(value, inlined)
 
 
 def _inline_item(item: object, inlined: "Mapping[str, str]") -> object:
     if not isinstance(item, dict):
         return item
-    inlined_fields: Final = {
+    inlined_fields: Final = {  # mutable-ok: outgoing JSON request item
         key: _inline_value(item[key], inlined) for key in IMAGE_BLOCK_KEYS if isinstance(item.get(key), (list, dict))
     }
     if not inlined_fields:
         return item
-    return {**item, **inlined_fields}
+    return {**item, **inlined_fields}  # mutable-ok: same
 
 
 def inline_remote_image_urls(
@@ -145,7 +145,7 @@ def inline_remote_image_urls(
     """``input`` with every http(s) image URL replaced by its entry in ``inlined``."""
     if not isinstance(input, list) or not inlined:
         return input
-    items: Final = [_inline_item(item, inlined) for item in input]
+    items: Final = [_inline_item(item, inlined) for item in input]  # mutable-ok: downstream narrows on isinstance(list)
     return items  # pyright: ignore[reportReturnType]  # items keep the caller's input union
 
 
@@ -219,7 +219,7 @@ class BedrockOpenAIResponsesConfig(BaseAWSLLM, OpenAIResponsesAPIConfig):
         bearer: Final = resolve_bedrock_bearer_token(api_key)
         if not bearer:
             return headers
-        return {**headers, "Authorization": f"Bearer {bearer}"}
+        return {**headers, "Authorization": f"Bearer {bearer}"}  # mutable-ok: dict return per the contract
 
     def sign_request(
         self,
@@ -261,7 +261,9 @@ class BedrockOpenAIResponsesConfig(BaseAWSLLM, OpenAIResponsesAPIConfig):
                 "Bedrock Runtime Responses API: dropping unsupported parameter(s) %s that the endpoint rejects.",
                 unsupported,
             )
-        params: Final = {key: value for key, value in mapped.items() if key not in unsupported}
+        params: Final = {  # mutable-ok: outgoing JSON request params
+            key: value for key, value in mapped.items() if key not in unsupported
+        }
         tools: Final = params.get("tools")
         if not isinstance(tools, list):
             return params

litellm/llms/bedrock/search/transformation.py

@@ -178,7 +178,7 @@ class AgentCoreSearchConfig(BaseSearchConfig, BaseAWSLLM):
         Authentication itself happens in sign_request(): bearer token for
         CUSTOM_JWT gateways, AWS SigV4 for AWS_IAM gateways.
         """
-        return {
+        return {  # mutable-ok: httpx request headers are a dict
             **headers,
             "Content-Type": "application/json",
             "Accept": "application/json, text/event-stream",
@@ -234,13 +234,13 @@ class AgentCoreSearchConfig(BaseSearchConfig, BaseAWSLLM):
                 "Other gateway tools cannot be invoked through this provider."
             )
 
-        return {
+        return {  # mutable-ok: JSON-RPC request bodies are JSON objects
             "jsonrpc": "2.0",
             "id": 1,
             "method": "tools/call",
-            "params": {
+            "params": {  # mutable-ok: JSON-RPC request bodies are JSON objects
                 "name": tool_name,
-                "arguments": {
+                "arguments": {  # mutable-ok: JSON-RPC request bodies are JSON objects
                     "query": joined_query[:AGENTCORE_MAX_QUERY_LENGTH],
                     "maxResults": optional_params.get("max_results", AGENTCORE_DEFAULT_MAX_RESULTS),
                 },
@@ -286,7 +286,7 @@ class AgentCoreSearchConfig(BaseSearchConfig, BaseAWSLLM):
             default_api_base=api_base if gateway_host_match else None,
         )
         if bearer_token:
-            bearer_headers: Final = {
+            bearer_headers: Final = {  # mutable-ok: httpx request headers are a dict
                 **headers,
                 "Authorization": f"Bearer {bearer_token}",
             }
@@ -302,7 +302,7 @@ class AgentCoreSearchConfig(BaseSearchConfig, BaseAWSLLM):
         signing_params: Final = (
             optional_params
             if optional_params.get("aws_region_name") is not None
-            else {
+            else {  # mutable-ok: BaseAWSLLM._sign_request takes optional params as a dict
                 **optional_params,
                 "aws_region_name": self._signing_region(api_base),
             }
@@ -398,7 +398,7 @@ class AgentCoreSearchConfig(BaseSearchConfig, BaseAWSLLM):
         structured: Final = result.get("structuredContent") if isinstance(result, Mapping) else None
         items: Final = text_items or _result_items(structured)
 
-        results: Final = [_to_search_result(item) for item in items]
+        results: Final = [_to_search_result(item) for item in items]  # mutable-ok: pydantic list field
 
         return SearchResponse(results=results, object="search")
 

litellm/llms/bedrock_mantle/chat/transformation.py

@@ -117,7 +117,7 @@ class BedrockMantleChatConfig(BedrockMantleAuthMixin, OpenAILikeChatConfig):
             )
             if supported and param not in base_params
         )
-        return [*base_params, *extra_params]
+        return [*base_params, *extra_params]  # mutable-ok: fresh list required by the inherited signature
 
     def _supports_reasoning(self, model: str) -> bool:
         try:

litellm/llms/bedrock_mantle/responses/transformation.py

@@ -173,11 +173,15 @@ class BedrockMantleResponsesAPIConfig(BedrockMantleAuthMixin, OpenAIResponsesAPI
             summary,
             sorted(_BEDROCK_MANTLE_OPENAI_PATH_SUPPORTED_REASONING_SUMMARIES),
         )
-        stripped: Final = {key: value for key, value in reasoning.items() if key != "summary"}
+        stripped: Final = {  # mutable-ok: map_openai_params contract returns a plain dict
+            key: value for key, value in reasoning.items() if key != "summary"
+        }
         return (
-            {**params, "reasoning": stripped}
+            {**params, "reasoning": stripped}  # mutable-ok: map_openai_params contract returns a plain dict
             if stripped
-            else {key: value for key, value in params.items() if key != "reasoning"}
+            else {  # mutable-ok: map_openai_params contract returns a plain dict
+                key: value for key, value in params.items() if key != "reasoning"
+            }
         )
 
     def transform_responses_api_request(

litellm/llms/custom_httpx/llm_http_handler.py

@@ -363,7 +363,7 @@ def _mask_presigned_request_headers(transformed_request: bytes | str | dict) ->
         _get_masked_values,  # pyright: ignore[reportPrivateUsage]  # the shared header-masking helper has no public name
     )
 
-    return {
+    return {  # mutable-ok: logging's curl and raw-request builders take dict
         **transformed_request,
         "headers": _get_masked_values(request_headers),
     }
@@ -2559,7 +2559,7 @@ class BaseLLMHTTPHandler:
         )
 
         if self._has_agentic_completion_hook(logging_obj):
-            agentic_kwargs: Final = dict(litellm_params)
+            agentic_kwargs: Final = dict(litellm_params)  # mutable-ok: agentic hooks mutate kwargs in place
             final_response: Final = run_async_function(
                 self._call_agentic_completion_hooks,
                 response=initial_response,
@@ -2754,7 +2754,7 @@ class BaseLLMHTTPHandler:
             logging_obj=logging_obj,
         )
 
-        agentic_kwargs: Final = dict(litellm_params)
+        agentic_kwargs: Final = dict(litellm_params)  # mutable-ok: agentic hooks mutate kwargs in place
         final_response: Final = await self._call_agentic_completion_hooks(
             response=initial_response,
             model=model,
@@ -4735,7 +4735,9 @@ class BaseLLMHTTPHandler:
         files_per_page: Final = self._files_per_listing_page(
             response, provider_config, logging_obj, litellm_params, headers, sync_httpx_client, timeout
         )
-        return [listed_file for page_files in files_per_page for listed_file in page_files]
+        return [  # mutable-ok: the files contract returns the listing as a list
+            listed_file for page_files in files_per_page for listed_file in page_files
+        ]
 
     async def async_list_files(
         self,
@@ -4790,7 +4792,9 @@ class BaseLLMHTTPHandler:
         files_per_page: Final = self._files_per_async_listing_page(
             response, provider_config, logging_obj, litellm_params, headers, async_httpx_client, timeout
         )
-        return [listed_file async for page_files in files_per_page for listed_file in page_files]
+        return [  # mutable-ok: the files contract returns the listing as a list
+            listed_file async for page_files in files_per_page for listed_file in page_files
+        ]
 
     def _files_per_listing_page(
         self,
@@ -9700,7 +9704,7 @@ class BaseLLMHTTPHandler:
         logging_obj.pre_call(
             input="",
             api_key="",
-            additional_args={
+            additional_args={  # mutable-ok: pre_call's additional_args contract is a dict
                 "query": query,
                 "vector_store_id": vector_store_id,
                 "api_base": endpoint,
@@ -9736,7 +9740,7 @@ class BaseLLMHTTPHandler:
                 query=query,
                 vector_store_search_optional_params=vector_store_search_optional_params,
                 litellm_logging_obj=logging_obj,
-                litellm_params=dict(litellm_params),
+                litellm_params=dict(litellm_params),  # mutable-ok: snapshot GenericLiteLLMParams into the Mapping shape
                 embedding_executor=embedding_executor,
                 timeout=timeout,
             )
@@ -9876,7 +9880,7 @@ class BaseLLMHTTPHandler:
                 query=query,
                 vector_store_search_optional_params=vector_store_search_optional_params,
                 litellm_logging_obj=logging_obj,
-                litellm_params=dict(litellm_params),
+                litellm_params=dict(litellm_params),  # mutable-ok: snapshot GenericLiteLLMParams into the Mapping shape
                 embedding_executor=embedding_executor,
                 timeout=timeout,
             )

litellm/llms/dashscope/chat/transformation.py

@@ -13,7 +13,7 @@ from ...openai.chat.gpt_transformation import OpenAIGPTConfig
 
 class DashScopeChatConfig(OpenAIGPTConfig):
     def get_supported_openai_params(self, model: str) -> list[str]:  # mutable-ok: base class contract returns a list
-        return [
+        return [  # mutable-ok: base class contract returns a list
             *super().get_supported_openai_params(model=model),
             "reasoning_effort",
         ]

litellm/llms/deepseek/chat/transformation.py

@@ -129,7 +129,7 @@ class DeepSeekChatConfig(OpenAIGPTConfig):
         forward_images: Final = any(
             isinstance(message.get("content"), list) for message in messages
         ) and supports_vision(model=model, custom_llm_provider="deepseek")
-        transformed: Final = [
+        transformed: Final = [  # mutable-ok: provider messages must stay JSON-array lists the base transform mutates
             self._forward_or_collapse_content(message=message, forward_images=forward_images) for message in messages
         ]
 
@@ -155,7 +155,7 @@ class DeepSeekChatConfig(OpenAIGPTConfig):
         collapsed: Final = convert_content_list_to_str(message=message)
         if not collapsed or collapsed == content:
             return message
-        collapsed_message: Final = {**message, "content": collapsed}
+        collapsed_message: Final = {**message, "content": collapsed}  # mutable-ok: wire messages are plain JSON dicts
         return cast(AllMessageValues, collapsed_message)  # cast-ok: TypedDict spread narrows to dict
 
     def _is_vision_forwardable_content(self, message: AllMessageValues, content: Sequence[object]) -> bool:
@@ -204,8 +204,8 @@ class DeepSeekChatConfig(OpenAIGPTConfig):
         search_text: Final = extract_search_results_text(message_fields.get("search_results"))
         if not search_text:
             return message
-        forwarded_content: Final = [*content, {"type": "text", "text": search_text}]
-        forwarded: Final = {
+        forwarded_content: Final = [*content, {"type": "text", "text": search_text}]  # mutable-ok: JSON-array content
+        forwarded: Final = {  # mutable-ok: wire messages are plain JSON dicts
             **{key: value for key, value in message_fields.items() if key != "search_results"},
             "content": forwarded_content,
         }

litellm/llms/edenai/audio_transcription/transformation.py

@@ -28,7 +28,7 @@ def _form_fields(model: str, optional_params: Mapping[str, object]) -> dict[str,
     extras: Final = optional_params.get("extra_body")
     nested: Final = extras.items() if isinstance(extras, Mapping) else ()
     fields: Final = (*optional_params.items(), *nested, ("model", model))
-    return {key: value for key, value in fields if key != "extra_body"}
+    return {key: value for key, value in fields if key != "extra_body"}  # mutable-ok: httpx form data
 
 
 class EdenAIAudioTranscriptionConfig(OpenAIWhisperAudioTranscriptionConfig):
@@ -69,7 +69,7 @@ class EdenAIAudioTranscriptionConfig(OpenAIWhisperAudioTranscriptionConfig):
         """Eden reports `duration` and `cost` on every body, so the Whisper default of `verbose_json`,
         which the gpt-4o-transcribe models reject, is not needed for cost tracking."""
         audio: Final = process_audio_file(audio_file)
-        files: Final = {"file": (audio.filename, audio.file_content, audio.content_type)}
+        files: Final = {"file": (audio.filename, audio.file_content, audio.content_type)}  # mutable-ok: httpx contract
         return AudioTranscriptionRequestData(data=_form_fields(model, optional_params), files=files)
 
     def transform_audio_transcription_response(self, raw_response: httpx.Response) -> TranscriptionResponse: