scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-02 02:11:58 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 102

chore(lint): remove the LIT002 mutable-construction rule (#43971)

Browse source 
* chore(lint): remove the LIT002 mutable-construction rule

Drop LIT002 from scripts/check_type_discipline.py along with its helpers,
its budget entry, its unit tests, and the AGENTS.md and gate docstring
mentions. `# mutable-ok` now only suppresses LIT001, so the markers that
only existed to silence LIT002 became LIT013 stale suppressions and are
removed. The files whose layout depended on those trailing comments are
reformatted with ruff format.

Every other LIT rule count is unchanged and the ASTs of all touched
litellm/ files match main apart from one docstring.

* 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.

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

This reverts commit c35bc0b84e.

---------

Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
This commit is contained in:
yuneng-jiang 2026-10-01 12:24:02 -07:00 • committed by GitHub
parent fcf87972fd
commit bba85f0b6c
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
367 changed files with 1434 additions and 2285 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
AGENTS.md
View file

@ -62,7 +62,7 @@ Never edit or commit `ruff-strict-budget.json`, `type-discipline-budget.json`, `
If you're trying to create a new function that relies on untyped stuff, instead of adding more Any's and pushing `reportAny` / `reportExplicitAny` closer to their basedpyright ceilings, just validate it in the caller with Pydantic (a model or `TypeAdapter` that returns the typed thing or raises will do) and then pass the now typed variable in
If you get an LIT001 or LIT002 fail, refactor the code to follow functional programming best practices rather than introducing mutable data structures. For example, build values in one shot with comprehensions or generators wrapped in `tuple()` / `MappingProxyType()` / `frozenset()` instead of seeding an empty `list`/`dict`/`set` and mutating it over time. Ideally, `# mutable-ok` is never used; reach for it only as a genuine last resort when an immutable rewrite is truly impossible, and always pair it with a real reason
If you get an LIT001 fail, refactor the code to follow functional programming best practices rather than introducing mutable data structures. For example, build values in one shot with comprehensions or generators wrapped in `tuple()` / `MappingProxyType()` / `frozenset()` instead of seeding an empty `list`/`dict`/`set` and mutating it over time. Ideally, `# mutable-ok` is never used; reach for it only as a genuine last resort when an immutable rewrite is truly impossible, and always pair it with a real reason
Every lint or type suppression must name the exact rule inside brackets and carry a reason comment, e.g. `# pyright: ignore[reportArgumentType] # stubs lack async overload` or `# noqa: TID251 # <reason>`. `# type: ignore` is banned (LIT009): pyrightconfig.json sets `enableTypeIgnoreComments` to false, so it silently does nothing

2
enterprise/litellm_enterprise/enterprise_callbacks/secret_detection.py
View file

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

4
litellm/__init__.py
View file

@ -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() # mutable-ok: provider registry, filled from model_cost at import like every sibling provider
nadir_models: Set = set()
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() # mutable-ok: filled from the price map at import, like the sibling provider sets
edenai_models: Set = set()
volcengine_models: Set = set()
wandb_models: Set = set(WANDB_MODELS)
ovhcloud_models: Set = set()

16
litellm/_logging.py
View file

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

4
litellm/_redis.py
View file

@ -233,7 +233,7 @@ def _coerce_redis_kwargs_types(
"socket_keepalive": bool,
}
)
result: Final = dict(redis_kwargs) # mutable-ok: per-key try/except coercion below needs to drop individual keys
result: Final = dict(redis_kwargs)
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) # mutable-ok: the branches below mutate these kwargs
return dict(kept, credential_provider=credential_provider)
def get_redis_client(**env_overrides):

4
litellm/a2a_protocol/main.py
View file

@ -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 # mutable-ok: a2a-sdk's get_agent_card takes a dict
return {"headers": extra_headers} if extra_headers else None
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( # mutable-ok: Logging.litellm_params is declared as a dict
_litellm_params: Final = dict(
(*_a2a_cost_params(litellm_params).items(), *((key, value) for key, value in _request_context if value))
)

2
litellm/batches/batch_utils.py
View file

@ -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=[], # mutable-ok: no output file means no model was ever priced; BatchCostUsageResult.models requires list[str]
models=[],
successful_requests=0,
failed_requests=await count_error_file_failed_requests(
batch, custom_llm_provider=custom_llm_provider, litellm_params=litellm_params

4
litellm/caching/affinity_cache.py
View file

@ -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)), # mutable-ok: JSON serialization requires dict, not a generic Mapping
json.dumps(dict(pin_value)),
int(ttl_seconds),
*(
(json.dumps(tuple(dict(value) for value in eligible_values)),) # mutable-ok: JSON requires dict
(json.dumps(tuple(dict(value) for value in eligible_values)),)
if eligible_values is not None
else ()
),

4
litellm/caching/caching_handler.py
View file

@ -702,14 +702,14 @@ class LLMCachingHandler:
)
merged: Final = EmbeddingResponse(
model=cached.model,
data=[ # mutable-ok: EmbeddingResponse.data is a pydantic list field
data=[
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={ # mutable-ok: EmbeddingResponse._hidden_params is a mutable dict field
hidden_params={
**cached._hidden_params,
"cache_hit": True,
},

12
litellm/caching/dual_cache.py
View file

@ -252,9 +252,7 @@ class DualCache(BaseCache):
if value is not None:
self.in_memory_cache.set_cache(key, value, **self._backfill_kwargs(kwargs))
return list( # mutable-ok: public list contract
redis_result.get(key) if value is None else value for key, value in zip(keys, result)
)
return list(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
@ -329,8 +327,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 [], {} # mutable-ok: API contract returns an empty list and dictionary
key_list: Final = list(keys) # mutable-ok: batch_get_cache takes a list
return [], {}
key_list: Final = list(keys)
memory: Final = self.in_memory_cache
in_memory_result: Final = (
None
@ -386,7 +384,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), # mutable-ok: the shared batch read takes a list
list(keys),
local_only=False,
throttle_redis=False,
)
@ -627,7 +625,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) # mutable-ok: both increment pipelines take a list
operations: Final = list(increment_list)
if batch is None:
await self.async_increment_cache_pipeline(operations, parent_otel_span=parent_otel_span)
return

2
litellm/caching/evicted_client_closer.py
View file

@ -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()) # mutable-ok: FIFO by design
bucket: Final = self._buckets.setdefault(_bucket_key(pending), deque())
while bucket and bucket[0].client_ref() is None:
bucket.popleft()
self._pending_count -= 1

4
litellm/caching/redis_batch.py
View file

@ -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] # mutable-ok: Callable params
SettledHook = Callable[[asyncio.Future[_T]], Awaitable[None] | None]
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 # mutable-ok: the cache API takes a list
found: Mapping[str, object] = await self._redis_cache.async_batch_get_cache(key_list=list(self._keys)) # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType] # untyped cache API
if any(key not in found for key in self._keys):
raise ConnectionError("batch get did not return every key")
return found

16
litellm/completion_extras/litellm_responses_transformation/transformation.py
View file

@ -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] # mutable-ok: API message payload
stored: Final = [_reasoning_item_to_response_input(item) for item in items]
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] # mutable-ok: API message payload
return [dict(item) for item in replayed]
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 {}, # mutable-ok: one-shot merge seed
dict(responses_api_request).get("text") or {},
)
return cast( # cast-ok: merged mapping is a valid ResponseText shape
"ResponseText",
{**existing, **update}, # mutable-ok: one-shot merged payload
{**existing, **update},
)
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=[ # mutable-ok: ModelResponseStream coerces only list choices
choices=[
StreamingChoices(
index=0,
delta=Delta(
@ -1612,7 +1612,7 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
)
],
usage=usage,
provider_specific_fields=dict(provider_metadata) or None, # mutable-ok: field is typed dict
provider_specific_fields=dict(provider_metadata) or None,
**(
MappingProxyType({"service_tier": served_service_tier})
if isinstance(served_service_tier, str)

4
litellm/cost_calculator.py
View file

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

4
litellm/experimental_mcp_client/client.py
View file

@ -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=[]) # mutable-ok: MCP result payload
return ListResourceTemplatesResult(resource_templates=[])
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=[]) # mutable-ok: MCP result payload
return ListResourceTemplatesResult(resource_templates=[])
try:
result: Final = await self.run_with_session(_list_resource_templates_operation)

4
litellm/experimental_mcp_client/tools.py
View file

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

8
litellm/integrations/SlackAlerting/slack_alerting.py
View file

@ -1131,7 +1131,7 @@ Model Info:
message=message,
level=level,
alert_type=AlertType.model_deprecation_warnings,
alerting_metadata={ # mutable-ok: send_alert takes a dict payload
alerting_metadata={
"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}, # mutable-ok: prisma find_many requires a dict where filter
order={"created_at": "desc"}, # mutable-ok: prisma find_many requires a dict order arg
where={"user_id": recipient_user_id},
order={"created_at": "desc"},
),
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={}, # mutable-ok: send_alert takes a dict payload
alerting_metadata={},
)
for event in typed_events:
await self.internal_usage_cache.async_set_cache(

6
litellm/integrations/azure_sentinel/azure_sentinel.py
View file

@ -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 = [] # mutable-ok: queue ownership is detached before the async send
self.log_queue = []
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 = [] # mutable-ok: queue ownership is detached before the async send
self.audit_log_queue = []
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] # mutable-ok: queue trimming returns a mutable logger queue
merged: Final = [*undelivered, *queue]
overflow: Final = len(merged) - self.max_queue_size
if overflow <= 0:
return merged

8
litellm/integrations/clickhouse/clickhouse_spend_logger.py
View file

@ -58,7 +58,7 @@ def _json(value: object) -> str:
def _json_mapping(value: Mapping[str, Any]) -> str:
return _json(dict(value)) # mutable-ok: [LIT002] JSON serialization requires a dict
return _json(dict(value))
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 [] # mutable-ok: [LIT002] empty spend-log tag payload
return [str(tag) for tag in value] # mutable-ok: [LIT002] SpendLogRecord schema
return []
return [str(tag) for tag in value]
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)]) # mutable-ok: [LIT002] batch logger API
self.enqueue([dict(row)])
except Exception as e:
verbose_logger.exception("ClickHouseSpendLogger: failed to log request: %s", e)

