fix(bedrock): send native structured output first on Opus 4.8 and Sonnet 5 and fall back to the json tool when Bedrock rejects it (#47100)

* fix(bedrock): send native structured output first on Opus 4.8 and Sonnet 5 and fall back to the json tool when Bedrock rejects it

* fix(bedrock): retry with the json tool when Bedrock refuses the native output schema

Count a 400 under output_config.format.schema: as a native output rejection, so a schema Bedrock cannot compile (like a pattern with a nested {n,m} group) gets the json tool retry instead of reaching the caller

* fix(bedrock): make the native output retry send what an opted-out deployment sends

The retry after a native output rejection now reruns the same optional param
mapping completion() ran, with the deployment model_info opted out of
supports_bedrock_native_structured_output, so tool_choice, effort and the json
tool schema match an opted-out deployment instead of a hand-built tool. Each
config keeps only the predicate that decides when native output came from the
opt-in key.

Bedrock batch records for Anthropic models map with the opt-in turned off,
since a batch job has no retry. Models on supports_native_structured_output
keep native output in batch

* fix(bedrock): build the messages retry from the caller's request and keep other errors unchanged

* fix(bedrock): give sync converse stream errors the upstream body as text

* fix(bedrock): let a malformed caller outputConfig reach Bedrock instead of raising

A Converse json_schema request whose caller outputConfig failed litellm's
shape check raised a pydantic error before the request was sent, so the
caller got a 500 where the merge base passed Bedrock's 400 through. The
opt-in check now treats that outputConfig as carrying no native output.

Also opts the vertex wire test's pinned Opus 4.8 json tool deployment out of
native output, and extends the Bedrock native output integration tests.

* fix(bedrock): send a refused caller outputConfig once and give opt-out advice only when native output is unsupported

A caller can send its own outputConfig on Converse. When Bedrock refused that
schema on an opt-in deployment, the retry still carried the caller's textFormat,
so Bedrock got the same request twice and the warning told the operator to turn
off an opt-in that was not the cause. The retry is now skipped when the opted-out
request would still carry a native textFormat, so Bedrock's 400 reaches the
caller after one call as on the merge base. The opt-out advice is logged only for
the account-level "Extra inputs are not permitted" rejection.

* fix(bedrock): keep the caller's continuation message on the native output retry

The Converse transform pops user_continue_message from litellm_params on the first attempt, so a retry built from those params afterwards dropped it. Both Bedrock chat routes now hand the fallback a copy taken before the first attempt.

* test(bedrock): pin additional_drop_params on the /v1/messages native output retry

* fix(bedrock): give the Converse native output retry the fake stream timeout
This commit is contained in:
Mateo Wang 2026-10-18 00:06:07 +00:00 • committed by Joshua Valluru
parent 21d80a13f0
commit a9bad2b5e0
24 changed files with 1859 additions and 88 deletions

View file

@ -1,5 +1,5 @@
from abc import ABC, abstractmethod
from collections.abc import AsyncIterator
from collections.abc import AsyncIterator, Mapping
from typing import TYPE_CHECKING, Any, Final
import httpx
@ -180,7 +180,7 @@ class BaseAnthropicMessagesConfig(ABC):
return 2
def should_retry_anthropic_messages_on_http_error(
self, e: httpx.HTTPStatusError, litellm_params: dict, request_body: object = None
self, e: httpx.HTTPStatusError, litellm_params: Mapping[str, object], request_body: object = None
) -> bool:
"""
When True, async_anthropic_messages_handler will transform the request body
@ -202,7 +202,9 @@ class BaseAnthropicMessagesConfig(ABC):
) and is_anthropic_thinking_binding_mismatch_error(error_text)
return not caller_asked_for_this_error and is_anthropic_invalid_thinking_block_error(error_text)
def transform_anthropic_messages_request_on_http_error(self, e: httpx.HTTPStatusError, request_data: dict) -> dict:
def transform_anthropic_messages_request_on_http_error(
self, e: httpx.HTTPStatusError, request_data: dict[str, object]
) -> dict[str, object]:
"""
Mutates request_data in place when retrying after a recoverable HTTP error.
"""
@ -214,3 +216,16 @@ class BaseAnthropicMessagesConfig(ABC):
if e.response.status_code == 400 and is_anthropic_invalid_thinking_block_error(e.response.text):
strip_thinking_blocks_from_anthropic_messages_request_dict(request_data)
return request_data
def native_output_fallback_litellm_params(
self,
model: str,
anthropic_messages_optional_request_params: Mapping[str, object],
litellm_params: GenericLiteLLMParams,
) -> GenericLiteLLMParams | None:
return None
def should_retry_anthropic_messages_without_native_output(
self, e: httpx.HTTPStatusError, request_body: Mapping[str, object], model: str
) -> bool:
return False

View file

@ -39,14 +39,22 @@ def make_sync_call(
if client is None:
client = get_httpx_client() # Create a new client if none provided
response: Final = client.post(
api_base,
headers=headers,
data=data,
stream=not fake_stream,
logging_obj=logging_obj,
timeout=timeout,
)
try:
response: Final = client.post(
api_base,
headers=headers,
data=data,
stream=not fake_stream,
logging_obj=logging_obj,
timeout=timeout,
)
except httpx.HTTPStatusError as err:
raise BedrockError(
status_code=err.response.status_code,
message=error_response_text(err.response),
headers=err.response.headers,
response=err.response,
) from err
if response.status_code != 200:
raise BedrockError(

View file

@ -12,7 +12,7 @@ from itertools import chain
from typing import TYPE_CHECKING, Final, Literal, cast, overload
import httpx
from pydantic import TypeAdapter
from pydantic import TypeAdapter, ValidationError
import litellm
from litellm._logging import verbose_logger
@ -58,7 +58,9 @@ from litellm.llms.bedrock.common_utils import (
bedrock_cleaned_output_schema,
bedrock_model_supports_regex_lookaround,
bedrock_native_output_format,
bedrock_native_output_is_opt_in,
deployment_model_info,
requests_output_config_format,
)
from litellm.llms.bedrock.request_metadata import (
bedrock_request_metadata_headers,
@ -151,6 +153,18 @@ def _tools_the_model_accepts(
return [tool_with_sanitized_parameters(tool, drop_lookaround_regex_patterns) for tool in tools]
_OUTPUT_CONFIG_BLOCK: Final = TypeAdapter(OutputConfigBlock)
def _has_native_text_format(output_config: object) -> bool:
if not isinstance(output_config, Mapping):
return False
try:
return _OUTPUT_CONFIG_BLOCK.validate_python(output_config).get("textFormat") is not None
except ValidationError:
return False
class AmazonConverseConfig(BaseConfig):
"""
Reference - https://docs.aws.amazon.com/bedrock/latest/APIReference/API_runtime_Converse.html
@ -1358,6 +1372,22 @@ class AmazonConverseConfig(BaseConfig):
optional_params["json_mode"] = True
return optional_params
@staticmethod
def sends_native_text_format(optional_params: Mapping[str, object]) -> bool:
return _has_native_text_format(optional_params.get("outputConfig"))
def uses_opt_in_native_output(
self, model: str, optional_params: Mapping[str, object], litellm_params: Mapping[str, object]
) -> bool:
native_output_requested: Final = (
optional_params.get("json_mode") is True and self.sends_native_text_format(optional_params)
) or requests_output_config_format(optional_params)
return native_output_requested and bedrock_native_output_is_opt_in(
bedrock_capability_model(model, litellm_params.get("base_model")),
self.custom_llm_provider or "bedrock_converse",
deployment_model_info(litellm_params),
)
def update_optional_params_with_thinking_tokens(self, non_default_params: dict, optional_params: dict):
"""
Handles scenario where max tokens is not specified. For anthropic models (anthropic api/bedrock/vertex ai), this requires having the max tokens being set and being greater than the thinking token budget.

View file

@ -26,6 +26,7 @@ from litellm.llms.bedrock.common_utils import (
normalize_bedrock_opus_output_config_effort,
normalize_custom_field_on_tools,
normalize_tool_input_schema_types_for_bedrock_invoke,
requests_opt_in_native_output,
strip_unsupported_bedrock_invoke_output_config_keys,
tools_without_eager_input_streaming,
)
@ -99,6 +100,11 @@ class AmazonAnthropicClaudeConfig(AmazonInvokeConfig, AnthropicConfig):
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 uses_opt_in_native_output(
self, model: str, optional_params: Mapping[str, object], litellm_params: Mapping[str, object]
) -> bool:
return requests_opt_in_native_output(model, optional_params, deployment_model_info(litellm_params))
def map_openai_params(
self,
non_default_params: dict,

View file

@ -0,0 +1,62 @@
from collections.abc import Callable, Coroutine
from typing import Final, TypeAlias
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.llms.bedrock.common_utils import (
NativeOutputFallback,
is_bedrock_native_output_rejection,
warn_bedrock_native_output_rejected,
)
from litellm.types.utils import ModelResponse
from litellm.utils import CustomStreamWrapper
NativeOutputResponse: TypeAlias = ModelResponse | CustomStreamWrapper
NativeOutputAttemptResult: TypeAlias = NativeOutputResponse | Coroutine[object, object, NativeOutputResponse]
NativeOutputAttempt: TypeAlias = Callable[[NativeOutputFallback | None], NativeOutputAttemptResult]
NativeOutputFallbackFactory: TypeAlias = Callable[[], NativeOutputFallback | None]
def _fallback_for(
error: BaseLLMException, fallback: NativeOutputFallbackFactory, model: str
) -> NativeOutputFallback | None:
if not is_bedrock_native_output_rejection(error.status_code, error.message):
return None
retry: Final = fallback()
if retry is not None:
warn_bedrock_native_output_rejected(model, error.message)
return retry
async def _await_with_native_output_fallback(
first_attempt: Coroutine[object, object, NativeOutputResponse],
attempt: NativeOutputAttempt,
fallback: NativeOutputFallbackFactory,
model: str,
) -> NativeOutputResponse:
try:
return await first_attempt
except BaseLLMException as error:
retry: Final = _fallback_for(error, fallback, model)
if retry is None:
raise
second_attempt: Final = attempt(retry)
return await second_attempt if isinstance(second_attempt, Coroutine) else second_attempt
def call_with_native_output_fallback(
attempt: NativeOutputAttempt,
fallback: NativeOutputFallbackFactory | None,
model: str,
) -> NativeOutputAttemptResult:
if fallback is None:
return attempt(None)
try:
first_attempt: Final = attempt(None)
except BaseLLMException as error:
retry: Final = _fallback_for(error, fallback, model)
if retry is None:
raise
return attempt(retry)
if isinstance(first_attempt, Coroutine):
return _await_with_native_output_fallback(first_attempt, attempt, fallback, model)
return first_attempt

View file

@ -10,6 +10,7 @@ import json
import os
import re
from collections.abc import Iterator, Mapping, Sequence
from dataclasses import dataclass
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, Literal, TypeAlias
@ -181,7 +182,7 @@ def _get_local_model_cost_map() -> dict:
return GetModelCostMap.load_local_model_cost_map()
def pop_bedrock_invoke_output_config_format(request_body: dict) -> dict | None:
def pop_bedrock_invoke_output_config_format(request_body: dict[str, object]) -> dict[str, object] | None:
"""
Remove and return Anthropic's nested ``output_config.format`` field.
@ -202,8 +203,8 @@ def pop_bedrock_invoke_output_config_format(request_body: dict) -> dict | None:
def convert_bedrock_invoke_output_format_to_inline_schema(
output_format: dict,
request_body: dict,
output_format: Mapping[str, object],
request_body: dict[str, object],
) -> None:
"""
Embed an Anthropic structured-output schema into the last user message.
@ -289,6 +290,77 @@ def bedrock_chat_uses_native_structured_output(
)
def bedrock_native_output_is_opt_in(
model: str, custom_llm_provider: str = "bedrock", model_info: Mapping[str, object] | None = None
) -> bool:
return not _bedrock_model_supports(
model, "supports_native_structured_output", custom_llm_provider
) and _bedrock_native_output_opt_in(model, custom_llm_provider, model_info)
def without_bedrock_native_output_opt_in(litellm_params: Mapping[str, object]) -> Mapping[str, object]:
return {
**litellm_params,
"model_info": {**deployment_model_info(litellm_params), "supports_bedrock_native_structured_output": False},
}
_BEDROCK_NATIVE_OUTPUT_UNSUPPORTED: Final = "output_config.format: Extra inputs are not permitted"
_BEDROCK_NATIVE_OUTPUT_REJECTIONS: Final = (
_BEDROCK_NATIVE_OUTPUT_UNSUPPORTED,
"compilation exceeded limits: Schema is too complex",
"output_config.format.schema:",
)
def is_bedrock_native_output_rejection(status_code: int, error_text: str) -> bool:
return status_code == 400 and any(rejection in error_text for rejection in _BEDROCK_NATIVE_OUTPUT_REJECTIONS)
def warn_bedrock_native_output_rejected(model: str, error_text: str) -> None:
opt_out_advice: Final = (
" Set `supports_bedrock_native_structured_output: false` in this deployment's model_info to skip the "
"rejected call."
if _BEDROCK_NATIVE_OUTPUT_UNSUPPORTED in error_text
else ""
)
verbose_logger.warning(
"Bedrock rejected native structured output for model=%s, so litellm sent the request again without it.%s",
model,
opt_out_advice,
)
@dataclass(frozen=True, slots=True)
class NativeOutputFallback:
optional_params: Mapping[str, object]
litellm_params: Mapping[str, object]
def requests_output_config_format(optional_params: Mapping[str, object]) -> bool:
match optional_params:
case {"output_config": {"format": object()}}:
return True
case _:
return False
def requests_opt_in_native_output(
model: str, anthropic_optional_params: Mapping[str, object], model_info: Mapping[str, object] | None
) -> bool:
native_output_requested: Final = isinstance(
anthropic_optional_params.get("output_format"), Mapping
) or requests_output_config_format(anthropic_optional_params)
return native_output_requested and bedrock_native_output_is_opt_in(model, model_info=model_info)
_JSON_OBJECT: Final = TypeAdapter(dict[str, object])
def json_object_or_none(value: object) -> Mapping[str, object] | None:
return _JSON_OBJECT.validate_python(value) if isinstance(value, Mapping) else None
BEDROCK_NATIVE_OUTPUT_MAX_PROPERTIES: Final = 64
@ -355,7 +427,8 @@ def apply_bedrock_invoke_structured_output(
"""
legacy_output_format: Final = request_body.pop("output_format", None)
output_config_format: Final = pop_bedrock_invoke_output_config_format(request_body)
schema_format: Final = legacy_output_format if isinstance(legacy_output_format, dict) else output_config_format
legacy_format: Final = json_object_or_none(legacy_output_format)
schema_format: Final = legacy_format if legacy_format is not None else output_config_format
if schema_format is None:
return

View file

@ -66,8 +66,10 @@ from litellm.utils import get_llm_provider, get_optional_params
from ..base_aws_llm import BaseAWSLLM
from ..common_utils import (
BedrockError,
deployment_model_info,
merge_bedrock_aws_request_params,
resolve_s3_encryption_key_id,
without_bedrock_native_output_opt_in,
)
S3_SIGNED_REQUEST_HEADERS_PARAM: Final = "_s3_signed_request_headers"
@ -982,10 +984,12 @@ class BedrockFilesConfig(BaseAWSLLM, BaseFilesConfig):
)
config: Final = AmazonAnthropicClaudeConfig()
litellm_params: Final = without_bedrock_native_output_opt_in({})
mapped_params = get_optional_params(
model=_invoke_route_model(model),
custom_llm_provider="bedrock",
messages=messages,
model_info=deployment_model_info(litellm_params),
**MappingProxyType(
{
k: v
@ -1000,7 +1004,7 @@ class BedrockFilesConfig(BaseAWSLLM, BaseFilesConfig):
optional_params={
k: v for k, v in mapped_params.items() if k not in _MAPPED_PARAMS_THE_REQUEST_HANDLER_STRIPS
},
litellm_params={},
litellm_params=dict(litellm_params),
headers={},
)

View file

@ -40,11 +40,15 @@ from litellm.llms.bedrock.common_utils import (
ensure_bedrock_anthropic_messages_tool_names,
get_anthropic_beta_from_headers,
is_claude_4_5_on_bedrock,
is_bedrock_native_output_rejection,
normalize_bedrock_opus_output_config_effort,
normalize_custom_field_on_tools,
normalize_tool_input_schema_types_for_bedrock_invoke,
requests_opt_in_native_output,
requests_output_config_format,
strip_unsupported_bedrock_invoke_output_config_keys,
tools_without_eager_input_streaming,
warn_bedrock_native_output_rejected,
)
from litellm.llms.bedrock.request_metadata import (
bedrock_request_metadata_headers,
@ -111,6 +115,29 @@ class AmazonAnthropicClaudeMessagesConfig(
BaseAnthropicMessagesConfig.__init__(self, **kwargs)
AmazonInvokeConfig.__init__(self, **kwargs)
def native_output_fallback_litellm_params(
self,
model: str,
anthropic_messages_optional_request_params: Mapping[str, object],
litellm_params: GenericLiteLLMParams,
) -> GenericLiteLLMParams | None:
model_info: Final = _MODEL_INFO.validate_python(litellm_params.model_info)
if not requests_opt_in_native_output(model, anthropic_messages_optional_request_params, model_info):
return None
return litellm_params.model_copy(
update={"model_info": {**(model_info or {}), "supports_bedrock_native_structured_output": False}}
)
def should_retry_anthropic_messages_without_native_output(
self, e: httpx.HTTPStatusError, request_body: Mapping[str, object], model: str
) -> bool:
if not requests_output_config_format(request_body) or not is_bedrock_native_output_rejection(
e.response.status_code, e.response.text
):
return False
warn_bedrock_native_output_rejected(model, e.response.text)
return True
def validate_anthropic_messages_environment(
self,
headers: dict,

View file

@ -1,8 +1,9 @@
import asyncio
import copy
import inspect
import json
import ssl
from collections.abc import AsyncGenerator, AsyncIterator, Awaitable, Coroutine, Iterator, Mapping, Sequence
from collections.abc import AsyncGenerator, AsyncIterator, Awaitable, Callable, Coroutine, Iterator, Mapping, Sequence
from contextlib import asynccontextmanager
from functools import lru_cache, partial
from types import MappingProxyType
@ -464,6 +465,9 @@ async def _aiter_bytes_then_close(response: httpx.Response, *, chunk_size: int)
_DECODED_BODY_STALE_HEADERS: Final[frozenset[str]] = frozenset({"content-encoding", "content-length"})
_OPTIONAL_PARAMS: Final = TypeAdapter(Mapping[str, object])
_JSON_OBJECT: Final = TypeAdapter(dict[str, object])
_JSON_OBJECTS: Final = TypeAdapter(list[dict[str, object]])
def _decoded_body_headers(response: httpx.Response) -> httpx.Headers:
@ -2099,11 +2103,13 @@ class BaseLLMHTTPHandler:
api_key: str | None,
model: str,
timeout: float | httpx.Timeout | None = None,
rebuild_without_native_output: Callable[[], Mapping[str, object]] | None = None,
) -> httpx.Response:
max_attempts: Final = max(provider_config.max_retry_on_anthropic_messages_http_error, 1)
litellm_params_dict: Final = dict(litellm_params)
optional_params_dict: Final = dict(litellm_params)
for attempt_idx in range(max_attempts):
recovered_errors: tuple[httpx.HTTPStatusError, ...] = () # rebind-ok: a rebuilt body replays every recovery
while True:
try:
response = await async_httpx_client.post(
url=request_url,
@ -2116,40 +2122,54 @@ class BaseLLMHTTPHandler:
response.raise_for_status()
return response
except httpx.HTTPStatusError as e:
hit_max_attempt = attempt_idx + 1 == max_attempts
rebuild_body = (
rebuild_without_native_output
if rebuild_without_native_output is not None
and provider_config.should_retry_anthropic_messages_without_native_output(
e=e, request_body=request_body, model=model
)
else None
)
hit_max_attempt = len(recovered_errors) + 1 == max_attempts
should_retry = provider_config.should_retry_anthropic_messages_on_http_error(
e=e, litellm_params=litellm_params_dict, request_body=request_body
)
if should_retry and not hit_max_attempt:
if logging_obj.baseline_cache_context is not None:
await logging_obj.invalidate_baseline_cache_estimate("retried_request")
if rebuild_body is None and (hit_max_attempt or not should_retry):
raise self._handle_error(e=e, provider_config=provider_config)
if logging_obj.baseline_cache_context is not None:
await logging_obj.invalidate_baseline_cache_estimate("retried_request")
if rebuild_body is not None:
request_body.clear()
request_body.update(rebuild_body())
for recovered_error in recovered_errors:
provider_config.transform_anthropic_messages_request_on_http_error(
e=recovered_error, request_data=request_body
)
else:
verbose_logger.debug(
"Anthropic /v1/messages: invalid thinking signature; "
"stripping thinking blocks and retrying (attempt %s/%s).",
attempt_idx + 2,
len(recovered_errors) + 2,
max_attempts,
)
provider_config.transform_anthropic_messages_request_on_http_error(e=e, request_data=request_body)
headers, signed_json_body = await sign_request_off_loop_if_aws(
provider_config,
provider_config.sign_request,
headers=headers,
optional_params=optional_params_dict,
request_data=request_body,
api_base=request_url,
api_key=api_key,
stream=stream,
fake_stream=False,
model=model,
)
logging_obj.model_call_details.update(request_body)
continue
raise self._handle_error(e=e, provider_config=provider_config)
recovered_errors = (*recovered_errors, e)
headers, signed_json_body = await sign_request_off_loop_if_aws(
provider_config,
provider_config.sign_request,
headers=headers,
optional_params=optional_params_dict,
request_data=request_body,
api_base=request_url,
api_key=api_key,
stream=stream,
fake_stream=False,
model=model,
)
logging_obj.model_call_details.update(request_body)
except Exception as e:
raise self._handle_error(e=e, provider_config=provider_config)
raise RuntimeError("unreachable: anthropic messages HTTP retry loop exited without return")
@staticmethod
def _resolve_anthropic_messages_timeout(
litellm_params: GenericLiteLLMParams,
@ -2270,6 +2290,24 @@ class BaseLLMHTTPHandler:
anthropic_messages_optional_request_params, path
)
native_output_fallback_params: Final = anthropic_messages_provider_config.native_output_fallback_litellm_params(
model=model,
anthropic_messages_optional_request_params=_OPTIONAL_PARAMS.validate_python(anthropic_messages_optional_request_params),
litellm_params=litellm_params,
)
rebuild_without_native_output: Final = (
None
if native_output_fallback_params is None
else partial[Mapping[str, object]](
anthropic_messages_provider_config.transform_anthropic_messages_request,
model=model,
messages=copy.deepcopy(_JSON_OBJECTS.validate_python(messages)),
anthropic_messages_optional_request_params=copy.deepcopy(_JSON_OBJECT.validate_python(anthropic_messages_optional_request_params)),
litellm_params=native_output_fallback_params,
headers=_JSON_OBJECT.validate_python(headers),
)
)
# Prepare request body
request_body: Final = anthropic_messages_provider_config.transform_anthropic_messages_request(
model=model,
@ -2343,6 +2381,7 @@ class BaseLLMHTTPHandler:
stream=stream or False,
custom_llm_provider=custom_llm_provider,
),
rebuild_without_native_output=rebuild_without_native_output,
)
# used for logging + cost tracking

View file

@ -121,7 +121,10 @@ from litellm.llms.base_llm.base_model_iterator import (
from litellm.llms.base_llm.chat.transformation import with_attribution_headers
from litellm.llms.bedrock.common_utils import (
BedrockModelInfo,
NativeOutputFallback,
bedrock_route_for_request,
deployment_model_info,
without_bedrock_native_output_opt_in,
without_bedrock_route_prefix,
)
from litellm.llms.bedrock_mantle.chat.claude_transformation import bedrock_mantle_chat_config
@ -219,6 +222,9 @@ from .llms.azure.completion.handler import AzureTextCompletion
from .llms.azure_ai.anthropic.handler import AzureAnthropicChatCompletion
from .llms.azure_ai.embed import AzureAIEmbedding
from .llms.bedrock.chat import BedrockConverseLLM
from .llms.bedrock.chat.converse_transformation import AmazonConverseConfig
from .llms.bedrock.chat.invoke_transformations.anthropic_claude3_transformation import AmazonAnthropicClaudeConfig
from .llms.bedrock.chat.native_output_fallback import NativeOutputAttemptResult, call_with_native_output_fallback
from .llms.bedrock.embed.embedding import BedrockEmbedding
from .llms.bedrock.image_edit.handler import BedrockImageEdit
from .llms.bedrock.image_generation.image_handler import BedrockImageGeneration
@ -4188,6 +4194,43 @@ _OPTIONAL_PARAMS_ADAPTER: Final = TypeAdapter(dict[str, object])
_MODEL_INFO_ADAPTER: Final = TypeAdapter(Mapping[str, object])
_AWS_BEDROCK_CLIENT_CREDENTIAL_KEYS: Final = (
"aws_access_key_id",
"aws_secret_access_key",
"aws_session_token",
"aws_region_name",
)
def _native_output_fallback(
ctx: CompletionDispatchContext,
first_attempt_params: Mapping[str, object],
first_attempt_litellm_params: Mapping[str, object],
) -> NativeOutputFallback:
litellm_params: Final = without_bedrock_native_output_opt_in(first_attempt_litellm_params)
remapped_params: Final = ctx.optional_params_for_model_info(deployment_model_info(litellm_params))
return NativeOutputFallback(
optional_params={
**{key: value for key, value in remapped_params.items() if key != "aws_bedrock_client"},
**{
key: first_attempt_params[key]
for key in _AWS_BEDROCK_CLIENT_CREDENTIAL_KEYS
if key in first_attempt_params
},
},
litellm_params=litellm_params,
)
def _converse_native_output_fallback(
ctx: CompletionDispatchContext,
first_attempt_params: Mapping[str, object],
first_attempt_litellm_params: Mapping[str, object],
) -> NativeOutputFallback | None:
fallback: Final = _native_output_fallback(ctx, first_attempt_params, first_attempt_litellm_params)
return None if AmazonConverseConfig.sends_native_text_format(fallback.optional_params) else fallback
def _complete_bedrock(ctx: CompletionDispatchContext) -> _CompletionDispatchResult:
acompletion: Final = ctx.acompletion
api_base: Final = ctx.api_base
@ -4259,22 +4302,36 @@ def _complete_bedrock(ctx: CompletionDispatchContext) -> _CompletionDispatchResu
)
elif bedrock_route == "converse":
model = without_bedrock_route_prefix(model)
response = bedrock_converse_chat_completion.completion(
model=model,
messages=messages,
custom_prompt_dict=custom_prompt_dict,
model_response=model_response,
optional_params=optional_params,
litellm_params=litellm_params,
logger_fn=logger_fn,
encoding=_get_encoding(),
logging_obj=logging,
extra_headers=headers, # Use merged headers instead of original extra_headers
timeout=timeout,
acompletion=acompletion,
client=client,
api_base=api_base,
api_key=api_key,
converse_model: Final = model
def converse_attempt(fallback: NativeOutputFallback | None) -> NativeOutputAttemptResult:
return bedrock_converse_chat_completion.completion(
model=converse_model,
messages=messages,
custom_prompt_dict=custom_prompt_dict,
model_response=model_response,
optional_params=optional_params if fallback is None else dict(fallback.optional_params),
litellm_params=litellm_params if fallback is None else dict(fallback.litellm_params),
logger_fn=logger_fn,
encoding=_get_encoding(),
logging_obj=logging,
extra_headers=headers, # Use merged headers instead of original extra_headers
timeout=timeout,
acompletion=acompletion,
client=client,
api_base=api_base,
api_key=api_key,
)
converse_params: Final = _OPTIONAL_PARAMS_ADAPTER.validate_python(optional_params)
response = call_with_native_output_fallback(
converse_attempt,
(
partial(_converse_native_output_fallback, ctx, converse_params, dict(litellm_params))
if AmazonConverseConfig().uses_opt_in_native_output(converse_model, converse_params, litellm_params)
else None
),
converse_model,
)
elif bedrock_route == "converse_like":
model = model.replace("converse_like/", "")
@ -4296,22 +4353,36 @@ def _complete_bedrock(ctx: CompletionDispatchContext) -> _CompletionDispatchResu
client=client,
)
else:
response = base_llm_http_handler.completion(
model=model,
stream=stream,
messages=messages,
acompletion=acompletion,
api_base=api_base,
model_response=model_response,
optional_params=optional_params,
litellm_params=litellm_params,
custom_llm_provider="bedrock",
timeout=timeout,
headers=headers,
encoding=_get_encoding(),
api_key=api_key,
logging_obj=logging,
client=client,
def invoke_attempt(fallback: NativeOutputFallback | None) -> NativeOutputAttemptResult:
return base_llm_http_handler.completion(
model=model,
stream=stream,
messages=messages,
acompletion=acompletion,
api_base=api_base,
model_response=model_response,
optional_params=optional_params if fallback is None else dict(fallback.optional_params),
litellm_params=litellm_params if fallback is None else dict(fallback.litellm_params),
custom_llm_provider="bedrock",
timeout=timeout,
headers=headers,
encoding=_get_encoding(),
api_key=api_key,
logging_obj=logging,
client=client,
)
invoke_params: Final = _OPTIONAL_PARAMS_ADAPTER.validate_python(optional_params)
response = call_with_native_output_fallback(
invoke_attempt,
(
partial(_native_output_fallback, ctx, invoke_params, dict(litellm_params))
if isinstance(provider_config, AmazonAnthropicClaudeConfig)
and provider_config.uses_opt_in_native_output(model, invoke_params, litellm_params)
else None
),
model,
)
return response
@ -5657,10 +5728,14 @@ def completion(
"allowed_openai_params": allowed_openai_params,
"base_model": base_model,
}
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,
mapping_args: Final = MappingProxyType(dict(optional_param_args))
mapping_special_params: Final = MappingProxyType(dict(non_default_params))
def optional_params_for_model_info(deployment_info: Mapping[str, object] | None) -> dict[str, object]:
return get_optional_params(**mapping_args, model_info=deployment_info, **mapping_special_params)
optional_params = optional_params_for_model_info(
_MODEL_INFO_ADAPTER.validate_python(model_info) if isinstance(model_info, dict) else None
)
processed_non_default_params: Final = pre_process_non_default_params(
model=model,
@ -5883,6 +5958,7 @@ def completion(
model=model,
model_response=model_response,
optional_params=optional_params,
optional_params_for_model_info=optional_params_for_model_info,
organization=organization,
provider_config=provider_config,
request_params=MappingProxyType(

View file

@ -229,6 +229,7 @@ class CompletionDispatchContext:
model: str
model_response: ModelResponse
optional_params: dict
optional_params_for_model_info: Callable[[Mapping[str, object] | None], dict[str, object]]
organization: str | None
provider_config: BaseConfig | None
request_params: Mapping[str, object]

View file

@ -2,7 +2,7 @@ import json
from typing import Final
import pytest
from integration._support.client import Gateway
from integration._support.client import JSON_OBJECT, Gateway, list_value, object_value
from integration._support.wire import Reply, Request, wire_server
MODEL: Final = "bedrock/anthropic.claude-3-haiku-20240307-v1:0"
@ -76,3 +76,56 @@ def test_completions_and_responses_batch_records_upload_as_anthropic_user_messag
assert len(uploads) == 1, f"Expected exactly one S3 PUT, saw {[upload.target for upload in uploads]}"
stored: Final = tuple(json.loads(line) for line in uploads[0].body.decode().splitlines() if line.strip())
assert stored == EXPECTED_S3_OBJECT
_JSON_TOOL: Final = "json_tool_call"
_SCHEMA: Final = {
"type": "object",
"properties": {"city": {"type": "string"}},
"required": ["city"],
"additionalProperties": False,
}
@pytest.mark.parametrize(
"backend", ["us.anthropic.claude-haiku-5-5", "us.anthropic.claude-opus-4-8"], ids=["haiku_5_5", "opus_4_8"]
)
def test_json_schema_batch_record_with_thinking_uploads_the_json_tool_and_no_native_output(
gateway: Gateway, backend: str
) -> None:
record: Final = {
"custom_id": "schema-record",
"method": "POST",
"url": "/v1/chat/completions",
"body": {
"model": f"bedrock/{backend}",
"messages": [{"role": "user", "content": "Name a city."}],
"max_tokens": 2048,
"thinking": {"type": "adaptive"},
"response_format": {"type": "json_schema", "json_schema": {"name": "city", "schema": _SCHEMA}},
},
}
with wire_server(s3_peer) as wire, gateway.scenario() as scenario:
model: Final = scenario.model(
model=f"bedrock/{backend}",
api_key=None,
api_base=None,
aws_access_key_id="AKIAIOSFODNN7EXAMPLE",
aws_secret_access_key="wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
aws_region_name="us-east-1",
s3_bucket_name=BUCKET,
s3_endpoint_url=wire.url,
)
response: Final = gateway.request_multipart(
"/v1/files",
{"purpose": "batch", "model": model},
{"file": ("in.jsonl", (json.dumps(record) + "\n").encode(), "application/jsonl")},
)
assert response.status_code == 200, response.text
(upload,) = wire.drain()
(stored,) = tuple(JSON_OBJECT.validate_json(line) for line in upload.body.decode().splitlines() if line.strip())
model_input: Final = object_value(stored["modelInput"])
assert "format" not in object_value(model_input.get("output_config", {})), model_input
(tool,) = (object_value(entry) for entry in list_value(model_input["tools"]))
assert tool["name"] == _JSON_TOOL, model_input
assert object_value(tool["input_schema"])["properties"] == _SCHEMA["properties"], model_input

View file

@ -1,14 +1,16 @@
import base64
import json
import re
import uuid
from collections.abc import Iterable, Iterator, Mapping
from collections.abc import Callable, 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.client import Gateway, Scenario, held, list_value, object_value, string_value
from integration._support.database import read_rows
from integration._support.upstream import aws_event_stream_frame
from integration._support.wire import Reply, Request, Wire, wire_server
from pydantic import JsonValue, TypeAdapter
@ -17,7 +19,8 @@ _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"
_OPT_IN: Final = _OPUS
_OPUS_4_8: Final = "us.anthropic.claude-opus-4-8"
_DEEPSEEK: Final = "deepseek.v3.2"
_AWS: Final[dict[str, JsonValue]] = {
"aws_access_key_id": "AKIASCRIPTEDPROVIDER",
@ -38,6 +41,20 @@ _JSON_VALUE: Final[TypeAdapter[JsonValue]] = TypeAdapter(JsonValue)
_FABLE_REJECTION: Final = (
"The model returned the following errors: output_config.format: Extra inputs are not permitted"
)
_SCHEMA_TOO_COMPLEX: Final = (
"The model returned the following errors: compilation exceeded limits: Schema is too complex."
)
_PATTERN_TOO_COMPLEX: Final = (
"The model returned the following errors: output_config.format.schema: Unsupported regex feature in pattern "
"field: pattern is too complex for structured output: reduce the {n,m} upper bound, narrow the character class "
"range, or avoid nesting quantified groups"
)
_MAX_TOKENS_400: Final = (
"The model returned the following errors: max_tokens: 2048 > 1024, the maximum for this request"
)
_SIGNATURE_400: Final = (
"The model returned the following errors: messages.1.content.0: Invalid `signature` in `thinking` block"
)
_UNSUPPORTED_KEYWORDS: Final = (
"minimum",
"maximum",
@ -327,6 +344,54 @@ def _converse_frames(blocks: tuple[dict[str, JsonValue], ...], stop_reason: str)
)
def _invoke_chunk(event: Mapping[str, JsonValue]) -> bytes:
return aws_event_stream_frame(
{":event-type": "chunk", ":content-type": "application/json", ":message-type": "event"},
json.dumps({"bytes": base64.b64encode(json.dumps(event).encode()).decode()}).encode(),
)
def _invoke_block_events(index: int, block: Mapping[str, JsonValue]) -> tuple[dict[str, JsonValue], ...]:
def start(content_block: dict[str, JsonValue]) -> dict[str, JsonValue]:
return {"type": "content_block_start", "index": index, "content_block": content_block}
def delta(payload: dict[str, JsonValue]) -> dict[str, JsonValue]:
return {"type": "content_block_delta", "index": index, "delta": payload}
stop: Final[dict[str, JsonValue]] = {"type": "content_block_stop", "index": index}
match block.get("type"):
case "thinking":
return (
start({"type": "thinking", "thinking": ""}),
delta({"type": "thinking_delta", "thinking": block["thinking"]}),
delta({"type": "signature_delta", "signature": block["signature"]}),
stop,
)
case "tool_use":
return (
start({"type": "tool_use", "id": block["id"], "name": block["name"], "input": {}}),
delta({"type": "input_json_delta", "partial_json": json.dumps(block["input"])}),
stop,
)
case _:
return (start({"type": "text", "text": ""}), delta({"type": "text_delta", "text": block["text"]}), stop)
def _invoke_frames(message: Mapping[str, JsonValue]) -> bytes:
blocks: Final = tuple(object_value(block) for block in list_value(message["content"]))
events: Final[tuple[Mapping[str, JsonValue], ...]] = (
{"type": "message_start", "message": {**message, "content": [], "stop_reason": None}},
*chain.from_iterable(_invoke_block_events(index, block) for index, block in enumerate(blocks)),
{
"type": "message_delta",
"delta": {"stop_reason": message["stop_reason"], "stop_sequence": None},
"usage": {"output_tokens": 7},
},
{"type": "message_stop"},
)
return b"".join(_invoke_chunk(event) for event in events)
def _called_tool(tools: Iterable[str], native: bool) -> str | None:
names: Final = frozenset(tools)
if _USER_TOOL in names:
@ -365,6 +430,8 @@ def _bedrock(request: Request) -> Reply:
"stop_sequence": None,
"usage": {"input_tokens": 11, "output_tokens": 7},
}
if _is_stream(request):
return Reply(body=_invoke_frames(message), content_type=_EVENT_STREAM)
return Reply(body=json.dumps(message).encode())
if _is_stream(request):
return Reply(body=_converse_frames(blocks, stop_reason), content_type=_EVENT_STREAM)
@ -446,6 +513,20 @@ def _messages_text(blocks: tuple[dict[str, JsonValue], ...]) -> JsonValue:
return _JSON_VALUE.validate_json(string_value(text["text"]))
def _messages_stream_text(response: httpx.Response) -> JsonValue:
assert response.status_code == 200, response.text
events: Final = tuple(
_JSON.validate_json(line.removeprefix("data: "))
for line in response.text.splitlines()
if line.startswith("data: {")
)
deltas: Final = tuple(
object_value(event["delta"]) for event in events if event.get("type") == "content_block_delta"
)
text: Final = "".join(string_value(delta["text"]) for delta in deltas if delta.get("type") == "text_delta")
return _JSON_VALUE.validate_json(text)
def _refs(schema: JsonValue) -> frozenset[str]:
return frozenset(string_value(node["$ref"]) for node in _schema_nodes(schema) if "$ref" in node)
@ -551,15 +632,18 @@ def test_bedrock_native_model_without_thinking_keeps_the_forced_json_tool(gatewa
@pytest.mark.parametrize(
("backend", "model_info"),
[(_OPUS, None), (_OPT_IN, {"supports_bedrock_native_structured_output": True})],
ids=["neither_key", "deployment_opt_in"],
("backend", "model_info", "native"),
[
(_OPUS, None, False),
(_OPT_IN, {"supports_bedrock_native_structured_output": True}, True),
(_OPUS_4_8, {"supports_bedrock_native_structured_output": False}, False),
],
ids=["neither_key", "deployment_opt_in", "deployment_opt_out"],
)
def test_thinking_request_follows_the_bedrock_native_key_the_model_resolves_to(
gateway: Gateway, backend: str, model_info: Mapping[str, JsonValue] | None
gateway: Gateway, backend: str, model_info: Mapping[str, JsonValue] | None, native: bool
) -> 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(
@ -818,3 +902,293 @@ def test_numeric_formats_and_wide_schemas_follow_the_bedrock_native_key(
tools: Final = _tools(body, invoke)
assert (_JSON_TOOL in tools) is not native, body
assert native or tools[_JSON_TOOL]["properties"] == schema["properties"], body
def _rejecting_native_output(message: str) -> Callable[[Request], Reply]:
def reply(request: Request) -> Reply:
native: Final = _native_schema(_JSON.validate_json(request.body), _is_invoke(request))
return _bedrock(request) if native is None else _error_reply(message)
return reply
def _fallback_answer(gateway: Gateway, entry: str, model: str, marker: str) -> JsonValue:
match entry:
case "converse_chat_stream" | "invoke_chat_stream":
streamed: Final = _chat_body(
model, marker, stream=True, thinking=_ADAPTIVE, response_format=_response_format(_SCHEMA)
)
return _content_json(_stream_message(gateway.request("POST", "/v1/chat/completions", streamed)))
case "invoke_messages_stream":
streamed_messages: Final = _messages_body(
model, marker, stream=True, output_config={"format": {"type": "json_schema", "schema": _SCHEMA}}
)
return _messages_stream_text(gateway.request("POST", "/v1/messages", streamed_messages))
case "converse_output_config":
chat: Final = _chat_body(
model, marker, output_config={"format": {"type": "json_schema", "schema": _SCHEMA}}
)
return _content_json(_chat_message(gateway.request("POST", "/v1/chat/completions", chat)))
case _:
return _opt_in_answer(gateway, entry, model, marker)
@pytest.mark.parametrize(
("entry", "rejection"),
[
("converse_chat", _FABLE_REJECTION),
("converse_chat", _PATTERN_TOO_COMPLEX),
("converse_chat_stream", _SCHEMA_TOO_COMPLEX),
("converse_responses", _FABLE_REJECTION),
("converse_messages", _FABLE_REJECTION),
("converse_output_config", _FABLE_REJECTION),
("invoke_chat", _FABLE_REJECTION),
("invoke_chat_stream", _FABLE_REJECTION),
("invoke_messages", _SCHEMA_TOO_COMPLEX),
("invoke_messages_stream", _FABLE_REJECTION),
],
)
def test_bedrock_rejecting_opt_in_native_output_gets_one_retry_without_it(
gateway: Gateway, entry: str, rejection: str
) -> None:
marker, marker_hex = _marker()
route: Final = entry.split("_", 1)[0]
invoke: Final = route == "invoke"
with wire_server(_rejecting_native_output(rejection)) as wire, gateway.scenario() as scenario:
model: Final = _deployment(scenario, wire, route, _OPUS_4_8)
assert _fallback_answer(gateway, entry, model, marker) == _payload(marker_hex)
rejected, retried = (_JSON.validate_json(request.body) for request in wire.drain())
assert _native_schema(rejected, invoke) is not None, rejected
assert _native_schema(retried, invoke) is None, retried
tools: Final = _tools(retried, invoke)
match entry:
case "converse_output_config":
assert not tools, retried
case "invoke_messages" | "invoke_messages_stream":
assert not tools, retried
prompt: Final = json.dumps(retried["messages"])
assert all(name in prompt for name in object_value(_SCHEMA["properties"])), retried
case _:
assert tools[_JSON_TOOL]["properties"] == _SCHEMA["properties"], retried
@pytest.mark.parametrize(
("backend", "rejection"),
[(_OPUS_4_8, _MAX_TOKENS_400), (_SONNET, _FABLE_REJECTION)],
ids=["other_400", "native_model"],
)
@pytest.mark.parametrize("route", ["converse", "invoke"])
def test_bedrock_400_is_not_retried_unless_it_rejects_opt_in_native_output(
gateway: Gateway, route: str, backend: str, rejection: str
) -> None:
marker, marker_hex = _marker()
with wire_server(_rejecting_native_output(rejection)) 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, thinking=_ADAPTIVE, response_format=_response_format(_SCHEMA)),
)
assert response.status_code == 400, response.text
assert rejection.split(": ", 1)[1] in response.text, response.text
_, body = _one_call(wire, marker_hex)
assert _native_schema(body, route == "invoke") == _SCHEMA, body
def _always_rejecting(message: str) -> Callable[[Request], Reply]:
def reply(_: Request) -> Reply:
return _error_reply(message)
return reply
@pytest.mark.parametrize(
("backend", "rejection"),
[(_OPUS_4_8, _MAX_TOKENS_400), (_SONNET, _FABLE_REJECTION)],
ids=["other_400", "native_model"],
)
def test_messages_400_on_bedrock_invoke_reaches_the_caller_after_one_call_unless_it_rejects_opt_in_native_output(
gateway: Gateway, backend: str, rejection: str
) -> None:
marker, marker_hex = _marker()
with wire_server(_always_rejecting(rejection)) as wire, gateway.scenario() as scenario:
model: Final = _deployment(scenario, wire, "invoke", backend)
response: Final = gateway.request(
"POST",
"/v1/messages",
_messages_body(model, marker, output_config={"format": {"type": "json_schema", "schema": _SCHEMA}}),
)
assert response.status_code == 400, response.text
assert rejection.split(": ", 1)[1] in response.text, response.text
_one_call(wire, marker_hex)
def test_a_messages_retry_without_native_output_still_omits_the_deployments_dropped_params(gateway: Gateway) -> None:
marker, marker_hex = _marker()
with wire_server(_rejecting_native_output(_FABLE_REJECTION)) as wire, gateway.scenario() as scenario:
model: Final = scenario.model(
model=f"bedrock/invoke/{_OPUS_4_8}",
api_base=wire.url,
model_info=None,
additional_drop_params=["stop_sequences"],
**_AWS,
)
request: Final = _messages_body(
model, marker, stop_sequences=["END"], output_config={"format": {"type": "json_schema", "schema": _SCHEMA}}
)
assert _messages_text(_messages_blocks(gateway.request("POST", "/v1/messages", request))) == _payload(
marker_hex
)
rejected, retried = (_JSON.validate_json(call.body) for call in wire.drain())
assert _native_schema(rejected, True) is not None, rejected
assert _native_schema(retried, True) is None, retried
assert "stop_sequences" not in rejected, rejected
assert "stop_sequences" not in retried, retried
def _carries_thinking_block(message: JsonValue) -> bool:
content: Final = object_value(message).get("content")
return isinstance(content, list) and any(object_value(block).get("type") == "thinking" for block in content)
def _replays_thinking(body: Mapping[str, JsonValue]) -> bool:
return any(_carries_thinking_block(message) for message in list_value(body["messages"]))
def _rejecting_native_output_and_foreign_signatures(signature_first: bool) -> Callable[[Request], Reply]:
def reply(request: Request) -> Reply:
body: Final = _JSON.validate_json(request.body)
checks: Final = (
(_replays_thinking(body), _SIGNATURE_400),
(_native_schema(body, True) is not None, _FABLE_REJECTION),
)
failing: Final = tuple(message for failed, message in (checks if signature_first else checks[::-1]) if failed)
return _error_reply(failing[0]) if failing else _bedrock(request)
return reply
@pytest.mark.parametrize("signature_first", [False, True], ids=["rejection_first", "signature_first"])
def test_messages_retry_without_native_output_keeps_the_thinking_signature_fix(
gateway: Gateway, signature_first: bool
) -> None:
marker, marker_hex = _marker()
history: Final[list[JsonValue]] = [
{"role": "user", "content": [{"type": "text", "text": f"first question {marker}"}]},
{
"role": "assistant",
"content": [
{"type": "thinking", "thinking": "from another deployment", "signature": _SIGNATURE},
{"type": "text", "text": "first answer"},
],
},
{"role": "user", "content": [{"type": "text", "text": "second question"}]},
]
with (
wire_server(_rejecting_native_output_and_foreign_signatures(signature_first)) as wire,
gateway.scenario() as scenario,
):
model: Final = _deployment(scenario, wire, "invoke", _OPUS_4_8)
request: Final = {
**_messages_body(
model, marker, thinking=_ADAPTIVE, output_config={"format": {"type": "json_schema", "schema": _SCHEMA}}
),
"messages": history,
}
assert _messages_text(_messages_blocks(gateway.request("POST", "/v1/messages", request))) == _payload(
marker_hex
)
sent: Final = tuple(_JSON.validate_json(call.body) for call in wire.drain())
assert len(sent) == 3, sent
assert _native_schema(sent[0], True) is not None and _replays_thinking(sent[0]), sent[0]
assert _native_schema(sent[-1], True) is None and not _replays_thinking(sent[-1]), sent[-1]
assert all(name in json.dumps(sent[-1]["messages"]) for name in object_value(_SCHEMA["properties"])), sent[-1]
@pytest.mark.parametrize("entry", ["converse_chat", "invoke_messages"])
def test_a_request_retried_without_native_output_lands_one_success_spend_row(gateway: Gateway, entry: str) -> None:
marker, marker_hex = _marker()
invoke: Final = entry.startswith("invoke")
with wire_server(_rejecting_native_output(_FABLE_REJECTION)) as wire, gateway.scenario() as scenario:
model: Final = _deployment(scenario, wire, entry.split("_", 1)[0], _OPUS_4_8)
assert _fallback_answer(gateway, entry, model, marker) == _payload(marker_hex)
sent: Final = tuple(_JSON.validate_json(call.body) for call in wire.drain())
assert len(sent) == (1 if _native_schema(sent[0], invoke) is None else 2), sent
rows: Final = held(
lambda: read_rows('SELECT status FROM "LiteLLM_SpendLogs" WHERE model_group=%s', (model,)),
lambda found: len(found) == 1,
holding=2,
seconds=70,
)
assert rows[0]["status"] == "success", rows
_CALLER_OUTPUT_CONFIG_WITH_A_SCHEMA_OBJECT: Final[dict[str, JsonValue]] = {
"textFormat": {"type": "json_schema", "structure": {"jsonSchema": {"name": "risk", "schema": {"type": "object"}}}}
}
_BEDROCK_SCHEMA_OBJECT_400: Final = "Start of structure or map found where not expected"
def _converse_json_schema_field(body: Mapping[str, JsonValue]) -> JsonValue:
text_format: Final = object_value(body.get("outputConfig") or {}).get("textFormat") or {}
structure: Final = object_value(text_format).get("structure") or {}
return object_value(object_value(structure).get("jsonSchema") or {}).get("schema")
def _deserializing_like_bedrock(request: Request) -> Reply:
schema: Final = _converse_json_schema_field(_JSON.validate_json(request.body))
if schema is None or isinstance(schema, str):
return _bedrock(request)
return Reply(status=400, body=json.dumps({"Message": _BEDROCK_SCHEMA_OBJECT_400}).encode())
def test_a_malformed_caller_output_config_on_an_opt_in_model_gets_bedrocks_own_400_after_one_call(
gateway: Gateway,
) -> None:
marker, marker_hex = _marker()
with wire_server(_deserializing_like_bedrock) as wire, gateway.scenario() as scenario:
model: Final = _deployment(scenario, wire, "converse", _OPUS_4_8)
response: Final = gateway.request(
"POST",
"/v1/chat/completions",
_chat_body(
model,
marker,
thinking=_ADAPTIVE,
response_format=_response_format(_SCHEMA),
outputConfig=_CALLER_OUTPUT_CONFIG_WITH_A_SCHEMA_OBJECT,
),
)
assert response.status_code == 400, response.text
assert _BEDROCK_SCHEMA_OBJECT_400 in response.text, response.text
_, body = _one_call(wire, marker_hex)
assert body["outputConfig"] == _CALLER_OUTPUT_CONFIG_WITH_A_SCHEMA_OBJECT, body
_CALLER_OUTPUT_CONFIG: Final[dict[str, JsonValue]] = {
"textFormat": {"type": "json_schema", "structure": {"jsonSchema": {"name": "risk", "schema": json.dumps(_SCHEMA)}}}
}
@pytest.mark.parametrize("thinking", [None, _ADAPTIVE], ids=["no_thinking", "adaptive_thinking"])
def test_a_caller_output_config_bedrock_refuses_is_sent_once_on_an_opt_in_model(
gateway: Gateway, thinking: dict[str, JsonValue] | None
) -> None:
marker, marker_hex = _marker()
with wire_server(_rejecting_native_output(_PATTERN_TOO_COMPLEX)) as wire, gateway.scenario() as scenario:
model: Final = _deployment(scenario, wire, "converse", _OPUS_4_8)
response: Final = gateway.request(
"POST",
"/v1/chat/completions",
_chat_body(
model,
marker,
response_format=_response_format(_SCHEMA),
outputConfig=_CALLER_OUTPUT_CONFIG,
**({} if thinking is None else {"thinking": thinking}),
),
)
assert response.status_code == 400, response.text
assert "pattern is too complex for structured output" in response.text, response.text
_, body = _one_call(wire, marker_hex)
assert body["outputConfig"] == _CALLER_OUTPUT_CONFIG, body

View file

@ -757,12 +757,15 @@ def _vertex_deployment(gateway: Gateway, scenario: Scenario, api_base: str, back
)
def _bedrock_invoke_deployment(scenario: Scenario, api_base: str, backend: str) -> str:
def _bedrock_invoke_deployment(
scenario: Scenario, api_base: str, backend: str, model_info: Mapping[str, JsonValue] | None = None
) -> str:
return scenario.model(
model=f"bedrock/invoke/{backend}",
api_base=api_base,
api_key=_BEDROCK_TOKEN,
aws_region_name="us-east-1",
model_info=model_info,
)
@ -1521,7 +1524,9 @@ def _json_tool_deployment(gateway: Gateway, scenario: Scenario, api_base: str, p
case "vertex_ai":
return _vertex_deployment(gateway, scenario, api_base, _TOOL_MODEL)
case "bedrock_invoke":
return _bedrock_invoke_deployment(scenario, api_base, _BEDROCK_JSON_TOOL_MODEL)
return _bedrock_invoke_deployment(
scenario, api_base, _BEDROCK_JSON_TOOL_MODEL, {"supports_bedrock_native_structured_output": False}
)
case _:
raise AssertionError(provider)

View file

@ -98,3 +98,17 @@ def test_k03_sync_stream_with_normal_frames_delivers_the_text_and_a_stop() -> No
assert consumed.text == _ANSWER, consumed
assert consumed.finish_reasons[-1] == "stop", consumed
assert len(wire.drain()) == 1
def test_k04_sync_stream_open_400_carries_the_bedrock_body_as_text() -> None:
body: Final = '{"message":"The request is invalid."}'
def reject(_: Request) -> Reply:
return Reply(status=400, body=body.encode())
with wire_server(reject) as wire:
with pytest.raises(BadRequestError) as raised:
_consume_sync_stream(wire)
assert raised.value.status_code == 400, raised.value
assert f"BedrockException - {body}" in raised.value.message, raised.value.message
assert len(wire.drain()) == 1

View file

@ -0,0 +1,336 @@
import json
import re
import uuid
from collections.abc import Mapping
from typing import Final
import boto3
import pytest
from integration._support.client import list_value, object_value, string_value
from integration._support.wire import Reply, Request, Wire, wire_server
from pydantic import JsonValue, TypeAdapter
import litellm
from litellm import CustomStreamWrapper, ModelResponse
from litellm.types.utils import Choices, StreamingChoices
_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})")
_REJECTION: Final = "The model returned the following errors: output_config.format: Extra inputs are not permitted"
_SCHEMA: Final = {
"type": "object",
"properties": {"city": {"type": "string"}},
"required": ["city"],
"additionalProperties": False,
}
_JSON: Final = TypeAdapter(dict[str, JsonValue])
_ROUTES: Final = pytest.mark.parametrize(
("route", "stream"),
[("converse", False), ("converse", True), ("invoke", False)],
ids=["converse", "converse_stream", "invoke"],
)
@pytest.fixture
def marker_hex() -> str:
return uuid.uuid4().hex
def _is_invoke(request: Request) -> bool:
return "/invoke" in request.target
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 _tool_names(body: Mapping[str, JsonValue], invoke: bool) -> frozenset[str]:
if invoke:
return frozenset(string_value(object_value(tool)["name"]) for tool in list_value(body.get("tools", [])))
tools: Final = list_value(object_value(body.get("toolConfig") or {"tools": []})["tools"])
return frozenset(string_value(object_value(object_value(tool)["toolSpec"])["name"]) for tool in tools)
def _tool_reply(invoke: bool, answer: Mapping[str, JsonValue]) -> Reply:
if invoke:
message: Final = {
"id": "msg_scripted",
"type": "message",
"role": "assistant",
"model": "claude-scripted",
"content": [{"type": "tool_use", "id": "toolu_scripted", "name": _JSON_TOOL, "input": answer}],
"stop_reason": "tool_use",
"stop_sequence": None,
"usage": {"input_tokens": 11, "output_tokens": 7},
}
return Reply(body=json.dumps(message).encode())
converse: Final = {
"output": {
"message": {
"role": "assistant",
"content": [{"toolUse": {"toolUseId": "tooluse_scripted", "name": _JSON_TOOL, "input": answer}}],
}
},
"stopReason": "tool_use",
"usage": {"inputTokens": 11, "outputTokens": 7, "totalTokens": 18},
"metrics": {"latencyMs": 1},
}
return Reply(body=json.dumps(converse).encode())
def _bedrock_without_native_output(request: Request) -> Reply:
body: Final = _JSON.validate_json(request.body)
invoke: Final = _is_invoke(request)
if _native(body, invoke):
return Reply(status=400, body=json.dumps({"message": _REJECTION}).encode())
found: Final = _MARKER.search(json.dumps(body))
assert found is not None and _JSON_TOOL in _tool_names(body, invoke), body
return _tool_reply(invoke, {"city": found.group(1)})
def _request(route: str, stream: bool, wire: Wire, marker_hex: str) -> Mapping[str, object]:
return {
"model": f"bedrock/{route}/{_OPUS_4_8}",
"messages": [{"role": "user", "content": f"Name a city. probe {marker_hex}"}],
"api_base": wire.url,
"api_key": "synthetic-bedrock-bearer",
"aws_region_name": "us-east-1",
"max_tokens": 2048,
"thinking": {"type": "adaptive"},
"response_format": {"type": "json_schema", "json_schema": {"name": "city", "schema": _SCHEMA}},
"num_retries": 0,
"stream": stream,
}
def _message_content(response: ModelResponse) -> str:
(choice,) = response.choices
assert isinstance(choice, Choices), response
return string_value(choice.message.content)
def _delta_content(chunk: ModelResponse) -> str:
(choice,) = chunk.choices
assert isinstance(choice, StreamingChoices), chunk
return choice.delta.content or ""
def _assert_one_retry_without_native_output(wire: Wire, route: str) -> None:
rejected, retried = (_JSON.validate_json(request.body) for request in wire.drain())
invoke: Final = route == "invoke"
assert _native(rejected, invoke), rejected
assert not _native(retried, invoke), retried
assert _JSON_TOOL in _tool_names(retried, invoke), retried
@_ROUTES
def test_sync_call_retries_a_native_output_rejection_once_with_the_json_tool(
route: str, stream: bool, marker_hex: str
) -> None:
with wire_server(_bedrock_without_native_output) as wire:
response: Final = litellm.completion(**_request(route, stream, wire, marker_hex))
content: Final = (
"".join(_delta_content(chunk) for chunk in response)
if isinstance(response, CustomStreamWrapper)
else _message_content(response)
)
assert json.loads(content) == {"city": marker_hex}
_assert_one_retry_without_native_output(wire, route)
@_ROUTES
async def test_async_call_retries_a_native_output_rejection_once_with_the_json_tool(
route: str, stream: bool, marker_hex: str
) -> None:
with wire_server(_bedrock_without_native_output) as wire:
response: Final = await litellm.acompletion(**_request(route, stream, wire, marker_hex))
content: Final = (
"".join([_delta_content(chunk) async for chunk in response])
if isinstance(response, CustomStreamWrapper)
else _message_content(response)
)
assert json.loads(content) == {"city": marker_hex}
_assert_one_retry_without_native_output(wire, route)
_CONSTRAINED_SCHEMA: Final = {
"type": "object",
"properties": {
"city": {"type": "string", "maxLength": 20},
"population": {"type": "integer", "minimum": 0},
"districts": {"type": "array", "items": {"type": "string"}, "minItems": 2, "uniqueItems": True},
"code": {"oneOf": [{"type": "string"}, {"type": "integer"}]},
},
"required": ["city", "population", "districts", "code"],
"additionalProperties": False,
}
def _response_format(schema: Mapping[str, object]) -> Mapping[str, object]:
return {"type": "json_schema", "json_schema": {"name": "city", "schema": schema}}
def _consume(response: ModelResponse | CustomStreamWrapper) -> None:
if isinstance(response, CustomStreamWrapper):
tuple(response)
def _sent_bodies(
route: str,
stream: bool,
marker_hex: str,
request_params: Mapping[str, object],
model_info: Mapping[str, object] | None,
) -> tuple[Mapping[str, JsonValue], ...]:
with wire_server(_bedrock_without_native_output) as wire:
request: Final = {
**_request(route, stream, wire, marker_hex),
**{"thinking": None, "response_format": None},
**request_params,
**({} if model_info is None else {"model_info": model_info}),
}
_consume(litellm.completion(**{key: value for key, value in request.items() if value is not None}))
return tuple(_JSON.validate_json(sent.body) for sent in wire.drain())
@_ROUTES
@pytest.mark.parametrize(
"request_params",
[
{"reasoning_effort": "low", "response_format": _response_format(_SCHEMA)},
{"thinking": {"type": "adaptive"}, "response_format": _response_format(_CONSTRAINED_SCHEMA)},
{
"thinking": {"type": "adaptive"},
"response_format": _response_format(_SCHEMA),
"output_config": {"format": {"type": "json_schema", "schema": _SCHEMA}},
},
{
"thinking": {"type": "adaptive"},
"output_config": {"effort": "low"},
"response_format": _response_format(_SCHEMA),
},
],
ids=[
"reasoning_effort",
"constrained_schema",
"response_format_and_output_config_format",
"adaptive_thinking_with_effort",
],
)
def test_the_retry_sends_what_a_deployment_without_native_output_sends(
route: str, stream: bool, marker_hex: str, request_params: Mapping[str, object]
) -> None:
rejected, retried = _sent_bodies(route, stream, marker_hex, request_params, None)
(opted_out,) = _sent_bodies(
route, stream, marker_hex, request_params, {"supports_bedrock_native_structured_output": False}
)
assert _native(rejected, route == "invoke"), rejected
assert retried == opted_out
@_ROUTES
def test_the_retry_keeps_the_callers_continuation_message(route: str, stream: bool, marker_hex: str) -> None:
request_params: Final = {
"thinking": {"type": "adaptive"},
"response_format": _response_format(_SCHEMA),
"messages": [
{"role": "user", "content": f"Name a city. probe {marker_hex}"},
{"role": "assistant", "content": "Which country should it be in?"},
],
"user_continue_message": {"role": "user", "content": "Any country in South America"},
}
rejected, retried = _sent_bodies(route, stream, marker_hex, request_params, None)
(opted_out,) = _sent_bodies(
route, stream, marker_hex, request_params, {"supports_bedrock_native_structured_output": False}
)
assert _native(rejected, route == "invoke"), rejected
assert retried == opted_out
_MAX_LENGTH_ONE_OF_SCHEMA: Final = {
"type": "object",
"properties": {
"city": {"type": "string", "maxLength": 20},
"code": {"oneOf": [{"type": "string"}, {"type": "integer"}]},
},
"required": ["city", "code"],
"additionalProperties": False,
}
def _messages_bedrock_without_native_output(request: Request) -> Reply:
if not _is_invoke(request) or _native(_JSON.validate_json(request.body), invoke=True):
return _bedrock_without_native_output(request)
message: Final = {
"id": "msg_scripted",
"type": "message",
"role": "assistant",
"model": "claude-scripted",
"content": [{"type": "text", "text": '{"city": "Paris"}'}],
"stop_reason": "end_turn",
"stop_sequence": None,
"usage": {"input_tokens": 11, "output_tokens": 7},
}
return Reply(body=json.dumps(message).encode())
async def _messages_bodies(
route: str, marker_hex: str, model_info: Mapping[str, object] | None
) -> tuple[Mapping[str, JsonValue], ...]:
with wire_server(_messages_bedrock_without_native_output) as wire:
await litellm.anthropic_messages(
model=f"bedrock/{route}/{_OPUS_4_8}",
messages=[{"role": "user", "content": f"Name a city. probe {marker_hex}"}],
api_base=wire.url,
api_key="synthetic-bedrock-bearer",
aws_region_name="us-east-1",
max_tokens=2048,
thinking={"type": "adaptive"},
output_config={"effort": "low", "format": {"type": "json_schema", "schema": _MAX_LENGTH_ONE_OF_SCHEMA}},
**({} if model_info is None else {"model_info": model_info}),
)
return tuple(_JSON.validate_json(sent.body) for sent in wire.drain())
@pytest.mark.parametrize("route", ["converse", "invoke"])
async def test_a_messages_retry_sends_what_a_deployment_without_native_output_sends(
route: str, marker_hex: str
) -> None:
rejected, retried = await _messages_bodies(route, marker_hex, None)
(opted_out,) = await _messages_bodies(route, marker_hex, {"supports_bedrock_native_structured_output": False})
assert _native(rejected, invoke=route == "invoke"), rejected
assert retried == opted_out
@_ROUTES
def test_the_retry_signs_with_the_credentials_of_a_deprecated_aws_bedrock_client(
route: str, stream: bool, marker_hex: str
) -> None:
aws_bedrock_client: Final = boto3.client(
"bedrock-runtime",
region_name="us-west-2",
aws_access_key_id="AKIASYNTHETICCLIENT",
aws_secret_access_key="synthetic-client-secret",
aws_session_token="synthetic-client-session",
)
with wire_server(_bedrock_without_native_output) as wire:
request: Final = {
key: value
for key, value in _request(route, stream, wire, marker_hex).items()
if key not in {"api_key", "aws_region_name"}
}
_consume(litellm.completion(**request, aws_bedrock_client=aws_bedrock_client))
sent: Final = wire.drain()
assert len(sent) == 2
for attempt in sent:
headers = {name.lower(): value for name, value in attempt.headers.items()}
assert "Credential=AKIASYNTHETICCLIENT/" in headers["authorization"]
assert "/us-west-2/bedrock/aws4_request" in headers["authorization"]
assert headers["x-amz-security-token"] == "synthetic-client-session"

View file

@ -1930,3 +1930,66 @@ class TestBedrockAnthropicCombinedRegressions:
user_msg = result["messages"][0]
has_cache_control = any(isinstance(c, dict) and "cache_control" in c for c in user_msg["content"])
assert has_cache_control
_CITY_SCHEMA: Final = {
"type": "object",
"properties": {"city": {"type": "string"}},
"required": ["city"],
"additionalProperties": False,
}
_CITY_RESPONSE_FORMAT: Final = {"type": "json_schema", "json_schema": {"name": "city", "schema": _CITY_SCHEMA}}
def _invoke_chat_params(
model: str, model_info: Mapping[str, object] | None, non_default_params: Mapping[str, object]
) -> Mapping[str, object]:
return AmazonAnthropicClaudeConfig(model_info=model_info).map_openai_params(
non_default_params=dict(non_default_params), optional_params={}, model=model, drop_params=False
)
@pytest.mark.parametrize(
"thinking_params",
[{"thinking": {"type": "adaptive"}}, {"reasoning_effort": "low"}],
ids=["adaptive", "reasoning_effort"],
)
@pytest.mark.parametrize("model", ["us.anthropic.claude-opus-4-8", "global.anthropic.claude-sonnet-5"])
def test_invoke_uses_opt_in_native_output_when_the_opt_in_drove_native_output(
local_model_cost_map, model, thinking_params
):
optional_params: Final = _invoke_chat_params(
model, None, {"response_format": _CITY_RESPONSE_FORMAT, **thinking_params}
)
assert AmazonAnthropicClaudeConfig().uses_opt_in_native_output(model, optional_params, {})
@pytest.mark.parametrize(
("model", "model_info", "request_params"),
[
("us.anthropic.claude-opus-4-8", None, {"response_format": _CITY_RESPONSE_FORMAT}),
(
"us.anthropic.claude-opus-4-8",
{"supports_bedrock_native_structured_output": False},
{"response_format": _CITY_RESPONSE_FORMAT, "thinking": {"type": "adaptive"}},
),
(
"us.anthropic.claude-sonnet-4-6",
None,
{"response_format": _CITY_RESPONSE_FORMAT, "thinking": {"type": "adaptive"}},
),
],
ids=["no_thinking", "deployment_opt_out", "native_model"],
)
def test_invoke_uses_opt_in_native_output_is_false_when_native_output_did_not_come_from_the_opt_in(
local_model_cost_map, model, model_info, request_params
):
optional_params: Final = _invoke_chat_params(model, model_info, request_params)
assert not AmazonAnthropicClaudeConfig().uses_opt_in_native_output(
model, optional_params, {"model_info": model_info}
)

View file

@ -302,3 +302,49 @@ async def test_async_converse_streaming_fails_at_the_request_timeout_not_the_ups
def test_sync_converse_streaming_fails_at_the_request_timeout_not_the_upstreams_pace() -> None:
with pytest.raises(litellm.Timeout):
litellm.completion(client=slow_upstream_sync_client(), **_converse_streaming_kwargs())
_UPSTREAM_ERROR_BODY: Final = '{"message":"The request is invalid."}'
@pytest.mark.usefixtures("local_model_cost_map")
@pytest.mark.parametrize(
"model", ["us.anthropic.claude-sonnet-4-5-20250929-v1:0", "us.anthropic.claude-opus-4-8"], ids=["sonnet", "opus"]
)
@pytest.mark.parametrize(
("upstream_status", "error_class", "status_code"),
[
(400, litellm.BadRequestError, 400),
(429, litellm.RateLimitError, 429),
(500, litellm.ServiceUnavailableError, 503),
(503, litellm.ServiceUnavailableError, 503),
],
)
def test_sync_converse_streaming_upstream_error_carries_the_upstream_body_as_text(
model: str,
upstream_status: int,
error_class: type[litellm.BadRequestError | litellm.RateLimitError | litellm.ServiceUnavailableError],
status_code: int,
) -> None:
client: Final = HTTPHandler(
client=httpx.Client(
transport=httpx.MockTransport(
lambda request: httpx.Response(upstream_status, content=_UPSTREAM_ERROR_BODY.encode())
)
)
)
with pytest.raises(error_class) as raised:
litellm.completion(
model=f"bedrock/converse/{model}",
messages=[{"role": "user", "content": "hi"}],
stream=True,
num_retries=0,
aws_access_key_id="fake",
aws_secret_access_key="fake",
aws_region_name="us-east-1",
client=client,
)
assert raised.value.status_code == status_code
assert raised.value.message == f"litellm.{error_class.__name__}: BedrockException - {_UPSTREAM_ERROR_BODY}"

View file

@ -9904,3 +9904,156 @@ def test_bedrock_nova_web_search_options_ignored_for_non_nova():
system_tool2 = result2.get("systemTool")
assert system_tool2 is not None
assert system_tool2["name"] == "nova_grounding"
def _converse_params(
model: str, model_info: Mapping[str, object] | None, non_default_params: Mapping[str, object]
) -> Mapping[str, object]:
return AmazonConverseConfig().map_openai_params(
non_default_params=dict(non_default_params),
optional_params={},
model=model,
drop_params=False,
model_info=model_info,
)
_OUTPUT_CONFIG_FORMAT: Final = {
"output_config": {"format": {"type": "json_schema", "schema": CAPITAL_RESPONSE_FORMAT["json_schema"]["schema"]}}
}
@pytest.mark.parametrize(
"request_params",
[
{"response_format": CAPITAL_RESPONSE_FORMAT, "thinking": ADAPTIVE_THINKING},
{"response_format": CAPITAL_RESPONSE_FORMAT, "reasoning_effort": "low"},
{"response_format": CAPITAL_RESPONSE_FORMAT, "thinking": {"type": "enabled", "budget_tokens": 2048}},
_OUTPUT_CONFIG_FORMAT,
],
ids=["adaptive", "reasoning_effort", "budget", "output_config_format"],
)
@pytest.mark.parametrize("model", ["us.anthropic.claude-opus-4-8", "global.anthropic.claude-sonnet-5"])
def test_uses_opt_in_native_output_when_the_opt_in_drove_native_output(local_model_cost_map, model, request_params):
optional_params: Final = _converse_params(model, None, request_params)
assert AmazonConverseConfig().uses_opt_in_native_output(model, optional_params, {"model_info": {"id": "d1"}})
@pytest.mark.parametrize(
("model", "model_info", "request_params"),
[
("us.anthropic.claude-opus-4-8", None, {"response_format": CAPITAL_RESPONSE_FORMAT}),
(
"us.anthropic.claude-opus-4-8",
{"supports_bedrock_native_structured_output": False},
{"response_format": CAPITAL_RESPONSE_FORMAT, "thinking": ADAPTIVE_THINKING},
),
(
"us.anthropic.claude-sonnet-4-6",
{"supports_bedrock_native_structured_output": True},
{"response_format": CAPITAL_RESPONSE_FORMAT, "thinking": ADAPTIVE_THINKING},
),
],
ids=["no_thinking", "deployment_opt_out", "native_model"],
)
def test_uses_opt_in_native_output_is_false_when_native_output_did_not_come_from_the_opt_in(
local_model_cost_map, model, model_info, request_params
):
optional_params: Final = _converse_params(model, model_info, request_params)
assert not AmazonConverseConfig().uses_opt_in_native_output(model, optional_params, {"model_info": model_info})
def test_uses_opt_in_native_output_reads_the_base_model_of_an_application_inference_profile(local_model_cost_map):
profile: Final = "arn:aws:bedrock:us-east-1:000000000000:application-inference-profile/profile-id"
litellm_params: Final = {"base_model": "bedrock/us.anthropic.claude-opus-4-8"}
optional_params: Final = _converse_params(
"us.anthropic.claude-opus-4-8",
None,
{"response_format": CAPITAL_RESPONSE_FORMAT, "thinking": ADAPTIVE_THINKING},
)
assert AmazonConverseConfig().uses_opt_in_native_output(profile, optional_params, litellm_params)
_CALLER_OUTPUT_CONFIG_WITH_A_SCHEMA_OBJECT: Final = {
"textFormat": {"type": "json_schema", "structure": {"jsonSchema": {"name": "capital", "schema": {"type": "object"}}}}
}
_BEDROCK_SCHEMA_TYPE_ERROR: Final = "outputConfig.textFormat.structure.jsonSchema.schema must be a string"
@pytest.mark.parametrize("model", [_OPUS, _LLAMA], ids=["opt_in", "not_opt_in"])
@pytest.mark.respx(assert_all_called=True)
def test_a_malformed_caller_output_config_gets_bedrocks_own_400(
model: str, respx_mock: respx.Router, local_model_cost_map: None
) -> None:
route: Final = respx_mock.post(url__regex=r"/converse$").mock(
return_value=httpx.Response(400, json={"message": _BEDROCK_SCHEMA_TYPE_ERROR})
)
with pytest.raises(litellm.BadRequestError, match=re.escape(_BEDROCK_SCHEMA_TYPE_ERROR)):
completion(
model=f"bedrock/converse/{model}",
messages=[{"role": "user", "content": "What is the capital of France?"}],
response_format=CAPITAL_RESPONSE_FORMAT,
outputConfig=_CALLER_OUTPUT_CONFIG_WITH_A_SCHEMA_OBJECT,
num_retries=0,
aws_access_key_id="fake",
aws_secret_access_key="fake",
aws_region_name="us-east-1",
client=HTTPHandler(),
)
assert route.call_count == 1
_CALLER_OUTPUT_CONFIG: Final = {
"textFormat": {
"type": "json_schema",
"structure": {
"jsonSchema": {"name": "capital", "schema": json.dumps(CAPITAL_RESPONSE_FORMAT["json_schema"]["schema"])}
},
}
}
_BEDROCK_SCHEMA_REFUSAL: Final = "output_config.format.schema: pattern is too complex for structured output"
@pytest.mark.parametrize(
("caller_params", "expected_calls"),
[
({"thinking": ADAPTIVE_THINKING}, 2),
({"outputConfig": _CALLER_OUTPUT_CONFIG}, 1),
({"outputConfig": _CALLER_OUTPUT_CONFIG, "thinking": ADAPTIVE_THINKING}, 1),
],
ids=["litellm_native_output", "caller_output_config", "caller_output_config_thinking"],
)
@pytest.mark.respx(assert_all_called=True)
def test_a_refused_native_schema_is_sent_again_only_when_the_retry_drops_native_output(
caller_params: dict, expected_calls: int, respx_mock: respx.Router, local_model_cost_map: None
) -> None:
route: Final = respx_mock.post(url__regex=r"/converse$").mock(
return_value=httpx.Response(400, json={"message": _BEDROCK_SCHEMA_REFUSAL})
)
with pytest.raises(litellm.BadRequestError, match=re.escape(_BEDROCK_SCHEMA_REFUSAL)):
completion(
model=f"bedrock/converse/{_OPUS}",
messages=[{"role": "user", "content": "What is the capital of France?"}],
response_format=CAPITAL_RESPONSE_FORMAT,
num_retries=0,
aws_access_key_id="fake",
aws_secret_access_key="fake",
aws_region_name="us-east-1",
client=HTTPHandler(),
**caller_params,
)
assert route.call_count == expected_calls
assert all("outputConfig" in json.loads(call.request.content) for call in route.calls[:1])
assert all("outputConfig" not in json.loads(call.request.content) for call in route.calls[1:])

View file

@ -0,0 +1,123 @@
import asyncio
from collections.abc import Coroutine
from dataclasses import dataclass
from typing import Final
import pytest
from litellm.llms.bedrock.chat.native_output_fallback import NativeOutputAttemptResult, call_with_native_output_fallback
from litellm.llms.bedrock.common_utils import BedrockError, NativeOutputFallback
from litellm.types.utils import ModelResponse
_MODEL: Final = "us.anthropic.claude-opus-4-8"
_FALLBACK: Final = NativeOutputFallback(
optional_params={"tools": "json_tool_call"}, litellm_params={"model_info": {"opted_out": True}}
)
_REJECTION: Final = BedrockError(
status_code=400,
message='{"message":"The model returned the following errors: output_config.format: Extra inputs are not permitted"}',
)
_SCHEMA_REFUSAL: Final = BedrockError(
status_code=400, message='{"message":"output_config.format.schema: pattern is too complex for structured output"}'
)
_OTHER_400: Final = BedrockError(status_code=400, message='{"message":"messages.0.content: Field required"}')
@dataclass(slots=True)
class ScriptedBedrock:
errors: tuple[BedrockError | None, ...]
asynchronous: bool
sent: tuple[NativeOutputFallback | None, ...] = ()
def _answer(self, error: BedrockError | None) -> ModelResponse:
if error is not None:
raise error
return ModelResponse(id=f"answer-{len(self.sent)}")
async def _answer_later(self, error: BedrockError | None) -> ModelResponse:
await asyncio.sleep(0)
return self._answer(error)
def __call__(self, fallback: NativeOutputFallback | None) -> NativeOutputAttemptResult:
self.sent = (*self.sent, fallback)
error: Final = self.errors[len(self.sent) - 1]
return self._answer_later(error) if self.asynchronous else self._answer(error)
def _resolve(result: NativeOutputAttemptResult) -> object:
if isinstance(result, Coroutine):
return asyncio.run(result)
return result
def _call(bedrock: ScriptedBedrock, fallback: NativeOutputFallback | None) -> object:
return _resolve(call_with_native_output_fallback(bedrock, None if fallback is None else lambda: fallback, _MODEL))
_MODES: Final = pytest.mark.parametrize("asynchronous", [False, True], ids=["sync", "async"])
@_MODES
def test_a_native_output_rejection_is_sent_again_once_with_the_fallback_params(asynchronous: bool) -> None:
bedrock: Final = ScriptedBedrock(errors=(_REJECTION, None), asynchronous=asynchronous)
response: Final = _call(bedrock, _FALLBACK)
assert isinstance(response, ModelResponse) and response.id == "answer-2"
assert bedrock.sent == (None, _FALLBACK)
@_MODES
def test_a_second_native_output_rejection_reaches_the_caller(asynchronous: bool) -> None:
bedrock: Final = ScriptedBedrock(errors=(_REJECTION, _REJECTION), asynchronous=asynchronous)
with pytest.raises(BedrockError):
_call(bedrock, _FALLBACK)
assert len(bedrock.sent) == 2
@_MODES
@pytest.mark.parametrize(
("error", "fallback"),
[(_OTHER_400, _FALLBACK), (_REJECTION, None)],
ids=["other_400", "no_fallback"],
)
def test_no_second_request_without_both_a_rejection_and_a_fallback(
asynchronous: bool, error: BedrockError, fallback: NativeOutputFallback | None
) -> None:
bedrock: Final = ScriptedBedrock(errors=(error,), asynchronous=asynchronous)
with pytest.raises(BedrockError) as raised:
_call(bedrock, fallback)
assert raised.value is error
assert bedrock.sent == (None,)
@_MODES
def test_a_rejection_reaches_the_caller_when_the_retry_would_send_native_output_again(asynchronous: bool) -> None:
bedrock: Final = ScriptedBedrock(errors=(_REJECTION,), asynchronous=asynchronous)
with pytest.raises(BedrockError) as raised:
_resolve(call_with_native_output_fallback(bedrock, lambda: None, _MODEL))
assert raised.value is _REJECTION
assert bedrock.sent == (None,)
@pytest.mark.parametrize(
("error", "advises_opting_out"),
[(_REJECTION, True), (_SCHEMA_REFUSAL, False)],
ids=["native_output_unsupported", "schema_refused"],
)
def test_a_rejection_logs_the_retry_and_advises_opting_out_only_when_native_output_is_unsupported(
error: BedrockError, advises_opting_out: bool, caplog: pytest.LogCaptureFixture
) -> None:
bedrock: Final = ScriptedBedrock(errors=(error, None), asynchronous=False)
with caplog.at_level("WARNING", logger="LiteLLM"):
_call(bedrock, _FALLBACK)
assert _MODEL in caplog.text
assert ("supports_bedrock_native_structured_output: false" in caplog.text) is advises_opting_out

View file

@ -2024,6 +2024,31 @@ class TestBedrockBatchAnthropicRowParams:
assert "response_format" not in model_input
assert ("output_config" in model_input) == (response_format["type"] == "json_schema")
@pytest.mark.parametrize(
("model", "native_output"),
[("bedrock/us.anthropic.claude-opus-4-8", False), (MODEL, True)],
ids=["opt_in_native_output_model", "native_output_model"],
)
def test_thinking_json_schema_rows_use_native_output_only_on_models_that_always_take_it(
self, local_model_cost_map, model, native_output
):
model_input: Final = self._transform(
"/v1/chat/completions",
{
"messages": [{"role": "user", "content": "Weather in Paris?"}],
"thinking": {"type": "adaptive"},
"response_format": {
"type": "json_schema",
"json_schema": {"name": "weather", "schema": self.PARAMETERS},
},
},
model=model,
)
tool_names: Final = frozenset(tool["name"] for tool in model_input.get("tools", []))
assert ("format" in model_input.get("output_config", {})) is native_output
assert (tool_names == {"json_tool_call"}) is not native_output
def test_provider_native_params_still_pass_through(self):
model_input = self._transform(
"/v1/chat/completions",

View file

@ -6,6 +6,7 @@ import os
import struct
import zlib
from collections.abc import AsyncIterator, Mapping, Sequence
from dataclasses import dataclass
from datetime import datetime
from types import SimpleNamespace
from typing import Final
@ -3783,3 +3784,162 @@ def test_bedrock_removed_tool_change_does_not_add_beta(explicit_beta: bool) -> N
assert result["messages"] == [{"role": "user", "content": "Reply with OK"}]
assert result.get("anthropic_beta", []).count(beta) == int(explicit_beta)
_NATIVE_OUTPUT_REJECTION: Final = {
"message": "The model returned the following errors: compilation exceeded limits: Schema is too complex."
}
_STALE_SIGNATURE_ERROR: Final = {"message": "messages.1.content.0: Invalid `signature` in `thinking` block"}
_OPTED_OUT: Final = {"supports_bedrock_native_structured_output": False}
_CONSTRAINED_SCHEMA: Final = {
"type": "object",
"properties": {
"summary": {"type": "string", "maxLength": 50},
"kind": {"oneOf": [{"type": "string"}, {"type": "integer"}]},
},
"required": ["summary", "kind"],
"additionalProperties": False,
}
_ANTHROPIC_MESSAGE: Final = {
"id": "msg_scripted",
"type": "message",
"role": "assistant",
"model": "claude-scripted",
"content": [{"type": "text", "text": '{"summary": "sunny", "kind": 1}'}],
"stop_reason": "end_turn",
"stop_sequence": None,
"usage": {"input_tokens": 1, "output_tokens": 1},
}
_THINKING_HISTORY: Final = (
{"role": "user", "content": "Describe the weather."},
{
"role": "assistant",
"content": [
{"type": "thinking", "thinking": "checking", "signature": "stale"},
{"type": "text", "text": "Sunny."},
],
},
{"role": "user", "content": "Summarise it as JSON."},
)
@dataclass(slots=True)
class ScriptedBedrockInvoke:
replies: tuple[tuple[int, Mapping[str, object]], ...]
bodies: tuple[dict[str, object], ...] = ()
def __call__(self, request: httpx.Request) -> httpx.Response:
self.bodies = (*self.bodies, REASONING_BODY_ADAPTER.validate_json(request.content))
status_code, body = self.replies[len(self.bodies) - 1]
return httpx.Response(status_code, json=body)
async def _send_bedrock_messages(
model: str,
model_info: Mapping[str, object] | None,
replies: Sequence[tuple[int, Mapping[str, object]]],
messages: Sequence[Mapping[str, object]] = ({"role": "user", "content": "Summarise the weather as JSON."},),
) -> tuple[ScriptedBedrockInvoke, Exception | None]:
bedrock: Final = ScriptedBedrockInvoke(replies=tuple(replies))
try:
await litellm.anthropic_messages(
model=f"bedrock/invoke/{model}",
messages=[dict(message) for message in messages],
max_tokens=4000,
system="Answer in French.",
thinking={"type": "adaptive"},
output_config={"format": {"type": "json_schema", "schema": _CONSTRAINED_SCHEMA}},
aws_access_key_id="test-access-key",
aws_secret_access_key="test-secret-key",
aws_region_name="us-east-1",
client=AsyncHTTPHandler(transport=httpx.MockTransport(bedrock)),
**({} if model_info is None else {"model_info": model_info}),
)
except Exception as error:
return bedrock, error
return bedrock, None
_OPT_IN_DEPLOYMENTS: Final = pytest.mark.parametrize(
("model", "model_info"),
[
("us.anthropic.claude-opus-4-8", None),
("us.anthropic.claude-opus-5-5", {"supports_bedrock_native_structured_output": True}),
],
ids=["mapped_opt_in", "deployment_opt_in"],
)
@pytest.mark.usefixtures("local_model_cost_map")
@_OPT_IN_DEPLOYMENTS
async def test_bedrock_messages_native_output_rejection_resends_the_body_an_opted_out_deployment_sends(
model: str, model_info: Mapping[str, object] | None
) -> None:
opt_in, opt_in_error = await _send_bedrock_messages(
model, model_info, [(400, _NATIVE_OUTPUT_REJECTION), (200, _ANTHROPIC_MESSAGE)]
)
opted_out, opted_out_error = await _send_bedrock_messages(model, _OPTED_OUT, [(200, _ANTHROPIC_MESSAGE)])
assert opt_in_error is None and opted_out_error is None
native, retried = opt_in.bodies
assert "format" in REASONING_BODY_ADAPTER.validate_python(native["output_config"])
assert opted_out.bodies == (retried,)
@pytest.mark.usefixtures("local_model_cost_map")
@pytest.mark.parametrize("model_info", [None, _OPTED_OUT], ids=["native_output", "opted_out"])
async def test_bedrock_messages_thinking_signature_400_that_persists_reaches_bedrock_twice_as_before(
model_info: Mapping[str, object] | None,
) -> None:
bedrock, error = await _send_bedrock_messages(
"us.anthropic.claude-opus-4-8", model_info, [(400, _STALE_SIGNATURE_ERROR)] * 3, _THINKING_HISTORY
)
assert isinstance(error, litellm.BadRequestError)
assert len(bedrock.bodies) == 2
@pytest.mark.usefixtures("local_model_cost_map")
@pytest.mark.parametrize(
"first_error, second_error",
[(_NATIVE_OUTPUT_REJECTION, _STALE_SIGNATURE_ERROR), (_STALE_SIGNATURE_ERROR, _NATIVE_OUTPUT_REJECTION)],
ids=["rejection_first", "signature_first"],
)
async def test_bedrock_messages_native_output_rejection_and_signature_400_each_get_one_retry_like_an_opted_out_deployment(
first_error: Mapping[str, object], second_error: Mapping[str, object]
) -> None:
opt_in, opt_in_error = await _send_bedrock_messages(
"us.anthropic.claude-opus-4-8",
None,
[(400, first_error), (400, second_error), (200, _ANTHROPIC_MESSAGE)],
_THINKING_HISTORY,
)
opted_out, opted_out_error = await _send_bedrock_messages(
"us.anthropic.claude-opus-4-8",
_OPTED_OUT,
[(400, _STALE_SIGNATURE_ERROR), (200, _ANTHROPIC_MESSAGE)],
_THINKING_HISTORY,
)
assert opt_in_error is None and opted_out_error is None
assert len(opt_in.bodies) == 3
assert opt_in.bodies[-1] == opted_out.bodies[-1]
@pytest.mark.usefixtures("local_model_cost_map")
@pytest.mark.parametrize(
("model", "model_info", "error"),
[
("us.anthropic.claude-opus-4-8", None, {"message": "messages.0.content: Field required"}),
("us.anthropic.claude-sonnet-4-6", None, _NATIVE_OUTPUT_REJECTION),
("us.anthropic.claude-opus-4-8", _OPTED_OUT, _NATIVE_OUTPUT_REJECTION),
],
ids=["other_400", "native_model", "deployment_opt_out"],
)
async def test_bedrock_messages_400_is_not_retried_unless_it_rejects_opt_in_native_output(
model: str, model_info: Mapping[str, object] | None, error: Mapping[str, object]
) -> None:
bedrock, raised = await _send_bedrock_messages(model, model_info, [(400, error), (200, _ANTHROPIC_MESSAGE)])
assert isinstance(raised, litellm.BadRequestError)
assert len(bedrock.bodies) == 1

View file

@ -1,15 +1,20 @@
import asyncio, importlib, litellm, litellm.litellm_core_utils.get_model_cost_map as bedrock_govcloud_model_cost_map, pytest
from typing import Final
from litellm.llms.bedrock.common_utils import(
AmazonBedrockGlobalConfig,
BedrockModelInfo,
bedrock_native_output_is_opt_in,
deployment_model_info,
extract_model_name_from_bedrock_arn,
get_bedrock_base_model,
get_bedrock_cross_region_inference_regions,
is_bedrock_native_output_rejection,
strip_bedrock_routing_prefix,
strip_bedrock_throughput_suffix,
without_bedrock_native_output_opt_in,
)
from collections.abc import Iterator
from litellm import completion
@ -2011,3 +2016,75 @@ def test_event_decoder_reports_only_missing_aws_dependency(missing):
assert error.value.__cause__ is failure
else:
assert error.value is failure
_FORMAT_REJECTION: Final = "The model returned the following errors: output_config.format: Extra inputs are not permitted"
_COMPILE_REJECTION: Final = "The model returned the following errors: compilation exceeded limits: Schema is too complex."
_SCHEMA_PATTERN_REJECTION: Final = (
"The model returned the following errors: output_config.format.schema: Unsupported regex feature in pattern "
"field: pattern is too complex for structured output: reduce the {n,m} upper bound, narrow the character class "
"range, or avoid nesting quantified groups"
)
@pytest.mark.parametrize(
("status_code", "error_text", "expected"),
[
(400, f'{{"message":"{_FORMAT_REJECTION}"}}', True),
(400, f'{{"message":"{_COMPILE_REJECTION}"}}', True),
(400, f'{{"message":"{_SCHEMA_PATTERN_REJECTION}"}}', True),
(400, "Malformed input request: #: extraneous key [output_config] is not permitted", False),
(400, "The model returned the following errors: messages.0.content: Field required", False),
(500, _FORMAT_REJECTION, False),
(429, _COMPILE_REJECTION, False),
],
ids=[
"format_not_permitted",
"schema_too_complex",
"schema_pattern_too_complex",
"other_key_not_permitted",
"other_400",
"500",
"429",
],
)
def test_is_bedrock_native_output_rejection_matches_only_the_400s_that_reject_native_output(
status_code, error_text, expected
):
assert is_bedrock_native_output_rejection(status_code, error_text) is expected
@pytest.mark.parametrize(
("model", "model_info", "expected"),
[
("us.anthropic.claude-opus-4-8", None, True),
("global.anthropic.claude-sonnet-5", None, True),
("us-gov.anthropic.claude-opus-4-8", None, False),
("us.anthropic.claude-opus-4-8", {"supports_bedrock_native_structured_output": False}, False),
("us.anthropic.claude-opus-5-5", {"supports_bedrock_native_structured_output": True}, True),
("us.anthropic.claude-sonnet-4-6", {"supports_bedrock_native_structured_output": True}, False),
],
ids=["opus_4_8", "sonnet_5", "govcloud_opus_4_8", "deployment_opt_out", "deployment_opt_in", "native_model"],
)
def test_bedrock_native_output_is_opt_in_only_where_the_opt_in_key_turns_native_output_on(
local_model_cost_map, model, model_info, expected
):
assert bedrock_native_output_is_opt_in(model, model_info=model_info) is expected
def test_without_bedrock_native_output_opt_in_turns_the_opt_in_off_and_keeps_the_deployment(local_model_cost_map):
litellm_params: Final = {"api_base": "https://bedrock.example", "model_info": {"id": "deployment-1"}}
turned_off: Final = without_bedrock_native_output_opt_in(litellm_params)
assert not bedrock_native_output_is_opt_in(
"us.anthropic.claude-opus-4-8", model_info=deployment_model_info(turned_off)
)
assert turned_off["api_base"] == "https://bedrock.example"
assert deployment_model_info(turned_off)["id"] == "deployment-1"
assert litellm_params["model_info"] == {"id": "deployment-1"}

View file

@ -179,6 +179,7 @@ def _build_dispatch_context() -> CompletionDispatchContext:
model="gpt-4o",
model_response=None, # type: ignore[arg-type]
optional_params={},
optional_params_for_model_info=lambda model_info: {},
organization=None,
provider_config=None,
request_params={},