scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-30 06:43:27 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 249

fix(anthropic)!: drop beta-gated body fields under drop_params (#48793)

Browse source 
* 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>

* test(integration): prove a leg that drops a beta sends neither the beta nor clear_at

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

* fix(bedrock): strip beta-gated fields against the final body anthropic_beta list

Converse counts the body anthropic_beta list as active, Invoke strips after
its auto-injected betas are known and injects thinking-binding-controls for
thinking.block_binding, test helpers are fully typed

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

* fix(anthropic): strip beta-gated system fields before an older model's system turn becomes a user turn

Models without supports_mid_conversation_system send no mid-conversation beta,
so tool_addition and tool_removal blocks are dropped (and logged) before the
turn is converted instead of riding along as user content. The Invoke
output_config regression test reads the real cost map instead of patching
_bedrock_model_supports

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

* test(anthropic): pin the integration proxy to the bundled beta allowlist and type the helper locals

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

* test(anthropic): annotate every new test local with Final and type the fixtures and returns

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

* fix(anthropic): strip beta-gated body fields only under drop_params

Without drop_params the request now goes out as sent and the provider
answers its own 400, like any other unsupported param. inline-tools-2026-09-15
also unlocks tool_addition and tool_removal blocks, so a leg that carries it
keeps them.

* fix(anthropic): strip beta-gated fields on Foundry chat and Bedrock chat Invoke under drop_params

Foundry chat and Bedrock chat Invoke now drop beta-gated body fields the same way the other legs do when drop_params is on and the request does not send the gating beta. The mid-conversation system normalizer validates only on the conversion path, and the strip WARNING names the beta the caller can send to keep the field.

* fix(anthropic): send the thinking betas for block_binding and display updates, strip Databricks chat thinking

Every Claude leg now adds the thinking-binding-controls beta when thinking carries block_binding and the display-updates beta for display updates, so those fields go through instead of a 400. Databricks chat rejects both fields with or without the beta, so it drops them under drop_params. A system turn the strip empties stays when it still carries output_config, and the drop warning names both causes.

* fix(anthropic): send the thinking betas on every leg and keep fields where the beta is unknown

Vertex messages and Claude Platform chat now add the thinking betas when thinking carries
block_binding or display updates. Converse sends only the caller's header betas when it has them.
A provider with no row in the beta config forwards beta-gated fields instead of dropping them,
an emptied trailing system turn that would leave an assistant prefill is sent as is, and the
prompt cache baseline body matches the body that goes out. The drop WARNING now names the beta
that unlocks each field

* fix(anthropic): filter Converse body betas, keep Foundry auto betas, read the bundled beta config for row checks

* fix(anthropic): validate messages only when a beta-gated drop runs

A non-object messages entry on a Claude model without mid-conversation
system support raised a raw pydantic ValidationError before the
beta-config and drop_params checks. The strip now passes the messages
through untouched unless it drops something, so the provider's 400
reaches SDK callers as BadRequestError again

* fix(anthropic): keep system blocks with a non-string type instead of raising

---------

Co-authored-by: mateo <mateo@berri.ai>
Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
This commit is contained in:
tavren27 2026-10-29 06:25:31 -07:00 • committed by GitHub
parent 1a05aab72a
commit b25ea48c4b
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
34 changed files with 2005 additions and 179 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

325
litellm/anthropic_beta_gated_fields.py Normal file
View file

@ -0,0 +1,325 @@
"""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. Under ``drop_params`` the same request degrades on
that leg by dropping those fields too; without it the request goes out as sent
and the provider answers its own 400, like any unsupported param. 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.litellm_core_utils.core_helpers import drops_unsupported_params
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_BINDING_CONTROLS_BETA_HEADER,
ANTHROPIC_THINKING_DISPLAY_UPDATES_BETA_HEADER,
)
ANTHROPIC_MID_CONVERSATION_SYSTEM_CLEAR_AT_BETA_HEADER: Final = "mid-conversation-system-clear-at-2026-08-21"
ANTHROPIC_INLINE_TOOLS_BETA_HEADER: Final = "inline-tools-2026-09-15"
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,),
ANTHROPIC_INLINE_TOOLS_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)
block_type: Final = mapping.get("type") if mapping is not None else None
return isinstance(block_type, str) and block_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], *, drop_params: object) -> 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 or not drops_unsupported_params(drop_params):
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 neither content nor ``output_config``, since such a 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, "", []) and "output_config" not in rewritten:
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 _ends_on_assistant(messages: Sequence[object]) -> bool:
last: Final = _as_mapping(messages[-1]) if messages else None
return last is not None and last.get("role") == "assistant"
def strip_beta_gated_system_messages(
messages: Sequence[object], active_betas: Set[str], *, drop_params: object
) -> StrippedMessages:
"""Drop the gated fields of every mid-conversation ``role: system`` message. Returns the same
sequence when nothing is gated, or when dropping an emptied system turn would leave the conversation
ending on an assistant turn, which the model would read as a prefill to continue."""
if not drops_unsupported_params(drop_params):
return StrippedMessages(messages=messages, stripped=())
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=())
kept: Final = [
message for result in per_message for message in result.messages
] # comprehension-ok: flattening per-message results
if _ends_on_assistant(kept) and not _ends_on_assistant(messages):
return StrippedMessages(messages=messages, stripped=())
return StrippedMessages(messages=kept, stripped=stripped)
def strip_beta_gated_request_fields(
request: Mapping[str, object], active_betas: Set[str], *, drop_params: object
) -> 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, drop_params=drop_params)
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, drop_params=drop_params)
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 unstripped_native_request(optional_params: object, messages: object) -> StrippedNativeRequest:
"""The identity result for legs that strip later, once their final ``anthropic_beta`` list exists."""
params: Final = _as_mapping(optional_params) or MappingProxyType({})
message_sequence: Final = _as_sequence(messages) if not isinstance(messages, str) else None
return StrippedNativeRequest(
thinking=params.get("thinking"),
thinking_stripped=False,
messages=message_sequence if message_sequence is not None else (),
messages_stripped=False,
stripped=(),
)
def strip_beta_gated_native_request(
headers: object, optional_params: object, messages: object, *, body_betas: object = (), drop_params: object
) -> StrippedNativeRequest:
"""The ``/v1/messages`` seam, fed the final (already filtered) headers, the final body ``anthropic_beta``
list on legs that carry betas in the body, plus the body parts it may rewrite."""
if not drops_unsupported_params(drop_params):
return unstripped_native_request(optional_params, messages)
active_betas: Final = active_beta_values(headers, body_betas)
params: Final = _as_mapping(optional_params) or MappingProxyType({})
thinking: Final = strip_beta_gated_thinking(params.get("thinking"), active_betas, drop_params=drop_params)
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, drop_params=drop_params)
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
described: Final = ", ".join(
f"{field} (unlocked by anthropic-beta {' or '.join(sorted(_FIELD_BETAS[field]))})"
if field in _FIELD_BETAS
else field
for field in sorted(set(stripped))
)
verbose_logger.warning(
"Dropped %s for model=%s provider=%s because drop_params is on and the provider would reject them on "
"this request, either since this provider leg does not send that anthropic-beta (see "
"anthropic_beta_headers_config.json) or since the model turns mid-conversation system messages into user "
"messages.",
described,
model,
provider,
)
def strip_beta_gated_request_body(
request: Mapping[str, object], headers: object, *, drop_params: object, model: str | None, provider: str
) -> Mapping[str, object]:
gated: Final = strip_beta_gated_request_fields(
request, active_beta_values(headers, request.get("anthropic_beta") or ()), drop_params=drop_params
)
log_beta_gated_strips(gated.stripped, model=model, provider=provider)
return gated.request

1
litellm/anthropic_beta_headers_config.json
View file

@ -248,6 +248,7 @@
"text_editor_20241022": null,
"text_editor_20250124": null,
"thinking-binding-controls-2026-08-01": "thinking-binding-controls-2026-08-01",
"thinking-display-updates-2026-08-18": "thinking-display-updates-2026-08-18",
"token-efficient-tools-2025-02-19": "token-efficient-tools-2025-02-19",
"web-fetch-2025-09-10": null,
"web-search-2025-03-05": "web-search-2025-03-05"

9
litellm/anthropic_beta_headers_manager.py
View file

@ -241,6 +241,13 @@ def get_provider_name(provider: str) -> str:
return aliases.get(provider, provider)
def has_beta_headers_config(provider: str) -> bool:
config: Final = _bundled_beta_headers_config()
aliases: Final = config.get("provider_aliases", {})
canonical_provider: Final = (aliases.get(provider) if isinstance(aliases, Mapping) else None) or provider
return canonical_provider in config
def filter_and_transform_beta_headers(
beta_headers: list[str],
provider: str,
@ -397,7 +404,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.

12
litellm/llms/anthropic/chat/handler.py
View file

@ -17,6 +17,7 @@ import litellm
import litellm.litellm_core_utils
import litellm.types
import litellm.types.utils
from litellm.anthropic_beta_gated_fields import strip_beta_gated_request_body
from litellm.anthropic_beta_headers_manager import (
update_request_with_filtered_beta,
)
@ -506,11 +507,20 @@ 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,
)
data: Final = dict(
strip_beta_gated_request_body(
filtered_data,
request_headers,
drop_params=litellm_params.get("drop_params"), # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType] # litellm_params is the handler's bare dict
model=model,
provider=custom_llm_provider,
)
)
logging_obj.pre_call(
input=messages,
api_key=api_key,

28
litellm/llms/anthropic/common_utils.py
View file

@ -51,6 +51,7 @@ from litellm.types.llms.anthropic import (
ANTHROPIC_OAUTH_BETA_HEADER,
ANTHROPIC_OAUTH_TOKEN_PREFIX,
ANTHROPIC_SERVER_TOOL_USE_ID_PREFIX,
ANTHROPIC_THINKING_BINDING_CONTROLS_BETA_HEADER,
ANTHROPIC_THINKING_DISPLAY_UPDATES_BETA_HEADER,
ANTHROPIC_TOOL_SEARCH_TOOL_TYPES,
AllAnthropicToolsValues,
@ -600,6 +601,20 @@ class AnthropicModelInfo(BaseLLMModelInfo):
return False
return thinking.get("type") in ("adaptive", "enabled") and thinking.get("display") == "updates"
def thinking_betas(self, thinking: AnthropicThinkingParam | None) -> tuple[str, ...]:
return (
*(
(ANTHROPIC_THINKING_DISPLAY_UPDATES_BETA_HEADER,)
if self.is_thinking_display_updates_used(thinking)
else ()
),
*(
(ANTHROPIC_THINKING_BINDING_CONTROLS_BETA_HEADER,)
if isinstance(thinking, dict) and "block_binding" in thinking
else ()
),
)
def is_mid_conversation_tool_change_used(self, messages: Sequence[object]) -> bool:
for message in messages:
if message_field(message, "role") != "system":
@ -1180,7 +1195,7 @@ class AnthropicModelInfo(BaseLLMModelInfo):
*,
custom_llm_provider: str,
is_mid_conversation_output_config_used: bool = False,
is_thinking_display_updates_used: bool = False,
thinking: AnthropicThinkingParam | None = None,
is_mid_conversation_tool_change_used: bool = False,
) -> list[str]:
"""
@ -1217,13 +1232,10 @@ class AnthropicModelInfo(BaseLLMModelInfo):
if is_mid_conversation_output_config_used:
betas.append(ANTHROPIC_MID_CONVERSATION_OUTPUT_CONFIG_BETA_HEADER)
thinking_display_betas: Final = (
(ANTHROPIC_THINKING_DISPLAY_UPDATES_BETA_HEADER,) if is_thinking_display_updates_used else ()
)
tool_change_betas: Final = (
(ANTHROPIC_MID_CONVERSATION_TOOL_CHANGES_BETA_HEADER,) if is_mid_conversation_tool_change_used else ()
)
return list(set(betas).union(thinking_display_betas, tool_change_betas))
return list(set(betas).union(self.thinking_betas(thinking), tool_change_betas))
@staticmethod
def _make_api_key_auth_header(
@ -1285,7 +1297,7 @@ class AnthropicModelInfo(BaseLLMModelInfo):
api_base: str | None = None,
use_bearer_for_custom_base: bool = False,
is_mid_conversation_output_config_used: bool = False,
is_thinking_display_updates_used: bool = False,
thinking: AnthropicThinkingParam | None = None,
is_mid_conversation_tool_change_used: bool = False,
wif_minted: bool = False,
) -> dict:
@ -1346,7 +1358,7 @@ class AnthropicModelInfo(BaseLLMModelInfo):
betas.update(user_anthropic_beta_headers)
all_betas: Final = betas.union(
(ANTHROPIC_THINKING_DISPLAY_UPDATES_BETA_HEADER,) if is_thinking_display_updates_used else (),
self.thinking_betas(thinking),
(ANTHROPIC_MID_CONVERSATION_TOOL_CHANGES_BETA_HEADER,) if is_mid_conversation_tool_change_used else (),
)
@ -1433,7 +1445,7 @@ class AnthropicModelInfo(BaseLLMModelInfo):
auth_token=auth_token,
file_id_used=file_id_used,
is_mid_conversation_output_config_used=is_mid_conversation_output_config_used,
is_thinking_display_updates_used=self.is_thinking_display_updates_used(optional_params.get("thinking")),
thinking=optional_params.get("thinking"),
is_mid_conversation_tool_change_used=self.is_mid_conversation_tool_change_used(messages),
web_search_tool_used=web_search_tool_used,
is_vertex_request=optional_params.get("is_vertex_request", False),

87
litellm/llms/anthropic/messages/native/transformation.py
View file

@ -2,10 +2,21 @@ 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_headers_manager import get_unsupported_headers, is_beta_header_supported_by_bundled_config
from litellm.anthropic_beta_gated_fields import (
log_beta_gated_strips,
strip_beta_gated_native_request,
strip_beta_gated_system_messages,
unstripped_native_request,
)
from litellm.anthropic_beta_headers_manager import (
get_unsupported_headers,
has_beta_headers_config,
is_beta_header_supported_by_bundled_config,
)
from litellm.constants import ANTHROPIC_BILLING_METADATA_PREFIX
from litellm.exceptions import AuthenticationError
from litellm.litellm_core_utils.core_helpers import drops_unsupported_params
@ -20,7 +31,6 @@ from litellm.types.llms.anthropic import (
ANTHROPIC_ADVISOR_TOOL_TYPE,
ANTHROPIC_BETA_HEADER_VALUES,
ANTHROPIC_MID_CONVERSATION_TOOL_CHANGES_BETA_HEADER,
ANTHROPIC_THINKING_DISPLAY_UPDATES_BETA_HEADER,
AnthropicMessagesRequest,
)
from litellm.types.llms.anthropic_messages.anthropic_response import (
@ -51,6 +61,8 @@ DEFAULT_ANTHROPIC_API_VERSION: Final = "2023-06-01"
_CALLER_CREDENTIAL_HEADERS: Final = frozenset({"x-api-key", "authorization"})
_MESSAGE_MAPPINGS: Final = TypeAdapter(tuple[Mapping[str, object], ...])
def _carries_caller_credential(headers: Mapping[str, str]) -> bool:
"""Whether the caller sent their own Anthropic credential, in which case this passthrough
@ -85,6 +97,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
@ -229,7 +244,18 @@ class AnthropicMessagesConfig(BaseAnthropicMessagesConfig):
else:
return system_param
def _normalize_system_role_messages(self, anthropic_messages_request: dict, model: str) -> None:
def _strip_beta_gated_system_turns(
self, messages: Sequence[Mapping[str, object]], model: str, drop_params: object
) -> Sequence[Mapping[str, object]]:
if not has_beta_headers_config(self._resolved_provider):
return messages
gated: Final = strip_beta_gated_system_messages(messages, frozenset(), drop_params=drop_params)
log_beta_gated_strips(gated.stripped, model=model, provider=self._resolved_provider)
return _MESSAGE_MAPPINGS.validate_python(gated.messages) if gated.stripped else messages
def _normalize_system_role_messages(
self, anthropic_messages_request: dict, model: str, drop_params: object
) -> None:
"""Normalize ``role: "system"`` entries in ``messages`` per the Anthropic
``/v1/messages`` contract, which the first-party API, Bedrock Invoke,
Vertex, and Azure Foundry all enforce identically.
@ -277,7 +303,15 @@ class AnthropicMessagesConfig(BaseAnthropicMessagesConfig):
custom_llm_provider=self.custom_llm_provider,
key="supports_mid_conversation_system",
)
else list(convert_mid_conversation_system_turns(messages[leading_count:]))
else list(
convert_mid_conversation_system_turns(
self._strip_beta_gated_system_turns(
messages[leading_count:], # pyright: ignore[reportUnknownArgumentType] # messages comes from the bare request dict
model=model,
drop_params=drop_params,
)
)
)
)
if hoisted or remaining != messages:
anthropic_messages_request["messages"] = remaining
@ -336,7 +370,7 @@ class AnthropicMessagesConfig(BaseAnthropicMessagesConfig):
model=model,
),
)
return self._finalize_messages_headers(headers, optional_params, messages), api_base
return self.finalize_messages_headers(headers, optional_params, messages), api_base
async def avalidate_anthropic_messages_environment(
self,
@ -376,7 +410,7 @@ class AnthropicMessagesConfig(BaseAnthropicMessagesConfig):
model=model,
),
)
return self._finalize_messages_headers(oauth_headers, optional_params, messages), api_base
return self.finalize_messages_headers(oauth_headers, optional_params, messages), api_base
def _require_auth_header(self, auth_header: Mapping[str, str] | None, model: str) -> Mapping[str, str]:
if auth_header is None:
@ -410,7 +444,7 @@ class AnthropicMessagesConfig(BaseAnthropicMessagesConfig):
return config_allows_workload_identity(self)
def _finalize_messages_headers(
def finalize_messages_headers(
self,
headers: dict, # mutable-ok: out-param
optional_params: dict, # mutable-ok: out-param
@ -619,6 +653,12 @@ class AnthropicMessagesConfig(BaseAnthropicMessagesConfig):
if thinking_enabled or effort_enabled:
optional_params.pop("temperature", None)
def strips_beta_gated_fields_from_final_headers(self) -> bool:
"""False on legs whose final ``anthropic_beta`` list is only known after this transform (Bedrock Invoke),
which strip beta-gated fields themselves once that list exists, and on providers with no row in
``anthropic_beta_headers_config.json``, whose filter drops every beta and so cannot say which fields fail."""
return has_beta_headers_config(self._resolved_provider)
def transform_anthropic_messages_request(
self,
model: str,
@ -720,18 +760,37 @@ class AnthropicMessagesConfig(BaseAnthropicMessagesConfig):
if not _has_advisor:
messages = strip_advisor_blocks_from_messages(messages)
gated: Final = (
strip_beta_gated_native_request(
headers, # pyright: ignore[reportUnknownArgumentType] # the base signature types this as a bare dict
anthropic_messages_optional_request_params, # pyright: ignore[reportUnknownArgumentType] # the base signature types this as a bare dict
messages, # pyright: ignore[reportUnknownArgumentType] # the base signature types this as a bare list
drop_params=getattr(litellm_params, "drop_params", None),
)
if self.strips_beta_gated_fields_from_final_headers()
else unstripped_native_request(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,
)
request: Final = (
{**dict(anthropic_messages_request), "thinking": gated.thinking}
if gated.thinking_stripped
else dict(anthropic_messages_request) # pyright: ignore[reportUnknownVariableType] # built from the base signature's bare dict
)
if not isinstance(tools, list):
return dict(anthropic_messages_request)
return request
return {
**anthropic_messages_request,
**request,
"tools": self.tools_for_upstream(
OBJECT_LIST_ADAPTER.validate_python(tools), model, litellm_params.drop_params
),
@ -866,17 +925,13 @@ class AnthropicMessagesConfig(BaseAnthropicMessagesConfig):
if AnthropicModelInfo().is_tool_search_used(tools):
beta_values.add(get_tool_search_beta_header(custom_llm_provider))
thinking_display_betas: Final = (
(ANTHROPIC_THINKING_DISPLAY_UPDATES_BETA_HEADER,)
if AnthropicModelInfo().is_thinking_display_updates_used(optional_params.get("thinking"))
else ()
)
thinking_betas: Final = AnthropicModelInfo().thinking_betas(optional_params.get("thinking"))
tool_change_betas: Final = (
(ANTHROPIC_MID_CONVERSATION_TOOL_CHANGES_BETA_HEADER,)
if AnthropicModelInfo().is_mid_conversation_tool_change_used(messages)
else ()
)
all_beta_values: Final = beta_values.union(thinking_display_betas, tool_change_betas)
all_beta_values: Final = beta_values.union(thinking_betas, tool_change_betas)
dangerous_tool_use: Final = ANTHROPIC_BETA_HEADER_VALUES.DANGEROUS_TOOL_USE_2026_09_03.value
forwarded_beta_values: Final = ( # pyright: ignore[reportUnknownVariableType] # derived from the bare beta set
all_beta_values - {dangerous_tool_use}

5
litellm/llms/anthropic/prompt_cache_prediction.py
View file

@ -661,12 +661,13 @@ def prepare_native_baseline_body(request: Mapping[str, object], model: str) -> M
TypeAdapter(tuple[str, ...]).validate_python(owned.get("additional_drop_params") or ()),
dict(options),
)
body: Final = AnthropicMessagesConfig().transform_anthropic_messages_request(
config: Final = AnthropicMessagesConfig()
body: Final = config.transform_anthropic_messages_request(
model=resolved_model,
messages=strip_provider_specific_fields_from_anthropic_messages(messages),
anthropic_messages_optional_request_params=filtered,
litellm_params=GenericLiteLLMParams(),
headers={},
headers=config.finalize_messages_headers({}, dict(filtered), list(messages)), # pyright: ignore[reportUnknownMemberType] # the shared helper takes and returns bare dicts
)
return MappingProxyType(_JSON_OBJECT.validate_python(body))
except Exception: # noqa: BLE001 # an unsupported hypothetical request is unavailable, never an inference failure

14
litellm/llms/azure_ai/anthropic/handler.py
View file

@ -4,11 +4,12 @@ Azure Anthropic handler - reuses AnthropicChatCompletion logic with Azure authen
import copy
import json
from collections.abc import Callable
from collections.abc import Callable, Mapping
from typing import Final
import httpx
from litellm.anthropic_beta_gated_fields import strip_beta_gated_request_body
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.llms.anthropic.chat.handler import AnthropicChatCompletion
from litellm.llms.anthropic.chat.transformation import JsonModeRequest
@ -75,13 +76,22 @@ class AzureAnthropicChatCompletion(AnthropicChatCompletion):
litellm_params=litellm_params,
)
data: Final = config.transform_request(
request_body: Final[Mapping[str, object]] = config.transform_request(
model=model,
messages=messages,
optional_params=optional_params,
litellm_params=litellm_params,
headers=headers,
)
data: Final = dict(
strip_beta_gated_request_body(
request_body,
headers, # pyright: ignore[reportUnknownArgumentType] # validate_environment returns a bare dict
drop_params=litellm_params.get("drop_params"), # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType] # litellm_params is the handler's bare dict
model=model,
provider=custom_llm_provider,
)
)
## LOGGING
logging_obj.pre_call(

4
litellm/llms/azure_ai/anthropic/messages_transformation.py
View file

@ -145,6 +145,8 @@ class AzureAnthropicMessagesConfig(AnthropicMessagesConfig):
litellm_params=litellm_params,
headers=headers,
)
self._normalize_system_role_messages(anthropic_messages_request, model=model)
self._normalize_system_role_messages(
anthropic_messages_request, model=model, drop_params=getattr(litellm_params, "drop_params", None)
)
self._remove_scope_from_cache_control(anthropic_messages_request)
return anthropic_messages_request

4
litellm/llms/azure_ai/anthropic/transformation.py
View file

@ -107,9 +107,11 @@ class AzureAnthropicConfig(AnthropicConfig):
is_vertex_request=optional_params.get("is_vertex_request", False),
user_anthropic_beta_headers=user_anthropic_beta_headers,
mcp_server_used=mcp_server_used,
thinking=optional_params.get("thinking"), # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType] # optional_params is the base config's bare dict
)
# Merge headers - Azure auth (api-key or Authorization) takes precedence
headers = {**anthropic_headers, **headers}
merged_beta_header: Final = {k: v for k, v in anthropic_headers.items() if k == "anthropic-beta"}
headers = {**anthropic_headers, **headers, **merged_beta_header}
# Ensure anthropic-version header is set
if "anthropic-version" not in headers:

36
litellm/llms/bedrock/chat/converse_transformation.py
View file

@ -17,6 +17,8 @@ from typing_extensions import ReadOnly, TypedDict
import litellm
from litellm._logging import verbose_logger
from litellm.anthropic_beta_gated_fields import active_beta_values, log_beta_gated_strips, strip_beta_gated_thinking
from litellm.anthropic_beta_headers_manager import filter_and_transform_beta_headers
from litellm.constants import (
BEDROCK_MIN_THINKING_BUDGET_TOKENS,
RESPONSE_FORMAT_TOOL_NAME,
@ -1825,7 +1827,16 @@ class AmazonConverseConfig(BaseConfig):
filtered_tools.append(tool)
base_model: Final = BedrockModelInfo.get_base_model(model)
client_beta_list: Final = get_anthropic_beta_from_headers(headers or {})
header_beta_list: Final = get_anthropic_beta_from_headers(headers or {})
client_beta_list: Final = header_beta_list or filter_and_transform_beta_headers(
sorted(
active_beta_values(
{},
additional_request_params.get("anthropic_beta"), # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType] # additional_request_params is a bare dict
)
),
provider="bedrock_converse",
)
eager_beta: Final = (
(ANTHROPIC_FINE_GRAINED_TOOL_STREAMING_BETA_HEADER,)
if base_model.startswith("anthropic")
@ -1833,7 +1844,15 @@ class AmazonConverseConfig(BaseConfig):
and ANTHROPIC_FINE_GRAINED_TOOL_STREAMING_BETA_HEADER not in client_beta_list
else ()
)
anthropic_beta_list: Final = [*client_beta_list, *eager_beta]
thinking_betas: Final = AnthropicModelInfo().thinking_betas(
additional_request_params.get("thinking"), # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType] # additional_request_params is a bare dict
)
missing_thinking_betas: Final = (
tuple(beta for beta in thinking_betas if beta not in client_beta_list)
if base_model.startswith("anthropic")
else ()
)
anthropic_beta_list: Final = [*client_beta_list, *eager_beta, *missing_thinking_betas]
# Only separate tools if computer use tools are actually present
if filtered_tools and self.is_computer_use_tool_used(filtered_tools, model):
@ -1903,7 +1922,8 @@ class AmazonConverseConfig(BaseConfig):
else:
computer_use_header = "computer-use-2024-10-22"
anthropic_beta_list.append(computer_use_header)
if computer_use_header not in anthropic_beta_list:
anthropic_beta_list.append(computer_use_header)
# Transform computer use tools to proper Bedrock format
transformed_computer_tools: Final = self._transform_computer_use_tools(computer_use_tools)
additional_request_params["tools"] = transformed_computer_tools
@ -2028,6 +2048,16 @@ 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
body_betas: Final = additional_request_params.get("anthropic_beta") # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType] # same bare dict
gated_thinking: Final = strip_beta_gated_thinking(
requested_thinking, # pyright: ignore[reportUnknownArgumentType] # same bare dict
active_beta_values({}, body_betas) | frozenset(anthropic_beta_list), # pyright: ignore[reportUnknownArgumentType] # _process_tools_and_beta hands back a bare list
drop_params=drop_params,
)
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)

14
litellm/llms/bedrock/chat/invoke_transformations/anthropic_claude_transformation.py
View file

@ -4,6 +4,7 @@ from typing import TYPE_CHECKING, Any, Final
import httpx
from pydantic import TypeAdapter
from litellm.anthropic_beta_gated_fields import strip_beta_gated_request_body
from litellm.anthropic_beta_headers_manager import filter_and_transform_beta_headers
from litellm.litellm_core_utils.llm_cost_calc.guardrail_cost import object_mapping
from litellm.litellm_core_utils.prompt_templates.factory import (
@ -210,7 +211,15 @@ class AmazonAnthropicClaudeConfig(AmazonInvokeConfig, AnthropicConfig):
if beta_list:
_anthropic_request["anthropic_beta"] = beta_list
return _anthropic_request
return dict(
strip_beta_gated_request_body(
_anthropic_request, # pyright: ignore[reportUnknownArgumentType] # built by the base signature's bare dict transform
{},
drop_params=litellm_params.get("drop_params"), # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType] # litellm_params is the base signature's bare dict
model=model,
provider="bedrock_invoke",
)
)
@property
def uses_async_transform_request(self) -> bool:
@ -294,7 +303,6 @@ class AmazonAnthropicClaudeConfig(AmazonInvokeConfig, AnthropicConfig):
programmatic_tool_calling_used: Final = self.is_programmatic_tool_calling_used(tools)
input_examples_used: Final = self.is_input_examples_used(tools)
is_mid_conversation_output_config_used: Final = self.is_mid_conversation_output_config_used(messages)
is_thinking_display_updates_used: Final = self.is_thinking_display_updates_used(thinking)
client_beta_set: Final = set(get_anthropic_beta_from_headers(headers))
user_beta_set: Final = (
@ -312,7 +320,7 @@ class AmazonAnthropicClaudeConfig(AmazonInvokeConfig, AnthropicConfig):
mcp_server_used=self.is_mcp_server_used(optional_params.get("mcp_servers")),
custom_llm_provider="bedrock",
is_mid_conversation_output_config_used=is_mid_conversation_output_config_used,
is_thinking_display_updates_used=is_thinking_display_updates_used,
thinking=thinking,
)
beta_set.update(auto_betas)

1
litellm/llms/bedrock/claude_platform/transformation.py
View file

@ -72,6 +72,7 @@ class BedrockClaudePlatformConfig(BedrockClaudePlatformMixin, AnthropicConfig):
),
code_execution_tool_used=self.is_code_execution_tool_used(tools=optional_params.get("tools")),
container_with_skills_used=self.is_container_with_skills_used(optional_params=optional_params),
thinking=optional_params.get("thinking"), # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType] # optional_params is the base config's bare dict
)
anthropic_headers["anthropic-workspace-id"] = workspace_id
return {**headers, **anthropic_headers}

26
litellm/llms/bedrock/messages/invoke_transformations/anthropic_claude_transformation.py
View file

@ -7,6 +7,7 @@ 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 filter_and_transform_beta_headers
from litellm.constants import (
BEDROCK_MIN_THINKING_BUDGET_TOKENS,
@ -549,6 +550,9 @@ class AmazonAnthropicClaudeMessagesConfig(
"edits": retained_edits,
}
def strips_beta_gated_fields_from_final_headers(self) -> bool:
return False
def _get_bedrock_invoke_anthropic_beta_headers(
self,
model: str,
@ -584,9 +588,7 @@ class AmazonAnthropicClaudeMessagesConfig(
),
custom_llm_provider="bedrock",
is_mid_conversation_output_config_used=is_mid_conversation_output_config_used,
is_thinking_display_updates_used=anthropic_model_info.is_thinking_display_updates_used(
anthropic_messages_request.get("thinking")
),
thinking=anthropic_messages_request.get("thinking"),
is_mid_conversation_tool_change_used=anthropic_model_info.is_mid_conversation_tool_change_used(
outgoing_messages_typed
),
@ -757,7 +759,9 @@ class AmazonAnthropicClaudeMessagesConfig(
),
"display": "updates",
}
self._normalize_system_role_messages(anthropic_messages_request, model=model)
self._normalize_system_role_messages(
anthropic_messages_request, model=model, drop_params=getattr(litellm_params, "drop_params", None)
)
#########################################################
############## BEDROCK Invoke SPECIFIC TRANSFORMATION ###
#########################################################
@ -825,6 +829,20 @@ class AmazonAnthropicClaudeMessagesConfig(
if filtered_betas:
anthropic_messages_request["anthropic_beta"] = filtered_betas
final_request: Final = OBJECT_MAPPING_ADAPTER.validate_python(anthropic_messages_request) # pyright: ignore[reportUnknownArgumentType] # built from the base signature's bare dict
gated: Final = strip_beta_gated_native_request(
{},
final_request,
final_request.get("messages"),
body_betas=filtered_betas,
drop_params=getattr(litellm_params, "drop_params", None),
)
log_beta_gated_strips(gated.stripped, model=model, provider="bedrock_invoke")
if gated.thinking_stripped:
anthropic_messages_request["thinking"] = gated.thinking
if gated.messages_stripped:
anthropic_messages_request["messages"] = list(gated.messages)
outbound_tools: Final = tools_without_eager_input_streaming(anthropic_messages_request)
if outbound_tools is not None:
anthropic_messages_request["tools"] = outbound_tools

9
litellm/llms/databricks/chat/transformation.py
View file

@ -10,6 +10,7 @@ from typing import TYPE_CHECKING, Any, Final, Literal, cast, overload
import httpx
from pydantic import BaseModel, ConfigDict, TypeAdapter
from litellm.anthropic_beta_gated_fields import log_beta_gated_strips, strip_beta_gated_thinking
from litellm.constants import RESPONSE_FORMAT_TOOL_NAME
from litellm.litellm_core_utils.llm_response_utils.convert_dict_to_response import (
handle_invalid_parallel_tool_calls,
@ -514,6 +515,14 @@ class DatabricksConfig(DatabricksBase, OpenAILikeChatConfig, AnthropicConfig):
self.update_optional_params_with_thinking_tokens(
non_default_params=non_default_params, optional_params=mapped_params
)
gated_thinking: Final = strip_beta_gated_thinking(
mapped_params.get("thinking"), # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType] # mapped_params is the base config's bare dict
frozenset(),
drop_params=drop_params,
)
if gated_thinking.stripped:
log_beta_gated_strips(gated_thinking.stripped, model=model, provider="databricks")
mapped_params["thinking"] = gated_thinking.thinking
return mapped_params

6
litellm/llms/databricks/messages/transformation.py
View file

@ -144,7 +144,7 @@ class DatabricksAnthropicMessagesConfig(DatabricksBase, AnthropicMessagesConfig)
("content-type" if key.lower() == "content-type" else key): value
for key, value in {**auth_headers, **client_anthropic_headers}.items()
}
finalized_headers: Final[dict[str, str]] = self._finalize_messages_headers( # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType] # the shared helper returns a bare dict # mutable-ok: becomes the outbound headers
finalized_headers: Final[dict[str, str]] = self.finalize_messages_headers( # pyright: ignore[reportUnknownMemberType, reportUnknownVariableType] # the shared helper returns a bare dict # mutable-ok: becomes the outbound headers
headers=normalized_headers,
optional_params=optional_params,
messages=messages,
@ -191,6 +191,8 @@ class DatabricksAnthropicMessagesConfig(DatabricksBase, AnthropicMessagesConfig)
litellm_params=litellm_params,
headers=headers,
)
self._normalize_system_role_messages(anthropic_messages_request, model=model) # pyright: ignore[reportUnknownMemberType] # the shared helper takes a bare dict
self._normalize_system_role_messages( # pyright: ignore[reportUnknownMemberType] # the shared helper takes a bare dict
anthropic_messages_request, model=model, drop_params=getattr(litellm_params, "drop_params", None)
)
self._remove_scope_from_cache_control(anthropic_messages_request) # pyright: ignore[reportUnknownMemberType] # the shared helper takes a bare dict
return anthropic_messages_request

10
litellm/llms/vertex_ai/vertex_ai_partner_models/anthropic/messages/transformation.py
View file

@ -211,6 +211,12 @@ class VertexAIPartnerModelsAnthropicMessagesConfig(AnthropicMessagesConfig, Vert
if messages_carry_output_config(messages):
beta_values.add(ANTHROPIC_BETA_HEADER_VALUES.PER_TURN_CONTROL_2026_07_01.value)
beta_values.update(
anthropic_model_info.thinking_betas(
optional_params.get("thinking"), # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType] # optional_params is the base config's bare dict
)
)
if beta_values:
request_headers["anthropic-beta"] = ",".join(beta_values)
@ -245,7 +251,9 @@ class VertexAIPartnerModelsAnthropicMessagesConfig(AnthropicMessagesConfig, Vert
headers=headers,
)
self._normalize_system_role_messages(anthropic_messages_request, model=model)
self._normalize_system_role_messages(
anthropic_messages_request, model=model, drop_params=getattr(litellm_params, "drop_params", None)
)
self._remove_scope_from_cache_control(anthropic_messages_request)

1
litellm/llms/vertex_ai/vertex_ai_partner_models/anthropic/transformation.py
View file

@ -145,6 +145,7 @@ class VertexAIAnthropicConfig(AnthropicConfig):
file_id_used=self.is_file_id_used(messages),
mcp_server_used=self.is_mcp_server_used(optional_params.get("mcp_servers")),
custom_llm_provider="vertex_ai",
thinking=optional_params.get("thinking"), # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType] # optional_params is the base config's bare dict
)
beta_set: Final = set(auto_betas)

1
litellm/types/llms/anthropic.py
View file

@ -832,6 +832,7 @@ ANTHROPIC_EFFORT_BETA_HEADER: Final = "effort-2025-11-24"
ANTHROPIC_MID_CONVERSATION_OUTPUT_CONFIG_BETA_HEADER: Final = "mid-conversation-output-config-2026-07-01"
ANTHROPIC_THINKING_DISPLAY_UPDATES_BETA_HEADER: Final = "thinking-display-updates-2026-08-18"
ANTHROPIC_THINKING_BINDING_CONTROLS_BETA_HEADER: Final = "thinking-binding-controls-2026-08-01"
ANTHROPIC_MID_CONVERSATION_TOOL_CHANGES_BETA_HEADER: Final = "mid-conversation-tool-changes-2026-07-01"
ANTHROPIC_FINE_GRAINED_TOOL_STREAMING_BETA_HEADER: Final = "fine-grained-tool-streaming-2025-05-14"

183
tests/integration/messages_endpoint/test_beta_gated_fields_wire.py Normal file
View file

@ -0,0 +1,183 @@
import json
import re
import uuid
from collections.abc import Iterator, Mapping
from dataclasses import dataclass
from pathlib import Path
from typing import Final
import pytest
from integration._support.client import Gateway, gateway_from_environment, list_value, object_value
from integration._support.process import owned_proxy
from integration._support.wire import Reply, Request, Wire, wire_server
from pydantic import JsonValue, TypeAdapter
pytestmark: Final = pytest.mark.shares_worker_state
ANTHROPIC_MODEL: Final = "beta-gated-anthropic"
FOUNDRY_MODEL: Final = "beta-gated-foundry"
FOUNDRY_DROP_PARAMS_MODEL: Final = "beta-gated-foundry-drop-params"
CLEAR_AT_BETA: Final = "mid-conversation-system-clear-at-2026-08-21"
_JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue])
_NONCE: Final = re.compile(rb"nonce-([0-9a-f]{32})")
def nonce() -> str:
return f"nonce-{uuid.uuid4().hex}"
def claude_reply(request: Request) -> Reply:
match: Final = _NONCE.search(request.body)
assert match is not None, request.body[:200]
identity: Final = match.group(1).decode()
return Reply(
body=json.dumps(
{
"id": f"msg_{identity}",
"type": "message",
"role": "assistant",
"model": "claude-opus-4-8",
"content": [{"type": "text", "text": f"answered {identity}"}],
"stop_reason": "end_turn",
"stop_sequence": None,
"usage": {"input_tokens": 12, "output_tokens": 3},
}
).encode()
)
def respond(request: Request) -> Reply:
if request.method != "POST":
return Reply(body=b'{"object":"list","data":[]}')
return claude_reply(request)
def deployment(name: str, model: str, api_base: str, drop_params: bool = False) -> dict[str, JsonValue]:
return {
"model_name": name,
"litellm_params": {
"model": model,
"api_base": api_base,
"api_key": f"synthetic-{name}-key",
**({"drop_params": True} if drop_params else {}),
},
}
def write_config(directory: Path, wire_url: str) -> Path:
path: Final = directory / f"beta-gated-{uuid.uuid4().hex}.json"
path.write_text(
json.dumps(
{
"model_list": [
deployment(ANTHROPIC_MODEL, "anthropic/claude-opus-4-8", f"{wire_url}/anthropic"),
deployment(FOUNDRY_MODEL, "azure_ai/claude-opus-4-8", f"{wire_url}/foundry"),
deployment(
FOUNDRY_DROP_PARAMS_MODEL, "azure_ai/claude-opus-4-8", f"{wire_url}/foundry", drop_params=True
),
],
"general_settings": {
"master_key": "os.environ/LITELLM_MASTER_KEY",
"database_url": "os.environ/DATABASE_URL",
},
}
)
)
return path
@dataclass(frozen=True, slots=True)
class Rig:
gateway: Gateway
wire: Wire
def send(self, model: str, request_nonce: str) -> tuple[int, bytes]:
response: Final = self.gateway.client.post(
"/v1/messages",
json={
"model": model,
"max_tokens": 64,
"messages": [
{"role": "user", "content": f"hi {request_nonce}"},
{"role": "assistant", "content": "Hello"},
{"role": "user", "content": "how are you"},
{"role": "system", "content": "Answer in French", "clear_at": "next_user_message"},
],
},
headers={"Authorization": f"Bearer {self.gateway.key}", "anthropic-beta": CLEAR_AT_BETA},
timeout=120,
)
return response.status_code, response.content
def received(self, request_nonce: str) -> Request:
requests: Final = tuple(request for request in self.wire.drain() if request_nonce.encode() in request.body)
assert len(requests) == 1, [request.target for request in requests]
return requests[0]
@pytest.fixture(scope="module")
def rig(tmp_path_factory: pytest.TempPathFactory) -> Iterator[Rig]:
directory: Final = tmp_path_factory.mktemp("beta-gated")
with (
gateway_from_environment() as shared,
wire_server(respond) as wire,
owned_proxy(
shared,
directory,
{"LITELLM_LOCAL_ANTHROPIC_BETA_HEADERS": "true"},
config=write_config(directory, wire.url),
workers=1,
) as gateway,
):
yield Rig(gateway, wire)
def reply_text(status: int, raw: bytes) -> str:
assert status == 200, raw[:500]
return "".join(str(object_value(block)["text"]) for block in list_value(_JSON_OBJECT.validate_json(raw)["content"]))
def beta_header(headers: Mapping[str, str]) -> str:
return ",".join(value for key, value in headers.items() if key.lower() == "anthropic-beta")
def system_message(body: bytes) -> dict[str, JsonValue]:
messages: Final = list_value(_JSON_OBJECT.validate_json(body)["messages"])
system: Final = [object_value(m) for m in messages if object_value(m).get("role") == "system"]
assert len(system) == 1, messages
return system[0]
def test_a_drop_params_leg_whose_row_lacks_the_beta_sends_neither_the_beta_nor_clear_at(rig: Rig) -> None:
request_nonce: Final = nonce()
status, raw = rig.send(FOUNDRY_DROP_PARAMS_MODEL, request_nonce)
assert reply_text(status, raw) == f"answered {request_nonce.removeprefix('nonce-')}"
upstream: Final = rig.received(request_nonce)
assert CLEAR_AT_BETA not in beta_header(upstream.headers), upstream.headers
assert system_message(upstream.body) == {"role": "system", "content": "Answer in French"}
def test_a_leg_whose_row_lacks_the_beta_forwards_clear_at_as_sent_without_drop_params(rig: Rig) -> None:
request_nonce: Final = nonce()
status, raw = rig.send(FOUNDRY_MODEL, request_nonce)
assert reply_text(status, raw) == f"answered {request_nonce.removeprefix('nonce-')}"
upstream: Final = rig.received(request_nonce)
assert CLEAR_AT_BETA not in beta_header(upstream.headers), upstream.headers
assert system_message(upstream.body) == {
"role": "system",
"content": "Answer in French",
"clear_at": "next_user_message",
}
def test_a_leg_that_carries_the_beta_forwards_clear_at_untouched(rig: Rig) -> None:
request_nonce: Final = nonce()
status, raw = rig.send(ANTHROPIC_MODEL, request_nonce)
assert reply_text(status, raw) == f"answered {request_nonce.removeprefix('nonce-')}"
upstream: Final = rig.received(request_nonce)
assert CLEAR_AT_BETA in beta_header(upstream.headers), upstream.headers
assert system_message(upstream.body) == {
"role": "system",
"content": "Answer in French",
"clear_at": "next_user_message",
}

88
tests/unit/llms/anthropic/chat/test_anthropic_chat_handler.py
View file

@ -537,7 +537,11 @@ def test_streamed_thinking_blocks_accumulate_in_the_openai_sdk_and_replay_withou
{"type": "content_block_delta", "index": 2, "delta": {"type": "thinking_delta", "thinking": "Step 3."}},
{"type": "content_block_delta", "index": 2, "delta": {"type": "signature_delta", "signature": "sig-b"}},
{"type": "content_block_stop", "index": 2},
{"type": "message_delta", "delta": {"stop_reason": "end_turn", "stop_sequence": None}, "usage": {"output_tokens": 9}},
{
"type": "message_delta",
"delta": {"stop_reason": "end_turn", "stop_sequence": None},
"usage": {"output_tokens": 9},
},
{"type": "message_stop"},
)
@ -558,7 +562,11 @@ def test_streamed_thinking_blocks_accumulate_in_the_openai_sdk_and_replay_withou
"thinking_blocks": message["thinking_blocks"],
}
replayed: Final = anthropic_messages_pt(
messages=[{"role": "user", "content": "Capital of France?"}, next_turn, {"role": "user", "content": "And Spain?"}],
messages=[
{"role": "user", "content": "Capital of France?"},
next_turn,
{"role": "user", "content": "And Spain?"},
],
model="claude-haiku-4-5",
llm_provider="anthropic",
)
@ -3650,3 +3658,79 @@ 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_request_for_bound_thinking(
extra_headers: dict[str, str] | None, drop_params: bool | None = None
) -> httpx.Request:
captured_requests: Final[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: Final = 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,
drop_params=drop_params,
)
finally:
client.close()
assert thinking == {"type": "adaptive", "block_binding": {"prefix_mismatch_behavior": "error"}}
return captured_requests[0]
def _anthropic_chat_body_for_bound_thinking(
extra_headers: dict[str, str] | None, drop_params: bool | None = None
) -> dict[str, object]:
return json.loads(_anthropic_chat_request_for_bound_thinking(extra_headers, drop_params).content)
def test_anthropic_chat_keeps_block_binding_when_the_final_header_carries_the_binding_beta(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setenv("LITELLM_LOCAL_ANTHROPIC_BETA_HEADERS", "true")
body: Final = _anthropic_chat_body_for_bound_thinking(
{"anthropic-beta": "thinking-binding-controls-2026-08-01"}, drop_params=True
)
assert body["thinking"] == {"type": "adaptive", "block_binding": {"prefix_mismatch_behavior": "error"}}
@pytest.mark.parametrize("drop_params", [True, False])
def test_anthropic_chat_adds_the_binding_beta_so_block_binding_is_kept(
drop_params: bool, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setenv("LITELLM_LOCAL_ANTHROPIC_BETA_HEADERS", "true")
monkeypatch.setattr(litellm, "drop_params", False)
request: Final = _anthropic_chat_request_for_bound_thinking(None, drop_params=drop_params)
assert json.loads(request.content)["thinking"] == {
"type": "adaptive",
"block_binding": {"prefix_mismatch_behavior": "error"},
}
assert "thinking-binding-controls-2026-08-01" in request.headers["anthropic-beta"].split(",")
def test_anthropic_chat_sends_block_binding_as_is_without_drop_params(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("LITELLM_LOCAL_ANTHROPIC_BETA_HEADERS", "true")
monkeypatch.setattr(litellm, "drop_params", False)
body: Final = _anthropic_chat_body_for_bound_thinking(None)
assert body["thinking"] == {"type": "adaptive", "block_binding": {"prefix_mismatch_behavior": "error"}}

133
tests/unit/llms/anthropic/messages/native/test_anthropic_messages_transformation.py
View file

@ -1,3 +1,4 @@
import logging
from collections.abc import Mapping
from datetime import datetime, timezone
from pathlib import Path
@ -6,9 +7,11 @@ from unittest.mock import patch
import httpx
import pytest
import respx
from pydantic import JsonValue
import litellm
from litellm import anthropic_beta_headers_manager
from litellm.litellm_core_utils.litellm_logging import Logging
from litellm.llms.anthropic.messages.native.transformation import (
AnthropicMessagesConfig,
@ -498,3 +501,133 @@ 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 test_messages_headers_add_the_binding_beta_when_thinking_carries_block_binding() -> None:
headers: Final = AnthropicMessagesConfig()._update_headers_with_anthropic_beta(
headers={},
optional_params={"thinking": {"type": "adaptive", "block_binding": {"prefix_mismatch_behavior": "error"}}},
messages=[],
)
assert "thinking-binding-controls-2026-08-01" in headers["anthropic-beta"].split(",")
def test_messages_headers_add_no_binding_beta_without_block_binding() -> None:
headers: Final = AnthropicMessagesConfig()._update_headers_with_anthropic_beta(
headers={}, optional_params={"thinking": {"type": "adaptive"}}, messages=[]
)
assert "thinking-binding-controls-2026-08-01" not in headers.get("anthropic-beta", "")
def _bound_adaptive_request(headers: dict[str, str], drop_params: bool | None = True) -> dict[str, object]:
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(drop_params=drop_params),
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() -> None:
request: Final = _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,
) -> None:
with caplog.at_level(logging.WARNING, logger="LiteLLM"):
request: Final = _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_sends_beta_gated_thinking_fields_as_is_without_drop_params(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(litellm, "drop_params", False)
request: Final = _bound_adaptive_request({}, drop_params=None)
assert request["thinking"] == {
"type": "adaptive",
"block_binding": {"prefix_mismatch_behavior": "error"},
"display": "updates",
}
def test_messages_request_drops_clear_at_from_a_mid_conversation_system_turn_without_its_beta() -> None:
system_turn: Final = {"role": "system", "content": "be terse", "clear_at": "end_turn"}
request: Final = 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(drop_params=True),
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"}
def test_a_provider_without_a_beta_headers_row_forwards_beta_gated_fields_under_drop_params(
local_beta_headers_config: None,
) -> None:
thinking: Final = {"type": "enabled", "budget_tokens": 1024, "display": "updates"}
request: Final = MinimaxMessagesConfig().transform_anthropic_messages_request(
model="MiniMax-M2",
messages=[
{"role": "user", "content": "hi"},
{"role": "assistant", "content": "Hello"},
{"role": "user", "content": "how are you"},
{"role": "system", "content": "Answer in French", "output_config": {"effort": "low"}},
],
anthropic_messages_optional_request_params={"max_tokens": 64, "thinking": dict(thinking)},
litellm_params=GenericLiteLLMParams(drop_params=True),
headers={},
)
assert request["thinking"] == thinking
assert request["messages"][-1]["output_config"] == {"effort": "low"}
def test_a_first_party_messages_request_never_fetches_the_remote_beta_headers_config(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.delenv("LITELLM_LOCAL_ANTHROPIC_BETA_HEADERS", raising=False)
monkeypatch.setattr(anthropic_beta_headers_manager, "_BETA_HEADERS_CONFIG", None)
with respx.mock(assert_all_called=False) as router:
remote_config: Final = router.get(litellm.anthropic_beta_headers_url).mock(
return_value=httpx.Response(200, json={})
)
request: Final = AnthropicMessagesConfig().transform_anthropic_messages_request(
model="claude-opus-4-8",
messages=[
{"role": "user", "content": "hi"},
{"role": "system", "content": "be terse", "clear_at": "end_turn"},
{"role": "user", "content": "go"},
],
anthropic_messages_optional_request_params={"max_tokens": 64},
litellm_params=GenericLiteLLMParams(drop_params=True),
headers={},
)
assert request["messages"][1] == {"role": "system", "content": "be terse"}
assert not remote_config.called

21
tests/unit/llms/anthropic/test_anthropic_common_utils.py
View file

@ -4295,7 +4295,14 @@ def test_shared_legacy_thinking_translation_carries_block_binding(
@pytest.mark.usefixtures("local_model_cost_map")
@pytest.mark.parametrize(
("custom_llm_provider", "model", "requested_thinking", "rejected_keys", "expected_thinking"),
(
"custom_llm_provider",
"model",
"requested_thinking",
"rejected_keys",
"expected_thinking",
"provider_rejects_block_binding",
),
(
(
"anthropic",
@ -4303,6 +4310,7 @@ def test_shared_legacy_thinking_translation_carries_block_binding(
{"type": "enabled", "budget_tokens": 2048},
{"not_a_thinking_key": True},
{"type": "adaptive"},
False,
),
(
"bedrock",
@ -4310,6 +4318,7 @@ def test_shared_legacy_thinking_translation_carries_block_binding(
{"type": "enabled", "budget_tokens": 2048},
{"not_a_thinking_key": True},
{"type": "adaptive"},
False,
),
(
"bedrock",
@ -4317,6 +4326,7 @@ def test_shared_legacy_thinking_translation_carries_block_binding(
{"type": "enabled", "budget_tokens": 2048},
{"not_a_thinking_key": True},
{"type": "adaptive"},
False,
),
(
"bedrock",
@ -4324,6 +4334,7 @@ def test_shared_legacy_thinking_translation_carries_block_binding(
{"type": "adaptive", "display": "updates"},
{"not_a_thinking_key": True, "budget_tokens": 2048},
{"type": "adaptive", "display": "updates"},
False,
),
(
"vertex_ai",
@ -4331,6 +4342,7 @@ def test_shared_legacy_thinking_translation_carries_block_binding(
{"type": "enabled", "budget_tokens": 2048},
{"not_a_thinking_key": True},
{"type": "adaptive"},
False,
),
(
"databricks",
@ -4338,6 +4350,7 @@ def test_shared_legacy_thinking_translation_carries_block_binding(
{"type": "enabled", "budget_tokens": 2048},
{"not_a_thinking_key": True},
{"type": "adaptive"},
True,
),
),
ids=(
@ -4356,6 +4369,7 @@ def test_chat_thinking_drops_unknown_keys_only_under_drop_params(
requested_thinking: dict[str, object],
rejected_keys: dict[str, object],
expected_thinking: dict[str, object],
provider_rejects_block_binding: bool,
drop_params_source: str | None,
monkeypatch: pytest.MonkeyPatch,
) -> None:
@ -4370,7 +4384,10 @@ def test_chat_thinking_drops_unknown_keys_only_under_drop_params(
)
kept_rejected_keys: Final = {} if drop_params_source else rejected_keys
assert optional_params["thinking"] == {**expected_thinking, "block_binding": block_binding, **kept_rejected_keys}
kept_block_binding: Final = (
{} if drop_params_source and provider_rejects_block_binding else {"block_binding": block_binding}
)
assert optional_params["thinking"] == {**expected_thinking, **kept_block_binding, **kept_rejected_keys}
@pytest.mark.usefixtures("local_model_cost_map", "local_beta_headers_config")

414
tests/unit/llms/anthropic/test_anthropic_prompt_cache_prediction.py
View file

@ -13,7 +13,7 @@ import litellm
from litellm.caching.dual_cache import DualCache
from litellm.caching.llm_caching_handler import LLMClientCache
from litellm.llms.anthropic.count_tokens import handler as count_handler
from litellm.llms.anthropic.messages.native.transformation import DEFAULT_ANTHROPIC_API_VERSION
from litellm.llms.anthropic.messages.native.transformation import DEFAULT_ANTHROPIC_API_VERSION, AnthropicMessagesConfig
from litellm.llms.anthropic.prompt_cache_prediction import (
CountedPromptCachePlan,
NativePredictionTarget,
@ -26,6 +26,7 @@ from litellm.llms.anthropic.prompt_cache_prediction import (
parse_cache_plan,
parse_observed_cache,
parse_prompt,
prepare_native_baseline_body,
resolve_baseline_prediction_target,
resolve_prediction_target,
supported_prediction_headers,
@ -36,7 +37,7 @@ from litellm.proxy import proxy_server
from litellm.proxy.hooks.prompt_cache_prediction import PromptCacheObserver
from litellm.proxy.management_endpoints.prompt_cache_prediction import predict_arm
from litellm.proxy.utils import InternalUsageCache
from litellm.types.router import Deployment, LiteLLM_Params, ModelInfo
from litellm.types.router import Deployment, GenericLiteLLMParams, LiteLLM_Params, ModelInfo
from litellm.types.utils import CacheCreationTokenDetails, ModelResponse, PromptTokensDetailsWrapper, Usage
_MODEL: Final = "claude-sonnet-5"
@ -50,9 +51,12 @@ def _body() -> dict[str, JsonValue]:
"model": _MODEL,
"system": "Keep this context",
"tools": [{"name": "lookup", "input_schema": {"type": "object"}}],
"messages": [{"role": "user", "content": [
{"type": "text", "text": "A cacheable prefix", "cache_control": {"type": "ephemeral"}}
]}],
"messages": [
{
"role": "user",
"content": [{"type": "text", "text": "A cacheable prefix", "cache_control": {"type": "ephemeral"}}],
}
],
}
@ -88,7 +92,8 @@ async def test_observer_records_only_version_supported_by_token_counter(version:
"httpx_response": httpx.Response(200, request=wire),
"first_api_call_start_time": datetime.fromtimestamp(1000.0),
"standard_logging_object": {
"status": "success", "model_id": _DEPLOYMENT,
"status": "success",
"model_id": _DEPLOYMENT,
"metadata": {"user_api_key_hash": _CALLER},
},
},
@ -104,13 +109,16 @@ async def test_observer_records_only_version_supported_by_token_counter(version:
assert await lookup(cache, other_scope, prefix, now=1010.0) is None
@pytest.mark.parametrize("headers, supported", [
({}, True),
({"Anthropic-Version": DEFAULT_ANTHROPIC_API_VERSION}, True),
({"anthropic-version": "2099-01-01"}, False),
({"Anthropic-Beta": ""}, False),
({"anthropic-beta": "future-feature"}, False),
])
@pytest.mark.parametrize(
"headers, supported",
[
({}, True),
({"Anthropic-Version": DEFAULT_ANTHROPIC_API_VERSION}, True),
({"anthropic-version": "2099-01-01"}, False),
({"Anthropic-Beta": ""}, False),
({"anthropic-beta": "future-feature"}, False),
],
)
def test_prediction_header_eligibility(headers: Mapping[str, str], supported: bool) -> None:
assert supported_prediction_headers(headers) is supported
@ -150,49 +158,89 @@ async def test_provider_count_uses_same_version_and_preserves_native_input(
@pytest.mark.parametrize("source", ["static", "database"])
@pytest.mark.asyncio
async def test_environment_credential_matches_native_count_and_observed_scope(
source: str, monkeypatch: pytest.MonkeyPatch,
source: str,
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setenv("LIT7658_PROVIDER_KEY", _KEY)
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
monkeypatch.setattr(litellm, "in_memory_llm_clients_cache", LLMClientCache())
params: Final = {
"model": f"anthropic/{_MODEL}", "api_key": "os.environ/LIT7658_PROVIDER_KEY",
"model": f"anthropic/{_MODEL}",
"api_key": "os.environ/LIT7658_PROVIDER_KEY",
"api_base": "https://api.anthropic.com",
}
router: Final = litellm.Router(model_list=[{
"model_name": "test-native", "litellm_params": dict(params), "model_info": {"id": _DEPLOYMENT},
}] if source == "static" else [], num_retries=0)
router: Final = litellm.Router(
model_list=[
{
"model_name": "test-native",
"litellm_params": dict(params),
"model_info": {"id": _DEPLOYMENT},
}
]
if source == "static"
else [],
num_retries=0,
)
if source == "database":
monkeypatch.setattr(proxy_server, "llm_router", router)
assert proxy_server.ProxyConfig()._add_deployment([SimpleNamespace(
model_id=_DEPLOYMENT, model_name="test-native", model_info={}, litellm_params=dict(params),
)]) == 1
assert (
proxy_server.ProxyConfig()._add_deployment(
[
SimpleNamespace(
model_id=_DEPLOYMENT,
model_name="test-native",
model_info={},
litellm_params=dict(params),
)
]
)
== 1
)
deployment: Final = router.get_deployment(_DEPLOYMENT)
assert deployment is not None
target: Final = resolve_prediction_target(deployment.litellm_params)
assert isinstance(target, NativePredictionTarget)
body: Final = _body()
with respx.mock() as upstream:
native: Final = upstream.post("https://api.anthropic.com/v1/messages").respond(200, json={
"id": "msg_test", "type": "message", "role": "assistant", "model": _MODEL,
"content": [{"type": "text", "text": "Hello"}], "stop_reason": "end_turn", "stop_sequence": None,
"usage": {"input_tokens": 11, "output_tokens": 1, "cache_read_input_tokens": 300},
})
native: Final = upstream.post("https://api.anthropic.com/v1/messages").respond(
200,
json={
"id": "msg_test",
"type": "message",
"role": "assistant",
"model": _MODEL,
"content": [{"type": "text", "text": "Hello"}],
"stop_reason": "end_turn",
"stop_sequence": None,
"usage": {"input_tokens": 11, "output_tokens": 1, "cache_read_input_tokens": 300},
},
)
counter: Final = upstream.post("https://api.anthropic.com/v1/messages/count_tokens").respond(
200, json={"input_tokens": 311},
200,
json={"input_tokens": 311},
)
await router.aanthropic_messages(
model="test-native", max_tokens=1, **{key: value for key, value in body.items() if key != "model"},
model="test-native",
max_tokens=1,
**{key: value for key, value in body.items() if key != "model"},
)
assert await count_prompt_tokens(target.model, target.api_key, body) == 311
assert native.call_count == counter.call_count == 1
assert native.calls.last.request.headers["x-api-key"] == counter.calls.last.request.headers["x-api-key"] == _KEY
observed: Final = parse_observed_cache(native.calls.last.request, ModelResponse(
model=_MODEL, usage=Usage(
prompt_tokens=311, completion_tokens=1, total_tokens=312,
prompt_tokens_details=PromptTokensDetailsWrapper(cached_tokens=300),
observed: Final = parse_observed_cache(
native.calls.last.request,
ModelResponse(
model=_MODEL,
usage=Usage(
prompt_tokens=311,
completion_tokens=1,
total_tokens=312,
prompt_tokens_details=PromptTokensDetailsWrapper(cached_tokens=300),
),
),
), _CALLER, _DEPLOYMENT)
_CALLER,
_DEPLOYMENT,
)
assert observed is not None
assert observed.scope == cache_scope(_CALLER, _DEPLOYMENT, target.api_key, target.model)
@ -200,15 +248,26 @@ async def test_environment_credential_matches_native_count_and_observed_scope(
@pytest.mark.parametrize("inline_key", [None, _KEY])
@pytest.mark.asyncio
async def test_named_credential_is_explicitly_unsupported_before_count(
inline_key: str | None, monkeypatch: pytest.MonkeyPatch,
inline_key: str | None,
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(litellm, "credential_list", [CredentialItem(
credential_name="test-named", credential_info={}, credential_values={"api_key": "test-named-provider-key"},
)])
monkeypatch.setattr(
litellm,
"credential_list",
[
CredentialItem(
credential_name="test-named",
credential_info={},
credential_values={"api_key": "test-named-provider-key"},
)
],
)
deployment: Final = Deployment(
model_name="test-native",
litellm_params=LiteLLM_Params(
model=f"anthropic/{_MODEL}", api_key=inline_key, litellm_credential_name="test-named",
model=f"anthropic/{_MODEL}",
api_key=inline_key,
litellm_credential_name="test-named",
),
model_info=ModelInfo(id=_DEPLOYMENT),
)
@ -232,8 +291,7 @@ def _cache_plan(body: Mapping[str, JsonValue]) -> PromptCachePlan:
def _text(text: str, ttl: str | None = None) -> dict[str, JsonValue]:
return {"type": "text", "text": text,
**({"cache_control": {"type": "ephemeral", "ttl": ttl}} if ttl else {})}
return {"type": "text", "text": text, **({"cache_control": {"type": "ephemeral", "ttl": ttl}} if ttl else {})}
def _prompt(*blocks: dict[str, JsonValue], role: str = "user", **options: JsonValue) -> dict[str, JsonValue]:
@ -251,16 +309,19 @@ def test_public_predictor_preserves_string_message_policy(text: str, supported:
def test_cache_plan_preserves_hierarchical_prefixes_and_public_policy() -> None:
body: Final = _prompt(
_text("First turn", "5m"), system=[_text("Stable instructions", "1h")],
tools=[{"name": "lookup", "input_schema": {"type": "object"},
"cache_control": {"type": "ephemeral", "ttl": "1h"}}],
_text("First turn", "5m"),
system=[_text("Stable instructions", "1h")],
tools=[
{"name": "lookup", "input_schema": {"type": "object"}, "cache_control": {"type": "ephemeral", "ttl": "1h"}}
],
)
plan: Final = _cache_plan(body)
changed: Final = _cache_plan({**body, "system": [_text("Changed instructions", "1h")]})
assert tuple(marker.ttl_seconds for marker in plan.breakpoints) == (3600, 3600, 300)
assert plan.breakpoints[0].fingerprint == changed.breakpoints[0].fingerprint
assert all(left.fingerprint != right.fingerprint for left, right
in zip(plan.breakpoints[1:], changed.breakpoints[1:]))
assert all(
left.fingerprint != right.fingerprint for left, right in zip(plan.breakpoints[1:], changed.breakpoints[1:])
)
assert plan.breakpoints[0].prefix_body == {
"tools": [{"name": "lookup", "input_schema": {"type": "object"}}],
"messages": [],
@ -268,51 +329,80 @@ def test_cache_plan_preserves_hierarchical_prefixes_and_public_policy() -> None:
assert parse_prompt(body) is None
@pytest.mark.parametrize("kind, added, matches", [
("text", 19, True), ("text", 20, False), ("tool_use", 30, True),
("tool_result", 30, True),
])
@pytest.mark.parametrize(
"kind, added, matches",
[
("text", 19, True),
("text", 20, False),
("tool_use", 30, True),
("tool_result", 30, True),
],
)
def test_cache_plan_lookback_counts_native_positions(
kind: str, added: int, matches: bool,
kind: str,
added: int,
matches: bool,
) -> None:
previous: Final = _cache_plan(_body())
appended: Final[list[dict[str, JsonValue]]] = [
{"type": "tool_use", "id": f"tool_{index}", "name": "lookup", "input": {}}
if kind == "tool_use" else
{"type": "tool_result", "tool_use_id": f"tool_{index}", "content": "done"}
if kind == "tool_result" else
{"type": "text", "text": f"Added {index}"}
if kind == "tool_use"
else {"type": "tool_result", "tool_use_id": f"tool_{index}", "content": "done"}
if kind == "tool_result"
else {"type": "text", "text": f"Added {index}"}
for index in range(added)
]
current: Final = _cache_plan({**_body(), **_prompt(
_text("A cacheable prefix"), *appended[:-1],
{**appended[-1], "cache_control": {"type": "ephemeral"}},
)})
assert (previous.breakpoints[0].fingerprint
in current.breakpoints[0].lookback_fingerprints) is matches
current: Final = _cache_plan(
{
**_body(),
**_prompt(
_text("A cacheable prefix"),
*appended[:-1],
{**appended[-1], "cache_control": {"type": "ephemeral"}},
),
}
)
assert (previous.breakpoints[0].fingerprint in current.breakpoints[0].lookback_fingerprints) is matches
@pytest.mark.parametrize("change, same_prefix, same_content", [
("tool_order", False, False), ("effort", False, False),
("standard_speed", True, True), ("ttl", False, True),
])
@pytest.mark.parametrize(
"change, same_prefix, same_content",
[
("tool_order", False, False),
("effort", False, False),
("standard_speed", True, True),
("ttl", False, True),
],
)
def test_cache_plan_identity_respects_settings_and_preserves_content(
change: str, same_prefix: bool, same_content: bool,
change: str,
same_prefix: bool,
same_content: bool,
) -> None:
tool_input: Final[dict[str, JsonValue]] = {"a": 1, "b": 2, "cache_control": {"ttl": "user-data"}}
block: Final[dict[str, JsonValue]] = {
"type": "tool_use", "id": "tool_1", "name": "lookup", "input": tool_input,
"type": "tool_use",
"id": "tool_1",
"name": "lookup",
"input": tool_input,
"cache_control": {"type": "ephemeral", "ttl": "5m"},
}
changed_block: Final = (
{**block, "input": dict(reversed(tool_input.items()))} if change == "tool_order" else
{**block, "cache_control": {"type": "ephemeral", "ttl": "1h"}} if change == "ttl" else block
{**block, "input": dict(reversed(tool_input.items()))}
if change == "tool_order"
else {**block, "cache_control": {"type": "ephemeral", "ttl": "1h"}}
if change == "ttl"
else block
)
before: Final = _cache_plan(_prompt(block, role="assistant", output_config={"effort": "low"})).breakpoints[0]
after: Final = _cache_plan(_prompt(
changed_block, role="assistant", output_config={"effort": "high" if change == "effort" else "low"},
**({"speed": "standard"} if change == "standard_speed" else {}),
)).breakpoints[0]
after: Final = _cache_plan(
_prompt(
changed_block,
role="assistant",
output_config={"effort": "high" if change == "effort" else "low"},
**({"speed": "standard"} if change == "standard_speed" else {}),
)
).breakpoints[0]
assert (before.fingerprint == after.fingerprint) is same_prefix
assert (before.fingerprint in after.lookback_fingerprints) is same_prefix
assert (before.content_fingerprint == after.content_fingerprint) is same_content
@ -323,30 +413,41 @@ def test_cache_plan_identity_respects_settings_and_preserves_content(
def test_cache_plan_automatic_cache_and_thinking_use_last_cacheable_block() -> None:
body: Final = _prompt(
_text("A stable answer"), {"type": "thinking", "thinking": "Thinking", "signature": "signature"},
role="assistant", thinking={"type": "adaptive"}, cache_control={"type": "ephemeral", "ttl": "1h"},
_text("A stable answer"),
{"type": "thinking", "thinking": "Thinking", "signature": "signature"},
role="assistant",
thinking={"type": "adaptive"},
cache_control={"type": "ephemeral", "ttl": "1h"},
)
plan: Final = _cache_plan(body)
assert len(plan.breakpoints) == 1
assert plan.breakpoints[0].ttl_seconds == 3600
assert plan.breakpoints[0].prefix_body == {
"thinking": {"type": "adaptive"},
"messages": [{"role": "assistant", "content": [
{"type": "text", "text": "A stable answer"},
]}],
"messages": [
{
"role": "assistant",
"content": [
{"type": "text", "text": "A stable answer"},
],
}
],
}
assert parse_prompt(body) is None
@pytest.mark.parametrize("body, reason", [
(_prompt({"type": "image"}), "unsupported_prompt_shape"),
({**_body(), "unknown_native_setting": True}, "unsupported_prompt_shape"),
({**_body(), "cache_control": {"type": "ephemeral", "ttl": "1h"}},
"conflicting_cache_ttl"),
(_prompt(_text("five", "5m"), _text("hour", "1h")), "invalid_cache_ttl_order"),
])
@pytest.mark.parametrize(
"body, reason",
[
(_prompt({"type": "image"}), "unsupported_prompt_shape"),
({**_body(), "unknown_native_setting": True}, "unsupported_prompt_shape"),
({**_body(), "cache_control": {"type": "ephemeral", "ttl": "1h"}}, "conflicting_cache_ttl"),
(_prompt(_text("five", "5m"), _text("hour", "1h")), "invalid_cache_ttl_order"),
],
)
def test_cache_plan_unsupported_is_explicit(
body: Mapping[str, JsonValue], reason: str,
body: Mapping[str, JsonValue],
reason: str,
) -> None:
result: Final = parse_cache_plan(body)
assert isinstance(result, UnsupportedCachePlan)
@ -354,30 +455,46 @@ def test_cache_plan_unsupported_is_explicit(
@pytest.mark.asyncio
@pytest.mark.parametrize("model, counts, reason", [
(None, (100, 150, 200), None),
(None, (100, 201, 200), "inconsistent_prefix_token_count"),
(None, (151, 150, 200), "inconsistent_prefix_token_count"),
(None, (None, 150, 200), "token_count_unavailable"),
("claude-opus-5", (100, 150, 200), None),
("claude-sonnet-5", (100, 150, 200), None),
("declared-cache-model", (100, 150, 200), None),
("unknown-cache-model", (100, 150, 200), "unsupported_thinking_cache_semantics"),
("claude-haiku-4-5", (100, 150, 200), "unsupported_thinking_cache_semantics"),
("claude-sonnet-4-5", (100, 150, 200), "unsupported_thinking_cache_semantics"),
])
@pytest.mark.parametrize(
"model, counts, reason",
[
(None, (100, 150, 200), None),
(None, (100, 201, 200), "inconsistent_prefix_token_count"),
(None, (151, 150, 200), "inconsistent_prefix_token_count"),
(None, (None, 150, 200), "token_count_unavailable"),
("claude-opus-5", (100, 150, 200), None),
("claude-sonnet-5", (100, 150, 200), None),
("declared-cache-model", (100, 150, 200), None),
("unknown-cache-model", (100, 150, 200), "unsupported_thinking_cache_semantics"),
("claude-haiku-4-5", (100, 150, 200), "unsupported_thinking_cache_semantics"),
("claude-sonnet-4-5", (100, 150, 200), "unsupported_thinking_cache_semantics"),
],
)
async def test_cache_plan_count_conserves_total_and_rejects_unknown(
model: str | None, counts: tuple[int | None, int | None, int], reason: str | None,
model: str | None,
counts: tuple[int | None, int | None, int],
reason: str | None,
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setitem(litellm.model_cost, "declared-cache-model", {
"litellm_provider": "anthropic", "mode": "chat", "supports_thinking_cache_preservation": True,
})
plan: Final = _cache_plan(_prompt(
{"type": "thinking", "thinking": "Retained thought", "signature": "signature"}
if model else _text("first", "5m"),
_text("second", "5m"), _text("uncached"), role="assistant" if model else "user",
))
monkeypatch.setitem(
litellm.model_cost,
"declared-cache-model",
{
"litellm_provider": "anthropic",
"mode": "chat",
"supports_thinking_cache_preservation": True,
},
)
plan: Final = _cache_plan(
_prompt(
{"type": "thinking", "thinking": "Retained thought", "signature": "signature"}
if model
else _text("first", "5m"),
_text("second", "5m"),
_text("uncached"),
role="assistant" if model else "user",
)
)
async def count(model: str, api_key: str, body: Mapping[str, JsonValue]) -> int | None:
assert reason != "unsupported_thinking_cache_semantics", "Unverified thinking retention must skip counting"
@ -399,12 +516,15 @@ async def test_cache_plan_count_conserves_total_and_rejects_unknown(
@pytest.mark.parametrize("section", ["system", "tools"])
@pytest.mark.parametrize("rejects_prefix", (False, True))
async def test_native_count_preserves_settings_and_requires_every_prefix(
section: str, rejects_prefix: bool, monkeypatch: pytest.MonkeyPatch,
section: str,
rejects_prefix: bool,
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
monkeypatch.setattr(litellm, "in_memory_llm_clients_cache", LLMClientCache())
params: Final = LiteLLM_Params(
model=f"anthropic/{_MODEL}", api_key=_KEY,
model=f"anthropic/{_MODEL}",
api_key=_KEY,
api_base="https://gateway.example/v1/messages",
)
target: Final = resolve_baseline_prediction_target(params)
@ -413,19 +533,38 @@ async def test_native_count_preserves_settings_and_requires_every_prefix(
assert not isinstance(resolve_prediction_target(params), NativePredictionTarget)
body: Final = _body()
marker: Final[dict[str, JsonValue]] = {"type": "ephemeral", "ttl": "1h"}
body[section] = ([_text("A cached system", "1h")] if section == "system" else [{
"name": "lookup", "input_schema": {"type": "object"}, "cache_control": marker,
}])
plan: Final = _cache_plan({**body, **_prompt(
_text("A later prefix", "5m"), _text("An uncached suffix"),
thinking={"type": "adaptive"}, tool_choice={"type": "auto"}, output_config={"effort": "high"},
)})
body[section] = (
[_text("A cached system", "1h")]
if section == "system"
else [
{
"name": "lookup",
"input_schema": {"type": "object"},
"cache_control": marker,
}
]
)
plan: Final = _cache_plan(
{
**body,
**_prompt(
_text("A later prefix", "5m"),
_text("An uncached suffix"),
thinking={"type": "adaptive"},
tool_choice={"type": "auto"},
output_config={"effort": "high"},
),
}
)
assert len(plan.breakpoints) == 2
assert plan.breakpoints[0].prefix_body["messages"] == []
async def count(model: str, api_key: str, body: Mapping[str, JsonValue]) -> int | None:
return await count_prompt_tokens(
model, api_key, {**body, "max_tokens": 100}, api_base=target.api_base,
model,
api_key,
{**body, "max_tokens": 100},
api_base=target.api_base,
)
with respx.mock(assert_all_called=False) as upstream:
@ -434,12 +573,16 @@ async def test_native_count_preserves_settings_and_requires_every_prefix(
upstream.post(endpoint, json={**body, "model": _MODEL}).respond(
400 if rejects_prefix and index == 1 else 200,
json={"detail": {"error": "messages parameter is required"}}
if rejects_prefix and index == 1 else {"input_tokens": tokens},
if rejects_prefix and index == 1
else {"input_tokens": tokens},
)
for index, (body, tokens) in enumerate(
(
(plan.full_body, 6000),
(plan.breakpoints[0].prefix_body, 5000),
(plan.breakpoints[1].prefix_body, 5800),
)
)
for index, (body, tokens) in enumerate((
(plan.full_body, 6000), (plan.breakpoints[0].prefix_body, 5000),
(plan.breakpoints[1].prefix_body, 5800),
))
)
unexpected: Final = upstream.post(endpoint).respond(200, json={"input_tokens": 1})
result: Final = await count_cache_plan(target.model, target.api_key, plan, count)
@ -452,3 +595,32 @@ async def test_native_count_preserves_settings_and_requires_every_prefix(
assert tuple(marker.prefix_tokens for marker in result.breakpoints) == (5000, 5800)
assert tuple(route.call_count for route in routes) == (1, 1, 1)
assert unexpected.call_count == 0
def test_native_baseline_body_matches_the_sent_body_when_drop_params_meets_an_auto_added_beta(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setattr(litellm, "drop_params", True)
thinking: Final = {"type": "adaptive", "block_binding": {"prefix_mismatch_behavior": "error"}}
messages: Final = [{"role": "user", "content": "Say hi in French"}]
predicted: Final = prepare_native_baseline_body(
{"model": "claude-opus-4-8", "messages": messages, "max_tokens": 64, "thinking": thinking}, "claude-opus-4-8"
)
config: Final = AnthropicMessagesConfig()
headers, _ = config.validate_anthropic_messages_environment(
headers={},
model="claude-opus-4-8",
messages=[dict(message) for message in messages],
optional_params={"max_tokens": 64, "thinking": dict(thinking)},
litellm_params={},
api_key="sk-ant-test",
)
sent: Final = config.transform_anthropic_messages_request(
model="claude-opus-4-8",
messages=[dict(message) for message in messages],
anthropic_messages_optional_request_params={"max_tokens": 64, "thinking": dict(thinking)},
litellm_params=GenericLiteLLMParams(),
headers=headers,
)
assert predicted is not None
assert predicted["thinking"] == sent["thinking"] == thinking

69
tests/unit/llms/azure_ai/claude/test_azure_anthropic_handler.py
View file

@ -6,6 +6,7 @@ sys.path.insert(
)
import json
from typing import Final
from unittest.mock import MagicMock, patch
import pytest
@ -272,3 +273,71 @@ def test_completion_surfaces_the_status_and_headers_of_a_rejected_request():
assert rejected.value.status_code == 400
assert rejected.value.litellm_response_headers["x-request-id"] == "req-1"
assert "max_tokens too large" in str(rejected.value)
@respx.mock
@pytest.mark.parametrize("drop_params", [True, False])
def test_completion_adds_the_binding_beta_so_block_binding_is_kept(
drop_params: bool, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(litellm, "drop_params", False)
route: Final = respx.post("https://example-resource.services.ai.azure.com/anthropic/v1/messages").mock(
return_value=Response(
200,
json={
"id": "msg_bound",
"type": "message",
"role": "assistant",
"model": "claude-sonnet-4-6",
"content": [{"type": "text", "text": "Hello"}],
"stop_reason": "end_turn",
"stop_sequence": None,
"usage": {"input_tokens": 1, "output_tokens": 1},
},
)
)
block_binding: Final = {"prefix_mismatch_behavior": "error"}
litellm.completion(
model="azure_ai/claude-sonnet-4-6",
messages=[{"role": "user", "content": "hi"}],
api_base="https://example-resource.services.ai.azure.com/anthropic",
api_key="azure-test-key",
thinking={"type": "enabled", "budget_tokens": 1024, "block_binding": block_binding},
drop_params=drop_params,
)
sent: Final = json.loads(route.calls.last.request.content)
assert sent["thinking"] == {"type": "enabled", "budget_tokens": 1024, "block_binding": block_binding}
assert "thinking-binding-controls-2026-08-01" in route.calls.last.request.headers["anthropic-beta"].split(",")
@respx.mock
def test_completion_sends_the_binding_beta_next_to_the_callers_own_betas(local_beta_headers_config: None) -> None:
route: Final = respx.post("https://example-resource.services.ai.azure.com/anthropic/v1/messages").mock(
return_value=Response(
200,
json={
"id": "msg_bound",
"type": "message",
"role": "assistant",
"model": "claude-sonnet-4-6",
"content": [{"type": "text", "text": "Hello"}],
"stop_reason": "end_turn",
"stop_sequence": None,
"usage": {"input_tokens": 1, "output_tokens": 1},
},
)
)
litellm.completion(
model="azure_ai/claude-sonnet-4-6",
messages=[{"role": "user", "content": "hi"}],
api_base="https://example-resource.services.ai.azure.com/anthropic",
api_key="azure-test-key",
thinking={"type": "enabled", "budget_tokens": 1024, "block_binding": {"prefix_mismatch_behavior": "error"}},
extra_headers={"anthropic-beta": "context-1m-2025-08-07"},
)
sent_betas: Final = set(route.calls.last.request.headers["anthropic-beta"].split(","))
assert {"context-1m-2025-08-07", "thinking-binding-controls-2026-08-01"} <= sent_betas

22
tests/unit/llms/bedrock/chat/invoke_transformations/test_bedrock_chat_invoke_transformations_anthropic_claude_transformation.py
View file

@ -2008,3 +2008,25 @@ def test_invoke_uses_opt_in_native_output_is_false_when_native_output_did_not_co
assert not AmazonAnthropicClaudeConfig().uses_opt_in_native_output(
model, optional_params, {"model_info": model_info}
)
@pytest.mark.usefixtures("local_model_cost_map", "local_beta_headers_config")
@pytest.mark.parametrize("drop_params", [True, False])
def test_bedrock_chat_invoke_adds_the_binding_beta_so_block_binding_is_kept(
drop_params: bool, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(litellm, "drop_params", False)
block_binding: Final = {"prefix_mismatch_behavior": "error"}
result: Final = AmazonAnthropicClaudeConfig().transform_request(
model="us.anthropic.claude-sonnet-4-6",
messages=[{"role": "user", "content": "Hello"}],
optional_params={
"max_tokens": 2048,
"thinking": {"type": "enabled", "budget_tokens": 1024, "block_binding": block_binding},
},
litellm_params={"drop_params": drop_params},
headers={},
)
assert result["thinking"] == {"type": "enabled", "budget_tokens": 1024, "block_binding": block_binding}
assert "thinking-binding-controls-2026-08-01" in result["anthropic_beta"]

126
tests/unit/llms/bedrock/chat/test_converse_transformation.py
View file

@ -24,9 +24,9 @@ from litellm.llms.anthropic.chat.transformation import (
NATIVE_OUTPUT_MAX_OPTIONAL_PARAMETERS,
NATIVE_OUTPUT_MAX_UNION_PARAMETERS,
)
from litellm.llms.bedrock.common_utils import BEDROCK_NATIVE_OUTPUT_MAX_PROPERTIES
from litellm.llms.bedrock.base_aws_llm import BaseAWSLLM
from litellm.llms.bedrock.chat.converse_transformation import AmazonConverseConfig
from litellm.llms.bedrock.common_utils import BEDROCK_NATIVE_OUTPUT_MAX_PROPERTIES
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler, HTTPHandler
from litellm.types.llms.bedrock import ContentBlock, ConverseTokenUsageBlock, RequestObject
from litellm.types.llms.openai import AllMessageValues
@ -5355,7 +5355,6 @@ def test_recursive_output_config_format_follows_the_profile_native_key(
assert ("outputConfig" in request) is forwarded
def test_sonnet_4_6_json_schema_still_uses_native_output_config(monkeypatch):
monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True")
monkeypatch.setattr(litellm, "model_cost", litellm.get_model_cost_map(url=""))
@ -10672,7 +10671,10 @@ def test_uses_opt_in_native_output_reads_the_base_model_of_an_application_infere
_CALLER_OUTPUT_CONFIG_WITH_A_SCHEMA_OBJECT: Final = {
"textFormat": {"type": "json_schema", "structure": {"jsonSchema": {"name": "capital", "schema": {"type": "object"}}}}
"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"
@ -10796,3 +10798,121 @@ 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[str, str] | None, body_betas: tuple[str, ...] = (), drop_params: bool = True
) -> dict[str, object]:
thinking: Final = {
"type": "adaptive",
"block_binding": {"prefix_mismatch_behavior": "error"},
"display": "updates",
}
data: Final = AmazonConverseConfig()._transform_request_helper(
model="anthropic.claude-opus-4-8-v1:0",
system_content_blocks=[],
optional_params={"thinking": thinking, **({"anthropic_beta": list(body_betas)} if body_betas else {})},
messages=[{"role": "user", "content": "hi"}],
headers=headers,
drop_params=drop_params,
)
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: pytest.MonkeyPatch,
) -> None:
monkeypatch.setenv("LITELLM_LOCAL_ANTHROPIC_BETA_HEADERS", "true")
fields: Final = _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_adds_the_thinking_betas_so_gated_thinking_fields_are_kept(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setenv("LITELLM_LOCAL_ANTHROPIC_BETA_HEADERS", "true")
fields: Final = _converse_bound_thinking_request(None)
assert fields["thinking"] == {
"type": "adaptive",
"block_binding": {"prefix_mismatch_behavior": "error"},
"display": "updates",
}
assert sorted(fields["anthropic_beta"]) == [
"thinking-binding-controls-2026-08-01",
"thinking-display-updates-2026-08-18",
]
def test_converse_keeps_the_callers_body_betas_next_to_the_added_thinking_betas(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setenv("LITELLM_LOCAL_ANTHROPIC_BETA_HEADERS", "true")
fields: Final = _converse_bound_thinking_request(None, body_betas=("context-1m-2025-08-07",))
assert sorted(fields["anthropic_beta"]) == [
"context-1m-2025-08-07",
"thinking-binding-controls-2026-08-01",
"thinking-display-updates-2026-08-18",
]
def test_converse_sends_beta_gated_thinking_fields_as_is_without_drop_params(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("LITELLM_LOCAL_ANTHROPIC_BETA_HEADERS", "true")
monkeypatch.setattr(litellm, "drop_params", False)
fields: Final = _converse_bound_thinking_request(None, drop_params=False)
assert fields["thinking"] == {
"type": "adaptive",
"block_binding": {"prefix_mismatch_behavior": "error"},
"display": "updates",
}
def test_converse_keeps_beta_gated_thinking_fields_the_body_anthropic_beta_carries(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setenv("LITELLM_LOCAL_ANTHROPIC_BETA_HEADERS", "true")
fields: Final = _converse_bound_thinking_request(
None, body_betas=("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_body_betas_the_bedrock_converse_row_rejects_when_adding_thinking_betas(
local_beta_headers_config: None,
) -> None:
fields: Final = _converse_bound_thinking_request(
None, body_betas=("per-turn-control-2026-07-01", "context-1m-2025-08-07")
)
assert sorted(fields["anthropic_beta"]) == [
"context-1m-2025-08-07",
"thinking-binding-controls-2026-08-01",
"thinking-display-updates-2026-08-18",
]
def test_converse_sends_only_the_header_betas_when_header_and_body_both_carry_betas(
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.setenv("LITELLM_LOCAL_ANTHROPIC_BETA_HEADERS", "true")
fields: Final = _converse_bound_thinking_request(
{"anthropic-beta": "context-1m-2025-08-07"}, body_betas=("not-a-real-beta-2099",)
)
assert sorted(fields["anthropic_beta"]) == [
"context-1m-2025-08-07",
"thinking-binding-controls-2026-08-01",
"thinking-display-updates-2026-08-18",
]

94
tests/unit/llms/bedrock/messages/invoke_transformations/test_anthropic_claude_transformation.py
View file

@ -16,7 +16,7 @@ import httpx
import pytest
import requests_mock as requests_mock_lib
import respx
from pydantic import TypeAdapter
from pydantic import BaseModel, ConfigDict, TypeAdapter
import litellm
from litellm.constants import (
@ -4200,6 +4200,98 @@ def test_bedrock_messages_without_guardrail_config_sends_no_guardrail_headers()
assert not any(name.startswith("X-Amzn-Bedrock-Guardrail") for name in headers)
@pytest.mark.usefixtures("local_model_cost_map", "local_beta_headers_config")
def test_bedrock_messages_keeps_mid_conversation_output_config_its_auto_added_beta_gates() -> None:
"""Invoke learns its body betas after the shared transform, so the gated field it supports must survive."""
from litellm.types.router import GenericLiteLLMParams
messages: Final = [
{"role": "user", "content": [{"type": "text", "text": "Hello"}]},
{"role": "assistant", "content": [{"type": "text", "text": "Hi"}]},
{"role": "system", "content": "Answer briefly", "output_config": {"effort": "high"}},
{"role": "user", "content": [{"type": "text", "text": "Go on"}]},
]
result: Final = AmazonAnthropicClaudeMessagesConfig().transform_anthropic_messages_request(
model="anthropic.claude-opus-4-8-v1:0",
messages=messages,
anthropic_messages_optional_request_params={"max_tokens": 4096},
litellm_params=GenericLiteLLMParams(drop_params=True),
headers={},
)
assert "mid-conversation-output-config-2026-07-01" in result["anthropic_beta"]
assert {"role": "system", "content": "Answer briefly", "output_config": {"effort": "high"}} in result["messages"]
class _SentMessage(BaseModel):
model_config = ConfigDict(frozen=True)
role: str
content: list[dict[str, object]]
_SENT_MESSAGES: Final = TypeAdapter(list[_SentMessage])
@pytest.mark.usefixtures("local_model_cost_map", "local_beta_headers_config")
@pytest.mark.parametrize("drop_params", [True, False])
def test_bedrock_messages_drops_tool_changes_from_a_system_turn_an_older_model_gets_as_a_user_turn(
drop_params: bool, monkeypatch: pytest.MonkeyPatch
) -> None:
"""A model without mid-conversation system support sends no tool-changes beta, so under drop_params the
converted turn must not carry the blocks that beta gates, and without it they go out as sent."""
monkeypatch.setattr(litellm, "drop_params", False)
from litellm.types.router import GenericLiteLLMParams
messages: Final = [
{"role": "user", "content": [{"type": "text", "text": "Hello"}]},
{"role": "assistant", "content": [{"type": "text", "text": "Hi"}]},
{
"role": "system",
"content": [
{"type": "text", "text": "A weather tool is now available"},
{"type": "tool_addition", "tool": {"type": "tool_reference", "name": "get_weather"}},
],
},
{"role": "user", "content": [{"type": "text", "text": "Go on"}]},
]
result: Final = AmazonAnthropicClaudeMessagesConfig().transform_anthropic_messages_request(
model="anthropic.claude-3-7-sonnet-20250219-v1:0",
messages=messages,
anthropic_messages_optional_request_params={"max_tokens": 4096},
litellm_params=GenericLiteLLMParams(drop_params=drop_params),
headers={},
)
sent: Final = _SENT_MESSAGES.validate_python(result["messages"])
blocks: Final = [
block for message in sent for block in message.content
] # comprehension-ok: flattening the sent content blocks
assert "mid-conversation-tool-changes-2026-07-01" not in result.get("anthropic_beta", [])
assert any(block["type"] == "tool_addition" for block in blocks) is not drop_params
assert {"type": "text", "text": "A weather tool is now available"} in blocks
assert all(message.role != "system" for message in sent)
@pytest.mark.usefixtures("local_model_cost_map", "local_beta_headers_config")
@pytest.mark.parametrize("drop_params", (True, False))
def test_bedrock_messages_sends_the_binding_beta_with_block_binding_whatever_drop_params(drop_params: bool) -> None:
from litellm.types.router import GenericLiteLLMParams
result: Final = AmazonAnthropicClaudeMessagesConfig().transform_anthropic_messages_request(
model="us.anthropic.claude-sonnet-4-6",
messages=[{"role": "user", "content": "Reply with OK"}],
anthropic_messages_optional_request_params={
"max_tokens": 2048,
"thinking": {"type": "enabled", "budget_tokens": 1024, "block_binding": BLOCK_BINDING},
},
litellm_params=GenericLiteLLMParams(drop_params=drop_params),
headers={},
)
assert result["thinking"]["block_binding"] == BLOCK_BINDING
assert "thinking-binding-controls-2026-08-01" in result["anthropic_beta"]
@dataclass(slots=True)
class BedrockDownAnthropicUp:
anthropic_bodies: tuple[dict[str, object], ...] = ()

20
tests/unit/llms/bedrock/test_claude_platform_provider.py
View file

@ -207,6 +207,26 @@ def test_claude_platform_api_key_auth_sets_workspace_and_key_headers():
assert headers["anthropic-beta"] == "skills-2025-10-02"
def test_claude_platform_chat_sends_the_binding_beta_when_thinking_carries_block_binding():
from litellm.llms.bedrock.claude_platform.transformation import (
BedrockClaudePlatformConfig,
)
headers: Final = BedrockClaudePlatformConfig().validate_environment(
api_key="fake-platform-key",
headers={},
model="claude-opus-5-5",
messages=[{"role": "user", "content": "hello"}],
optional_params={
"workspace_id": "wrkspc_test",
"thinking": {"type": "adaptive", "block_binding": {"prefix_mismatch_behavior": "error"}},
},
litellm_params={},
)
assert "thinking-binding-controls-2026-08-01" in headers["anthropic-beta"].split(",")
def test_claude_platform_does_not_use_standard_anthropic_api_key(monkeypatch):
from litellm.llms.bedrock.claude_platform.transformation import (
BedrockClaudePlatformConfig,

35
tests/unit/llms/databricks/chat/test_databricks_chat_transformation.py
View file

@ -2465,6 +2465,41 @@ def test_databricks_claude_drops_the_openai_tool_search_tool_under_drop_params(m
]
@pytest.mark.parametrize("drop_params", [True, False])
def test_databricks_chat_drops_beta_gated_thinking_fields_only_under_drop_params(
drop_params: bool, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(litellm, "drop_params", False)
thinking: Final = {
"type": "adaptive",
"block_binding": {"prefix_mismatch_behavior": "error"},
"display": "updates",
}
mapped: Final = DatabricksConfig().map_openai_params(
non_default_params={"thinking": thinking},
optional_params={},
model="databricks-claude-opus-5-5",
drop_params=drop_params,
)
assert mapped["thinking"] == ({"type": "adaptive"} if drop_params else thinking)
assert thinking == {
"type": "adaptive",
"block_binding": {"prefix_mismatch_behavior": "error"},
"display": "updates",
}
def test_databricks_chat_keeps_ungated_thinking_display_under_drop_params(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(litellm, "drop_params", False)
mapped: Final = DatabricksConfig().map_openai_params(
non_default_params={"thinking": {"type": "adaptive", "display": "summarized"}},
optional_params={},
model="databricks-claude-opus-5-5",
drop_params=True,
)
assert mapped["thinking"] == {"type": "adaptive", "display": "summarized"}
_JSON_TOOL_ARGUMENT_FRAGMENTS: Final = ("", "", '{"name": ', '"Ada"', ', "met', 'a": ', "{}", ', "tags": []}')
_RECORD_RESPONSE_FORMAT: Final = {
"type": "json_schema",

29
tests/unit/llms/databricks/messages/test_databricks_messages_transformation.py
View file

@ -281,7 +281,7 @@ SERVER_TOOLS: Final = [
]
async def _workspace_request_body(**request_params: object) -> dict[str, object]:
async def _workspace_request(max_tokens: int = 64, **request_params: object) -> httpx.Request:
received: Final[list[httpx.Request]] = [] # mutable-ok: the fake workspace records what litellm sent
def workspace_answers(request: httpx.Request) -> httpx.Response:
@ -303,7 +303,7 @@ async def _workspace_request_body(**request_params: object) -> dict[str, object]
upstream: Final = AsyncHTTPHandler(transport=httpx.MockTransport(workspace_answers))
try:
await litellm.anthropic.messages.acreate(
max_tokens=64,
max_tokens=max_tokens,
messages=USER_TURN,
model=f"databricks/{CLAUDE_MODEL}",
api_key=DATABRICKS_API_KEY,
@ -314,7 +314,30 @@ async def _workspace_request_body(**request_params: object) -> dict[str, object]
finally:
await upstream.close()
assert len(received) == 1
return json.loads(received[0].content)
return received[0]
async def _workspace_request_body(**request_params: object) -> dict[str, object]:
return json.loads((await _workspace_request(**request_params)).content)
@pytest.mark.asyncio
@pytest.mark.parametrize("drop_params", [True, False])
async def test_bound_thinking_with_display_updates_reaches_the_workspace_with_both_betas(
drop_params: bool, monkeypatch: pytest.MonkeyPatch, local_beta_headers_config: None
) -> None:
monkeypatch.setattr(litellm, "drop_params", False)
thinking: Final = {
"type": "enabled",
"budget_tokens": 1024,
"block_binding": {"prefix_mismatch_behavior": "error"},
"display": "updates",
}
request: Final = await _workspace_request(max_tokens=4096, thinking=thinking, drop_params=drop_params)
assert json.loads(request.content)["thinking"] == thinking
assert {"thinking-binding-controls-2026-08-01", "thinking-display-updates-2026-08-18"} <= set(
request.headers["anthropic-beta"].split(",")
)
@pytest.mark.asyncio

37
tests/unit/llms/vertex_ai/vertex_ai_partner_models/anthropic/test_vertex_ai_partner_models_anthropic_messages_config.py
View file

@ -289,7 +289,7 @@ def test_no_safeguards_leaves_dangerous_tool_use_beta_header_out():
assert "dangerous-tool-use-2026-09-03" not in updated_headers.get("anthropic-beta", "")
def _validate_vertex_headers(client_headers, messages):
def _validate_vertex_headers(client_headers, messages, thinking: Mapping[str, object] | None = None):
config = VertexAIPartnerModelsAnthropicMessagesConfig()
litellm_params = {
"vertex_ai_project": "test-project",
@ -305,7 +305,7 @@ def _validate_vertex_headers(client_headers, messages):
headers=client_headers,
model="claude-opus-5-5",
messages=messages,
optional_params={"max_tokens": 64},
optional_params={"max_tokens": 64, **({"thinking": thinking} if thinking is not None else {})},
litellm_params=litellm_params,
api_base=None,
)
@ -337,6 +337,25 @@ def test_per_message_output_config_reaches_vertex_with_per_turn_control_beta(cli
assert filtered["anthropic-beta"].split(",").count("per-turn-control-2026-07-01") == 1
def test_block_binding_reaches_vertex_messages_with_its_beta_under_drop_params(local_beta_headers_config: None) -> None:
from litellm.anthropic_beta_headers_manager import update_headers_with_filtered_beta
thinking: Final = {"type": "adaptive", "block_binding": {"prefix_mismatch_behavior": "error"}}
messages: Final = [{"role": "user", "content": "Say hi in French"}]
headers: Final = update_headers_with_filtered_beta(
headers=_validate_vertex_headers({}, messages, thinking=thinking), provider="vertex_ai"
)
request: Final = VertexAIPartnerModelsAnthropicMessagesConfig().transform_anthropic_messages_request(
model="claude-opus-5-5",
messages=[dict(message) for message in messages],
anthropic_messages_optional_request_params={"max_tokens": 64, "thinking": dict(thinking)},
litellm_params=GenericLiteLLMParams(drop_params=True),
headers=headers,
)
assert "thinking-binding-controls-2026-08-01" in headers["anthropic-beta"].split(",")
assert request["thinking"] == thinking
def test_no_per_message_output_config_leaves_per_turn_control_beta_out():
headers = _validate_vertex_headers({}, [{"role": "user", "content": "Hello"}])
@ -913,6 +932,20 @@ class TestVertexAnthropicMidConversationSystem:
{"type": "text", "text": "Cite sources."},
]
@pytest.mark.parametrize("drop_params", (False, True))
def test_unsupported_model_sends_a_non_object_message_on_for_the_provider_to_reject(
self, local_model_cost_map, local_beta_headers_config, drop_params
):
messages: Final = [{"role": "user", "content": "hi"}, "oops"]
result: Final = VertexAIPartnerModelsAnthropicMessagesConfig().transform_anthropic_messages_request(
model="claude-sonnet-4-6",
messages=list(messages),
anthropic_messages_optional_request_params={"max_tokens": 64},
litellm_params=GenericLiteLLMParams(drop_params=drop_params),
headers={},
)
assert result["messages"] == messages
def test_unsupported_model_converts_mid_conversation_system_in_place(self, local_model_cost_map):
messages = [
{"role": "user", "content": "read the file"},

19
tests/unit/llms/vertex_ai/vertex_ai_partner_models/anthropic/test_vertex_ai_partner_models_anthropic_transformation.py
View file

@ -155,6 +155,25 @@ def test_vertex_ai_anthropic_web_search_header_in_completion():
), "anthropic-beta with web-search should not be present for non-Vertex requests"
def test_vertex_ai_anthropic_chat_adds_the_betas_for_bound_thinking_with_display_updates() -> None:
thinking: Final = {
"type": "adaptive",
"block_binding": {"prefix_mismatch_behavior": "error"},
"display": "updates",
}
result: Final = VertexAIAnthropicConfig().transform_request(
model="claude-opus-4-8",
messages=[{"role": "user", "content": "Hello"}],
optional_params={"thinking": thinking, "max_tokens": 100, "is_vertex_request": True},
litellm_params={},
headers={},
)
assert result["thinking"] == thinking
assert {"thinking-binding-controls-2026-08-01", "thinking-display-updates-2026-08-18"} <= set(
result["anthropic_beta"]
)
def test_vertex_ai_anthropic_context_management_compact_beta_header():
"""Test that context_management with compact adds the correct beta header for Vertex AI"""
config = VertexAIAnthropicConfig()

301
tests/unit/test_anthropic_beta_gated_fields.py Normal file
View file

@ -0,0 +1,301 @@
import logging
from typing import Final
import pytest
import litellm
from litellm.anthropic_beta_gated_fields import (
ANTHROPIC_INLINE_TOOLS_BETA_HEADER,
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_native_request,
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[str, object]:
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() -> None:
result: Final = strip_beta_gated_thinking(BOUND_THINKING, EVERY_BETA, drop_params=True)
assert result.thinking is BOUND_THINKING
assert result.stripped == ()
def test_thinking_loses_only_the_field_whose_beta_is_absent() -> None:
without_binding: Final = strip_beta_gated_thinking(
BOUND_THINKING, EVERY_BETA - {ANTHROPIC_THINKING_BINDING_CONTROLS_BETA_HEADER}, drop_params=True
)
assert without_binding.thinking == {"type": "adaptive", "display": "updates"}
assert without_binding.stripped == ("thinking.block_binding",)
without_display: Final = strip_beta_gated_thinking(
BOUND_THINKING, EVERY_BETA - {ANTHROPIC_THINKING_DISPLAY_UPDATES_BETA_HEADER}, drop_params=True
)
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() -> None:
thinking: Final = {"type": "adaptive", "display": "summarized"}
result: Final = strip_beta_gated_thinking(thinking, frozenset(), drop_params=True)
assert result.thinking is thinking
@pytest.mark.parametrize("thinking", [None, "enabled", 12, ["adaptive"]])
def test_non_mapping_thinking_passes_through(thinking: object) -> None:
result: Final = strip_beta_gated_thinking(thinking, frozenset(), drop_params=True)
assert result.thinking is thinking
assert result.stripped == ()
def test_system_message_keeps_every_gated_field_when_the_betas_are_active() -> None:
messages: Final = [{"role": "user", "content": "hi"}, _system_turn()]
result: Final = strip_beta_gated_system_messages(messages, EVERY_BETA, drop_params=True)
assert result.messages is messages
assert result.stripped == ()
def test_system_message_loses_exactly_the_fields_whose_betas_are_absent() -> None:
original: Final = _system_turn()
messages: Final = [{"role": "user", "content": "hi"}, original]
result: Final = strip_beta_gated_system_messages(
messages,
EVERY_BETA
- {
ANTHROPIC_MID_CONVERSATION_SYSTEM_CLEAR_AT_BETA_HEADER,
ANTHROPIC_MID_CONVERSATION_TOOL_CHANGES_BETA_HEADER,
ANTHROPIC_INLINE_TOOLS_BETA_HEADER,
},
drop_params=True,
)
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() -> None:
messages: Final = [{"role": "system", "output_config": {"effort": "low"}, "content": "terse"}]
assert (
strip_beta_gated_system_messages(
messages, frozenset({"per-turn-control-2026-07-01"}), drop_params=True
).messages
is messages
)
assert (
strip_beta_gated_system_messages(
messages, frozenset({ANTHROPIC_MID_CONVERSATION_OUTPUT_CONFIG_BETA_HEADER}), drop_params=True
).messages
is messages
)
assert strip_beta_gated_system_messages(messages, frozenset(), drop_params=True).messages == [
{"role": "system", "content": "terse"}
]
def test_inline_tools_beta_also_unlocks_system_tool_changes() -> None:
messages: Final = [
{
"role": "system",
"content": [{"type": "tool_removal", "tool_names": ["lookup"]}, {"type": "text", "text": "x"}],
}
]
assert (
strip_beta_gated_system_messages(
messages, frozenset({ANTHROPIC_INLINE_TOOLS_BETA_HEADER}), drop_params=True
).messages
is messages
)
assert (
strip_beta_gated_system_messages(
messages, frozenset({ANTHROPIC_MID_CONVERSATION_TOOL_CHANGES_BETA_HEADER}), drop_params=True
).messages
is messages
)
assert strip_beta_gated_system_messages(messages, frozenset(), drop_params=True).messages == [
{"role": "system", "content": [{"type": "text", "text": "x"}]}
]
def test_system_block_with_unhashable_type_is_kept_for_the_provider_to_reject() -> None:
messages: Final = [{"role": "system", "content": [{"type": [], "text": "x"}]}]
result: Final = strip_beta_gated_system_messages(messages, frozenset(), drop_params=True)
assert result.messages is messages
assert result.stripped == ()
def test_system_message_emptied_by_the_strip_is_dropped() -> None:
messages: Final = [
{"role": "user", "content": "hi"},
{"role": "system", "content": [{"type": "tool_removal", "tool_names": ["lookup"]}]},
{"role": "assistant", "content": "ok"},
]
result: Final = strip_beta_gated_system_messages(messages, frozenset(), drop_params=True)
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_emptied_trailing_system_turn_after_an_assistant_turn_is_sent_as_is() -> None:
messages: Final = [
{"role": "user", "content": "hi"},
{"role": "assistant", "content": "ok"},
{"role": "system", "content": [{"type": "tool_removal", "tool_names": ["lookup"]}]},
]
result: Final = strip_beta_gated_system_messages(messages, frozenset(), drop_params=True)
assert result.messages is messages
assert result.stripped == ()
def test_emptied_system_turn_is_dropped_when_a_user_turn_still_ends_the_conversation() -> None:
messages: Final = [
{"role": "user", "content": "hi"},
{"role": "assistant", "content": "ok"},
{"role": "system", "content": [{"type": "tool_removal", "tool_names": ["lookup"]}]},
{"role": "user", "content": "go on"},
]
result: Final = strip_beta_gated_system_messages(messages, frozenset(), drop_params=True)
assert [message["role"] for message in result.messages] == ["user", "assistant", "user"]
def test_system_turn_emptied_by_the_strip_stays_as_a_directive_when_its_output_config_survives() -> None:
directive: Final = {
"role": "system",
"output_config": {"effort": "low"},
"content": [{"type": "tool_removal", "tool_names": ["lookup"]}],
}
result: Final = strip_beta_gated_system_messages(
[directive], frozenset({ANTHROPIC_MID_CONVERSATION_OUTPUT_CONFIG_BETA_HEADER}), drop_params=True
)
assert result.messages == [{"role": "system", "output_config": {"effort": "low"}, "content": []}]
assert result.stripped == ("messages[].system.content[].tool_addition|tool_removal",)
def test_system_turn_emptied_by_the_strip_is_dropped_when_its_output_config_is_gated_too() -> None:
directive: Final = {
"role": "system",
"output_config": {"effort": "low"},
"content": [{"type": "tool_removal", "tool_names": ["lookup"]}],
}
result: Final = strip_beta_gated_system_messages([directive], frozenset(), drop_params=True)
assert result.messages == []
assert result.stripped[-1] == "messages[].system (emptied by the strip)"
def test_request_strip_replaces_only_thinking_and_messages_and_keeps_the_dict_otherwise() -> None:
request: Final = {
"model": "claude",
"thinking": dict(BOUND_THINKING),
"messages": [_system_turn()],
"max_tokens": 5,
}
result: Final = strip_beta_gated_request_fields(request, frozenset(), drop_params=True)
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() -> None:
request: Final = {
"thinking": {"type": "enabled", "budget_tokens": 1024},
"messages": [{"role": "user", "content": "hi"}],
}
result: Final = strip_beta_gated_request_fields(request, frozenset(), drop_params=True)
assert result.request is request
assert result.stripped == ()
@pytest.mark.parametrize("drop_params", [None, False, "false"])
def test_without_drop_params_every_gated_field_is_sent_as_is(
drop_params: object, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(litellm, "drop_params", False)
messages: Final = [{"role": "user", "content": "hi"}, _system_turn()]
request: Final = {"thinking": BOUND_THINKING, "messages": messages}
assert strip_beta_gated_thinking(BOUND_THINKING, frozenset(), drop_params=drop_params).thinking is BOUND_THINKING
assert strip_beta_gated_system_messages(messages, frozenset(), drop_params=drop_params).messages is messages
assert strip_beta_gated_request_fields(request, frozenset(), drop_params=drop_params).request is request
native: Final = strip_beta_gated_native_request({}, {"thinking": BOUND_THINKING}, messages, drop_params=drop_params)
assert (native.thinking, list(native.messages), native.thinking_stripped, native.messages_stripped) == (
BOUND_THINKING,
messages,
False,
False,
)
def test_global_drop_params_strips_without_a_per_call_flag(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(litellm, "drop_params", True)
assert strip_beta_gated_thinking(BOUND_THINKING, frozenset(), drop_params=None).thinking == {"type": "adaptive"}
def test_active_beta_values_reads_the_header_case_insensitively_and_merges_body_betas() -> None:
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,
) -> None:
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: Final = [record for record in caplog.records if record.levelno == logging.WARNING]
assert len(warnings) == 1
assert (
"Dropped thinking.block_binding (unlocked by anthropic-beta thinking-binding-controls-2026-08-01) for "
"model=claude provider=azure_ai" in warnings[0].getMessage()
)