10
litellm/integrations/custom_guardrail.py
View file

@ -372,7 +372,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} # mutable-ok: plain dict for live request
advisory_message: Final = {"role": "system", "content": message}
existing_messages: Final = data.get("messages")
existing_input: Final = data.get("input")
existing_instructions: Final = data.get("instructions")
@ -383,7 +383,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 = [ # mutable-ok: fresh list
messages_with_instructions_note: Final = [
*existing_messages,
advisory_message,
]
@ -395,7 +395,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] # mutable-ok: fresh list
messages_with_input_note: Final = [*existing_messages, advisory_message]
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.
@ -409,10 +409,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] # mutable-ok: fresh list
messages_without_input_note: Final = [*existing_messages, advisory_message]
data["messages"] = messages_without_input_note # rebind-ok: mutates caller's dict by design
return True
sole_message: Final = [advisory_message] # mutable-ok: plain list for the live JSON request
sole_message: Final = [advisory_message]
data["messages"] = sole_message # rebind-ok: mutates caller's dict by design
return True

4
litellm/integrations/datadog/datadog.py
View file

@ -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 # mutable-ok: logger queue remains appendable
self.log_queue = list(cancelled.undelivered) + self.log_queue
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) # mutable-ok: caller prepends records to the logger queue
return list(undelivered)
@staticmethod
def _exceeds_intake_limits(chunk: Sequence[DatadogPayload]) -> bool:

2
litellm/integrations/datadog/datadog_llm_obs.py
View file

