fix(bedrock): adapt fallback tests and legacy lint contracts

This commit is contained in:
Joshua Valluru 2026-10-20 18:15:18 -07:00
parent 70b4a8c0ce
commit 6d484fe297
15 changed files with 222 additions and 1619 deletions

View file

@ -1383,7 +1383,11 @@ def schema_has_unresolvable_ref(schema: Mapping[str, object]) -> bool:
if not _refs_in(schema) <= definitions.keys():
return True
try:
tuple(TopologicalSorter({pointer: _refs_in(body) for pointer, body in definitions.items()}).static_order())
tuple(
TopologicalSorter(
MappingProxyType({pointer: _refs_in(body) for pointer, body in definitions.items()})
).static_order()
)
except CycleError:
return True
return False

View file

@ -5129,7 +5129,9 @@ def _bedrock_tools_pt(tools: list, model: str | None = None) -> list[BedrockTool
description = name
defs = parameters.pop("$defs", {})
defs_copy = copy.deepcopy(defs) if isinstance(defs, dict) else {}
defs_copy = (
copy.deepcopy(defs) if isinstance(defs, dict) else {} # mutable-ok: unpack_defs expands references in place
)
# Expand $ref references in parameters using the definitions
# Note: We don't pre-flatten defs as that causes exponential memory growth
# with circular references (see issue #19098). unpack_defs handles nested

View file

@ -615,7 +615,9 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
enforced_min_items: Final = (
isinstance(min_items, int) and not isinstance(min_items, bool) and min_items in (0, 1)
)
unsupported_fields: Final = set(constraint_labels) - ({"minItems"} if enforced_min_items else set())
unsupported_fields: Final = frozenset(constraint_labels) - (
frozenset(("minItems",)) if enforced_min_items else frozenset()
)
# Build description additions from removed constraints. Iterating
# constraint_labels (not the set) keeps the note order deterministic across
@ -1362,7 +1364,9 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
@staticmethod
def _defs_unpacked(json_schema: Mapping[str, object]) -> Mapping[str, object]:
unpacked: Final = copy.deepcopy(dict(json_schema))
unpacked: Final = copy.deepcopy(
dict(json_schema) # mutable-ok: unpack_defs expands this private schema copy in place
)
defs: Final = unpacked.pop("$defs", unpacked.pop("definitions", None))
if isinstance(defs, dict) and defs:
unpack_defs(unpacked, defs, max_inlined_bytes=DEFS_MAX_INLINED_BYTES)
@ -1380,7 +1384,9 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
or schema_grammar_branch_count(json_schema) > NATIVE_OUTPUT_MAX_GRAMMAR_BRANCHES
):
return None
return AnthropicConfig.filter_anthropic_output_schema(dict(json_schema))
return AnthropicConfig.filter_anthropic_output_schema(
dict(json_schema) # mutable-ok: schema filter mutates its input
)
def uses_native_structured_output(self, model: str) -> bool:
return AnthropicConfig._supports_model_capability(
@ -1397,14 +1403,16 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
json_schema is not None
and schema_has_open_object(json_schema)
and not self.is_thinking_requested(request_params)
and not self.is_thinking_enabled(dict(request_params))
and not self.is_thinking_enabled(dict(request_params)) # mutable-ok: legacy helper requires a dict
and not request_params.get("tools")
and not AnthropicModelInfo.forced_tool_use_unsupported(model)
)
return self.uses_native_structured_output(model) and not open_object_takes_forced_json_tool
def native_output_format_schema(self, json_schema: Mapping[str, object]) -> Mapping[str, object] | None:
return self.filter_anthropic_output_schema(dict(self._defs_unpacked(json_schema)))
return self.filter_anthropic_output_schema(
dict(self._defs_unpacked(json_schema)) # mutable-ok: filter mutates its input
)
def map_response_format_to_anthropic_output_format(self, value: dict | None) -> AnthropicOutputSchema | None:
json_schema: Final = self._extract_json_schema_from_response_format(value)
@ -1413,7 +1421,10 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
native_schema: Final = self.native_output_format_schema(json_schema)
if native_schema is None:
return None
return AnthropicOutputSchema(type="json_schema", schema=dict(native_schema))
return AnthropicOutputSchema(
type="json_schema",
schema=dict(native_schema), # mutable-ok: wire schema is a JSON dict
)
def map_response_format_to_anthropic_tool(
self, value: dict | None, optional_params: dict, is_thinking_enabled: bool

View file

@ -1,4 +1,5 @@
from collections.abc import Mapping
"""
Common base config for all LLM providers
"""

View file

@ -959,7 +959,7 @@ class AmazonConverseConfig(BaseConfig):
if not isinstance(schema, dict):
return schema
result: Final[dict[str, object]] = dict(schema)
result: Final[dict[str, object]] = dict(schema) # mutable-ok: recursively normalize a private JSON schema copy
if is_object_schema(result) and "additionalProperties" not in result:
result["additionalProperties"] = False
@ -978,7 +978,9 @@ class AmazonConverseConfig(BaseConfig):
}
for key in ("anyOf", "allOf", "oneOf"):
if isinstance(variants := result.get(key), list):
result[key] = [AmazonConverseConfig._add_additional_properties_to_schema(item) for item in variants]
result[key] = [ # mutable-ok: schema unions serialize as JSON arrays
AmazonConverseConfig._add_additional_properties_to_schema(item) for item in variants
]
return result
@ -1007,7 +1009,9 @@ class AmazonConverseConfig(BaseConfig):
return None
schema, name, description = (native_format.get(key) for key in ("schema", "name", "description"))
return AmazonConverseConfig._create_output_config_for_response_format(
json_schema={**schema} if isinstance(schema, Mapping) else None,
json_schema={**schema} # mutable-ok: legacy schema normalizer requires a dict
if isinstance(schema, Mapping)
else None,
name=name if isinstance(name, str) else None,
description=description if isinstance(description, str) else None,
)
@ -1305,7 +1309,7 @@ class AmazonConverseConfig(BaseConfig):
# Requires an explicit schema — json_object with no schema falls through
# to the tool-call path below.
output_config: Final = self._create_output_config_for_response_format(
json_schema=dict(native_schema),
json_schema=dict(native_schema), # mutable-ok: legacy schema normalizer requires a dict
name=name,
description=description,
)

