fix(anthropic): strip beta-gated body fields when a leg drops the anthropic-beta

When a provider leg drops an anthropic-beta value (the chat handler's allowlist
filter, /v1/messages' filtered headers, Bedrock Converse's additionalModelRequestFields),
also drop the body fields that beta gates (thinking.block_binding, thinking.display=updates,
mid-conversation system clear_at, output_config, tool_addition/tool_removal blocks) so the
request degrades on that leg instead of failing with a provider 400. Copy-on-write so a
fallback leg still sends the full request, and each strip is logged at WARNING.

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

litellm/anthropic_beta_gated_fields.py

@@ -0,0 +1,274 @@
+"""Body fields that an ``anthropic-beta`` value gates.
+
+``anthropic_beta_headers_config.json`` says which beta each provider leg
+carries. When a leg drops a beta, the request fields that beta unlocks
+would make the provider 400, so the same request degrades on that leg by
+dropping those fields too. Every helper here is copy-on-write: the caller's
+containers are returned untouched when nothing is stripped, and a shallow
+copy is returned otherwise, so a fallback leg still sends the full request.
+"""
+
+from collections.abc import Mapping, Sequence, Set
+from dataclasses import dataclass
+from types import MappingProxyType
+from typing import Final
+
+from pydantic import TypeAdapter, ValidationError
+
+from litellm._logging import verbose_logger
+from litellm.types.llms.anthropic import (
+    ANTHROPIC_BETA_HEADER_VALUES,
+    ANTHROPIC_MID_CONVERSATION_OUTPUT_CONFIG_BETA_HEADER,
+    ANTHROPIC_MID_CONVERSATION_TOOL_CHANGES_BETA_HEADER,
+    ANTHROPIC_THINKING_DISPLAY_UPDATES_BETA_HEADER,
+)
+
+ANTHROPIC_THINKING_BINDING_CONTROLS_BETA_HEADER: Final = "thinking-binding-controls-2026-08-01"
+ANTHROPIC_MID_CONVERSATION_SYSTEM_CLEAR_AT_BETA_HEADER: Final = "mid-conversation-system-clear-at-2026-08-21"
+
+THINKING_BLOCK_BINDING: Final = "thinking.block_binding"
+THINKING_DISPLAY_UPDATES: Final = "thinking.display=updates"
+SYSTEM_CLEAR_AT: Final = "messages[].system.clear_at"
+SYSTEM_OUTPUT_CONFIG: Final = "messages[].system.output_config"
+SYSTEM_TOOL_CHANGES: Final = "messages[].system.content[].tool_addition|tool_removal"
+SYSTEM_MESSAGE_EMPTIED: Final = "messages[].system (emptied by the strip)"
+
+_TOOL_CHANGE_BLOCK_TYPES: Final = frozenset({"tool_addition", "tool_removal"})
+
+BETA_GATED_FIELDS: Final[Mapping[str, tuple[str, ...]]] = MappingProxyType(
+    {
+        ANTHROPIC_THINKING_BINDING_CONTROLS_BETA_HEADER: (THINKING_BLOCK_BINDING,),
+        ANTHROPIC_THINKING_DISPLAY_UPDATES_BETA_HEADER: (THINKING_DISPLAY_UPDATES,),
+        ANTHROPIC_MID_CONVERSATION_SYSTEM_CLEAR_AT_BETA_HEADER: (SYSTEM_CLEAR_AT,),
+        ANTHROPIC_MID_CONVERSATION_OUTPUT_CONFIG_BETA_HEADER: (SYSTEM_OUTPUT_CONFIG,),
+        ANTHROPIC_BETA_HEADER_VALUES.PER_TURN_CONTROL_2026_07_01.value: (SYSTEM_OUTPUT_CONFIG,),
+        ANTHROPIC_MID_CONVERSATION_TOOL_CHANGES_BETA_HEADER: (SYSTEM_TOOL_CHANGES,),
+    }
+)
+
+
+def _betas_unlocking(field: str) -> frozenset[str]:
+    return frozenset(beta for beta, fields in BETA_GATED_FIELDS.items() if field in fields)
+
+
+_FIELD_BETAS: Final[Mapping[str, frozenset[str]]] = MappingProxyType(
+    {
+        field: _betas_unlocking(field) for fields in BETA_GATED_FIELDS.values() for field in fields
+    }  # comprehension-ok: flattening the beta-to-fields map into a field-to-betas index
+)
+
+
+def field_is_gated(field: str, active_betas: Set[str]) -> bool:
+    """True when no active beta unlocks ``field``, so the field must be stripped."""
+    return _FIELD_BETAS[field].isdisjoint(active_betas)
+
+
+_MAPPING_ADAPTER: Final = TypeAdapter(Mapping[str, object])
+_SEQUENCE_ADAPTER: Final = TypeAdapter(tuple[object, ...])
+_HEADERS_ADAPTER: Final = TypeAdapter(Mapping[str, str])
+
+
+def _as_mapping(value: object) -> Mapping[str, object] | None:
+    try:
+        return _MAPPING_ADAPTER.validate_python(value, strict=True)
+    except ValidationError:
+        return None
+
+
+def _as_sequence(value: object) -> tuple[object, ...] | None:
+    try:
+        return _SEQUENCE_ADAPTER.validate_python(value)
+    except ValidationError:
+        return None
+
+
+def _is_tool_change_block(block: object) -> bool:
+    mapping: Final = _as_mapping(block)
+    return mapping is not None and mapping.get("type") in _TOOL_CHANGE_BLOCK_TYPES
+
+
+def _header_beta_values(headers: Mapping[str, str]) -> frozenset[str]:
+    joined: Final = ",".join(value for key, value in headers.items() if key.lower() == "anthropic-beta")
+    return frozenset(piece.strip() for piece in joined.split(",") if piece.strip())
+
+
+def _validated_headers(headers: object) -> Mapping[str, str]:
+    try:
+        return _HEADERS_ADAPTER.validate_python(headers)
+    except ValidationError:
+        return MappingProxyType({})
+
+
+def active_beta_values(headers: object, body_betas: object = ()) -> frozenset[str]:
+    """The ``anthropic-beta`` values a request will send: the comma-joined header (key matched
+    case-insensitively) plus the body's ``anthropic_beta`` list on legs that carry betas in the body."""
+    body_values: Final = _as_sequence(body_betas) or ()
+    body_betas_as_strings: Final = frozenset(beta for beta in body_values if isinstance(beta, str))
+    return _header_beta_values(_validated_headers(headers)) | body_betas_as_strings
+
+
+@dataclass(frozen=True, slots=True)
+class StrippedThinking:
+    thinking: object
+    stripped: tuple[str, ...]
+
+
+@dataclass(frozen=True, slots=True)
+class StrippedMessages:
+    messages: Sequence[object]
+    stripped: tuple[str, ...]
+
+
+@dataclass(frozen=True, slots=True)
+class StrippedRequest:
+    request: Mapping[str, object]
+    stripped: tuple[str, ...]
+
+
+def strip_beta_gated_thinking(thinking: object, active_betas: Set[str]) -> StrippedThinking:
+    """Drop ``thinking`` sub-fields whose beta is absent. Returns the same object when nothing is gated."""
+    mapping: Final = _as_mapping(thinking)
+    if mapping is None:
+        return StrippedThinking(thinking=thinking, stripped=())
+    binding_gated: Final = "block_binding" in mapping and field_is_gated(THINKING_BLOCK_BINDING, active_betas)
+    display_gated: Final = mapping.get("display") == "updates" and field_is_gated(
+        THINKING_DISPLAY_UPDATES, active_betas
+    )
+    if not binding_gated and not display_gated:
+        return StrippedThinking(thinking=thinking, stripped=())
+    gated_keys: Final = frozenset(
+        {*(("block_binding",) if binding_gated else ()), *(("display",) if display_gated else ())}
+    )
+    stripped: Final = (
+        *((THINKING_BLOCK_BINDING,) if binding_gated else ()),
+        *((THINKING_DISPLAY_UPDATES,) if display_gated else ()),
+    )
+    return StrippedThinking(
+        thinking={key: value for key, value in mapping.items() if key not in gated_keys},
+        stripped=stripped,
+    )
+
+
+def _strip_beta_gated_system_message(message: Mapping[str, object], active_betas: Set[str]) -> StrippedMessages:
+    """One mid-conversation ``role: system`` message. ``messages`` holds the rewritten message, or nothing
+    when the strip left it with no content, since an empty system turn is itself a 400."""
+    clear_at_gated: Final = "clear_at" in message and field_is_gated(SYSTEM_CLEAR_AT, active_betas)
+    output_config_gated: Final = "output_config" in message and field_is_gated(SYSTEM_OUTPUT_CONFIG, active_betas)
+    content: Final = message.get("content")
+    blocks: Final = _as_sequence(content) if not isinstance(content, str) else None
+    tool_changes_gated: Final = (
+        blocks is not None
+        and any(_is_tool_change_block(block) for block in blocks)
+        and field_is_gated(SYSTEM_TOOL_CHANGES, active_betas)
+    )
+    if not clear_at_gated and not output_config_gated and not tool_changes_gated:
+        return StrippedMessages(messages=(message,), stripped=())
+    kept_content: Final[object] = (
+        [block for block in blocks if not _is_tool_change_block(block)]
+        if tool_changes_gated and blocks is not None
+        else content
+    )
+    gated_keys: Final = frozenset(
+        {*(("clear_at",) if clear_at_gated else ()), *(("output_config",) if output_config_gated else ())}
+    )
+    rewritten: Final = {
+        key: (kept_content if key == "content" else value) for key, value in message.items() if key not in gated_keys
+    }
+    stripped: Final = (
+        *((SYSTEM_CLEAR_AT,) if clear_at_gated else ()),
+        *((SYSTEM_OUTPUT_CONFIG,) if output_config_gated else ()),
+        *((SYSTEM_TOOL_CHANGES,) if tool_changes_gated else ()),
+    )
+    if kept_content in (None, "", []):
+        return StrippedMessages(messages=(), stripped=(*stripped, SYSTEM_MESSAGE_EMPTIED))
+    return StrippedMessages(messages=(rewritten,), stripped=stripped)
+
+
+def _strip_beta_gated_message(message: object, active_betas: Set[str]) -> StrippedMessages:
+    mapping: Final = _as_mapping(message)
+    if mapping is None or mapping.get("role") != "system":
+        return StrippedMessages(messages=(message,), stripped=())
+    return _strip_beta_gated_system_message(mapping, active_betas)
+
+
+def strip_beta_gated_system_messages(messages: Sequence[object], active_betas: Set[str]) -> StrippedMessages:
+    """Drop the gated fields of every mid-conversation ``role: system`` message. Returns the same
+    sequence when nothing is gated."""
+    per_message: Final = tuple(_strip_beta_gated_message(message, active_betas) for message in messages)
+    stripped: Final = tuple(
+        name for result in per_message for name in result.stripped
+    )  # comprehension-ok: flattening per-message strip reports
+    if not stripped:
+        return StrippedMessages(messages=messages, stripped=())
+    return StrippedMessages(
+        messages=[
+            message for result in per_message for message in result.messages
+        ],  # comprehension-ok: flattening per-message results
+        stripped=stripped,
+    )
+
+
+def strip_beta_gated_request_fields(request: Mapping[str, object], active_betas: Set[str]) -> StrippedRequest:
+    """Apply both strips to an Anthropic ``/v1/messages`` shaped body. Returns the same mapping when
+    nothing is gated, otherwise a shallow copy with ``thinking`` and ``messages`` replaced."""
+    thinking: Final = strip_beta_gated_thinking(request.get("thinking"), active_betas)
+    messages: Final = request.get("messages")
+    message_sequence: Final = _as_sequence(messages) if not isinstance(messages, str) else None
+    system_messages: Final = (
+        strip_beta_gated_system_messages(message_sequence, active_betas)
+        if message_sequence is not None
+        else StrippedMessages(messages=(), stripped=())
+    )
+    stripped: Final = (*thinking.stripped, *system_messages.stripped)
+    if not stripped:
+        return StrippedRequest(request=request, stripped=())
+    rewritten: Final = {
+        **request,
+        **({"thinking": thinking.thinking} if thinking.stripped else {}),
+        **({"messages": system_messages.messages} if system_messages.stripped else {}),
+    }
+    return StrippedRequest(request=rewritten, stripped=stripped)
+
+
+@dataclass(frozen=True, slots=True)
+class StrippedNativeRequest:
+    thinking: object
+    thinking_stripped: bool
+    messages: Sequence[object]
+    messages_stripped: bool
+    stripped: tuple[str, ...]
+
+
+def strip_beta_gated_native_request(
+    headers: object, optional_params: object, messages: object
+) -> StrippedNativeRequest:
+    """The ``/v1/messages`` seam, fed the final (already filtered) headers plus the body parts it may rewrite."""
+    active_betas: Final = active_beta_values(headers)
+    params: Final = _as_mapping(optional_params) or MappingProxyType({})
+    thinking: Final = strip_beta_gated_thinking(params.get("thinking"), active_betas)
+    message_sequence: Final = _as_sequence(messages) if not isinstance(messages, str) else None
+    system_messages: Final = (
+        strip_beta_gated_system_messages(message_sequence, active_betas)
+        if message_sequence is not None
+        else StrippedMessages(messages=(), stripped=())
+    )
+    return StrippedNativeRequest(
+        thinking=thinking.thinking,
+        thinking_stripped=bool(thinking.stripped),
+        messages=system_messages.messages,
+        messages_stripped=bool(system_messages.stripped),
+        stripped=(*thinking.stripped, *system_messages.stripped),
+    )
+
+
+def log_beta_gated_strips(stripped: Sequence[str], model: str | None, provider: str) -> None:
+    if not stripped:
+        return
+    verbose_logger.warning(
+        "Stripped %s from the request for model=%s provider=%s because this leg does not carry the "
+        "anthropic-beta that gates them (see anthropic_beta_headers_config.json). The request is sent "
+        "without those fields instead of failing with a provider 400.",
+        sorted(set(stripped)),
+        model,
+        provider,
+    )