@ -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 { # mutable-ok: a fresh record built per entry, handed straight to the span serializer
return {
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

6
litellm/integrations/langfuse/langfuse.py
View file

@ -868,10 +868,8 @@ 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} # 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
"cost_details": {"total": cost} if usage is not None and isinstance(cost, (int, float)) else None,
"metadata": {
**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),

6
litellm/integrations/newrelic/newrelic_metrics.py
View file

@ -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]] = { # mutable-ok: JSON leaf; safe_dumps stringifies MappingProxyType
attributes: Final[Mapping[str, str]] = {
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]] = { # mutable-ok: JSON leaf; safe_dumps stringifies MappingProxyType
attributes: Final[Mapping[str, str]] = {
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) # mutable-ok: leave late arrivals for the next serialized drain
self.log_queue[:] = list(survivors)
async def _drain_flush_once(self) -> None:
"""Attempt every queued record once, in ``batch_size`` chunks, without

2
litellm/integrations/otel/model/metadata.py
View file

@ -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]) # mutable-ok: iterative worklist keeps the walk off the call stack
stack: Final = list(tuple(raw.items())[::-1])
while stack:
key, value = stack.pop()
if (nested := as_str_mapping(value)) is not None:

2
litellm/integrations/otel/model/payloads.py
View file

@ -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"} # mutable-ok: mappers json.dumps and isinstance(dict) it
return {k: v for k, v in merged if k != "text"}
def _completion_choices(response: Mapping[str, object]) -> tuple[Mapping[str, object], ...]:

2
litellm/integrations/otel/model/request_io.py
View file

@ -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), # mutable-ok: stream_chunk_builder takes a list
chunks=list(chunks),
messages=_MESSAGES.validate_python(data.get("messages")),
)
except (litellm.APIError, ValidationError):

6
litellm/integrations/otel/plumbing/context.py
View file

@ -380,10 +380,8 @@ 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) # 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
}
return dict(headers)
carrier: Final = {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

4
litellm/integrations/otel/plumbing/providers.py
View file

@ -636,9 +636,7 @@ 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( # mutable-ok: the same bounded map, keeping only what is still exporting
(ident, p) for ident, p in live if ident in self._exporting
)
self._retired = OrderedDict((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()))

6
litellm/integrations/otel/plumbing/routing.py
View file

@ -171,9 +171,7 @@ 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() # mutable-ok: bounded LRU; eviction needs in-place ordered mutation
)
self._providers: OrderedDict[_RouteKey, TracerProvider] = OrderedDict()
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
@ -393,7 +391,7 @@ class TenantTracerCache:
if project_headers and kind not in _GRPC_KINDS
else base
)
update: Final = { # mutable-ok: model_copy(update=...) requires a plain dict
update: Final = {
field: value
for field, value in (("headers", routed), ("endpoint", endpoint))
if (field == "headers" and routed != spec.headers)

2
litellm/integrations/otel/presets/langfuse.py
View file

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

4
litellm/integrations/otel/presets/signoz.py
View file

@ -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 {} # mutable-ok: same registry contract
return {"signoz-ingestion-key": key} # mutable-ok: same registry contract
return {}
return {"signoz-ingestion-key": key}

2
litellm/integrations/otel/presets/weave.py
View file

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

4
litellm/integrations/pointfive/logger.py
View file

@ -207,9 +207,7 @@ 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) # mutable-ok: both framework helpers take the call details as a dict
)
details: Final = self.redact_standard_logging_payload_from_model_call_details(dict(kwargs))
payload: Final = details.get("standard_logging_object")
if not isinstance(payload, dict):
return None

4
litellm/integrations/pointfive/upload_client.py
View file

@ -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={ # mutable-ok: AsyncHTTPHandler.post types headers as dict
headers={
"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) # mutable-ok: put wants dict
headers: Final = dict(PUT_HEADERS, Host=host) if host else dict(PUT_HEADERS)
try:
await self.http_client.put(url, data=body, headers=headers, follow_redirects=False)
except httpx.HTTPStatusError as e:

2
litellm/integrations/prometheus.py
View file

@ -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} # mutable-ok: ** needs a mapping to override labelnames
kept_kwargs: Final = {**kwargs, "labelnames": kept}
real_metric: Final = metric_class(*args, **kept_kwargs)
return _ExcludedLabelMetric(real_metric, original_labelnames, self.exclude_labels)

4
litellm/integrations/s3_v2.py
View file

@ -642,7 +642,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 = {} # mutable-ok: per-flush drop marks read back by _upload_bounded
self._flush_dropped = {}
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))
@ -694,7 +694,7 @@ class S3Logger(CustomBatchLogger, BaseAWSLLM):
self.max_queue_size,
overflow,
)
self.log_queue = [ # mutable-ok: log_queue is the flush buffer shared with custom_batch_logger
self.log_queue = [
*requeued,
*arrivals,
][overflow:]

34
litellm/integrations/shadow_eval_logger.py
View file

@ -357,11 +357,7 @@ class GuardrailRequestSnapshot:
if fingerprint is None:
return None
return GuardrailRequestSnapshot(
body=MappingProxyType(
_CHAT_REQUEST_ADAPTER.validate_python(
independent_snapshot(dict(body)) # mutable-ok: snapshot helper requires a plain dictionary
)
),
body=MappingProxyType(_CHAT_REQUEST_ADAPTER.validate_python(independent_snapshot(dict(body)))),
fingerprint=fingerprint,
)
@ -813,7 +809,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}) # mutable-ok: pydantic update payload
return job.model_copy(update={"attempts": attempts, "spend": spend})
_jobs_cache: Final = InMemoryCache(max_size_in_memory=4, default_ttl=_JOBS_CACHE_TTL_SECONDS)
@ -864,9 +860,9 @@ class ShadowEvalLogger(CustomLogger):
return _EMPTY_JOBS
try:
records: Final = await prisma.db.litellm_shadowevaljob.find_many(
where={ # mutable-ok: Prisma filter
where={
"stopped_at": None,
"ends_at": {"gt": datetime.now(timezone.utc)}, # mutable-ok: Prisma filter
"ends_at": {"gt": datetime.now(timezone.utc)},
},
)
grouped: Final = (
@ -874,12 +870,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]}}, # mutable-ok: Prisma filter
where={"job_id": {"in": [str(record.id) for record in records]}},
)
if records
else ()
)
attempt_stats: Final = { # mutable-ok: frozen snapshot of the grouped read
attempt_stats: Final = {
str(row["job_id"]): (
int(row["_count"]["_all"]),
_leg_eval_spend(row["_sum"] or _EMPTY_METADATA),
@ -952,13 +948,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)) # mutable-ok: helper needs dict
raw_meta: Final = get_litellm_metadata_from_kwargs(dict(kwargs))
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)): # mutable-ok: predicate takes a plain dict
if should_redact_message_logging(dict(kwargs)):
return
metadata: Final = payload.get("metadata") or _EMPTY_METADATA
# Each identity the request resolved to is a candidate target; JWT-auth
@ -999,7 +995,7 @@ class ShadowEvalLogger(CustomLogger):
sample: Final = _judgeable_sample(
ops,
sample_kwargs,
MappingProxyType(dict(payload.get("model_parameters") or {})), # mutable-ok: frozen snapshot
MappingProxyType(dict(payload.get("model_parameters") or {})),
response_obj,
)
if sample is None:
@ -1246,7 +1242,7 @@ class ShadowEvalLogger(CustomLogger):
return
try:
await prisma.db.litellm_shadowevalattempt.create(
data={ # mutable-ok: Prisma payload
data={
"job_id": job.id,
"request_id": request_id,
"router_name": router_name,
@ -1287,12 +1283,10 @@ class ShadowEvalLogger(CustomLogger):
try:
response: Final = await router.acompletion(
model=target_model,
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
messages=[dict(m) for m in messages], # pyright: ignore[reportArgumentType] # snapshot of the SDK's own message dicts
metadata=shadow_metadata,
num_retries=0,
fallbacks=[], # mutable-ok: SDK kwarg; a failed shadow is a recorded error, never a spend multiplier
fallbacks=[],
**shadow_params,
)
except Exception as e: # noqa: BLE001 # provider errors become error rows, not crashes
@ -1341,8 +1335,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 = [ # mutable-ok: SDK takes a list
{"role": "system", "content": PAIRWISE_JUDGE_SYSTEM_PROMPT}, # mutable-ok: SDK message
judge_messages: Final = [
{"role": "system", "content": PAIRWISE_JUDGE_SYSTEM_PROMPT},
{
"role": "user",
"content": _judge_user_prompt(conversation, response_a, response_b, _tool_definitions_text(tools)),

2
litellm/integrations/websearch_interception/handler.py
View file

@ -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 { # mutable-ok: litellm's metadata channel is a plain dict its logging path reads and enriches
return {
**user_api_key_metadata,
**parent_correlation.as_search_metadata(),
"model_group": search_tool_name,

4
litellm/integrations/zerobus/logger.py
View file

@ -188,9 +188,7 @@ 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) # mutable-ok: both framework helpers take the call details as a dict
)
details: Final = self.redact_standard_logging_payload_from_model_call_details(dict(kwargs))
payload: Final = details.get("standard_logging_object")
if not isinstance(payload, dict):
return None

2
litellm/litellm_core_utils/agentic_followup_kwargs.py
View file

@ -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] # mutable-ok: the chat loop's settings reader only accepts a list
seen: Final = [*fingerprints, fingerprint]
return MappingProxyType(
{
key: value

2
litellm/litellm_core_utils/chat_completion_agentic_loop.py
View file

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

8
litellm/litellm_core_utils/core_helpers.py
View file

@ -586,7 +586,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()
}
@ -608,15 +608,13 @@ def independent_snapshot(
and isinstance(original_value, dict)
and "litellm_parent_otel_span" in original_value
):
return { # mutable-ok: same request-payload shape as data
return {
**copied_value,
"litellm_parent_otel_span": original_value["litellm_parent_otel_span"],
}
return copied_value
return { # mutable-ok: same request-payload shape as data
key: _copied_value(key, value) for key, value in sanitized.items()
}
return {key: _copied_value(key, value) for key, value in sanitized.items()}
def filter_exceptions_from_params(data: object, max_depth: int = 20) -> Any:

8
litellm/litellm_core_utils/get_litellm_params.py
View file

@ -99,9 +99,7 @@ class InvalidControlOption:
def parse_control_options(kwargs: Mapping[str, object]) -> ControlOptions | InvalidControlOption:
given: Final = { # mutable-ok: TypeAdapter.validate_python takes a dict
name: kwargs[name] for name in _CONTROL_OPTION_NAMES if name in kwargs
}
given: Final = {name: kwargs[name] for name in _CONTROL_OPTION_NAMES if name in kwargs}
try:
return _CONTROL_OPTIONS.validate_python(given)
except ValidationError as e:
@ -118,8 +116,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) # 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
return dict(litellm_params)
return {**litellm_params, CONTROL_OPTIONS_KEY: control}
def _get_base_model_from_litellm_call_metadata(

2
litellm/litellm_core_utils/get_model_cost_map.py
View file

@ -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={ # mutable-ok: threading requires a mutable keyword-arguments mapping
kwargs={
"url": url,
"timeout": timeout,
"max_attempts": max_attempts,

14
litellm/litellm_core_utils/internal_call_metadata.py
View file

@ -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"} # mutable-ok: SDK metadata value
return {k: v for k, v in auth.items() if k != "budget_reservation"}
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}) # mutable-ok: pydantic update payload
return model_copy(update={"budget_reservation": None})
return auth
def _sanitized(parent_metadata: Mapping[str, object]) -> dict[str, object]: # mutable-ok: SDK metadata kwarg
return { # mutable-ok: SDK metadata kwarg
return {
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,10 +138,8 @@ def forwarded_internal_call_metadata(
parent's full context still describes the call being made.
"""
if not parent_metadata:
return {} # mutable-ok: SDK metadata kwarg
return _sanitized(parent_metadata) | { # mutable-ok: SDK metadata kwarg
INTERNAL_CALL_ORIGIN_METADATA_KEY: call_origin
}
return {}
return _sanitized(parent_metadata) | {INTERNAL_CALL_ORIGIN_METADATA_KEY: call_origin}
def parent_session_kwargs(request_kwargs: Mapping[str, object] | None) -> Mapping[str, str]:
@ -167,4 +165,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} # mutable-ok: SDK metadata kwarg
return _sanitized(identity) | {INTERNAL_CALL_ORIGIN_METADATA_KEY: call_origin}

4
litellm/litellm_core_utils/json_fragment_accumulator.py
View file

@ -50,7 +50,7 @@ class JSONFragmentAccumulator:
unconsumed: Final = self._buffer[self._offset :]
self._buffer = unconsumed + "".join(self._chunks)
self._offset = 0
self._chunks = [] # mutable-ok: see __init__
self._chunks = []
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 = [] # mutable-ok: see __init__
self._chunks = []
self._buffer = value
self._offset = 0
stripped: Final = value.rstrip()

20
litellm/litellm_core_utils/litellm_logging.py
View file

