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

litellm/llms/base_llm/anthropic_messages/transformation.py

@@ -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
 
 import httpx
@@ -167,7 +167,9 @@ class BaseAnthropicMessagesConfig(ABC):
 
         return e.response.status_code == 400 and is_anthropic_invalid_thinking_block_error(e.response.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.
         """
@@ -179,3 +181,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

litellm/llms/bedrock/chat/converse_handler.py

@@ -37,13 +37,21 @@ 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,
-    )
+    try:
+        response: Final = client.post(
+            api_base,
+            headers=headers,
+            data=data,
+            stream=not fake_stream,
+            logging_obj=logging_obj,
+        )
+    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(

litellm/llms/bedrock/chat/converse_transformation.py

@@ -12,6 +12,7 @@ from itertools import chain
 from typing import TYPE_CHECKING, Final, Literal, cast, overload
 
 import httpx
+from pydantic import TypeAdapter, ValidationError
 
 import litellm
 from litellm._logging import verbose_logger
@@ -54,7 +55,9 @@ from litellm.llms.bedrock.common_utils import (
     bedrock_chat_uses_native_structured_output,
     bedrock_cleaned_output_schema,
     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,
@@ -135,6 +138,18 @@ UNSUPPORTED_BEDROCK_CONVERSE_BETA_PATTERNS: Final = [
 ]
 
 
+_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
@@ -1323,6 +1338,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.

litellm/llms/bedrock/chat/invoke_transformations/anthropic_claude3_transformation.py

@@ -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,
 )
@@ -94,6 +95,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,

litellm/llms/bedrock/chat/native_output_fallback.py

@@ -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

litellm/llms/bedrock/common_utils.py

@@ -10,6 +10,8 @@ import json
 import os
 import re
 from collections.abc import Mapping, Sequence
+from dataclasses import dataclass
+from types import MappingProxyType
 from typing import TYPE_CHECKING, Any, Final, Literal, TypedDict
 
 if TYPE_CHECKING:
@@ -169,7 +171,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.
 
@@ -190,8 +192,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.
@@ -277,6 +279,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
 
 
@@ -343,7 +416,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
 

litellm/llms/bedrock/files/transformation.py

@@ -64,8 +64,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"
@@ -899,10 +901,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
@@ -917,7 +921,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={},
             )
 

litellm/llms/bedrock/messages/invoke_transformations/anthropic_claude3_transformation.py

@@ -36,11 +36,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,
@@ -103,6 +107,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,

litellm/llms/custom_httpx/llm_http_handler.py

@@ -1,7 +1,8 @@
 import asyncio
+import copy
 import json
 import ssl
-from collections.abc import AsyncGenerator, AsyncIterator, Coroutine, Iterator, Mapping, Sequence
+from collections.abc import AsyncGenerator, AsyncIterator, Callable, Coroutine, Iterator, Mapping, Sequence
 from contextlib import asynccontextmanager
 from functools import lru_cache, partial
 from types import MappingProxyType, ModuleType
@@ -376,6 +377,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:
@@ -1992,11 +1996,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,
@@ -2009,40 +2015,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
                 )
-                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,
@@ -2163,6 +2183,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,
@@ -2236,6 +2274,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

litellm/main.py

@@ -112,7 +112,12 @@ from litellm.llms.base_llm import BaseConfig, BaseImageGenerationConfig
 from litellm.llms.base_llm.base_model_iterator import (
     convert_model_response_to_streaming,
 )
-from litellm.llms.bedrock.common_utils import BedrockModelInfo
+from litellm.llms.bedrock.common_utils import (
+    BedrockModelInfo,
+    NativeOutputFallback,
+    deployment_model_info,
+    without_bedrock_native_output_opt_in,
+)
 from litellm.llms.cohere.common_utils import CohereModelInfo
 from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler, http2_enabled
 from litellm.llms.openai.chat.gpt_5_transformation import OpenAIGPT5Config
@@ -205,6 +210,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
@@ -4152,6 +4160,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
@@ -4218,22 +4263,36 @@ def _complete_bedrock(ctx: _CompletionDispatchContext) -> _CompletionDispatchRes
         )
     elif bedrock_route == "converse":
         model = model.replace("converse/", "")
-        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/", "")
@@ -4255,22 +4314,36 @@ def _complete_bedrock(ctx: _CompletionDispatchContext) -> _CompletionDispatchRes
             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
@@ -5607,10 +5680,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,
@@ -5823,6 +5900,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,
             shared_session=shared_session,

litellm/types/completion.py

@@ -227,6 +227,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
     shared_session: ClientSession | None

tests/integration/providers/test_bedrock_batch_files_wire.py

@@ -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

tests/integration/providers/test_bedrock_native_structured_output_wire.py

@@ -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

tests/integration/providers/test_vertex_anthropic_structured_output_wire.py

@@ -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)
 

tests/integration/sdk/test_bedrock_converse_stream_sync_decoder_wire.py

@@ -0,0 +1,114 @@
+import re
+from collections.abc import Callable, Mapping
+from dataclasses import dataclass
+from typing import Final
+from urllib.parse import unquote
+
+import pytest
+from integration._support.upstream import _aws_event_frame
+from integration._support.wire import Reply, Request, Wire, wire_server
+from pydantic import JsonValue
+
+import litellm
+from litellm.exceptions import BadGatewayError, BadRequestError, MidStreamFallbackError
+from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper
+
+_MODEL_ID: Final = "global.moonshotai.kimi-k3"
+_CONVERSE_MODEL: Final = f"bedrock/converse/{_MODEL_ID}"
+_STREAM_TARGET: Final = f"/model/{_MODEL_ID}/converse-stream"
+_EVENT_STREAM: Final = "application/vnd.amazon.eventstream"
+_ANSWER: Final = "bedrock sync decoder control"
+_REJECTION: Final = "structured output schema uses unsupported regex negative look-ahead"
+_UNKNOWN_TYPE: Final = "somethingBedrockAddedLater"
+_USAGE: Final[dict[str, JsonValue]] = {"usage": {"inputTokens": 11, "outputTokens": 4, "totalTokens": 15}}
+
+
+def _frame(event_type: str, payload: Mapping[str, JsonValue]) -> bytes:
+    return _aws_event_frame(event_type, payload, "sc", "u")
+
+
+_NORMAL: Final = b"".join(
+    (
+        _frame("messageStart", {"role": "assistant"}),
+        _frame("contentBlockDelta", {"delta": {"text": _ANSWER}, "contentBlockIndex": 0}),
+        _frame("contentBlockStop", {"contentBlockIndex": 0}),
+        _frame("messageStop", {"stopReason": "end_turn"}),
+        _frame("metadata", _USAGE),
+    )
+)
+_VALIDATION_FRAME: Final = _frame("validationException", {"message": _REJECTION})
+_UNKNOWN_FRAME: Final = _frame(_UNKNOWN_TYPE, {"future": True})
+
+
+@dataclass(frozen=True, slots=True)
+class _Consumed:
+    text: str
+    finish_reasons: tuple[str | None, ...]
+
+
+def _peer(frames: bytes) -> Callable[[Request], Reply]:
+    def respond(request: Request) -> Reply:
+        assert unquote(request.target) == _STREAM_TARGET, request.target
+        return Reply(body=frames, content_type=_EVENT_STREAM)
+
+    return respond
+
+
+def _consume_sync_stream(wire: Wire) -> _Consumed:
+    response: Final = litellm.completion(
+        model=_CONVERSE_MODEL,
+        messages=[{"role": "user", "content": "What does the sync decoder do with this stream?"}],
+        max_tokens=16,
+        stream=True,
+        api_base=wire.url,
+        aws_access_key_id="AKIASCRIPTEDPROVIDER",
+        aws_secret_access_key="scripted-secret",
+        aws_region_name="us-east-1",
+        num_retries=0,
+    )
+    assert isinstance(response, CustomStreamWrapper), type(response)
+    chunks: Final = tuple(response)
+    return _Consumed(
+        text="".join(str(chunk.choices[0].delta.content or "") for chunk in chunks),
+        finish_reasons=tuple(chunk.choices[0].finish_reason for chunk in chunks),
+    )
+
+
+def test_k01_sync_stream_with_a_validation_exception_frame_first_raises_a_400() -> None:
+    with wire_server(_peer(_VALIDATION_FRAME)) as wire:
+        with pytest.raises(BadRequestError, match=re.escape(_REJECTION)) as raised:
+            _consume_sync_stream(wire)
+        assert raised.value.status_code == 400, raised.value
+        assert len(wire.drain()) == 1
+
+
+def test_k02_sync_stream_whose_only_frame_has_an_unknown_event_type_raises_a_502_naming_it() -> None:
+    with wire_server(_peer(_UNKNOWN_FRAME)) as wire:
+        with pytest.raises(MidStreamFallbackError, match=re.escape(_UNKNOWN_TYPE)) as raised:
+            _consume_sync_stream(wire)
+        assert raised.value.status_code == 502, raised.value
+        assert isinstance(raised.value.original_exception, BadGatewayError), raised.value.original_exception
+        assert "none of its 1 events carried a known event type" in str(raised.value), raised.value
+        assert len(wire.drain()) == 1
+
+
+def test_k03_sync_stream_with_normal_frames_delivers_the_text_and_a_stop() -> None:
+    with wire_server(_peer(_NORMAL)) as wire:
+        consumed: Final = _consume_sync_stream(wire)
+        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

tests/integration/sdk/test_bedrock_native_output_fallback_sdk.py

@@ -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"

tests/unit/llms/bedrock/chat/invoke_transformations/test_bedrock_chat_invoke_transformations_anthropic_claude3_transformation.py

@@ -1264,3 +1264,66 @@ def test_chat_flagged_model_replays_a_byte_identical_prefix_around_a_mid_convers
     _assert_prefix_stable(requests)
     assert [m["role"] for m in requests[1]["messages"]] == ["user", "assistant", "user", "system"]
     assert [m["role"] for m in requests[2]["messages"]] == ["user", "assistant", "user", "system", "assistant", "user"]
+
+
+_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}
+    )
+

tests/unit/llms/bedrock/chat/test_bedrock_converse_handler.py

@@ -273,3 +273,50 @@ def test_session_tags_sign_the_request_and_stay_out_of_the_body(monkeypatch):
     sent = client.post.call_args.kwargs
     assert "Credential=ASIACONVERSETAGGED/" in sent["headers"]["Authorization"]
     assert "aws_session_tags" not in sent["data"]
+
+
+_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}"
+

tests/unit/llms/bedrock/chat/test_converse_transformation.py

@@ -8340,3 +8340,156 @@ def test_supports_sampling_params_prefixed_and_anthropic_fallback(monkeypatch: p
     )
     assert AmazonConverseConfig._supports_sampling_params("custom-test-reasoning-model") is False
     assert AmazonConverseConfig._supports_sampling_params("anthropic.claude-custom-unregistered") is True
+
+
+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:])
+

tests/unit/llms/bedrock/chat/test_native_output_fallback.py

@@ -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

tests/unit/llms/bedrock/files/test_bedrock_files_transformation.py

@@ -2032,6 +2032,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",

tests/unit/llms/bedrock/messages/invoke_transformations/test_anthropic_claude3_transformation.py

@@ -5,6 +5,8 @@ import json
 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 collections.abc import AsyncIterator, Mapping, Sequence
@@ -3609,3 +3611,173 @@ async def test_get_async_streaming_response_iterator_yields_small_frame_before_u
     remaining: Final = tuple([chunk async for chunk in iterator])
     assert any(chunk.startswith(b"event: message_stop\n") for chunk in remaining), remaining
     await iterator.aclose()
+
+
+_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
+

tests/unit/llms/bedrock/test_bedrock_common_utils.py

@@ -1,8 +1,26 @@
 
-import pytest
+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 BedrockModelInfo
+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
+from litellm.llms.bedrock.count_tokens.bedrock_token_counter import BedrockTokenCounter
+from tests._vcr_conftest_common import install_live_call_probe, record_vcr_outcome
+from unittest.mock import Mock, patch
 
 # --------------------------------------------------------------------------- #
 # get_bedrock_response_stream_shape lazy-load tests                           #
@@ -987,3 +1005,75 @@ def test_unmapped_openai_family_model_routes_to_converse():
     assert BedrockModelInfo.get_bedrock_route(unmapped) == "converse"
     imported: Final = "bedrock/openai/arn:aws:bedrock:us-east-1:123456789012:imported-model/abc123"
     assert BedrockModelInfo.get_bedrock_route(imported) == "openai"
+
+
+_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"}
+

tests/unit/types/test_completion.py

@@ -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,
         shared_session=None,