View file

@ -12,8 +12,12 @@ from litellm.utils import CustomStreamWrapper
NativeOutputResponse: TypeAlias = ModelResponse | CustomStreamWrapper
NativeOutputAttemptResult: TypeAlias = NativeOutputResponse | Coroutine[object, object, NativeOutputResponse]
NativeOutputAttempt: TypeAlias = Callable[[NativeOutputFallback | None], NativeOutputAttemptResult]
NativeOutputFallbackFactory: TypeAlias = Callable[[], NativeOutputFallback | None]
NativeOutputAttempt: TypeAlias = Callable[
[NativeOutputFallback | None], NativeOutputAttemptResult # mutable-ok: Callable argument type syntax
]
NativeOutputFallbackFactory: TypeAlias = Callable[
[], NativeOutputFallback | None # mutable-ok: Callable argument type syntax
]
def _fallback_for(

View file

@ -247,14 +247,18 @@ _DEPLOYMENT_MODEL_INFO: Final = TypeAdapter(dict[str, object])
def deployment_model_info(litellm_params: Mapping[str, object] | None) -> Mapping[str, object]:
model_info: Final = (litellm_params or {}).get("model_info")
return _DEPLOYMENT_MODEL_INFO.validate_python(model_info) if isinstance(model_info, Mapping) else {}
model_info: Final = litellm_params.get("model_info") if litellm_params is not None else None
return (
_DEPLOYMENT_MODEL_INFO.validate_python(model_info) if isinstance(model_info, Mapping) else MappingProxyType({})
)
def _bedrock_native_output_opt_in(
model: str, custom_llm_provider: str, model_info: Mapping[str, object] | None
) -> bool:
deployment_flag: Final = (model_info or {}).get("supports_bedrock_native_structured_output")
deployment_flag: Final = (
model_info.get("supports_bedrock_native_structured_output") if model_info is not None else None
)
if isinstance(deployment_flag, bool):
return deployment_flag
return _bedrock_model_supports(model, "supports_bedrock_native_structured_output", custom_llm_provider)
@ -288,9 +292,12 @@ def bedrock_native_output_is_opt_in(
def without_bedrock_native_output_opt_in(litellm_params: Mapping[str, object]) -> Mapping[str, object]:
return {
return { # mutable-ok: request metadata must remain JSON serializable and deep-copyable
**litellm_params,
"model_info": {**deployment_model_info(litellm_params), "supports_bedrock_native_structured_output": False},
"model_info": { # mutable-ok: nested request metadata is serialized as JSON
**deployment_model_info(litellm_params),
"supports_bedrock_native_structured_output": False,
},
}
@ -394,7 +401,7 @@ def bedrock_native_output_format(
native_schema: Final = bedrock_native_output_schema(model, schema, custom_llm_provider, model_info)
if native_schema is None:
return None
return {**output_format, "schema": native_schema}
return {**output_format, "schema": native_schema} # mutable-ok: output format is serialized directly as JSON
def apply_bedrock_invoke_structured_output(
@ -427,7 +434,9 @@ def apply_bedrock_invoke_structured_output(
if isinstance(existing_output_config, dict):
existing_output_config["format"] = native_format
else:
request_body["output_config"] = {"format": native_format} # rebind-ok: out-param
request_body["output_config"] = { # mutable-ok: wire request JSON object
"format": native_format
} # rebind-ok: out-param
return
verbose_logger.warning(

View file

@ -901,7 +901,7 @@ class BedrockFilesConfig(BaseAWSLLM, BaseFilesConfig):
)
config: Final = AmazonAnthropicClaudeConfig()
litellm_params: Final = without_bedrock_native_output_opt_in({})
litellm_params: Final = without_bedrock_native_output_opt_in(MappingProxyType({}))
mapped_params = get_optional_params(
model=_invoke_route_model(model),
custom_llm_provider="bedrock",
@ -918,10 +918,10 @@ class BedrockFilesConfig(BaseAWSLLM, BaseFilesConfig):
return config.transform_request(
model=model,
messages=messages,
optional_params={
optional_params={ # mutable-ok: provider transform consumes and removes request options
k: v for k, v in mapped_params.items() if k not in _MAPPED_PARAMS_THE_REQUEST_HANDLER_STRIPS
},
litellm_params=dict(litellm_params),
litellm_params=dict(litellm_params), # mutable-ok: legacy provider transform requires a private dict
headers={},
)

View file

@ -117,7 +117,9 @@ class AmazonAnthropicClaudeMessagesConfig(
if not requests_opt_in_native_output(model, anthropic_messages_optional_request_params, model_info):
return None
return litellm_params.model_copy(
update={"model_info": {**(model_info or {}), "supports_bedrock_native_structured_output": False}}
update={ # mutable-ok: nested model metadata must remain JSON serializable
"model_info": {**(model_info or {}), "supports_bedrock_native_structured_output": False}
}
)
def should_retry_anthropic_messages_without_native_output(

View file

@ -1,4 +1,3 @@
from pydantic import TypeAdapter
# LiteLLM main module: public completion, embedding, streaming, and moderation entrypoints.
#
# +-----------------------------------------------+
@ -9,7 +8,6 @@ from pydantic import TypeAdapter
# +-----------------------------------------------+
#
# Thank you ! We ❤️ you! - Krrish & Ishaan
import asyncio
import contextvars
import datetime
@ -29,6 +27,8 @@ from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, Literal, Optional, Protocol, Union, cast, get_args
from urllib.parse import urlsplit
from pydantic import TypeAdapter
from litellm._logging import _redact_string
from litellm._uuid import uuid
@ -4177,9 +4177,11 @@ def _native_output_fallback(
litellm_params: Final = without_bedrock_native_output_opt_in(first_attempt_litellm_params)
remapped_params: Final = ctx.optional_params_for_model_info(deployment_model_info(litellm_params))
return NativeOutputFallback(
optional_params={
**{key: value for key, value in remapped_params.items() if key != "aws_bedrock_client"},
**{
optional_params={ # mutable-ok: provider handlers consume request options in place
**{ # mutable-ok: request options need independent ownership
key: value for key, value in remapped_params.items() if key != "aws_bedrock_client"
},
**{ # mutable-ok: preserve only credentials resolved by the first attempt
key: first_attempt_params[key]
for key in _AWS_BEDROCK_CLIENT_CREDENTIAL_KEYS
if key in first_attempt_params
@ -4272,8 +4274,12 @@ def _complete_bedrock(ctx: _CompletionDispatchContext) -> _CompletionDispatchRes
messages=messages,
custom_prompt_dict=custom_prompt_dict,
model_response=model_response,
optional_params=optional_params if fallback is None else dict(fallback.optional_params),
litellm_params=litellm_params if fallback is None else dict(fallback.litellm_params),
optional_params=optional_params
if fallback is None
else dict(fallback.optional_params), # mutable-ok: handler pops request options
litellm_params=litellm_params
if fallback is None
else dict(fallback.litellm_params), # mutable-ok: handler mutates private request metadata
logger_fn=logger_fn,
encoding=_get_encoding(),
logging_obj=logging,
@ -4289,7 +4295,7 @@ def _complete_bedrock(ctx: _CompletionDispatchContext) -> _CompletionDispatchRes
response = call_with_native_output_fallback(
converse_attempt,
(
partial(_converse_native_output_fallback, ctx, converse_params, dict(litellm_params))
partial(_converse_native_output_fallback, ctx, converse_params, MappingProxyType(litellm_params.copy()))
if AmazonConverseConfig().uses_opt_in_native_output(converse_model, converse_params, litellm_params)
else None
),
@ -4324,8 +4330,12 @@ def _complete_bedrock(ctx: _CompletionDispatchContext) -> _CompletionDispatchRes
acompletion=acompletion,
api_base=api_base,
model_response=model_response,
optional_params=optional_params if fallback is None else dict(fallback.optional_params),
litellm_params=litellm_params if fallback is None else dict(fallback.litellm_params),
optional_params=optional_params
if fallback is None
else dict(fallback.optional_params), # mutable-ok: handler pops request options
litellm_params=litellm_params
if fallback is None
else dict(fallback.litellm_params), # mutable-ok: handler mutates private request metadata
custom_llm_provider="bedrock",
timeout=timeout,
headers=headers,
@ -4339,7 +4349,7 @@ def _complete_bedrock(ctx: _CompletionDispatchContext) -> _CompletionDispatchRes
response = call_with_native_output_fallback(
invoke_attempt,
(
partial(_native_output_fallback, ctx, invoke_params, dict(litellm_params))
partial(_native_output_fallback, ctx, invoke_params, MappingProxyType(litellm_params.copy()))
if isinstance(provider_config, AmazonAnthropicClaudeConfig)
and provider_config.uses_opt_in_native_output(model, invoke_params, litellm_params)
else None

View file

@ -1,7 +1,6 @@
from __future__ import annotations
from collections.abc import Mapping
from collections.abc import Callable, Coroutine, Iterable
from collections.abc import Callable, Coroutine, Iterable, Mapping
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any, Literal, Union

View file

@ -3831,9 +3831,7 @@ class CustomPricingLiteLLMParams(MirroredPricingParams):
DEPLOYMENT_SCOPED_PRICING_FIELDS: Final[frozenset[str]] = frozenset({"off_peak_pricing"})
DEPLOYMENT_SCOPED_CAPABILITY_FIELDS: Final[frozenset[str]] = frozenset(
{"supports_bedrock_native_structured_output"}
)
DEPLOYMENT_SCOPED_CAPABILITY_FIELDS: Final[frozenset[str]] = frozenset({"supports_bedrock_native_structured_output"})
DEPLOYMENT_SCOPED_TOKEN_COUNT_FIELDS: Final[frozenset[str]] = frozenset({"token_count_framing", "token_count_model"})

View file

@ -3,7 +3,7 @@ from __future__ import annotations
import os
import time
import uuid
from collections.abc import Sequence, Callable, Iterator, Mapping
from collections.abc import Callable, Iterator, Mapping, Sequence
from contextlib import ExitStack, contextmanager
from dataclasses import dataclass
from hashlib import sha256
@ -219,3 +219,25 @@ def gateway_from_environment() -> Iterator[Gateway]:
upstream: Final = os.environ["INTEGRATION_UPSTREAM_URL"]
with httpx.Client(base_url=url, timeout=15, trust_env=False) as client:
yield Gateway(client, os.environ["INTEGRATION_MASTER_KEY"], upstream)
def held(read: Callable[[], T], satisfied: Callable[[T], bool], *, holding: float, seconds: float = 10) -> T:
"""`eventually`, and then `satisfied` must hold on every read for `holding` more seconds; a miss restarts the hold."""
deadline: Final = time.monotonic() + seconds
while True:
converged: Final = eventually(read, satisfied, seconds=max(deadline - time.monotonic(), 0.0))
misses: Final = _misses_within(read, satisfied, holding)
if not misses:
return converged
assert time.monotonic() < deadline, f"State did not hold: {misses[0]!r}"
time.sleep(0.1)
def _misses_within(read: Callable[[], T], satisfied: Callable[[T], bool], holding: float) -> tuple[T, ...]:
until: Final = time.monotonic() + holding
while time.monotonic() < until:
time.sleep(0.1)
observed: Final = read()
if not satisfied(observed):
return (observed,)
return ()

View file

@ -89,9 +89,12 @@ def _aws_event_frame(
scenario_id: str,
unique_id: str,
) -> bytes:
payload_bytes: Final = json.dumps(payload, separators=(",", ":")).replace(
"$REQUEST_ID", scenario_id
).replace("$UNIQUE_ID", unique_id).encode()
payload_bytes: Final = (
json.dumps(payload, separators=(",", ":"))
.replace("$REQUEST_ID", scenario_id)
.replace("$UNIQUE_ID", unique_id)
.encode()
)
headers_bytes: Final = (
_aws_str_header(":event-type", event_type)
+ _aws_str_header(":content-type", "application/json")
@ -147,7 +150,13 @@ class Provider:
status: Final = script.popleft()
if status != 200:
return JSONResponse(
{"error": {"message": "Controlled provider failure", "type": error_type(status), "code": str(status)}},
{
"error": {
"message": "Controlled provider failure",
"type": error_type(status),
"code": str(status),
}
},
status_code=status,
)
return await chat_completions(request)
@ -244,9 +253,7 @@ class Provider:
if raw_body:
body: Final = JSON_OBJECT.validate_json(raw_body)
if isinstance(body, dict):
self.observations.put(
Observation(request.url.path, request.headers.get("authorization", ""), body)
)
self.observations.put(Observation(request.url.path, request.headers.get("authorization", ""), body))
if isinstance(response, RoutedResponse):
route_key: Final = f"{request.method} /{'/'.join(segments[1:])}"
route: Final = next(
@ -300,11 +307,10 @@ class Provider:
match response:
case JsonResponse():
return Response(
content=json.dumps(response.body, separators=(",", ":")).replace(
"$REQUEST_ID", scenario_id
).replace(
"$UNIQUE_ID", unique_id
).encode(),
content=json.dumps(response.body, separators=(",", ":"))
.replace("$REQUEST_ID", scenario_id)
.replace("$UNIQUE_ID", unique_id)
.encode(),
media_type=response.content_type,
status_code=response.status,
)
@ -321,6 +327,7 @@ class Provider:
)
case SseResponse():
if response.frame_delay_ms > 0:
async def stream() -> AsyncIterator[bytes]:
for frame in response.frames:
yield (
@ -329,9 +336,11 @@ class Provider:
await asyncio.sleep(response.frame_delay_ms / 1000)
return StreamingResponse(stream(), media_type=response.content_type)
stream_body: Final = ("\n\n".join(response.frames) + "\n\n").replace(
"$REQUEST_ID", scenario_id
).replace("$UNIQUE_ID", unique_id)
stream_body: Final = (
("\n\n".join(response.frames) + "\n\n")
.replace("$REQUEST_ID", scenario_id)
.replace("$UNIQUE_ID", unique_id)
)
return Response(content=stream_body.encode(), media_type=response.content_type)
case EventStreamResponse():
events: Final = (
@ -423,3 +432,21 @@ def main() -> None:
if __name__ == "__main__":
main()
def _aws_int_header(name: str, value: int) -> bytes:
name_bytes: Final = name.encode()
return struct.pack("!B", len(name_bytes)) + name_bytes + struct.pack("!B", 4) + struct.pack("!i", value)
def aws_event_stream_frame(headers: Mapping[str, str | int], payload: bytes) -> bytes:
"""One AWS event-stream frame: a string header is wire type 7, an int header wire type 4 (int32)."""
headers_bytes: Final = b"".join(
_aws_str_header(name, value) if isinstance(value, str) else _aws_int_header(name, value)
for name, value in headers.items()
)
total_length: Final = 12 + len(headers_bytes) + len(payload) + 4
prelude: Final = struct.pack("!II", total_length, len(headers_bytes))
prelude_crc: Final = struct.pack("!I", zlib.crc32(prelude) & 0xFFFFFFFF)
message: Final = prelude + prelude_crc + headers_bytes + payload
return message + struct.pack("!I", zlib.crc32(message) & 0xFFFFFFFF)