@ -697,7 +697,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] = {} # mutable-ok: kept deep-copyable
self.response_timing_metrics: Mapping[str, float] = {}
# Passthrough endpoint guardrails config for field targeting
self.passthrough_guardrails_config: dict[str, object] | None = None
@ -721,7 +721,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) # mutable-ok: kept deep-copyable
self.response_timing_metrics = dict(timing_metrics)
def add_dynamic_callback(self, callback: CustomLogger) -> None:
self.dynamic_input_callbacks = self._with_dynamic_callback(self.dynamic_input_callbacks, callback)
@ -4144,7 +4144,7 @@ class Logging(LiteLLMLoggingBaseClass):
if result.status == "completed":
return InteractionsAPIResponse.model_validate(
result.model_dump(
exclude={ # mutable-ok: pydantic types exclude as set[str], which a frozenset does not satisfy
exclude={
"event_type",
"delta",
"index",
@ -5247,9 +5247,7 @@ 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)]} # mutable-ok: model_copy update
)
return config.model_copy(update={"exporters": [spec for spec in config.exporters if _is_gated(spec)]})
def _is_gated(spec: "ExporterSpec") -> bool:
@ -5721,7 +5719,7 @@ class StandardLoggingPayloadSetup:
if key not in user_metadata
}
)
return {**user_metadata, **model_metadata} # mutable-ok: function contract returns a plain dict
return {**user_metadata, **model_metadata}
@staticmethod
def get_standard_logging_metadata(
@ -6583,9 +6581,7 @@ 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 {} # mutable-ok: empty fallback
)
timing_metrics: Final = getattr(logging_obj, "response_timing_metrics", None) or {}
clean_hidden_params["litellm_overhead_time_ms"] = timing_metrics.get("litellm_overhead_time_ms")
model_cost_information: Final = StandardLoggingPayloadSetup.get_model_cost_information(
@ -6700,14 +6696,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",

2
litellm/litellm_core_utils/llm_cost_calc/guardrail_cost.py
View file

@ -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 { # mutable-ok: stamped into guardrail_information, which safe_dumps only serializes as a plain dict
return {
counter: _priced_units(units, pricing.guardrail_cost_per_unit.get(counter))
for counter, units in usage_units.items()
}

2
litellm/litellm_core_utils/llm_judge.py
View file

@ -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()} # mutable-ok: plain parsed-JSON payload
return {str(k): v for k, v in parsed.items()}
def extract_text_from_content(content: object) -> str:

8
litellm/litellm_core_utils/llm_request_utils.py
View file

@ -16,9 +16,7 @@ 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]]
] = [ # mutable-ok: depth-capped stack walks nested JSON into multipart names
(key, value, 0)
]
] = [(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()
@ -48,9 +46,7 @@ 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]]
] = [ # mutable-ok: depth-capped stack walks nested JSON into multipart names
(key, value, 0)
]
] = [(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()

2
litellm/litellm_core_utils/llm_response_utils/response_metadata.py
View file

@ -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} # mutable-ok: read-only timing result
return {"_response_ms": total_response_time_ms}
caching_details: Final = logging_obj.caching_details
cache_duration_ms: Final = (
caching_details.get("cache_duration_ms")

2
litellm/litellm_core_utils/logging_utils.py
View file

@ -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 { # mutable-ok: loggers isinstance-check the request body as a dict
return {
"model": model,
"voice": voice,
**{key: value for key, value in optional_params.items() if key != "extra_headers"},

52
litellm/litellm_core_utils/prompt_templates/common_utils.py
View file

@ -1274,7 +1274,7 @@ def _flatten_schema_against_root(
if not is_object_schema:
return schema
merged_properties: Final = { # mutable-ok: tool parameters are JSON dicts
merged_properties: Final = {
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 { # mutable-ok: tool parameters are JSON dicts
return {
**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, {}) # mutable-ok: fresh per-call $ref memo
return _flatten_schema_against_root(schema, schema, frozenset(), 0, {})
_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 = { # mutable-ok: tool parameters are JSON dicts
kept: Final = {
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 = { # mutable-ok: tool parameters are JSON dicts
kept: Final = {
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] # mutable-ok: tool parameters are JSON lists
items: Final = [rebuilt.get(id(sub), sub) for sub in value]
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}} # mutable-ok: request tools are JSON dicts
return {**tool, "function": {**function, "parameters": sanitized}}
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} # mutable-ok: API message payload
marked: Final = {**target, "prompt_cache_breakpoint": marker}
return cast(_MarkedT, marked) # cast-ok: same block shape as the input plus the marker key
@ -1703,9 +1703,7 @@ def strip_litellm_internal_message_fields(message: AllMessageValues) -> AllMessa
return message
return cast( # cast-ok: same TypedDict minus internal keys
AllMessageValues,
{ # 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
},
{key: value for key, value in message.items() if key not in LITELLM_INTERNAL_MESSAGE_FIELDS},
)
@ -2194,11 +2192,9 @@ def _split_images_from_tool_message(
)
if not image_parts:
return message, ()
remaining_parts = [ # mutable-ok: tool message content must stay a json list
part for part in content if not _is_image_url_part(part)
]
remaining_parts = [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} # mutable-ok: chat messages are plain json dicts
rewritten = {**message, "content": new_content}
return cast(AllMessageValues, rewritten), image_parts # cast-ok: dict spread keeps keys like cache_control
@ -2206,14 +2202,12 @@ 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 = [ # 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
hoisted_images = [image for _, images in split_results for image in images]
rewritten_messages = [message for message, _ in split_results]
if not hoisted_images:
return rewritten_messages
boundary_part = ChatCompletionTextObject(type="text", text=TOOL_RESULT_IMAGE_BOUNDARY)
hoisted_content = [boundary_part, *hoisted_images] # mutable-ok: user message content must be a json list
hoisted_content = [boundary_part, *hoisted_images]
rewritten_messages.append(ChatCompletionUserMessage(role="user", content=hoisted_content))
return rewritten_messages
@ -2237,7 +2231,7 @@ def hoist_images_from_tool_messages(
"""
if not any(_tool_message_carries_image(message) for message in messages):
return messages
return [ # mutable-ok: pipelines mutate message lists
return [
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)
@ -2259,11 +2253,9 @@ 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 = [ # mutable-ok: tool message content must stay a json list
part for part in content if not _is_tool_reference_part(part)
]
remaining_parts = [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} # mutable-ok: chat messages are plain json dicts
rewritten = {**message, "content": new_content}
return cast(AllMessageValues, rewritten) # cast-ok: dict spread keeps keys like cache_control
@ -2281,7 +2273,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] # mutable-ok: pipelines mutate message lists
return [_drop_tool_reference_parts(message) for message in messages]
INSTRUCTION_MESSAGE_ROLES: Final = frozenset({"system", "developer"})
@ -2294,7 +2286,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 [ # mutable-ok: pipelines mutate message lists
return [
*(message for message in messages if _is_instruction_message(message)),
*(message for message in messages if not _is_instruction_message(message)),
]
@ -2315,16 +2307,14 @@ 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 = [ # mutable-ok: chat message content must stay a json list
part for content in contents for part in _system_content_as_text_parts(content)
]
merged_parts: Final = [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 [ # mutable-ok: pipelines mutate message lists
return [
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)

2
litellm/litellm_core_utils/prompt_templates/factory.py
View file

@ -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] = [] # mutable-ok: accumulator behind the mutable return contract
new_messages: Final[_AnthropicMessageList] = []
if len(messages) == 0:
if not litellm.modify_params:

24
litellm/litellm_core_utils/prompt_templates/image_handling.py
View file

@ -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 # mutable-ok: json-serialized part
return {**image_url, "url": data_url} if image_url is not None else data_url
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"} # mutable-ok: json-serialized message part
return {**kept, **_inferred_format(file, url), "file_data": data_url} # mutable-ok: json-serialized part
kept: Final = {k: v for k, v in file.items() if k != "file_id"}
return {**kept, **_inferred_format(file, url), "file_data": data_url}
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} # mutable-ok: json-serialized message part
return {"type": "base64", "media_type": media_type, "data": data}
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)} # mutable-ok: json-serialized part
return {**part, "image_url": _inlined_image_url(image_url, data_url)}
case _RemoteFile(part, file, url):
return {**part, "file": _inlined_file(file, url, data_url)} # mutable-ok: json-serialized message part
return {**part, "file": _inlined_file(file, url, data_url)}
case _RemoteSource(part, _, url):
return {**part, "source": _base64_source(url, data_url)} # mutable-ok: json-serialized message part
return {**part, "source": _base64_source(url, data_url)}
def _content_parts(message: Mapping[str, object]) -> tuple[object, ...]:
@ -286,10 +286,8 @@ def _inline_message(
parts: Final = _content_parts(message)
if not parts:
return message
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
inlined_parts: Final = [_inline_part(part, data_urls, should_inline) for part in parts]
inlined_message: Final = {**message, "content": inlined_parts}
return inlined_message # pyright: ignore[reportReturnType] # the same message with its remote parts inlined
@ -326,6 +324,4 @@ 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 [ # mutable-ok: transform_request takes a list
_inline_message(message, inlined, should_inline) for message in messages
]
return [_inline_message(message, inlined, should_inline) for message in messages]

2
litellm/litellm_core_utils/prompt_templates/mid_conversation_system.py
View file

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

6
litellm/litellm_core_utils/provider_affinity.py
View file

@ -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) # mutable-ok: downstream handlers add auth and signing headers
return dict(headers)
session_id: Final = get_stable_session_id(litellm_params)
if session_id is None:
return dict(headers) # mutable-ok: downstream handlers add auth and signing headers
return dict(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} # mutable-ok: downstream handlers add auth and signing headers
return {**headers, header_name: session_id}

8
litellm/litellm_core_utils/sentry_scrubbing.py
View file

@ -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 { # mutable-ok: JSON object
return {
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] # mutable-ok: JSON array
return [scrub_json_strings(item, scrub, (*path, "*")) for item in value]
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), # mutable-ok: EventScrubber appends pii_denylist onto denylist in place
pii_denylist=list(PII_FIELD_NAMES), # mutable-ok: EventScrubber takes List[str]
denylist=list(SECRET_FIELD_NAMES),
pii_denylist=list(PII_FIELD_NAMES),
recursive=True,
send_default_pii=send_default_pii,
),

4
litellm/litellm_core_utils/streaming_chunk_builder_utils.py
View file

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

4
litellm/litellm_core_utils/streaming_handler.py
View file

@ -189,9 +189,7 @@ 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) # mutable-ok: stream assembly merges provider metadata into this dict
if parsed is not None and parsed.provider_specific_fields
else None
dict(parsed.provider_specific_fields) if parsed is not None and parsed.provider_specific_fields else None
)
params: Final[Mapping[str, object]] = MappingProxyType(
{

66
litellm/litellm_core_utils/tokenizer.py
View file

@ -72,7 +72,7 @@ class OpenAIEncoding:
return self._special_tokens["<|endoftext|>"]
@property
def special_tokens_set(self) -> set[str]: # mutable-ok: [LIT001, LIT002] SDK return type
def special_tokens_set(self) -> set[str]: # mutable-ok: [LIT001] 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, LIT002] SDK return type
def encode_ordinary(self, text: str) -> list[int]: # mutable-ok: [LIT001] 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, LIT002] SDK return type
) -> list[int]: # mutable-ok: [LIT001] SDK return type
allowed: Final = self._allowed(text, allowed_special, disallowed_special)
if not allowed:
return self.encode_ordinary(text)
@ -111,11 +111,9 @@ class OpenAIEncoding:
def encode_ordinary_batch(
self, text: Sequence[str], *, num_threads: int = 8
) -> list[list[int]]: # mutable-ok: [LIT001, LIT002] SDK return type
) -> list[list[int]]: # mutable-ok: [LIT001] SDK return type
with ThreadPoolExecutor(num_threads) as executor:
return list( # mutable-ok: [LIT002] SDK returns a list
executor.map(self.encode_ordinary, text)
)
return list(executor.map(self.encode_ordinary, text))
def encode_batch(
self,
@ -124,12 +122,10 @@ class OpenAIEncoding:
num_threads: int = 8,
allowed_special: AllowedSpecial = frozenset(),
disallowed_special: SpecialTokens = "all",
) -> list[list[int]]: # mutable-ok: [LIT001, LIT002] SDK return type
) -> list[list[int]]: # mutable-ok: [LIT001] SDK return type
encode: Final = partial(self.encode, allowed_special=allowed_special, disallowed_special=disallowed_special)
with ThreadPoolExecutor(num_threads) as executor:
return list( # mutable-ok: [LIT002] SDK returns a list
executor.map(encode, text)
)
return list(executor.map(encode, text))
def encode_with_unstable(
self,
@ -137,7 +133,7 @@ class OpenAIEncoding:
*,
allowed_special: AllowedSpecial = frozenset(),
disallowed_special: SpecialTokens = "all",
) -> tuple[list[int], list[list[int]]]: # mutable-ok: [LIT001, LIT002] SDK return type
) -> tuple[list[int], list[list[int]]]: # mutable-ok: [LIT001] 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."""
@ -164,14 +160,12 @@ 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, LIT002] SDK return type
return [ # mutable-ok: [LIT002] SDK returns a list
self.decode_single_token_bytes(token) for token in tokens
]
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_with_offsets(
self, tokens: Sequence[int]
) -> tuple[str, list[int]]: # mutable-ok: [LIT001, LIT002] SDK return type
) -> tuple[str, list[int]]: # mutable-ok: [LIT001] 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."""
@ -185,21 +179,17 @@ class OpenAIEncoding:
def decode_batch(
self, batch: Sequence[Sequence[int]], *, errors: str = "replace", num_threads: int = 8
) -> list[str]: # mutable-ok: [LIT001, LIT002] SDK return type
) -> list[str]: # mutable-ok: [LIT001] SDK return type
with ThreadPoolExecutor(num_threads) as executor:
return list( # mutable-ok: [LIT002] SDK returns a list
executor.map(partial(self.decode, errors=errors), batch)
)
return 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, LIT002] SDK return type
) -> list[bytes]: # mutable-ok: [LIT001] SDK return type
with ThreadPoolExecutor(num_threads) as executor:
return list( # mutable-ok: [LIT002] SDK returns a list
executor.map(self.decode_bytes, batch)
)
return list(executor.map(self.decode_bytes, batch))
def token_byte_values(self) -> list[bytes]: # mutable-ok: [LIT001, LIT002] SDK return type
def token_byte_values(self) -> list[bytes]: # mutable-ok: [LIT001] SDK return type
return self._native.token_byte_values()
def __reduce__(self) -> tuple[Callable[[str], OpenAIEncoding], tuple[str]]:
@ -273,16 +263,14 @@ 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, LIT002] SDK return type
def get_vocab(self, with_added_tokens: bool = True) -> dict[str, int]: # mutable-ok: [LIT001] 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, LIT002] SDK return type
return { # mutable-ok: [LIT002] SDK returns a dict
def get_added_tokens_decoder(self) -> dict[int, AddedToken]: # mutable-ok: [LIT001] SDK return type
return {
token_id: AddedToken(
content, single_word=single_word, lstrip=lstrip, rstrip=rstrip, normalized=normalized, special=special
)
@ -300,11 +288,11 @@ class HuggingFaceTokenizer:
return self._native.num_special_tokens_to_add(is_pair)
@property
def padding(self) -> dict[str, object] | None: # mutable-ok: [LIT001, LIT002] SDK return type
def padding(self) -> dict[str, object] | None: # mutable-ok: [LIT001] SDK return type
return self._native.padding()
@property
def truncation(self) -> dict[str, object] | None: # mutable-ok: [LIT001, LIT002] SDK return type
def truncation(self) -> dict[str, object] | None: # mutable-ok: [LIT001] SDK return type
return self._native.truncation()
@property
@ -327,7 +315,7 @@ class HuggingFaceTokenizer:
input: Sequence[HuggingFaceBatchInput],
is_pretokenized: bool = False,
add_special_tokens: bool = True,
) -> list[HuggingFaceEncoding]: # mutable-ok: [LIT001, LIT002] SDK return type
) -> list[HuggingFaceEncoding]: # mutable-ok: [LIT001] SDK return type
return self._encode_batch(input, is_pretokenized, add_special_tokens, fast=False)
def encode_batch_fast(
@ -335,12 +323,12 @@ class HuggingFaceTokenizer:
input: Sequence[HuggingFaceBatchInput],
is_pretokenized: bool = False,
add_special_tokens: bool = True,
) -> list[HuggingFaceEncoding]: # mutable-ok: [LIT001, LIT002] SDK return type
) -> list[HuggingFaceEncoding]: # mutable-ok: [LIT001] 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, LIT002] SDK return type
) -> list[HuggingFaceEncoding]: # mutable-ok: [LIT001] 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)
@ -353,10 +341,8 @@ class HuggingFaceTokenizer:
def decode_batch(
self, sequences: Sequence[Sequence[int]], skip_special_tokens: bool = True
) -> 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
]
) -> list[str]: # mutable-ok: [LIT001] SDK return type
return [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(),))