litellm/anthropic_beta_headers_manager.py

@@ -397,7 +397,7 @@ def update_request_with_filtered_beta(
     headers: dict,
     request_data: dict,
     provider: str,
-) -> tuple[dict, dict]:
+) -> tuple[dict[str, str], dict[str, object]]:
     """
     Update both headers and request body beta fields based on provider support.
     Modifies both dicts in place and returns them.

litellm/llms/anthropic/chat/handler.py

@@ -17,6 +17,11 @@ import litellm
 import litellm.litellm_core_utils
 import litellm.types
 import litellm.types.utils
+from litellm.anthropic_beta_gated_fields import (
+    active_beta_values,
+    log_beta_gated_strips,
+    strip_beta_gated_request_fields,
+)
 from litellm.anthropic_beta_headers_manager import (
     update_request_with_filtered_beta,
 )
@@ -505,11 +510,17 @@ class AnthropicChatCompletion(BaseLLM):
             The pair stays mutable because the streaming path rewrites it in
             place (`data["stream"] = True`) before sending.
             """
-            request_headers, data = update_request_with_filtered_beta(
+            request_headers, filtered_data = update_request_with_filtered_beta(
                 headers=headers,
                 request_data=request_data,
                 provider=custom_llm_provider,
             )
+            gated: Final = strip_beta_gated_request_fields(
+                filtered_data,
+                active_beta_values(request_headers, filtered_data.get("anthropic_beta") or ()),
+            )
+            log_beta_gated_strips(gated.stripped, model=model, provider=custom_llm_provider)
+            data: Final = dict(gated.request)
             logging_obj.pre_call(
                 input=messages,
                 api_key=api_key,

litellm/llms/anthropic/messages/native/transformation.py

@@ -2,9 +2,11 @@ from collections.abc import AsyncIterator, Mapping, Sequence
 from typing import ClassVar, Final
 
 import httpx
+from pydantic import TypeAdapter
 
 import litellm
 from litellm._logging import verbose_logger
+from litellm.anthropic_beta_gated_fields import log_beta_gated_strips, strip_beta_gated_native_request
 from litellm.anthropic_beta_headers_manager import get_unsupported_headers, is_beta_header_supported_by_bundled_config
 from litellm.exceptions import AuthenticationError
 from litellm.litellm_core_utils.core_helpers import drops_unsupported_params
@@ -81,6 +83,9 @@ def _ensure_stop_sequence(response: AnthropicMessagesResponse) -> AnthropicMessa
     return response
 
 
+_MESSAGES_ADAPTER: Final = TypeAdapter(list[dict[str, object]])
+
+
 class AnthropicMessagesConfig(BaseAnthropicMessagesConfig):
     _workload_identity_eligible: ClassVar[bool] = True
 
@@ -705,15 +710,22 @@ class AnthropicMessagesConfig(BaseAnthropicMessagesConfig):
         if not _has_advisor:
             messages = strip_advisor_blocks_from_messages(messages)
 
+        gated: Final = strip_beta_gated_native_request(headers, anthropic_messages_optional_request_params, messages)  # pyright: ignore[reportUnknownArgumentType]  # the base signature types these as bare dicts
+        log_beta_gated_strips(gated.stripped, model=model, provider=self._resolved_provider)
         anthropic_messages_request: Final[AnthropicMessagesRequest] = AnthropicMessagesRequest(
             messages=strip_encrypted_reasoning_blocks_from_anthropic_messages(
-                drop_non_string_tool_result_texts(messages, getattr(litellm_params, "drop_params", None))
+                drop_non_string_tool_result_texts(
+                    _MESSAGES_ADAPTER.validate_python(gated.messages) if gated.messages_stripped else messages,
+                    getattr(litellm_params, "drop_params", None),
+                )
             ),
             max_tokens=max_tokens,
             model=model,
             **anthropic_messages_optional_request_params,
         )
-        return dict(anthropic_messages_request)
+        if gated.thinking_stripped:
+            return {**dict(anthropic_messages_request), "thinking": gated.thinking}
+        return dict(anthropic_messages_request)  # pyright: ignore[reportUnknownVariableType]  # built from the base signature's bare dict
 
     def transform_anthropic_messages_response(
         self,

litellm/llms/bedrock/chat/converse_transformation.py

@@ -17,6 +17,7 @@ from typing_extensions import ReadOnly, TypedDict
 
 import litellm
 from litellm._logging import verbose_logger
+from litellm.anthropic_beta_gated_fields import log_beta_gated_strips, strip_beta_gated_thinking
 from litellm.constants import (
     BEDROCK_MIN_THINKING_BUDGET_TOKENS,
     RESPONSE_FORMAT_TOOL_NAME,
@@ -2031,6 +2032,11 @@ class AmazonConverseConfig(BaseConfig):
         bedrock_tools, anthropic_beta_list = self._process_tools_and_beta(
             original_tools, model, headers, additional_request_params, litellm_params, deployment_model
         )
+        requested_thinking: Final = additional_request_params.get("thinking")  # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType]  # _prepare_request_params hands back a bare dict
+        gated_thinking: Final = strip_beta_gated_thinking(requested_thinking, frozenset(anthropic_beta_list))  # pyright: ignore[reportUnknownArgumentType]  # _process_tools_and_beta hands back a bare list
+        if gated_thinking.stripped:
+            log_beta_gated_strips(gated_thinking.stripped, model=model, provider="bedrock_converse")
+            additional_request_params["thinking"] = gated_thinking.thinking
 
         # Append cachePoint to tools if cache_control_injection_points has tool_config
         cache_injection_points: Final = additional_request_params.pop("cache_control_injection_points", None)

tests/unit/llms/anthropic/chat/test_anthropic_chat_handler.py

@@ -3607,3 +3607,51 @@ def test_streamed_end_turn_carries_no_stop_details():
     provider_fields: Final = assembled.choices[0].message.provider_specific_fields or {}
     assert "stop_details" not in provider_fields
     assert "stop_reason" not in provider_fields
+
+
+def _anthropic_chat_body_for_bound_thinking(extra_headers: dict | None) -> dict:
+    captured_requests: list[httpx.Request] = []
+
+    def respond(request: httpx.Request) -> httpx.Response:
+        captured_requests.append(request)
+        return httpx.Response(
+            200,
+            json={
+                "id": "msg_bound",
+                "type": "message",
+                "role": "assistant",
+                "model": "claude-opus-4-8",
+                "content": [{"type": "text", "text": "Hello"}],
+                "stop_reason": "end_turn",
+                "stop_sequence": None,
+                "usage": {"input_tokens": 1, "output_tokens": 1},
+            },
+        )
+
+    thinking: Final = {"type": "adaptive", "block_binding": {"prefix_mismatch_behavior": "error"}}
+    client = HTTPHandler(client=httpx.Client(transport=httpx.MockTransport(respond)))
+    try:
+        litellm.completion(
+            model="anthropic/claude-opus-4-8",
+            messages=[{"role": "user", "content": "Hello"}],
+            api_key="test-key",
+            client=client,
+            thinking=thinking,
+            extra_headers=extra_headers,
+        )
+    finally:
+        client.close()
+    assert thinking == {"type": "adaptive", "block_binding": {"prefix_mismatch_behavior": "error"}}
+    return json.loads(captured_requests[0].content)
+
+
+def test_anthropic_chat_keeps_block_binding_when_the_final_header_carries_the_binding_beta(monkeypatch):
+    monkeypatch.setenv("LITELLM_LOCAL_ANTHROPIC_BETA_HEADERS", "true")
+    body = _anthropic_chat_body_for_bound_thinking({"anthropic-beta": "thinking-binding-controls-2026-08-01"})
+    assert body["thinking"] == {"type": "adaptive", "block_binding": {"prefix_mismatch_behavior": "error"}}
+
+
+def test_anthropic_chat_drops_block_binding_when_no_header_carries_the_binding_beta(monkeypatch):
+    monkeypatch.setenv("LITELLM_LOCAL_ANTHROPIC_BETA_HEADERS", "true")
+    body = _anthropic_chat_body_for_bound_thinking(None)
+    assert body["thinking"] == {"type": "adaptive"}

tests/unit/llms/anthropic/messages/native/test_anthropic_messages_transformation.py

@@ -1,3 +1,4 @@
+import logging
 from collections.abc import Mapping
 from datetime import datetime, timezone
 from pathlib import Path
@@ -416,3 +417,59 @@ def test_drop_params_forwards_a_non_object_message_item_as_sent_on_the_native_re
 
     assert request["messages"][1] == 7
     assert request["messages"][2]["content"][0]["content"] == [{"type": "text", "text": "sunny"}]
+
+
+BINDING_AND_DISPLAY_BETAS: Final = "thinking-binding-controls-2026-08-01,thinking-display-updates-2026-08-18"
+
+
+def _bound_adaptive_request(headers: dict) -> dict:
+    thinking: Final = {
+        "type": "adaptive",
+        "block_binding": {"prefix_mismatch_behavior": "error"},
+        "display": "updates",
+    }
+    optional_params: Final = {"max_tokens": 64, "thinking": thinking}
+    request: Final = AnthropicMessagesConfig().transform_anthropic_messages_request(
+        model="claude-opus-4-8",
+        messages=[{"role": "user", "content": "hi"}],
+        anthropic_messages_optional_request_params=optional_params,
+        litellm_params=GenericLiteLLMParams(),
+        headers=headers,
+    )
+    assert thinking == {
+        "type": "adaptive",
+        "block_binding": {"prefix_mismatch_behavior": "error"},
+        "display": "updates",
+    }, "the caller's thinking dict must survive for a fallback leg"
+    return request
+
+
+def test_messages_request_keeps_beta_gated_thinking_fields_when_the_final_header_carries_the_betas():
+    request = _bound_adaptive_request({"anthropic-beta": BINDING_AND_DISPLAY_BETAS})
+    assert request["thinking"] == {
+        "type": "adaptive",
+        "block_binding": {"prefix_mismatch_behavior": "error"},
+        "display": "updates",
+    }
+
+
+def test_messages_request_drops_beta_gated_thinking_fields_the_final_header_does_not_carry(
+    caplog: pytest.LogCaptureFixture,
+):
+    with caplog.at_level(logging.WARNING, logger="LiteLLM"):
+        request = _bound_adaptive_request({"anthropic-beta": "thinking-binding-controls-2026-08-01"})
+    assert request["thinking"] == {"type": "adaptive", "block_binding": {"prefix_mismatch_behavior": "error"}}
+    assert any("thinking.display=updates" in record.getMessage() for record in caplog.records)
+
+
+def test_messages_request_drops_clear_at_from_a_mid_conversation_system_turn_without_its_beta():
+    system_turn: Final = {"role": "system", "content": "be terse", "clear_at": "end_turn"}
+    request = AnthropicMessagesConfig().transform_anthropic_messages_request(
+        model="claude-opus-4-8",
+        messages=[{"role": "user", "content": "hi"}, system_turn, {"role": "user", "content": "go"}],
+        anthropic_messages_optional_request_params={"max_tokens": 64},
+        litellm_params=GenericLiteLLMParams(),
+        headers={"anthropic-beta": "mid-conversation-system-2026-07-01"},
+    )
+    assert request["messages"][1] == {"role": "system", "content": "be terse"}
+    assert system_turn == {"role": "system", "content": "be terse", "clear_at": "end_turn"}

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

@@ -1,3 +1,4 @@
+import logging
 import base64
 import copy
 import json
@@ -10789,3 +10790,47 @@ def test_a_deployment_with_a_batch_control_plane_endpoint_keeps_it_out_of_chat_r
 
     assert response.choices[0].message.content == "ok"
     assert "aws_bedrock_control_plane_endpoint" not in route.calls[0].request.content.decode()
+
+
+def _converse_bound_thinking_request(headers: dict | None) -> dict:
+    thinking: Final = {
+        "type": "adaptive",
+        "block_binding": {"prefix_mismatch_behavior": "error"},
+        "display": "updates",
+    }
+    data = AmazonConverseConfig()._transform_request_helper(
+        model="anthropic.claude-opus-4-8-v1:0",
+        system_content_blocks=[],
+        optional_params={"thinking": thinking},
+        messages=[{"role": "user", "content": "hi"}],
+        headers=headers,
+    )
+    assert thinking == {
+        "type": "adaptive",
+        "block_binding": {"prefix_mismatch_behavior": "error"},
+        "display": "updates",
+    }, "the caller's thinking dict must survive for a fallback leg"
+    return data["additionalModelRequestFields"]
+
+
+def test_converse_keeps_beta_gated_thinking_fields_the_bedrock_converse_row_carries(monkeypatch):
+    monkeypatch.setenv("LITELLM_LOCAL_ANTHROPIC_BETA_HEADERS", "true")
+    fields = _converse_bound_thinking_request(
+        {"anthropic-beta": "thinking-binding-controls-2026-08-01,thinking-display-updates-2026-08-18"}
+    )
+    assert fields["thinking"] == {
+        "type": "adaptive",
+        "block_binding": {"prefix_mismatch_behavior": "error"},
+        "display": "updates",
+    }
+
+
+def test_converse_drops_beta_gated_thinking_fields_when_the_request_carries_no_beta(monkeypatch, caplog):
+    monkeypatch.setenv("LITELLM_LOCAL_ANTHROPIC_BETA_HEADERS", "true")
+    with caplog.at_level(logging.WARNING, logger="LiteLLM"):
+        fields = _converse_bound_thinking_request(None)
+    assert fields["thinking"] == {"type": "adaptive"}
+    assert any(
+        "thinking.block_binding" in record.getMessage() and "provider=bedrock_converse" in record.getMessage()
+        for record in caplog.records
+    )

tests/unit/test_anthropic_beta_gated_fields.py

@@ -0,0 +1,174 @@
+import logging
+from typing import Final
+
+import pytest
+
+from litellm.anthropic_beta_gated_fields import (
+    ANTHROPIC_MID_CONVERSATION_SYSTEM_CLEAR_AT_BETA_HEADER,
+    ANTHROPIC_THINKING_BINDING_CONTROLS_BETA_HEADER,
+    BETA_GATED_FIELDS,
+    active_beta_values,
+    log_beta_gated_strips,
+    strip_beta_gated_request_fields,
+    strip_beta_gated_system_messages,
+    strip_beta_gated_thinking,
+)
+from litellm.types.llms.anthropic import (
+    ANTHROPIC_MID_CONVERSATION_OUTPUT_CONFIG_BETA_HEADER,
+    ANTHROPIC_MID_CONVERSATION_TOOL_CHANGES_BETA_HEADER,
+    ANTHROPIC_THINKING_DISPLAY_UPDATES_BETA_HEADER,
+)
+
+EVERY_BETA: Final = frozenset(BETA_GATED_FIELDS)
+BOUND_THINKING: Final = {
+    "type": "adaptive",
+    "block_binding": {"prefix_mismatch_behavior": "error"},
+    "display": "updates",
+}
+
+
+def _system_turn() -> dict:
+    return {
+        "role": "system",
+        "clear_at": "end_turn",
+        "output_config": {"effort": "low"},
+        "content": [
+            {"type": "text", "text": "stay terse"},
+            {"type": "tool_addition", "tools": [{"name": "lookup"}]},
+            {"type": "tool_removal", "tool_names": ["lookup"]},
+        ],
+    }
+
+
+def test_thinking_with_every_beta_active_is_returned_as_is():
+    result = strip_beta_gated_thinking(BOUND_THINKING, EVERY_BETA)
+    assert result.thinking is BOUND_THINKING
+    assert result.stripped == ()
+
+
+def test_thinking_loses_only_the_field_whose_beta_is_absent():
+    without_binding = strip_beta_gated_thinking(
+        BOUND_THINKING, EVERY_BETA - {ANTHROPIC_THINKING_BINDING_CONTROLS_BETA_HEADER}
+    )
+    assert without_binding.thinking == {"type": "adaptive", "display": "updates"}
+    assert without_binding.stripped == ("thinking.block_binding",)
+
+    without_display = strip_beta_gated_thinking(
+        BOUND_THINKING, EVERY_BETA - {ANTHROPIC_THINKING_DISPLAY_UPDATES_BETA_HEADER}
+    )
+    assert without_display.thinking == {"type": "adaptive", "block_binding": {"prefix_mismatch_behavior": "error"}}
+    assert without_display.stripped == ("thinking.display=updates",)
+    assert BOUND_THINKING == {
+        "type": "adaptive",
+        "block_binding": {"prefix_mismatch_behavior": "error"},
+        "display": "updates",
+    }, "copy-on-write: the caller's thinking dict must survive for the fallback leg"
+
+
+def test_thinking_display_other_than_updates_is_not_gated():
+    thinking = {"type": "adaptive", "display": "summarized"}
+    result = strip_beta_gated_thinking(thinking, frozenset())
+    assert result.thinking is thinking
+
+
+@pytest.mark.parametrize("thinking", [None, "enabled", 12, ["adaptive"]])
+def test_non_mapping_thinking_passes_through(thinking: object):
+    result = strip_beta_gated_thinking(thinking, frozenset())
+    assert result.thinking is thinking
+    assert result.stripped == ()
+
+
+def test_system_message_keeps_every_gated_field_when_the_betas_are_active():
+    messages = [{"role": "user", "content": "hi"}, _system_turn()]
+    result = strip_beta_gated_system_messages(messages, EVERY_BETA)
+    assert result.messages is messages
+    assert result.stripped == ()
+
+
+def test_system_message_loses_exactly_the_fields_whose_betas_are_absent():
+    original = _system_turn()
+    messages = [{"role": "user", "content": "hi"}, original]
+    result = strip_beta_gated_system_messages(
+        messages,
+        EVERY_BETA
+        - {ANTHROPIC_MID_CONVERSATION_SYSTEM_CLEAR_AT_BETA_HEADER, ANTHROPIC_MID_CONVERSATION_TOOL_CHANGES_BETA_HEADER},
+    )
+    assert result.messages == [
+        {"role": "user", "content": "hi"},
+        {
+            "role": "system",
+            "output_config": {"effort": "low"},
+            "content": [{"type": "text", "text": "stay terse"}],
+        },
+    ]
+    assert result.stripped == (
+        "messages[].system.clear_at",
+        "messages[].system.content[].tool_addition|tool_removal",
+    )
+    assert original == _system_turn(), "copy-on-write: the caller's system message must survive"
+
+
+def test_per_turn_control_beta_also_unlocks_system_output_config():
+    messages = [{"role": "system", "output_config": {"effort": "low"}, "content": "terse"}]
+    assert strip_beta_gated_system_messages(messages, frozenset({"per-turn-control-2026-07-01"})).messages is messages
+    assert (
+        strip_beta_gated_system_messages(
+            messages, frozenset({ANTHROPIC_MID_CONVERSATION_OUTPUT_CONFIG_BETA_HEADER})
+        ).messages
+        is messages
+    )
+    assert strip_beta_gated_system_messages(messages, frozenset()).messages == [
+        {"role": "system", "content": "terse"}
+    ]
+
+
+def test_system_message_emptied_by_the_strip_is_dropped():
+    messages = [
+        {"role": "user", "content": "hi"},
+        {"role": "system", "content": [{"type": "tool_removal", "tool_names": ["lookup"]}]},
+        {"role": "assistant", "content": "ok"},
+    ]
+    result = strip_beta_gated_system_messages(messages, frozenset())
+    assert result.messages == [{"role": "user", "content": "hi"}, {"role": "assistant", "content": "ok"}]
+    assert result.stripped == (
+        "messages[].system.content[].tool_addition|tool_removal",
+        "messages[].system (emptied by the strip)",
+    )
+
+
+def test_request_strip_replaces_only_thinking_and_messages_and_keeps_the_dict_otherwise():
+    request = {"model": "claude", "thinking": dict(BOUND_THINKING), "messages": [_system_turn()], "max_tokens": 5}
+    result = strip_beta_gated_request_fields(request, frozenset())
+    assert result.request == {
+        "model": "claude",
+        "thinking": {"type": "adaptive"},
+        "messages": [{"role": "system", "content": [{"type": "text", "text": "stay terse"}]}],
+        "max_tokens": 5,
+    }
+    assert result.request is not request
+    assert request["thinking"] == BOUND_THINKING
+    assert request["messages"] == [_system_turn()]
+
+
+def test_request_strip_is_identity_when_nothing_is_gated():
+    request = {"thinking": {"type": "enabled", "budget_tokens": 1024}, "messages": [{"role": "user", "content": "hi"}]}
+    result = strip_beta_gated_request_fields(request, frozenset())
+    assert result.request is request
+    assert result.stripped == ()
+
+
+def test_active_beta_values_reads_the_header_case_insensitively_and_merges_body_betas():
+    assert active_beta_values({"Anthropic-Beta": "a-2026-01-01, b-2026-01-01 ,"}, ["c-2026-01-01", 7]) == frozenset(
+        {"a-2026-01-01", "b-2026-01-01", "c-2026-01-01"}
+    )
+    assert active_beta_values({}) == frozenset()
+
+
+def test_every_strip_is_logged_once_at_warning_with_the_fields_model_and_provider(caplog: pytest.LogCaptureFixture):
+    with caplog.at_level(logging.WARNING, logger="LiteLLM"):
+        log_beta_gated_strips(("thinking.block_binding", "thinking.block_binding"), model="claude", provider="azure_ai")
+        log_beta_gated_strips((), model="claude", provider="azure_ai")
+    warnings = [record for record in caplog.records if record.levelno == logging.WARNING]
+    assert len(warnings) == 1
+    assert "['thinking.block_binding']" in warnings[0].getMessage()
+    assert "model=claude provider=azure_ai" in warnings[0].getMessage()