scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-08 03:08:45 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 115

refactor(lint): allow dict and mapping annotations in LIT001

Browse source 
Main already retired LIT002 (#43971). This narrows LIT001 to mutable sequences and sets, so dict, Dict,
DefaultDict, OrderedDict, Counter, ChainMap, defaultdict and MutableMapping annotations are allowed, nested
list/set inside a mapping still trips, and the 945 mutable-ok suppressions that only covered mapping
annotations are deleted (LIT013 now flags them). AGENTS.md guidance updated to match

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
mateo 2026-10-07 17:11:20 +00:00
parent 086bcd2a47
commit 22d584a4c1
272 changed files with 1157 additions and 1247 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
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 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 sequences or sets. For example, build values in one shot with comprehensions or generators wrapped in `tuple()` / `frozenset()` instead of seeding an empty `list`/`set` and appending to it over time. Ideally, `# mutable-ok` is never used; reach for it only as a last resort when an immutable rewrite is impossible, and always pair it with a real reason. Plain `dict` is allowed: most of the Python ecosystem takes and returns dicts, and converting to `MappingProxyType` at every boundary costs more than it protects. Never deep copy a value just to hand out an immutable view; a defensive copy of a self-referential or large object is worse than the mutation it guards against
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
@ -93,7 +93,7 @@ Follow these coding conventions for new/updated code (a three-line fix in a lega
- Composition over inheritance
- Never-nester: early returns over deep nesting
- Don't throw; model failures as values (One function (e.g., raise_public) maps error union to existing public exception contracts via exhaustive match + assert_never)
- No mutation; don't reassign variables, global or local. Instead of mutable lists and dicts, prefer tuples, frozen dataclasses (with slots=True), `MappingProxyType`, etc.
- No mutation; don't reassign variables, global or local. Prefer tuples over lists and frozensets over sets. For structured records prefer frozen dataclasses (with slots=True) or `ReadOnly` TypedDicts over dicts. A `dict` is fine for a mapping with arbitrary keys, especially at a library boundary; `MappingProxyType` is optional there, and never worth a copy
- Annotate every variable with `: Final` (LIT010). Unpacking and walrus targets cannot carry the annotation, so they are implicitly final. Don't rebind them. Never rebind or mutate function parameters (LIT011); `self`/`cls` attribute stores are the exception. If rebinding or in-place mutation is truly unavoidable, suppress with `# rebind-ok: <reason>`
- Qualify every TypedDict field with `ReadOnly[...]` (LIT012), which nests freely with `Required` / `NotRequired` / `Annotated` in any order. If making the key writable is truly unavoidable, suppress with `# writable-ok: <reason>`
- Comprehensions take at most one `for` clause and one `if` clause (LIT014); split stacked clauses into a helper generator, a named intermediate, or a plain loop. Suppress with `# comprehension-ok: <reason>` only when unavoidable

2
litellm/_redis.py
View file

@ -205,7 +205,7 @@ def _str_to_bool(value: str) -> bool:
def _coerce_redis_kwargs_types(
redis_kwargs: Mapping[str, object],
client: type | tuple[type, ...] = redis.Redis,
) -> dict[str, object]: # mutable-ok: a caller mutates the returned kwargs before constructing its client
) -> dict[str, object]:
"""Coerces string values to the numeric/boolean type ``client``'s constructor
declares for that parameter. ``client`` may be a tuple of client classes; a
parameter's type is taken from the first signature that declares it, which

6
litellm/caching/redis_batch.py
View file

@ -466,13 +466,13 @@ class RequestRedisBatches:
)
def __init__(self, post_call_deadline: float = POST_CALL_FLUSH_DEADLINE_SECONDS) -> None:
self._batches: Final[dict[object, RedisBatch]] = {} # mutable-ok: lazily filled per backend
self._post_call: Final[dict[object, RedisBatch]] = {} # mutable-ok: lazily filled per backend
self._batches: Final[dict[object, RedisBatch]] = {}
self._post_call: Final[dict[object, RedisBatch]] = {}
self.post_call_deadline: Final = post_call_deadline
self._deadline: asyncio.TimerHandle | None = None
self._deadline_flush: asyncio.Task[None] | None = None
# Reads declared early for a consumer that runs later in the request, keyed by consumer name.
self.prefetched: Final[dict[str, object]] = {} # mutable-ok: armed pre-admission, taken at use
self.prefetched: Final[dict[str, object]] = {}
def batch(self, redis_cache: RedisCache) -> RedisBatch:
key: Final = _backend_key(redis_cache)

4
litellm/caching/redis_cluster_node_isolation.py
View file

@ -142,9 +142,7 @@ def get_litellm_async_redis_cluster_class( # noqa: C901 # supports redis-py ve
self._litellm_reinit_requests = 0
self._litellm_tolerated_timeouts = 0
super().__init__(*args, **kwargs)
self._litellm_consecutive_timeouts: dict[ # mutable-ok: per-node counter updated on the command hot path
str, int
] = {}
self._litellm_consecutive_timeouts: dict[str, int] = {}
@property
def _initialize(self) -> bool:

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

@ -1382,7 +1382,7 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
super().__init__(streaming_response, sync_stream, json_mode)
self._chat_completion_id: str | None = None
self._served_service_tier: str | None = None
self._tool_call_index_map: dict[int, int] = {} # mutable-ok: per-stream accumulator state
self._tool_call_index_map: dict[int, int] = {}
def _handle_string_chunk(
self, str_line: Union[str, "BaseModel"]
@ -1403,7 +1403,7 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
@staticmethod
def _sequential_tool_call_index(
tool_call_index_map: dict[int, int] | None, # mutable-ok: per-stream state, remapped in place
tool_call_index_map: dict[int, int] | None,
output_index: int,
) -> int:
"""Chat-completions tool_call indices must be 0-based and sequential, but
@ -1422,7 +1422,7 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
@staticmethod
def translate_responses_chunk_to_openai_stream(
parsed_chunk: dict | BaseModel,
tool_call_index_map: dict[int, int] | None = None, # mutable-ok: per-stream state, remapped in place
tool_call_index_map: dict[int, int] | None = None,
) -> "ModelResponseStream":
"""
Translate a Responses API streaming chunk to OpenAI chat completion streaming format.

4
litellm/harness/context.py
View file

@ -15,9 +15,7 @@ from litellm.harness.types import Approval, Harness, PermissionMode
if TYPE_CHECKING:
from litellm.harness.endpoint import ModelEndpoint
ApprovalHandler: TypeAlias = Callable[
[Approval], bool | Awaitable[bool] # mutable-ok: Callable parameter list in a type alias, not a runtime collection
]
ApprovalHandler: TypeAlias = Callable[[Approval], bool | Awaitable[bool]]
@dataclass(frozen=True)

43
litellm/harness/endpoint.py
View file

@ -303,7 +303,7 @@ def gateway_headers(
for name, value in incoming.items()
if name.lower() not in DROPPED_REQUEST_HEADERS and not name.lower().startswith("x-litellm-")
)
metadata_json = json.dumps(dict(metadata), default=str) if metadata else None # mutable-ok: for json.dumps
metadata_json = json.dumps(dict(metadata), default=str) if metadata else None
metadata_header = (("x-litellm-spend-logs-metadata", metadata_json),) if metadata_json is not None else ()
added = (
("authorization", f"Bearer {gateway.api_key}"),
@ -333,15 +333,15 @@ def error_status(exc: BaseException) -> int:
return 500
def error_body(exc: BaseException, message: str) -> dict[str, dict[str, str]]: # mutable-ok: JSONResponse body
return {"error": {"type": type(exc).__name__, "message": message}} # mutable-ok: JSONResponse body
def error_body(exc: BaseException, message: str) -> dict[str, dict[str, str]]:
return {"error": {"type": type(exc).__name__, "message": message}}
def to_jsonable(obj: object) -> object:
if hasattr(obj, "model_dump"):
return obj.model_dump(mode="json", exclude_none=True)
if isinstance(obj, Mapping):
return dict(obj) # mutable-ok: plain-dict copy so json.dumps can serialize any Mapping
return dict(obj)
return obj
@ -457,9 +457,7 @@ class ModelEndpoint:
return self._injected_client
handler = get_async_httpx_client(
llm_provider=httpxSpecialProvider.AgentHarness,
params={ # mutable-ok: get_async_httpx_client takes a dict params argument
"timeout": HARNESS_ENDPOINT_REQUEST_TIMEOUT_SECONDS
},
params={"timeout": HARNESS_ENDPOINT_REQUEST_TIMEOUT_SECONDS},
)
return handler.client
@ -494,17 +492,12 @@ class ModelEndpoint:
Route(
f"{prefix}/{route}",
self._handle,
methods=["POST"], # mutable-ok: Starlette Route takes a methods list
methods=["POST"],
)
for prefix, route in itertools.product(ROUTE_PREFIXES, POST_ROUTES)
)
get_routes = tuple(
Route(f"{prefix}/models", self._models, methods=["GET"]) # mutable-ok: Starlette Route takes a methods list
for prefix in ROUTE_PREFIXES
)
return deps.applications.Starlette(
routes=[*post_routes, *get_routes] # mutable-ok: Starlette takes a routes list
)
get_routes = tuple(Route(f"{prefix}/models", self._models, methods=["GET"]) for prefix in ROUTE_PREFIXES)
return deps.applications.Starlette(routes=[*post_routes, *get_routes])
@property
def _responses(self) -> _ResponsesModule:
@ -521,7 +514,7 @@ class ModelEndpoint:
def _unauthorized(self) -> Response:
return self._json(
{"error": {"type": "authentication_error", "message": "invalid token"}}, # mutable-ok: JSONResponse body
{"error": {"type": "authentication_error", "message": "invalid token"}},
401,
)
@ -536,9 +529,9 @@ class ModelEndpoint:
async def _models(self, request: Request) -> Response:
if not self._authorized(request):
return self._unauthorized()
entry = {"id": self.model, "object": "model", "created": 0, "owned_by": "litellm"} # mutable-ok: JSON body
entry = {"id": self.model, "object": "model", "created": 0, "owned_by": "litellm"}
data = (entry,) if self.model else ()
return self._json({"object": "list", "data": data}) # mutable-ok: JSON response body for Starlette JSONResponse
return self._json({"object": "list", "data": data})
async def _handle(self, request: Request) -> Response:
if not self._authorized(request):
@ -574,7 +567,7 @@ class ModelEndpoint:
if self._client is None or self.gateway is None:
raise HarnessError("gateway client is not started")
if self.model:
body = {**body, "model": self.model} # mutable-ok: JSON request body re-sent upstream via httpx json=
body = {**body, "model": self.model}
upstream_request = self._client.build_request(
"POST",
f"{self.gateway.api_base}/v1/{route}",
@ -626,10 +619,8 @@ class ModelEndpoint:
tokens = (0, 0)
self._record(model, tokens[0], tokens[1], header_cost(upstream.headers))
def _sdk_kwargs(
self, body: Mapping[str, object]
) -> dict[str, object]: # mutable-ok: SDK call kwargs, mutated by _invoke_sdk then splatted
kwargs: dict[str, object] = {**body} # mutable-ok: SDK call kwargs built from the JSON body, then overridden
def _sdk_kwargs(self, body: Mapping[str, object]) -> dict[str, object]:
kwargs: dict[str, object] = {**body}
if self.model:
kwargs["model"] = self.model
if self.api_key:
@ -641,14 +632,14 @@ class ModelEndpoint:
async def _invoke_sdk(
self,
route: str,
kwargs: dict[str, Any], # mutable-ok: injects stream_options into the SDK kwargs
kwargs: dict[str, Any],
) -> object:
if route == ROUTE_MESSAGES:
return await litellm.anthropic.messages.acreate(**kwargs)
if route == ROUTE_CHAT:
if kwargs.get("stream"):
stream_options = kwargs.get("stream_options") or {} # mutable-ok: empty default for a JSON field
kwargs["stream_options"] = { # mutable-ok: JSON field sent to litellm.acompletion
stream_options = kwargs.get("stream_options") or {}
kwargs["stream_options"] = {
"include_usage": True,
**stream_options,
}

4
litellm/harness/handlers/cli_handler.py
View file

@ -73,7 +73,7 @@ def sandbox_path(private_dir: str, path: str) -> str:
async def persist_dir(sandbox: Sandbox, link_path: str, cache_subpath: str) -> None:
script_args: Final = ("-c", PERSIST_DIR_SCRIPT, "sh", link_path, cache_subpath)
cmd: Final = ["sh", *script_args] # mutable-ok: Sandbox.run takes list[str]
cmd: Final = ["sh", *script_args]
run = await sandbox.run(cmd)
if run.exit_code != 0:
verbose_logger.debug(
@ -120,7 +120,7 @@ class CLIHarnessHandler(BaseHarnessHandler):
if self._setup is None or self._private_dir is None:
raise RuntimeError("CLIHarnessHandler.turn() called before start()")
request = self.config.transform_turn_request(ctx, self._setup, self._private_dir, prompt, self._native_id)
argv: Final = list(request.argv) # mutable-ok: Sandbox.exec takes list[str]
argv: Final = list(request.argv)
proc = await ctx.sandbox.exec(argv, env=request.env, cwd=request.cwd)
self._proc = proc
tail: Final[deque[str]] = deque(maxlen=HARNESS_STDERR_TAIL_LINES) # mutable-ok: bounded stderr ring buffer

28
litellm/harness/handlers/deepagents_handler.py
View file

@ -91,7 +91,7 @@ def load_deps() -> DeepAgentsDeps:
)
_SHARED_CHECKPOINTER: dict[str, Any] = {} # mutable-ok: process-wide lazy singleton slot for the in-memory checkpointer
_SHARED_CHECKPOINTER: dict[str, Any] = {}
def shared_checkpointer(deps: DeepAgentsDeps) -> BaseCheckpointSaver:
@ -129,7 +129,7 @@ class DeepAgentsHandler(BaseHarnessHandler):
)
self._agent = deps.create_deep_agent(
model=build_chat_model(ctx, deps),
tools=list(ctx.tools), # mutable-ok: deepagents create_deep_agent(tools=) takes a list
tools=list(ctx.tools),
system_prompt=ctx.instructions,
middleware=self._middleware(deps, blocked),
subagents=self._subagents(ctx, deps, blocked),
@ -166,39 +166,39 @@ class DeepAgentsHandler(BaseHarnessHandler):
async def turn(self, ctx: SessionContext, prompt: str) -> AsyncIterator[Event]:
agent, deps = self._require_agent()
run_config = self._run_config(ctx, deps.backend.UsageCallback(ctx, ctx.model))
user_message = {"role": "user", "content": prompt} # mutable-ok: LangGraph input message dict
payload: dict[str, object] | Command = {"messages": [user_message]} # mutable-ok: LangGraph input state
user_message = {"role": "user", "content": prompt}
payload: dict[str, object] | Command = {"messages": [user_message]}
while True:
state = TurnState()
async for event in self._stream_pass(agent, payload, run_config, state):
yield event
if not state.interrupts:
break
resume: dict[str, object] = {} # mutable-ok: Command(resume=) payload, filled per answered approval
resume: dict[str, object] = {}
for interrupt in state.interrupts:
decisions: list[dict[str, Any]] = [] # mutable-ok: HITL decisions collected across awaited approvals
for request in approval_requests(getattr(interrupt, "value", None)):
approval = Approval(
tool=normalized_tool_name(str(request.get("name") or "")),
input=dict(request.get("args") or ()), # mutable-ok: Approval.input is a public dict field
input=dict(request.get("args") or ()),
)
yield approval
decisions.append(decision(*await approval.wait()))
resume[interrupt.id] = {"decisions": decisions} # mutable-ok: LangGraph HITL resume payload
resume[interrupt.id] = {"decisions": decisions}
payload = deps.command_cls(resume=resume)
await self._finish_turn(ctx, agent, run_config)
async def _stream_pass(
self,
agent: CompiledStateGraph,
payload: dict[str, object] | Command, # mutable-ok: LangGraph astream input type
payload: dict[str, object] | Command,
run_config: RunnableConfig,
state: TurnState,
) -> AsyncIterator[Event]:
async for part in agent.astream(
payload,
run_config,
stream_mode=["messages", "updates"], # mutable-ok: LangGraph stream_mode takes a list
stream_mode=["messages", "updates"],
):
# A list stream_mode yields (mode, chunk) tuples; LangGraph's overloads don't say so.
if not isinstance(part, tuple) or len(part) != 2:
@ -232,13 +232,13 @@ class DeepAgentsHandler(BaseHarnessHandler):
"recursion_limit": recursion_limit(ctx),
}
if usage_callback is not None:
run_config["callbacks"] = [usage_callback] # mutable-ok: LangChain RunnableConfig.callbacks is a list
run_config["callbacks"] = [usage_callback]
return run_config
@staticmethod
def _middleware(deps: DeepAgentsDeps, blocked: frozenset[str]) -> list[object]: # mutable-ok: deepagents API
filters = (deps.backend.ToolFilterMiddleware(blocked),) if blocked else ()
return list(filters) # mutable-ok: deepagents create_deep_agent(middleware=) takes a list
return list(filters)
def _subagents(
self, ctx: SessionContext, deps: DeepAgentsDeps, blocked: frozenset[str]
@ -249,13 +249,13 @@ class DeepAgentsHandler(BaseHarnessHandler):
has_general = any(
isinstance(s, Mapping) and s.get("name") == deps.subagent_defaults["name"] for s in user_subagents
)
spec = {**deps.subagent_defaults, "middleware": self._middleware(deps, blocked)} # mutable-ok: SubAgent dict
spec = {**deps.subagent_defaults, "middleware": self._middleware(deps, blocked)}
general = (spec,) if blocked and not has_general else ()
return [*general, *user_subagents] # mutable-ok: deepagents create_deep_agent(subagents=) takes a list
return [*general, *user_subagents]
@staticmethod
async def _install_skills(ctx: SessionContext) -> list[str] | None: # mutable-ok: deepagents skills= takes a list
if not ctx.skills:
return None
await copy_skills(ctx.sandbox, ctx.skills, f"{ctx.sandbox.workdir}/{SKILLS_DIR}")
return [f"/{SKILLS_DIR}/"] # mutable-ok: deepagents create_deep_agent(skills=) takes a list
return [f"/{SKILLS_DIR}/"]

6
litellm/harness/handlers/tool_loop_handler.py
View file

@ -89,7 +89,7 @@ def _normalize_tool_call(
return _FunctionToolCall(id=call.id, name=call.custom.name, arguments=call.custom.input)
def _parse_arguments(raw: str) -> tuple[dict[str, object], str | None]: # mutable-ok: event input uses a dict
def _parse_arguments(raw: str) -> tuple[dict[str, object], str | None]:
text: Final = raw.strip()
try:
raw_decoded: object = _JSON_DECODER.raw_decode(text)
@ -195,7 +195,7 @@ async def _execute_tool(tool: FunctionTool, arguments: Mapping[str, object]) ->
positional_args: Final = tuple(
validated[parameter.name] for parameter in parameters if parameter.kind is inspect.Parameter.POSITIONAL_ONLY
)
keyword_args: Final[dict[str, object]] = { # mutable-ok: tool calls need keyword arguments
keyword_args: Final[dict[str, object]] = {
parameter.name: validated[parameter.name]
for parameter in parameters
if parameter.kind is not inspect.Parameter.POSITIONAL_ONLY
@ -268,7 +268,7 @@ class ToolLoopHandler(BaseHarnessHandler):
tool_specs: list[ChatCompletionToolParam] = copy.deepcopy( # mutable-ok: acompletion takes tool list
list(self._tool_specs)
)
kwargs: dict[str, object] = { # mutable-ok: acompletion takes keyword arguments
kwargs: dict[str, object] = {
**{key: value for key, value in self._completion_kwargs.items() if key not in {"messages", "tools"}},
"messages": messages,
**({"tools": tool_specs} if tool_specs else {}),

8
litellm/harness/runtime.py
View file

@ -461,9 +461,7 @@ class _Turn:
sandbox = self.ctx.sandbox
after = await sandbox.snapshot()
files = await build_file_changes(sandbox, self.before, after, self.before_contents)
seen = { # mutable-ok: dedupe set grown while merging streamed FileChange events
change.path for change in files
}
seen = {change.path for change in files}
for event in self.events:
if isinstance(event, FileChange) and event.path not in seen:
files.append(event)
@ -496,9 +494,7 @@ class _Turn:
text=self._text(),
output=output,
files=files,
events=list( # mutable-ok: Result.events is a public list field; copy detaches it from the accumulator
self.events
),
events=list(self.events),
usage=usage,
cost=cost,
stop_reason=self.stop_reason,

8
litellm/harness/sandbox/docker.py
View file

@ -124,7 +124,7 @@ class DockerSandbox:
env_args = tuple(
arg for key, value in self.env.items() for arg in ("-e", f"{key}={value}")
) # comprehension-ok: flattens (flag, value) pairs into argv
return [ # mutable-ok: argv is returned as a list, the shape callers and tests compare against
return [
"run",
"-d",
"--rm",
@ -150,7 +150,7 @@ class DockerSandbox:
env_args = tuple(
arg for key, value in (env.items() if env else ()) for arg in ("-e", f"{key}={value}")
) # comprehension-ok: flattens (flag, value) pairs into argv
return [ # mutable-ok: argv is returned as a list, the shape callers and tests compare against
return [
"exec",
"-i",
"-w",
@ -257,9 +257,7 @@ class DockerSandbox:
self._processes.clear()
if self.container_id is not None:
container_id, self.container_id = self.container_id, None
await self._docker(
["rm", "-f", container_id] # mutable-ok: argv list, the shape _spawn records and tests assert on
)
await self._docker(["rm", "-f", container_id])
async def __aenter__(self) -> DockerSandbox:
await self.start()

14
litellm/harness/sandbox/snapshot.py
View file

@ -60,9 +60,7 @@ def _hash_entry(root: str, dirpath: str, filename: str) -> tuple[str, str] | Non
def _walk_entries(root: str) -> Iterator[tuple[str, str]]:
for dirpath, dirnames, filenames in os.walk(root, followlinks=False):
dirnames[:] = [ # mutable-ok: os.walk prunes only via in-place mutation of its dirnames list
d for d in dirnames if d not in HARNESS_SNAPSHOT_SKIP_DIRS
]
dirnames[:] = [d for d in dirnames if d not in HARNESS_SNAPSHOT_SKIP_DIRS]
for filename in filenames:
entry = _hash_entry(root, dirpath, filename)
if entry is not None:
@ -94,9 +92,7 @@ def diff_snapshots(
) -> list[tuple[str, FileChangeKind]]: # mutable-ok: public sandbox helper; callers compare against a list
"""Return (path, kind) for every changed file, sorted by path."""
kinds = ((path, _change_kind(path, before, after)) for path in sorted(frozenset(before) | frozenset(after)))
return [ # mutable-ok: public sandbox helper returns a list
(path, kind) for path, kind in kinds if kind is not None
]
return [(path, kind) for path, kind in kinds if kind is not None]
def _as_text(data: bytes) -> str | None:
@ -134,7 +130,7 @@ async def capture_text_contents(sandbox: Sandbox, paths_hashes: Mapping[str, str
Each kept file is <= HARNESS_MAX_DIFF_BYTES; every byte read (kept or not) counts
toward HARNESS_SNAPSHOT_MAX_TOTAL_BYTES, after which capture stops.
"""
captured: dict[str, bytes] = {} # mutable-ok: async accumulator (awaits per read), frozen on return
captured: dict[str, bytes] = {}
total = 0
for path in sorted(paths_hashes):
if total >= HARNESS_SNAPSHOT_MAX_TOTAL_BYTES:
@ -178,6 +174,4 @@ async def build_file_changes(
) -> list[FileChange]: # mutable-ok: feeds the public Result.files list
"""FileChange per changed path. diff is None when it cannot be built as text."""
contents: Mapping[str, bytes] = before_contents or MappingProxyType({})
return [ # mutable-ok: feeds the public Result.files list
await _change_for(sandbox, path, kind, contents) for path, kind in diff_snapshots(before, after)
]
return [await _change_for(sandbox, path, kind, contents) for path, kind in diff_snapshots(before, after)]

2
litellm/harness/sync.py
View file

@ -199,7 +199,7 @@ class Session:
@property
def results(self) -> list[Result]: # mutable-ok: public property; returns a detached copy of the session's results
return list(self._inner.results) # mutable-ok: detached copy so callers cannot mutate the session's accumulator
return list(self._inner.results)
@property
def session_id(self) -> str:

2
litellm/harness/types.py
View file

@ -187,7 +187,7 @@ class State:
def dumps(self) -> bytes:
return json.dumps(
{ # mutable-ok: JSON payload serialized immediately by json.dumps
{
"harness": self.harness.value,
"native_session_id": self.native_session_id,
"workdir": self.workdir,

14
litellm/integrations/custom_guardrail.py
View file

@ -345,7 +345,7 @@ class CustomGuardrail(CustomLogger):
def inject_advisory_message(
self,
data: dict[str, Any], # mutable-ok: caller's dict is mutated in place, matching mark_pre_call_hook_ran
data: dict[str, Any],
message: str,
) -> bool:
"""
@ -872,7 +872,7 @@ class CustomGuardrail(CustomLogger):
)
target: Final = self._deployment_hook_target()
scan_request: Final[dict[str, object]] = { # mutable-ok: async_pre_call_hook writes into the dict it is handed
scan_request: Final[dict[str, object]] = {
**{key: value for key, value in request_data.items() if key not in _PRE_CALL_CONTENT_KEYS},
"messages": list(messages),
**({} if target is self else _unified_hook_fields(self, request_data)),
@ -944,10 +944,10 @@ class CustomGuardrail(CustomLogger):
async def async_logging_hook(
self,
kwargs: dict, # mutable-ok: CustomLogger.async_logging_hook contract
kwargs: dict,
result: object,
call_type: str,
) -> tuple[dict, object]: # mutable-ok: CustomLogger.async_logging_hook contract
) -> tuple[dict, object]:
"""logging_only: run apply_guardrail on copies of the logged request/response and record the verdict."""
from litellm.llms import get_guardrail_translation_mapping
@ -997,11 +997,11 @@ class CustomGuardrail(CustomLogger):
async def _scan_logged_call(
self,
kwargs: dict, # mutable-ok: CustomLogger.async_logging_hook contract
kwargs: dict,
response: object | None,
translation: "BaseTranslation",
output_translation: "BaseTranslation",
scratch_metadata: dict, # mutable-ok: apply_guardrail records its verdict into request metadata
scratch_metadata: dict,
) -> None:
optional_params: Final = kwargs.get("optional_params") or {}
scratch_input: Final = copy.deepcopy(kwargs.get("messages") or kwargs.get("input"))
@ -1567,7 +1567,7 @@ def _original_inputs_for(
kwargs: Mapping[str, object],
request_data: Mapping[str, object],
event_type: GuardrailEventHooks | None,
) -> dict | None: # mutable-ok: matches _process_response(original_inputs=) signature
) -> dict | None:
"""Baseline the hook's return value is compared against to decide "allow" vs "mask".
Hooks may edit their argument in place and return it, so the baseline is always a deep

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

@ -884,7 +884,7 @@ class _Lease:
_TRACING_LOCK: Final = threading.Lock()
_TRACING: Final[dict[_TracingKey, _Lease]] = {} # mutable-ok: process-wide channel cache, guarded by _TRACING_LOCK
_TRACING: Final[dict[_TracingKey, _Lease]] = {}
def acquire_langfuse_tracing(
@ -1108,9 +1108,9 @@ class LangfuseApiClient:
def __init__(self, api: LangfuseAPI, *, prompt_cache_ttl_seconds: float) -> None:
self.api: Final = api
self.prompt_cache_ttl_seconds: Final = prompt_cache_ttl_seconds
# mutable-ok: per-client prompt cache, guarded by _lock
self._prompts: Final[dict[_PromptKey, _CachedPrompt]] = {}
# mutable-ok: keys with a refresh in flight, guarded by _lock
self._refreshing: Final[set[_PromptKey]] = set()
self._lock: Final = threading.Lock()

2
litellm/integrations/otel/plumbing/otlp_json.py
View file

@ -66,7 +66,7 @@ class OTLPJsonSpanExporter(OTLPSpanExporter):
def __init__(
self,
endpoint: str | None,
headers: dict[str, str], # mutable-ok: SDK __init__ takes Dict
headers: dict[str, str],
certificate_file: str | None = None,
session: "requests.Session | None" = None,
) -> None:

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

@ -624,12 +624,12 @@ class TenantFanOutSpanProcessor(SpanProcessor):
self._closed = False # guarded by ``_lock``: an unlocked read races the teardown it gates
self._draining = False # guarded by ``_lock``: shutdown is flushing, nothing is held any more
self._build: Final = processor_factory if processor_factory is not None else _destination_processor
self._processors: OrderedDict[object, SpanProcessor] = OrderedDict() # mutable-ok: bounded LRU
self._retired: OrderedDict[int, SpanProcessor] = OrderedDict() # mutable-ok: drains as exports finish
self._exporting: dict[int, int] = {} # mutable-ok: per-processor in-flight export count
self._pending: OrderedDict[int, _PendingTree] = OrderedDict() # mutable-ok: bounded, by trace id
self._open: OrderedDict[int, int] = OrderedDict() # mutable-ok: bounded, spans in flight by trace id
self._verdicts: OrderedDict[_VerdictKey, bool] = OrderedDict() # mutable-ok: bounded LRU
self._processors: OrderedDict[object, SpanProcessor] = OrderedDict()
self._retired: OrderedDict[int, SpanProcessor] = OrderedDict()
self._exporting: dict[int, int] = {}
self._pending: OrderedDict[int, _PendingTree] = OrderedDict()
self._open: OrderedDict[int, int] = OrderedDict()
self._verdicts: OrderedDict[_VerdictKey, bool] = OrderedDict()
self._drain: Final = drain_pool if drain_pool is not None else _DrainPool(capacity=pending_drains)
def on_start(self, span: SDKSpan, parent_context: Context | None = None) -> None:
@ -730,7 +730,7 @@ class TenantFanOutSpanProcessor(SpanProcessor):
tree: Final = self._pending.pop(trace_id, None)
if tree is None:
return ()
verdicts: dict[_VerdictKey, bool] = {} # mutable-ok: one draw per destination of this tree
verdicts: dict[_VerdictKey, bool] = {}
for _, destination in tree.held:
key = _verdict_key(trace_id, destination)
if key not in verdicts:

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

@ -172,9 +172,9 @@ class TenantTracerCache:
# updates, span counts, and retirement must be atomic.
self._lock: Final = threading.Lock()
self._providers: OrderedDict[_RouteKey, TracerProvider] = OrderedDict()
self._open_span_counts: dict[TracerProvider, int] = {} # mutable-ok: live refcount state
self._open_span_counts: dict[TracerProvider, int] = {}
# Oldest-first so an overflow of draining providers sheds the stalest.
self._retired: OrderedDict[TracerProvider, None] = OrderedDict() # mutable-ok: draining evicted providers
self._retired: OrderedDict[TracerProvider, None] = OrderedDict()
# An owned exporter is routable only when its kind actually resolves to a
# header-carrying OTLP exporter. A denylist would accept a typo'd or
# unavailable kind, which ``_exporter_from_spec`` falls back to a

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

@ -88,7 +88,7 @@ def signoz_dynamic_endpoint(params: StandardCallbackDynamicParams) -> str | None
def signoz_dynamic_headers(
params: StandardCallbackDynamicParams,
) -> dict[str, str]: # mutable-ok: DYNAMIC_HEADERS_BY_CALLBACK returns a dict
) -> dict[str, str]:
key: Final = params.get("signoz_ingestion_key")
if _tenant_endpoint_is_unusable(params) or not key:
return {}

2
litellm/integrations/prometheus_helpers/shared_prometheus_series_admissions.py
View file

@ -79,7 +79,7 @@ class SharedPrometheusSeriesAdmissions:
def __init__(self, directory: str) -> None:
self._directory = directory
self._admissions: dict[str, _MetricAdmissions] = {} # mutable-ok: one entry per metric, added on first use
self._admissions: dict[str, _MetricAdmissions] = {}
self.lock = RLock()
def admit_series(self, metric_name: str, label_values: tuple[str, ...], max_series: int) -> bool:

2
litellm/integrations/s3_v2.py
View file

@ -412,7 +412,7 @@ class S3Logger(CustomBatchLogger, BaseAWSLLM):
def _sign_put(
self, credentials: "Credentials", url: str, json_string: str, headers: Mapping[str, str]
) -> dict[str, str]: # mutable-ok: [LIT001] AsyncHTTPHandler.put/HTTPHandler.put only accept dict headers
) -> dict[str, str]:
"""
``RefreshableCredentials`` (IMDS roles) may refresh between the access key, secret and token
reads SigV4 performs, producing a mixed-generation signature that S3 rejects with 403.

6
litellm/integrations/shadow_eval_logger.py
View file

@ -847,7 +847,7 @@ class ShadowEvalLogger(CustomLogger):
self._inflight_shadow_tasks: int = 0
# Starts per job since the last cache fill, never decremented within a
# generation; the refill absorbs written rows and resets.
self._job_starts: dict[str, int] = {} # mutable-ok: per-generation counter
self._job_starts: dict[str, int] = {}
async def _active_jobs(self) -> Mapping[tuple[str, str], tuple[ActiveShadowEvalJob, ...]]:
"""Active jobs by (target_type, target_id), cache-first. A target holds at most
@ -1278,8 +1278,8 @@ class ShadowEvalLogger(CustomLogger):
router: Final = self._router_provider()
if router is None:
return _CallFailure("no router configured on this pod")
shadow_metadata: Final[dict[str, object]] = ( # mutable-ok: router writes its routing decision back
sanitized_forwardable_call_metadata(parent_metadata, SHADOW_EVAL_ROUTER_CALL_ORIGIN)
shadow_metadata: Final[dict[str, object]] = sanitized_forwardable_call_metadata(
parent_metadata, SHADOW_EVAL_ROUTER_CALL_ORIGIN
)
try:
response: Final = await router.acompletion(

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

@ -1834,7 +1834,7 @@ class WebSearchInterceptionLogger(CustomLogger):
for tool_call in tool_calls:
# Handle both Anthropic-style input and OpenAI-style function.arguments
query = None
tool_args: dict[str, object] | None = None # mutable-ok: the tool call's own arguments dict
tool_args: dict[str, object] | None = None
if "input" in tool_call and isinstance(tool_call["input"], dict):
tool_args = tool_call["input"]
query = tool_args.get("query")

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

@ -17,7 +17,7 @@ _WEB_SEARCH_TOOL_DESCRIPTION: Final = (
)
def _web_search_input_schema() -> dict[str, object]: # mutable-ok: plain-dict tool shape, as the get_* builders
def _web_search_input_schema() -> dict[str, object]:
"""
JSON schema for the web search tool's input, shared by every tool format.

6
litellm/interactions/background_cost_polling.py
View file

@ -126,9 +126,7 @@ class InMemoryBackgroundSettlementStore:
registration relies on.
"""
_rows: dict[str, PendingBackgroundInteraction | None] = field( # mutable-ok: the registry every settler shares
default_factory=dict
)
_rows: dict[str, PendingBackgroundInteraction | None] = field(default_factory=dict)
async def register(self, pending: PendingBackgroundInteraction) -> None:
self._rows[pending.interaction_id] = pending
@ -470,7 +468,7 @@ class _ActiveBackgroundPoll:
context: BackgroundInteractionPollContext
_ACTIVE_POLLS: Final[dict[str, _ActiveBackgroundPoll]] = {} # mutable-ok: asyncio needs strong refs to poll tasks
_ACTIVE_POLLS: Final[dict[str, _ActiveBackgroundPoll]] = {}
def _discard_poll(interaction_id: str, task: "asyncio.Task[SettlementOutcome | None]") -> None:

16
litellm/litellm_core_utils/core_helpers.py
View file

@ -579,8 +579,8 @@ def safe_deep_copy(data):
def independent_snapshot(
data: dict, # mutable-ok: caller-defined request-payload shape
) -> dict: # mutable-ok: caller-defined request-payload shape
data: dict,
) -> dict:
"""
A copy of ``data`` whose top-level keys are deep-copied independently
where possible -- always attempted, regardless of
@ -771,13 +771,13 @@ _NO_HEADERS: Final[Mapping[str, object]] = MappingProxyType({})
class _CarriesHiddenParams(Protocol):
_hidden_params: dict[str, object] # mutable-ok: the responses billed here keep hidden params in a plain dict
_hidden_params: dict[str, object]
@property
def hidden_params(self) -> dict[str, object]: ... # mutable-ok: API requires mutation
def hidden_params(self) -> dict[str, object]: ...
@hidden_params.setter
def hidden_params(self, hidden_params: dict[str, object]) -> None: ... # mutable-ok: API requires mutation
def hidden_params(self, hidden_params: dict[str, object]) -> None: ...
def set_response_cost_in_hidden_params(response: _CarriesHiddenParams, cost: float | None) -> None:
@ -786,7 +786,7 @@ def set_response_cost_in_hidden_params(response: _CarriesHiddenParams, cost: flo
return
hidden_params: Final = response.hidden_params
additional_headers: Final[object] = hidden_params.get("additional_headers")
merged: Final[dict[str, object]] = { # mutable-ok: assigned into the plain-dict hidden params
merged: Final[dict[str, object]] = {
**(additional_headers if isinstance(additional_headers, Mapping) else _NO_HEADERS),
RESPONSE_COST_HEADER: cost,
}
@ -802,8 +802,8 @@ def set_provider_response_headers_in_hidden_params(
) -> None:
hidden_params: Final = response.hidden_params
existing_additional_headers: Final[object] = hidden_params.get("additional_headers")
raw_headers: Final[dict[str, str]] = dict(headers) # mutable-ok: stored as the plain-dict hidden param
additional_headers: Final[dict[str, object]] = { # mutable-ok: assigned into the plain-dict hidden params
raw_headers: Final[dict[str, str]] = dict(headers)
additional_headers: Final[dict[str, object]] = {
**process_response_headers(raw_headers),
**(existing_additional_headers if isinstance(existing_additional_headers, Mapping) else _NO_HEADERS),
}

4
litellm/litellm_core_utils/get_model_cost_map.py
View file

@ -213,7 +213,7 @@ def mark_litellm_import_complete() -> None:
@dataclass(frozen=True, slots=True)
class ModelCostMapReloaded:
model_cost_map: dict # mutable-ok: adopted as litellm.model_cost, whose consumer contract is a plain mutable dict
model_cost_map: dict
revision: str | None = None
etag: str | None = None
@ -557,7 +557,7 @@ def _finalize_loaded_model_cost_map(loaded: ModelCostMapReloaded) -> ModelCostMa
def adopt_model_cost_map(
new_model_cost_map: dict, # mutable-ok: public API preserves the mutable cost-map contract
new_model_cost_map: dict,
) -> int:
import litellm
from litellm import utils

6
litellm/litellm_core_utils/internal_call_metadata.py
View file

@ -120,7 +120,7 @@ def sanitize_user_api_key_auth(auth: object) -> object:
return auth
def _sanitized(parent_metadata: Mapping[str, object]) -> dict[str, object]: # mutable-ok: SDK metadata kwarg
def _sanitized(parent_metadata: Mapping[str, object]) -> dict[str, object]:
return {
k: sanitize_user_api_key_auth(v) if k == _USER_API_KEY_AUTH_KEY else v
for k, v in parent_metadata.items()
@ -131,7 +131,7 @@ def _sanitized(parent_metadata: Mapping[str, object]) -> dict[str, object]: # m
def forwarded_internal_call_metadata(
parent_metadata: Mapping[str, object] | None,
call_origin: InternalCallOrigin,
) -> dict[str, object]: # mutable-ok: SDK metadata kwarg
) -> dict[str, object]:
"""Parent metadata, minus its budget reservation, stamped with the sub-call's origin.
For sub-calls made inside the parent request (classifier, embeddings), where the
@ -158,7 +158,7 @@ def effective_turn_off_message_logging(request_kwargs: Mapping[str, object] | No
def sanitized_forwardable_call_metadata(
parent_metadata: Mapping[str, object],
call_origin: InternalCallOrigin,
) -> dict[str, object]: # mutable-ok: SDK metadata kwarg
) -> dict[str, object]:
"""Just the caller's identity, stamped with the sub-call's origin.
For sub-calls detached from the parent request (shadow eval), which outlive it and

4
litellm/litellm_core_utils/litellm_logging.py
View file

@ -1107,7 +1107,7 @@ class Logging(LiteLLMLoggingBaseClass):
prompt_management_logger: CustomLogger | None = None,
prompt_label: str | None = None,
prompt_version: int | None = None,
request_kwargs: dict[str, object] | None = None, # mutable-ok: marker stamped into live request kwargs
request_kwargs: dict[str, object] | None = None,
injected_for_every_deployment: bool = False,
) -> tuple[str, list[AllMessageValues], dict]:
from litellm.integrations.anthropic_cache_control_hook import AnthropicCacheControlHook
@ -1158,7 +1158,7 @@ class Logging(LiteLLMLoggingBaseClass):
tools: list[dict] | None = None,
prompt_label: str | None = None,
prompt_version: int | None = None,
request_kwargs: dict[str, object] | None = None, # mutable-ok: marker stamped into live request kwargs
request_kwargs: dict[str, object] | None = None,
injected_for_every_deployment: bool = False,
) -> tuple[str, list[AllMessageValues], dict]:
from litellm.integrations.anthropic_cache_control_hook import AnthropicCacheControlHook

2
litellm/litellm_core_utils/llm_judge.py
View file

@ -27,7 +27,7 @@ def default_router_provider() -> Router | None:
return llm_router
def parse_json_verdict(raw: str) -> dict[str, object]: # mutable-ok: plain parsed-JSON payload
def parse_json_verdict(raw: str) -> dict[str, object]:
"""Parse a judge's JSON verdict, tolerating markdown fences and surrounding prose."""
text = raw.strip() # rebind-ok: progressively narrowed to the JSON payload
fenced: Final = JSON_FENCE_RE.search(text)

2
litellm/litellm_core_utils/logging_worker.py
View file

@ -61,7 +61,7 @@ class LoggingWorker:
self._queue: asyncio.Queue[LoggingTask] | None = None
self._worker_task: asyncio.Task | None = None
self._running_tasks: set[asyncio.Task] = set()
self._dequeued_tasks: dict[int, LoggingTask] = {} # mutable-ok: refs so flush can rescue never-started tasks
self._dequeued_tasks: dict[int, LoggingTask] = {}
self._sem: asyncio.Semaphore | None = None
self._bound_loop: asyncio.AbstractEventLoop | None = None
self._last_aggressive_clear_time: float = 0.0

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

@ -1212,7 +1212,7 @@ def _mergeable_branch(
branch: object,
seen_refs: frozenset[str],
depth: int,
expanded_refs: dict[str, Mapping[str, object] | None], # mutable-ok: per-call memo bounding repeated $ref work
expanded_refs: dict[str, Mapping[str, object] | None],
) -> Mapping[str, object] | None:
if not isinstance(branch, dict) or depth > _MAX_SCHEMA_FLATTEN_DEPTH:
return None
@ -1245,7 +1245,7 @@ def _flatten_schema_against_root(
root: Mapping[str, object],
seen_refs: frozenset[str],
depth: int,
expanded_refs: dict[str, Mapping[str, object] | None], # mutable-ok: per-call memo bounding repeated $ref work
expanded_refs: dict[str, Mapping[str, object] | None],
) -> Mapping[str, object]:
raw_branch_groups: Final = tuple(
(
@ -1377,7 +1377,7 @@ def drop_lookaround_regex_patterns(schema: Mapping[str, object]) -> Mapping[str,
def _schema_without_rejected_regex(
schema: Mapping[str, object], rejected: Callable[[str], bool]
) -> Mapping[str, object]:
rebuilt: dict[int, Mapping[str, object]] = {} # mutable-ok: per-call memo of rewritten nodes, deepest level first
rebuilt: dict[int, Mapping[str, object]] = {}
for level in reversed(tuple(islice(_schema_levels(schema), _MAX_SCHEMA_NESTING))):
rebuilt.update(
(id(node), rewritten)

4
litellm/litellm_core_utils/provider_affinity.py
View file

@ -83,9 +83,7 @@ def get_stable_session_id(litellm_params: object | None) -> str | None:
return None
def add_provider_affinity_header(
headers: Mapping[str, object], litellm_params: object | None
) -> dict[str, object]: # mutable-ok: downstream handlers add auth and signing headers
def add_provider_affinity_header(headers: Mapping[str, object], litellm_params: object | None) -> dict[str, object]:
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)

4
litellm/litellm_core_utils/sensitive_data_masker.py
View file

@ -194,9 +194,7 @@ def mask_credentials_in_payload(data: object) -> object:
@dataclass(frozen=True, slots=True)
class _PayloadWalker:
_memo: dict[tuple[int, bool], tuple[object, object]] = field( # mutable-ok: memo of one walk, pins each keyed node
default_factory=dict
)
_memo: dict[tuple[int, bool], tuple[object, object]] = field(default_factory=dict)
def walk(self, node: object, key_is_sensitive: bool, depth: int) -> object:
if not isinstance(node, (Mapping, list, tuple, BaseModel)):

4
litellm/litellm_core_utils/streaming_handler.py
View file

@ -208,11 +208,11 @@ def _provider_hidden_params(
class CustomStreamWrapper:
@property
def hidden_params(self) -> dict[str, object]: # mutable-ok: API requires mutation
def hidden_params(self) -> dict[str, object]:
return self._hidden_params
@hidden_params.setter
def hidden_params(self, hidden_params: dict[str, object]) -> None: # mutable-ok: API requires mutation
def hidden_params(self, hidden_params: dict[str, object]) -> None:
self._hidden_params = hidden_params
def __init__(

8
litellm/litellm_core_utils/tokenizer.py
View file

@ -263,13 +263,13 @@ class HuggingFaceTokenizer:
def id_to_token(self, id: int) -> str | None:
return self._native.id_to_token(id)
def get_vocab(self, with_added_tokens: bool = True) -> dict[str, int]: # mutable-ok: [LIT001] SDK return type
def get_vocab(self, with_added_tokens: bool = True) -> dict[str, int]:
return self._native.get_vocab(with_added_tokens)
def get_vocab_size(self, with_added_tokens: bool = True) -> int:
return self._native.get_vocab_size(with_added_tokens)
def get_added_tokens_decoder(self) -> dict[int, AddedToken]: # mutable-ok: [LIT001] SDK return type
def get_added_tokens_decoder(self) -> dict[int, AddedToken]:
return {
token_id: AddedToken(
content, single_word=single_word, lstrip=lstrip, rstrip=rstrip, normalized=normalized, special=special
@ -288,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] SDK return type
def padding(self) -> dict[str, object] | None:
return self._native.padding()
@property
def truncation(self) -> dict[str, object] | None: # mutable-ok: [LIT001] SDK return type
def truncation(self) -> dict[str, object] | None:
return self._native.truncation()
@property

4
litellm/llms/anthropic/batches/handler.py
View file

@ -42,7 +42,7 @@ class AnthropicBatchesHandler:
timeout: float | httpx.Timeout,
max_retries: int | None,
logging_obj: LiteLLMLoggingObj | None = None,
litellm_params: dict | None = None, # mutable-ok: handed straight to validate_environment
litellm_params: dict | None = None,
) -> LiteLLMBatch:
"""
Async: Retrieve a batch from Anthropic.
@ -131,7 +131,7 @@ class AnthropicBatchesHandler:
timeout: float | httpx.Timeout,
max_retries: int | None,
logging_obj: LiteLLMLoggingObj | None = None,
litellm_params: dict | None = None, # mutable-ok: handed straight to validate_environment
litellm_params: dict | None = None,
) -> LiteLLMBatch | Coroutine[Any, Any, LiteLLMBatch]:
"""
Retrieve a batch from Anthropic.

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

@ -719,7 +719,7 @@ class AnthropicMessagesHandler(BaseTranslation):
@staticmethod
def _openai_system_message_to_anthropic(
message: Mapping[str, object],
) -> dict[str, object] | None: # mutable-ok: API message payload
) -> dict[str, object] | None:
"""Convert an OpenAI system message to the client's Anthropic-shaped entry."""
content: Final = message.get("content")
if isinstance(content, str):
@ -733,7 +733,7 @@ class AnthropicMessagesHandler(BaseTranslation):
text = block.get("text")
if not isinstance(text, str) or not text:
continue
anthropic_block: dict[str, object] = { # mutable-ok: API message payload
anthropic_block: dict[str, object] = {
"type": "text",
"text": text,
}
@ -745,7 +745,7 @@ class AnthropicMessagesHandler(BaseTranslation):
@staticmethod
def _fold_leading_systems_into_top_level(
data: dict[str, object], # mutable-ok: API message payload
data: dict[str, object],
leading_systems: Sequence[object],
include_existing_system: bool,
) -> None:
@ -826,7 +826,7 @@ class AnthropicMessagesHandler(BaseTranslation):
@staticmethod
def _write_back_structured_messages(
data: dict, # mutable-ok: API message payload
data: dict,
structured_messages: list, # mutable-ok: API message payload
hoisted_system_message: object = None,
preserve_system_messages: bool = False,
@ -1070,7 +1070,7 @@ class AnthropicMessagesHandler(BaseTranslation):
async def _apply_guardrail_responses_to_input(
self,
data: dict[str, object], # mutable-ok: API message payload
data: dict[str, object],
responses: Sequence[str],
scanned: tuple[ScannedText, ...],
) -> None:

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

@ -369,7 +369,7 @@ class AnthropicChatCompletion(BaseLLM):
transform_params: Final = {**optional_params, "is_vertex_request": is_vertex_request}
def finish_request(request_data: dict) -> tuple[dict, dict]: # mutable-ok: rewritten in place downstream
def finish_request(request_data: dict) -> tuple[dict, dict]:
"""Filter beta headers and emit pre_call, returning `(headers, data)`.
The pair stays mutable because the streaming path rewrites it in
@ -586,7 +586,7 @@ class ModelResponseIterator:
# Accumulate web_search_tool_result blocks for multi-turn reconstruction
# See: https://github.com/BerriAI/litellm/issues/17737
self.web_search_results: list[dict[str, object]] = []
self._web_search_calls: dict[str, object] = {} # mutable-ok: provider call state by id
self._web_search_calls: dict[str, object] = {}
# Accumulate compaction blocks for multi-turn reconstruction
self.compaction_blocks: list[dict[str, object]] = []

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

@ -493,7 +493,7 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
def maybe_drop_speed_param(
cls,
model: str,
optional_params: dict[str, object], # mutable-ok: mirrors override contract
optional_params: dict[str, object],
drop_params: bool,
custom_llm_provider: str | None = None,
) -> None:
@ -1949,10 +1949,10 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
self,
model: str,
messages: list[AllMessageValues], # mutable-ok: BaseConfig signature
optional_params: dict[str, object], # mutable-ok: BaseConfig signature
litellm_params: dict[str, object], # mutable-ok: BaseConfig signature
headers: dict[str, object], # mutable-ok: BaseConfig signature
) -> dict[str, object]: # mutable-ok: BaseConfig signature
optional_params: dict[str, object],
litellm_params: dict[str, object],
headers: dict[str, object],
) -> dict[str, object]:
return self.transform_request(
model=model,
messages=await async_inline_remote_media(messages, should_inline=self.inlines_remote_media),

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

@ -667,7 +667,7 @@ class AnthropicModelInfo(BaseLLMModelInfo):
@classmethod
def apply_sampling_param(
cls,
optional_params: dict[str, object], # mutable-ok: mirrors override contract
optional_params: dict[str, object],
model: str,
param: str,
value: object,
@ -892,7 +892,7 @@ class AnthropicModelInfo(BaseLLMModelInfo):
@staticmethod
def maybe_drop_disabled_thinking(
model: str,
optional_params: MutableMapping[str, object], # mutable-ok: in-place out-param, as in _maybe_drop_speed_param
optional_params: MutableMapping[str, object],
custom_llm_provider: str,
) -> None:
"""Omit ``thinking={'type': 'disabled'}`` for always-on-thinking models
@ -912,7 +912,7 @@ class AnthropicModelInfo(BaseLLMModelInfo):
@staticmethod
def translate_legacy_thinking_for_adaptive_model(
model: str,
optional_params: MutableMapping[str, object], # mutable-ok: in-place out-param like the sibling helpers
optional_params: MutableMapping[str, object],
custom_llm_provider: str,
) -> None:
"""Translate legacy ``thinking.type=enabled`` to adaptive for the
@ -1634,7 +1634,7 @@ def strip_thinking_blocks_from_anthropic_messages(messages: Sequence[object]) ->
return out
def _without_encrypted_reasoning_blocks(message: dict) -> dict | None: # mutable-ok: Anthropic message payload shape
def _without_encrypted_reasoning_blocks(message: dict) -> dict | None:
if not isinstance(message, Mapping):
return message
content: Final = message.get("content")
@ -1649,7 +1649,7 @@ def _without_encrypted_reasoning_blocks(message: dict) -> dict | None: # mutabl
def strip_encrypted_reasoning_blocks_from_anthropic_messages(
messages: Sequence[dict], # mutable-ok: Anthropic message payload shape
messages: Sequence[dict],
) -> list[dict]: # mutable-ok: AnthropicMessagesRequest.messages is typed list[dict]
"""
Drop thinking / redacted_thinking blocks that carry another provider's encrypted
@ -1999,14 +1999,14 @@ def strip_provider_specific_fields_from_anthropic_messages(
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
def _normalized_cache_control(cache_control: object) -> dict[str, str] | None:
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"}
def _with_portable_cache_control(block: Mapping[str, object]) -> dict[str, object]: # mutable-ok: JSON wire format
def _with_portable_cache_control(block: Mapping[str, object]) -> dict[str, object]:
if "cache_control" not in block:
return dict(block)
normalized: Final = _normalized_cache_control(block["cache_control"])
@ -2062,7 +2062,7 @@ def _with_portable_cache_control_in_scoped_value(key: str, value: object) -> obj
def normalize_cache_control_in_anthropic_payload(
payload: Mapping[str, object],
) -> dict[str, object]: # mutable-ok: JSON wire format
) -> dict[str, object]:
"""
Return a copy of an Anthropic /v1/messages payload with every
``cache_control`` entry reduced to ``{"type": <its type, or "ephemeral">}``

2
litellm/llms/anthropic/count_tokens/transformation.py
View file

@ -51,7 +51,7 @@ class AnthropicCountTokensConfig:
tools: list[dict[str, JsonValue]] | None = None,
system: JsonValue = None,
optional_params: Mapping[str, JsonValue] | None = None,
) -> dict[str, JsonValue]: # mutable-ok: provider transport requires JSON dictionaries
) -> dict[str, JsonValue]:
"""
Transform request to Anthropic CountTokens format.

10
litellm/llms/anthropic/files/transformation.py
View file

@ -111,14 +111,14 @@ class AnthropicFilesConfig(BaseFilesConfig):
async def avalidate_environment(
self,
headers: dict, # mutable-ok: mirrors the sync validate_environment contract this overrides
headers: dict,
model: str,
messages: list, # mutable-ok: mirrors the sync validate_environment contract this overrides
optional_params: dict, # mutable-ok: mirrors the sync validate_environment contract this overrides
litellm_params: dict, # mutable-ok: mirrors the sync validate_environment contract this overrides
optional_params: dict,
litellm_params: dict,
api_key: str | None = None,
api_base: str | None = None,
) -> dict: # mutable-ok: mirrors the sync validate_environment contract this overrides
) -> dict:
"""Async counterpart of validate_environment: the WIF tier can block on a token
exchange POST, so async callers await it off the event loop."""
params_mapping, resolved_api_base = self._resolve_params(litellm_params, api_base)
@ -136,7 +136,7 @@ class AnthropicFilesConfig(BaseFilesConfig):
return params_mapping, resolved_api_base
@staticmethod
def _finalize_headers(headers: dict, auth_header: Mapping[str, str] | None) -> dict: # mutable-ok: out-param
def _finalize_headers(headers: dict, auth_header: Mapping[str, str] | None) -> dict:
if auth_header is None:
raise ValueError(
"Anthropic API key is required. Set ANTHROPIC_API_KEY or ANTHROPIC_AUTH_TOKEN environment variable or pass api_key parameter."

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

@ -44,7 +44,7 @@ class AnthropicMessagesStreamCacheWriter:
self.caching_handler = caching_handler
self.collected_chunks: list[bytes] = [] # mutable-ok: rebuilding a tuple per SSE chunk is quadratic
self.persisted = False
self._hidden_params: dict[str, object] = dict( # mutable-ok: callers stamp cache_key in here
self._hidden_params: dict[str, object] = dict(
stream.hidden_params if isinstance(stream, AnthropicMessagesStreamingResponse) else _EMPTY_MAPPING
)
@ -139,7 +139,7 @@ class CachedAnthropicMessagesStreamIterator(BaseAnthropicMessagesStreamingIterat
self.chunks: Final[tuple[bytes, ...]] = tuple(event.encode("utf-8") for event in events)
self.current_index = 0
self.logged = False
self._hidden_params: dict[str, object] = {"cache_hit": True} # mutable-ok: callers stamp cache_key in here
self._hidden_params: dict[str, object] = {"cache_hit": True}
litellm_logging_obj.model_call_details["cache_hit"] = True
def __aiter__(self) -> "CachedAnthropicMessagesStreamIterator":

16
litellm/llms/anthropic/pass_through/messages/transformation.py
View file

@ -290,14 +290,14 @@ class AnthropicMessagesConfig(BaseAnthropicMessagesConfig):
async def avalidate_anthropic_messages_environment(
self,
headers: dict, # mutable-ok: mirrors the sync validate_anthropic_messages_environment contract
headers: dict,
model: str,
messages: list[Any], # mutable-ok: mirrors the sync validate_anthropic_messages_environment contract
optional_params: dict, # mutable-ok: mirrors the sync validate_anthropic_messages_environment contract
litellm_params: dict, # mutable-ok: mirrors the sync validate_anthropic_messages_environment contract
optional_params: dict,
litellm_params: dict,
api_key: str | None = None,
api_base: str | None = None,
) -> tuple[dict, str | None]: # mutable-ok: mirrors the sync validate_anthropic_messages_environment contract
) -> tuple[dict, str | None]:
if type(self).validate_anthropic_messages_environment is not (
AnthropicMessagesConfig.validate_anthropic_messages_environment
):
@ -342,7 +342,7 @@ class AnthropicMessagesConfig(BaseAnthropicMessagesConfig):
return auth_header
@staticmethod
def _apply_env_auth_header(headers: dict, auth_header: Mapping[str, str] | None) -> None: # mutable-ok: out-param
def _apply_env_auth_header(headers: dict, auth_header: Mapping[str, str] | None) -> None:
if auth_header is None:
return
merged_beta: Final = merge_anthropic_beta_headers(
@ -362,10 +362,10 @@ class AnthropicMessagesConfig(BaseAnthropicMessagesConfig):
def _finalize_messages_headers(
self,
headers: dict, # mutable-ok: out-param
optional_params: dict, # mutable-ok: out-param
headers: dict,
optional_params: dict,
messages: list[object], # mutable-ok: mirrors the validate_anthropic_messages_environment contract
) -> dict: # mutable-ok: out-param
) -> dict:
if "anthropic-version" not in headers:
headers["anthropic-version"] = DEFAULT_ANTHROPIC_API_VERSION
if "content-type" not in headers:

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

@ -98,7 +98,7 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
@staticmethod
def _translate_anthropic_document_block_to_file_part(
block: Mapping[str, object],
) -> dict[str, str] | None: # mutable-ok: API message payload
) -> dict[str, str] | None:
"""Convert an Anthropic document block to a Responses input_file part."""
raw_source: Final = block.get("source")
if not isinstance(raw_source, Mapping):
@ -130,7 +130,7 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
@staticmethod
def _tool_result_output_value(
output_text: str,
file_parts: tuple[dict[str, str], ...], # mutable-ok: json content parts
file_parts: tuple[dict[str, str], ...],
) -> str | list[dict[str, str]]: # mutable-ok: API message payload
"""Plain string output, or a part list when document file parts are present."""
if not file_parts:
@ -165,7 +165,7 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
cls,
summary: Iterable[object],
encrypted_content: object,
) -> dict[str, object] | None: # mutable-ok: API message payload
) -> dict[str, object] | None:
"""The one Anthropic block for a Responses reasoning item.
The item's encrypted reasoning rides the block's opaque field (`signature`, or
@ -192,9 +192,7 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
return "thinking" if block.get("type") in ("thinking", "redacted_thinking") else f"block:{index}"
@classmethod
def _assistant_group_to_input_items(
cls, group: tuple[Mapping[str, object], ...]
) -> tuple[dict[str, object], ...]: # mutable-ok: API message payload
def _assistant_group_to_input_items(cls, group: tuple[Mapping[str, object], ...]) -> tuple[dict[str, object], ...]:
first: Final = group[0]
btype: Final = first.get("type")
if btype in ("thinking", "redacted_thinking"):

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

@ -425,7 +425,7 @@ def _digest(value: object) -> str:
).hexdigest()
def _json_object(value: object) -> dict[str, JsonValue]: # mutable-ok: JSON serialization requires a dictionary
def _json_object(value: object) -> dict[str, JsonValue]:
return _JSON_OBJECT.validate_python(value)

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

@ -1269,7 +1269,7 @@ class AzureChatCompletion(BaseAzureLLM, BaseLLM):
if not isinstance(max_retries, int):
raise AzureOpenAIError(status_code=422, message="max retries must be an int")
auth_params: Final[dict[str, object]] = {**(litellm_params or {})} # mutable-ok: SDK init takes a dict
auth_params: Final[dict[str, object]] = {**(litellm_params or {})}
if azure_ad_token is not None:
auth_params["azure_ad_token"] = azure_ad_token
if azure_ad_token_provider is not None:

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

@ -789,9 +789,9 @@ class BaseAzureLLM(BaseOpenAILLM):
@classmethod
def base_validate_azure_environment(
cls,
headers: dict[str, str], # mutable-ok: mirrors override contract
headers: dict[str, str],
litellm_params: GenericLiteLLMParams | None,
) -> dict[str, str]: # mutable-ok: mirrors override contract
) -> dict[str, str]:
return cls._base_validate_azure_environment(headers, litellm_params)
@staticmethod

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

@ -279,11 +279,11 @@ class BingGroundingSearchConfig(BaseSearchConfig):
def validate_environment(
self,
headers: dict[str, str], # mutable-ok: BaseSearchConfig.validate_environment signature
headers: dict[str, str],
api_key: str | None = None,
api_base: str | None = None,
**kwargs: object, # kwargs-ok: BaseSearchConfig.validate_environment signature
) -> dict[str, str]: # mutable-ok: the http handler passes this straight to httpx as headers
) -> dict[str, str]:
"""
Validate environment and return headers.
@ -331,7 +331,7 @@ class BingGroundingSearchConfig(BaseSearchConfig):
def get_complete_url(
self,
api_base: str | None,
optional_params: dict[str, object], # mutable-ok: BaseSearchConfig.get_complete_url signature
optional_params: dict[str, object],
data: dict[str, object] | list[dict[str, object]] | None = None, # mutable-ok: base signature
**kwargs: object, # kwargs-ok: BaseSearchConfig.get_complete_url signature
) -> str:
@ -349,9 +349,9 @@ class BingGroundingSearchConfig(BaseSearchConfig):
def transform_search_request(
self,
query: str | list[str], # mutable-ok: BaseSearchConfig.transform_search_request signature
optional_params: dict[str, object], # mutable-ok: base signature
optional_params: dict[str, object],
**kwargs: object, # kwargs-ok: BaseSearchConfig.transform_search_request signature
) -> dict[str, object]: # mutable-ok: the http handler passes this straight to httpx as the JSON body
) -> dict[str, object]:
"""
Transform Search request to the Foundry Responses API format.
@ -429,7 +429,7 @@ class BingGroundingSearchConfig(BaseSearchConfig):
self,
error_message: str,
status_code: int,
headers: dict[str, str], # mutable-ok: BaseSearchConfig.get_error_class signature
headers: dict[str, str],
) -> Exception:
detail: Final = _unwrap_error_detail(error_message).rstrip(". ")
return BaseLLMException(

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

@ -187,7 +187,7 @@ class AzureAIAgentsConfig(BaseConfig):
def get_agent_id(
self,
model: str,
optional_params: dict[str, object], # mutable-ok: mirrors override contract
optional_params: dict[str, object],
) -> str:
return self._get_agent_id(model, optional_params)

6
litellm/llms/azure_ai/chat/transformation.py
View file

@ -91,11 +91,11 @@ class AzureAIStudioConfig(OpenAIConfig):
def map_openai_params(
self,
non_default_params: dict[str, object], # mutable-ok: OpenAIConfig.map_openai_params signature
optional_params: dict[str, object], # mutable-ok: OpenAIConfig.map_openai_params signature
non_default_params: dict[str, object],
optional_params: dict[str, object],
model: str,
drop_params: bool,
) -> dict[str, object]: # mutable-ok: OpenAIConfig.map_openai_params signature
) -> dict[str, object]:
if not azureAIGPT5Config.is_model_gpt_5_model(model):
return super().map_openai_params(
non_default_params=non_default_params,

2
litellm/llms/azure_ai/embed/cohere_transformation.py
View file

@ -70,7 +70,7 @@ class AzureAICohereConfig:
def transform_request(
self,
input: list[str], # mutable-ok: mirrors override contract
optional_params: dict[str, object], # mutable-ok: mirrors override contract
optional_params: dict[str, object],
model: str,
) -> tuple[ImageEmbeddingRequest, EmbeddingCreateParams, list[int]]: # mutable-ok: mirrors override contract
return self._transform_request(input, optional_params, model)

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

@ -123,7 +123,7 @@ class AzureFoundryFluxImageGenerationConfig(GPTImageGenerationConfig):
optional_params: Mapping[str, object],
model: str,
drop_params: bool,
) -> dict[str, object]: # mutable-ok: inherited config contract returns a dict
) -> dict[str, object]:
if not self.is_flux2_model(model):
return super().map_openai_params(
non_default_params=dict(non_default_params),

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

@ -128,7 +128,7 @@ class AzureAIPassthroughConfig(AzureFoundryModelInfo, BasePassthroughConfig):
litellm_params: Mapping[str, object],
api_key: str | None = None,
api_base: str | None = None,
) -> dict[str, str]: # mutable-ok: base class contract returns dict for httpx
) -> dict[str, str]:
auth_headers: Final = get_azure_ai_auth_headers(
api_key=api_key,
litellm_params=litellm_params,

8
litellm/llms/base_llm/anthropic_messages/transformation.py
View file

@ -43,14 +43,14 @@ class BaseAnthropicMessagesConfig(ABC):
async def avalidate_anthropic_messages_environment(
self,
headers: dict, # mutable-ok: mirrors the sync validate_anthropic_messages_environment contract
headers: dict,
model: str,
messages: list[Any], # mutable-ok: mirrors the sync validate_anthropic_messages_environment contract
optional_params: dict, # mutable-ok: mirrors the sync validate_anthropic_messages_environment contract
litellm_params: dict, # mutable-ok: mirrors the sync validate_anthropic_messages_environment contract
optional_params: dict,
litellm_params: dict,
api_key: str | None = None,
api_base: str | None = None,
) -> tuple[dict, str | None]: # mutable-ok: mirrors the sync validate_anthropic_messages_environment contract
) -> tuple[dict, str | None]:
"""Async counterpart used by the async handler. The default delegates to the
sync implementation; providers whose sync path can block the event loop
(e.g. a WIF token exchange) override this."""

2
litellm/llms/base_llm/auth/token_exchange.py
View file

@ -653,7 +653,7 @@ class JwtBearerTokenExchangeEngine:
self._shared_store: Final = shared_store
self._wall_clock: Final = wall_clock
self._lock: Final = threading.Lock()
self._entries: Final[dict[str, _Entry]] = {} # mutable-ok: engine-owned map guarded by _lock
self._entries: Final[dict[str, _Entry]] = {}
def get_token(self, spec: TokenExchangeSpec) -> ExchangeResult:
"""A follower whose leader published nothing re-classifies rather than recursing, so a

2
litellm/llms/base_llm/base_model_iterator.py
View file

@ -112,7 +112,7 @@ class BaseModelResponseIterator:
def string_to_dict_parser(
cls,
str_line: str,
) -> dict[str, object] | None: # mutable-ok: mirrors override contract
) -> dict[str, object] | None:
return cls._string_to_dict_parser(str_line)
def _handle_string_chunk(self, str_line: str) -> GenericStreamingChunk | ModelResponseStream:

4
litellm/llms/base_llm/chat/transformation.py
View file

@ -148,9 +148,9 @@ class BaseConfig(ABC):
def add_tools_to_optional_params(
self,
optional_params: dict[str, object], # mutable-ok: mirrors override contract
optional_params: dict[str, object],
tools: list[ChatCompletionToolParam], # mutable-ok: mirrors override contract
) -> dict[str, object]: # mutable-ok: mirrors override contract
) -> dict[str, object]:
return self._add_tools_to_optional_params(optional_params, tools)
def translate_developer_role_to_system_role(

2
litellm/llms/base_llm/files/batch_records.py
View file

@ -18,7 +18,7 @@ def _responses_request_keys() -> frozenset[str]:
def responses_batch_body_to_chat_body(
openai_request_body: Mapping[str, object],
custom_llm_provider: str | None = None,
) -> dict[str, object]: # mutable-ok: provider transforms take the bridged chat body as a plain dict
) -> dict[str, object]:
"""
Rewrite the body of an OpenAI `/v1/responses` batch record as a Chat Completions body.

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

@ -166,7 +166,7 @@ class BaseFilesConfig(BaseConfig):
self,
raw_response: httpx.Response,
optional_params: Mapping[str, object],
litellm_params: dict, # mutable-ok: carries provider stashes from the request transform to the response one
litellm_params: dict,
) -> tuple[str, dict[str, str]] | None:
"""Request for the page after `raw_response`, or None once the listing is complete."""
return None

2
litellm/llms/base_llm/guardrail_translation/base_translation.py
View file

@ -121,7 +121,7 @@ class BaseTranslation(ABC):
@staticmethod
def merge_user_api_key_metadata_into_request(
request_data: dict[str, Any], # mutable-ok: proxy hooks share and mutate the request payload dict in place
request_data: dict[str, Any],
user_api_key_dict: Optional["UserAPIKeyAuth"],
) -> None:
"""

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

@ -33,7 +33,7 @@ StreamStateT = TypeVar("StreamStateT")
def event_list(*events: Event) -> Sequence[Event]:
"""A transform_stream_line result. One place builds it so every parser returns the same shape."""
return list(events) # mutable-ok: stream-line results are list-shaped; callers and tests compare with list literals
return list(events)
class HarnessTurnError(HarnessError):

14
litellm/llms/base_llm/harness/utils.py
View file

@ -26,7 +26,7 @@ _JSON_OBJECT_ADAPTER: Final = TypeAdapter(dict[str, object])
_RAW_DECODE_ADAPTER: Final = TypeAdapter(tuple[object, int])
def gateway_headers(ctx: SessionContext) -> dict[str, str]: # mutable-ok: acompletion(extra_headers=) requires dict
def gateway_headers(ctx: SessionContext) -> dict[str, str]:
metadata_json: Final = json.dumps(dict(ctx.metadata), default=str) if ctx.metadata else None
return {
"x-litellm-tags": f"harness,{ctx.harness.value}",
@ -42,7 +42,7 @@ def normalize_tool_name(native_name: str, mapping: Mapping[str, str]) -> str:
def native_tool_names(normalized: Sequence[str], mapping: Mapping[str, Sequence[str]]) -> Sequence[str]:
"""Native names for normalized tool names, de-duplicated, order kept."""
expanded: Final = itertools.chain.from_iterable(mapping.get(name, (name,)) for name in normalized)
return list(dict.fromkeys(expanded)) # mutable-ok: public helper whose callers/tests compare against list literals
return list(dict.fromkeys(expanded))
def last_json_object(text: str) -> str | None:
@ -80,22 +80,22 @@ def strict_json_schema(schema: JSONValue, depth: int = 0) -> JSONValue:
if depth > DEFAULT_MAX_RECURSE_DEPTH:
raise ValueError(f"output schema is nested deeper than {DEFAULT_MAX_RECURSE_DEPTH} levels")
if isinstance(schema, list):
return [strict_json_schema(entry, depth + 1) for entry in schema] # mutable-ok: JSON document output
return [strict_json_schema(entry, depth + 1) for entry in schema]
if not isinstance(schema, dict):
return schema
entries: Final = ((key, strict_json_schema(value, depth + 1)) for key, value in schema.items())
result = dict(entries) # mutable-ok: JSON document; "default" is popped below
result = dict(entries)
if "$ref" in result:
return {"$ref": result["$ref"]} # mutable-ok: JSONValue output is a plain JSON document
return {"$ref": result["$ref"]}
result.pop("default", None)
properties = result.get("properties")
if result.get("type") == "object" or isinstance(properties, dict):
props = properties if isinstance(properties, dict) else {} # mutable-ok: JSONValue object member
props = properties if isinstance(properties, dict) else {}
required: Final[list[JSONValue]] = list(props) # mutable-ok: JSON array in the output schema
strict: Final[Mapping[str, JSONValue]] = MappingProxyType(
{"properties": props, "required": required, "additionalProperties": False}
)
result = {**result, **strict} # mutable-ok: JSON document output
result = {**result, **strict}
return result

4
litellm/llms/base_llm/ocr/transformation.py
View file

@ -92,11 +92,11 @@ class OCRResponse(LiteLLMPydanticObjectBase):
_hidden_params: dict = PrivateAttr(default_factory=dict)
@property
def hidden_params(self) -> dict[str, builtins.object]: # mutable-ok: API requires mutation
def hidden_params(self) -> dict[str, builtins.object]:
return self._hidden_params
@hidden_params.setter
def hidden_params(self, hidden_params: dict[str, builtins.object]) -> None: # mutable-ok: API requires mutation
def hidden_params(self, hidden_params: dict[str, builtins.object]) -> None:
self._hidden_params = hidden_params
def set_provider_native_response(self, native_response: Mapping[str, builtins.object]) -> None:

8
litellm/llms/base_llm/sandbox/transformation.py
View file

@ -29,11 +29,11 @@ class ContainerHandle(LiteLLMPydanticObjectBase):
_hidden_params: dict = PrivateAttr(default_factory=dict)
@property
def hidden_params(self) -> dict[str, builtins.object]: # mutable-ok: API requires mutation
def hidden_params(self) -> dict[str, builtins.object]:
return self._hidden_params
@hidden_params.setter
def hidden_params(self, hidden_params: dict[str, builtins.object]) -> None: # mutable-ok: API requires mutation
def hidden_params(self, hidden_params: dict[str, builtins.object]) -> None:
self._hidden_params = hidden_params
@ -52,11 +52,11 @@ class CodeExecutionResult(LiteLLMPydanticObjectBase):
_hidden_params: dict = PrivateAttr(default_factory=dict)
@property
def hidden_params(self) -> dict[str, builtins.object]: # mutable-ok: API requires mutation
def hidden_params(self) -> dict[str, builtins.object]:
return self._hidden_params
@hidden_params.setter
def hidden_params(self, hidden_params: dict[str, builtins.object]) -> None: # mutable-ok: API requires mutation
def hidden_params(self, hidden_params: dict[str, builtins.object]) -> None:
self._hidden_params = hidden_params

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

@ -79,11 +79,11 @@ class SearchResponse(LiteLLMPydanticObjectBase):
_hidden_params: dict = PrivateAttr(default_factory=dict)
@property
def hidden_params(self) -> dict[str, builtins.object]: # mutable-ok: API requires mutation
def hidden_params(self) -> dict[str, builtins.object]:
return self._hidden_params
@hidden_params.setter
def hidden_params(self, hidden_params: dict[str, builtins.object]) -> None: # mutable-ok: API requires mutation
def hidden_params(self, hidden_params: dict[str, builtins.object]) -> None:
self._hidden_params = hidden_params
@ -206,12 +206,12 @@ class BaseSearchConfig:
def sign_request(
self,
headers: dict[str, str], # mutable-ok: matches the request header dict every other hook on this base takes
optional_params: dict[str, object], # mutable-ok: matches every other hook on this base
headers: dict[str, str],
optional_params: dict[str, object],
request_data: dict[str, object] | list[dict[str, object]], # mutable-ok: transform_search_request's body
api_base: str,
api_key: str | None = None,
) -> tuple[dict[str, str], bytes | None]: # mutable-ok: the handler passes these headers straight to httpx
) -> tuple[dict[str, str], bytes | None]:
"""
OPTIONAL

8
litellm/llms/bedrock/base_aws_llm.py
View file

@ -170,7 +170,7 @@ def build_web_identity_session_policy() -> WebIdentitySessionPolicy:
def pop_aws_auth_params(
optional_params: MutableMapping[str, object], # mutable-ok: pops the aws_* keys out of the caller's mapping
optional_params: MutableMapping[str, object],
) -> AwsAuthParams:
return AwsAuthParams.model_validate(
MappingProxyType({key: optional_params.pop(key, None) for key in AWS_AUTH_PARAM_KEYS})
@ -792,7 +792,7 @@ class BaseAWSLLM(SignsRequestsWithAWS):
def get_aws_region_name(
self,
optional_params: dict[str, object], # mutable-ok: mirrors override contract
optional_params: dict[str, object],
model: str | None = None,
model_id: str | None = None,
) -> str:
@ -1535,7 +1535,7 @@ class BaseAWSLLM(SignsRequestsWithAWS):
@overload
def _get_boto_credentials_from_optional_params(
self,
optional_params: dict, # mutable-ok: the implementation pops the aws_* keys out of the caller's dict in place
optional_params: dict,
model: str | None = None,
bearer_token: None = None,
) -> Boto3CredentialsInfo: ...
@ -1543,7 +1543,7 @@ class BaseAWSLLM(SignsRequestsWithAWS):
@overload
def _get_boto_credentials_from_optional_params(
self,
optional_params: dict, # mutable-ok: the implementation pops the aws_* keys out of the caller's dict in place
optional_params: dict,
model: str | None = None,
*,
bearer_token: str,

4
litellm/llms/bedrock/chat/agentcore/transformation.py
View file

@ -195,7 +195,7 @@ class AmazonAgentCoreConfig(BaseConfig, BaseAWSLLM):
def get_runtime_session_id(
self,
optional_params: dict[str, object], # mutable-ok: mirrors override contract
optional_params: dict[str, object],
) -> str:
return self._get_runtime_session_id(optional_params)
@ -210,7 +210,7 @@ class AmazonAgentCoreConfig(BaseConfig, BaseAWSLLM):
def get_runtime_user_id(
self,
optional_params: dict[str, object], # mutable-ok: mirrors override contract
optional_params: dict[str, object],
) -> str | None:
return self._get_runtime_user_id(optional_params)

60
litellm/llms/bedrock/chat/chat_completions/transformation.py
View file

@ -224,7 +224,7 @@ class BedrockRuntimeChatCompletionsStreamingHandler(OpenAIChatCompletionStreamin
super().__init__(streaming_response=streaming_response, sync_stream=sync_stream, json_mode=json_mode)
self._splitters: Mapping[int, ReasoningTagSplitter] = MappingProxyType({})
def chunk_parser(self, chunk: dict) -> ModelResponseStream: # mutable-ok: BaseModelResponseIterator signature
def chunk_parser(self, chunk: dict) -> ModelResponseStream:
parsed: Final = super().chunk_parser(chunk)
for choice in parsed.choices:
next_state, reasoning, content = _split_streamed_content(
@ -275,20 +275,20 @@ class AmazonBedrockRuntimeChatCompletionsConfig(OpenAILikeChatConfig):
self,
error_message: str,
status_code: int,
headers: dict[str, object] | httpx.Headers, # mutable-ok: BaseConfig signature
headers: dict[str, object] | httpx.Headers,
) -> BaseLLMException:
return BedrockError(status_code=status_code, message=error_message, headers=headers)
def validate_environment(
self,
headers: dict, # mutable-ok: BaseConfig signature
headers: dict,
model: str,
messages: list[AllMessageValues],
optional_params: dict, # mutable-ok: BaseConfig signature
litellm_params: dict, # mutable-ok: BaseConfig signature
optional_params: dict,
litellm_params: dict,
api_key: str | None = None,
api_base: str | None = None,
) -> dict: # mutable-ok: BaseConfig signature
) -> dict:
return super().validate_environment(
headers=headers,
model=model,
@ -304,8 +304,8 @@ class AmazonBedrockRuntimeChatCompletionsConfig(OpenAILikeChatConfig):
api_base: str | None,
api_key: str | None,
model: str,
optional_params: dict, # mutable-ok: BaseConfig signature
litellm_params: dict, # mutable-ok: BaseConfig signature
optional_params: dict,
litellm_params: dict,
stream: bool | None = None,
) -> str:
if api_base is not None and "chat/completions" in api_base:
@ -328,9 +328,7 @@ class AmazonBedrockRuntimeChatCompletionsConfig(OpenAILikeChatConfig):
return f"{base}/chat/completions"
return f"{base}/openai/v1/chat/completions"
def _params_with_region_from_path(
self, optional_params: dict, model: str | None
) -> dict: # mutable-ok: BaseAWSLLM's region resolver and signer take a plain dict
def _params_with_region_from_path(self, optional_params: dict, model: str | None) -> dict:
region_from_path, _ = split_bedrock_region_path(model or "")
if region_from_path is None or optional_params.get("aws_region_name") is not None:
return optional_params
@ -338,15 +336,15 @@ class AmazonBedrockRuntimeChatCompletionsConfig(OpenAILikeChatConfig):
def sign_request(
self,
headers: dict, # mutable-ok: BaseConfig signature
optional_params: dict, # mutable-ok: BaseConfig signature
request_data: dict, # mutable-ok: BaseConfig signature
headers: dict,
optional_params: dict,
request_data: dict,
api_base: str,
api_key: str | None = None,
model: str | None = None,
stream: bool | None = None,
fake_stream: bool | None = None,
) -> tuple[dict, bytes | None]: # mutable-ok: BaseConfig signature
) -> tuple[dict, bytes | None]:
return self._aws_signer._sign_request( # pyright: ignore[reportPrivateUsage] # BaseAWSLLM has no public signer
service_name="bedrock",
headers=headers,
@ -361,12 +359,12 @@ class AmazonBedrockRuntimeChatCompletionsConfig(OpenAILikeChatConfig):
def map_openai_params(
self,
non_default_params: dict, # mutable-ok: BaseConfig signature
optional_params: dict, # mutable-ok: BaseConfig signature
non_default_params: dict,
optional_params: dict,
model: str,
drop_params: bool,
replace_max_completion_tokens_with_max_tokens: bool = False,
) -> dict: # mutable-ok: BaseConfig signature
) -> dict:
mapped: Final = _PARAMS_DICT_ADAPTER.validate_python(
super().map_openai_params(
non_default_params=non_default_params,
@ -404,9 +402,7 @@ class AmazonBedrockRuntimeChatCompletionsConfig(OpenAILikeChatConfig):
)
)
def _inference_params(
self, optional_params: Mapping[str, object]
) -> dict[str, object]: # mutable-ok: BaseConfig signature of transform_request
def _inference_params(self, optional_params: Mapping[str, object]) -> dict[str, object]:
return {
key: value
for key, value in optional_params.items()
@ -417,10 +413,10 @@ class AmazonBedrockRuntimeChatCompletionsConfig(OpenAILikeChatConfig):
self,
model: str,
messages: list[AllMessageValues], # mutable-ok: BaseConfig signature
optional_params: dict, # mutable-ok: BaseConfig signature
litellm_params: dict, # mutable-ok: BaseConfig signature
headers: dict, # mutable-ok: BaseConfig signature
) -> dict: # mutable-ok: BaseConfig signature
optional_params: dict,
litellm_params: dict,
headers: dict,
) -> dict:
optional_params_view: Final = _PARAMS_DICT_ADAPTER.validate_python(optional_params)
return super().transform_request(
model=split_bedrock_region_path(model)[1],
@ -434,10 +430,10 @@ class AmazonBedrockRuntimeChatCompletionsConfig(OpenAILikeChatConfig):
self,
model: str,
messages: list[AllMessageValues], # mutable-ok: BaseConfig signature
optional_params: dict, # mutable-ok: BaseConfig signature
litellm_params: dict, # mutable-ok: BaseConfig signature
headers: dict, # mutable-ok: BaseConfig signature
) -> dict: # mutable-ok: BaseConfig signature
optional_params: dict,
litellm_params: dict,
headers: dict,
) -> dict:
optional_params_view: Final = _PARAMS_DICT_ADAPTER.validate_python(optional_params)
return await super().async_transform_request(
model=split_bedrock_region_path(model)[1],
@ -453,10 +449,10 @@ class AmazonBedrockRuntimeChatCompletionsConfig(OpenAILikeChatConfig):
raw_response: httpx.Response,
model_response: ModelResponse,
logging_obj: "LiteLLMLoggingObj",
request_data: dict, # mutable-ok: BaseConfig signature
request_data: dict,
messages: list[AllMessageValues], # mutable-ok: BaseConfig signature
optional_params: dict, # mutable-ok: BaseConfig signature
litellm_params: dict, # mutable-ok: BaseConfig signature
optional_params: dict,
litellm_params: dict,
encoding: "tiktoken.Encoding | None",
api_key: str | None = None,
json_mode: bool | None = None,

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

@ -779,8 +779,8 @@ class AWSEventStreamDecoder:
def chunk_parser(
self,
chunk_data: dict[str, object], # mutable-ok: mirrors override contract
) -> GChunk | ModelResponseStream | dict[str, object]: # mutable-ok: mirrors override contract
chunk_data: dict[str, object],
) -> GChunk | ModelResponseStream | dict[str, object]:
return self._chunk_parser(chunk_data)
def iter_bytes(

2
litellm/llms/bedrock/claude_platform/common_utils.py
View file

@ -83,7 +83,7 @@ class BedrockClaudePlatformMixin(BaseAWSLLM):
self,
error_message: str,
status_code: int,
headers: dict[str, object] | httpx.Headers, # mutable-ok: base passes response headers as a dict
headers: dict[str, object] | httpx.Headers,
) -> BedrockError:
return BedrockError(status_code=status_code, message=error_message, headers=headers)

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

@ -253,7 +253,7 @@ def _bedrock_model_supports(model: str, key: str) -> bool:
def apply_bedrock_invoke_structured_output(
model: str,
request_body: dict[str, object], # mutable-ok: edited in place like siblings
request_body: dict[str, object],
) -> None:
"""
Route Anthropic structured-output params to what the Bedrock model supports.
@ -295,7 +295,7 @@ def apply_bedrock_invoke_structured_output(
def strip_unsupported_bedrock_invoke_output_config_keys(
model: str,
request_body: dict[str, object], # mutable-ok: edited in place like siblings
request_body: dict[str, object],
) -> None:
"""
Drop ``output_config`` keys the Bedrock model does not accept.

6
litellm/llms/bedrock/embed/amazon_nova_transformation.py
View file

@ -207,11 +207,11 @@ class AmazonNovaEmbeddingConfig:
def transform_request(
self,
input: str,
inference_params: dict[str, object], # mutable-ok: mirrors override contract
inference_params: dict[str, object],
async_invoke_route: bool = False,
model_id: str | None = None,
output_s3_uri: str | None = None,
) -> dict[str, object]: # mutable-ok: mirrors override contract
) -> dict[str, object]:
return self._transform_request(input, inference_params, async_invoke_route, model_id, output_s3_uri)
def _wrap_async_invoke_request(
@ -372,7 +372,7 @@ class AmazonNovaEmbeddingConfig:
def transform_async_invoke_response(
self,
response: dict[str, object], # mutable-ok: mirrors override contract
response: dict[str, object],
model: str,
) -> EmbeddingResponse:
return self._transform_async_invoke_response(response, model)

2
litellm/llms/bedrock/embed/amazon_titan_g1_transformation.py
View file

@ -63,7 +63,7 @@ class AmazonTitanG1Config:
def transform_request(
self,
input: str,
inference_params: dict[str, object], # mutable-ok: mirrors override contract
inference_params: dict[str, object],
) -> AmazonTitanG1EmbeddingRequest:
return self._transform_request(input, inference_params)

2
litellm/llms/bedrock/embed/amazon_titan_multimodal_transformation.py
View file

@ -55,7 +55,7 @@ class AmazonTitanMultimodalEmbeddingG1Config:
def transform_request(
self,
input: str,
inference_params: dict[str, object], # mutable-ok: mirrors override contract
inference_params: dict[str, object],
) -> AmazonTitanMultimodalEmbeddingRequest:
return self._transform_request(input, inference_params)

2
litellm/llms/bedrock/embed/amazon_titan_v2_transformation.py
View file

@ -80,7 +80,7 @@ class AmazonTitanV2Config:
def transform_request(
self,
input: str,
inference_params: dict[str, object], # mutable-ok: mirrors override contract
inference_params: dict[str, object],
) -> AmazonTitanV2EmbeddingRequest:
return self._transform_request(input, inference_params)

2
litellm/llms/bedrock/embed/cohere_transformation.py
View file

@ -46,6 +46,6 @@ class BedrockCohereEmbeddingConfig:
self,
model: str,
input: list[str], # mutable-ok: mirrors override contract
inference_params: dict[str, object], # mutable-ok: mirrors override contract
inference_params: dict[str, object],
) -> CohereEmbeddingRequest:
return self._transform_request(model, input, inference_params)

4
litellm/llms/bedrock/embed/embedding.py
View file

@ -67,14 +67,14 @@ class BedrockEmbedding(BaseAWSLLM):
@overload
def _load_credentials(
self,
optional_params: dict, # mutable-ok: the implementation pops the aws_* keys out of the caller's dict in place
optional_params: dict,
bearer_token: None = None,
) -> tuple[Credentials, str]: ...
@overload
def _load_credentials(
self,
optional_params: dict, # mutable-ok: the implementation pops the aws_* keys out of the caller's dict in place
optional_params: dict,
bearer_token: str,
) -> tuple[None, str]: ...

4
litellm/llms/bedrock/embed/twelvelabs_marengo_transformation.py
View file

@ -275,7 +275,7 @@ class TwelveLabsMarengoEmbeddingConfig:
def transform_request(
self,
input: str,
inference_params: dict[str, object], # mutable-ok: mirrors override contract
inference_params: dict[str, object],
async_invoke_route: bool = False,
model_id: str | None = None,
output_s3_uri: str | None = None,
@ -390,7 +390,7 @@ class TwelveLabsMarengoEmbeddingConfig:
def transform_async_invoke_response(
self,
response: dict[str, object], # mutable-ok: mirrors override contract
response: dict[str, object],
model: str,
) -> EmbeddingResponse:
return self._transform_async_invoke_response(response, model)

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

@ -1356,9 +1356,7 @@ class BedrockFilesConfig(BaseAWSLLM, BaseFilesConfig):
url: Final = f"{target.endpoint_url}/{bucket_name}/"
listing_query: Final = _listing_query(configured_prefix, purpose)
continuation_query: Final = (("continuation-token", continuation_token),) if continuation_token else ()
query: Final[dict[str, str]] = dict( # mutable-ok: the base files contract returns the query as a dict
listing_query + continuation_query
)
query: Final[dict[str, str]] = dict(listing_query + continuation_query)
signed_headers: Final = self._sign_s3_request_without_body(
method="GET",
api_base=f"{url}?{urlencode(query, quote_via=quote, safe='')}",

2
litellm/llms/bedrock/image_edit/amazon_nova_canvas_image_edit_transformation.py
View file

@ -240,7 +240,7 @@ class BedrockAmazonNovaCanvasImageEditConfig(BaseImageEditConfig):
self,
error_message: str,
status_code: int,
headers: dict[str, object] | httpx.Headers, # mutable-ok: base passes response headers as a dict
headers: dict[str, object] | httpx.Headers,
) -> BedrockError:
return BedrockError(status_code=status_code, message=error_message, headers=headers)

2
litellm/llms/bedrock/image_edit/stability_transformation.py
View file

@ -97,7 +97,7 @@ class BedrockStabilityImageEditConfig(BaseImageEditConfig):
self,
error_message: str,
status_code: int,
headers: dict[str, object] | httpx.Headers, # mutable-ok: base passes response headers as a dict
headers: dict[str, object] | httpx.Headers,
) -> BedrockError:
return BedrockError(status_code=status_code, message=error_message, headers=headers)

2
litellm/llms/bedrock/messages/invoke_transformations/anthropic_claude3_transformation.py
View file

@ -92,7 +92,7 @@ class AmazonAnthropicClaudeMessagesConfig(
self,
error_message: str,
status_code: int,
headers: dict[str, object] | httpx.Headers, # mutable-ok: base passes response headers as a dict
headers: dict[str, object] | httpx.Headers,
) -> BedrockError:
return BedrockError(status_code=status_code, message=error_message, headers=headers)

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

@ -129,7 +129,7 @@ class BedrockPassthroughConfig(BaseAWSLLM, BedrockModelInfo, BedrockEventStreamD
self,
error_message: str,
status_code: int,
headers: dict[str, object] | httpx.Headers, # mutable-ok: base passes response headers as a dict
headers: dict[str, object] | httpx.Headers,
) -> BedrockError:
return BedrockError(status_code=status_code, message=error_message, headers=headers)

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

@ -89,7 +89,7 @@ def _should_log_event(openai_message: Mapping[str, object]) -> bool:
class RealtimeClientWebSocket(Protocol):
"""The client-facing websocket surface the realtime bridge talks to."""
scope: MutableMapping[str, object] # mutable-ok: the ASGI scope is the per-connection state store
scope: MutableMapping[str, object]
async def receive_text(self) -> str: ...

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

@ -128,7 +128,7 @@ class BedrockRealtimeConfig(BaseRealtimeConfig):
self,
error_message: str,
status_code: int,
headers: dict[str, object] | httpx.Headers, # mutable-ok: base passes response headers as a dict
headers: dict[str, object] | httpx.Headers,
) -> BedrockError:
return BedrockError(status_code=status_code, message=error_message, headers=headers)

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

@ -117,6 +117,6 @@ class BedrockRerankConfig:
def transform_response(
self,
response: dict[str, object], # mutable-ok: mirrors override contract
response: dict[str, object],
) -> RerankResponse:
return self._transform_response(response)

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

@ -157,7 +157,7 @@ def inline_remote_image_urls(
def _without_disabled_reasoning_effort(
params: Mapping[str, object], model: str, drop_params: bool
) -> dict[str, object]: # mutable-ok: becomes the map_openai_params return value
) -> dict[str, object]:
reasoning: Final = params.get("reasoning")
effort: Final = reasoning.get("effort") if isinstance(reasoning, Mapping) else None
if not isinstance(reasoning, Mapping) or not isinstance(effort, str):
@ -220,7 +220,7 @@ class BedrockOpenAIResponsesConfig(BaseAWSLLM, OpenAIResponsesAPIConfig):
def get_complete_url(
self,
api_base: str | None,
litellm_params: dict, # mutable-ok: signature fixed by the BaseResponsesAPIConfig override contract
litellm_params: dict,
) -> str:
region: Final = self._get_aws_region_name(optional_params=litellm_params, model=None)
override: Final = (
@ -244,10 +244,10 @@ class BedrockOpenAIResponsesConfig(BaseAWSLLM, OpenAIResponsesAPIConfig):
def validate_environment(
self,
headers: dict, # mutable-ok: signature fixed by the BaseResponsesAPIConfig override contract
headers: dict,
model: str,
litellm_params: GenericLiteLLMParams | None,
) -> dict: # mutable-ok: signature fixed by the BaseResponsesAPIConfig override contract
) -> dict:
api_key: Final = litellm_params.api_key if litellm_params is not None else None
bearer: Final = resolve_bedrock_bearer_token(api_key)
if not bearer:
@ -256,15 +256,15 @@ class BedrockOpenAIResponsesConfig(BaseAWSLLM, OpenAIResponsesAPIConfig):
def sign_request(
self,
headers: dict, # mutable-ok: signature fixed by the BaseResponsesAPIConfig override contract
optional_params: dict, # mutable-ok: same
request_data: dict, # mutable-ok: same
headers: dict,
optional_params: dict,
request_data: dict,
api_base: str,
api_key: str | None = None,
model: str | None = None,
stream: bool | None = None,
fake_stream: bool | None = None,
) -> "tuple[dict, bytes | None]": # mutable-ok: signature fixed by the override contract
) -> "tuple[dict, bytes | None]":
if resolve_bedrock_bearer_token(api_key):
# Bedrock API keys are Bearer credentials; SigV4 on top would be wrong.
return headers, None
@ -284,7 +284,7 @@ class BedrockOpenAIResponsesConfig(BaseAWSLLM, OpenAIResponsesAPIConfig):
response_api_optional_params: ResponsesAPIOptionalRequestParams,
model: str,
drop_params: bool,
) -> dict: # mutable-ok: signature fixed by the override contract
) -> dict:
mapped: Final = super().map_openai_params(
response_api_optional_params=response_api_optional_params, model=model, drop_params=drop_params
)
@ -316,10 +316,10 @@ class BedrockOpenAIResponsesConfig(BaseAWSLLM, OpenAIResponsesAPIConfig):
self,
model: str,
input: "str | ResponseInputParam",
response_api_optional_request_params: dict, # mutable-ok: signature fixed by the override contract
response_api_optional_request_params: dict,
litellm_params: GenericLiteLLMParams,
headers: dict, # mutable-ok: same
) -> dict: # mutable-ok: same
headers: dict,
) -> dict:
inlined: Final = MappingProxyType({url: self.fetch_image(url) for url in collect_remote_image_urls(input)})
return self._transform_inlined_request(
model=model,
@ -333,10 +333,10 @@ class BedrockOpenAIResponsesConfig(BaseAWSLLM, OpenAIResponsesAPIConfig):
self,
model: str,
input: "str | ResponseInputParam",
response_api_optional_request_params: dict, # mutable-ok: signature fixed by the override contract
response_api_optional_request_params: dict,
litellm_params: GenericLiteLLMParams,
headers: dict, # mutable-ok: same
) -> dict: # mutable-ok: same
headers: dict,
) -> dict:
remote_urls: Final = collect_remote_image_urls(input)
data_uris: Final = await asyncio.gather(*(self.async_fetch_image(url) for url in remote_urls))
return self._transform_inlined_request(
@ -351,10 +351,10 @@ class BedrockOpenAIResponsesConfig(BaseAWSLLM, OpenAIResponsesAPIConfig):
self,
model: str,
input: "str | ResponseInputParam",
response_api_optional_request_params: dict, # mutable-ok: signature fixed by the override contract
response_api_optional_request_params: dict,
litellm_params: GenericLiteLLMParams,
headers: dict, # mutable-ok: same
) -> dict: # mutable-ok: same
headers: dict,
) -> dict:
normalized_input, rewritten_types = normalize_codex_input_items(input)
if rewritten_types:
verbose_logger.warning(

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

@ -168,11 +168,11 @@ class AgentCoreSearchConfig(BaseSearchConfig, BaseAWSLLM):
def validate_environment(
self,
headers: dict, # mutable-ok: BaseSearchConfig hands providers the mutable request header dict
headers: dict,
api_key: str | None = None,
api_base: str | None = None,
**kwargs: object, # kwargs-ok: BaseSearchConfig.validate_environment forwards provider-specific extras
) -> dict: # mutable-ok: the handler passes these headers straight to httpx, which wants a dict
) -> dict:
"""
Set MCP transport headers. Per the MCP Streamable HTTP transport spec,
the client MUST accept both application/json and text/event-stream, and
@ -192,7 +192,7 @@ class AgentCoreSearchConfig(BaseSearchConfig, BaseAWSLLM):
def get_complete_url(
self,
api_base: str | None,
optional_params: dict, # mutable-ok: BaseSearchConfig passes optional params as a dict
optional_params: dict,
data: dict | list[dict] | None = None, # mutable-ok: BaseSearchConfig request bodies are JSON dicts
**kwargs: object, # kwargs-ok: BaseSearchConfig.get_complete_url forwards provider-specific extras
) -> str:
@ -208,9 +208,9 @@ class AgentCoreSearchConfig(BaseSearchConfig, BaseAWSLLM):
def transform_search_request(
self,
query: str | list[str], # mutable-ok: BaseSearchConfig accepts a list of queries
optional_params: dict, # mutable-ok: BaseSearchConfig passes optional params as a dict
optional_params: dict,
**kwargs: object, # kwargs-ok: BaseSearchConfig.transform_search_request forwards provider-specific extras
) -> dict: # mutable-ok: the JSON-RPC body is serialized as a JSON object
) -> dict:
"""
Transform Search request to an MCP tools/call request.
@ -252,12 +252,12 @@ class AgentCoreSearchConfig(BaseSearchConfig, BaseAWSLLM):
def sign_request(
self,
headers: dict[str, str], # mutable-ok: BaseSearchConfig hands providers the mutable request header dict
optional_params: dict[str, object], # mutable-ok: BaseSearchConfig passes optional params as a dict
headers: dict[str, str],
optional_params: dict[str, object],
request_data: dict[str, object] | list[dict[str, object]], # mutable-ok: request bodies are JSON dicts
api_base: str,
api_key: str | None = None,
) -> tuple[dict[str, str], bytes | None]: # mutable-ok: BaseSearchConfig.sign_request returns httpx headers
) -> tuple[dict[str, str], bytes | None]:
"""
Authenticate the MCP request.
@ -451,7 +451,7 @@ class AgentCoreSearchConfig(BaseSearchConfig, BaseAWSLLM):
self,
error_message: str,
status_code: int,
headers: dict, # mutable-ok: BaseSearchConfig.get_error_class takes the response headers as a dict
headers: dict,
) -> Exception:
return BedrockError(
status_code=status_code,

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

@ -45,7 +45,7 @@ class BedrockVectorStoreConfig(BaseVectorStoreConfig, BaseAWSLLM):
self,
error_message: str,
status_code: int,
headers: dict[str, object] | httpx.Headers, # mutable-ok: base passes response headers as a dict
headers: dict[str, object] | httpx.Headers,
) -> BedrockError:
return BedrockError(status_code=status_code, message=error_message, headers=headers)

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

@ -55,7 +55,7 @@ class BedrockMantlePassthroughConfig(BedrockPassthroughConfig):
model: str,
custom_llm_provider: str,
httpx_response: Response,
request_data: dict, # mutable-ok: mirrors the inherited BedrockPassthroughConfig signature
request_data: dict,
logging_obj: Logging,
endpoint: str,
) -> Optional["CostResponseTypes"]:

11
litellm/llms/claude_code/harness/transformation.py
View file

@ -164,7 +164,7 @@ def _message_blocks(event: Mapping[str, object]) -> Sequence[object]:
message: Final = event.get("message")
content: Final = message.get("content") if isinstance(message, Mapping) else None
if isinstance(content, str):
return ({"type": "text", "text": content},) # mutable-ok: JSON content block, like the stream's
return ({"type": "text", "text": content},)
return content if isinstance(content, list) else ()
@ -182,8 +182,7 @@ def _assistant_block_events(block: Mapping[str, Any], state: ClaudeCodeStreamSta
id=str(block.get("id", "")),
name=normalize_tool_name(native, NATIVE_TO_NORMALIZED),
native_name=native,
input=block.get("input")
or {}, # mutable-ok: ToolCall.input is a dict field; empty default for a missing input
input=block.get("input") or {},
builtin=not native.startswith("mcp__"),
),
)
@ -236,7 +235,7 @@ def _record_result(event: Mapping[str, Any], state: ClaudeCodeStreamState) -> Se
state.is_error = bool(event.get("is_error", False))
result = event.get("result")
state.result_text = result if isinstance(result, str) else None
state.errors = [str(e) for e in event.get("errors") or ()] # mutable-ok: mirrors the JSON errors array
state.errors = [str(e) for e in event.get("errors") or ()]
state.structured_output = event.get("structured_output")
if event.get("session_id"):
state.session_id = str(event["session_id"])
@ -318,7 +317,7 @@ class ClaudeCodeHarnessConfig(BaseCLIHarnessConfig):
}
)
return HarnessSessionSetup(
persisted_dirs=[("projects", "claude_code/projects")], # mutable-ok: tests compare to a list
persisted_dirs=[("projects", "claude_code/projects")],
skills_dir="skills",
env=env,
)
@ -335,7 +334,7 @@ class ClaudeCodeHarnessConfig(BaseCLIHarnessConfig):
schema = ctx.output.model_json_schema() if ctx.output is not None else None
system_prompt: Final = build_system_prompt(ctx.instructions, schema)
disallowed: Final = native_tool_names(ctx.disable_tools, NORMALIZED_TO_NATIVE)
config: Final = dict(options.config) # mutable-ok: json.dumps needs a plain dict
config: Final = dict(options.config)
settings: Final = json.dumps(config) if config else None
argv: Final = (
CLAUDE_BINARY,

6
litellm/llms/codex/harness/transformation.py
View file

@ -83,7 +83,7 @@ def _tool_input(item: Mapping[str, Any]) -> tuple[str, str, Mapping[str, object]
if item_type == "command_execution":
return "bash", "command_execution", MappingProxyType({"command": item.get("command", "")}), True
if item_type == "file_change":
changes: Final = list(item.get("changes") or ()) # mutable-ok: JSON array, as codex reports it
changes: Final = list(item.get("changes") or ())
return "edit", "apply_patch", MappingProxyType({"changes": changes}), True
if item_type == "web_search":
return "web_search", "web_search", MappingProxyType({"query": item.get("query", "")}), True
@ -188,7 +188,7 @@ def _config_override(key: object, value: object) -> str:
def config_overrides(config: Mapping[str, object]) -> Sequence[str]:
"""`-c` override strings for CodexOptions.config, rejecting managed keys."""
overrides: Final = (_config_override(key, value) for key, value in config.items())
return list(overrides) # mutable-ok: public helper; tests compare to a list
return list(overrides)
def _flag_pairs(flag: str, values: Sequence[str]) -> tuple[str, ...]:
@ -232,7 +232,7 @@ class CodexHarnessConfig(BaseCLIHarnessConfig):
)
return HarnessSessionSetup(
files=files,
persisted_dirs=[("sessions", "codex/sessions")], # mutable-ok: tests compare to a list
persisted_dirs=[("sessions", "codex/sessions")],
skills_dir="skills",
env=MappingProxyType({**options.env, CODEX_TOKEN_ENV: ctx.endpoint.token, "CODEX_HOME": private_dir}),
)

6
litellm/llms/cohere/embed/v1_transformation.py
View file

@ -72,7 +72,7 @@ class CohereEmbeddingConfig:
self,
model: str,
input: list[str], # mutable-ok: mirrors override contract
inference_params: dict[str, object], # mutable-ok: mirrors override contract
inference_params: dict[str, object],
) -> CohereEmbeddingRequestWithModel:
return self._transform_request(model, input, inference_params)
@ -136,7 +136,7 @@ class CohereEmbeddingConfig:
response: httpx.Response,
api_key: str | None,
logging_obj: LiteLLMLoggingObj,
data: dict[str, object] | CohereEmbeddingRequest, # mutable-ok: mirrors override contract
data: dict[str, object] | CohereEmbeddingRequest,
model_response: EmbeddingResponse,
model: str,
encoding: _SupportsEncode,
@ -202,7 +202,7 @@ class CohereEmbeddingConfig:
def populate_embedding_response(
self,
response_json: dict[str, object], # mutable-ok: mirrors override contract
response_json: dict[str, object],
model_response: EmbeddingResponse,
model: str,
encoding: _SupportsEncode,

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

@ -283,26 +283,26 @@ class _MediaUploadKwargs(TypedDict, total=False):
class _AsyncFilesEnvironmentValidator(Protocol):
async def avalidate_environment(
self,
headers: dict, # mutable-ok: mirrors the sync validate_environment contract this overrides
headers: dict,
model: str,
messages: list, # mutable-ok: mirrors the sync validate_environment contract this overrides
optional_params: dict, # mutable-ok: mirrors the sync validate_environment contract this overrides
litellm_params: dict, # mutable-ok: mirrors the sync validate_environment contract this overrides
optional_params: dict,
litellm_params: dict,
api_key: str | None = None,
api_base: str | None = None,
) -> dict: ... # mutable-ok: mirrors the sync validate_environment contract this overrides
) -> dict: ...
async def _avalidate_files_environment(
provider_config: BaseFilesConfig | BaseBatchesConfig,
*,
headers: dict, # mutable-ok: mirrors the sync validate_environment contract this overrides
headers: dict,
model: str,
messages: list, # mutable-ok: mirrors the sync validate_environment contract this overrides
optional_params: dict, # mutable-ok: mirrors the sync validate_environment contract this overrides
litellm_params: dict, # mutable-ok: mirrors the sync validate_environment contract this overrides
optional_params: dict,
litellm_params: dict,
api_key: str | None,
) -> dict: # mutable-ok: mirrors the sync validate_environment contract this overrides
) -> dict:
"""Await the provider's async credential hook when it has one (e.g. Anthropic's workload
identity token exchange); otherwise offload the sync hook to a worker thread. Either way
the caller, an async file handler, never blocks the event loop on it."""
@ -736,8 +736,8 @@ class BaseLLMHTTPHandler:
)
def sign_and_log(
transformed: dict[str, object], # mutable-ok: async_completion takes dict
) -> tuple[dict[str, object], dict[str, object], bytes | None]: # mutable-ok: async_completion takes dict
transformed: dict[str, object],
) -> tuple[dict[str, object], dict[str, object], bytes | None]:
data: Final = (
{
**transformed,
@ -778,8 +778,8 @@ class BaseLLMHTTPHandler:
return data, signed[0], signed[1]
def dispatch_async(
data: dict[str, object], # mutable-ok: async_completion takes dict
signed_headers: dict[str, object], # mutable-ok: async_completion takes dict
data: dict[str, object],
signed_headers: dict[str, object],
signed_json_body: bytes | None,
) -> Coroutine[object, object, ModelResponse | CustomStreamWrapper]:
async_client: Final = client if isinstance(client, AsyncHTTPHandler) else None
@ -3569,9 +3569,9 @@ class BaseLLMHTTPHandler:
self,
*,
create_file_data: CreateFileRequest,
litellm_params: dict, # mutable-ok: mirrors the create_file contract this dispatches for
litellm_params: dict,
provider_config: BaseFilesConfig,
headers: dict, # mutable-ok: mirrors the create_file contract this dispatches for
headers: dict,
api_base: str | None,
api_key: str | None,
logging_obj: LiteLLMLoggingObj,
@ -4046,9 +4046,9 @@ class BaseLLMHTTPHandler:
self,
*,
create_batch_data: "CreateBatchRequest",
litellm_params: dict, # mutable-ok: mirrors the create_batch contract this dispatches for
litellm_params: dict,
provider_config: "BaseBatchesConfig",
headers: dict, # mutable-ok: mirrors the create_batch contract this dispatches for
headers: dict,
api_base: str | None,
api_key: str | None,
logging_obj: "LiteLLMLoggingObj",
@ -4980,8 +4980,8 @@ class BaseLLMHTTPHandler:
first_page: httpx.Response,
provider_config: BaseFilesConfig,
logging_obj: LiteLLMLoggingObj,
litellm_params: dict, # mutable-ok: handed to the files contract, which types it as a dict
headers: dict, # mutable-ok: handed to validate_environment, which types it as a dict
litellm_params: dict,
headers: dict,
client: HTTPHandler,
timeout: float | httpx.Timeout | None,
) -> Iterator[list[OpenAIFileObject]]: # mutable-ok: each page arrives as the list the files contract returns
@ -5008,8 +5008,8 @@ class BaseLLMHTTPHandler:
first_page: httpx.Response,
provider_config: BaseFilesConfig,
logging_obj: LiteLLMLoggingObj,
litellm_params: dict, # mutable-ok: handed to the files contract, which types it as a dict
headers: dict, # mutable-ok: handed to validate_environment, which types it as a dict
litellm_params: dict,
headers: dict,
client: AsyncHTTPHandler,
timeout: float | httpx.Timeout | None,
) -> AsyncIterator[list[OpenAIFileObject]]: # mutable-ok: each page arrives as the list the files contract returns
@ -5034,9 +5034,9 @@ class BaseLLMHTTPHandler:
def _next_listing_page_headers(
self,
provider_config: BaseFilesConfig,
headers: dict, # mutable-ok: handed to validate_environment, which types it as a dict
litellm_params: dict, # mutable-ok: handed to the files contract, which types it as a dict
) -> dict: # mutable-ok: validate_environment returns the header dict the files contract types
headers: dict,
litellm_params: dict,
) -> dict:
return provider_config.validate_environment(
api_key=litellm_params.get("api_key"),
headers=headers,
@ -5050,9 +5050,9 @@ class BaseLLMHTTPHandler:
self,
latest_page: httpx.Response,
provider_config: BaseFilesConfig,
litellm_params: dict, # mutable-ok: handed to the files contract, which types it as a dict
litellm_params: dict,
listed_count: int,
) -> tuple[str, dict[str, str]] | None: # mutable-ok: the base files contract returns the query as a dict
) -> tuple[str, dict[str, str]] | None:
if listed_count >= MAX_FILE_LIST_LIMIT:
return None
return provider_config.transform_list_files_next_request(

18
litellm/llms/deepagents/harness/sandbox_backend.py
View file

@ -314,7 +314,7 @@ class SandboxBackend(SandboxBackendProtocol): # pyright: ignore[reportUntypedBa
except (InvalidGlobPatternError, ValueError, SandboxError) as e:
return GlobResult(error=str(e), matches=None)
if done.exit_code != 0 and not done.stdout:
return GlobResult(matches=[]) # mutable-ok: deepagents GlobResult.matches is typed list[FileInfo]
return GlobResult(matches=[])
matches = sorted(
(
FileInfo(path=self.to_virtual(real), is_dir=False)
@ -353,9 +353,7 @@ class SandboxBackend(SandboxBackendProtocol): # pyright: ignore[reportUntypedBa
return GrepResult(error=f"Path '{path or '/'}': {e}")
if done.exit_code not in (0, 1) and not done.stdout:
return GrepResult(error=f"Path '{path or '/'}': {done.stderr.strip() or 'grep failed'}")
matches = list( # mutable-ok: GrepResult.matches is list[GrepMatch]
self._grep_matches(done.stdout, root, include)
)
matches = list(self._grep_matches(done.stdout, root, include))
if max_count is not None and len(matches) > max_count:
return GrepResult(matches=matches[:max_count], truncated=True)
return GrepResult(matches=matches)
@ -385,9 +383,7 @@ class SandboxBackend(SandboxBackendProtocol): # pyright: ignore[reportUntypedBa
self,
files: list[tuple[str, bytes]], # mutable-ok: signature fixed by deepagents BackendProtocol
) -> list[FileUploadResponse]: # mutable-ok: return type fixed by deepagents BackendProtocol
return [ # mutable-ok: BackendProtocol returns a list
await self._upload_one(path, data) for path, data in files
]
return [await self._upload_one(path, data) for path, data in files]
def upload_files(
self,
@ -407,7 +403,7 @@ class SandboxBackend(SandboxBackendProtocol): # pyright: ignore[reportUntypedBa
self,
paths: list[str], # mutable-ok: signature fixed by deepagents BackendProtocol
) -> list[FileDownloadResponse]: # mutable-ok: return type fixed by deepagents BackendProtocol
return [await self._download_one(path) for path in paths] # mutable-ok: BackendProtocol returns a list
return [await self._download_one(path) for path in paths]
def download_files(
self,
@ -464,11 +460,7 @@ class ToolFilterMiddleware(AgentMiddleware): # pyright: ignore[reportUntypedBas
return _FILTER_MIDDLEWARE_NAME
def _filtered(self, request: ModelRequest) -> ModelRequest:
return request.override(
tools=[ # mutable-ok: ModelRequest.tools is a list
t for t in request.tools if _tool_name(t) not in self._blocked
]
)
return request.override(tools=[t for t in request.tools if _tool_name(t) not in self._blocked])
def wrap_model_call(
self, request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse]

39
litellm/llms/deepagents/harness/transformation.py
View file

@ -73,18 +73,18 @@ _APPROVAL_DECISIONS: Final = ("approve", "reject")
def chat_model_kwargs(
ctx: SessionContext,
) -> dict[str, Any]: # mutable-ok: ChatLiteLLM constructor kwargs, splatted as **kwargs
) -> dict[str, Any]:
"""ChatLiteLLM constructor kwargs for gateway or SDK mode."""
if not ctx.model:
raise ValueError("Harness.DEEPAGENTS needs model=")
if ctx.gateway is not None:
return { # mutable-ok: ChatLiteLLM constructor kwargs, splatted as **kwargs
return {
"model": f"litellm_proxy/{ctx.model}",
"api_base": ctx.gateway.api_base,
"api_key": ctx.gateway.api_key,
"extra_headers": gateway_headers(ctx),
}
return {"model": ctx.model, "api_key": ctx.api_key, "api_base": ctx.api_base} # mutable-ok: ChatLiteLLM kwargs
return {"model": ctx.model, "api_key": ctx.api_key, "api_base": ctx.api_base}
def native_tool_name(name: str) -> str:
@ -105,16 +105,11 @@ def blocked_tools(permissions: str, disable_tools: Sequence[str]) -> frozenset[s
return disabled
def interrupt_config(
permissions: str, blocked: frozenset[str]
) -> dict[str, Any] | None: # mutable-ok: deepagents create_deep_agent(interrupt_on=) takes a dict
def interrupt_config(permissions: str, blocked: frozenset[str]) -> dict[str, Any] | None:
"""interrupt_on for permissions='ask': approve/reject every mutating built-in."""
if permissions != "ask":
return None
return { # mutable-ok: deepagents interrupt_on config (dict of InterruptOnConfig with list allowed_decisions)
name: {"allowed_decisions": list(_APPROVAL_DECISIONS)} # mutable-ok: deepagents InterruptOnConfig shape
for name in sorted(APPROVAL_TOOLS - blocked)
}
return {name: {"allowed_decisions": list(_APPROVAL_DECISIONS)} for name in sorted(APPROVAL_TOOLS - blocked)}
def recursion_limit(ctx: SessionContext) -> int:
@ -162,12 +157,12 @@ def stream_events(
) -> list[Event]: # mutable-ok: returns a list; existing callers/tests compare it to list literals
"""Text / Reasoning deltas for one streamed message chunk."""
if getattr(message, "type", None) not in ("AIMessageChunk", "ai"):
return [] # mutable-ok: returns a list; existing callers/tests compare it to list literals
return []
reasoning = reasoning_text(message)
text = content_text(getattr(message, "content", ""))
reasoning_events: tuple[Event, ...] = (Reasoning(delta=reasoning),) if reasoning else ()
text_events: tuple[Event, ...] = (Text(delta=text),) if text else ()
return [ # mutable-ok: returns a list; existing callers/tests compare it to list literals
return [
*reasoning_events,
*text_events,
]
@ -180,7 +175,7 @@ def tool_call_event(call: Mapping[str, object]) -> ToolCall:
id=str(call.get("id") or ""),
name=normalized_tool_name(native),
native_name=native,
input=dict(args) if isinstance(args, Mapping) else {"args": args}, # mutable-ok: ToolCall.input is a dict
input=dict(args) if isinstance(args, Mapping) else {"args": args},
builtin=native in BUILTIN_TOOLS,
)
@ -199,8 +194,8 @@ def update_events(
) -> list[Event]: # mutable-ok: returns a list; existing callers/tests compare it to list literals
"""ToolCall / ToolResult events from one `updates` stream chunk (node -> state delta)."""
if not isinstance(update, Mapping):
return [] # mutable-ok: returns a list; existing callers/tests compare it to list literals
return list( # mutable-ok: returns a list; existing callers/tests compare it to list literals
return []
return list(
itertools.chain.from_iterable(_message_events(message, skip_tools) for message in _node_messages(update))
)
@ -225,10 +220,10 @@ def interrupts_in(
update: object,
) -> list[Any]: # mutable-ok: returns a list; existing callers/tests compare it to list literals
if not isinstance(update, Mapping):
return [] # mutable-ok: returns a list; existing callers/tests compare it to list literals
return []
found = update.get("__interrupt__")
items = tuple(found) if isinstance(found, (list, tuple)) else ()
return list(items) # mutable-ok: list return; callers/tests compare to lists
return list(items)
def final_ai_text(messages: Sequence[object]) -> str:
@ -254,16 +249,16 @@ def approval_requests(
) -> list[Mapping[str, Any]]: # mutable-ok: returns a list; existing callers/tests compare it to list literals
"""action_requests of a HumanInTheLoopMiddleware interrupt payload."""
if not isinstance(interrupt_value, Mapping):
return [] # mutable-ok: returns a list; existing callers/tests compare it to list literals
return []
requests = interrupt_value.get("action_requests")
kept = tuple(r for r in requests if isinstance(r, Mapping)) if isinstance(requests, list) else ()
return list(kept) # mutable-ok: list return; callers/tests compare to lists
return list(kept)
def decision(allowed: bool, reason: str) -> dict[str, Any]: # mutable-ok: LangGraph resume payload (HITL decision dict)
def decision(allowed: bool, reason: str) -> dict[str, Any]:
if allowed:
return {"type": "approve"} # mutable-ok: LangGraph resume payload (HITL decision dict)
return { # mutable-ok: LangGraph HITL decision
return {"type": "approve"}
return {
"type": "reject",
"message": reason or "The user denied this tool call.",
}

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