2
litellm/llms/a2a/chat/transformation.py
View file

@ -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 { # mutable-ok: completion() and httpx take the request headers as a dict
return {
name: value
for name, value in stored_headers.items()
if not (entra_owns_authorization and str(name).lower() == "authorization")

24
litellm/llms/anthropic/chat/guardrail_translation/handler.py
View file

@ -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}} # mutable-ok: json.dumps needs a dict
return {**event, rewrite.section: {**section, rewrite.field: rewrite.value}}
def _tool_call_shapes(tool_calls: Sequence[object]) -> tuple[_ToolCallShape, ...]:
@ -539,9 +539,7 @@ 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 = { # mutable-ok: API message payload
key: value for key, value in data.items() if key != "system"
}
translation_source: Final = {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(
@ -594,7 +592,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] # mutable-ok: GenericGuardrailAPIInputs takes list[str]
texts_to_check: Final = [item.text for item in scanned]
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]
@ -691,13 +689,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": [], # mutable-ok: API message payload
"messages": [],
"system": system,
}
)
hoisted: Final = probe.get("messages") or [] # mutable-ok: API message payload
hoisted: Final = probe.get("messages") or []
return hoisted[0] if hoisted else None
@staticmethod
@ -720,9 +718,7 @@ 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 # mutable-ok: API message payload
)
return {"role": "system", "content": content} if content else None
if not isinstance(content, list):
return None
blocks: Final[list[dict[str, object]]] = [] # mutable-ok: API message payload
@ -740,9 +736,7 @@ 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 # mutable-ok: API message payload
)
return {"role": "system", "content": blocks} if blocks else None
@staticmethod
def _fold_leading_systems_into_top_level(
@ -846,7 +840,7 @@ class AnthropicMessagesHandler(BaseTranslation):
for group in group_tool_exchanges(run):
converted.extend(
anthropic_messages_pt(
messages=[run[index] for index in group], # mutable-ok: API message payload
messages=[run[index] for index in group],
model=model,
llm_provider="anthropic",
)

6
litellm/llms/anthropic/chat/handler.py
View file

@ -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) # mutable-ok: stream payload snapshot
return dict(self._web_search_calls)
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, {}), # mutable-ok: empty provider input
tool_input=self._server_tool_inputs.get(tool_use_id, {}),
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": []}, # mutable-ok: no provider result yet
{"content": []},
status="in_progress",
)
provider_specific_fields["web_search_calls"] = self._web_search_call_snapshot()

