fix(bedrock): use native structured output on Haiku 5.5 and Sonnet 5.5, and fix Fable 5 and Sonnet 4.6 schema 400s (#46870)

* fix(bedrock): use native structured output on every Claude model whose Bedrock profiles accept it

Bedrock accepts outputConfig structured output on every reachable profile of
Claude Opus 4.7, 4.8, 5, 5.5, Sonnet 5, 5.5, Haiku 5.5, and Fable 5.1, so
json_schema now goes out natively there instead of as a forced tool, which kept
Bedrock from returning adaptive thinking. Fable 5 rejects outputConfig, so its
keys go back to the tool path. GovCloud keys keep their flags, and the drift
tests let GovCloud lag its commercial base but never lead it.

* fix(bedrock): clean Claude schemas for native structured output and keep recursive schemas on the old path

Bedrock native output rejects minimum/maximum, minItems, uniqueItems, oneOf,
additionalProperties true, and recursive or unresolved $ref, all of which the
forced tool accepted. Claude schemas on the Bedrock native paths now go through
the cleaner the Anthropic native path already uses, and a schema with a
recursive or unresolved $ref keeps the forced tool, inline, or drop path it had.

* fix(bedrock): keep older installs on the json tool with a Bedrock-only native output key

Older installs fetch the model map from main, so flipping supports_native_structured_output
moved them onto native output with their old schema handling. The new
supports_bedrock_native_structured_output key is ignored by older code. Schemas native output
cannot hold (recursive refs, open objects such as dict fields) keep the json tool on the new
key, models on the old key keep their current behavior for those shapes, and the Anthropic
API, Vertex AI, and Azure keep their current behavior.

* chore(cost-map): regenerate the model prices schema for the Bedrock native output key

* fix(bedrock): keep native output to profiles every account accepts and route schemas Bedrock rejects to the json tool

Only Opus 5.5, Sonnet 5.5, Haiku 5.5 and Fable 5.1 keep the Bedrock native output key, since
Opus 4.8 and Sonnet 5 answer 400 on some accounts and Opus 5 garbles escaped characters.
Lookaround patterns and string formats outside Bedrock's list now keep the json tool, objects
typed as a union with null are closed, and the cleaner keeps minItems of 0 or 1.

* fix(bedrock): keep Opus 5.5 and Fable 5.1 on the json tool, since native output stops their adaptive thinking

* fix(bedrock): keep shared $defs in native schemas and send refused schemas to the json tool on every native model

Native output now keeps $defs instead of inlining each use, so a submodel reused across a schema counts once against Bedrock's optional and union limits. Schemas native output cannot hold use the json tool on Converse and Invoke chat for both native keys, and minItems is only kept when it is an integer 0 or 1

* fix(bedrock): keep the json tool unless thinking is requested and bound native output schemas

The Bedrock native output key now applies only when the call asks for thinking, so calls without
thinking keep the json tool as before. Schemas past Bedrock's documented union and optional counts,
or past a grammar size estimate fitted to live probes, keep the json tool on both map keys. Open
objects keep the json tool on the Anthropic API, Vertex AI and Azure AI, where native output returned
them empty. The $ref check no longer expands $defs, so doubling definitions no longer stall the request.

* fix(bedrock): resolve root-path $defs refs so nested Pydantic models keep native output on Invoke

* test(bedrock): cover native structured output and json tool routing through the proxy

Integration cells drive the proxy against a scripted Bedrock peer that enforces Bedrock's structured output schema rules, covering thinking with native output on Converse, Converse streaming, Invoke, /v1/responses and /v1/messages, the deployment opt-in, Fable 5, schema cleaning on every native entry, open objects, and malformed schemas. Vertex, Anthropic and Azure AI cells cover open objects on native-capable models and the cleaner's minItems and nullable object handling. The unit table now covers reasoning_effort none and disabled thinking

* fix(bedrock): keep native output for numeric formats and wide old-key schemas, and keep thinking in json mode

Check string formats only on nodes that can be strings, apply the 64-field cap only to the Bedrock-only key, carry thinking blocks into the json mode message, and send malformed json tool schemas as object schemas.

* fix(anthropic): keep native output for dict fields unless the json tool can be forced

* fix(anthropic): keep native output for dict fields when the caller sends tools

* test(bedrock): close the dangerous-tool-use test schema, since Bedrock native output requires closed objects

* test(anthropic): cover forced and unforced open-object routing, json mode thinking, numeric formats and wide schemas

Add proxy cells for open-object schemas that keep native output when the json tool cannot be forced (a model that refuses forced tools on the Anthropic API and Vertex AI, thinking, caller tools), for the thinking block a json mode answer keeps on Vertex AI and Bedrock Invoke, and for integer and number formats and 65-field schemas on each Bedrock map key. Point the forced open-object cells at the Anthropic API and Azure AI only, since no Vertex AI model in the cost map takes both native output and a forced tool

* fix(bedrock): keep the native output opt-in on the deployment that sets it, and route effort none open objects to native output

* test(bedrock): cover deployment-scoped native output opt-in, register_model and effort none open objects
This commit is contained in:
Mateo Wang 2026-10-17 03:57:23 +00:00 • committed by Joshua Valluru
parent 11ab893c26
commit b6bdc0dfc8
28 changed files with 5103 additions and 159 deletions

View file

@ -7,7 +7,8 @@ import json
import mimetypes
import re
from collections.abc import Callable, Iterable, Iterator, Mapping, Sequence
from itertools import groupby, islice
from graphlib import CycleError, TopologicalSorter
from itertools import chain, groupby, islice
from os import PathLike
from pathlib import Path
from types import MappingProxyType
@ -1371,6 +1372,123 @@ def _schema_levels(schema: Mapping[str, object]) -> Iterator[tuple[Mapping[str,
frontier = tuple(child for node in frontier for child in _subschemas(node))
def schema_fragment_ref(ref: str) -> str:
return f"#{ref}" if ref.startswith(("/$defs/", "/definitions/")) else ref
def schema_has_unresolvable_ref(schema: Mapping[str, object]) -> bool:
definitions: Final = MappingProxyType(dict(_root_definitions(schema)))
if not _refs_in(schema) <= definitions.keys():
return True
try:
tuple(TopologicalSorter({pointer: _refs_in(body) for pointer, body in definitions.items()}).static_order())
except CycleError:
return True
return False
def schema_union_property_count(schema: Mapping[str, object]) -> int:
return sum(1 for _, node in _property_schemas(schema) if _is_union(node))
def schema_optional_property_count(schema: Mapping[str, object]) -> int:
return sum(len(_optional_property_names(node)) for node in _schema_nodes(schema))
def schema_property_count(schema: Mapping[str, object]) -> int:
return sum(len(properties) for node in _schema_nodes(schema) if (properties := _schema_at(node, "properties")))
def schema_grammar_branch_count(schema: Mapping[str, object]) -> int:
return sum(_grammar_branches(node) for node in _schema_nodes(schema))
def schema_has_open_object(schema: Mapping[str, object]) -> bool:
return any(_is_open_object(node) for node in _schema_nodes(schema))
def schema_has_lookaround_pattern(schema: Mapping[str, object]) -> bool:
return any(
isinstance(pattern := node.get("pattern"), str) and _uses_regex_lookaround(pattern)
for node in _schema_nodes(schema)
)
def schema_string_formats(schema: Mapping[str, object]) -> frozenset[str]:
return frozenset(
value for node in _schema_nodes(schema) if _may_be_string(node) and isinstance(value := node.get("format"), str)
)
def _may_be_string(node: Mapping[str, object]) -> bool:
node_type: Final = node.get("type")
return node_type is None or node_type == "string" or "string" in _list_at(node, "type")
def is_object_schema(node: Mapping[str, object]) -> bool:
node_type: Final = node.get("type")
return node_type == "object" or (isinstance(node_type, list) and "object" in node_type)
def _schema_nodes(schema: Mapping[str, object]) -> Iterator[Mapping[str, object]]:
return chain.from_iterable(islice(_schema_levels(schema), _MAX_SCHEMA_NESTING))
_SCHEMA_LIST: Final = TypeAdapter(tuple[object, ...])
def _list_at(container: Mapping[str, object], key: str) -> tuple[object, ...]:
value: Final = container.get(key)
return _SCHEMA_LIST.validate_python(value) if isinstance(value, list) else ()
def _named_schemas(container: Mapping[str, object] | None) -> Iterator[tuple[str, Mapping[str, object]]]:
if container is None:
return iter(())
return ((key, value) for key in container if (value := _schema_at(container, key)) is not None)
def _root_definitions(schema: Mapping[str, object]) -> Iterator[tuple[str, Mapping[str, object]]]:
for defs_key in ("$defs", "definitions"):
yield from ((f"#/{defs_key}/{name}", body) for name, body in _named_schemas(_schema_at(schema, defs_key)))
def _refs_in(schema: Mapping[str, object]) -> frozenset[str]:
return frozenset(
schema_fragment_ref(ref) for node in _schema_nodes(schema) if isinstance(ref := node.get("$ref"), str)
)
def _property_schemas(schema: Mapping[str, object]) -> Iterator[tuple[str, Mapping[str, object]]]:
return chain.from_iterable(_named_schemas(_schema_at(node, "properties")) for node in _schema_nodes(schema))
def _is_union(node: Mapping[str, object]) -> bool:
return isinstance(node.get("anyOf"), list) or isinstance(node.get("oneOf"), list) or len(_list_at(node, "type")) > 1
def _grammar_branches(node: Mapping[str, object]) -> int:
properties: Final = _schema_at(node, "properties")
if properties is None:
return 0
optional: Final = len(_optional_property_names(node))
return 2**optional * (len(properties) - optional + 1)
def _optional_property_names(node: Mapping[str, object]) -> frozenset[str]:
properties: Final = _schema_at(node, "properties")
if properties is None:
return frozenset()
return frozenset(properties) - frozenset(name for name in _list_at(node, "required") if isinstance(name, str))
def _is_open_object(node: Mapping[str, object]) -> bool:
additional_properties: Final = node.get("additionalProperties")
if additional_properties is True or isinstance(additional_properties, dict) or "patternProperties" in node:
return True
return is_object_schema(node) and "properties" not in node and additional_properties is not False
def _subschemas(node: Mapping[str, object]) -> Iterator[Mapping[str, object]]:
for key, value in node.items():
if key in _SUBSCHEMA_MAP_KEYWORDS and isinstance(value, dict):

View file

@ -5129,14 +5129,14 @@ def _bedrock_tools_pt(tools: list, model: str | None = None) -> list[BedrockTool
description = name
defs = parameters.pop("$defs", {})
defs_copy = copy.deepcopy(defs)
defs_copy = copy.deepcopy(defs) if isinstance(defs, dict) else {}
# 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
# refs recursively and correctly detects/skips circular references.
unpack_defs(parameters, defs_copy)
normalize_json_schema_custom_types_to_object(parameters)
if parameters.get("type") not in _valid_json_schema_root_types:
if not (isinstance(root_type := parameters.get("type"), str) and root_type in _valid_json_schema_root_types):
parameters["type"] = "object"
tool_block = cast(
BedrockToolBlock,

View file

@ -1,3 +1,4 @@
import copy
import json
import re
import time
@ -24,7 +25,18 @@ from litellm.constants import (
)
from litellm.litellm_core_utils.core_helpers import map_finish_reason
from litellm.litellm_core_utils.prompt_templates.common_utils import (
DEFS_MAX_INLINED_BYTES,
is_object_schema,
sanitize_input_schema_for_anthropic,
schema_fragment_ref,
schema_grammar_branch_count,
schema_has_lookaround_pattern,
schema_has_open_object,
schema_has_unresolvable_ref,
schema_optional_property_count,
schema_string_formats,
schema_union_property_count,
unpack_defs,
)
from litellm.litellm_core_utils.prompt_templates.image_handling import (
RemoteMedia,
@ -36,7 +48,7 @@ from litellm.litellm_core_utils.prompt_templates.mid_conversation_system import
split_leading_system_run,
)
from litellm.llms.base_llm.base_utils import type_to_response_format_param
from litellm.llms.base_llm.chat.transformation import BaseConfig, BaseLLMException
from litellm.llms.base_llm.chat.transformation import REQUEST_PARAMS, BaseConfig, BaseLLMException
from litellm.types.llms.anthropic import (
ANTHROPIC_ADVISOR_TOOL_TYPE,
ANTHROPIC_BETA_HEADER_VALUES,
@ -137,6 +149,12 @@ else:
# response side.
_ANTHROPIC_TOOL_NAME_INVALID_CHARS: Final = re.compile(r"[^a-zA-Z0-9_-]")
_ANTHROPIC_TOOL_NAME_MAX_LEN: Final = 128
_NATIVE_OUTPUT_STRING_FORMATS: Final = frozenset(
("date-time", "time", "date", "duration", "email", "hostname", "uri", "ipv4", "ipv6", "uuid")
)
NATIVE_OUTPUT_MAX_UNION_PARAMETERS: Final = 16
NATIVE_OUTPUT_MAX_OPTIONAL_PARAMETERS: Final = 24
NATIVE_OUTPUT_MAX_GRAMMAR_BRANCHES: Final = 2048
class _AnthropicUsageIteration(TypedDict, total=False):
@ -536,7 +554,7 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
grammar Anthropic compiles the schema into, so the API rejects them with a
400 ``invalid_request_error`` (e.g. "output_format.schema: For 'array' type,
property 'uniqueItems' is not supported"):
- maxItems/minItems/uniqueItems/contains/minContains/maxContains/prefixItems: array constraints
- maxItems/minItems above 1/uniqueItems/contains/minContains/maxContains/prefixItems: array constraints
- minimum/maximum/exclusiveMinimum/exclusiveMaximum/multipleOf: numeric constraints
- minLength/maxLength: string constraints
- minProperties/maxProperties/patternProperties/propertyNames: object constraints
@ -593,7 +611,11 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
"else": "conditional (else): {}",
"not": "must not match: {}",
}
unsupported_fields: Final = set(constraint_labels)
min_items: Final = schema.get("minItems")
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())
# Build description additions from removed constraints. Iterating
# constraint_labels (not the set) keeps the note order deterministic across
@ -601,7 +623,7 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
# PYTHONHASHSEED and stay cache-friendly.
constraint_descriptions: Final[list] = []
for field, label in constraint_labels.items():
if field not in schema:
if field not in schema or field not in unsupported_fields:
continue
value = schema[field]
# A falsy boolean constraint (e.g. ``uniqueItems: false``) imposes no
@ -636,7 +658,7 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
# Already handled above
continue
if key == "properties" and isinstance(value, dict):
if key in ("properties", "$defs", "definitions") and isinstance(value, dict):
result[key] = {k: AnthropicConfig.filter_anthropic_output_schema(v) for k, v in value.items()}
elif key == "items" and isinstance(value, dict):
result[key] = AnthropicConfig.filter_anthropic_output_schema(value)
@ -648,6 +670,8 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
]
elif key == "allOf" and isinstance(value, list):
result[key] = [AnthropicConfig.filter_anthropic_output_schema(item) for item in value]
elif key == "$ref" and isinstance(value, str):
result[key] = schema_fragment_ref(value)
elif key == "oneOf" and isinstance(value, list):
# Anthropic rejects oneOf ("Schema type 'oneOf' is not supported");
# the Anthropic SDK rewrites it to anyOf, so do the same.
@ -659,7 +683,7 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
# Anthropic requires additionalProperties=false for object schemas
# See: https://docs.anthropic.com/en/docs/build-with-claude/structured-outputs
if result.get("type") == "object":
if is_object_schema(result):
result["additionalProperties"] = False
return result
@ -1336,32 +1360,55 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
return json_schema
@staticmethod
def _defs_unpacked(json_schema: Mapping[str, object]) -> Mapping[str, object]:
unpacked: Final = copy.deepcopy(dict(json_schema))
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)
return unpacked
@staticmethod
def native_output_schema(json_schema: Mapping[str, object]) -> Mapping[str, object] | None:
if (
schema_has_unresolvable_ref(json_schema)
or schema_has_open_object(json_schema)
or schema_has_lookaround_pattern(json_schema)
or not schema_string_formats(json_schema) <= _NATIVE_OUTPUT_STRING_FORMATS
or schema_union_property_count(json_schema) > NATIVE_OUTPUT_MAX_UNION_PARAMETERS
or schema_optional_property_count(json_schema) > NATIVE_OUTPUT_MAX_OPTIONAL_PARAMETERS
or schema_grammar_branch_count(json_schema) > NATIVE_OUTPUT_MAX_GRAMMAR_BRANCHES
):
return None
return AnthropicConfig.filter_anthropic_output_schema(dict(json_schema))
def uses_native_structured_output_for_schema(
self,
model: str,
json_schema: Mapping[str, object] | None,
request_params: Mapping[str, object],
) -> bool:
open_object_takes_forced_json_tool: Final = (
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 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)))
def map_response_format_to_anthropic_output_format(self, value: dict | None) -> AnthropicOutputSchema | None:
json_schema: dict | None = self._extract_json_schema_from_response_format(value)
json_schema: Final = self._extract_json_schema_from_response_format(value)
if json_schema is None:
return None
# Resolve $ref/$defs before filtering — Anthropic doesn't support
# external schema references (e.g., /$defs/CalendarEvent).
import copy
from litellm.litellm_core_utils.prompt_templates.common_utils import (
DEFS_MAX_INLINED_BYTES,
unpack_defs,
)
json_schema = copy.deepcopy(json_schema)
defs: Final = json_schema.pop("$defs", json_schema.pop("definitions", {}))
if defs:
unpack_defs(json_schema, defs, max_inlined_bytes=DEFS_MAX_INLINED_BYTES)
# Filter out unsupported fields for Anthropic's output_format API
filtered_schema: Final = self.filter_anthropic_output_schema(json_schema)
return AnthropicOutputSchema(
type="json_schema",
schema=filtered_schema,
)
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))
def map_response_format_to_anthropic_tool(
self, value: dict | None, optional_params: dict, is_thinking_enabled: bool
@ -1526,10 +1573,10 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
output_key=param,
)
elif param == "response_format" and isinstance(value, dict):
if AnthropicConfig._supports_model_capability(
if self.uses_native_structured_output_for_schema(
model,
"supports_native_structured_output",
self._resolved_provider,
self._extract_json_schema_from_response_format(value),
REQUEST_PARAMS.validate_python(non_default_params),
):
_output_format = self.map_response_format_to_anthropic_output_format(value)
if _output_format is not None:
@ -2665,6 +2712,9 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
if json_mode_message is not None:
completion_response["stop_reason"] = "stop"
if thinking_blocks:
json_mode_message.thinking_blocks = thinking_blocks
json_mode_message.reasoning_content = reasoning_content
_message = json_mode_message
model_response.choices[0].message = _message

View file

@ -8,7 +8,7 @@ from collections.abc import AsyncIterator, Iterator
from typing import TYPE_CHECKING, Any, Final, Union
import httpx
from pydantic import BaseModel
from pydantic import BaseModel, TypeAdapter
from litellm.constants import DEFAULT_MAX_TOKENS, RESPONSE_FORMAT_TOOL_NAME
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler
@ -38,6 +38,9 @@ else:
LiteLLMLoggingObj = Any
REQUEST_PARAMS: Final = TypeAdapter(Mapping[str, object])
class BaseLLMException(Exception):
def __init__(
self,
@ -97,6 +100,17 @@ class BaseConfig(ABC):
thinking_type: Final = thinking.get("type") if isinstance(thinking, dict) else None
return thinking is True or thinking_type == "enabled" or non_default_params.get("reasoning_effort") is not None
def is_thinking_requested(self, non_default_params: Mapping[str, object]) -> bool:
match non_default_params:
case {"thinking": True | {"type": "enabled" | "adaptive"}}:
return True
case {"reasoning_effort": None | "none"}:
return False
case {"reasoning_effort": object()}:
return True
case _:
return False
def is_max_tokens_in_request(self, non_default_params: dict) -> bool:
"""
OpenAI spec allows max_tokens or max_completion_tokens to be specified.

View file

@ -28,6 +28,7 @@ from litellm.litellm_core_utils.core_helpers import (
from litellm.litellm_core_utils.litellm_logging import Logging
from litellm.litellm_core_utils.prompt_templates.common_utils import (
_parse_content_for_reasoning,
is_object_schema,
)
from litellm.litellm_core_utils.prompt_templates.factory import (
BedrockConverseMessagesProcessor,
@ -48,7 +49,13 @@ from litellm.llms.anthropic.chat.transformation import (
AnthropicConfig,
)
from litellm.llms.anthropic.common_utils import AnthropicModelInfo
from litellm.llms.base_llm.chat.transformation import BaseConfig, BaseLLMException
from litellm.llms.base_llm.chat.transformation import REQUEST_PARAMS, BaseConfig, BaseLLMException
from litellm.llms.bedrock.common_utils import (
bedrock_chat_uses_native_structured_output,
bedrock_cleaned_output_schema,
bedrock_native_output_format,
deployment_model_info,
)
from litellm.llms.bedrock.request_metadata import (
bedrock_request_metadata_headers,
bedrock_request_metadata_is_owned,
@ -906,17 +913,22 @@ class AmazonConverseConfig(BaseConfig):
return _tool
@staticmethod
def _supports_native_structured_outputs(model: str, custom_llm_provider: str | None = None) -> bool:
"""Check if the Bedrock model supports native structured outputs (outputConfig.textFormat).
def _supports_native_structured_outputs(
model: str,
custom_llm_provider: str | None,
thinking_requested: bool,
model_info: Mapping[str, object] | None,
) -> bool:
"""Check if the Bedrock model takes native structured outputs (outputConfig.textFormat) for this request.
Delegates to the standard ``supports_native_structured_output`` utility
which looks up the flag in ``litellm.model_cost`` via
``_get_model_info_helper``.
``supports_native_structured_output`` in ``litellm.model_cost`` turns it on for every request, and
``supports_bedrock_native_structured_output`` (the deployment's ``model_info`` first) only when the
request asks for thinking, which the json tool path would drop.
Ref: https://docs.aws.amazon.com/bedrock/latest/userguide/structured-output.html
"""
from litellm.utils import supports_native_structured_output
return supports_native_structured_output(model=model, custom_llm_provider=custom_llm_provider)
return bedrock_chat_uses_native_structured_output(
model, thinking_requested, custom_llm_provider or "bedrock_converse", model_info
)
@staticmethod
def _add_additional_properties_to_schema(schema: dict) -> dict:
@ -931,9 +943,9 @@ class AmazonConverseConfig(BaseConfig):
if not isinstance(schema, dict):
return schema
result: Final = dict(schema)
result: Final[dict[str, object]] = dict(schema)
if result.get("type") == "object" and "additionalProperties" not in result:
if is_object_schema(result) and "additionalProperties" not in result:
result["additionalProperties"] = False
# Recurse into nested schemas
@ -944,16 +956,46 @@ class AmazonConverseConfig(BaseConfig):
if "items" in result and isinstance(result["items"], dict):
result["items"] = AmazonConverseConfig._add_additional_properties_to_schema(result["items"])
for defs_key in ("$defs", "definitions"):
if defs_key in result and isinstance(result[defs_key], dict):
if isinstance(definitions := result.get(defs_key), dict):
result[defs_key] = {
k: AmazonConverseConfig._add_additional_properties_to_schema(v) for k, v in result[defs_key].items()
k: AmazonConverseConfig._add_additional_properties_to_schema(v) for k, v in definitions.items()
}
for key in ("anyOf", "allOf", "oneOf"):
if key in result and isinstance(result[key], list):
result[key] = [AmazonConverseConfig._add_additional_properties_to_schema(item) for item in result[key]]
if isinstance(variants := result.get(key), list):
result[key] = [AmazonConverseConfig._add_additional_properties_to_schema(item) for item in variants]
return result
def _native_output_schema(
self,
model: str,
json_schema: Mapping[str, object],
thinking_requested: bool,
model_info: Mapping[str, object] | None,
) -> Mapping[str, object] | None:
if not self._supports_native_structured_outputs(
model, self.custom_llm_provider, thinking_requested, model_info
):
return None
if BedrockModelInfo.get_base_model(model).startswith("anthropic"):
return bedrock_cleaned_output_schema(model, json_schema, self.custom_llm_provider or "bedrock_converse")
return json_schema
def _output_config_for_anthropic_format(
self, model: str, output_config_format: Mapping[str, object], model_info: Mapping[str, object] | None
) -> OutputConfigBlock | None:
native_format: Final = bedrock_native_output_format(
model, output_config_format, self.custom_llm_provider or "bedrock_converse", model_info
)
if native_format is None:
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,
name=name if isinstance(name, str) else None,
description=description if isinstance(description, str) else None,
)
@staticmethod
def _create_output_config_for_response_format(
json_schema: dict | None = None,
@ -1013,6 +1055,7 @@ class AmazonConverseConfig(BaseConfig):
optional_params: dict,
model: str,
drop_params: bool,
model_info: Mapping[str, object] | None = None,
) -> dict:
is_thinking_enabled: Final = self.is_thinking_enabled(non_default_params)
base_model: Final = BedrockModelInfo.get_base_model(model)
@ -1029,6 +1072,7 @@ class AmazonConverseConfig(BaseConfig):
optional_params=optional_params,
non_default_params=non_default_params,
is_thinking_enabled=is_thinking_enabled,
model_info=model_info,
)
if param == "max_tokens" or param == "max_completion_tokens":
optional_params["maxTokens"] = (
@ -1205,6 +1249,7 @@ class AmazonConverseConfig(BaseConfig):
optional_params: dict,
non_default_params: dict,
is_thinking_enabled: bool,
model_info: Mapping[str, object] | None = None,
) -> dict:
"""
Handles translation of response_format parameter to Bedrock format.
@ -1228,13 +1273,23 @@ class AmazonConverseConfig(BaseConfig):
if "type" in value and value["type"] == "text":
return optional_params
if self._supports_native_structured_outputs(model, self.custom_llm_provider) and json_schema is not None:
native_schema: Final = (
self._native_output_schema(
model,
json_schema,
self.is_thinking_requested(REQUEST_PARAMS.validate_python(non_default_params)),
model_info,
)
if json_schema is not None
else None
)
if native_schema is not None:
# Use Bedrock's native structured outputs API (outputConfig.textFormat)
# No synthetic tool injection, no fake_stream needed.
# 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=json_schema,
json_schema=dict(native_schema),
name=name,
description=description,
)
@ -1566,7 +1621,11 @@ class AmazonConverseConfig(BaseConfig):
tool_choice.pop("type", None)
def _prepare_request_params(
self, optional_params: dict, model: str, drop_params: bool = False
self,
optional_params: dict,
model: str,
drop_params: bool = False,
model_info: Mapping[str, object] | None = None,
) -> tuple[dict, dict, dict, OutputConfigBlock | None]:
"""Prepare and separate request parameters."""
# Consume the internal ``_output_config_normalized`` marker set by
@ -1609,17 +1668,12 @@ class AmazonConverseConfig(BaseConfig):
and output_config_format is not None
and output_config_format.get("type") == "json_schema"
and base_model.startswith("anthropic")
and self._supports_native_structured_outputs(model, self.custom_llm_provider)
):
output_config = self._create_output_config_for_response_format(
json_schema=output_config_format.get("schema"),
name=output_config_format.get("name"),
description=output_config_format.get("description"),
)
elif output_config is None and output_config_format is not None:
output_config = self._output_config_for_anthropic_format(model, output_config_format, model_info)
if output_config is None and output_config_format is not None:
litellm.verbose_logger.warning(
"Bedrock Converse: dropping `output_config.format` for model=%s — "
"model does not advertise `supports_native_structured_output` in "
"model does not take native structured output for this schema per "
"model_prices_and_context_window.json. The schema will not be "
"enforced; pass `response_format` to use the synthetic tool-call "
"fallback.",
@ -1898,7 +1952,7 @@ class AmazonConverseConfig(BaseConfig):
additional_request_params,
request_metadata,
output_config,
) = self._prepare_request_params(optional_params, model, drop_params)
) = self._prepare_request_params(optional_params, model, drop_params, deployment_model_info(litellm_params))
original_tools: Final = inference_params.pop("tools", [])

View file

@ -92,8 +92,11 @@ class AmazonInvokeNovaConfig(AmazonInvokeConfig, AmazonConverseConfig):
optional_params: dict,
model: str,
drop_params: bool,
model_info: Mapping[str, object] | None = None,
) -> dict:
return AmazonConverseConfig.map_openai_params(self, non_default_params, optional_params, model, drop_params)
return AmazonConverseConfig.map_openai_params(
self, non_default_params, optional_params, model, drop_params, model_info
)
def transform_request(
self,

View file

@ -18,7 +18,10 @@ from litellm.llms.bedrock.chat.invoke_transformations.base_invoke_transformation
)
from litellm.llms.bedrock.common_utils import (
apply_bedrock_invoke_structured_output,
bedrock_chat_uses_native_structured_output,
bedrock_cleaned_output_schema,
bedrock_supports_tool_search,
deployment_model_info,
get_anthropic_beta_from_headers,
normalize_bedrock_opus_output_config_effort,
normalize_custom_field_on_tools,
@ -41,6 +44,8 @@ if TYPE_CHECKING:
else:
LiteLLMLoggingObj = Any
_LITELLM_PARAMS: Final = TypeAdapter(Mapping[str, object])
class AmazonAnthropicClaudeConfig(AmazonInvokeConfig, AnthropicConfig):
"""
@ -62,6 +67,10 @@ class AmazonAnthropicClaudeConfig(AmazonInvokeConfig, AnthropicConfig):
anthropic_version: str = "bedrock-2023-05-31"
def __init__(self, model_info: Mapping[str, object] | None = None) -> None:
super().__init__()
self._deployment_model_info: Final = model_info
@property
def custom_llm_provider(self) -> str | None:
return "bedrock"
@ -72,6 +81,19 @@ class AmazonAnthropicClaudeConfig(AmazonInvokeConfig, AnthropicConfig):
def get_supported_openai_params(self, model: str) -> list[str]:
return AnthropicConfig.get_supported_openai_params(self, model)
def native_output_format_schema(self, json_schema: Mapping[str, object]) -> Mapping[str, object] | None:
return AnthropicConfig.native_output_schema(json_schema)
def uses_native_structured_output_for_schema(
self,
model: str,
json_schema: Mapping[str, object] | None,
request_params: Mapping[str, object],
) -> bool:
return bedrock_chat_uses_native_structured_output(
model, self.is_thinking_requested(request_params), model_info=self._deployment_model_info
) and (json_schema is None or bedrock_cleaned_output_schema(model, json_schema) is not None)
def map_openai_params(
self,
non_default_params: dict,
@ -229,6 +251,7 @@ class AmazonAnthropicClaudeConfig(AmazonInvokeConfig, AnthropicConfig):
apply_bedrock_invoke_structured_output(
model=model,
request_body=anthropic_request,
model_info=deployment_model_info(_LITELLM_PARAMS.validate_python(litellm_params)),
)
strip_unsupported_bedrock_invoke_output_config_keys(
model=model,

View file

@ -23,6 +23,7 @@ from pydantic import ConfigDict, TypeAdapter, ValidationError
import litellm
from litellm import verbose_logger
from litellm.litellm_core_utils.aws_partition import get_aws_dns_suffix
from litellm.litellm_core_utils.prompt_templates.common_utils import schema_property_count
from litellm.llms.base_llm.anthropic_messages.transformation import (
BaseAnthropicMessagesConfig,
)
@ -234,15 +235,99 @@ def convert_bedrock_invoke_output_format_to_inline_schema(
request_body["messages"] = new_messages
def _bedrock_model_supports(model: str, key: str) -> bool:
def _bedrock_model_supports(model: str, key: str, custom_llm_provider: str = "bedrock") -> bool:
from litellm.utils import _supports_factory
return _supports_factory(model=model, custom_llm_provider="bedrock", key=key)
return _supports_factory(model=model, custom_llm_provider=custom_llm_provider, key=key)
_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 {}
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")
if isinstance(deployment_flag, bool):
return deployment_flag
return _bedrock_model_supports(model, "supports_bedrock_native_structured_output", custom_llm_provider)
def bedrock_supports_native_structured_output(
model: str, custom_llm_provider: str = "bedrock", model_info: Mapping[str, object] | None = None
) -> bool:
return _bedrock_model_supports(
model, "supports_native_structured_output", custom_llm_provider
) or _bedrock_native_output_opt_in(model, custom_llm_provider, model_info)
def bedrock_chat_uses_native_structured_output(
model: str,
thinking_requested: bool,
custom_llm_provider: str = "bedrock",
model_info: Mapping[str, object] | None = None,
) -> bool:
return _bedrock_model_supports(model, "supports_native_structured_output", custom_llm_provider) or (
thinking_requested and _bedrock_native_output_opt_in(model, custom_llm_provider, model_info)
)
BEDROCK_NATIVE_OUTPUT_MAX_PROPERTIES: Final = 64
def bedrock_cleaned_output_schema(
model: str, schema: Mapping[str, object], custom_llm_provider: str = "bedrock"
) -> Mapping[str, object] | None:
from litellm.llms.anthropic.chat.transformation import AnthropicConfig
if (
not _bedrock_model_supports(model, "supports_native_structured_output", custom_llm_provider)
and schema_property_count(schema) > BEDROCK_NATIVE_OUTPUT_MAX_PROPERTIES
):
return None
return AnthropicConfig.native_output_schema(schema)
def bedrock_native_output_schema(
model: str,
schema: Mapping[str, object],
custom_llm_provider: str = "bedrock",
model_info: Mapping[str, object] | None = None,
) -> Mapping[str, object] | None:
native_schema: Final = bedrock_cleaned_output_schema(model, schema, custom_llm_provider)
if _bedrock_model_supports(model, "supports_native_structured_output", custom_llm_provider):
return schema if native_schema is None else native_schema
if _bedrock_native_output_opt_in(model, custom_llm_provider, model_info):
return native_schema
return None
def bedrock_native_output_format(
model: str,
output_format: Mapping[str, object],
custom_llm_provider: str = "bedrock",
model_info: Mapping[str, object] | None = None,
) -> Mapping[str, object] | None:
schema: Final = output_format.get("schema")
if not isinstance(schema, Mapping):
return (
output_format if bedrock_supports_native_structured_output(model, custom_llm_provider, model_info) else None
)
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}
def apply_bedrock_invoke_structured_output(
model: str,
request_body: dict[str, object], # mutable-ok: edited in place like siblings
model_info: Mapping[str, object] | None = None,
) -> None:
"""
Route Anthropic structured-output params to what the Bedrock model supports.
@ -262,17 +347,18 @@ def apply_bedrock_invoke_structured_output(
if schema_format is None:
return
if _bedrock_model_supports(model, "supports_native_structured_output"):
native_format: Final = bedrock_native_output_format(model, schema_format, model_info=model_info)
if native_format is not None:
existing_output_config: Final = request_body.get("output_config")
if isinstance(existing_output_config, dict):
existing_output_config["format"] = schema_format
existing_output_config["format"] = native_format
else:
request_body["output_config"] = {"format": schema_format} # rebind-ok: out-param # mutable-ok: json
request_body["output_config"] = {"format": native_format} # rebind-ok: out-param
return
verbose_logger.warning(
"Bedrock Invoke: model=%s does not advertise `supports_native_structured_output` "
"in model_prices_and_context_window.json, so the JSON schema was inlined into "
"Bedrock Invoke: model=%s does not take native structured output for this schema "
"per model_prices_and_context_window.json, so the JSON schema was inlined into "
"the last user message and is NOT enforced by the model.",
model,
)

View file

@ -3,6 +3,7 @@ from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, cast
import httpx
from pydantic import TypeAdapter
import litellm
from litellm.anthropic_beta_headers_manager import filter_and_transform_beta_headers
@ -64,6 +65,9 @@ else:
LiteLLMLoggingObj = Any
_MODEL_INFO: Final[TypeAdapter[Mapping[str, object] | None]] = TypeAdapter(Mapping[str, object] | None)
class AmazonAnthropicClaudeMessagesConfig(
AnthropicMessagesConfig,
AmazonInvokeConfig,
@ -706,6 +710,7 @@ class AmazonAnthropicClaudeMessagesConfig(
apply_bedrock_invoke_structured_output(
model=model,
request_body=anthropic_messages_request,
model_info=_MODEL_INFO.validate_python(getattr(litellm_params, "model_info", None)),
)
normalize_bedrock_opus_output_config_effort(
model=model,

View file

@ -4148,6 +4148,10 @@ def _complete_sagemaker(ctx: _CompletionDispatchContext) -> _CompletionDispatchR
)
_OPTIONAL_PARAMS_ADAPTER: Final = TypeAdapter(dict[str, object])
_MODEL_INFO_ADAPTER: Final = TypeAdapter(Mapping[str, object])
def _complete_bedrock(ctx: _CompletionDispatchContext) -> _CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base: Final = ctx.api_base
@ -5603,7 +5607,11 @@ def completion(
"allowed_openai_params": allowed_openai_params,
"base_model": base_model,
}
optional_params = get_optional_params(**optional_param_args, **non_default_params)
optional_params = get_optional_params(
**optional_param_args,
model_info=_MODEL_INFO_ADAPTER.validate_python(model_info) if isinstance(model_info, dict) else None,
**non_default_params,
)
processed_non_default_params: Final = pre_process_non_default_params(
model=model,
passed_params=optional_param_args,

View file

@ -188,6 +188,7 @@ class ProviderSpecificModelInfo(TypedDict, total=False):
supports_pdf_input: bool | None
supports_native_streaming: bool | None
supports_native_structured_output: bool | None
supports_bedrock_native_structured_output: ReadOnly[bool | None]
supports_parallel_function_calling: bool | None
supports_web_search: bool | None
supports_reasoning: bool | None
@ -3830,6 +3831,10 @@ 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_TOKEN_COUNT_FIELDS: Final[frozenset[str]] = frozenset({"token_count_framing", "token_count_model"})
SHARED_BACKEND_MODEL_INFO_FIELDS: Final[frozenset[str]] = (

View file

@ -4469,7 +4469,8 @@ def get_optional_params(
store: bool | None = None,
prompt_cache_key: str | None = None,
base_model: str | None = None,
**kwargs,
model_info: Mapping[str, object] | None = None,
**kwargs: object,
):
drop_params = normalize_drop_params(drop_params) # rebind-ok: config and DB deployments pass "true" as a string
passed_params: Final = locals().copy()
@ -4478,6 +4479,7 @@ def get_optional_params(
# non_default_params / _check_valid_arg — it's a routing hint, not an
# OpenAI param.
passed_params.pop("base_model", None)
passed_params.pop("model_info", None)
provider_config: BaseConfig | None = None
if custom_llm_provider is not None and custom_llm_provider in [provider.value for provider in LlmProviders]:
provider_config = ProviderConfigManager.get_provider_chat_config(
@ -4710,6 +4712,7 @@ def get_optional_params(
non_default_params=non_default_params,
optional_params=optional_params,
drop_params=bool(drop_params),
model_info=model_info,
)
elif bedrock_route == "openai":
optional_params = litellm.AmazonBedrockOpenAIConfig().map_openai_params(
@ -4727,7 +4730,7 @@ def get_optional_params(
drop_params=bool(drop_params),
)
else:
optional_params = litellm.AmazonAnthropicClaudeConfig().map_openai_params(
optional_params = litellm.AmazonAnthropicClaudeConfig(model_info=model_info).map_openai_params(
non_default_params=non_default_params,
optional_params=optional_params,
model=model,
@ -6293,6 +6296,9 @@ def _get_model_info_helper(
supports_embedding_image_input=_model_info.get("supports_embedding_image_input", None),
supports_native_streaming=_model_info.get("supports_native_streaming", None),
supports_native_structured_output=_model_info.get("supports_native_structured_output", None),
supports_bedrock_native_structured_output=_model_info.get(
"supports_bedrock_native_structured_output", None
),
supports_web_search=_model_info.get("supports_web_search", None),
supports_url_context=_model_info.get("supports_url_context", None),
supports_reasoning=_model_info.get("supports_reasoning", None),

View file

@ -918,6 +918,9 @@
"supports_audio_output": {
"type": "boolean"
},
"supports_bedrock_native_structured_output": {
"type": "boolean"
},
"supports_computer_use": {
"type": "boolean"
},

View file

@ -0,0 +1,820 @@
import json
import re
import uuid
from collections.abc import Iterable, Iterator, Mapping
from itertools import chain
from typing import Final
from urllib.parse import unquote
import httpx
import pytest
from integration._support.client import Gateway, Scenario, list_value, object_value, string_value
from integration._support.upstream import aws_event_stream_frame
from integration._support.wire import Reply, Request, Wire, wire_server
from pydantic import JsonValue, TypeAdapter
_HAIKU: Final = "us.anthropic.claude-haiku-5-5"
_FABLE: Final = "us.anthropic.claude-fable-5"
_SONNET: Final = "us.anthropic.claude-sonnet-4-6"
_OPUS: Final = "us.anthropic.claude-opus-5-5"
_OPT_IN: Final = "au.anthropic.claude-opus-4-8"
_DEEPSEEK: Final = "deepseek.v3.2"
_AWS: Final[dict[str, JsonValue]] = {
"aws_access_key_id": "AKIASCRIPTEDPROVIDER",
"aws_secret_access_key": "scripted-secret",
"aws_region_name": "us-east-1",
}
_NO_CACHE: Final[dict[str, JsonValue]] = {"cache": {"no-cache": True}, "num_retries": 0}
_JSON_TOOL: Final = "json_tool_call"
_USER_TOOL: Final = "lookup"
_SIGNATURE: Final = "c2NyaXB0ZWQtc2lnbmF0dXJl"
_ADAPTIVE: Final[dict[str, JsonValue]] = {"type": "adaptive"}
_THINKING_TYPES: Final = frozenset({"enabled", "adaptive"})
_EVENT_STREAM: Final = "application/vnd.amazon.eventstream"
_USAGE: Final[dict[str, JsonValue]] = {"inputTokens": 11, "outputTokens": 7, "totalTokens": 18}
_MARKER: Final = re.compile(r"probe ([0-9a-f]{32})")
_JSON: Final = TypeAdapter(dict[str, JsonValue])
_JSON_VALUE: Final[TypeAdapter[JsonValue]] = TypeAdapter(JsonValue)
_FABLE_REJECTION: Final = (
"The model returned the following errors: output_config.format: Extra inputs are not permitted"
)
_UNSUPPORTED_KEYWORDS: Final = (
"minimum",
"maximum",
"exclusiveMinimum",
"exclusiveMaximum",
"multipleOf",
"maxItems",
"uniqueItems",
"minProperties",
"maxProperties",
"patternProperties",
"oneOf",
)
_SCHEMA_MAPS: Final = ("properties", "$defs", "definitions")
_SCHEMA_LISTS: Final = ("anyOf", "allOf", "oneOf")
_SCHEMA: Final[dict[str, JsonValue]] = {
"type": "object",
"properties": {"tier": {"type": "integer"}, "summary": {"type": "string"}},
"required": ["tier", "summary"],
"additionalProperties": False,
}
_LIMITS_SCHEMA: Final[dict[str, JsonValue]] = {
"type": "object",
"properties": {
"tier": {"type": "integer", "minimum": 1, "maximum": 5},
"summary": {"type": "string"},
"tags": {"type": "array", "items": {"type": "string"}, "minItems": 1},
"owner": {"anyOf": [{"$ref": "#/$defs/Person"}, {"type": "null"}]},
"backup": {"type": ["object", "null"], "properties": {"name": {"type": "string"}}, "required": ["name"]},
},
"required": ["tier", "summary", "tags", "owner", "backup"],
"$defs": {"Person": {"type": "object", "properties": {"name": {"type": "string"}}, "required": ["name"]}},
}
_OPEN_COUNTS: Final[dict[str, JsonValue]] = {"type": "object", "additionalProperties": {"type": "integer"}}
_DICT_SCHEMA: Final[dict[str, JsonValue]] = {
"type": "object",
"properties": {"store": {"type": "string"}, "counts": _OPEN_COUNTS},
"required": ["store", "counts"],
"additionalProperties": False,
}
_NULLABLE_SCHEMA: Final[dict[str, JsonValue]] = {
"type": "object",
"properties": {
"summary": {"type": "string"},
"backup": {"type": ["object", "null"], "properties": {"name": {"type": "string"}}, "required": ["name"]},
},
"required": ["summary", "backup"],
"additionalProperties": False,
}
_CLOSED_NULLABLE_SCHEMA: Final[dict[str, JsonValue]] = {
"type": "object",
"properties": {
"summary": {"type": "string"},
"backup": {
"type": ["object", "null"],
"properties": {"name": {"type": "string"}},
"required": ["name"],
"additionalProperties": False,
},
},
"required": ["summary", "backup"],
"additionalProperties": False,
}
_UNCLEANED_SCHEMA: Final[dict[str, JsonValue]] = {
"type": "object",
"properties": {"tier": {"type": "integer", "minimum": 1, "maximum": 5}, "summary": {"type": "string"}},
"required": ["tier", "summary"],
"additionalProperties": False,
}
_NUMERIC_FORMATS_SCHEMA: Final[dict[str, JsonValue]] = {
"type": "object",
"properties": {
"tier": {"type": "integer", "format": "int32"},
"score": {"type": "number", "format": "double"},
"summary": {"type": "string"},
},
"required": ["tier", "score", "summary"],
"additionalProperties": False,
}
_WIDE_FIELDS: Final = tuple(f"field_{index:02d}" for index in range(65))
_WIDE_SCHEMA: Final[dict[str, JsonValue]] = {
"type": "object",
"properties": {name: {"type": "string"} for name in _WIDE_FIELDS},
"required": list(_WIDE_FIELDS),
"additionalProperties": False,
}
_USER_TOOL_SPEC: Final[dict[str, JsonValue]] = {
"type": "function",
"function": {
"name": _USER_TOOL,
"description": "Look up a record",
"parameters": {"type": "object", "properties": {"q": {"type": "string"}}, "required": ["q"]},
},
}
def _payload(marker_hex: str) -> dict[str, JsonValue]:
return {"tier": 2, "summary": marker_hex}
def _marker() -> tuple[str, str]:
marker_hex: Final = uuid.uuid4().hex
return f"probe {marker_hex}", marker_hex
def _response_format(schema: Mapping[str, JsonValue]) -> dict[str, JsonValue]:
return {"type": "json_schema", "json_schema": {"name": "risk", "schema": dict(schema)}}
def _model_id(request: Request) -> str:
return unquote(request.target).split("/model/", 1)[1].rsplit("/", 1)[0]
def _is_invoke(request: Request) -> bool:
return unquote(request.target).rsplit("/", 1)[1].startswith("invoke")
def _is_stream(request: Request) -> bool:
return unquote(request.target).endswith(("/converse-stream", "/invoke-with-response-stream"))
def _thinking_of(body: Mapping[str, JsonValue], invoke: bool) -> JsonValue:
if invoke:
return body.get("thinking")
return object_value(body.get("additionalModelRequestFields") or {}).get("thinking")
def _thinking_requested(body: Mapping[str, JsonValue], invoke: bool) -> bool:
thinking: Final = _thinking_of(body, invoke)
return isinstance(thinking, dict) and thinking.get("type") in _THINKING_TYPES
def _native_schema(body: Mapping[str, JsonValue], invoke: bool) -> dict[str, JsonValue] | None:
if invoke:
output_config: Final = body.get("output_config")
if not isinstance(output_config, dict) or "format" not in output_config:
return None
return object_value(object_value(object_value(output_config)["format"])["schema"])
text_format: Final = object_value(body.get("outputConfig") or {}).get("textFormat")
if text_format is None:
return None
json_schema: Final = object_value(object_value(object_value(text_format)["structure"])["jsonSchema"])
return _JSON.validate_json(string_value(json_schema["schema"]))
def _tools(body: Mapping[str, JsonValue], invoke: bool) -> dict[str, dict[str, JsonValue]]:
if invoke:
return {
string_value(object_value(tool)["name"]): object_value(object_value(tool)["input_schema"])
for tool in list_value(body.get("tools", []))
}
specs: Final = tuple(
object_value(object_value(tool)["toolSpec"])
for tool in list_value(object_value(body.get("toolConfig") or {"tools": []})["tools"])
)
return {string_value(spec["name"]): object_value(object_value(spec["inputSchema"])["json"]) for spec in specs}
def _is_object_type(node: Mapping[str, JsonValue]) -> bool:
node_type: Final = node.get("type")
return node_type == "object" or (isinstance(node_type, list) and "object" in node_type)
def _child_schemas(node: Mapping[str, JsonValue]) -> Iterable[JsonValue]:
mapped: Final = (node.get(key) for key in _SCHEMA_MAPS)
listed: Final = (node.get(key) for key in _SCHEMA_LISTS)
single: Final = (node.get(key) for key in ("items", "additionalProperties"))
return chain(
chain.from_iterable(children.values() for children in mapped if isinstance(children, dict)),
chain.from_iterable(variants for variants in listed if isinstance(variants, list)),
(child for child in single if isinstance(child, dict)),
)
def _schema_nodes(schema: JsonValue) -> Iterator[dict[str, JsonValue]]:
if not isinstance(schema, dict):
return
yield schema
yield from chain.from_iterable(_schema_nodes(child) for child in _child_schemas(schema))
def _node_fault(node: Mapping[str, JsonValue]) -> str | None:
node_type: Final = node.get("type")
unsupported: Final = tuple(key for key in _UNSUPPORTED_KEYWORDS if key in node)
if unsupported:
return f"For '{node_type}' type, property '{unsupported[0]}' is not supported"
min_items: Final = node.get("minItems")
if isinstance(min_items, int) and min_items > 1:
return "For 'array' type, property 'minItems' is not supported"
if _is_object_type(node) and node.get("additionalProperties") is not False:
return "For 'object' type, 'additionalProperties' must be explicitly set to false"
return None
def _bedrock_schema_fault(schema: JsonValue) -> str | None:
faults: Final = tuple(fault for node in _schema_nodes(schema) if (fault := _node_fault(node)) is not None)
if not faults:
return None
return f"The model returned the following errors: output_config.format.schema: {faults[0]}"
def _rejection(model_id: str, native: dict[str, JsonValue] | None) -> str | None:
if native is None:
return None
if model_id == _FABLE:
return _FABLE_REJECTION
if "anthropic." not in model_id:
return None
return _bedrock_schema_fault(native)
def _reasoning_block(marker_hex: str, invoke: bool) -> dict[str, JsonValue]:
if invoke:
return {"type": "thinking", "thinking": f"reasoning {marker_hex}", "signature": _SIGNATURE}
return {"reasoningContent": {"reasoningText": {"text": f"reasoning {marker_hex}", "signature": _SIGNATURE}}}
def _answer_block(name: str | None, marker_hex: str, invoke: bool) -> dict[str, JsonValue]:
tool_input: Final[dict[str, JsonValue]] = {"q": marker_hex} if name == _USER_TOOL else _payload(marker_hex)
if name is None and invoke:
return {"type": "text", "text": json.dumps(_payload(marker_hex))}
if name is None:
return {"text": json.dumps(_payload(marker_hex))}
if invoke:
return {"type": "tool_use", "id": f"toolu_{marker_hex}", "name": name, "input": tool_input}
return {"toolUse": {"toolUseId": f"tooluse_{marker_hex}", "name": name, "input": tool_input}}
def _converse_frames(blocks: tuple[dict[str, JsonValue], ...], stop_reason: str) -> bytes:
def frame(event_type: str, payload: Mapping[str, JsonValue]) -> bytes:
return aws_event_stream_frame(
{":event-type": event_type, ":content-type": "application/json", ":message-type": "event"},
json.dumps(payload).encode(),
)
def block_frames(index: int, block: dict[str, JsonValue]) -> bytes:
if "reasoningContent" in block:
reasoning: Final = object_value(object_value(block["reasoningContent"])["reasoningText"])
return b"".join(
(
frame(
"contentBlockDelta",
{"delta": {"reasoningContent": {"text": reasoning["text"]}}, "contentBlockIndex": index},
),
frame(
"contentBlockDelta",
{
"delta": {"reasoningContent": {"signature": reasoning["signature"]}},
"contentBlockIndex": index,
},
),
frame("contentBlockStop", {"contentBlockIndex": index}),
)
)
if "toolUse" in block:
tool_use: Final = object_value(block["toolUse"])
return b"".join(
(
frame(
"contentBlockStart",
{
"start": {"toolUse": {"toolUseId": tool_use["toolUseId"], "name": tool_use["name"]}},
"contentBlockIndex": index,
},
),
frame(
"contentBlockDelta",
{"delta": {"toolUse": {"input": json.dumps(tool_use["input"])}}, "contentBlockIndex": index},
),
frame("contentBlockStop", {"contentBlockIndex": index}),
)
)
return b"".join(
(
frame("contentBlockDelta", {"delta": {"text": block["text"]}, "contentBlockIndex": index}),
frame("contentBlockStop", {"contentBlockIndex": index}),
)
)
return b"".join(
(
frame("messageStart", {"role": "assistant"}),
*(block_frames(index, block) for index, block in enumerate(blocks)),
frame("messageStop", {"stopReason": stop_reason}),
frame("metadata", {"usage": _USAGE, "metrics": {"latencyMs": 1}}),
)
)
def _called_tool(tools: Iterable[str], native: bool) -> str | None:
names: Final = frozenset(tools)
if _USER_TOOL in names:
return _USER_TOOL
if native or _JSON_TOOL not in names:
return None
return _JSON_TOOL
def _error_reply(message: str) -> Reply:
return Reply(status=400, body=json.dumps({"message": message}).encode())
def _bedrock(request: Request) -> Reply:
body: Final = _JSON.validate_json(request.body)
invoke: Final = _is_invoke(request)
found: Final = _MARKER.search(json.dumps(body))
assert found is not None, body
marker_hex: Final = found.group(1)
native: Final = _native_schema(body, invoke)
rejection: Final = _rejection(_model_id(request), native)
if rejection is not None:
return _error_reply(rejection)
called: Final = _called_tool(_tools(body, invoke).keys(), native is not None)
reasoning: Final = (_reasoning_block(marker_hex, invoke),) if _thinking_requested(body, invoke) else ()
blocks: Final = (*reasoning, _answer_block(called, marker_hex, invoke))
stop_reason: Final = "tool_use" if called is not None else "end_turn"
if invoke:
message: Final[dict[str, JsonValue]] = {
"id": f"msg_{marker_hex}",
"type": "message",
"role": "assistant",
"model": "claude-scripted",
"content": list(blocks),
"stop_reason": stop_reason,
"stop_sequence": None,
"usage": {"input_tokens": 11, "output_tokens": 7},
}
return Reply(body=json.dumps(message).encode())
if _is_stream(request):
return Reply(body=_converse_frames(blocks, stop_reason), content_type=_EVENT_STREAM)
converse: Final[dict[str, JsonValue]] = {
"output": {"message": {"role": "assistant", "content": list(blocks)}},
"stopReason": stop_reason,
"usage": _USAGE,
"metrics": {"latencyMs": 1},
}
return Reply(body=json.dumps(converse).encode())
def _deployment(
scenario: Scenario, wire: Wire, route: str, backend: str, model_info: Mapping[str, JsonValue] | None = None
) -> str:
return scenario.model(model=f"bedrock/{route}/{backend}", api_base=wire.url, model_info=model_info, **_AWS)
def _chat_body(model: str, marker: str, **extra: JsonValue) -> dict[str, JsonValue]:
return {"model": model, "messages": [{"role": "user", "content": marker}], "max_tokens": 2048, **_NO_CACHE, **extra}
def _messages_body(model: str, marker: str, **extra: JsonValue) -> dict[str, JsonValue]:
return {"model": model, "max_tokens": 2048, "messages": [{"role": "user", "content": marker}], **_NO_CACHE, **extra}
def _one_call(wire: Wire, marker_hex: str) -> tuple[Request, dict[str, JsonValue]]:
(request,) = wire.drain()
body: Final = _JSON.validate_json(request.body)
assert marker_hex in json.dumps(body), body
return request, body
def _chat_message(response: httpx.Response) -> dict[str, JsonValue]:
assert response.status_code == 200, response.text
choices: Final = list_value(_JSON.validate_json(response.content)["choices"])
return object_value(object_value(choices[0])["message"])
def _stream_message(response: httpx.Response) -> dict[str, JsonValue]:
assert response.status_code == 200, response.text
deltas: Final = tuple(
object_value(object_value(list_value(_JSON.validate_json(line[6:])["choices"])[0])["delta"])
for line in response.text.splitlines()
if line.startswith("data: {")
)
return {
"content": "".join(string_value(delta["content"]) for delta in deltas if delta.get("content")),
"reasoning_content": "".join(
string_value(delta["reasoning_content"]) for delta in deltas if delta.get("reasoning_content")
),
}
def _content_json(message: Mapping[str, JsonValue]) -> JsonValue:
return _JSON_VALUE.validate_json(string_value(message["content"]))
def _responses_items(response: httpx.Response) -> tuple[dict[str, JsonValue], ...]:
assert response.status_code == 200, response.text
return tuple(object_value(item) for item in list_value(_JSON.validate_json(response.content)["output"]))
def _responses_text(items: tuple[dict[str, JsonValue], ...]) -> JsonValue:
(message,) = tuple(item for item in items if item.get("type") == "message")
(text,) = tuple(
object_value(part) for part in list_value(message["content"]) if object_value(part).get("type") == "output_text"
)
return _JSON_VALUE.validate_json(string_value(text["text"]))
def _messages_blocks(response: httpx.Response) -> tuple[dict[str, JsonValue], ...]:
assert response.status_code == 200, response.text
return tuple(object_value(block) for block in list_value(_JSON.validate_json(response.content)["content"]))
def _messages_text(blocks: tuple[dict[str, JsonValue], ...]) -> JsonValue:
(text,) = tuple(block for block in blocks if block.get("type") == "text")
return _JSON_VALUE.validate_json(string_value(text["text"]))
def _refs(schema: JsonValue) -> frozenset[str]:
return frozenset(string_value(node["$ref"]) for node in _schema_nodes(schema) if "$ref" in node)
def _resolves(schema: Mapping[str, JsonValue], ref: str) -> bool:
prefix: Final = "#/$defs/"
return ref.startswith(prefix) and ref.removeprefix(prefix) in object_value(schema.get("$defs") or {})
def _assert_native_without_tools(body: Mapping[str, JsonValue], invoke: bool, schema: Mapping[str, JsonValue]) -> None:
assert _native_schema(body, invoke) == schema, body
assert _JSON_TOOL not in _tools(body, invoke), body
@pytest.mark.parametrize(
("route", "stream"),
[("converse", False), ("converse", True), ("invoke", False)],
ids=["converse", "converse_stream", "invoke"],
)
def test_thinking_request_on_a_bedrock_native_model_gets_native_output_and_keeps_reasoning(
gateway: Gateway, route: str, stream: bool
) -> None:
marker, marker_hex = _marker()
with wire_server(_bedrock) as wire, gateway.scenario() as scenario:
model: Final = _deployment(scenario, wire, route, _HAIKU)
response: Final = gateway.request(
"POST",
"/v1/chat/completions",
_chat_body(model, marker, stream=stream, thinking=_ADAPTIVE, response_format=_response_format(_SCHEMA)),
)
message: Final = _stream_message(response) if stream else _chat_message(response)
assert _content_json(message) == _payload(marker_hex), response.text
assert message["reasoning_content"] == f"reasoning {marker_hex}", response.text
request, body = _one_call(wire, marker_hex)
invoke: Final = route == "invoke"
assert _is_stream(request) is stream, request.target
_assert_native_without_tools(body, invoke, _SCHEMA)
assert _thinking_requested(body, invoke), body
def test_responses_reasoning_effort_on_a_bedrock_native_model_gets_native_output_and_keeps_reasoning(
gateway: Gateway,
) -> None:
marker, marker_hex = _marker()
with wire_server(_bedrock) as wire, gateway.scenario() as scenario:
model: Final = _deployment(scenario, wire, "converse", _HAIKU)
response: Final = gateway.request(
"POST",
"/v1/responses",
{
"model": model,
"input": marker,
"max_output_tokens": 2048,
"reasoning": {"effort": "low"},
"text": {"format": {"type": "json_schema", "name": "risk", "schema": _SCHEMA}},
**_NO_CACHE,
},
)
items: Final = _responses_items(response)
assert _responses_text(items) == _payload(marker_hex), response.text
(reasoning,) = tuple(item for item in items if item.get("type") == "reasoning")
assert f"reasoning {marker_hex}" in json.dumps(reasoning), response.text
_, body = _one_call(wire, marker_hex)
_assert_native_without_tools(body, False, _SCHEMA)
assert _thinking_requested(body, False), body
def test_messages_thinking_on_a_bedrock_converse_native_model_gets_native_output_and_keeps_reasoning(
gateway: Gateway,
) -> None:
marker, marker_hex = _marker()
with wire_server(_bedrock) as wire, gateway.scenario() as scenario:
model: Final = _deployment(scenario, wire, "converse", _HAIKU)
response: Final = gateway.request(
"POST",
"/v1/messages",
_messages_body(model, marker, thinking=_ADAPTIVE, output_format={"type": "json_schema", "schema": _SCHEMA}),
)
blocks: Final = _messages_blocks(response)
assert _messages_text(blocks) == _payload(marker_hex), response.text
(thinking,) = tuple(block for block in blocks if block.get("type") == "thinking")
assert thinking["thinking"] == f"reasoning {marker_hex}", response.text
_, body = _one_call(wire, marker_hex)
_assert_native_without_tools(body, False, _SCHEMA)
assert _thinking_requested(body, False), body
@pytest.mark.parametrize("route", ["converse", "invoke"])
def test_bedrock_native_model_without_thinking_keeps_the_forced_json_tool(gateway: Gateway, route: str) -> None:
marker, marker_hex = _marker()
with wire_server(_bedrock) as wire, gateway.scenario() as scenario:
model: Final = _deployment(scenario, wire, route, _HAIKU)
response: Final = gateway.request(
"POST", "/v1/chat/completions", _chat_body(model, marker, response_format=_response_format(_SCHEMA))
)
message: Final = _chat_message(response)
assert _content_json(message) == _payload(marker_hex), response.text
assert not message.get("tool_calls"), response.text
_, body = _one_call(wire, marker_hex)
invoke: Final = route == "invoke"
assert _native_schema(body, invoke) is None, body
assert _tools(body, invoke)[_JSON_TOOL]["properties"] == _SCHEMA["properties"], body
@pytest.mark.parametrize(
("backend", "model_info"),
[(_OPUS, None), (_OPT_IN, {"supports_bedrock_native_structured_output": True})],
ids=["neither_key", "deployment_opt_in"],
)
def test_thinking_request_follows_the_bedrock_native_key_the_model_resolves_to(
gateway: Gateway, backend: str, model_info: Mapping[str, JsonValue] | None
) -> None:
marker, marker_hex = _marker()
native: Final = model_info is not None
with wire_server(_bedrock) as wire, gateway.scenario() as scenario:
model: Final = _deployment(scenario, wire, "converse", backend, model_info)
response: Final = gateway.request(
"POST",
"/v1/chat/completions",
_chat_body(model, marker, thinking=_ADAPTIVE, response_format=_response_format(_SCHEMA)),
)
assert _content_json(_chat_message(response)) == _payload(marker_hex), response.text
_, body = _one_call(wire, marker_hex)
assert (_native_schema(body, False) == _SCHEMA) is native, body
assert (_JSON_TOOL in _tools(body, False)) is not native, body
def _opt_in_answer(gateway: Gateway, entry: str, model: str, marker: str) -> JsonValue:
output_format: Final[dict[str, JsonValue]] = {"type": "json_schema", "schema": _SCHEMA}
match entry:
case "converse_chat" | "invoke_chat":
chat: Final = _chat_body(model, marker, thinking=_ADAPTIVE, response_format=_response_format(_SCHEMA))
return _content_json(_chat_message(gateway.request("POST", "/v1/chat/completions", chat)))
case "converse_responses":
responses: Final[dict[str, JsonValue]] = {
"model": model,
"input": marker,
"max_output_tokens": 2048,
"reasoning": {"effort": "low"},
"text": {"format": {**output_format, "name": "risk"}},
**_NO_CACHE,
}
return _responses_text(_responses_items(gateway.request("POST", "/v1/responses", responses)))
case "converse_messages":
adapted: Final = _messages_body(model, marker, thinking=_ADAPTIVE, output_format=output_format)
return _messages_text(_messages_blocks(gateway.request("POST", "/v1/messages", adapted)))
case _:
invoked: Final = _messages_body(model, marker, output_config={"format": output_format})
return _messages_text(_messages_blocks(gateway.request("POST", "/v1/messages", invoked)))
def _sent_native_schema(gateway: Gateway, wire: Wire, entry: str, model: str) -> dict[str, JsonValue] | None:
marker, marker_hex = _marker()
assert _opt_in_answer(gateway, entry, model, marker) == _payload(marker_hex)
_, body = _one_call(wire, marker_hex)
invoke: Final = entry.startswith("invoke")
native: Final = _native_schema(body, invoke)
assert native is None or _JSON_TOOL not in _tools(body, invoke), body
return native
@pytest.mark.parametrize(
"entry", ["converse_chat", "converse_responses", "converse_messages", "invoke_chat", "invoke_messages"]
)
def test_a_deployment_native_output_opt_in_stays_on_the_deployment_that_sets_it(gateway: Gateway, entry: str) -> None:
route: Final = entry.split("_", 1)[0]
with wire_server(_bedrock) as wire, gateway.scenario() as scenario:
opted_in: Final = _deployment(
scenario, wire, route, _OPT_IN, {"supports_bedrock_native_structured_output": True}
)
sibling: Final = _deployment(scenario, wire, route, _OPT_IN)
assert _sent_native_schema(gateway, wire, entry, sibling) is None
assert _sent_native_schema(gateway, wire, entry, opted_in) == _SCHEMA
def test_fable_5_json_schema_takes_the_json_tool_instead_of_the_native_output_bedrock_rejects(
gateway: Gateway,
) -> None:
marker, marker_hex = _marker()
with wire_server(_bedrock) as wire, gateway.scenario() as scenario:
model: Final = _deployment(scenario, wire, "converse", _FABLE)
response: Final = gateway.request(
"POST", "/v1/chat/completions", _chat_body(model, marker, response_format=_response_format(_SCHEMA))
)
assert _content_json(_chat_message(response)) == _payload(marker_hex), response.text
_, body = _one_call(wire, marker_hex)
assert "outputConfig" not in body, body
assert _tools(body, False)[_JSON_TOOL]["properties"] == _SCHEMA["properties"], body
def _limits_answer(gateway: Gateway, entry: str, model: str, marker: str) -> JsonValue:
output_config: Final[dict[str, JsonValue]] = {"format": {"type": "json_schema", "schema": _LIMITS_SCHEMA}}
match entry:
case "converse_chat" | "invoke_chat":
body: Final = _chat_body(model, marker, response_format=_response_format(_LIMITS_SCHEMA))
return _content_json(_chat_message(gateway.request("POST", "/v1/chat/completions", body)))
case "converse_output_config":
chat: Final = _chat_body(model, marker, output_config=output_config)
return _content_json(_chat_message(gateway.request("POST", "/v1/chat/completions", chat)))
case _:
messages: Final = _messages_body(model, marker, output_config=output_config)
return _messages_text(_messages_blocks(gateway.request("POST", "/v1/messages", messages)))
@pytest.mark.parametrize("entry", ["converse_chat", "converse_output_config", "invoke_chat", "invoke_messages"])
def test_native_schema_reaches_bedrock_with_limits_in_descriptions_and_every_object_closed(
gateway: Gateway, entry: str
) -> None:
marker, marker_hex = _marker()
route: Final = entry.split("_", 1)[0]
with wire_server(_bedrock) as wire, gateway.scenario() as scenario:
model: Final = _deployment(scenario, wire, route, _SONNET)
assert _limits_answer(gateway, entry, model, marker) == _payload(marker_hex)
_, body = _one_call(wire, marker_hex)
schema: Final = _native_schema(body, route == "invoke")
assert schema is not None, body
assert _bedrock_schema_fault(schema) is None, schema
properties: Final = object_value(schema["properties"])
assert object_value(properties["tags"])["minItems"] == 1, schema
assert {"1", "5"} <= set(re.findall(r"\d+", string_value(object_value(properties["tier"])["description"])))
assert all(_resolves(schema, ref) for ref in _refs(schema)), schema
assert _JSON_TOOL not in _tools(body, route == "invoke"), body
@pytest.mark.parametrize("route", ["converse", "invoke"])
def test_open_object_schema_takes_the_json_tool_on_a_native_model_and_keeps_the_dict_open(
gateway: Gateway, route: str
) -> None:
marker, marker_hex = _marker()
with wire_server(_bedrock) as wire, gateway.scenario() as scenario:
model: Final = _deployment(scenario, wire, route, _SONNET)
response: Final = gateway.request(
"POST", "/v1/chat/completions", _chat_body(model, marker, response_format=_response_format(_DICT_SCHEMA))
)
assert _content_json(_chat_message(response)) == _payload(marker_hex), response.text
_, body = _one_call(wire, marker_hex)
invoke: Final = route == "invoke"
assert _native_schema(body, invoke) is None, body
tool_schema: Final = _tools(body, invoke)[_JSON_TOOL]
assert object_value(tool_schema["properties"])["counts"] == _OPEN_COUNTS, tool_schema
def test_thinking_with_a_caller_tool_on_a_bedrock_native_model_sends_only_the_callers_tool(gateway: Gateway) -> None:
marker, marker_hex = _marker()
with wire_server(_bedrock) as wire, gateway.scenario() as scenario:
model: Final = _deployment(scenario, wire, "converse", _HAIKU)
response: Final = gateway.request(
"POST",
"/v1/chat/completions",
_chat_body(
model,
marker,
thinking=_ADAPTIVE,
tools=[_USER_TOOL_SPEC],
response_format=_response_format(_SCHEMA),
),
)
(call,) = list_value(_chat_message(response)["tool_calls"])
function: Final = object_value(object_value(call)["function"])
assert function["name"] == _USER_TOOL, response.text
assert json.loads(string_value(function["arguments"])) == {"q": marker_hex}, response.text
_, body = _one_call(wire, marker_hex)
assert _tools(body, False).keys() == {_USER_TOOL}, body
assert _native_schema(body, False) == _SCHEMA, body
assert _thinking_requested(body, False), body
@pytest.mark.parametrize(
("schema", "expected"),
[(_UNCLEANED_SCHEMA, _UNCLEANED_SCHEMA), (_NULLABLE_SCHEMA, _CLOSED_NULLABLE_SCHEMA)],
ids=["limits_kept", "nullable_object_closed"],
)
def test_non_anthropic_native_model_gets_the_callers_schema_with_objects_closed(
gateway: Gateway, schema: dict[str, JsonValue], expected: dict[str, JsonValue]
) -> None:
marker, marker_hex = _marker()
with wire_server(_bedrock) as wire, gateway.scenario() as scenario:
model: Final = _deployment(scenario, wire, "converse", _DEEPSEEK)
response: Final = gateway.request(
"POST", "/v1/chat/completions", _chat_body(model, marker, response_format=_response_format(schema))
)
assert _content_json(_chat_message(response)) == _payload(marker_hex), response.text
_, body = _one_call(wire, marker_hex)
assert _native_schema(body, False) == expected, body
@pytest.mark.parametrize(
("schema", "native"),
[
(_SCHEMA, True),
(_DICT_SCHEMA, False),
],
ids=["closed_schema_native", "open_object_inlined"],
)
def test_messages_on_bedrock_invoke_follows_the_bedrock_native_key_without_thinking(
gateway: Gateway, schema: dict[str, JsonValue], native: bool
) -> None:
marker, marker_hex = _marker()
with wire_server(_bedrock) as wire, gateway.scenario() as scenario:
model: Final = _deployment(scenario, wire, "invoke", _HAIKU)
response: Final = gateway.request(
"POST",
"/v1/messages",
_messages_body(model, marker, output_config={"format": {"type": "json_schema", "schema": schema}}),
)
assert _messages_text(_messages_blocks(response)) == _payload(marker_hex), response.text
_, body = _one_call(wire, marker_hex)
sent: Final = _native_schema(body, True)
assert (sent is not None) is native, body
prompt: Final = json.dumps(body["messages"])
assert all((name in prompt) is not native for name in object_value(schema["properties"])), body
@pytest.mark.parametrize(
"schema",
[
{"type": "object", "properties": ["tier", "summary"]},
{"type": "object", "properties": {"tier": {"$ref": 5}}, "additionalProperties": False},
{"type": "object", "$defs": [1, 2], "properties": {"tier": {"$ref": "#/$defs/Tier"}}},
{"type": ["object", "null"]},
{"type": "object", "properties": {"tier": {"type": "integer"}}, "required": "tier"},
],
ids=["properties_list", "ref_not_string", "defs_list", "type_list_without_properties", "required_string"],
)
def test_malformed_schema_on_a_native_model_reaches_bedrock_once_without_a_proxy_error(
gateway: Gateway, schema: dict[str, JsonValue]
) -> None:
marker, marker_hex = _marker()
with wire_server(_bedrock) as wire, gateway.scenario() as scenario:
model: Final = _deployment(scenario, wire, "converse", _SONNET)
response: Final = gateway.request(
"POST", "/v1/chat/completions", _chat_body(model, marker, response_format=_response_format(schema))
)
assert response.status_code < 500, response.text
_, body = _one_call(wire, marker_hex)
native: Final = _native_schema(body, False)
fault: Final = None if native is None else _bedrock_schema_fault(native)
assert response.status_code == (200 if fault is None else 400), response.text
assert fault is None or fault.split(": ", 1)[1] in response.text, response.text
assert (native is None) is (_JSON_TOOL in _tools(body, False)), body
assert gateway.request("GET", "/health/liveliness").status_code == 200
@pytest.mark.parametrize(
("route", "backend", "thinking", "schema", "native"),
[
("converse", _SONNET, False, _NUMERIC_FORMATS_SCHEMA, True),
("invoke", _HAIKU, True, _NUMERIC_FORMATS_SCHEMA, True),
("converse", _SONNET, False, _WIDE_SCHEMA, True),
("converse", _HAIKU, True, _WIDE_SCHEMA, False),
],
ids=["numeric_formats_old_key", "numeric_formats_new_key_invoke", "wide_old_key", "wide_new_key"],
)
def test_numeric_formats_and_wide_schemas_follow_the_bedrock_native_key(
gateway: Gateway, route: str, backend: str, thinking: bool, schema: dict[str, JsonValue], native: bool
) -> None:
marker, marker_hex = _marker()
extra: Final[dict[str, JsonValue]] = {"thinking": _ADAPTIVE} if thinking else {}
with wire_server(_bedrock) as wire, gateway.scenario() as scenario:
model: Final = _deployment(scenario, wire, route, backend)
response: Final = gateway.request(
"POST",
"/v1/chat/completions",
_chat_body(model, marker, response_format=_response_format(schema), **extra),
)
assert _content_json(_chat_message(response)) == _payload(marker_hex), response.text
_, body = _one_call(wire, marker_hex)
invoke: Final = route == "invoke"
assert (_native_schema(body, invoke) == schema) is native, body
tools: Final = _tools(body, invoke)
assert (_JSON_TOOL in tools) is not native, body
assert native or tools[_JSON_TOOL]["properties"] == schema["properties"], body

File diff suppressed because it is too large Load diff

View file

@ -0,0 +1,139 @@
import json
import re
import uuid
from collections.abc import Callable, Iterator, Mapping
from typing import Final
import pytest
from integration._support.wire import Reply, Request, Wire, wire_server
from pydantic import JsonValue, TypeAdapter
import litellm
from litellm import ModelResponse, Router
from litellm.utils import _invalidate_model_cost_lowercase_map # pyright: ignore[reportPrivateUsage] # required cache invalidation
_OPUS_4_8: Final = "us.anthropic.claude-opus-4-8"
_JSON_TOOL: Final = "json_tool_call"
_MARKER: Final = re.compile(r"probe ([0-9a-f]{32})")
_SCHEMA: Final[dict[str, JsonValue]] = {
"type": "object",
"properties": {"city": {"type": "string"}},
"required": ["city"],
"additionalProperties": False,
}
_JSON: Final = TypeAdapter(dict[str, JsonValue])
_COST_MAP: Final = TypeAdapter(dict[str, dict[str, object]])
_OPTIONS: Final[dict[str, JsonValue]] = {
"max_tokens": 2048,
"thinking": {"type": "adaptive"},
"response_format": {"type": "json_schema", "json_schema": {"name": "city", "schema": _SCHEMA}},
"num_retries": 0,
}
def _native(body: Mapping[str, JsonValue], invoke: bool) -> bool:
output_config: Final = body.get("output_config" if invoke else "outputConfig")
return isinstance(output_config, dict) and ("format" if invoke else "textFormat") in output_config
def _invoke_reply(answer: dict[str, JsonValue], native: bool) -> Reply:
block: Final[dict[str, JsonValue]] = (
{"type": "text", "text": json.dumps(answer)}
if native
else {"type": "tool_use", "id": "toolu_scripted", "name": _JSON_TOOL, "input": answer}
)
message: Final[dict[str, JsonValue]] = {
"id": f"msg_{uuid.uuid4().hex}",
"type": "message",
"role": "assistant",
"model": "claude-scripted",
"content": [block],
"stop_reason": "end_turn" if native else "tool_use",
"stop_sequence": None,
"usage": {"input_tokens": 11, "output_tokens": 7},
}
return Reply(body=json.dumps(message).encode())
def _converse_reply(answer: dict[str, JsonValue], native: bool) -> Reply:
block: Final[dict[str, JsonValue]] = (
{"text": json.dumps(answer)}
if native
else {"toolUse": {"toolUseId": "tooluse_scripted", "name": _JSON_TOOL, "input": answer}}
)
converse: Final[dict[str, JsonValue]] = {
"output": {"message": {"role": "assistant", "content": [block]}},
"stopReason": "end_turn" if native else "tool_use",
"usage": {"inputTokens": 11, "outputTokens": 7, "totalTokens": 18},
"metrics": {"latencyMs": 1},
}
return Reply(body=json.dumps(converse).encode())
def _bedrock(request: Request) -> Reply:
body: Final = _JSON.validate_json(request.body)
found: Final = _MARKER.search(json.dumps(body))
assert found is not None, body
invoke: Final = "/invoke" in request.target
reply: Final[Callable[[dict[str, JsonValue], bool], Reply]] = _invoke_reply if invoke else _converse_reply
return reply({"city": found.group(1)}, _native(body, invoke))
@pytest.fixture
def opus_4_8_opted_in_for_the_process(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
monkeypatch.setattr(litellm, "model_cost", _COST_MAP.validate_python(litellm.model_cost)) # pyright: ignore[reportUnknownMemberType] # the cost map is untyped
litellm.register_model(
{_OPUS_4_8: {"supports_bedrock_native_structured_output": True}}, persist_across_reloads=False
)
yield
monkeypatch.undo()
_invalidate_model_cost_lowercase_map()
def _model(route: str) -> str:
return f"bedrock/{route}/{_OPUS_4_8}"
def _connection(wire: Wire) -> dict[str, JsonValue]:
return {"api_base": wire.url, "api_key": "synthetic-bedrock-bearer", "aws_region_name": "us-east-1"}
def _sent_native(wire: Wire, route: str, call: Callable[[list[dict[str, str]]], object]) -> bool:
marker_hex: Final = uuid.uuid4().hex
response: Final = call([{"role": "user", "content": f"Name a city. probe {marker_hex}"}])
assert isinstance(response, ModelResponse), response
assert json.loads(str(response.choices[0].message.content)) == {"city": marker_hex}, response # pyright: ignore[reportAttributeAccessIssue] # choices also types streaming chunks
(request,) = wire.drain()
return _native(_JSON.validate_json(request.body), route == "invoke")
@pytest.mark.parametrize("route", ["converse", "invoke"])
@pytest.mark.usefixtures("opus_4_8_opted_in_for_the_process")
def test_a_register_model_native_output_opt_in_reaches_every_call_but_a_deployment_that_opts_out(route: str) -> None:
with wire_server(_bedrock) as wire:
router: Final = Router(
model_list=[
{
"model_name": "opted-out",
"litellm_params": {"model": _model(route), **_connection(wire)},
"model_info": {"supports_bedrock_native_structured_output": False},
},
{"model_name": "inherits", "litellm_params": {"model": _model(route), **_connection(wire)}},
],
num_retries=0,
)
assert _sent_native(
wire,
route,
lambda messages: router.completion(model="inherits", messages=messages, **_OPTIONS), # pyright: ignore[reportUnknownMemberType] # Router.completion leaves **kwargs untyped
)
assert not _sent_native(
wire,
route,
lambda messages: router.completion(model="opted-out", messages=messages, **_OPTIONS), # pyright: ignore[reportUnknownMemberType] # Router.completion leaves **kwargs untyped
)
assert _sent_native(
wire,
route,
lambda messages: litellm.completion(_model(route), messages, **_connection(wire), **_OPTIONS),
)

View file

@ -2,10 +2,10 @@ import asyncio
import base64
import importlib
import os
from collections.abc import Coroutine, Iterator
from collections.abc import Callable, Coroutine, Iterator
from dataclasses import dataclass, field
from pathlib import Path
from typing import Final
from typing import Final, Literal, TypeAlias
import boto3
import httpx
@ -299,6 +299,22 @@ def local_model_cost_map(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
litellm.get_model_info.cache_clear()
NativeStructuredOutputKey: TypeAlias = Literal["legacy", "bedrock"] | None
@pytest.fixture
def set_native_structured_output(
local_model_cost_map: None, monkeypatch: pytest.MonkeyPatch
) -> Callable[[str, NativeStructuredOutputKey], None]:
def set_flags(model_cost_key: str, native_key: NativeStructuredOutputKey) -> None:
entry: Final = litellm.model_cost[model_cost_key]
monkeypatch.setitem(entry, "supports_native_structured_output", native_key == "legacy")
monkeypatch.setitem(entry, "supports_bedrock_native_structured_output", native_key == "bedrock")
litellm.get_model_info.cache_clear()
return set_flags
@pytest.fixture
def local_beta_headers_config(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
monkeypatch.setenv("LITELLM_LOCAL_ANTHROPIC_BETA_HEADERS", "True")

View file

@ -3,6 +3,7 @@ import functools
import json
import os
import sys
from collections.abc import Mapping
from typing import Final
import pytest
@ -21,6 +22,11 @@ from litellm.litellm_core_utils.prompt_templates.common_utils import (
is_encrypted_reasoning_block,
merge_consecutive_system_messages,
responses_reasoning_items_from_thinking_blocks,
salvage_concatenated_tool_arguments,
schema_grammar_branch_count,
schema_has_open_object,
schema_has_unresolvable_ref,
schema_property_count,
split_concatenated_json_objects,
strip_encrypted_reasoning_from_messages,
system_messages_first,
@ -1952,3 +1958,76 @@ class TestMergeConsecutiveSystemMessages:
merged = merge_consecutive_system_messages([{"role": "system"}, {"role": "system"}, {"role": "user", "content": "Hi"}])
assert merged == [{"role": "system"}, {"role": "user", "content": "Hi"}]
@pytest.mark.parametrize(
("schema", "is_open"),
[
({"type": "object", "properties": {"a": {"type": "string"}}}, False),
({"type": "object", "properties": {}}, False),
({"type": "object", "additionalProperties": False}, False),
({"type": "string"}, False),
({"type": "object"}, True),
({"type": ["object", "null"]}, True),
({"type": "object", "properties": {"a": {"type": "string"}}, "additionalProperties": True}, True),
({"type": "object", "additionalProperties": {"type": "integer"}}, True),
({"type": "object", "properties": {}, "patternProperties": {"^x": {"type": "string"}}}, True),
({"type": "object", "properties": {"inner": {"anyOf": [{"type": "object"}, {"type": "null"}]}}}, True),
({"type": "array", "items": {"type": "object", "additionalProperties": {"type": "string"}}}, True),
],
)
def test_schema_has_open_object_flags_any_node_that_accepts_unlisted_keys(schema, is_open):
assert schema_has_open_object(schema) is is_open
def _object_with(properties: Mapping[str, object], **defs: Mapping[str, object]) -> Mapping[str, object]:
return {"type": "object", "properties": properties, **defs}
@pytest.mark.parametrize(
("schema", "unresolvable"),
[
(_object_with({"a": {"type": "string"}}), False),
(_object_with({"a": {"$ref": "#/$defs/A"}, "b": {"$ref": "#/$defs/A"}}, **{"$defs": {"A": {"type": "string"}}}), False),
(_object_with({"a": {"$ref": "#/definitions/A"}}, definitions={"A": {"$ref": "#/definitions/B"}, "B": {"type": "integer"}}), False),
(_object_with({"a": {"$ref": "/$defs/A"}}, **{"$defs": {"A": {"items": {"$ref": "/$defs/B"}}, "B": {"type": "string"}}}), False),
(_object_with({"a": {"$ref": "#/$defs/Missing"}}, **{"$defs": {"A": {"type": "string"}}}), True),
(_object_with({"a": {"$ref": "other.json#/A"}}), True),
(_object_with({"parent": {"$ref": "#"}}), True),
(_object_with({"a": {"$ref": "#/$defs/A"}}, **{"$defs": {"A": {"type": "array", "items": {"$ref": "#/$defs/A"}}}}), True),
(_object_with({"a": {"$ref": "/$defs/A"}}, **{"$defs": {"A": {"type": "array", "items": {"$ref": "/$defs/A"}}}}), True),
(
_object_with(
{"a": {"$ref": "#/$defs/A"}},
**{"$defs": {"A": {"properties": {"b": {"$ref": "#/$defs/B"}}}, "B": {"properties": {"a": {"$ref": "#/$defs/A"}}}}},
),
True,
),
],
ids=["no-refs", "shared-def", "chained-definitions", "root-path-defs", "missing-def", "external", "root", "self-cycle", "root-path-self-cycle", "mutual-cycle"],
)
def test_schema_has_unresolvable_ref_flags_refs_that_cannot_be_inlined(schema, unresolvable):
assert schema_has_unresolvable_ref(schema) is unresolvable
@pytest.mark.parametrize(
("schema", "properties", "branches"),
[
(_object_with({"a": {"type": "string"}, "b": {"type": "string"}, "c": {"type": "string"}}, required=["a"]), 3, 8),
(
_object_with(
{"x": {"$ref": "#/$defs/Pair"}, "y": {"$ref": "#/$defs/Pair"}},
required=["x", "y"],
**{"$defs": {"Pair": _object_with({"l": {"type": "string"}, "r": {"type": "string"}}, required=["l"])}},
),
4,
7,
),
(_object_with({"inner": _object_with({"z": {"type": "string"}})}, required=["inner"]), 2, 4),
],
ids=["flat", "shared-def-counted-once", "nested"],
)
def test_schema_size_counts_every_written_object_once(schema, properties, branches):
assert schema_property_count(schema) == properties
assert schema_grammar_branch_count(schema) == branches

View file

@ -1012,6 +1012,25 @@ def test_bedrock_tools_unpack_defs():
_bedrock_tools_pt(tools=tools)
@pytest.mark.parametrize(
"parameters",
[
{"type": ["object", "null"]},
{"type": "object", "properties": {"tier": {"type": "integer"}}, "$defs": [{"type": "string"}]},
],
ids=["root_type_list", "defs_list"],
)
def test_bedrock_tools_pt_sends_malformed_parameters_as_an_object_schema(parameters: dict[str, object]) -> None:
from litellm.litellm_core_utils.prompt_templates.factory import _bedrock_tools_pt
tools: Final = [{"type": "function", "function": {"name": "json_tool_call", "parameters": parameters}}]
input_schema: Final = _bedrock_tools_pt(tools=tools)[0]["toolSpec"]["inputSchema"]["json"]
assert input_schema["type"] == "object"
assert "$defs" not in input_schema
def test_bedrock_tools_pt_strict_parameter():
"""Regression for strict tools on the Bedrock Converse path.

View file

@ -1302,6 +1302,74 @@ def test_opus_uses_native_structured_output(model_name):
assert optional_params.get("json_mode") is True
@pytest.mark.parametrize(
("field_schema", "native"),
[
({"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]}, True),
({"type": "object", "additionalProperties": {"type": "integer"}}, False),
({"anyOf": [{"type": "object"}, {"type": "null"}]}, False),
],
ids=["closed-object", "dict-field", "nullable-dict-field"],
)
def test_dict_fields_take_the_json_tool_on_a_native_structured_output_model(field_schema, native):
optional_params: Final = AnthropicConfig().map_openai_params(
non_default_params={
"response_format": {
"type": "json_schema",
"json_schema": {
"name": "out",
"schema": {"type": "object", "properties": {"field": field_schema}, "required": ["field"]},
},
}
},
optional_params={},
model="claude-opus-4-8",
drop_params=False,
)
assert ("output_format" in optional_params) is native
assert ([tool["name"] for tool in optional_params.get("tools", [])] == ["json_tool_call"]) is not native
@pytest.mark.parametrize(
("model", "thinking_params"),
[
("claude-opus-4-8", {"thinking": {"type": "adaptive"}}),
("claude-opus-4-8", {"reasoning_effort": "low"}),
("claude-opus-4-8", {"reasoning_effort": "none"}),
("claude-sonnet-5-5", {}),
(
"claude-opus-4-8",
{"tools": [{"type": "function", "function": {"name": "lookup", "parameters": {"type": "object"}}}]},
),
],
ids=["adaptive-thinking", "reasoning-effort", "reasoning-effort-none", "model-without-forced-tools", "caller-tools"],
)
def test_dict_fields_keep_native_output_when_the_json_tool_would_not_be_forced(model, thinking_params):
optional_params: Final = AnthropicConfig().map_openai_params(
non_default_params={
**thinking_params,
"response_format": {
"type": "json_schema",
"json_schema": {
"name": "out",
"schema": {
"type": "object",
"properties": {"field": {"type": "object", "additionalProperties": {"type": "integer"}}},
"required": ["field"],
},
},
},
},
optional_params={},
model=model,
drop_params=False,
)
assert "output_format" in optional_params
assert "json_tool_call" not in [tool["name"] for tool in optional_params.get("tools", [])]
def test_native_structured_output_uses_bundled_capability_when_remote_map_lags(
monkeypatch: pytest.MonkeyPatch,
) -> None:
@ -6798,3 +6866,15 @@ def test_chat_dummy_tool_result_for_an_orphaned_tool_call_replays_a_byte_identic
_assert_prefix_stable(requests)
assert [m["role"] for m in requests[0]["messages"]] == ["user", "assistant", "user"]
assert requests[0]["messages"][2]["content"][0]["type"] == "tool_result"
def test_json_mode_keeps_the_thinking_that_preceded_the_json_tool_call():
thinking_block: Final = {"type": "thinking", "thinking": "The user wants a greeting.", "signature": "sig-1"}
body: Final = _json_mode_message_body("tool_use", None)
result: Final = _transform_json_mode_body({**body, "content": [thinking_block, *body["content"]]})
message: Final = result.choices[0].message
assert json.loads(message.content) == {"answer": "Hello"}
assert message.reasoning_content == thinking_block["thinking"]
assert [block["signature"] for block in message.thinking_blocks] == [thinking_block["signature"]]

View file

@ -12,6 +12,18 @@ from litellm.llms.anthropic.chat.transformation import AnthropicConfig
class TestOutputFormatArrayObjectConstraints:
def test_closes_an_object_whose_type_is_a_union_with_null(self):
schema = {
"type": "object",
"properties": {
"address": {"type": ["object", "null"], "properties": {"city": {"type": "string"}}},
},
}
result = AnthropicConfig.filter_anthropic_output_schema(schema)
assert result["properties"]["address"]["additionalProperties"] is False
def test_removes_uniqueitems(self):
schema = {
"type": "object",
@ -129,7 +141,7 @@ class TestOutputFormatArrayObjectConstraints:
schema = {
"type": "array",
"items": {"type": "string"},
"minItems": 1,
"minItems": 2,
"maxItems": 10,
"uniqueItems": True,
"minContains": 2,
@ -139,7 +151,7 @@ class TestOutputFormatArrayObjectConstraints:
result = AnthropicConfig.filter_anthropic_output_schema(schema)
assert result["description"] == (
"Note: minimum number of items: 1, maximum number of items: 10, "
"Note: minimum number of items: 2, maximum number of items: 10, "
"all array items must be unique, minimum number of matching items: 2, "
"maximum number of matching items: 3."
)

View file

@ -69,7 +69,7 @@ class TestFilterAnthropicOutputSchema:
schema = {
"type": "array",
"items": {"type": "string"},
"minItems": 1,
"minItems": 2,
"maxItems": 10,
}
@ -80,9 +80,27 @@ class TestFilterAnthropicOutputSchema:
assert result["type"] == "array"
assert result["items"] == {"type": "string"}
# Description should contain constraint info
assert "minimum number of items: 1" in result["description"]
assert "minimum number of items: 2" in result["description"]
assert "maximum number of items: 10" in result["description"]
def test_keeps_min_items_of_zero_or_one(self):
for min_items in (0, 1):
schema = {"type": "array", "items": {"type": "string"}, "minItems": min_items}
result = AnthropicConfig.filter_anthropic_output_schema(schema)
assert result["minItems"] == min_items
assert "description" not in result
def test_moves_a_min_items_that_is_not_an_integer_into_the_description(self):
for min_items in (True, 1.0):
schema = {"type": "array", "items": {"type": "string"}, "minItems": min_items}
result = AnthropicConfig.filter_anthropic_output_schema(schema)
assert "minItems" not in result
assert "minimum number of items" in result["description"]
def test_handles_nested_schemas(self):
"""Test that nested schemas are also filtered."""
schema = {
@ -100,7 +118,7 @@ class TestFilterAnthropicOutputSchema:
}
},
},
"minItems": 1,
"minItems": 2,
}
},
}
@ -336,7 +354,7 @@ class TestFilterAnthropicOutputSchema:
schema = {
"type": "array",
"items": {"type": "string"},
"minItems": 1,
"minItems": 2,
"maxItems": 10,
"uniqueItems": True,
"minContains": 2,
@ -346,7 +364,7 @@ class TestFilterAnthropicOutputSchema:
result = AnthropicConfig.filter_anthropic_output_schema(schema)
assert result["description"] == (
"Note: minimum number of items: 1, maximum number of items: 10, "
"Note: minimum number of items: 2, maximum number of items: 10, "
"all array items must be unique, minimum number of matching items: 2, "
"maximum number of matching items: 3."
)

View file

@ -4,11 +4,13 @@ import copy
import json
import uuid
from types import SimpleNamespace
from collections.abc import Mapping
from typing import Final
from unittest.mock import patch
import httpx
import pytest
from pydantic import BaseModel, ConfigDict
# Ensure the project root is on the import path so `litellm` can be imported when
# tests are executed from any working directory.
@ -521,7 +523,7 @@ def test_output_config_format_converted_for_bedrock_chat_invoke_request():
with patch( # test-quality-ok: pin non-native path
"litellm.llms.bedrock.common_utils._bedrock_model_supports",
side_effect=lambda _model, key: key == "supports_output_config",
side_effect=lambda _model, key, *_: key == "supports_output_config",
):
result = config.transform_request(
model="anthropic.claude-opus-4-7",
@ -543,10 +545,12 @@ def test_output_config_format_converted_for_bedrock_chat_invoke_request():
@pytest.mark.parametrize("model", ["anthropic.claude-opus-4-7", "us.anthropic.claude-opus-4-8"])
def test_output_config_format_inlined_for_bedrock_chat_invoke_opus_4_7_and_4_8(local_model_cost_map, model):
"""Bedrock rejects ``output_config.format`` on Claude Opus 4.7 and 4.8, so the
Invoke chat path inlines the schema into the last user message and keeps effort,
driven by the cost map alone (no capability stub)."""
def test_output_config_format_inlined_for_bedrock_chat_invoke_profile_without_native_output(
set_native_structured_output, model
):
"""On a profile without native structured output the Invoke chat path inlines the
schema into the last user message and keeps effort."""
set_native_structured_output(model, None)
schema = {"type": "object", "properties": {"answer": {"type": "string"}}}
result = AmazonAnthropicClaudeConfig().transform_request(
@ -564,12 +568,155 @@ def test_output_config_format_inlined_for_bedrock_chat_invoke_opus_4_7_and_4_8(l
assert json.loads(result["messages"][-1]["content"][-1]["text"]) == schema
TREE_SCHEMA: Final = {
"type": "object",
"properties": {"root": {"$ref": "#/$defs/Node"}},
"$defs": {
"Node": {
"type": "object",
"properties": {"children": {"type": "array", "items": {"$ref": "#/$defs/Node"}}},
}
},
}
OPEN_OBJECT_SCHEMA: Final = {
"type": "object",
"properties": {"counts": {"type": "object", "additionalProperties": {"type": "integer"}}},
}
def _invoke_params_for_schema(
model: str, schema: Mapping[str, object], thinking_params: Mapping[str, object] | None = None
) -> Mapping[str, object]:
return AmazonAnthropicClaudeConfig().map_openai_params(
non_default_params={
"response_format": {"type": "json_schema", "json_schema": {"name": "out", "schema": schema}},
**({"thinking": {"type": "adaptive"}} if thinking_params is None else thinking_params),
},
optional_params={},
model=model,
drop_params=False,
)
@pytest.mark.parametrize(
("native_key", "thinking_params", "native"),
[
("legacy", {}, True),
("bedrock", {}, False),
("bedrock", {"thinking": {"type": "adaptive"}}, True),
("bedrock", {"reasoning_effort": "low"}, True),
],
ids=["legacy-key", "bedrock-key-no-thinking", "bedrock-key-adaptive", "bedrock-key-effort"],
)
def test_bedrock_native_output_key_applies_only_when_thinking_is_requested(
set_native_structured_output, native_key, thinking_params, native
):
model = "us.anthropic.claude-haiku-5-5"
set_native_structured_output(model, native_key)
schema = {"type": "object", "properties": {"answer": {"type": "string"}}, "required": ["answer"]}
result = _invoke_params_for_schema(model, schema, thinking_params)
assert ("output_format" in result) is native
assert ("tools" in result) is not native
@pytest.mark.parametrize("native_key", ["legacy", "bedrock"])
@pytest.mark.parametrize("schema", [TREE_SCHEMA, OPEN_OBJECT_SCHEMA], ids=["recursive", "open-object"])
def test_schema_native_output_cannot_hold_keeps_the_tool_path_on_a_native_profile(
set_native_structured_output, schema, native_key
):
model = "us.anthropic.claude-opus-4-8"
set_native_structured_output(model, native_key)
result = _invoke_params_for_schema(model, schema)
assert "output_format" not in result
assert [tool["name"] for tool in result["tools"]] == ["json_tool_call"]
@pytest.mark.parametrize("native_key", ["legacy", "bedrock"])
def test_closed_schema_goes_native_on_either_native_key(set_native_structured_output, native_key):
model = "us.anthropic.claude-opus-4-8"
set_native_structured_output(model, native_key)
schema = {"type": "object", "properties": {"age": {"type": "integer", "minimum": 0}}, "required": ["age"]}
result = _invoke_params_for_schema(model, schema)
assert result["output_format"]["schema"]["properties"]["age"]["type"] == "integer"
assert "minimum" not in result["output_format"]["schema"]["properties"]["age"]
assert "tools" not in result
@pytest.mark.parametrize("native_key", ["legacy", "bedrock"])
def test_shared_defs_reach_native_output_once_instead_of_inlined_per_use(set_native_structured_output, native_key):
model = "us.anthropic.claude-opus-4-8"
set_native_structured_output(model, native_key)
party_ref = {"$ref": "#/$defs/Party"}
schema = {
"type": "object",
"properties": {"buyer": party_ref, "seller": party_ref},
"required": ["buyer", "seller"],
"$defs": {
"Party": {
"type": "object",
"properties": {"email": {"anyOf": [{"type": "string"}, {"type": "null"}]}},
"required": ["email"],
}
},
}
result = _invoke_params_for_schema(model, schema)
sent = result["output_format"]["schema"]
assert sent["properties"]["buyer"] == party_ref
assert json.dumps(sent).count('"anyOf"') == 1
assert "tools" not in result
class Relabel(BaseModel):
model_config = ConfigDict(frozen=True)
box: str
new_label: str
class PuzzleAnswer(BaseModel):
model_config = ConfigDict(frozen=True)
relabels: list[Relabel]
@pytest.mark.parametrize("native_key", ["legacy", "bedrock"])
def test_nested_pydantic_response_format_reaches_native_output_with_refs_that_resolve(
set_native_structured_output, native_key
):
model = "us.anthropic.claude-haiku-5-5"
set_native_structured_output(model, native_key)
config: Final = AmazonAnthropicClaudeConfig()
result: Final = config.map_openai_params(
non_default_params={
"response_format": config.get_json_schema_from_pydantic_object(PuzzleAnswer),
"thinking": {"type": "adaptive"},
},
optional_params={},
model=model,
drop_params=False,
)
sent: Final = result["output_format"]["schema"]
item_ref: Final = sent["properties"]["relabels"]["items"]["$ref"]
assert item_ref.removeprefix("#/$defs/") in sent["$defs"]
assert "tools" not in result
def test_output_config_format_forwarded_for_bedrock_chat_invoke_request(local_model_cost_map):
"""Bedrock Invoke chat path forwards ``output_config.format`` alongside effort
for models with native structured-output support (Claude Sonnet 4.6)."""
schema_format = {
"type": "json_schema",
"schema": {"type": "object", "properties": {"answer": {"type": "string"}}},
"schema": {"type": "object", "properties": {"answer": {"type": "string"}}, "additionalProperties": False},
}
result = AmazonAnthropicClaudeConfig().transform_request(
@ -712,7 +859,7 @@ def test_bedrock_chat_invoke_drop_params_keeps_native_output_config_format(local
monkeypatch.setattr(litellm, "drop_params", True)
schema_format = {
"type": "json_schema",
"schema": {"type": "object", "properties": {"zebra_count": {"type": "integer"}}},
"schema": {"type": "object", "properties": {"zebra_count": {"type": "integer"}}, "additionalProperties": False},
}
result = AmazonAnthropicClaudeConfig().transform_request(
@ -754,10 +901,11 @@ def test_bedrock_chat_invoke_drop_params_still_inlines_for_non_native(local_mode
"model",
["us.anthropic.claude-fable-5-1", "anthropic.claude-fable-5-1"],
)
def test_bedrock_chat_invoke_fable_5_1_response_format_avoids_forced_tool_choice(local_model_cost_map, model):
"""Regression: Bedrock rejects both native ``output_config.format`` and forced
tool_choice for Fable 5.1, so invoke must use the tool-based path without a
def test_bedrock_chat_invoke_fable_5_1_response_format_avoids_forced_tool_choice(set_native_structured_output, model):
"""Regression: Bedrock rejects forced tool_choice for Fable 5.1, so on a profile
without native structured output invoke must use the tool-based path without a
forced ``tool_choice``."""
set_native_structured_output(model, None)
result = AmazonAnthropicClaudeConfig().map_openai_params(
non_default_params={
"response_format": {
@ -779,7 +927,10 @@ def test_bedrock_chat_invoke_fable_5_1_response_format_avoids_forced_tool_choice
@pytest.mark.parametrize("model", ["us.anthropic.claude-sonnet-5", "us.anthropic.claude-fable-5-1"])
def test_bedrock_chat_invoke_tool_based_response_format_still_upgrades_legacy_thinking(local_model_cost_map, model):
def test_bedrock_chat_invoke_tool_based_response_format_still_upgrades_legacy_thinking(
set_native_structured_output, model
):
set_native_structured_output(model, None)
result = AmazonAnthropicClaudeConfig().map_openai_params(
non_default_params={
"response_format": {
@ -802,6 +953,9 @@ def test_bedrock_chat_invoke_tool_based_response_format_still_upgrades_legacy_th
assert result["output_config"] == {"effort": "high"}
def test_bedrock_chat_invoke_response_format_stub_still_upgrades_legacy_thinking(local_model_cost_map):
"""Regression: the tool-based ``response_format`` path swaps in a Claude 3 stub
model before the shared Anthropic mapping, which hid the adaptive-only model

View file

@ -1,6 +1,7 @@
import copy
import json
import os
from collections.abc import Mapping
from typing import Final
from unittest.mock import MagicMock, patch
@ -8,8 +9,19 @@ import httpx
import pytest
import litellm
from litellm import ModelResponse
from litellm import ModelResponse, Router, completion
from litellm.litellm_core_utils.prompt_templates.factory import (
_bedrock_converse_messages_pt,
_bedrock_tools_pt,
)
from litellm.litellm_core_utils.prompt_templates.mid_conversation_system import CONVERTED_SYSTEM_NOTE
from litellm.llms.anthropic.chat.transformation import (
NATIVE_OUTPUT_MAX_GRAMMAR_BRANCHES,
NATIVE_OUTPUT_MAX_OPTIONAL_PARAMETERS,
NATIVE_OUTPUT_MAX_UNION_PARAMETERS,
)
from litellm.llms.bedrock.common_utils import BEDROCK_NATIVE_OUTPUT_MAX_PROPERTIES
from litellm.llms.bedrock.base_aws_llm import BaseAWSLLM
from litellm.llms.bedrock.chat.converse_transformation import AmazonConverseConfig
from litellm.types.llms.bedrock import ConverseTokenUsageBlock
@ -637,6 +649,9 @@ def test_output_config_effort_forwarded_into_additional_request_fields(model):
assert additional.get("output_config") == {"effort": "high"}
def test_reasoning_effort_requests_summarized_display_converse():
"""Regression LIT-5714: adaptive thinking synthesized from reasoning_effort must
request the summarized display, otherwise the provider returns a blank thinking
@ -831,32 +846,25 @@ def test_output_config_format_dropped_on_unsupported_converse_model_warns(caplog
"""When Converse model lacks native structured-output support, the silently
dropped ``output_config.format`` must surface as a warning so callers can
diagnose plain-text responses."""
from unittest.mock import patch
config = AmazonConverseConfig()
schema = {
"type": "object",
"properties": {"answer": {"type": "string"}},
}
with patch.object(
AmazonConverseConfig,
"_supports_native_structured_outputs",
return_value=False,
):
with caplog.at_level("WARNING"):
result = config._transform_request(
model="bedrock/converse/us.anthropic.claude-3-haiku-20240307-v1:0",
messages=[{"role": "user", "content": "hi"}],
optional_params={
"maxTokens": 256,
"output_config": {
"format": {"type": "json_schema", "schema": schema},
},
with caplog.at_level("WARNING"):
result = config._transform_request(
model="bedrock/converse/us.anthropic.claude-3-haiku-20240307-v1:0",
messages=[{"role": "user", "content": "hi"}],
optional_params={
"maxTokens": 256,
"output_config": {
"format": {"type": "json_schema", "schema": schema},
},
litellm_params={},
headers={},
)
},
litellm_params={},
headers={},
)
assert "outputConfig" not in result
assert any(
@ -4293,24 +4301,6 @@ def test_translate_response_format_native_output_config(monkeypatch):
monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", old_env)
BEDROCK_OPUS_4_7_AND_4_8_MODELS: Final = (
"anthropic.claude-opus-4-7",
"global.anthropic.claude-opus-4-7",
"us.anthropic.claude-opus-4-7",
"eu.anthropic.claude-opus-4-7",
"au.anthropic.claude-opus-4-7",
"jp.anthropic.claude-opus-4-7",
"anthropic.claude-opus-4-8",
"global.anthropic.claude-opus-4-8",
"us.anthropic.claude-opus-4-8",
"eu.anthropic.claude-opus-4-8",
"au.anthropic.claude-opus-4-8",
"jp.anthropic.claude-opus-4-8",
"us-gov.anthropic.claude-opus-4-8",
"us-gov-west-1/anthropic.claude-opus-4-8",
"us-gov-east-1/anthropic.claude-opus-4-8",
)
CAPITAL_RESPONSE_FORMAT: Final = {
"type": "json_schema",
"json_schema": {
@ -4325,10 +4315,15 @@ CAPITAL_RESPONSE_FORMAT: Final = {
}
def _converse_request_for_json_schema(model: str, stream: bool) -> tuple[dict, dict]:
ADAPTIVE_THINKING: Final = {"type": "adaptive"}
def _converse_request_for_json_schema(
model: str, stream: bool, thinking_params: Mapping[str, object] | None = None
) -> tuple[Mapping[str, object], RequestObject]:
config = AmazonConverseConfig()
optional_params = config.map_openai_params(
non_default_params={"response_format": CAPITAL_RESPONSE_FORMAT, "stream": stream},
non_default_params={"response_format": CAPITAL_RESPONSE_FORMAT, "stream": stream, **(thinking_params or {})},
optional_params={},
model=model,
drop_params=False,
@ -4343,21 +4338,452 @@ def _converse_request_for_json_schema(model: str, stream: bool) -> tuple[dict, d
return optional_params, request
@pytest.mark.parametrize("model", BEDROCK_OPUS_4_7_AND_4_8_MODELS)
@pytest.mark.parametrize(
("native_profile", "tool_profile"),
[
("us.anthropic.claude-opus-4-8", "anthropic.claude-opus-4-8"),
("global.anthropic.claude-opus-4-7", "us.anthropic.claude-opus-4-7"),
("anthropic.claude-opus-4-8", "us-gov.anthropic.claude-opus-4-8"),
],
)
@pytest.mark.parametrize("stream", [False, True])
def test_opus_4_7_and_4_8_json_schema_sent_as_forced_tool_not_output_config(monkeypatch, model, stream):
"""Regression for issue #27846: Bedrock rejects outputConfig on Opus 4.7 and 4.8
(``output_config.format: Extra inputs are not permitted``), so json_schema has to
go out as the forced json_tool_call tool, streamed through fake_stream."""
monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True")
monkeypatch.setattr(litellm, "model_cost", litellm.get_model_cost_map(url=""))
def test_json_schema_shape_follows_each_inference_profile_own_native_output_flag(
set_native_structured_output, native_profile, tool_profile, stream
):
set_native_structured_output(native_profile, "bedrock")
set_native_structured_output(tool_profile, None)
optional_params, request = _converse_request_for_json_schema(model=model, stream=stream)
thinking_params: Final = {"thinking": ADAPTIVE_THINKING}
native_params, native_request = _converse_request_for_json_schema(native_profile, stream, thinking_params)
tool_params, tool_request = _converse_request_for_json_schema(tool_profile, stream, thinking_params)
assert native_request["outputConfig"]["textFormat"]["structure"]["jsonSchema"]["name"] == "capital"
assert "toolConfig" not in native_request
assert "fake_stream" not in native_params
assert "outputConfig" not in tool_request
assert [tool["toolSpec"]["name"] for tool in tool_request["toolConfig"]["tools"]] == ["json_tool_call"]
assert tool_request["toolConfig"]["toolChoice"] == {"tool": {"name": "json_tool_call"}}
assert tool_params.get("fake_stream", False) is stream
@pytest.mark.parametrize(
("native_key", "thinking_params", "native"),
[
("legacy", {}, True),
("bedrock", {}, False),
("bedrock", {"thinking": ADAPTIVE_THINKING}, True),
("bedrock", {"thinking": {"type": "enabled", "budget_tokens": 1024}}, True),
("bedrock", {"reasoning_effort": "low"}, True),
("bedrock", {"reasoning_effort": "none"}, False),
("bedrock", {"thinking": {"type": "disabled"}}, False),
],
ids=[
"legacy-key",
"bedrock-key-no-thinking",
"bedrock-key-adaptive",
"bedrock-key-enabled",
"bedrock-key-effort",
"bedrock-key-effort-none",
"bedrock-key-thinking-disabled",
],
)
def test_bedrock_native_output_key_applies_only_when_thinking_is_requested(
set_native_structured_output, native_key, thinking_params, native
):
model: Final = "us.anthropic.claude-haiku-5-5"
set_native_structured_output(model, native_key)
_, request = _converse_request_for_json_schema(model, False, thinking_params)
assert ("outputConfig" in request) is native
assert ("toolConfig" in request) is not native
def test_adaptive_thinking_is_kept_next_to_json_schema_on_a_native_output_profile(set_native_structured_output):
model: Final = "us.anthropic.claude-opus-5"
set_native_structured_output(model, "bedrock")
config: Final = AmazonConverseConfig()
thinking: Final = {"type": "adaptive", "display": "summarized"}
optional_params: Final = config.map_openai_params(
non_default_params={"response_format": CAPITAL_RESPONSE_FORMAT, "thinking": thinking},
optional_params={},
model=model,
drop_params=False,
)
request: Final = config._transform_request(
model=model,
messages=[{"role": "user", "content": "Name the capital of France."}],
optional_params=optional_params,
litellm_params={},
headers={},
)
assert request["additionalModelRequestFields"]["thinking"] == thinking
assert request["outputConfig"]["textFormat"]["structure"]["jsonSchema"]["name"] == "capital"
assert "toolConfig" not in request
_OPUS_4_8_INVOKE_RESPONSE: Final = {
"id": "msg_1",
"type": "message",
"role": "assistant",
"model": "claude-opus-4-8",
"content": [{"type": "text", "text": '{"city": "Paris", "country": "France"}'}],
"stop_reason": "end_turn",
"usage": {"input_tokens": 10, "output_tokens": 5},
}
def _routed_request_body(router: Router, model_name: str, route: str) -> Mapping[str, object]:
response: Final = _mock_converse_response()
if route == "invoke":
response.json.return_value = _OPUS_4_8_INVOKE_RESPONSE
response.text = json.dumps(_OPUS_4_8_INVOKE_RESPONSE)
client: Final = HTTPHandler()
with patch.object(client, "post", return_value=response) as mock_post:
router.completion(
model=model_name,
messages=[{"role": "user", "content": "Name the capital of France."}],
response_format=CAPITAL_RESPONSE_FORMAT,
thinking=ADAPTIVE_THINKING,
client=client,
)
return json.loads(mock_post.call_args.kwargs["data"])
def _sends_native_output(body: Mapping[str, object], route: str) -> bool:
if route == "converse":
return "outputConfig" in body and "toolConfig" not in body
output_config: Final = body.get("output_config")
return isinstance(output_config, dict) and "format" in output_config and "tools" not in body
@pytest.mark.parametrize("route", ["converse", "invoke"])
def test_a_deployment_native_output_opt_in_stays_on_that_deployment(set_native_structured_output, route):
set_native_structured_output("us.anthropic.claude-opus-4-8", None)
litellm_params: Final = {
"model": f"bedrock/{route}/us.anthropic.claude-opus-4-8",
"aws_access_key_id": "fake-access-key",
"aws_secret_access_key": "fake-secret-key",
"aws_region_name": "us-east-1",
}
router: Final = Router(
model_list=[
{
"model_name": "opted-in",
"litellm_params": litellm_params,
"model_info": {"supports_bedrock_native_structured_output": True},
},
{"model_name": "sibling", "litellm_params": litellm_params},
]
)
assert _sends_native_output(_routed_request_body(router, "opted-in", route), route)
assert not _sends_native_output(_routed_request_body(router, "sibling", route), route)
NATIVE_OUTPUT_REJECTED_KEYWORDS: Final = frozenset({"minimum", "maximum", "uniqueItems", "oneOf"})
CONSTRAINED_PERSON_SCHEMA: Final = {
"type": "object",
"properties": {
"age": {"type": "integer", "minimum": 0, "maximum": 150},
"tags": {"type": "array", "items": {"type": "string"}, "minItems": 2, "uniqueItems": True},
"aliases": {"type": "array", "items": {"type": "string"}, "minItems": 1},
"contact": {"oneOf": [{"type": "string"}, {"type": "integer"}]},
"address": {"type": ["object", "null"], "properties": {"city": {"type": "string"}}, "required": ["city"]},
},
"required": ["age", "tags", "aliases", "contact", "address"],
}
RECURSIVE_TREE_SCHEMA: Final = {
"type": "object",
"properties": {"root": {"$ref": "#/$defs/Node"}},
"required": ["root"],
"$defs": {
"Node": {
"type": "object",
"properties": {
"name": {"type": "string"},
"children": {"type": "array", "items": {"$ref": "#/$defs/Node"}},
},
"required": ["name", "children"],
}
},
}
OPEN_OBJECT_SCHEMA: Final = {
"type": "object",
"properties": {"counts": {"type": "object", "additionalProperties": {"type": "integer"}}},
"required": ["counts"],
}
LOOKAROUND_PATTERN_SCHEMA: Final = {
"type": "object",
"properties": {"amount": {"type": "string", "pattern": "^(?!^[-+.]*$)[+-]?0*\\d*\\.?\\d*$"}},
"required": ["amount"],
}
UNKNOWN_FORMAT_SCHEMA: Final = {
"type": "object",
"properties": {"blob": {"type": "string", "format": "x-litellm-unlisted-format"}},
"required": ["blob"],
}
ROOT_REF_SCHEMA: Final = {
"type": "object",
"properties": {"name": {"type": "string"}, "parent": {"anyOf": [{"$ref": "#"}, {"type": "null"}]}},
"required": ["name", "parent"],
}
def _schema_keys(node: object) -> frozenset[str]:
if isinstance(node, dict):
return frozenset(node).union(*(_schema_keys(value) for value in node.values()))
if isinstance(node, list):
return frozenset().union(*(_schema_keys(item) for item in node))
return frozenset()
def _converse_request_for_schema(model: str, schema: Mapping[str, object]) -> RequestObject:
config: Final = AmazonConverseConfig()
optional_params: Final = config.map_openai_params(
non_default_params={
"response_format": {"type": "json_schema", "json_schema": {"name": "out", "schema": schema}},
"thinking": ADAPTIVE_THINKING,
},
optional_params={},
model=model,
drop_params=False,
)
return config._transform_request(
model=model,
messages=[{"role": "user", "content": "Fill in a plausible example."}],
optional_params=optional_params,
litellm_params={},
headers={},
)
@pytest.mark.parametrize("native_key", ["legacy", "bedrock"])
def test_native_output_schema_drops_keywords_bedrock_rejects_and_closes_objects(
set_native_structured_output, native_key
):
model: Final = "us.anthropic.claude-opus-4-8"
set_native_structured_output(model, native_key)
request: Final = _converse_request_for_schema(model, CONSTRAINED_PERSON_SCHEMA)
sent: Final = json.loads(request["outputConfig"]["textFormat"]["structure"]["jsonSchema"]["schema"])
assert not _schema_keys(sent) & NATIVE_OUTPUT_REJECTED_KEYWORDS
assert sent["additionalProperties"] is False
assert sent["properties"]["contact"]["anyOf"] == [{"type": "string"}, {"type": "integer"}]
assert "minimum value: 0" in sent["properties"]["age"]["description"]
assert "minItems" not in sent["properties"]["tags"]
assert sent["properties"]["aliases"]["minItems"] == 1
assert sent["properties"]["address"]["additionalProperties"] is False
@pytest.mark.parametrize("native_key", ["legacy", "bedrock"])
@pytest.mark.parametrize(
("schema", "kept_keyword"),
[
(RECURSIVE_TREE_SCHEMA, "$ref"),
(ROOT_REF_SCHEMA, "$ref"),
(OPEN_OBJECT_SCHEMA, "additionalProperties"),
(LOOKAROUND_PATTERN_SCHEMA, "pattern"),
(UNKNOWN_FORMAT_SCHEMA, "format"),
],
ids=["recursive", "root-ref", "open-object", "lookaround-pattern", "unlisted-format"],
)
def test_schema_native_output_cannot_hold_keeps_the_tool_path_on_a_native_profile(
set_native_structured_output, schema, kept_keyword, native_key
):
model: Final = "us.anthropic.claude-opus-4-8"
set_native_structured_output(model, native_key)
request: Final = _converse_request_for_schema(model, schema)
assert "outputConfig" not in request
assert [tool["toolSpec"]["name"] for tool in request["toolConfig"]["tools"]] == ["json_tool_call"]
assert request["toolConfig"]["toolChoice"] == {"tool": {"name": "json_tool_call"}}
assert optional_params.get("fake_stream", False) is stream
tool_spec: Final = request["toolConfig"]["tools"][0]["toolSpec"]
assert tool_spec["name"] == "json_tool_call"
assert kept_keyword in _schema_keys(tool_spec["inputSchema"]["json"])
def _shared_defs_schema(defs_key: str) -> Mapping[str, object]:
party_ref: Final = {"$ref": f"#/{defs_key}/Party"}
return {
"type": "object",
"properties": {"buyer": party_ref, "seller": party_ref, "broker": {"anyOf": [party_ref, {"type": "null"}]}},
"required": ["buyer", "seller", "broker"],
defs_key: {
"Party": {
"type": ["object", "null"],
"properties": {"name": {"type": "string"}, "email": {"anyOf": [{"type": "string"}, {"type": "null"}]}},
"required": ["name", "email"],
}
},
}
@pytest.mark.parametrize("defs_key", ["$defs", "definitions"])
@pytest.mark.parametrize("native_key", ["legacy", "bedrock"])
def test_shared_defs_reach_native_output_once_instead_of_inlined_per_use(
set_native_structured_output, native_key, defs_key
):
model: Final = "us.anthropic.claude-opus-4-8"
set_native_structured_output(model, native_key)
request: Final = _converse_request_for_schema(model, _shared_defs_schema(defs_key))
sent: Final = json.loads(request["outputConfig"]["textFormat"]["structure"]["jsonSchema"]["schema"])
assert sent["properties"]["buyer"] == {"$ref": f"#/{defs_key}/Party"}
assert json.dumps(sent).count('"anyOf"') == 2
assert sent[defs_key]["Party"]["additionalProperties"] is False
def _flat_schema(property_schema: Mapping[str, object], count: int, required: bool) -> Mapping[str, object]:
names: Final = tuple(f"field_{index}" for index in range(count))
return {
"type": "object",
"properties": dict.fromkeys(names, property_schema),
"required": list(names) if required else [],
}
def _optional_groups_schema(count: int, group_size: int = 6) -> Mapping[str, object]:
groups: Final = {
f"group_{start}": _flat_schema({"type": "string"}, min(group_size, count - start), required=False)
for start in range(0, count, group_size)
}
return {"type": "object", "properties": groups, "required": list(groups)}
@pytest.mark.parametrize("native_key", ["legacy", "bedrock"])
@pytest.mark.parametrize(
("schema_for", "limit"),
[
(lambda count: _flat_schema({"type": ["string", "null"]}, count, True), NATIVE_OUTPUT_MAX_UNION_PARAMETERS),
(
lambda count: _flat_schema({"anyOf": [{"type": "string"}, {"type": "integer"}]}, count, True),
NATIVE_OUTPUT_MAX_UNION_PARAMETERS,
),
(_optional_groups_schema, NATIVE_OUTPUT_MAX_OPTIONAL_PARAMETERS),
(
lambda count: _flat_schema({"type": "string"}, count, False),
NATIVE_OUTPUT_MAX_GRAMMAR_BRANCHES.bit_length() - 1,
),
],
ids=["type-list-unions", "any-of-unions", "optional", "optional-grammar-branches"],
)
def test_schema_past_a_native_output_size_limit_keeps_the_tool_path(
set_native_structured_output, native_key, schema_for, limit
):
model: Final = "us.anthropic.claude-opus-4-8"
set_native_structured_output(model, native_key)
at_limit: Final = _converse_request_for_schema(model, schema_for(limit))
past_limit: Final = _converse_request_for_schema(model, schema_for(limit + 1))
sent: Final = json.loads(at_limit["outputConfig"]["textFormat"]["structure"]["jsonSchema"]["schema"])
assert sent["properties"].keys() == schema_for(limit)["properties"].keys()
assert "outputConfig" not in past_limit
assert past_limit["toolConfig"]["tools"][0]["toolSpec"]["name"] == "json_tool_call"
@pytest.mark.parametrize(
("native_key", "native_past_limit"), [("legacy", True), ("bedrock", False)], ids=["legacy-key", "bedrock-key"]
)
def test_field_count_limit_sends_the_json_tool_only_on_the_bedrock_native_key(
set_native_structured_output, native_key, native_past_limit
):
model: Final = "us.anthropic.claude-opus-4-8"
set_native_structured_output(model, native_key)
at_limit: Final = _converse_request_for_schema(
model, _flat_schema({"type": "string"}, BEDROCK_NATIVE_OUTPUT_MAX_PROPERTIES, True)
)
past_limit: Final = _converse_request_for_schema(
model, _flat_schema({"type": "string"}, BEDROCK_NATIVE_OUTPUT_MAX_PROPERTIES + 1, True)
)
assert "outputConfig" in at_limit
assert ("outputConfig" in past_limit) is native_past_limit
NUMERIC_FORMAT_SCHEMA: Final = {
"type": "object",
"properties": {
"count": {"type": "integer", "format": "int32"},
"total": {"type": "integer", "format": "int64"},
"ratio": {"type": "number", "format": "double"},
},
"required": ["count", "total", "ratio"],
}
@pytest.mark.parametrize("native_key", ["legacy", "bedrock"])
def test_numeric_formats_keep_native_output(set_native_structured_output, native_key):
model: Final = "us.anthropic.claude-opus-4-8"
set_native_structured_output(model, native_key)
request: Final = _converse_request_for_schema(model, NUMERIC_FORMAT_SCHEMA)
sent: Final = json.loads(request["outputConfig"]["textFormat"]["structure"]["jsonSchema"]["schema"])
assert {name: field["format"] for name, field in sent["properties"].items()} == {
"count": "int32",
"total": "int64",
"ratio": "double",
}
def _doubling_defs_schema(depth: int) -> Mapping[str, object]:
levels: Final = {
f"Level{index}": {
"type": "object",
"properties": {
"left": {"$ref": f"#/$defs/Level{index + 1}"},
"right": {"$ref": f"#/$defs/Level{index + 1}"},
},
"required": ["left", "right"],
}
for index in range(depth)
}
return {
"type": "object",
"properties": {"root": {"$ref": "#/$defs/Level0"}},
"required": ["root"],
"$defs": {**levels, f"Level{depth}": {"type": "string"}},
}
@pytest.mark.timeout(30)
def test_defs_that_double_at_every_level_reach_native_output_without_being_expanded(set_native_structured_output):
model: Final = "us.anthropic.claude-opus-4-8"
set_native_structured_output(model, "legacy")
schema: Final = _doubling_defs_schema(30)
request: Final = _converse_request_for_schema(model, schema)
sent: Final = json.loads(request["outputConfig"]["textFormat"]["structure"]["jsonSchema"]["schema"])
assert sent["$defs"].keys() == schema["$defs"].keys()
assert sent["properties"]["root"] == {"$ref": "#/$defs/Level0"}
@pytest.mark.parametrize(
("native_key", "forwarded"), [("legacy", True), ("bedrock", False)], ids=["legacy-key", "bedrock-key"]
)
def test_recursive_output_config_format_follows_the_profile_native_key(
set_native_structured_output, native_key, forwarded
):
model: Final = "us.anthropic.claude-opus-4-8"
set_native_structured_output(model, native_key)
request: Final = AmazonConverseConfig()._transform_request(
model=model,
messages=[{"role": "user", "content": "Fill in a plausible example."}],
optional_params={"output_config": {"format": {"type": "json_schema", "schema": RECURSIVE_TREE_SCHEMA}}},
litellm_params={},
headers={},
)
assert ("outputConfig" in request) is forwarded
def test_sonnet_4_6_json_schema_still_uses_native_output_config(monkeypatch):
@ -7399,11 +7825,12 @@ def test_unforced_tool_choice_unaffected_on_fable_5_1_converse(local_model_cost_
["anthropic.claude-fable-5-1", "us.anthropic.claude-fable-5-1"],
)
def test_response_format_avoids_native_and_forced_tool_choice_on_fable_5_1_converse(
local_model_cost_map, model
set_native_structured_output, model
):
"""Regression: Bedrock rejects both ``outputConfig`` structured output and forced
tool_choice for Fable 5.1, so response_format must map to a tool without a forced
tool_choice."""
"""Regression: Bedrock rejects forced tool_choice for Fable 5.1, so on a profile
without native structured output response_format must map to a tool without a
forced tool_choice."""
set_native_structured_output(model, None)
config = AmazonConverseConfig()
result = config.map_openai_params(

View file

@ -1071,6 +1071,7 @@ def test_bedrock_messages_forwards_output_config_with_output_format():
"schema": {
"type": "object",
"properties": {"answer": {"type": "string"}},
"additionalProperties": False,
},
}
optional_params = {
@ -1119,7 +1120,7 @@ def test_bedrock_messages_converts_output_config_format_to_inline_schema():
with patch(
"litellm.llms.bedrock.common_utils._bedrock_model_supports",
side_effect=lambda _model, key: key == "supports_output_config",
side_effect=lambda _model, key, *_: key == "supports_output_config",
):
result = cfg.transform_anthropic_messages_request(
model="anthropic.claude-opus-4-7",
@ -1193,7 +1194,7 @@ def test_bedrock_messages_does_not_mutate_callers_messages_when_embedding_schema
with patch(
"litellm.llms.bedrock.common_utils._bedrock_model_supports",
side_effect=lambda _model, key: key == "supports_output_config",
side_effect=lambda _model, key, *_: key == "supports_output_config",
):
result = cfg.transform_anthropic_messages_request(
model="anthropic.claude-opus-4-7",
@ -3227,11 +3228,12 @@ def test_bedrock_messages_forwards_output_config_format_natively(local_model_cos
assert "zebra_count" not in json.dumps(result["messages"])
def test_bedrock_messages_inlines_schema_for_claude_5(local_model_cost_map):
"""Bedrock rejects ``output_config.format`` for the Claude 5 family, so the
schema falls back to the inline-text path instead of a deterministic 400."""
def test_bedrock_messages_inlines_schema_for_a_profile_without_native_output(set_native_structured_output):
"""On a profile without native structured output the schema falls back to the
inline-text path instead of a deterministic 400."""
from litellm.types.router import GenericLiteLLMParams
set_native_structured_output("us.anthropic.claude-sonnet-5", None)
cfg = AmazonAnthropicClaudeMessagesConfig()
schema = {
"type": "object",
@ -3254,6 +3256,115 @@ def test_bedrock_messages_inlines_schema_for_claude_5(local_model_cost_map):
assert json.loads(last_content[-1]["text"]) == schema
def _bedrock_messages_request_with_schema(model: str, schema: Mapping[str, object]) -> Mapping[str, object]:
from litellm.types.router import GenericLiteLLMParams
return AmazonAnthropicClaudeMessagesConfig().transform_anthropic_messages_request(
model=model,
messages=[{"role": "user", "content": [{"type": "text", "text": "say hello"}]}],
anthropic_messages_optional_request_params={
"max_tokens": 100,
"output_config": {"format": {"type": "json_schema", "schema": schema}},
},
litellm_params=GenericLiteLLMParams(),
headers={},
)
def test_bedrock_messages_native_output_drops_schema_keywords_bedrock_rejects(set_native_structured_output):
model: Final = "us.anthropic.claude-sonnet-5"
set_native_structured_output(model, "bedrock")
schema: Final = {
"type": "object",
"properties": {
"count": {"type": "integer", "minimum": 1},
"ids": {"type": "array", "items": {"type": "string"}, "uniqueItems": True},
},
"required": ["count", "ids"],
}
result: Final = _bedrock_messages_request_with_schema(model, schema)
sent: Final = result["output_config"]["format"]["schema"]
assert "minimum" not in sent["properties"]["count"]
assert "uniqueItems" not in sent["properties"]["ids"]
assert sent["additionalProperties"] is False
@pytest.mark.parametrize("native_key", ["legacy", "bedrock"])
def test_bedrock_messages_shared_defs_reach_native_output_once_instead_of_inlined_per_use(
set_native_structured_output, native_key
):
model: Final = "us.anthropic.claude-sonnet-5"
set_native_structured_output(model, native_key)
party_ref: Final = {"$ref": "#/$defs/Party"}
schema: Final = {
"type": "object",
"properties": {"buyer": party_ref, "seller": party_ref},
"required": ["buyer", "seller"],
"additionalProperties": False,
"$defs": {
"Party": {
"type": "object",
"properties": {"email": {"anyOf": [{"type": "string"}, {"type": "null"}]}},
"required": ["email"],
"additionalProperties": False,
}
},
}
result: Final = _bedrock_messages_request_with_schema(model, schema)
sent: Final = result["output_config"]["format"]["schema"]
assert sent["properties"]["buyer"] == party_ref
assert json.dumps(sent).count('"anyOf"') == 1
MESSAGES_TREE_SCHEMA: Final = {
"type": "object",
"properties": {"root": {"$ref": "#/$defs/Node"}},
"$defs": {
"Node": {
"type": "object",
"properties": {"children": {"type": "array", "items": {"$ref": "#/$defs/Node"}}},
}
},
}
MESSAGES_OPEN_OBJECT_SCHEMA: Final = {
"type": "object",
"properties": {"counts": {"type": "object", "additionalProperties": {"type": "integer"}}},
}
@pytest.mark.parametrize(
"schema", [MESSAGES_TREE_SCHEMA, MESSAGES_OPEN_OBJECT_SCHEMA], ids=["recursive", "open-object"]
)
def test_bedrock_messages_schema_native_output_cannot_hold_is_inlined_on_a_bedrock_native_profile(
set_native_structured_output, schema
):
model: Final = "us.anthropic.claude-sonnet-5"
set_native_structured_output(model, "bedrock")
result: Final = _bedrock_messages_request_with_schema(model, schema)
assert "output_config" not in result
assert json.loads(result["messages"][-1]["content"][-1]["text"]) == schema
@pytest.mark.parametrize(
"schema", [MESSAGES_TREE_SCHEMA, MESSAGES_OPEN_OBJECT_SCHEMA], ids=["recursive", "open-object"]
)
def test_bedrock_messages_schema_native_output_cannot_hold_is_forwarded_on_a_legacy_native_profile(
set_native_structured_output, schema
):
model: Final = "us.anthropic.claude-sonnet-4-6"
set_native_structured_output(model, "legacy")
result: Final = _bedrock_messages_request_with_schema(model, schema)
assert result["output_config"]["format"]["schema"] == schema
def test_bedrock_messages_legacy_output_format_wins_over_output_config_format(local_model_cost_map):
"""When a request carries both schema forms, the legacy top-level
``output_format`` keeps winning, matching the pre-existing precedence."""
@ -3262,7 +3373,11 @@ def test_bedrock_messages_legacy_output_format_wins_over_output_config_format(lo
cfg = AmazonAnthropicClaudeMessagesConfig()
legacy_format = {
"type": "json_schema",
"schema": {"type": "object", "properties": {"legacy_field": {"type": "string"}}},
"schema": {
"type": "object",
"properties": {"legacy_field": {"type": "string"}},
"additionalProperties": False,
},
}
newer_format = {
"type": "json_schema",
@ -3296,7 +3411,7 @@ def test_bedrock_messages_drop_params_keeps_native_output_config_format(local_mo
cfg = AmazonAnthropicClaudeMessagesConfig()
schema_format = {
"type": "json_schema",
"schema": {"type": "object", "properties": {"zebra_count": {"type": "integer"}}},
"schema": {"type": "object", "properties": {"zebra_count": {"type": "integer"}}, "additionalProperties": False},
}
result = cfg.transform_anthropic_messages_request(
@ -3321,7 +3436,7 @@ def test_bedrock_messages_strips_effort_but_keeps_format_for_sonnet_4_5(local_mo
cfg = AmazonAnthropicClaudeMessagesConfig()
schema_format = {
"type": "json_schema",
"schema": {"type": "object", "properties": {"zebra_count": {"type": "integer"}}},
"schema": {"type": "object", "properties": {"zebra_count": {"type": "integer"}}, "additionalProperties": False},
}
result = cfg.transform_anthropic_messages_request(

View file

@ -557,8 +557,14 @@ def test_apply_structured_output_prefers_legacy_output_format(local_model_cost_m
apply_bedrock_invoke_structured_output,
)
legacy = {"type": "json_schema", "schema": {"type": "object", "properties": {"a": {"type": "string"}}}}
newer = {"type": "json_schema", "schema": {"type": "object", "properties": {"b": {"type": "string"}}}}
legacy = {
"type": "json_schema",
"schema": {"type": "object", "properties": {"a": {"type": "string"}}, "additionalProperties": False},
}
newer = {
"type": "json_schema",
"schema": {"type": "object", "properties": {"b": {"type": "string"}}, "additionalProperties": False},
}
body = {
"messages": [{"role": "user", "content": "hi"}],
"output_format": legacy,

View file

@ -1056,6 +1056,7 @@ def test_aaamodel_prices_and_context_window_json_is_valid():
"supports_native_streaming": {"type": "boolean"},
"supports_image_size": {"type": "boolean"},
"supports_native_structured_output": {"type": "boolean"},
"supports_bedrock_native_structured_output": {"type": "boolean"},
"use_openai_responses_path": {"type": "boolean"},
"off_peak_pricing": {
"type": "object",

View file

@ -0,0 +1,522 @@
import json
import re
from collections.abc import Collection, Iterator, Mapping
from pathlib import Path
from typing import Final, Literal, cast
import httpx
import pytest
import litellm
from litellm import get_model_info
from litellm.llms.bedrock.common_utils import BedrockModelInfo
from litellm.llms.custom_httpx.http_handler import HTTPHandler
from litellm.types.utils import ModelInfoBase
from litellm.utils import _invalidate_model_cost_lowercase_map
import os
from unittest.mock import MagicMock, patch
@pytest.fixture(autouse=True)
def isolate_model_info_state(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
monkeypatch.setattr(litellm, "model_cost", dict(litellm.model_cost))
monkeypatch.setattr(litellm, "custom_provider_map", list(litellm.custom_provider_map))
yield
_invalidate_model_cost_lowercase_map()
def test_get_model_info_simple_model_name():
"""
tests if model name given, and model exists in model info - the object is returned
"""
model = "claude-opus-5-5"
litellm.get_model_info(model)
def test_get_model_info_custom_llm_with_model_name():
"""
Tests if {custom_llm_provider}/{model_name} name given, and model exists in model info, the object is returned
"""
model = "anthropic/claude-opus-5-5"
litellm.get_model_info(model)
def test_get_model_info_custom_llm_with_same_name_vllm(monkeypatch):
"""
Tests if {custom_llm_provider}/{model_name} name given, and model exists in model info, the object is returned
"""
model = "command-r-plus"
provider = "openai" # vllm is openai-compatible
litellm.register_model(
{
"openai/command-r-plus": {
"input_cost_per_token": 0.0,
"output_cost_per_token": 0.0,
},
}
)
model_info = litellm.get_model_info(model, custom_llm_provider=provider)
print("model_info", model_info)
assert model_info["input_cost_per_token"] == 0.0
def test_get_model_info_ollama_chat():
from litellm.llms.ollama.completion.transformation import OllamaConfig
with patch.object(
litellm.module_level_client,
"post",
return_value=MagicMock(
json=lambda: {
"model_info": {"llama.context_length": 32768},
"template": "tools",
}
),
) as mock_client:
info = OllamaConfig().get_model_info("unknown-model")
assert info["supports_function_calling"] is True
info = get_model_info("ollama/unknown-model")
print("info", info)
assert info["supports_function_calling"] is True
mock_client.assert_called()
print(mock_client.call_args.kwargs)
assert mock_client.call_args.kwargs["json"]["name"] == "unknown-model"
def test_get_model_info_bedrock_region(monkeypatch):
regional_model = "us.anthropic.claude-haiku-4-5-20251001-v1:0"
monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True")
model_cost_without_regional_entry = {
key: value for key, value in litellm.get_model_cost_map(url="").items() if key != regional_model
}
monkeypatch.setattr(litellm, "model_cost", model_cost_without_regional_entry)
_invalidate_model_cost_lowercase_map()
info = litellm.get_model_info(model=regional_model, custom_llm_provider="bedrock")
print("info", info)
assert info["key"] == "anthropic.claude-haiku-4-5-20251001-v1:0"
assert info["litellm_provider"] == "bedrock_converse"
@pytest.mark.parametrize(
"model",
[
"ft:gpt-3.5-turbo:my-org:custom_suffix:id",
"ft:gpt-4-0613:my-org:custom_suffix:id",
"ft:davinci-002:my-org:custom_suffix:id",
"ft:babbage-002:my-org:custom_suffix:id",
"gpt-35-turbo",
"ada",
],
)
def test_get_model_info_completion_cost_unit_tests(model):
info = litellm.get_model_info(model)
print("info", info)
def test_get_model_info_ft_model_with_provider_prefix():
args = {
"model": "openai/ft:gpt-3.5-turbo:my-org:custom_suffix:id",
"custom_llm_provider": "openai",
}
info = litellm.get_model_info(**args)
print("info", info)
assert info["key"] == "ft:gpt-3.5-turbo"
def _enforce_bedrock_converse_models(
model_cost: Mapping[str, ModelInfoBase], whitelist_models: Collection[str]
) -> None:
for model, info in model_cost.items():
if (
info.get("litellm_provider") == "bedrock"
and info.get("mode") == "chat"
and model not in whitelist_models
and not (
(base_model := BedrockModelInfo.get_base_model(model)) != model
and model_cost.get(base_model, {}).get("litellm_provider") == "bedrock_converse"
and BedrockModelInfo.get_bedrock_route(model) == "converse"
)
):
raise AssertionError(f"Unlisted Bedrock chat model does not route to Converse: {model}")
def _read_whitelisted_bedrock_models() -> tuple[str, ...]:
path: Final = Path(__file__).resolve().parents[2] / "whitelisted_bedrock_models.txt"
return tuple(path.read_text().splitlines())
def _normalize_bedrock_model_key(model_key: str) -> str:
without_wildcard: Final = model_key.replace("*/", "")
return re.sub(r"(?:1-month-commitment|3-month-commitment|6-month-commitment)/", "", without_wildcard)
def test_model_info_bedrock_converse(monkeypatch):
"""
Assert unlisted Bedrock chat models declare or inherit Converse routing.
This ensures they are automatically routed to the converse endpoint.
"""
monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True")
litellm.model_cost = litellm.get_model_cost_map(url="")
try:
# Load whitelist models from file
with open("whitelisted_bedrock_models.txt", "r") as file:
whitelist_models = [line.strip() for line in file.readlines()]
except FileNotFoundError:
pytest.skip("whitelisted_bedrock_models.txt not found")
_enforce_bedrock_converse_models(
model_cost=litellm.model_cost, whitelist_models=whitelist_models
)
@pytest.mark.flaky(retries=6, delay=2)
def test_model_info_bedrock_converse_enforcement(monkeypatch):
"""
Test the enforcement of the whitelist by adding a fake model and ensuring the test fails.
"""
monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True")
litellm.model_cost = litellm.get_model_cost_map(url="")
# Add a fake unwhitelisted model
litellm.model_cost["fake.bedrock-chat-model"] = {
"litellm_provider": "bedrock",
"mode": "chat",
}
try:
# Load whitelist models from file
with open("whitelisted_bedrock_models.txt", "r") as file:
whitelist_models = [line.strip() for line in file.readlines()]
# Check for unwhitelisted models
with pytest.raises(AssertionError, match=r"fake\.bedrock-chat-model"):
_enforce_bedrock_converse_models(
model_cost=litellm.model_cost, whitelist_models=whitelist_models
)
except FileNotFoundError as e:
pytest.skip("whitelisted_bedrock_models.txt not found")
@pytest.mark.parametrize("region", ("us-gov-east-1", "us-gov-west-1"))
@pytest.mark.parametrize("base_provider", ("bedrock_converse", "bedrock"))
def test_regional_bedrock_alias_requires_canonical_converse_metadata(
region: str, base_provider: Literal["bedrock_converse", "bedrock"]
) -> None:
base_model: Final = next(
model for model in sorted(litellm.bedrock_converse_models) if BedrockModelInfo.get_base_model(model) == model
)
model: Final = f"bedrock/{region}/{base_model}"
model_cost: Final[Mapping[str, ModelInfoBase]] = {
model: {"litellm_provider": "bedrock", "mode": "chat"},
base_model: {"litellm_provider": base_provider, "mode": "chat"},
}
assert BedrockModelInfo.get_bedrock_route(model) == "converse"
if base_provider == "bedrock":
with pytest.raises(AssertionError, match=re.escape(model)):
_enforce_bedrock_converse_models(model_cost, ())
return
_enforce_bedrock_converse_models(model_cost, ())
GOVCLOUD_LAGGING_CAPABILITIES: Final = frozenset({"supports_bedrock_native_structured_output"})
def _govcloud_may_lag(model_key: str, capability: str) -> bool:
is_govcloud: Final = model_key.startswith("us-gov.") or any(
part.startswith("us-gov-") for part in model_key.split("/")
)
return is_govcloud and capability in GOVCLOUD_LAGGING_CAPABILITIES
def test_get_model_info_bedrock_models():
"""
Check for drift in base model info for bedrock models and regional model info for bedrock models.
GovCloud regions may lag their commercial base on Bedrock native structured output, never lead it.
"""
from litellm.llms.bedrock.common_utils import BedrockModelInfo
os.environ["LITELLM_LOCAL_MODEL_COST_MAP"] = "True"
litellm.model_cost = litellm.get_model_cost_map(url="")
for k, v in litellm.model_cost.items():
if v["litellm_provider"] == "bedrock":
k = k.replace("*/", "")
potential_commitments = [
"1-month-commitment",
"3-month-commitment",
"6-month-commitment",
]
if any(commitment in k for commitment in potential_commitments):
for commitment in potential_commitments:
k = k.replace(f"{commitment}/", "")
base_model = BedrockModelInfo.get_base_model(k)
# get_base_model() returns model id without "bedrock/" prefix; cost map keys use "bedrock/<model>"
base_model_key = (
base_model
if base_model in litellm.model_cost
else f"bedrock/{base_model}"
)
if base_model_key not in litellm.model_cost:
continue
base_model_info = litellm.model_cost[base_model_key]
for base_model_key, base_model_value in base_model_info.items():
if "invoke/" in k:
continue
if base_model_key.startswith("supports_"):
assert (
base_model_key in v
), f"{base_model_key} is not in model cost map for {k}"
if _govcloud_may_lag(k, base_model_key):
assert base_model_value or not v[base_model_key], f"{base_model_key} is on {k} but not its base"
continue
assert (
v[base_model_key] == base_model_value
), f"{base_model_key} is not equal to {base_model_value} for model {k}"
def _cross_region_base_model_key(
model_key: str,
prefixes: tuple[str, ...],
model_cost: Mapping[str, Mapping[str, object]],
) -> str | None:
matched_prefix: Final = next((prefix for prefix in prefixes if model_key.startswith(prefix)), None)
if matched_prefix is None:
return None
base_model_key: Final = model_key[len(matched_prefix) :]
return base_model_key if base_model_key in model_cost else None
def _cross_region_profiles(
model_cost: Mapping[str, Mapping[str, object]],
prefixes: tuple[str, ...],
) -> tuple[tuple[str, Mapping[str, object], str], ...]:
return tuple(
(model_key, model_info, base_model_key)
for model_key, model_info in model_cost.items()
if str(model_info.get("litellm_provider", "")).startswith("bedrock")
and (
base_model_key := _cross_region_base_model_key(model_key, prefixes, model_cost)
)
is not None
)
def test_get_model_info_bedrock_cross_region_capability_parity():
"""
Cross-region inference profiles carry litellm_provider "bedrock_converse", so the
regional drift check above (which filters on "bedrock") never reaches them.
GovCloud is a separate AWS partition that gets features later, so a us-gov profile
may lack Bedrock native structured output its commercial base has, never the reverse.
"""
os.environ["LITELLM_LOCAL_MODEL_COST_MAP"] = "True"
litellm.model_cost = litellm.get_model_cost_map(url="")
prefixes = ("us.", "eu.", "apac.", "us-gov.")
checked = 0
for k, v in litellm.model_cost.items():
if not str(v.get("litellm_provider", "")).startswith("bedrock"):
continue
base_model_key = next(
(k[len(p) :] for p in prefixes if k.startswith(p)),
None,
)
if base_model_key is None or base_model_key not in litellm.model_cost:
continue
checked += 1
for cap, base_value in litellm.model_cost[base_model_key].items():
if not cap.startswith("supports_"):
continue
assert cap in v, f"{cap} is on {base_model_key} but missing from {k}"
if _govcloud_may_lag(k, cap):
assert base_value or not v[cap], f"{cap} is on {k} but not on {base_model_key}"
continue
assert (
v[cap] == base_value
), f"{cap} is {v[cap]} on {k} but {base_value} on {base_model_key}"
assert checked > 0, "no cross-region bedrock profiles found - the filter is inert"
def _is_positive_cost(value: object) -> bool:
return isinstance(value, (int, float)) and value > 0
def test_get_model_info_bedrock_priced_cross_region_profile_has_priced_base():
os.environ["LITELLM_LOCAL_MODEL_COST_MAP"] = "True"
litellm.model_cost = litellm.get_model_cost_map(url="")
prefixes = ("us.", "eu.", "apac.", "us-gov.", "au.", "global.")
checked = 0
for k, v in litellm.model_cost.items():
if not str(v.get("litellm_provider", "")).startswith("bedrock"):
continue
base_model_key = next(
(k[len(p) :] for p in prefixes if k.startswith(p)),
None,
)
if base_model_key is None or base_model_key not in litellm.model_cost:
continue
checked += 1
base = litellm.model_cost[base_model_key]
for cost_key in ("input_cost_per_token", "output_cost_per_token"):
if (v.get(cost_key) or 0) > 0:
assert (
base.get(cost_key) or 0
) > 0, f"{k} charges {cost_key} but its base {base_model_key} is free"
assert checked > 0, "no cross-region bedrock profiles found - the filter is inert"
def test_get_model_info_case_insensitive_lookup(monkeypatch):
"""
Test that model info lookup is case-insensitive.
This ensures that users can use lowercase model names even when the model cost
map has mixed-case keys (e.g., "Qwen/Qwen3-Next-80B-A3B-Thinking").
Related Slack discussion: Users were getting "does not support parameters: ['tools']"
errors when using lowercase model names like "qwen/qwen3-next-80b-a3b-thinking"
because the lookup was case-sensitive.
"""
monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True")
litellm.model_cost = litellm.get_model_cost_map(url="")
# Register a test model with mixed-case name
litellm.register_model(
{
"together_ai/Qwen/Qwen3-Next-80B-A3B-Thinking": {
"input_cost_per_token": 0.0001,
"output_cost_per_token": 0.0002,
"litellm_provider": "together_ai",
"supports_function_calling": True,
}
}
)
# Test 1: Exact case should work
info = litellm.get_model_info(
model="Qwen/Qwen3-Next-80B-A3B-Thinking", custom_llm_provider="together_ai"
)
assert info is not None
assert info["supports_function_calling"] is True
# Test 2: Lowercase should also work (case-insensitive lookup)
info_lower = litellm.get_model_info(
model="qwen/qwen3-next-80b-a3b-thinking", custom_llm_provider="together_ai"
)
assert info_lower is not None
assert info_lower["supports_function_calling"] is True
# Test 3: Mixed case should also work
info_mixed = litellm.get_model_info(
model="QWEN/qwen3-NEXT-80b-a3b-thinking", custom_llm_provider="together_ai"
)
assert info_mixed is not None
assert info_mixed["supports_function_calling"] is True
def test_get_model_info_case_insensitive_supports_function_calling(monkeypatch):
"""
Test that supports_function_calling check works with case-insensitive model lookup.
"""
monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True")
litellm.model_cost = litellm.get_model_cost_map(url="")
# Register a model with mixed-case name that supports function calling
litellm.register_model(
{
"test_provider/TestModel-ABC": {
"input_cost_per_token": 0.0001,
"output_cost_per_token": 0.0002,
"litellm_provider": "test_provider",
"supports_function_calling": True,
}
}
)
# Test that supports_function_calling works with lowercase model name
from litellm.utils import supports_function_calling
# Exact case
assert (
supports_function_calling("TestModel-ABC", custom_llm_provider="test_provider")
is True
)
# Lowercase (should now work with case-insensitive lookup)
assert (
supports_function_calling("testmodel-abc", custom_llm_provider="test_provider")
is True
)
def test_get_model_info_custom_model_router():
from litellm import Router
from litellm import get_model_info
litellm.turn_on_debug()
router = Router(
model_list=[
{
"model_name": "ma-summary",
"litellm_params": {
"api_base": "http://ma-mix-llm-serving.cicero.svc.cluster.local/v1",
"input_cost_per_token": 1,
"output_cost_per_token": 1,
"model": "openai/meta-llama/Meta-Llama-3-8B-Instruct",
},
"model_info": {
"id": "c20d603e-1166-4e0f-aa65-ed9c476ad4ca",
},
}
]
)
info = get_model_info("c20d603e-1166-4e0f-aa65-ed9c476ad4ca")
print("info", info)
assert info is not None
def test_get_model_info_custom_provider():
# Custom provider example copied from https://docs.litellm.ai/docs/providers/custom_llm_server:
import litellm
from litellm import CustomLLM, completion
class MyCustomLLM(CustomLLM):
def completion(self, *args, **kwargs) -> litellm.ModelResponse:
return litellm.completion(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": "Hello world"}],
mock_response="Hi!",
) # type: ignore
my_custom_llm = MyCustomLLM()
litellm.custom_provider_map = [ # 👈 KEY STEP - REGISTER HANDLER
{"provider": "my-custom-llm", "custom_handler": my_custom_llm}
]
resp = completion(
model="my-custom-llm/my-fake-model",
messages=[{"role": "user", "content": "Hello world!"}],
)
assert resp.choices[0].message.content == "Hi!"
# Register model info
model_info = {"my-custom-llm/my-fake-model": {"max_tokens": 2048}}
litellm.register_model(model_info)
# Get registered model info
from litellm import get_model_info
get_model_info(
model="my-custom-llm/my-fake-model"
) # 💥 "Exception: This model isn't mapped yet." in v1.56.10