14
litellm/llms/anthropic/chat/transformation.py
View file

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

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

@ -1331,12 +1331,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)] # mutable-ok: API message payload
kept: Final = [b for b in content if not is_encrypted_reasoning_block(b)]
if len(kept) == len(content):
return message
if not kept:
return None
return {**message, "content": kept} # mutable-ok: API message payload
return {**message, "content": kept}
def strip_encrypted_reasoning_blocks_from_anthropic_messages(
@ -1348,7 +1348,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] # mutable-ok: API message payload
return [m for m in stripped if m is not None]
def strip_thinking_blocks_from_anthropic_messages_request_dict(
@ -1636,7 +1636,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]} # mutable-ok: JSON wire format
return {**message, "content": [b for b in rewritten if b is not None]}
def flatten_unencrypted_web_search_results_in_anthropic_messages(
@ -1654,49 +1654,47 @@ 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] # mutable-ok: JSON wire format
return [_flatten_web_search_results_in_message(m) for m in messages]
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"} # mutable-ok: JSON wire format
return {k: v for k, v in block.items() if k != "provider_specific_fields"}
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"]] # mutable-ok: JSON wire format
return {**message, "content": content} # mutable-ok: JSON wire format
content: Final = [_without_provider_specific_fields(b) for b in message["content"]]
return {**message, "content": content}
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] # mutable-ok: JSON wire format
return [_strip_provider_specific_fields_in_message(m) for m in messages]
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"} # mutable-ok: JSON wire format
return {"type": cache_type if isinstance(cache_type, str) else "ephemeral"}
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) # mutable-ok: JSON wire format
return dict(block)
normalized: Final = _normalized_cache_control(block["cache_control"])
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
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}
def _with_portable_cache_control_in_blocks(blocks: object) -> object:
if isinstance(blocks, str) or not isinstance(blocks, Sequence):
return blocks
return [ # mutable-ok: JSON wire format
_with_portable_cache_control(block) if isinstance(block, Mapping) else block for block in blocks
]
return [_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:
@ -1705,7 +1703,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 { # mutable-ok: JSON wire format
return {
**portable,
"content": _with_portable_cache_control_in_blocks(portable["content"]),
}
@ -1717,20 +1715,16 @@ 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 { # mutable-ok: JSON wire format
return {
**message,
"content": [ # mutable-ok: JSON wire format
_with_portable_cache_control_in_content_block(block) for block in content
],
"content": [_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 [ # mutable-ok: JSON wire format
_with_portable_cache_control_in_message(message) for message in messages
]
return [_with_portable_cache_control_in_message(message) for message in messages]
def _with_portable_cache_control_in_scoped_value(key: str, value: object) -> object:
@ -1762,9 +1756,7 @@ def normalize_cache_control_in_anthropic_payload(
dropped entirely. The caller's payload is never mutated.
"""
portable: Final = _with_portable_cache_control(payload)
return { # mutable-ok: JSON wire format
key: _with_portable_cache_control_in_scoped_value(key, value) for key, value in portable.items()
}
return {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:
@ -1791,7 +1783,7 @@ def _anthropic_model_entry(
source: Final[Mapping[str, object]] = (
MappingProxyType({"source_model": model["id"]}) if listed_id is not None else MappingProxyType({})
)
return { # mutable-ok: JSON response body, serialized by the route and never mutated
return {
"type": "model",
"id": listed_id or model["id"],
**source,
@ -1822,10 +1814,8 @@ 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 = [ # 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: Final = [_anthropic_model_entry(model, created_at, display_names, listed_ids) for model in models]
return {
"data": data,
"has_more": False,
"first_id": data[0]["id"] if data else None,

4
litellm/llms/anthropic/pass_through/adapters/streaming_iterator.py
View file

@ -1212,9 +1212,7 @@ 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
] = {} # mutable-ok: the proxy merges provider headers onto _hidden_params in place
self._hidden_params: dict[str, object] = {}
@property
def chunks(self) -> "list[ModelResponseStream] | None":

2
litellm/llms/anthropic/pass_through/adapters/transformation.py
View file

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

4
litellm/llms/anthropic/pass_through/messages/response_cache.py
View file

@ -132,7 +132,7 @@ class CachedAnthropicMessagesStreamIterator(BaseAnthropicMessagesStreamingIterat
litellm_logging_obj: "LiteLLMLoggingObj",
request_body: Mapping[str, object],
) -> None:
body: Final = dict(request_body) # mutable-ok: the base iterator takes a plain dict
body: Final = dict(request_body)
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) # mutable-ok: the logging handler takes a list
chunks: Final = list(self.chunks)
await self._handle_streaming_logging(chunks)
raise StopAsyncIteration
chunk: Final = self.chunks[self.current_index]

42
litellm/llms/anthropic/pass_through/messages/streaming_iterator.py
View file

@ -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": {}, # mutable-ok: one-shot payload
"input": {},
"type": "tool_use",
},
(
{ # mutable-ok: one-shot payload
"partial_json": json.dumps(block.get("input") or {}), # mutable-ok: one-shot payload
{
"partial_json": json.dumps(block.get("input") or {}),
"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"},) # mutable-ok: one-shot payload
({"signature": signature, "type": "signature_delta"},)
if isinstance(signature, str) and signature
else ()
)
return (
{"thinking": "", "signature": "", "type": "thinking"}, # mutable-ok: one-shot payload
{"thinking": "", "signature": "", "type": "thinking"},
(
{"thinking": block.get("thinking") or "", "type": "thinking_delta"}, # mutable-ok: one-shot payload
{"thinking": block.get("thinking") or "", "type": "thinking_delta"},
*signature_deltas,
),
)
case "redacted_thinking":
return ({"type": "redacted_thinking", "data": block.get("data")}, ()) # mutable-ok: one-shot JSON payload
return ({"type": "redacted_thinking", "data": block.get("data")}, ())
case _:
return (
{"type": "text", "text": ""}, # mutable-ok: one-shot JSON payload
({"type": "text_delta", "text": block.get("text") or ""},), # mutable-ok: one-shot JSON payload
{"type": "text", "text": ""},
({"type": "text_delta", "text": block.get("text") or ""},),
)
@ -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 = { # mutable-ok: one-shot JSON payload
message_start_usage: Final = {
**(response.get("usage") or {}),
"output_tokens": 0,
}
message_start_payload: Final = { # mutable-ok: one-shot JSON payload, never mutated after construction
message_start_payload: Final = {
"type": "message_start",
"message": { # mutable-ok: one-shot JSON payload
"message": {
**response,
"content": [], # mutable-ok: one-shot JSON payload
"content": [],
"stop_reason": None,
"stop_sequence": None,
"usage": message_start_usage,
},
}
message_delta_payload: Final = { # mutable-ok: one-shot JSON payload, never mutated after construction
message_delta_payload: Final = {
"type": "message_delta",
"delta": { # mutable-ok: one-shot JSON payload
"delta": {
"stop_reason": response.get("stop_reason"),
"stop_sequence": response.get("stop_sequence"),
},
"usage": response.get("usage") or {}, # mutable-ok: one-shot JSON payload
"usage": response.get("usage") or {},
}
return (
_sse_event("message_start", message_start_payload),
*content_events,
_sse_event("message_delta", message_delta_payload),
_sse_event("message_stop", {"type": "message_stop"}), # mutable-ok: one-shot JSON payload
_sse_event("message_stop", {"type": "message_stop"}),
)
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 = { # mutable-ok: one-shot payload
start_payload: Final = {
"type": "content_block_start",
"index": index,
"content_block": start_block,
}
stop_payload: Final = { # mutable-ok: one-shot payload
stop_payload: Final = {
"type": "content_block_stop",
"index": index,
}
delta_events: Final = tuple(
_sse_event(
"content_block_delta",
{"type": "content_block_delta", "index": index, "delta": delta}, # mutable-ok: one-shot payload
{"type": "content_block_delta", "index": index, "delta": delta},
)
for delta in deltas
)

26
litellm/llms/anthropic/pass_through/responses_adapters/streaming_iterator.py
View file

@ -191,20 +191,20 @@ class AnthropicResponsesStreamWrapper:
if block_idx < 0:
redacted_idx: Final = self._open_block(
item_id,
{"type": "redacted_thinking", "data": signature}, # mutable-ok: API message payload
{"type": "redacted_thinking", "data": signature},
)
stop: Final = {"type": "content_block_stop", "index": redacted_idx} # mutable-ok: API message payload
stop: Final = {"type": "content_block_stop", "index": redacted_idx}
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}, # mutable-ok: API message payload
"delta": {"type": "signature_delta", "signature": signature},
}
)
self._chunk_queue.append({"type": "content_block_stop", "index": block_idx}) # mutable-ok: API message payload
self._chunk_queue.append({"type": "content_block_stop", "index": block_idx})
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": { # mutable-ok: API message payload
"delta": {
"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": ""}, # mutable-ok: API message payload
{"type": "thinking", "thinking": "", "signature": ""},
)
self._chunk_queue.append(
{
@ -413,16 +413,10 @@ class AnthropicResponsesStreamWrapper:
else AnthropicUsage(input_tokens=0, output_tokens=0)
)
message_delta_payload: Final = { # mutable-ok: fresh message_delta payload built per chunk
message_delta_payload: Final = {
"stop_reason": stop_reason,
"stop_sequence": None,
**(
{ # 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
),
**({"stop_details": refusal_stop_details(refusal_text)} if stop_reason == "refusal" else {}),
}
self._chunk_queue.append(

41
litellm/llms/anthropic/pass_through/responses_adapters/transformation.py
View file

@ -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 { # mutable-ok: API message payload
return {
"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} # mutable-ok: API message payload
return {"type": "input_file", "file_url": url}
return None
@staticmethod
@ -135,10 +135,8 @@ 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 [] # mutable-ok: API message payload
)
return [*text_parts, *file_parts] # mutable-ok: API message payload
text_parts: Final = [{"type": "input_text", "text": output_text}] if output_text else []
return [*text_parts, *file_parts]
@staticmethod
def _translate_midturn_system_content_to_responses(
@ -146,12 +144,10 @@ 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 [] # mutable-ok: API message payload
)
return [{"type": "input_text", "text": content}] if content else []
if not isinstance(content, list):
return [] # mutable-ok: API message payload
return [ # mutable-ok: API message payload
return []
return [
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
@ -203,14 +199,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) # mutable-ok: API message payload
return tuple(dict(item) for item in replayed)
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", {})), # mutable-ok: API message payload
"arguments": json.dumps(first.get("input", {})),
},
)
return ()
@ -239,7 +235,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,
@ -322,8 +318,7 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
else TOOL_RESULT_IMAGE_PLACEHOLDER
)
tool_image_parts.extend(
{"type": "input_image", "image_url": url} # mutable-ok: json content part
for url in image_urls
{"type": "input_image", "image_url": url} for url in image_urls
)
else:
output_text = str(inner)
@ -336,15 +331,15 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
}
)
if tool_image_parts:
boundary_part = { # mutable-ok: json content part
boundary_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], # mutable-ok: json content list
"content": [boundary_part, *tool_image_parts],
}
)
if user_parts:
@ -373,7 +368,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", "")} # mutable-ok: API message payload
{"type": "output_text", "text": block.get("text", "")}
for block in blocks
if block.get("type") == "text"
]
@ -531,7 +526,7 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
if developer_parts:
input_items.insert(
0,
{ # mutable-ok: API message payload
{
"type": "message",
"role": "developer",
"content": developer_parts,
@ -543,7 +538,7 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
"input": input_items,
}
if include_encrypted_reasoning:
responses_kwargs["include"] = [RESPONSES_INCLUDE_ENCRYPTED_REASONING] # mutable-ok: API request payload
responses_kwargs["include"] = [RESPONSES_INCLUDE_ENCRYPTED_REASONING]
if system and not developer_parts:
if isinstance(system, str):

2
litellm/llms/anthropic/prompt_cache_prediction.py
View file

@ -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] # mutable-ok: serialize read-only inputs at the provider API boundary
return [dict(value) for value in values]
def _messages_url(model: str, api_key: str, api_base: str | None) -> str:

6
litellm/llms/azure/azure.py
View file

@ -1279,7 +1279,7 @@ class AzureChatCompletion(BaseAzureLLM, BaseLLM):
api_base=api_base,
is_async=False,
)
request_headers: Final = dict( # mutable-ok: the httpx request helpers take a dict
request_headers: Final = 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={ # mutable-ok: loggers isinstance-check this payload as a dict
additional_args={
"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={ # mutable-ok: loggers isinstance-check this payload as a dict
additional_args={
"complete_input_dict": speech_request_body(model, voice, optional_params),
"api_base": str(azure_client.base_url),
},

2
litellm/llms/azure/chat/gpt_transformation.py
View file

@ -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 = [ # mutable-ok: request tools are a JSON list
sanitized: Final = [
tool_with_sanitized_parameters(tool, flatten_combinators_and_drop_non_python_regex_patterns)
if isinstance(tool, dict)
else tool

2
litellm/llms/azure/chat/o_series_transformation.py
View file

@ -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 = { # mutable-ok: transform_request's contract takes a plain JSON params dict
flattened_params: Final = {
**optional_params,
**sanitized_tools_update(optional_params),
}

2
litellm/llms/azure/common_utils.py
View file

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

14
litellm/llms/azure/search/transformation.py
View file

@ -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 { # mutable-ok: httpx requires a plain dict of headers
return {
**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), # mutable-ok: BaseSearchConfig.get_error_class signature
headers=dict(raw_response.headers),
)
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), # mutable-ok: BaseSearchConfig.get_error_class signature
headers=dict(raw_response.headers),
)
def _priced(self, results: tuple[SearchResult, ...]) -> SearchResponse:
@ -416,16 +416,12 @@ 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), # mutable-ok: SearchResponse.results is list[SearchResult]
results=list(results),
object="search",
)
if get_secret_str(CONNECTION_ID_ENV):
return response
response._hidden_params[
"additional_headers"
] = { # mutable-ok: response_cost_calculator writes into _hidden_params
_RESPONSE_COST_HEADER: 0.0
}
response._hidden_params["additional_headers"] = {_RESPONSE_COST_HEADER: 0.0}
return response
def get_error_class(

2
litellm/llms/azure_ai/azure_model_router/transformation.py
View file

@ -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 # mutable-ok: ModelResponse requires _hidden_params to be a plain dict
transformed_response._hidden_params = { # pyright: ignore[reportPrivateUsage] # ModelResponse exposes no public hidden-params setter
**get_hidden_params_dict(transformed_response),
AZURE_MODEL_ROUTER_SELECTED_MODEL_KEY: selected_model,
}

4
litellm/llms/azure_ai/image_generation/flux_transformation.py
View file

@ -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 [ # mutable-ok: BaseImageGenerationConfig requires a list
return [
"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} # mutable-ok: inherited config contract returns a dict
return {**optional_params, **mapped_params}

4
litellm/llms/azure_ai/passthrough/transformation.py
View file

@ -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} # mutable-ok: base class contract returns dict for httpx
return {**headers, **auth_headers}
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), # mutable-ok: AzurePassthroughConfig wants a dict
request_data=dict(request_data),
logging_obj=logging_obj,
endpoint=endpoint,
)

2
litellm/llms/azure_ai/responses/transformation.py
View file

@ -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 { # mutable-ok: the handler updates the returned headers in place per the dict contract
return {
**headers,
**auth_headers,
"Content-Type": "application/json",

6
litellm/llms/base_llm/guardrail_translation/utils.py
View file

@ -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} # mutable-ok: content parts stay JSON-plain dicts
return {**part, "text": text}
def _content_with_slot_texts(content: Sequence[object], texts: Sequence[str]) -> Sequence[object]:
remaining_texts: Final = iter(texts)
return [ # mutable-ok: message content stays a JSON list
return [
_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} # mutable-ok: chat rows stay JSON-plain dicts
rewritten: Final = {**message, "content": rewritten_content}
return cast("AllMessageValues", rewritten) # cast-ok: the same row with only its text slots swapped

2
litellm/llms/base_llm/responses/codex_compat.py
View file

@ -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] # mutable-ok: downstream narrows on isinstance(list)
kept: Final = [i for i, _ in normalized if i is not None]
# Codex passthrough items sit outside the OpenAI input union.
return kept, rewritten_types # pyright: ignore[reportReturnType] # see above

2
litellm/llms/base_llm/search/transformation.py
View file

@ -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), # mutable-ok: provider error factories require dict headers
headers=dict(error.response.headers),
)
def get_error_class(

14
litellm/llms/base_llm/vector_store/transformation.py
View file

@ -48,8 +48,8 @@ class LiteLLMVectorStoreEmbeddingExecutor:
return litellm.embedding( # pyright: ignore[reportCallIssue, reportUnknownMemberType, reportUnknownVariableType] # provider kwargs are intentionally dynamic
model=model,
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
input=[query],
**dict(configuration), # pyright: ignore[reportArgumentType] # provider-specific embedding config is validated downstream
)
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], # 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
input=[query],
**dict(configuration), # pyright: ignore[reportArgumentType] # provider-specific embedding config is validated downstream
)
@ -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], # mutable-ok: Router embedding requires a mutable input list
input=[query],
**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], # mutable-ok: Router embedding requires a mutable input list
input=[query],
**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=[]) # mutable-ok: the TypedDict declares list fields
return VectorStoreIndexEndpoints(read=[], write=[])

4
litellm/llms/bedrock/chat/converse_transformation.py
View file

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

6
litellm/llms/bedrock/common_utils.py
View file

@ -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} # mutable-ok: AWS helpers require a plain dict
request_params: Final = {**optional_params, **litellm_params}
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 # mutable-ok: json
request_body["output_config"] = {"format": schema_format} # rebind-ok: out-param
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 # mutable-ok: json
request_body["output_config"] = {"format": preserved_format} # rebind-ok: out-param
def normalize_custom_field_on_tools(request_body: dict) -> None:

8
litellm/llms/bedrock/files/transformation.py
View file

@ -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 [ # mutable-ok: the base files contract returns a list
return [
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, {} # mutable-ok: the base files contract returns the query as a dict
return url, {}
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}, # mutable-ok: BaseAWSLLM takes a dict
optional_params={"aws_region_name": region_preference},
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}, # mutable-ok: botocore AWSRequest takes a dict
headers={"x-amz-content-sha256": empty_body_hash},
)
auth: Final = S3SigV4Auth(credentials, "s3", aws_region_name) # any-ok: botocore untyped
auth.add_auth(aws_request) # any-ok: botocore request mutation is untyped

4
litellm/llms/bedrock/messages/mantle_transformation.py
View file

@ -110,7 +110,7 @@ class AmazonMantleMessagesConfig(AmazonAnthropicClaudeMessagesConfig):
if value
}
)
return { # mutable-ok: the base class contract returns a dict the handler signs into in place
return {
**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 { # mutable-ok: the base class contract returns the dict the handler serializes as the body
return {
**body,
**mantle_fields,
}

2
litellm/llms/bedrock/realtime/handler.py
View file

@ -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), # mutable-ok: realtime spend logging requires a list result
list(logged_events),
prefer_async_handlers=True,
)
)

2
litellm/llms/bedrock/realtime/transformation.py
View file

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

16
litellm/llms/bedrock/responses/transformation.py
View file

@ -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]} # mutable-ok: outgoing JSON request item
return {**block, "image_url": inlined[url]}
def _inline_value(value: object, inlined: "Mapping[str, str]") -> object:
if isinstance(value, list):
return [_inline_block(block, inlined) for block in value] # mutable-ok: outgoing JSON request item
return [_inline_block(block, inlined) for block in value]
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 = { # mutable-ok: outgoing JSON request item
inlined_fields: Final = {
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} # mutable-ok: same
return {**item, **inlined_fields}
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] # mutable-ok: downstream narrows on isinstance(list)
items: Final = [_inline_item(item, inlined) for item in input]
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}"} # mutable-ok: dict return per the contract
return {**headers, "Authorization": f"Bearer {bearer}"}
def sign_request(
self,
@ -261,9 +261,7 @@ class BedrockOpenAIResponsesConfig(BaseAWSLLM, OpenAIResponsesAPIConfig):
"Bedrock Runtime Responses API: dropping unsupported parameter(s) %s that the endpoint rejects.",
unsupported,
)
params: Final = { # mutable-ok: outgoing JSON request params
key: value for key, value in mapped.items() if key not in unsupported
}
params: Final = {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

14
litellm/llms/bedrock/search/transformation.py
View file

@ -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 { # mutable-ok: httpx request headers are a dict
return {
**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 { # mutable-ok: JSON-RPC request bodies are JSON objects
return {
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": { # mutable-ok: JSON-RPC request bodies are JSON objects
"params": {
"name": tool_name,
"arguments": { # mutable-ok: JSON-RPC request bodies are JSON objects
"arguments": {
"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 = { # mutable-ok: httpx request headers are a dict
bearer_headers: Final = {
**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 { # mutable-ok: BaseAWSLLM._sign_request takes optional params as a dict
else {
**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] # mutable-ok: pydantic list field
results: Final = [_to_search_result(item) for item in items]
return SearchResponse(results=results, object="search")

2
litellm/llms/bedrock_mantle/chat/transformation.py
View file

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

10
litellm/llms/bedrock_mantle/responses/transformation.py
View file

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

20
litellm/llms/custom_httpx/llm_http_handler.py
View file

@ -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 { # mutable-ok: logging's curl and raw-request builders take dict
return {
**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) # mutable-ok: agentic hooks mutate kwargs in place
agentic_kwargs: Final = dict(litellm_params)
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) # mutable-ok: agentic hooks mutate kwargs in place
agentic_kwargs: Final = dict(litellm_params)
final_response: Final = await self._call_agentic_completion_hooks(
response=initial_response,
model=model,
@ -4735,9 +4735,7 @@ class BaseLLMHTTPHandler:
files_per_page: Final = self._files_per_listing_page(
response, provider_config, logging_obj, litellm_params, headers, sync_httpx_client, timeout
)
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
]
return [listed_file for page_files in files_per_page for listed_file in page_files]
async def async_list_files(
self,
@ -4792,9 +4790,7 @@ 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 [ # 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
]
return [listed_file async for page_files in files_per_page for listed_file in page_files]
def _files_per_listing_page(
self,
@ -9704,7 +9700,7 @@ class BaseLLMHTTPHandler:
logging_obj.pre_call(
input="",
api_key="",
additional_args={ # mutable-ok: pre_call's additional_args contract is a dict
additional_args={
"query": query,
"vector_store_id": vector_store_id,
"api_base": endpoint,
@ -9740,7 +9736,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), # mutable-ok: snapshot GenericLiteLLMParams into the Mapping shape
litellm_params=dict(litellm_params),
embedding_executor=embedding_executor,
timeout=timeout,
)
@ -9880,7 +9876,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), # mutable-ok: snapshot GenericLiteLLMParams into the Mapping shape
litellm_params=dict(litellm_params),
embedding_executor=embedding_executor,
timeout=timeout,
)

2
litellm/llms/dashscope/chat/transformation.py
View file

@ -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 [ # mutable-ok: base class contract returns a list
return [
*super().get_supported_openai_params(model=model),
"reasoning_effort",
]

8
litellm/llms/deepseek/chat/transformation.py
View file

@ -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 = [ # mutable-ok: provider messages must stay JSON-array lists the base transform mutates
transformed: Final = [
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} # mutable-ok: wire messages are plain JSON dicts
collapsed_message: Final = {**message, "content": collapsed}
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}] # mutable-ok: JSON-array content
forwarded: Final = { # mutable-ok: wire messages are plain JSON dicts
forwarded_content: Final = [*content, {"type": "text", "text": search_text}]
forwarded: Final = {
**{key: value for key, value in message_fields.items() if key != "search_results"},
"content": forwarded_content,
}

Some files were not shown because too many files have changed in this diff Show more