scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-13 08:52:02 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 174

Merge pull request #50456 from BerriAI/litellm_backport_49076_lts_1_108_x

Browse source 
fix(anthropic): backport #49076 encrypted reasoning round-trip to lts/1.108.x
This commit is contained in:
Mateo Wang 2026-11-09 19:36:36 +00:00 • committed by GitHub
commit 862a1fbeee
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
21 changed files with 2541 additions and 177 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

19
litellm/_internal_context.py
View file

@ -36,6 +36,8 @@ REDIS_FAMILIES_METADATA_KEY: Final = "families"
_service_caller: Final[ContextVar[str | None]] = ContextVar("service_caller", default=None)
_resent_on: Final[ContextVar[Callable[[Exception], bool] | None]] = ContextVar("resent_on", default=None)
@contextmanager
def post_response_phase() -> Generator[None]:
@ -113,6 +115,23 @@ def current_service_caller() -> str | None:
return _service_caller.get()
@contextmanager
def resent_on(rejection: Callable[[Exception], bool]) -> Generator[None]:
"""An attempt its caller sends again when it fails with ``rejection``. That failure is not the request's
outcome, so it is neither logged as the request's failure nor retried, and its budget reservation and rate-limit
slot stay with the attempt sent again."""
token: Final = _resent_on.set(rejection)
try:
yield
finally:
_restore(_resent_on, token)
def is_resent_failure(error: Exception) -> bool:
rejection: Final = _resent_on.get()
return rejection is not None and rejection(error)
@contextmanager
def pinned_billing_time(moment: datetime) -> Generator[None]:
"""Price every rate lookup inside this block at ``moment`` rather than at each one's own clock read."""

12
litellm/completion_extras/litellm_responses_transformation/handler.py
View file

@ -182,6 +182,9 @@ class ResponsesToCompletionBridgeHandler:
model_response: Final = validated_kwargs["model_response"]
logging_obj: Final = validated_kwargs["logging_obj"]
custom_llm_provider: Final = validated_kwargs["custom_llm_provider"]
from .transformation import serves_anthropic_messages
serves_messages_client: Final = serves_anthropic_messages(logging_obj)
if kwargs.get("stream") is True and "stream" not in optional_params:
optional_params = {**optional_params, "stream": True}
@ -222,6 +225,7 @@ class ResponsesToCompletionBridgeHandler:
encoding=validated_kwargs["encoding"],
api_key=kwargs.get("api_key"),
json_mode=kwargs.get("json_mode"),
keeps_every_reasoning_item=serves_messages_client,
)
elif isinstance(result, ModelResponse):
if not stream:
@ -247,6 +251,7 @@ class ResponsesToCompletionBridgeHandler:
encoding=validated_kwargs["encoding"],
api_key=kwargs.get("api_key"),
json_mode=kwargs.get("json_mode"),
keeps_every_reasoning_item=serves_messages_client,
)
else:
if self._is_preformatted_cached_chat_stream(result):
@ -260,6 +265,7 @@ class ResponsesToCompletionBridgeHandler:
structured_output=is_structured_output_request(
cast("Mapping[str, object]", optional_params) # cast-ok: optional_params is an untyped dict
),
streams_reasoning_signatures=serves_messages_client,
)
streamwrapper: Final = CustomStreamWrapper(
completion_stream=completion_stream,
@ -282,6 +288,9 @@ class ResponsesToCompletionBridgeHandler:
model_response: Final = validated_kwargs["model_response"]
logging_obj: Final = validated_kwargs["logging_obj"]
custom_llm_provider: Final = validated_kwargs["custom_llm_provider"]
from .transformation import serves_anthropic_messages
serves_messages_client: Final = serves_anthropic_messages(logging_obj)
if kwargs.get("stream") is True and "stream" not in optional_params:
optional_params = {**optional_params, "stream": True}
@ -325,6 +334,7 @@ class ResponsesToCompletionBridgeHandler:
encoding=validated_kwargs["encoding"],
api_key=kwargs.get("api_key"),
json_mode=kwargs.get("json_mode"),
keeps_every_reasoning_item=serves_messages_client,
)
elif isinstance(result, ModelResponse):
if not stream:
@ -350,6 +360,7 @@ class ResponsesToCompletionBridgeHandler:
encoding=validated_kwargs["encoding"],
api_key=kwargs.get("api_key"),
json_mode=kwargs.get("json_mode"),
keeps_every_reasoning_item=serves_messages_client,
)
else:
if self._is_preformatted_cached_chat_stream(result):
@ -363,6 +374,7 @@ class ResponsesToCompletionBridgeHandler:
structured_output=is_structured_output_request(
cast("Mapping[str, object]", optional_params) # cast-ok: optional_params is an untyped dict
),
streams_reasoning_signatures=serves_messages_client,
)
streamwrapper: Final = CustomStreamWrapper(
completion_stream=completion_stream,

136
litellm/completion_extras/litellm_responses_transformation/transformation.py
View file

@ -26,9 +26,11 @@ import litellm
from litellm import ModelResponse
from litellm._logging import verbose_logger
from litellm.integrations.anthropic_cache_control_hook import supports_openai_prompt_cache_breakpoint
from litellm.litellm_core_utils.core_helpers import as_str_mapping, attribute_or_none
from litellm.litellm_core_utils.hidden_params import get_hidden_params, get_or_create_hidden_params
from litellm.litellm_core_utils.prompt_templates.common_utils import (
responses_reasoning_items_from_thinking_blocks,
thinking_blocks_from_reasoning_items,
with_prompt_cache_breakpoint,
)
from litellm.llms.base_llm.base_model_iterator import BaseModelResponseIterator
@ -45,6 +47,8 @@ from litellm.responses.utils import ResponsesAPIRequestUtils, normalize_response
from litellm.types.llms.openai import (
ChatCompletionAnnotation,
ChatCompletionReasoningItem,
ChatCompletionRedactedThinkingBlock,
ChatCompletionThinkingBlock,
ChatCompletionToolCallChunk,
ChatCompletionToolCallFunctionChunk,
ChatCompletionToolParamFunctionChunk,
@ -54,7 +58,7 @@ from litellm.types.llms.openai import (
ResponsesAPIResponse,
ResponsesAPIStreamEvents,
)
from litellm.types.utils import GenericStreamingChunk, ModelResponseStream
from litellm.types.utils import CallTypes, Delta, GenericStreamingChunk, ModelResponseStream, StreamingChoices
if TYPE_CHECKING:
from openai.types.responses import ResponseInputImageParam, ResponseOutputItem
@ -70,8 +74,6 @@ if TYPE_CHECKING:
AllMessageValues,
ChatCompletionFileObject,
ChatCompletionImageObject,
ChatCompletionRedactedThinkingBlock,
ChatCompletionThinkingBlock,
ChatCompletionToolReferenceObject,
OpenAIMessageContentListBlock,
)
@ -225,6 +227,10 @@ class _BuiltReasoningItem(TypedDict):
summary: Sequence[_ReasoningSummaryText]
def serves_anthropic_messages(logging_obj: object) -> bool:
return attribute_or_none(logging_obj, "call_type") == CallTypes.anthropic_messages.value
def _get_reasoning_items(
msg: "AllMessageValues",
) -> list[ChatCompletionReasoningItem]:
@ -443,6 +449,77 @@ def _as_chat_reasoning_items(
return cast(list[ChatCompletionReasoningItem], list(reasoning_items))
def _with_stream_block_index(
block: Mapping[str, object],
index: int,
summary_streamed: bool,
) -> ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock:
if block.get("type") == "redacted_thinking":
data: Final = block.get("data")
return ChatCompletionRedactedThinkingBlock(
type="redacted_thinking",
data=data if isinstance(data, str) else "",
index=index,
)
thinking: Final = block.get("thinking")
signature: Final = block.get("signature")
return ChatCompletionThinkingBlock(
type="thinking",
thinking=thinking if isinstance(thinking, str) and not summary_streamed else "",
signature=signature if isinstance(signature, str) else None,
index=index,
)
def _stream_output_index(chunk: Mapping[str, object]) -> int | None:
output_index: Final = chunk.get("output_index")
return output_index if isinstance(output_index, int) and not isinstance(output_index, bool) else None
def _reasoning_done_thinking_blocks(
item: Mapping[str, object],
output_index: int | None,
summary_streamed: bool,
) -> tuple[ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock, ...]:
index: Final = output_index if output_index is not None else 0
return tuple(
_with_stream_block_index(block, index, summary_streamed)
for block in thinking_blocks_from_reasoning_items((item,))
)
def _reasoning_done_stream_chunk(
chunk: Mapping[str, object],
summary_streamed_outputs: frozenset[int | None],
) -> ModelResponseStream | None:
event_type: Final = chunk.get("type")
event_name: Final = (
event_type.value
if isinstance(event_type, ResponsesAPIStreamEvents)
else event_type
if isinstance(event_type, str)
else None
)
if event_name != ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE:
return None
item: Final = as_str_mapping(chunk.get("item"))
if item is None or item.get("type") != "reasoning":
return None
output_index: Final = _stream_output_index(chunk)
signed_blocks: Final = _reasoning_done_thinking_blocks(item, output_index, output_index in summary_streamed_outputs)
if not signed_blocks:
return None
return ModelResponseStream(
choices=[
StreamingChoices(
index=0,
delta=Delta(thinking_blocks=list(signed_blocks)),
finish_reason=None,
)
]
)
def _shift_annotation(annotation: ChatCompletionAnnotation, offset: int) -> ChatCompletionAnnotation:
shifted: Final = {
key: value + offset if key in ("start_index", "end_index") and isinstance(value, int) else value
@ -476,7 +553,7 @@ def _fold_choices_into_one(
choices: Sequence["Choices"],
tool_calls: Sequence[Mapping[str, object]],
trailing_reasoning_content: str | None,
trailing_reasoning_item: _BuiltReasoningItem | None,
trailing_reasoning_items: Sequence[_BuiltReasoningItem],
tool_search_items: Sequence[Mapping[str, object]],
structured_output: bool = False,
) -> "Choices":
@ -495,7 +572,7 @@ def _fold_choices_into_one(
reasoning_contents: Final = (*(_message_reasoning_content(m) for m in messages), trailing_reasoning_content)
reasoning_items: Final = (
*chain.from_iterable(_message_reasoning_items(m) for m in messages),
*(() if trailing_reasoning_item is None else (trailing_reasoning_item,)),
*trailing_reasoning_items,
)
return Choices(
message=Message(
@ -1149,6 +1226,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
output_items: Sequence[object],
handle_raw_dict_callback: Callable[..., tuple["Choices | None", int]] | None = None,
structured_output: bool = False,
keeps_every_reasoning_item: bool = False,
) -> list["Choices"]:
"""
Convert Responses API output items to chat completion choices.
@ -1158,6 +1236,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
handle_raw_dict_callback: Optional callback for handling raw dict items
structured_output: The chat request carried a json response_format, so only the
first message item becomes content
keeps_every_reasoning_item: Keep each reasoning item before an answer, not only the last
Returns:
List of Choices objects
@ -1180,7 +1259,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
choices: Final[list[Choices]] = []
index = 0
reasoning_content: str | None = None
pending_reasoning_item: _BuiltReasoningItem | None = None
pending_reasoning_items: tuple[_BuiltReasoningItem, ...] = ()
# Collect all tool calls to put them in a single choice
# (Chat Completions API expects all tool calls in one message)
@ -1189,12 +1268,19 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
for item in output_items:
if isinstance(item, ResponseReasoningItem):
pending_reasoning_item = _build_reasoning_item(
item_id=item.id,
encrypted_content=getattr(item, "encrypted_content", None),
summary_raw=item.summary,
pending_reasoning_items = (
*(pending_reasoning_items if keeps_every_reasoning_item else ()),
_build_reasoning_item(
item_id=item.id,
encrypted_content=getattr(item, "encrypted_content", None),
summary_raw=item.summary,
),
)
reasoning_content = " ".join(
s["text"]
for s in chain.from_iterable(built["summary"] for built in pending_reasoning_items)
if s.get("text")
)
reasoning_content = " ".join(s["text"] for s in pending_reasoning_item["summary"] if s.get("text"))
elif isinstance(item, ResponseOutputMessage):
for content in item.content:
@ -1211,7 +1297,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
annotations=annotations,
reasoning_items=cast(
list[ChatCompletionReasoningItem] | None,
([pending_reasoning_item] if pending_reasoning_item is not None else None),
(list(pending_reasoning_items) or None),
),
)
@ -1224,7 +1310,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
)
reasoning_content = None # flush
pending_reasoning_item = None # flush
pending_reasoning_items = () # flush
index += 1
elif isinstance(item, ResponseFunctionToolCall):
@ -1289,7 +1375,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
choices=choices,
tool_calls=accumulated_tool_calls,
trailing_reasoning_content=reasoning_content,
trailing_reasoning_item=pending_reasoning_item,
trailing_reasoning_items=pending_reasoning_items,
tool_search_items=tool_search_items,
structured_output=structured_output,
)
@ -1396,6 +1482,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
encoding: object,
api_key: str | None = None,
json_mode: bool | None = None,
keeps_every_reasoning_item: bool = False,
) -> "ModelResponse":
"""Transform Responses API response to chat completion response"""
from litellm.responses.utils import ResponseAPILoggingUtils
@ -1427,6 +1514,7 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
structured_output=is_structured_output_request(
cast("Mapping[str, object]", optional_params) # cast-ok: optional_params is an untyped dict
),
keeps_every_reasoning_item=keeps_every_reasoning_item,
)
response_is_incomplete: Final = raw_response.status == "incomplete" or (
@ -1490,9 +1578,14 @@ class LiteLLMResponsesTransformationHandler(CompletionTransformationBridge):
sync_stream: bool,
json_mode: bool | None = False,
structured_output: bool = False,
streams_reasoning_signatures: bool = False,
) -> BaseModelResponseIterator:
return OpenAiResponsesToChatCompletionStreamIterator(
streaming_response, sync_stream, json_mode, structured_output=structured_output
streaming_response,
sync_stream,
json_mode,
structured_output=structured_output,
streams_reasoning_signatures=streams_reasoning_signatures,
)
def _convert_content_str_to_input_text(self, content: str, role: str) -> dict[str, object]:
@ -1871,13 +1964,16 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
sync_stream: bool,
json_mode: bool | None = False,
structured_output: bool = False,
streams_reasoning_signatures: bool = False,
):
super().__init__(streaming_response, sync_stream, json_mode)
self._streams_reasoning_signatures: Final = streams_reasoning_signatures
self._chat_completion_id: str | None = None
self._served_service_tier: str | None = None
self._tool_call_index_map: dict[int, int] = {} # mutable-ok: per-stream accumulator state
self._structured_output: Final = structured_output
self._first_message_item_done: bool = False
self._summary_streamed_outputs: frozenset[int | None] = frozenset()
def _handle_string_chunk(
self, str_line: Union[str, "BaseModel"]
@ -2089,7 +2185,6 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
)
]
)
elif event_type == "response.output_text.delta":
# Content part added to output
content_part = parsed_chunk.get("delta", None)
@ -2193,15 +2288,24 @@ class OpenAiResponsesToChatCompletionStreamIterator(BaseModelResponseIterator):
"""
verbose_logger.debug("Chat provider: transform_streaming_response called with chunk: %s", chunk)
self._remember_served_service_tier(chunk)
reasoning_chunk: Final = (
_reasoning_done_stream_chunk(chunk, self._summary_streamed_outputs)
if self._streams_reasoning_signatures
else None
)
translated: Final = (
self._empty_content_chunk()
if self._is_text_delta_after_structured_answer(chunk)
else reasoning_chunk
if reasoning_chunk is not None
else OpenAiResponsesToChatCompletionStreamIterator.translate_responses_chunk_to_openai_stream(
chunk, tool_call_index_map=self._tool_call_index_map
)
)
if self._structured_output and _is_message_item_done(chunk):
self._first_message_item_done = True
if _responses_event_type(chunk) == "response.reasoning_summary_text.delta" and chunk.get("delta"):
self._summary_streamed_outputs = self._summary_streamed_outputs | {_stream_output_index(chunk)}
return self._with_served_service_tier(self._with_stream_scoped_id(translated))
def _is_text_delta_after_structured_answer(self, chunk: Mapping[str, object]) -> bool:

8
litellm/litellm_core_utils/get_llm_provider_logic.py
View file

@ -158,6 +158,14 @@ def declared_authenticating_provider(model: str | None, custom_llm_provider: str
return declared if declared in PROVIDERS_THAT_AUTHENTICATE_ON_PROVIDER_INFO else None
def resolve_model_and_provider_for_metadata(model: str, custom_llm_provider: str | None = None) -> tuple[str, str]:
declared: Final = declared_authenticating_provider(model, custom_llm_provider)
if declared is not None:
return model.removeprefix(f"{declared}/"), declared
resolved_model, resolved_provider, _, _ = get_llm_provider(model=model, custom_llm_provider=custom_llm_provider)
return resolved_model, resolved_provider
def inferred_provider(model: str | None) -> str | None:
if not model:
return None

2
litellm/litellm_core_utils/litellm_logging.py
View file

@ -4593,6 +4593,7 @@ class Logging(LiteLLMLoggingBaseClass):
from litellm.completion_extras.litellm_responses_transformation.transformation import (
LiteLLMResponsesTransformationHandler,
responses_provider_metadata,
serves_anthropic_messages,
)
served_id: Final = _provider_response_id(result)
@ -4607,6 +4608,7 @@ class Logging(LiteLLMLoggingBaseClass):
optional_params={},
litellm_params={},
encoding=litellm.encoding,
keeps_every_reasoning_item=serves_anthropic_messages(self),
)
translated.id = served_id or translated.id
return translated

92
litellm/litellm_core_utils/prompt_templates/common_utils.py
View file

@ -20,6 +20,7 @@ from pydantic import TypeAdapter
import litellm
from litellm import verbose_logger
from litellm.litellm_core_utils.core_helpers import as_str_mapping
from litellm.router_utils.batch_utils import InMemoryFile
from litellm.types.llms.openai import (
AllMessageValues,
@ -2289,6 +2290,89 @@ def encrypted_content_of_block(block: Mapping[str, object]) -> str | None:
return encrypted_content_from_signature(_encrypted_reasoning_field(block))
_REASONING_SUMMARY_SEPARATOR: Final = "\n\n"
_OBJECT_TUPLE: Final = TypeAdapter(tuple[object, ...])
def _object_sequence(value: object) -> tuple[object, ...]:
if not isinstance(value, Sequence) or isinstance(value, (str, bytes)):
return ()
return _OBJECT_TUPLE.validate_python(value)
def _iter_mappings(value: object) -> tuple[Mapping[str, object], ...]:
return tuple(mapping for item in _object_sequence(value) if (mapping := as_str_mapping(item)) is not None)
def _reasoning_summary_part_text(part: object) -> str:
mapping: Final = as_str_mapping(part)
if mapping is not None:
text: Final = mapping.get("text")
return text if isinstance(text, str) else ""
text_attr: Final = getattr(part, "text", None)
return text_attr if isinstance(text_attr, str) else ""
def _reasoning_item_summary_text(item: Mapping[str, object]) -> str:
parts: Final = tuple(_reasoning_summary_part_text(part) for part in _object_sequence(item.get("summary")))
return _REASONING_SUMMARY_SEPARATOR.join(part for part in parts if part)
def thinking_block_from_reasoning_item(
item: object,
) -> ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock | None:
"""The chat thinking block that carries one Responses `reasoning` item back to an Anthropic client.
`encrypted_content` rides the same `litellm_encrypted_reasoning:` signature the Responses
adapter uses. A summary becomes thinking text. Bytes with no summary become `redacted_thinking`,
because an empty thinking string is not something the client can replay.
"""
mapping: Final = as_str_mapping(item)
if mapping is None or mapping.get("type") not in (None, "reasoning"):
return None
encrypted_content: Final = mapping.get("encrypted_content")
if not isinstance(encrypted_content, str) or not encrypted_content:
return None
signature: Final = encrypted_reasoning_signature(encrypted_content)
text: Final = _reasoning_item_summary_text(mapping)
if not text:
return ChatCompletionRedactedThinkingBlock(type="redacted_thinking", data=signature)
return ChatCompletionThinkingBlock(type="thinking", thinking=text, signature=signature)
def thinking_blocks_from_reasoning_items(
items: object,
) -> tuple[ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock, ...]:
return tuple(
block for item in _object_sequence(items) if (block := thinking_block_from_reasoning_item(item)) is not None
)
def _block_encrypted_content(block: object) -> str | None:
mapping: Final = as_str_mapping(block)
if mapping is None:
return None
return encrypted_content_of_block(mapping)
def thinking_blocks_with_encrypted_reasoning(
thinking_blocks: object,
reasoning_items: object,
) -> tuple[Mapping[str, object], ...]:
"""Thinking blocks already on the message, plus any encrypted reasoning item they do not already carry."""
direct: Final = _iter_mappings(thinking_blocks)
represented: Final = frozenset(
content for block in direct if (content := _block_encrypted_content(block)) is not None
)
extra: Final = tuple(
block
for block in thinking_blocks_from_reasoning_items(_iter_mappings(reasoning_items))
if _block_encrypted_content(block) not in represented
)
return (*extra, *direct)
def is_encrypted_reasoning_block(block: object) -> bool:
"""A thinking or redacted_thinking block carrying Responses API encrypted reasoning.
@ -2380,6 +2464,14 @@ def anthropic_content_lists(messages: Sequence[object]) -> Iterator[object]:
)
def _content_carries_encrypted_reasoning(content: object) -> bool:
return any(is_encrypted_reasoning_block(block) for block in _object_sequence(content))
def messages_carry_encrypted_reasoning(messages: Sequence[object]) -> bool:
return any(_content_carries_encrypted_reasoning(content) for content in anthropic_content_lists(messages))
def _strip_encrypted_reasoning_from_blocks(
content: object,
*,

1
litellm/litellm_core_utils/streaming_handler.py
View file

@ -912,6 +912,7 @@ class CustomStreamWrapper:
and model_response.choices[0].delta.role is not None
)
or (getattr(model_response.choices[0].delta, "reasoning_items", None) is not None)
or getattr(model_response.choices[0].delta, "thinking_blocks", None)
):
return True
else:

227
litellm/llms/anthropic/pass_through/adapters/handler.py
View file

@ -1,5 +1,8 @@
from collections.abc import AsyncIterator, Coroutine, Iterator, Mapping, Sequence
import re
from collections.abc import AsyncIterator, Callable, Coroutine, Iterator, Mapping, Sequence
from contextlib import AbstractContextManager, nullcontext
from dataclasses import dataclass
from functools import partial
from typing import (
TYPE_CHECKING,
Final,
@ -11,16 +14,25 @@ from pydantic import TypeAdapter
from typing_extensions import TypedDict
import litellm
from litellm._internal_context import resent_on
from litellm._logging import verbose_logger
from litellm.litellm_core_utils.asyncify import run_async_function
from litellm.litellm_core_utils.core_helpers import as_str_mapping, iter_items
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObject
from litellm.litellm_core_utils.model_param_helper import peek_reasoning_summary_aliases
from litellm.litellm_core_utils.prompt_templates.common_utils import (
is_encrypted_reasoning_block,
thinking_blocks_with_encrypted_reasoning,
)
from litellm.litellm_core_utils.tool_fields import (
ROUTES_WITHOUT_MCP_SERVERS,
drops_mcp_servers,
tools_without_mcp_servers,
)
from litellm.llms.anthropic.common_utils import strip_encrypted_reasoning_blocks_from_anthropic_messages
from litellm.llms.anthropic.pass_through.adapters.transformation import (
AnthropicAdapter,
openai_thinking_routes_to_responses,
)
from litellm.llms.anthropic.pass_through.context_management import (
AnthropicContextManagementError,
@ -48,9 +60,9 @@ from litellm.types.llms.anthropic_messages.anthropic_response import (
)
from litellm.types.llms.openai import OpenAIWebSearchOptions
from litellm.types.utils import ModelResponse
from litellm.utils import get_model_info
if TYPE_CHECKING:
from litellm.chat_completions.dispatch import ChatResult
from litellm.router import Router
from litellm.types.proxy.auth.user_api_key_auth import UserAPIKeyAuth
@ -68,22 +80,156 @@ class _CompletionKwargs(TypedDict, total=False, extra_items=object):
custom_llm_provider: str
_ANTHROPIC_MESSAGES: Final = TypeAdapter(list[dict[str, object]])
_UNVERIFIED_ENCRYPTED_CONTENT: Final = re.compile(r"encrypted content \S+ could not be verified", re.IGNORECASE)
_PrepareReplay: TypeAlias = "Callable[[_AnthropicMessages], _CompletionKwargs]"
@dataclass(frozen=True, slots=True)
class _ChatTargetReasoning:
effort: str
summary: str | None = None
def _provider_error_code(body: object) -> str | None:
mapping: Final = as_str_mapping(body)
if mapping is None:
return None
nested: Final = as_str_mapping(mapping.get("error"))
source: Final = nested if nested is not None else mapping
code: Final = source.get("code")
return code if isinstance(code, str) else None
def _logged_rejection_code(error: litellm.BadRequestError) -> str:
if error.code == "invalid_encrypted_content" or _provider_error_code(error.body) == "invalid_encrypted_content":
return "invalid_encrypted_content"
if "invalid_encrypted_content" in str(error):
return "invalid_encrypted_content"
return "could_not_be_verified"
def _completion_text(completion_kwargs: _CompletionKwargs, key: str) -> str:
value: Final = completion_kwargs.get(key)
return value if isinstance(value, str) else ""
def _message_carries_encrypted_reasoning(message: object) -> bool:
mapping: Final = as_str_mapping(message)
if mapping is None:
return False
blocks: Final = thinking_blocks_with_encrypted_reasoning(
mapping.get("thinking_blocks"), mapping.get("reasoning_items")
)
return any(is_encrypted_reasoning_block(block) for block in blocks)
def _sends_encrypted_reasoning(completion_kwargs: _CompletionKwargs) -> bool:
messages: Final = completion_kwargs.get("messages")
sent: Final = iter_items(messages) if messages is not None else ()
return any(_message_carries_encrypted_reasoning(message) for message in sent)
def _rejects_encrypted_reasoning(error: litellm.BadRequestError) -> bool:
error_message: Final = str(error)
return (
"invalid_encrypted_content" in error_message or _UNVERIFIED_ENCRYPTED_CONTENT.search(error_message) is not None
)
def _is_rejected_reasoning(error: Exception) -> bool:
return isinstance(error, litellm.BadRequestError) and _rejects_encrypted_reasoning(error)
def resent_without_rejected_reasoning(sent_encrypted_reasoning: bool) -> AbstractContextManager[None]:
return resent_on(_is_rejected_reasoning) if sent_encrypted_reasoning else nullcontext()
def messages_without_rejected_reasoning(
messages: Sequence[Mapping[str, object]],
error: litellm.BadRequestError,
*,
sent_encrypted_reasoning: bool,
provider: str,
model: str,
) -> "_AnthropicMessages | None":
if not sent_encrypted_reasoning or not _rejects_encrypted_reasoning(error):
return None
verbose_logger.warning(
"Backend rejected replayed encrypted reasoning, retrying once without it. "
"Enable the encrypted_content_affinity pre-call check to drop such blocks before the call. "
"provider=%s model=%s status=%s code=%s",
provider,
model,
error.status_code,
_logged_rejection_code(error),
)
return _ANTHROPIC_MESSAGES.validate_python(
strip_encrypted_reasoning_blocks_from_anthropic_messages([dict(message) for message in messages])
)
def _messages_without_rejected_reasoning(
messages: _AnthropicMessages,
error: litellm.BadRequestError,
completion_kwargs: _CompletionKwargs,
sent_encrypted_reasoning: bool,
) -> "_AnthropicMessages | None":
return messages_without_rejected_reasoning(
messages,
error,
sent_encrypted_reasoning=sent_encrypted_reasoning,
provider=_completion_text(completion_kwargs, "custom_llm_provider"),
model=_completion_text(completion_kwargs, "model"),
)
async def _acompletion_without_rejected_reasoning(
completion_kwargs: _CompletionKwargs,
messages: _AnthropicMessages,
prepare_replay: _PrepareReplay,
) -> "ChatResult":
sent_encrypted_reasoning: Final = _sends_encrypted_reasoning(completion_kwargs)
try:
with resent_without_rejected_reasoning(sent_encrypted_reasoning):
return await litellm.acompletion(**completion_kwargs)
except litellm.BadRequestError as error:
replayed_messages: Final = _messages_without_rejected_reasoning(
messages, error, completion_kwargs, sent_encrypted_reasoning
)
if replayed_messages is None:
raise
return await litellm.acompletion(**prepare_replay(replayed_messages))
def _completion_without_rejected_reasoning(
completion_kwargs: _CompletionKwargs,
messages: _AnthropicMessages,
prepare_replay: _PrepareReplay,
) -> "ChatResult | Coroutine[object, object, ChatResult]":
sent_encrypted_reasoning: Final = _sends_encrypted_reasoning(completion_kwargs)
try:
with resent_without_rejected_reasoning(sent_encrypted_reasoning):
return litellm.completion(**completion_kwargs)
except litellm.BadRequestError as error:
replayed_messages: Final = _messages_without_rejected_reasoning(
messages, error, completion_kwargs, sent_encrypted_reasoning
)
if replayed_messages is None:
raise
return litellm.completion(**prepare_replay(replayed_messages))
def _is_compaction_block(block: object) -> bool:
mapping: Final = as_str_mapping(block)
return mapping is not None and mapping.get("type") == "compaction"
def _messages_have_compaction_block(messages: _AnthropicMessages) -> bool:
"""Return True when any message carries a ``compaction`` content block."""
for msg in messages:
content = msg.get("content")
if not isinstance(content, list):
continue
for block in content:
if isinstance(block, dict) and block.get("type") == "compaction":
return True
return False
return any(
any(map(_is_compaction_block, content)) for msg in messages if isinstance(content := msg.get("content"), list)
)
def _proxy_router_fallback() -> "Router | None":
@ -413,34 +559,14 @@ class LiteLLMMessagesToCompletionTransformationHandler:
If the user provides a `summary` field in the thinking dict, it is passed
through to the OpenAI reasoning params (opt-in per OpenAI spec).
"""
custom_llm_provider = completion_kwargs.get("custom_llm_provider")
if custom_llm_provider is None:
try:
_, inferred_provider, _, _ = litellm.utils.get_llm_provider(
model=cast(str, completion_kwargs.get("model"))
)
custom_llm_provider = inferred_provider
except Exception:
custom_llm_provider = None
if custom_llm_provider != "openai":
return
if not isinstance(thinking, dict) or thinking.get("type") != "enabled":
return
model: Final = completion_kwargs.get("model")
try:
model_info: Final = get_model_info(model=cast(str, model), custom_llm_provider=custom_llm_provider)
if model_info and model_info.get("supports_reasoning") is False:
# Model doesn't support reasoning/responses API, don't route
return
except Exception:
pass
if not isinstance(model, str) or thinking is None:
return
if not openai_thinking_routes_to_responses(model, completion_kwargs.get("custom_llm_provider"), thinking):
return
if isinstance(model, str) and model and "responses/" not in model:
local_model: Final = model.removeprefix(f"{custom_llm_provider}/")
completion_kwargs["model"] = f"{custom_llm_provider}/responses/{local_model}"
if "responses/" not in model:
completion_kwargs["model"] = f"openai/responses/{model.removeprefix('openai/')}"
auto_summary: Final = is_reasoning_auto_summary_enabled()
@ -646,12 +772,15 @@ class LiteLLMMessagesToCompletionTransformationHandler:
if "output_config" in extra_kwargs:
request_data["output_config"] = extra_kwargs["output_config"]
api_base: Final = provider_kwargs.get("api_base")
(
openai_request,
tool_name_mapping,
) = ANTHROPIC_ADAPTER.translate_completion_input_params_with_tool_mapping(
request_data,
custom_llm_provider=target_provider,
api_base=api_base if isinstance(api_base, str) else None,
reasoning_summary=peek_reasoning_summary_aliases(provider_kwargs),
)
if openai_request is None:
@ -779,12 +908,9 @@ class LiteLLMMessagesToCompletionTransformationHandler:
parent_trace_id=sub_call_trace_id,
)
(
completion_kwargs,
tool_name_mapping,
) = LiteLLMMessagesToCompletionTransformationHandler._prepare_completion_kwargs(
prepare: Final = partial(
LiteLLMMessagesToCompletionTransformationHandler._prepare_completion_kwargs,
max_tokens=max_tokens,
messages=effective_messages,
model=model,
metadata=metadata,
stop_sequences=stop_sequences,
@ -799,8 +925,10 @@ class LiteLLMMessagesToCompletionTransformationHandler:
output_format=output_format,
extra_kwargs=kwargs,
)
completion_response: Final = await litellm.acompletion(**completion_kwargs)
initial_kwargs, tool_name_mapping = prepare(messages=effective_messages)
completion_response: Final = await _acompletion_without_rejected_reasoning(
initial_kwargs, effective_messages, lambda replayed_messages: prepare(messages=replayed_messages)[0]
)
logging_obj: Final = litellm_logging_obj_from_kwargs(kwargs)
if stream:
@ -924,12 +1052,9 @@ class LiteLLMMessagesToCompletionTransformationHandler:
effective_messages: Final = polyfill_result.messages if polyfill_result is not None else messages
effective_system: Final = polyfill_result.system if polyfill_result is not None else system
(
completion_kwargs,
tool_name_mapping,
) = LiteLLMMessagesToCompletionTransformationHandler._prepare_completion_kwargs(
prepare: Final = partial(
LiteLLMMessagesToCompletionTransformationHandler._prepare_completion_kwargs,
max_tokens=max_tokens,
messages=effective_messages,
model=model,
metadata=metadata,
stop_sequences=stop_sequences,
@ -944,8 +1069,10 @@ class LiteLLMMessagesToCompletionTransformationHandler:
output_format=output_format,
extra_kwargs=kwargs,
)
completion_response: Final = litellm.completion(**completion_kwargs)
initial_kwargs, tool_name_mapping = prepare(messages=effective_messages)
completion_response: Final = _completion_without_rejected_reasoning(
initial_kwargs, effective_messages, lambda replayed_messages: prepare(messages=replayed_messages)[0]
)
logging_obj: Final = litellm_logging_obj_from_kwargs(kwargs)
if stream:

40
litellm/llms/anthropic/pass_through/adapters/streaming_iterator.py
View file

@ -24,6 +24,7 @@ from typing_extensions import assert_never
from litellm._logging import verbose_logger
from litellm._uuid import uuid
from litellm.exceptions import ContextWindowExceededError, MidStreamFallbackError
from litellm.litellm_core_utils.core_helpers import as_str_mapping
from litellm.litellm_core_utils.hidden_params import set_hidden_params
from litellm.llms.anthropic.pass_through.safeguards import StreamedSafeguardResults
from litellm.llms.anthropic.pass_through.utils import anthropic_sse_frame
@ -71,6 +72,21 @@ def _optional_attr_sequence(obj: object, name: str) -> Sequence[object]:
return value if value else ()
def _first_thinking_block(chunk: "ModelResponseStream") -> Mapping[str, object] | None:
delta: Final = _optional_attr(chunk.choices[0], "delta") if chunk.choices else None
blocks: Final = _optional_attr_sequence(delta, "thinking_blocks")
return as_str_mapping(blocks[0]) if blocks else None
def _thinking_block_index(block: Mapping[str, object] | None) -> int | None:
index: Final = block.get("index") if block is not None else None
return index if isinstance(index, int) else None
def _closes_thinking_block(block: Mapping[str, object] | None) -> bool:
return block is not None and _thinking_block_index(block) is not None and bool(block.get("signature"))
def _error_status_and_message(exc: Exception) -> tuple[int, str]:
if isinstance(exc, (BaseLLMException, MidStreamFallbackError, ContextWindowExceededError)):
return exc.status_code, exc.message
@ -458,12 +474,14 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
sent_first_chunk: bool = False
sent_content_block_start: bool = False
sent_content_block_finish: bool = False
current_content_block_type: Literal["text", "tool_use", "thinking"] = "text"
current_content_block_type: Literal["text", "tool_use", "thinking", "redacted_thinking"] = "text"
sent_last_message: bool = False
holding_chunk: ContentBlockDelta | None = None
holding_stop_reason_chunk: MessageBlockDelta | None = None
queued_usage_chunk: bool = False
current_content_block_index: int = 0
current_thinking_block_index: int | None = None
current_thinking_block_closed: bool = False
def __init__(
self,
@ -1352,11 +1370,31 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
original_name: Final = self.tool_name_mapping.get(truncated_name, truncated_name)
tool_block["name"] = original_name
thinking_block: Final = _first_thinking_block(chunk)
thinking_block_index: Final = _thinking_block_index(thinking_block)
previous_thinking_block_index: Final = self.current_thinking_block_index
follows_closed_thinking_block: Final = self.current_thinking_block_closed and block_type == "thinking"
self.current_thinking_block_closed = _closes_thinking_block(thinking_block)
if (
thinking_block_index is not None
or block_type != self.current_content_block_type
or follows_closed_thinking_block
):
self.current_thinking_block_index = thinking_block_index
if block_type != self.current_content_block_type:
self.current_content_block_type = block_type
self.current_content_block_start = content_block_start
return True
if follows_closed_thinking_block or (
thinking_block_index is not None
and previous_thinking_block_index is not None
and thinking_block_index != previous_thinking_block_index
):
self.current_content_block_start = content_block_start
return True
# For parallel tool calls, we'll necessarily have a new content block
# if we get a function name since it signals a new tool call
if block_type == "tool_use":

173
litellm/llms/anthropic/pass_through/adapters/transformation.py
View file

@ -2,12 +2,14 @@ import copy
import hashlib
import json
from collections.abc import AsyncIterator, Iterator, Mapping, Sequence
from dataclasses import dataclass
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, Literal, TypeAlias, TypeVar, cast
from pydantic import JsonValue, TypeAdapter
import litellm
from litellm.litellm_core_utils.core_helpers import as_str_mapping, attribute_items
from litellm.litellm_core_utils.hidden_params import get_hidden_params
from litellm.llms.anthropic.pass_through.utils import (
is_reasoning_auto_summary_enabled,
@ -122,6 +124,7 @@ from litellm.litellm_core_utils.prompt_templates.common_utils import (
anthropic_image_source_to_openai_url,
parse_tool_call_arguments,
reasoning_content_from_thinking_blocks,
thinking_blocks_with_encrypted_reasoning,
with_prompt_cache_breakpoint,
)
from litellm.litellm_core_utils.prompt_templates.factory import (
@ -219,6 +222,72 @@ def target_supports_mid_conversation_system(model: str | None, custom_llm_provid
return supports_mid_conversation_system(model=model, custom_llm_provider=custom_llm_provider)
@dataclass(frozen=True, slots=True)
class ChatRouteInputs:
tools: Sequence[Mapping[str, object]] | None = None
reasoning_effort: str | Mapping[str, object] | None = None
api_base: str | None = None
thinking: Mapping[str, object] | None = None
reasoning_summary: object | None = None
_NO_ROUTE_INPUTS: Final = ChatRouteInputs()
def _inferred_provider(model: str) -> str | None:
try:
return litellm.utils.get_llm_provider(model=model)[1]
except Exception: # noqa: BLE001 # an unknown model name has no inferred provider
return None
def openai_thinking_routes_to_responses(
model: str | None,
custom_llm_provider: str | None,
thinking: Mapping[str, object] | None,
) -> bool:
if not model or thinking is None or thinking.get("type") != "enabled":
return False
provider: Final = custom_llm_provider or _inferred_provider(model)
if provider != "openai":
return False
try:
model_info: Final = litellm.get_model_info(model=model, custom_llm_provider=provider)
except Exception: # noqa: BLE001 # a model missing from the cost map still routes, as before
return True
return model_info.get("supports_reasoning") is not False
def _keeps_encrypted_reasoning_for_responses(
model: str | None,
custom_llm_provider: str | None,
route: ChatRouteInputs,
) -> bool:
from litellm.responses.bridge_check import chat_completion_ends_in_responses_api
return openai_thinking_routes_to_responses(
model, custom_llm_provider, route.thinking
) or chat_completion_ends_in_responses_api(
model,
custom_llm_provider,
tools=route.tools,
reasoning_effort=route.reasoning_effort,
reasoning_summary=route.reasoning_summary,
api_base=route.api_base,
)
def _anthropic_thinking_blocks(source: object) -> tuple[Mapping[str, object], ...]:
return thinking_blocks_with_encrypted_reasoning(
getattr(source, "thinking_blocks", None),
getattr(source, "reasoning_items", None),
)
def _streamed_thinking_blocks(delta: object) -> tuple[Mapping[str, object], ...]:
return tuple(block for item in attribute_items(delta, "thinking_blocks") if (block := as_str_mapping(item)))
def _chat_tool_param(function_chunk: ChatCompletionToolParamFunctionChunk, tool: object) -> ChatCompletionToolParam:
eager_input_streaming: Final = eager_input_streaming_flag(tool)
if eager_input_streaming is None:
@ -246,7 +315,12 @@ class AnthropicAdapter:
return result
def translate_completion_input_params_with_tool_mapping(
self, kwargs, *, custom_llm_provider: str | None = None
self,
kwargs,
*,
custom_llm_provider: str | None = None,
api_base: str | None = None,
reasoning_summary: object | None = None,
) -> tuple[ChatCompletionRequest | None, dict[str, str]]:
"""
Translate Anthropic request params to OpenAI format, returning tool name mapping.
@ -280,6 +354,8 @@ class AnthropicAdapter:
) = LiteLLMAnthropicMessagesAdapter().translate_anthropic_to_openai(
anthropic_message_request=request_body,
custom_llm_provider=custom_llm_provider,
api_base=api_base,
reasoning_summary=reasoning_summary,
)
return translated_body, tool_name_mapping
@ -465,9 +541,14 @@ class LiteLLMAnthropicMessagesAdapter:
*,
custom_llm_provider: str | None = None,
preserve_midturn_system: bool = False,
route: ChatRouteInputs = _NO_ROUTE_INPUTS,
) -> list[AllMessageValues]:
new_messages: Final[list[AllMessageValues]] = []
replayable_messages: Final = strip_encrypted_reasoning_blocks_from_anthropic_messages(messages)
replayable_messages: Final = (
messages
if _keeps_encrypted_reasoning_for_responses(model, custom_llm_provider, route)
else strip_encrypted_reasoning_blocks_from_anthropic_messages(messages)
)
leading_count: Final = next(
(i for i, m in enumerate(replayable_messages) if not is_system_role_message(m)),
len(replayable_messages),
@ -1221,6 +1302,8 @@ class LiteLLMAnthropicMessagesAdapter:
anthropic_message_request: AnthropicMessagesRequest,
*,
custom_llm_provider: str | None = None,
api_base: str | None = None,
reasoning_summary: object | None = None,
preserve_midturn_system: bool = False,
) -> tuple[ChatCompletionRequest, dict[str, str]]:
"""
@ -1232,26 +1315,15 @@ class LiteLLMAnthropicMessagesAdapter:
for tools that exceeded OpenAI's 64-char limit
"""
# Debug: Processing Anthropic message request
new_messages: list[AllMessageValues] = []
tool_name_mapping: dict[str, str] = {}
## CONVERT ANTHROPIC MESSAGES TO OPENAI
messages_list: Final[list[AllAnthropicPassThroughMessageValues]] = cast(
list[AllAnthropicPassThroughMessageValues],
anthropic_message_request["messages"],
)
new_messages = self.translate_anthropic_messages_to_openai(
messages=messages_list,
model=anthropic_message_request.get("model"),
custom_llm_provider=custom_llm_provider,
preserve_midturn_system=preserve_midturn_system,
)
## ADD SYSTEM MESSAGE TO MESSAGES
self._add_system_message_to_messages(new_messages, anthropic_message_request)
new_kwargs: Final[ChatCompletionRequest] = {
"model": anthropic_message_request["model"],
"messages": new_messages,
"messages": [],
}
## CONVERT METADATA (user_id + litellm metadata)
self._translate_metadata_to_openai(
@ -1275,6 +1347,23 @@ class LiteLLMAnthropicMessagesAdapter:
new_kwargs=new_kwargs,
custom_llm_provider=custom_llm_provider,
)
## CONVERT ANTHROPIC MESSAGES TO OPENAI, after tools and thinking pick the route
new_messages: Final = self.translate_anthropic_messages_to_openai(
messages=messages_list,
model=anthropic_message_request.get("model"),
custom_llm_provider=custom_llm_provider,
preserve_midturn_system=preserve_midturn_system,
route=ChatRouteInputs(
tools=new_kwargs.get("tools"),
reasoning_effort=new_kwargs.get("reasoning_effort"),
api_base=api_base,
thinking=anthropic_message_request.get("thinking"),
reasoning_summary=reasoning_summary,
),
)
## ADD SYSTEM MESSAGE TO MESSAGES
self._add_system_message_to_messages(new_messages, anthropic_message_request)
new_kwargs["messages"] = new_messages
## CONVERT STOP_SEQUENCES
self._translate_stop_sequences_to_openai(
anthropic_message_request=anthropic_message_request,
@ -1358,9 +1447,9 @@ class LiteLLMAnthropicMessagesAdapter:
for choice in choices
):
new_content.extend(dict(_COMPACTION_BLOCK.validate_python(block)) for block in compaction_blocks)
# Handle thinking blocks first
if hasattr(choice.message, "thinking_blocks") and choice.message.thinking_blocks:
for thinking_block in choice.message.thinking_blocks:
signed_thinking_blocks = _anthropic_thinking_blocks(choice.message)
if signed_thinking_blocks:
for thinking_block in signed_thinking_blocks:
if thinking_block.get("type") == "thinking":
if is_empty_unsigned_thinking_block(thinking_block):
continue
@ -1627,7 +1716,7 @@ class LiteLLMAnthropicMessagesAdapter:
def translate_streaming_openai_chunk_to_anthropic_content_block(
self, choices: list[OpenAIStreamingChoice | StreamingChoices]
) -> tuple[
Literal["text", "tool_use", "thinking"],
Literal["text", "tool_use", "thinking", "redacted_thinking"],
"ContentBlockContentBlockDict",
]:
from litellm._uuid import uuid
@ -1660,20 +1749,24 @@ class LiteLLMAnthropicMessagesAdapter:
choice.delta
) is not None:
return "text", TextBlock(type="text", text="")
elif isinstance(choice, StreamingChoices) and hasattr(choice.delta, "thinking_blocks"):
thinking_blocks = choice.delta.thinking_blocks or []
if len(thinking_blocks) > 0:
thinking_block = thinking_blocks[0]
if thinking_block["type"] == "thinking":
thinking = thinking_block.get("thinking") or ""
signature = thinking_block.get("signature") or ""
elif isinstance(choice, StreamingChoices) and (thinking_blocks := _streamed_thinking_blocks(choice.delta)):
thinking_block = thinking_blocks[0]
if thinking_block["type"] == "redacted_thinking":
redacted_data = thinking_block.get("data")
return "redacted_thinking", ChatCompletionRedactedThinkingBlock(
type="redacted_thinking",
data=redacted_data if isinstance(redacted_data, str) else "",
)
if thinking_block["type"] == "thinking":
thinking = thinking_block.get("thinking") or ""
signature = thinking_block.get("signature") or ""
assert isinstance(thinking, str)
assert isinstance(signature, str)
assert isinstance(thinking, str)
assert isinstance(signature, str)
return "thinking", ChatCompletionThinkingBlock(
type="thinking", thinking=thinking, signature=signature
)
return "thinking", ChatCompletionThinkingBlock(
type="thinking", thinking=thinking, signature=signature
)
# OpenAI-compatible reasoning backends (e.g. vLLM/SGLang reasoning
# parsers) populate ``reasoning_content`` without ``thinking_blocks``.
# ``Delta`` deletes the ``thinking_blocks`` attribute when unset, so the
@ -1706,19 +1799,17 @@ class LiteLLMAnthropicMessagesAdapter:
for tool in choice.delta.tool_calls:
if tool.function is not None and tool.function.arguments is not None:
partial_json = (partial_json or "") + tool.function.arguments
elif isinstance(choice, StreamingChoices) and hasattr(choice.delta, "thinking_blocks"):
thinking_blocks = choice.delta.thinking_blocks or []
if len(thinking_blocks) > 0:
for thinking_block in thinking_blocks:
if thinking_block["type"] == "thinking":
thinking = thinking_block.get("thinking") or ""
signature = thinking_block.get("signature") or ""
elif isinstance(choice, StreamingChoices) and (thinking_blocks := _streamed_thinking_blocks(choice.delta)):
for thinking_block in thinking_blocks:
if thinking_block["type"] == "thinking":
thinking = thinking_block.get("thinking") or ""
signature = thinking_block.get("signature") or ""
assert isinstance(thinking, str)
assert isinstance(signature, str)
assert isinstance(thinking, str)
assert isinstance(signature, str)
reasoning_content += thinking
reasoning_signature += signature
reasoning_content += thinking
reasoning_signature += signature
# Handle reasoning_content when thinking_blocks is not present
# This handles providers like OpenRouter that return reasoning_content
elif isinstance(choice, StreamingChoices) and hasattr(choice.delta, "reasoning_content"):

165
litellm/llms/anthropic/pass_through/responses_adapters/handler.py
View file

@ -14,9 +14,10 @@ from litellm.integrations.anthropic_cache_control_hook import (
AnthropicCacheControlHook,
supports_openai_prompt_cache_breakpoint,
)
from litellm.litellm_core_utils.core_helpers import normalize_drop_params
from litellm.litellm_core_utils.core_helpers import as_str_mapping, iter_items, normalize_drop_params
from litellm.litellm_core_utils.prompt_templates.common_utils import cache_control_as_prompt_cache_breakpoint
from litellm.litellm_core_utils.tool_fields import drops_mcp_servers, tools_without_mcp_servers
from litellm.responses.streaming_iterator import BaseResponsesAPIStreamingIterator
from litellm.types.integrations.custom_logger import CONVERTED_STREAM_KEYS
from litellm.types.llms.anthropic import (
AllAnthropicMessageValues,
@ -31,7 +32,11 @@ from litellm.types.llms.anthropic_messages.anthropic_response import (
from litellm.types.llms.openai import ResponsesAPIResponse
from litellm.utils import ProviderConfigManager
from ..adapters.handler import request_safeguards_evaluator
from ..adapters.handler import (
messages_without_rejected_reasoning,
request_safeguards_evaluator,
resent_without_rejected_reasoning,
)
from ..safeguards import with_safeguard_results
from ..utils import litellm_logging_obj_from_kwargs, local_model_name
from .streaming_iterator import AnthropicResponsesStreamWrapper
@ -254,6 +259,78 @@ def _build_responses_kwargs(
return responses_kwargs
_ResponsesKwargsFor: TypeAlias = Callable[[AnthropicRequestMessages], Mapping[str, object]]
_ResponsesResult: TypeAlias = ResponsesAPIResponse | BaseResponsesAPIStreamingIterator
_SyncResponsesResult: TypeAlias = _ResponsesResult | Coroutine[object, object, _ResponsesResult]
def _is_encrypted_reasoning_item(item: object) -> bool:
mapping: Final = as_str_mapping(item)
return mapping is not None and mapping.get("type") == "reasoning" and bool(mapping.get("encrypted_content"))
def _sends_encrypted_reasoning(responses_kwargs: Mapping[str, object]) -> bool:
input_items: Final = responses_kwargs.get("input")
sent: Final = iter_items(input_items) if input_items is not None else ()
return any(map(_is_encrypted_reasoning_item, sent))
def _kwarg_text(responses_kwargs: Mapping[str, object], key: str) -> str:
value: Final = responses_kwargs.get(key)
return value if isinstance(value, str) else ""
def _messages_without_rejected_reasoning(
messages: AnthropicRequestMessages,
error: litellm.BadRequestError,
responses_kwargs: Mapping[str, object],
sent_encrypted_reasoning: bool,
) -> AnthropicRequestMessages | None:
return messages_without_rejected_reasoning(
messages,
error,
sent_encrypted_reasoning=sent_encrypted_reasoning,
provider=_kwarg_text(responses_kwargs, "custom_llm_provider"),
model=_kwarg_text(responses_kwargs, "model"),
)
async def _aresponses_without_rejected_reasoning(
messages: AnthropicRequestMessages, responses_kwargs_for: _ResponsesKwargsFor
) -> tuple[Mapping[str, object], _ResponsesResult]:
responses_kwargs: Final = responses_kwargs_for(messages)
sent_encrypted_reasoning: Final = _sends_encrypted_reasoning(responses_kwargs)
try:
with resent_without_rejected_reasoning(sent_encrypted_reasoning):
return responses_kwargs, await litellm.aresponses(**responses_kwargs)
except litellm.BadRequestError as error:
replayed_messages: Final = _messages_without_rejected_reasoning(
messages, error, responses_kwargs, sent_encrypted_reasoning
)
if replayed_messages is None:
raise
replay_kwargs: Final = responses_kwargs_for(replayed_messages)
return replay_kwargs, await litellm.aresponses(**replay_kwargs)
def _responses_without_rejected_reasoning(
messages: AnthropicRequestMessages, responses_kwargs_for: _ResponsesKwargsFor
) -> tuple[Mapping[str, object], _SyncResponsesResult]:
responses_kwargs: Final = responses_kwargs_for(messages)
sent_encrypted_reasoning: Final = _sends_encrypted_reasoning(responses_kwargs)
try:
with resent_without_rejected_reasoning(sent_encrypted_reasoning):
return responses_kwargs, litellm.responses(**responses_kwargs)
except litellm.BadRequestError as error:
replayed_messages: Final = _messages_without_rejected_reasoning(
messages, error, responses_kwargs, sent_encrypted_reasoning
)
if replayed_messages is None:
raise
replay_kwargs: Final = responses_kwargs_for(replayed_messages)
return replay_kwargs, litellm.responses(**replay_kwargs)
class LiteLLMMessagesToResponsesAPIHandler:
"""
Handles Anthropic /v1/messages requests for OpenAI / Azure models by
@ -281,28 +358,30 @@ class LiteLLMMessagesToResponsesAPIHandler:
output_format: AnthropicOutputSchema | None = None,
**kwargs: object,
) -> AnthropicMessagesResponse | AsyncIterator[bytes]:
responses_kwargs: Final = _build_responses_kwargs(
max_tokens=max_tokens,
messages=messages,
model=model,
context_management=context_management,
metadata=metadata,
output_config=output_config,
stop_sequences=stop_sequences,
stream=stream,
system=system,
temperature=temperature,
thinking=thinking,
tool_choice=tool_choice,
tools=tools,
top_k=top_k,
top_p=top_p,
output_format=output_format,
extra_kwargs=kwargs,
)
def responses_kwargs_for(sent_messages: AnthropicRequestMessages) -> Mapping[str, object]:
return _build_responses_kwargs(
max_tokens=max_tokens,
messages=sent_messages,
model=model,
context_management=context_management,
metadata=metadata,
output_config=output_config,
stop_sequences=stop_sequences,
stream=stream,
system=system,
temperature=temperature,
thinking=thinking,
tool_choice=tool_choice,
tools=tools,
top_k=top_k,
top_p=top_p,
output_format=output_format,
extra_kwargs=kwargs,
)
safeguards_evaluator: Final = request_safeguards_evaluator(kwargs, messages)
result: Final = await litellm.aresponses(**responses_kwargs)
responses_kwargs, result = await _aresponses_without_rejected_reasoning(messages, responses_kwargs_for)
if stream:
wrapper: Final = AnthropicResponsesStreamWrapper(
@ -366,28 +445,28 @@ class LiteLLMMessagesToResponsesAPIHandler:
**kwargs,
)
# Sync path
responses_kwargs: Final = _build_responses_kwargs(
max_tokens=max_tokens,
messages=messages,
model=model,
context_management=context_management,
metadata=metadata,
output_config=output_config,
stop_sequences=stop_sequences,
stream=stream,
system=system,
temperature=temperature,
thinking=thinking,
tool_choice=tool_choice,
tools=tools,
top_k=top_k,
top_p=top_p,
output_format=output_format,
extra_kwargs=kwargs,
)
def responses_kwargs_for(sent_messages: AnthropicRequestMessages) -> Mapping[str, object]:
return _build_responses_kwargs(
max_tokens=max_tokens,
messages=sent_messages,
model=model,
context_management=context_management,
metadata=metadata,
output_config=output_config,
stop_sequences=stop_sequences,
stream=stream,
system=system,
temperature=temperature,
thinking=thinking,
tool_choice=tool_choice,
tools=tools,
top_k=top_k,
top_p=top_p,
output_format=output_format,
extra_kwargs=kwargs,
)
result: Final = litellm.responses(**responses_kwargs)
responses_kwargs, result = _responses_without_rejected_reasoning(messages, responses_kwargs_for)
if stream:
wrapper: Final = AnthropicResponsesStreamWrapper(

33
litellm/main.py
View file

@ -27,6 +27,7 @@ from functools import partial
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, Literal, Optional, Protocol, Union, cast, get_args
from litellm._internal_context import is_resent_failure
from litellm._logging import redact_string
from litellm._uuid import uuid
@ -6235,6 +6236,10 @@ def completion(
)
def _is_not_resent_failure(error: BaseException) -> bool:
return isinstance(error, Exception) and not is_resent_failure(error)
def completion_with_retries(*args, **kwargs):
"""
Executes a litellm.completion() with 3 retries
@ -6254,10 +6259,15 @@ def completion_with_retries(*args, **kwargs):
retryer = tenacity.Retrying(
wait=tenacity.wait_exponential(multiplier=1, max=10),
stop=tenacity.stop_after_attempt(num_retries),
retry=tenacity.retry_if_exception(_is_not_resent_failure),
reraise=True,
)
else:
retryer = tenacity.Retrying(stop=tenacity.stop_after_attempt(num_retries), reraise=True)
retryer = tenacity.Retrying(
stop=tenacity.stop_after_attempt(num_retries),
retry=tenacity.retry_if_exception(_is_not_resent_failure),
reraise=True,
)
return retryer(original_function, *args, **kwargs)
@ -6280,10 +6290,15 @@ async def acompletion_with_retries(*args, **kwargs):
retryer = tenacity.AsyncRetrying(
wait=tenacity.wait_exponential(multiplier=1, max=10),
stop=tenacity.stop_after_attempt(num_retries),
retry=tenacity.retry_if_exception(_is_not_resent_failure),
reraise=True,
)
else:
retryer = tenacity.AsyncRetrying(stop=tenacity.stop_after_attempt(num_retries), reraise=True)
retryer = tenacity.AsyncRetrying(
stop=tenacity.stop_after_attempt(num_retries),
retry=tenacity.retry_if_exception(_is_not_resent_failure),
reraise=True,
)
return await retryer(original_function, *args, **kwargs)
@ -6308,10 +6323,15 @@ def responses_with_retries(*args, **kwargs):
retryer = tenacity.Retrying(
wait=tenacity.wait_exponential(multiplier=1, max=10),
stop=tenacity.stop_after_attempt(num_retries),
retry=tenacity.retry_if_exception(_is_not_resent_failure),
reraise=True,
)
else:
retryer = tenacity.Retrying(stop=tenacity.stop_after_attempt(num_retries), reraise=True)
retryer = tenacity.Retrying(
stop=tenacity.stop_after_attempt(num_retries),
retry=tenacity.retry_if_exception(_is_not_resent_failure),
reraise=True,
)
return retryer(original_function, *args, **kwargs)
@ -6335,10 +6355,15 @@ async def aresponses_with_retries(*args, **kwargs):
retryer = tenacity.AsyncRetrying(
wait=tenacity.wait_exponential(multiplier=1, max=10),
stop=tenacity.stop_after_attempt(num_retries),
retry=tenacity.retry_if_exception(_is_not_resent_failure),
reraise=True,
)
else:
retryer = tenacity.AsyncRetrying(stop=tenacity.stop_after_attempt(num_retries), reraise=True)
retryer = tenacity.AsyncRetrying(
stop=tenacity.stop_after_attempt(num_retries),
retry=tenacity.retry_if_exception(_is_not_resent_failure),
reraise=True,
)
return await retryer(original_function, *args, **kwargs)

42
litellm/responses/bridge_check.py
View file

@ -4,7 +4,10 @@ from urllib.parse import urlsplit
import litellm
from litellm._logging import verbose_logger
from litellm.litellm_core_utils.get_llm_provider_logic import get_llm_provider
from litellm.litellm_core_utils.get_llm_provider_logic import (
get_llm_provider,
resolve_model_and_provider_for_metadata,
)
from litellm.llms.azure_ai.chat.transformation import AzureAIGPT5Config
from litellm.llms.azure_ai.common_utils import (
azure_ai_supports_native_responses,
@ -69,7 +72,7 @@ def responses_api_bridge_check(
reasoning_effort: str | Mapping[str, object] | None = None,
reasoning_summary: object | None = None,
api_base: str | None = None,
) -> tuple[dict, str]:
) -> tuple[dict[str, object], str]:
model_info: dict[str, object] = {}
# Global flag: route ALL OpenAI chat completions through Responses API.
@ -186,3 +189,38 @@ def responses_api_bridge_check(
model = model.replace("responses/", "")
return model_info, model
def chat_completion_ends_in_responses_api(
model: str | None,
custom_llm_provider: str | None,
*,
tools: Sequence[Mapping[str, object]] | None = None,
reasoning_effort: str | Mapping[str, object] | None = None,
reasoning_summary: object | None = None,
api_base: str | None = None,
) -> bool:
"""Whether chat completion for this model is served by a Responses API.
Encrypted reasoning can be replayed only there. A normal chat provider cannot
verify the block, so the messages adapter strips it. ``chatgpt``, a ``responses/``
model, and a request whose tools or reasoning effort the bridge sends to Responses keep it.
"""
if model is None and custom_llm_provider is None:
return False
try:
resolved_model, provider = resolve_model_and_provider_for_metadata(model or "", custom_llm_provider)
except Exception as error: # noqa: BLE001 # unknown model names must not fail message translation
verbose_logger.debug("responses replay check failed for %s: %s", model, error)
return custom_llm_provider == LlmProviders.CHATGPT.value or (
isinstance(model, str) and (model == "chatgpt" or model.startswith("chatgpt/"))
)
model_info, _resolved = responses_api_bridge_check(
model=resolved_model,
custom_llm_provider=provider,
tools=tools,
reasoning_effort=reasoning_effort,
reasoning_summary=reasoning_summary,
api_base=api_base,
)
return model_info.get("mode") == "responses"

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

@ -545,7 +545,9 @@ class ContentBlockStartText(TypedDict):
content_block: TextBlock
ContentBlockContentBlockDict = ToolUseBlock | TextBlock | ChatCompletionThinkingBlock
ContentBlockContentBlockDict = (
ToolUseBlock | TextBlock | ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock
)
ContentBlockStart = ContentBlockStartToolUse | ContentBlockStartText

36
litellm/utils.py
View file

@ -51,7 +51,7 @@ import litellm.litellm_core_utils
# audio_utils.utils is lazy-loaded - only imported when needed for transcription calls
import litellm.litellm_core_utils.json_validation_rule
from litellm._internal_context import is_internal_call
from litellm._internal_context import is_internal_call, is_resent_failure
from litellm._lazy_imports import (
_get_modified_max_tokens,
get_default_encoding,
@ -1980,12 +1980,13 @@ def client(original_function):
retry_policy=kwargs.get("retry_policy"),
)
kwargs["retry_policy"] = reset_retry_policy() # prevent infinite loops
litellm.num_retries = None # set retries to None to prevent infinite loops
if not is_resent_failure(e):
litellm.num_retries = None # set retries to None to prevent infinite loops
context_window_fallback_dict: Final = kwargs.get("context_window_fallback_dict", {})
is_completion_litellm_router_call: Final = _is_litellm_router_call(kwargs, is_async=False)
if (
num_retries and not is_completion_litellm_router_call
num_retries and not is_completion_litellm_router_call and not is_resent_failure(e)
): # only enter this if call is not from litellm router/proxy. router has it's own logic for retrying
if (
isinstance(e, openai.APIError)
@ -2015,11 +2016,12 @@ def client(original_function):
retry_policy=kwargs.get("retry_policy"),
)
kwargs["retry_policy"] = reset_retry_policy() # prevent infinite loops
litellm.num_retries = None # set retries to None to prevent infinite loops
if not is_resent_failure(e):
litellm.num_retries = None # set retries to None to prevent infinite loops
is_responses_litellm_router_call: Final = _is_litellm_router_call(kwargs, is_async=False)
if (
num_retries and not is_responses_litellm_router_call
num_retries and not is_responses_litellm_router_call and not is_resent_failure(e)
): # only enter this if call is not from litellm router/proxy. router has it's own logic for retrying
if (
isinstance(e, openai.APIError)
@ -2032,7 +2034,7 @@ def client(original_function):
end_time = datetime.datetime.now()
# LOG FAILURE - handle streaming failure logging in the _next_ object, remove `handle_failure` once it's deprecated
if logging_obj:
if logging_obj and not is_resent_failure(e):
logging_obj.failure_handler(
e, traceback_exception, start_time, end_time
) # DO NOT MAKE THREADED - router retry fallback relies on this!
@ -2279,7 +2281,7 @@ def client(original_function):
# Reuse the timestamp taken right when the deployment call itself failed, before
# the failure hook ran, so a slow callback doesn't inflate the reported duration.
end_time = _deployment_call_end_time if _deployment_call_end_time is not None else datetime.datetime.now() # noqa: DTZ005 # matches the naive datetimes this whole function already times start_time/end_time with
if logging_obj and not _is_litellm_internal_call:
if logging_obj and not _is_litellm_internal_call and not is_resent_failure(e):
unbind_budget_reservation_from_callbacks(logging_obj.litellm_params)
try:
logging_obj.failure_handler(
@ -2300,7 +2302,10 @@ def client(original_function):
is_acompletion_litellm_router_call: Final = _is_litellm_router_call(kwargs, is_async=True)
if (
num_retries and not is_acompletion_litellm_router_call and not isinstance(e, ImportError)
num_retries
and not is_acompletion_litellm_router_call
and not isinstance(e, ImportError)
and not is_resent_failure(e)
): # only enter this if call is not from litellm router/proxy. router has it's own logic for retrying
try:
litellm.num_retries = None # set retries to None to prevent infinite loops
@ -2311,8 +2316,9 @@ def client(original_function):
elif isinstance(e, openai.APIError): # generic api error
kwargs["retry_strategy"] = "constant_retry"
result = await litellm.acompletion_with_retries(*args, **kwargs)
except Exception:
pass
except Exception as retry_error:
if is_resent_failure(retry_error):
raise
else:
return result
elif (
@ -2331,7 +2337,10 @@ def client(original_function):
is_aresponses_litellm_router_call: Final = _is_litellm_router_call(kwargs, is_async=True)
if (
num_retries and not is_aresponses_litellm_router_call and not isinstance(e, ImportError)
num_retries
and not is_aresponses_litellm_router_call
and not isinstance(e, ImportError)
and not is_resent_failure(e)
): # only enter this if call is not from litellm router/proxy. router has it's own logic for retrying
try:
litellm.num_retries = None # set retries to None to prevent infinite loops
@ -2342,8 +2351,9 @@ def client(original_function):
elif isinstance(e, openai.APIError): # generic api error
kwargs["retry_strategy"] = "constant_retry"
result = await litellm.aresponses_with_retries(*args, **kwargs)
except Exception:
pass
except Exception as retry_error:
if is_resent_failure(retry_error):
raise
else:
return result

113
tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_handler.py
View file

@ -1,7 +1,12 @@
import json
from datetime import datetime
from pathlib import Path
from typing import Final
from unittest.mock import AsyncMock, MagicMock, patch
import httpx
import pytest
import respx
import litellm
@ -11,6 +16,7 @@ from litellm.completion_extras.litellm_responses_transformation.handler import (
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLogging
from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper
from litellm.types.router import GenericLiteLLMParams
from litellm.litellm_core_utils.prompt_templates.common_utils import encrypted_reasoning_signature
from litellm.types.utils import ModelResponse
@ -335,3 +341,110 @@ async def test_acompletion_keeps_provider_native_model_id_through_responses(
handed_model = responses_call.call_args.kwargs["model"]
assert _upstream_model_for(handed_model, custom_llm_provider) == expected_upstream_model
_CHATGPT_RESPONSES_URL: Final = "https://chatgpt.com/backend-api/codex/responses"
def _reasoning_item(item_id: str) -> dict[str, object]:
return {"type": "reasoning", "id": item_id, "summary": [], "encrypted_content": f"gAAAA_{item_id}"}
def _weather_call(index: int, city: str) -> dict[str, object]:
return {
"type": "function_call",
"id": f"fc_{index}",
"status": "completed",
"arguments": json.dumps({"city": city}),
"call_id": f"call_{index}",
"name": "get_weather",
}
def _two_reasoning_tool_turn_stream() -> httpx.Response:
turn: Final = {
"id": "resp_two_reasoning",
"object": "response",
"created_at": 1,
"status": "completed",
"model": "gpt-5.6-sol",
"output": [
_reasoning_item("rs_first"),
_weather_call(1, "Oslo"),
_reasoning_item("rs_second"),
_weather_call(2, "Lyon"),
],
"usage": {"input_tokens": 10, "output_tokens": 20, "total_tokens": 30},
}
body: Final = f"data: {json.dumps({'type': 'response.completed', 'response': turn})}\n\ndata: [DONE]\n\n"
return httpx.Response(200, headers={"content-type": "text/event-stream"}, text=body)
_FAR_FUTURE_EXPIRY: Final = 4_102_444_800
@pytest.fixture
def _chatgpt_backend(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter) -> None:
auth: Final = {"access_token": "chatgpt-test-token", "expires_at": _FAR_FUTURE_EXPIRY, "account_id": "acct_test"}
(tmp_path / "auth.json").write_text(json.dumps(auth))
monkeypatch.setenv("CHATGPT_TOKEN_DIR", str(tmp_path))
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
litellm.in_memory_llm_clients_cache.flush_cache()
respx_mock.post(_CHATGPT_RESPONSES_URL).mock(side_effect=lambda _request: _two_reasoning_tool_turn_stream())
_WEATHER_QUESTION: Final = "Weather in Oslo and Lyon?"
_WEATHER_TOOL_INPUT: Final = {"type": "object", "properties": {"city": {"type": "string"}}}
_EVERY_REASONING_SIGNATURE: Final = [
encrypted_reasoning_signature("gAAAA_rs_first"),
encrypted_reasoning_signature("gAAAA_rs_second"),
]
def _thinking_signatures(reply: object) -> list[object]:
assert isinstance(reply, dict)
return [block.get("data") for block in reply["content"] if block["type"] == "redacted_thinking"]
@pytest.mark.asyncio
@pytest.mark.usefixtures("_chatgpt_backend")
async def test_async_messages_client_gets_every_reasoning_item_of_a_non_streamed_turn() -> None:
reply: Final = await litellm.anthropic.messages.acreate(
model="chatgpt/gpt-5.6-sol",
max_tokens=64,
messages=[{"role": "user", "content": _WEATHER_QUESTION}],
tools=[{"name": "get_weather", "input_schema": _WEATHER_TOOL_INPUT}],
)
assert _thinking_signatures(reply) == _EVERY_REASONING_SIGNATURE
_CHAT_WEATHER_TOOL: Final = {"type": "function", "function": {"name": "get_weather", "parameters": _WEATHER_TOOL_INPUT}}
def _reasoning_item_ids(response: object) -> list[object]:
assert isinstance(response, ModelResponse)
return [item["id"] for item in response.choices[0].message.reasoning_items]
@pytest.mark.asyncio
@pytest.mark.usefixtures("_chatgpt_backend")
async def test_async_chat_client_keeps_only_the_last_reasoning_item_of_a_turn() -> None:
response: Final = await litellm.acompletion(
model="chatgpt/gpt-5.6-sol",
messages=[{"role": "user", "content": _WEATHER_QUESTION}],
tools=[_CHAT_WEATHER_TOOL],
)
assert _reasoning_item_ids(response) == ["rs_second"]
@pytest.mark.usefixtures("_chatgpt_backend")
def test_sync_chat_client_keeps_only_the_last_reasoning_item_of_a_turn() -> None:
response: Final = litellm.completion(
model="chatgpt/gpt-5.6-sol",
messages=[{"role": "user", "content": _WEATHER_QUESTION}],
tools=[_CHAT_WEATHER_TOOL],
)
assert _reasoning_item_ids(response) == ["rs_second"]

396
tests/unit/completion_extras/litellm_responses_transformation/test_completion_extras_litellm_responses_transformation_transformation.py
View file

@ -3240,6 +3240,402 @@ def test_reasoning_items_streaming_emitted_on_response_completed():
assert ri["summary"][0]["text"] == summary_text
def test_reasoning_output_item_done_emits_signed_thinking_blocks_the_stream_keeps():
"""A chatgpt /v1/messages stream drops the reasoning item unless this delta carries thinking_blocks.
response.output_item.done is the only event that has encrypted_content. The chat delta has to
expose it as a signed thinking block, and that block alone has to count as stream content.
"""
from unittest.mock import MagicMock
from litellm.completion_extras.litellm_responses_transformation.transformation import (
OpenAiResponsesToChatCompletionStreamIterator,
)
from litellm.litellm_core_utils.prompt_templates.common_utils import encrypted_reasoning_signature
from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper
from litellm.llms.anthropic.pass_through.adapters.transformation import LiteLLMAnthropicMessagesAdapter
encrypted: Final = "gAAAA_round_trip"
done: Final = {
"type": "response.output_item.done",
"output_index": 0,
"item": {
"type": "reasoning",
"id": "rs_done",
"encrypted_content": encrypted,
"summary": [],
},
}
result: Final = OpenAiResponsesToChatCompletionStreamIterator(
streaming_response=None, sync_stream=True, streams_reasoning_signatures=True
).chunk_parser(done)
blocks: Final = result.choices[0].delta.thinking_blocks
assert blocks is not None
assert blocks[0]["type"] == "redacted_thinking"
assert blocks[0]["data"] == encrypted_reasoning_signature(encrypted)
assert blocks[0]["index"] == 0
assert getattr(result.choices[0].delta, "reasoning_content", None) in (None, "")
wrapper: Final = CustomStreamWrapper(
completion_stream=None,
model="chatgpt/gpt-6.1-sol",
logging_obj=MagicMock(),
custom_llm_provider="chatgpt",
)
assert wrapper.is_chunk_non_empty({}, result, {}) is True
block_type, start = LiteLLMAnthropicMessagesAdapter().translate_streaming_openai_chunk_to_anthropic_content_block(
choices=result.choices
)
assert block_type == "redacted_thinking"
assert start["data"] == encrypted_reasoning_signature(encrypted)
_SIGNED_REASONING_STREAM_BODY: Final[Mapping[str, object]] = {
"id": "resp_signed_reasoning",
"object": "response",
"created_at": 1789788253,
"status": "completed",
"model": "gpt-6-sol",
"output": [
{"type": "reasoning", "id": "rs_signed", "summary": [], "encrypted_content": "gAAAA_consumer"},
{
"type": "message",
"id": "msg_logged_call_type",
"status": "completed",
"role": "assistant",
"content": [{"type": "output_text", "text": "pong", "annotations": []}],
},
],
"usage": {"input_tokens": 10, "output_tokens": 5, "total_tokens": 15},
}
_GET_WEATHER_TOOL: Final = {
"type": "function",
"function": {"name": "get_weather", "parameters": {"type": "object", "properties": {}}},
}
def _signed_reasoning_sse() -> str:
events: Final = (
{
"type": "response.created",
"response": {**_SIGNED_REASONING_STREAM_BODY, "status": "in_progress", "output": [], "usage": None},
},
{"type": "response.output_item.done", "output_index": 0, "item": _SIGNED_REASONING_STREAM_BODY["output"][0]},
{
"type": "response.output_text.delta",
"item_id": "msg_logged_call_type",
"output_index": 1,
"content_index": 0,
"delta": "pong",
},
{"type": "response.completed", "response": _SIGNED_REASONING_STREAM_BODY},
)
return "".join(f"event: {event['type']}\ndata: {json.dumps(event)}\n\n" for event in events)
_AZURE_BRIDGE_KWARGS: Final[Mapping[str, object]] = {
"api_base": "https://example.invalid",
"api_version": "2025-04-01-preview",
"api_key": "test-key",
}
@pytest.mark.asyncio
async def test_streamed_reasoning_signature_reaches_messages_clients_and_never_chat_completions_deltas(
monkeypatch: pytest.MonkeyPatch,
) -> None:
from litellm.litellm_core_utils.prompt_templates.common_utils import encrypted_reasoning_signature
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
with respx.mock(assert_all_called=True) as router:
router.post(url__startswith="https://example.invalid/openai/responses").mock(
side_effect=lambda _request: httpx.Response(
200, text=_signed_reasoning_sse(), headers={"content-type": "text/event-stream"}
)
)
chat_stream: Final = await litellm.acompletion(
model="azure/gpt-6-sol",
messages=[{"role": "user", "content": "ping"}],
tools=[_GET_WEATHER_TOOL],
stream=True,
**_AZURE_BRIDGE_KWARGS,
)
chat_deltas: Final = [chunk.choices[0].delta async for chunk in chat_stream if chunk.choices]
messages_stream: Final = await litellm.anthropic.messages.acreate(
model="azure/gpt-6-sol",
messages=[{"role": "user", "content": "ping"}],
tools=[{"name": "get_weather", "input_schema": {"type": "object", "properties": {}}}],
max_tokens=64,
stream=True,
**_AZURE_BRIDGE_KWARGS,
)
sse: Final = b"".join([chunk async for chunk in messages_stream]).decode()
assert [getattr(delta, "thinking_blocks", None) for delta in chat_deltas if getattr(delta, "thinking_blocks", None)] == []
assert "".join(delta.content or "" for delta in chat_deltas) == "pong"
messages_events: Final = [json.loads(line[len("data: ") :]) for line in sse.splitlines() if line.startswith("data: ")]
started: Final = [event["content_block"] for event in messages_events if event.get("type") == "content_block_start"]
assert [block["type"] for block in started] == ["redacted_thinking", "text"]
assert started[0]["data"] == encrypted_reasoning_signature("gAAAA_consumer")
def test_reasoning_output_item_done_with_summary_signs_the_thinking_text():
from litellm.completion_extras.litellm_responses_transformation.transformation import (
OpenAiResponsesToChatCompletionStreamIterator,
)
from litellm.litellm_core_utils.prompt_templates.common_utils import encrypted_reasoning_signature
encrypted: Final = "gAAAA_with_summary"
done: Final = {
"type": "response.output_item.done",
"output_index": 2,
"item": {
"type": "reasoning",
"id": "rs_sum",
"encrypted_content": encrypted,
"summary": [{"type": "summary_text", "text": "Oslo is colder."}],
},
}
result: Final = OpenAiResponsesToChatCompletionStreamIterator(
streaming_response=None, sync_stream=True, streams_reasoning_signatures=True
).chunk_parser(done)
block: Final = result.choices[0].delta.thinking_blocks[0]
assert block["type"] == "thinking"
assert block["thinking"] == "Oslo is colder."
assert block["signature"] == encrypted_reasoning_signature(encrypted)
assert block["index"] == 2
def test_streamed_reasoning_summary_reaches_messages_clients_once_with_its_signature():
from litellm.completion_extras.litellm_responses_transformation.transformation import (
OpenAiResponsesToChatCompletionStreamIterator,
)
from litellm.litellm_core_utils.prompt_templates.common_utils import encrypted_reasoning_signature
from litellm.llms.anthropic.pass_through.adapters.streaming_iterator import AnthropicStreamWrapper
from litellm.types.utils import Delta, ModelResponseStream, StreamingChoices
encrypted: Final = "gAAAA_streamed_summary"
events: Final = (
{"type": "response.reasoning_summary_part.added", "output_index": 0, "summary_index": 0},
{"type": "response.reasoning_summary_text.delta", "output_index": 0, "delta": "Oslo is "},
{"type": "response.reasoning_summary_text.delta", "output_index": 0, "delta": "colder."},
{
"type": "response.output_item.done",
"output_index": 0,
"item": {
"type": "reasoning",
"id": "rs_streamed",
"encrypted_content": encrypted,
"summary": [{"type": "summary_text", "text": "Oslo is colder."}],
},
},
{"type": "response.output_text.delta", "output_index": 1, "delta": "Oslo."},
)
parser: Final = OpenAiResponsesToChatCompletionStreamIterator(
streaming_response=None, sync_stream=True, streams_reasoning_signatures=True
)
chunks: Final = (
*(parser.chunk_parser(event) for event in events),
ModelResponseStream(choices=[StreamingChoices(index=0, delta=Delta(), finish_reason="stop")]),
)
sse_events: Final = list(AnthropicStreamWrapper(completion_stream=iter(chunks), model="chatgpt/gpt-5.5"))
deltas: Final = [event["delta"] for event in sse_events if event.get("type") == "content_block_delta"]
assert "".join(d.get("thinking", "") for d in deltas if d["type"] == "thinking_delta") == "Oslo is colder."
assert [d["signature"] for d in deltas if d["type"] == "signature_delta"] == [
encrypted_reasoning_signature(encrypted)
]
def test_reasoning_item_repeated_at_stream_end_reaches_messages_clients_once():
from litellm.completion_extras.litellm_responses_transformation.transformation import (
OpenAiResponsesToChatCompletionStreamIterator,
)
from litellm.litellm_core_utils.prompt_templates.common_utils import encrypted_reasoning_signature
from litellm.llms.anthropic.pass_through.adapters.streaming_iterator import AnthropicStreamWrapper
from litellm.types.utils import Delta, ModelResponseStream, StreamingChoices
encrypted: Final = "gAAAA_completed_repeat"
reasoning_item: Final = {"type": "reasoning", "id": "rs_repeat", "summary": [], "encrypted_content": encrypted}
events: Final = (
{"type": "response.output_item.done", "output_index": 0, "item": reasoning_item},
{"type": "response.output_text.delta", "output_index": 1, "delta": "Oslo."},
)
parser: Final = OpenAiResponsesToChatCompletionStreamIterator(
streaming_response=None, sync_stream=True, streams_reasoning_signatures=True
)
chunks: Final = (
*(parser.chunk_parser(event) for event in events),
ModelResponseStream(
choices=[StreamingChoices(index=0, delta=Delta(reasoning_items=[reasoning_item]), finish_reason=None)]
),
ModelResponseStream(choices=[StreamingChoices(index=0, delta=Delta(), finish_reason="stop")]),
)
sse_events: Final = list(AnthropicStreamWrapper(completion_stream=iter(chunks), model="azure/gpt-5.4-nano"))
started: Final = [event["content_block"] for event in sse_events if event.get("type") == "content_block_start"]
assert [block["type"] for block in started] == ["redacted_thinking", "text"]
assert started[0]["data"] == encrypted_reasoning_signature(encrypted)
def test_reasoning_output_item_done_without_encrypted_content_stays_empty():
from litellm.completion_extras.litellm_responses_transformation.transformation import (
OpenAiResponsesToChatCompletionStreamIterator,
)
done: Final = {
"type": "response.output_item.done",
"output_index": 0,
"item": {
"type": "reasoning",
"id": "rs_plain",
"summary": [{"type": "summary_text", "text": "already streamed"}],
},
}
result: Final = OpenAiResponsesToChatCompletionStreamIterator(
streaming_response=None, sync_stream=True, streams_reasoning_signatures=True
).chunk_parser(done)
assert getattr(result.choices[0].delta, "thinking_blocks", None) in (None, [])
def test_transform_response_hands_encrypted_reasoning_to_messages_clients_only():
from unittest.mock import Mock
from litellm.litellm_core_utils.prompt_templates.common_utils import encrypted_reasoning_signature
from litellm.llms.anthropic.pass_through.adapters.transformation import (
LiteLLMAnthropicMessagesAdapter,
)
from litellm.types.llms.openai import ResponsesAPIResponse
from litellm.types.utils import ModelResponse, Usage
encrypted: Final = "gAAAA_non_stream"
reasoning: Final = ResponseReasoningItem(
id="rs_tool",
summary=[],
type="reasoning",
content=None,
encrypted_content=encrypted,
status=None,
)
function_call: Final = ResponseFunctionToolCall(
id="fc_1",
type="function_call",
status="completed",
arguments='{"city":"Oslo"}',
call_id="call_1",
name="get_weather",
)
raw: Final = ResponsesAPIResponse.model_construct(
id="resp_tool",
created_at=1,
error=None,
incomplete_details=None,
model="gpt-6.1-sol",
output=[reasoning, function_call],
status="completed",
usage=None,
)
model_response: Final = ModelResponse(
id="chatcmpl-tool",
created=1,
model=None,
object="chat.completion",
system_fingerprint=None,
choices=[],
usage=Usage(completion_tokens=0, prompt_tokens=0, total_tokens=0),
)
result: Final = LiteLLMResponsesTransformationHandler().transform_response(
model="gpt-6.1-sol",
raw_response=raw,
model_response=model_response,
logging_obj=Mock(),
request_data={"model": "gpt-6.1-sol"},
messages=[{"role": "user", "content": "Which city?"}],
optional_params={},
litellm_params={},
encoding=Mock(),
)
message: Final = result.choices[0].message
assert message.reasoning_items[0]["encrypted_content"] == encrypted
assert "thinking_blocks" not in result.model_dump()["choices"][0]["message"]
assert message.tool_calls[0].function.name == "get_weather"
anthropic_response: Final = LiteLLMAnthropicMessagesAdapter().translate_openai_response_to_anthropic(
response=result, tool_name_mapping=None
)
content: Final = anthropic_response["content"]
assert [block["type"] for block in content] == ["redacted_thinking", "tool_use"]
assert content[0]["data"] == encrypted_reasoning_signature(encrypted)
@pytest.mark.parametrize(
("keeps_every_reasoning_item", "kept_ids"),
[
pytest.param(True, ["rs_first", "rs_second"], id="messages_client_keeps_every_item"),
pytest.param(False, ["rs_second"], id="chat_client_keeps_the_last_item"),
],
)
def test_non_streamed_tool_turn_hands_messages_clients_every_reasoning_item(
keeps_every_reasoning_item: bool, kept_ids: list[str]
) -> None:
from unittest.mock import Mock
from litellm.types.llms.openai import ResponsesAPIResponse
from litellm.types.utils import ModelResponse, Usage
def reasoning(item_id: str) -> ResponseReasoningItem:
return ResponseReasoningItem(
id=item_id, summary=[], type="reasoning", content=None, encrypted_content=f"gAAAA_{item_id}", status=None
)
def call(index: int, city: str) -> ResponseFunctionToolCall:
return ResponseFunctionToolCall(
id=f"fc_{index}",
type="function_call",
status="completed",
arguments=json.dumps({"city": city}),
call_id=f"call_{index}",
name="get_weather",
)
raw: Final = ResponsesAPIResponse.model_construct(
id="resp_two_reasoning",
created_at=1,
error=None,
incomplete_details=None,
model="gpt-5.4-nano",
output=[reasoning("rs_first"), call(1, "Oslo"), reasoning("rs_second"), call(2, "Lyon")],
status="completed",
usage=None,
)
result: Final = LiteLLMResponsesTransformationHandler().transform_response(
model="gpt-5.4-nano",
raw_response=raw,
model_response=ModelResponse(
id="chatcmpl-two-reasoning",
created=1,
model=None,
object="chat.completion",
system_fingerprint=None,
choices=[],
usage=Usage(completion_tokens=0, prompt_tokens=0, total_tokens=0),
),
logging_obj=Mock(),
request_data={"model": "gpt-5.4-nano"},
messages=[{"role": "user", "content": "Oslo or Lyon?"}],
optional_params={},
litellm_params={"custom_llm_provider": "azure"},
encoding=Mock(),
keeps_every_reasoning_item=keeps_every_reasoning_item,
)
message: Final = result.choices[0].message
assert [item["id"] for item in message.reasoning_items] == kept_ids
assert [tool_call.function.name for tool_call in message.tool_calls] == ["get_weather", "get_weather"]
def test_streaming_function_call_tool_id_for_degenerate_call_id():
"""In streaming, Bedrock Mantle's function_call event carries a unique ``id``
(``fc_...``) and a non-unique, index-based ``call_id`` (``call_0``). For that

31
tests/unit/litellm_core_utils/test_streaming_handler.py
View file

@ -246,6 +246,37 @@ def test_is_chunk_non_empty(initialized_custom_stream_wrapper: CustomStreamWrapp
)
def test_is_chunk_non_empty_keeps_thinking_blocks_without_other_content(
initialized_custom_stream_wrapper: CustomStreamWrapper,
):
"""Encrypted reasoning arrives as thinking_blocks and nothing else, so that delta has to be yielded."""
chunk: Final = ModelResponseStream(
id="chunk-thinking",
object="chat.completion.chunk",
created=1,
model="chatgpt/gpt-6.1-sol",
choices=[
StreamingChoices(
index=0,
finish_reason=None,
delta=Delta(
thinking_blocks=[
{"type": "redacted_thinking", "data": "litellm_encrypted_reasoning:gAAAA"}
]
),
)
],
)
assert (
initialized_custom_stream_wrapper.is_chunk_non_empty(
completion_obj={},
model_response=chunk,
response_obj={},
)
is True
)
def test_is_chunk_non_empty_with_annotations(
initialized_custom_stream_wrapper: CustomStreamWrapper,
):

794
tests/unit/llms/anthropic/pass_through/adapters/test_anthropic_experimental_pass_through_adapters_transformation.py
View file

@ -1,15 +1,18 @@
import base64
import functools
import json
import logging
from collections.abc import Mapping, Sequence
from contextlib import nullcontext
from types import MappingProxyType
from typing import Any, Final, cast
import httpx
import pytest
import respx
import litellm
from litellm.integrations.custom_logger import CustomLogger
from litellm.litellm_core_utils.prompt_templates.common_utils import (
TOOL_RESULT_IMAGE_PLACEHOLDER,
encrypted_reasoning_signature,
@ -31,6 +34,7 @@ from litellm.llms.anthropic.pass_through.messages.mid_conversation_system import
from litellm.llms.openai.chat.gpt_transformation import OpenAIGPTConfig
from litellm.types.llms.anthropic import (
AnthopicMessagesAssistantMessageParam,
AnthropicMessagesRequest,
AnthropicMessagesUserMessageParam,
)
from litellm.types.llms.openai import ChatCompletionAssistantToolCall
@ -509,6 +513,218 @@ def test_translate_anthropic_messages_to_openai_drops_bridge_encrypted_reasoning
assert [b["signature"] for b in result[2]["thinking_blocks"]] == ["EqQBCkYIAxgCIkA_signed"]
def test_chatgpt_messages_keep_encrypted_reasoning_and_replay_it():
"""chatgpt /v1/messages ends in the Responses API, so the signed block has to survive into the next request."""
from litellm.completion_extras.litellm_responses_transformation.transformation import (
LiteLLMResponsesTransformationHandler,
)
summary_encrypted: Final = "gAAAA_summary"
bytes_encrypted: Final = "gAAAA_bytes"
messages: Final = [
AnthropicMessagesUserMessageParam(role="user", content="Which city is colder?"),
AnthopicMessagesAssistantMessageParam(
role="assistant",
content=[
{
"type": "thinking",
"thinking": "Oslo is colder.",
"signature": encrypted_reasoning_signature(summary_encrypted),
},
{"type": "redacted_thinking", "data": encrypted_reasoning_signature(bytes_encrypted)},
{"type": "tool_use", "id": "call_1", "name": "get_weather", "input": {"city": "Oslo"}},
],
),
AnthropicMessagesUserMessageParam(
role="user",
content=[{"type": "tool_result", "tool_use_id": "call_1", "content": "-7 C"}],
),
]
kept: Final = LiteLLMAnthropicMessagesAdapter().translate_anthropic_messages_to_openai(
messages=messages,
model="chatgpt/gpt-6.1-sol",
custom_llm_provider="chatgpt",
)
assistant: Final = kept[1]
assert [block["type"] for block in assistant["thinking_blocks"]] == ["thinking", "redacted_thinking"]
assert assistant["thinking_blocks"][0]["signature"] == encrypted_reasoning_signature(summary_encrypted)
assert assistant["thinking_blocks"][1]["data"] == encrypted_reasoning_signature(bytes_encrypted)
responses_input, _instructions = LiteLLMResponsesTransformationHandler().convert_chat_completion_messages_to_responses_api(
kept
)
reasoning: Final = tuple(item for item in responses_input if item.get("type") == "reasoning")
assert [item["encrypted_content"] for item in reasoning] == [summary_encrypted, bytes_encrypted]
types: Final = [item.get("type") for item in responses_input]
assert types.index("reasoning") < types.index("function_call")
stripped: Final = LiteLLMAnthropicMessagesAdapter().translate_anthropic_messages_to_openai(
messages=messages,
model="gpt-4o",
custom_llm_provider="openai",
)
assert not stripped[1].get("thinking_blocks")
prefixed: Final = LiteLLMAnthropicMessagesAdapter().translate_anthropic_messages_to_openai(
messages=messages,
model="responses/gpt-5-mini",
custom_llm_provider="openai",
)
assert prefixed[1]["thinking_blocks"][1]["data"] == encrypted_reasoning_signature(bytes_encrypted)
def test_translate_openai_reasoning_items_become_signed_thinking_blocks():
encrypted: Final = "gAAAA_outbound"
openai_choices: Final = [
Choices(
message=Message(
role="assistant",
content="Oslo.",
reasoning_items=[
{
"type": "reasoning",
"id": "rs_1",
"encrypted_content": encrypted,
"summary": [{"type": "summary_text", "text": "Oslo is colder."}],
}
],
tool_calls=[
ChatCompletionAssistantToolCall(
id="call_1",
type="function",
function=Function(name="get_weather", arguments='{"city": "Oslo"}'),
)
],
)
)
]
result: Final = LiteLLMAnthropicMessagesAdapter()._translate_openai_content_to_anthropic(choices=openai_choices)
assert result[0]["type"] == "thinking"
assert result[0]["thinking"] == "Oslo is colder."
assert result[0]["signature"] == encrypted_reasoning_signature(encrypted)
assert result[1]["type"] == "text"
assert result[2]["type"] == "tool_use"
def test_translate_openai_reasoning_items_without_summary_become_redacted_thinking():
encrypted: Final = "gAAAA_bytes_only"
openai_choices: Final = [
Choices(
message=Message(
role="assistant",
content=None,
reasoning_items=[
{"type": "reasoning", "id": "rs_1", "encrypted_content": encrypted, "summary": []}
],
)
)
]
result: Final = LiteLLMAnthropicMessagesAdapter()._translate_openai_content_to_anthropic(choices=openai_choices)
assert result == [
{"type": "redacted_thinking", "data": encrypted_reasoning_signature(encrypted)},
]
def test_translate_openai_reasoning_items_are_not_duplicated_when_thinking_blocks_already_carry_them():
encrypted: Final = "gAAAA_once"
signature: Final = encrypted_reasoning_signature(encrypted)
openai_choices: Final = [
Choices(
message=Message(
role="assistant",
content="Oslo.",
thinking_blocks=[{"type": "redacted_thinking", "data": signature}],
reasoning_items=[
{"type": "reasoning", "id": "rs_1", "encrypted_content": encrypted, "summary": []}
],
)
)
]
result: Final = LiteLLMAnthropicMessagesAdapter()._translate_openai_content_to_anthropic(choices=openai_choices)
assert [block["type"] for block in result] == ["redacted_thinking", "text"]
def _replayed_reasoning_request(tools: list[dict[str, object]]) -> AnthropicMessagesRequest:
return AnthropicMessagesRequest(
model="azure/gpt-5.4-nano",
max_tokens=64,
messages=[
AnthropicMessagesUserMessageParam(role="user", content="Which city is colder?"),
AnthopicMessagesAssistantMessageParam(
role="assistant",
content=[
{"type": "redacted_thinking", "data": encrypted_reasoning_signature("gAAAA_azure_tools")},
{"type": "tool_use", "id": "call_1", "name": "get_weather", "input": {"city": "Oslo"}},
],
),
AnthropicMessagesUserMessageParam(
role="user",
content=[{"type": "tool_result", "tool_use_id": "call_1", "content": "-7 C"}],
),
],
tools=tools,
)
@pytest.mark.parametrize(
("tools", "kept"),
[
pytest.param(
[{"name": "get_weather", "input_schema": {"type": "object", "properties": {}}}],
True,
id="function_tools_bridge_to_responses",
),
pytest.param([], False, id="no_tools_stay_on_chat"),
],
)
def test_encrypted_reasoning_is_kept_when_the_request_tools_route_it_to_responses(
tools: list[dict[str, object]], kept: bool
) -> None:
openai_request, _mapping = LiteLLMAnthropicMessagesAdapter().translate_anthropic_to_openai(
_replayed_reasoning_request(tools),
custom_llm_provider="azure",
api_base="https://keep-check.openai.azure.com",
)
assistant: Final = openai_request["messages"][1]
assert bool(assistant.get("thinking_blocks")) is kept
@pytest.mark.parametrize(
("deployment_params", "kept"),
[
pytest.param({"reasoning_summary": "auto"}, True, id="deployment_summary_bridges_to_responses"),
pytest.param({"extra_body": {"reasoningSummary": "auto"}}, True, id="extra_body_summary_bridges_to_responses"),
pytest.param({}, False, id="no_summary_stays_on_chat"),
],
)
def test_encrypted_reasoning_is_kept_when_the_deployment_reasoning_summary_routes_it_to_responses(
deployment_params: dict[str, object], kept: bool
) -> None:
from litellm.llms.anthropic.pass_through.adapters.handler import (
LiteLLMMessagesToCompletionTransformationHandler,
)
request: Final = _replayed_reasoning_request([])
completion_kwargs, _mapping = LiteLLMMessagesToCompletionTransformationHandler._prepare_completion_kwargs(
max_tokens=64,
messages=cast(list[dict[str, object]], request["messages"]),
model="azure/gpt-5.1",
thinking={"type": "enabled", "budget_tokens": 2048},
extra_kwargs={"api_base": "https://keep-check.openai.azure.com", **deployment_params},
)
assistant: Final = cast(list[Mapping[str, object]], completion_kwargs["messages"])[1]
assert bool(assistant.get("thinking_blocks")) is kept
def test_translate_anthropic_messages_to_openai_sets_reasoning_content():
"""Reasoning-aware chat providers read reasoning_content, so thinking text must land there.
@ -5235,3 +5451,575 @@ def test_eager_input_streaming_tool_reaches_bedrock_converse_as_beta():
assert data["additionalModelRequestFields"]["anthropic_beta"] == ["fine-grained-tool-streaming-2025-05-14"]
assert data["toolConfig"]["tools"][0]["toolSpec"]["inputSchema"]["json"] == EAGER_INPUT_SCHEMA
@pytest.fixture
def _chat_bridge_httpx(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
litellm.in_memory_llm_clients_cache.flush_cache()
_RETRY_API_BASE: Final = "https://encrypted-retry.openai.azure.com"
_RETRY_URL: Final = r".*encrypted-retry\.openai\.azure\.com/openai/.*responses.*"
_RETRY_TOOLS: Final = [{"name": "get_weather", "input_schema": {"type": "object", "properties": {}}}]
_RETRY_REJECTION: Final = httpx.Response(
400,
json={
"error": {
"message": "The encrypted content gAAA_foreign could not be verified.",
"type": "invalid_request_error",
"param": None,
"code": "invalid_encrypted_content",
}
},
)
_UNRELATED_REJECTION: Final = httpx.Response(
400, json={"error": {"message": "max_output_tokens is too small", "type": "invalid_request_error"}}
)
_UNRELATED_UNVERIFIED_REJECTION: Final = httpx.Response(
400, json={"error": {"message": "The organization could not be verified.", "type": "invalid_request_error"}}
)
_RETRY_REPLY: Final = {
"id": "resp_retry",
"object": "response",
"created_at": 1700000000,
"status": "completed",
"model": "gpt-5.4-nano",
"output": [
{
"type": "message",
"id": "msg_retry",
"status": "completed",
"role": "assistant",
"content": [{"type": "output_text", "text": "Oslo", "annotations": []}],
}
],
"usage": {"input_tokens": 10, "output_tokens": 2, "total_tokens": 12},
"parallel_tool_calls": True,
"tool_choice": "auto",
"tools": [],
}
def _history_with_foreign_reasoning() -> list[dict[str, object]]:
return [
{"role": "user", "content": "Which city is colder?"},
{
"role": "assistant",
"content": [
{"type": "redacted_thinking", "data": encrypted_reasoning_signature("gAAA_foreign")},
{"type": "tool_use", "id": "call_1", "name": "get_weather", "input": {"city": "Oslo"}},
],
},
{"role": "user", "content": [{"type": "tool_result", "tool_use_id": "call_1", "content": "-7 C"}]},
]
def _sent_reasoning_items(request: httpx.Request) -> list[object]:
sent: Final = json.loads(request.read())
return [item for item in sent["input"] if item.get("type") == "reasoning"]
@pytest.mark.asyncio
@pytest.mark.usefixtures("_chat_bridge_httpx")
async def test_async_messages_retry_once_without_reasoning_the_backend_cannot_decrypt(
respx_mock: respx.MockRouter,
caplog: pytest.LogCaptureFixture,
) -> None:
route: Final = respx_mock.post(url__regex=_RETRY_URL).mock(
side_effect=[_RETRY_REJECTION, httpx.Response(200, json=_RETRY_REPLY)]
)
with caplog.at_level(logging.WARNING, logger="LiteLLM"):
response: Final = await litellm.anthropic.messages.acreate(
model="azure/gpt-5.4-nano",
api_base=_RETRY_API_BASE,
api_key="fake-key",
api_version="2025-04-01-preview",
max_tokens=64,
tools=_RETRY_TOOLS,
messages=_history_with_foreign_reasoning(),
)
warnings: Final = tuple(
record.getMessage()
for record in caplog.records
if record.levelno >= logging.WARNING and record.name == "LiteLLM"
)
retry_warnings: Final = tuple(message for message in warnings if "retrying once without it" in message)
assert route.call_count == 2
assert len(_sent_reasoning_items(route.calls[0].request)) == 1
assert _sent_reasoning_items(route.calls[1].request) == []
assert [block["text"] for block in response["content"] if block["type"] == "text"] == ["Oslo"]
assert retry_warnings == (
"Backend rejected replayed encrypted reasoning, retrying once without it. "
"Enable the encrypted_content_affinity pre-call check to drop such blocks before the call. "
"provider=azure model=azure/gpt-5.4-nano status=400 code=invalid_encrypted_content",
)
assert all(
"gAAA_foreign" not in message and "litellm_encrypted_reasoning:" not in message for message in warnings
)
@pytest.mark.usefixtures("_chat_bridge_httpx")
def test_sync_messages_retry_once_without_reasoning_the_backend_cannot_decrypt(
respx_mock: respx.MockRouter,
) -> None:
route: Final = respx_mock.post(url__regex=_RETRY_URL).mock(
side_effect=[_RETRY_REJECTION, httpx.Response(200, json=_RETRY_REPLY)]
)
response: Final = litellm.anthropic.messages.create(
model="azure/gpt-5.4-nano",
api_base=_RETRY_API_BASE,
api_key="fake-key",
api_version="2025-04-01-preview",
max_tokens=64,
tools=_RETRY_TOOLS,
messages=_history_with_foreign_reasoning(),
)
assert route.call_count == 2
assert _sent_reasoning_items(route.calls[1].request) == []
assert [block["text"] for block in response["content"] if block["type"] == "text"] == ["Oslo"]
@pytest.mark.asyncio
@pytest.mark.usefixtures("_chat_bridge_httpx")
async def test_async_messages_with_num_retries_send_the_rejected_reasoning_only_once(
respx_mock: respx.MockRouter,
) -> None:
route: Final = respx_mock.post(url__regex=_RETRY_URL).mock(
side_effect=[_RETRY_REJECTION, httpx.Response(200, json=_RETRY_REPLY)]
)
await litellm.anthropic.messages.acreate(
model="azure/gpt-5.4-nano",
api_base=_RETRY_API_BASE,
api_key="fake-key",
api_version="2025-04-01-preview",
max_tokens=64,
tools=_RETRY_TOOLS,
messages=_history_with_foreign_reasoning(),
num_retries=2,
)
assert [len(_sent_reasoning_items(call.request)) for call in route.calls] == [1, 0]
@pytest.mark.usefixtures("_chat_bridge_httpx")
def test_sync_messages_with_global_num_retries_send_the_rejected_reasoning_once_and_keep_the_setting(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(litellm, "num_retries", 2)
route: Final = respx_mock.post(url__regex=_RETRY_URL).mock(
side_effect=[_RETRY_REJECTION, httpx.Response(200, json=_RETRY_REPLY)]
)
litellm.anthropic.messages.create(
model="azure/gpt-5.4-nano",
api_base=_RETRY_API_BASE,
api_key="fake-key",
api_version="2025-04-01-preview",
max_tokens=64,
tools=_RETRY_TOOLS,
messages=_history_with_foreign_reasoning(),
)
assert [len(_sent_reasoning_items(call.request)) for call in route.calls] == [1, 0]
assert litellm.num_retries == 2
_SERVER_ERROR: Final = httpx.Response(
500, json={"error": {"message": "The server had an error while processing your request.", "type": "server_error"}}
)
@pytest.mark.asyncio
@pytest.mark.usefixtures("_chat_bridge_httpx")
async def test_async_messages_retry_a_server_error_with_the_reasoning_and_drop_it_only_once_rejected(
respx_mock: respx.MockRouter,
) -> None:
route: Final = respx_mock.post(url__regex=_RETRY_URL).mock(
side_effect=[_SERVER_ERROR, _RETRY_REJECTION, httpx.Response(200, json=_RETRY_REPLY)]
)
response: Final = await litellm.anthropic.messages.acreate(
model="azure/gpt-5.4-nano",
api_base=_RETRY_API_BASE,
api_key="fake-key",
api_version="2025-04-01-preview",
max_tokens=64,
tools=_RETRY_TOOLS,
messages=_history_with_foreign_reasoning(),
num_retries=2,
)
assert [len(_sent_reasoning_items(call.request)) for call in route.calls] == [1, 1, 0]
assert [block["text"] for block in response["content"] if block["type"] == "text"] == ["Oslo"]
@pytest.mark.usefixtures("_chat_bridge_httpx")
def test_sync_messages_retry_a_server_error_with_the_reasoning_and_drop_it_only_once_rejected(
respx_mock: respx.MockRouter,
) -> None:
route: Final = respx_mock.post(url__regex=_RETRY_URL).mock(
side_effect=[_SERVER_ERROR, _RETRY_REJECTION, httpx.Response(200, json=_RETRY_REPLY)]
)
response: Final = litellm.anthropic.messages.create(
model="azure/gpt-5.4-nano",
api_base=_RETRY_API_BASE,
api_key="fake-key",
api_version="2025-04-01-preview",
max_tokens=64,
tools=_RETRY_TOOLS,
messages=_history_with_foreign_reasoning(),
num_retries=2,
)
assert [len(_sent_reasoning_items(call.request)) for call in route.calls] == [1, 1, 0]
assert [block["text"] for block in response["content"] if block["type"] == "text"] == ["Oslo"]
class _FailureLog(CustomLogger):
def __init__(self) -> None:
super().__init__()
self.rejections: tuple[str, ...] = ()
def _record(self, kwargs: Mapping[str, object]) -> None:
error: Final = str(kwargs.get("exception"))
self.rejections = (
*self.rejections,
next((phrase for phrase in _UPSTREAM_REJECTION_PHRASES if phrase in error), error),
)
def log_failure_event(
self, kwargs: Mapping[str, object], response_obj: object, start_time: object, end_time: object
) -> None:
self._record(kwargs)
async def async_log_failure_event(
self, kwargs: Mapping[str, object], response_obj: object, start_time: object, end_time: object
) -> None:
self._record(kwargs)
_UPSTREAM_REJECTION_PHRASES: Final = ("could not be verified", "max_output_tokens is too small")
_OUTCOMES_OF_THE_TURN_SENT_AGAIN: Final = (
pytest.param(httpx.Response(200, json=_RETRY_REPLY), (), id="served"),
pytest.param(_UNRELATED_REJECTION, ("max_output_tokens is too small",), id="rejected"),
)
@pytest.mark.asyncio
@pytest.mark.usefixtures("_chat_bridge_httpx")
@pytest.mark.parametrize(("sent_again", "logged_rejections"), _OUTCOMES_OF_THE_TURN_SENT_AGAIN)
async def test_async_messages_log_only_the_outcome_of_the_turn_sent_again(
respx_mock: respx.MockRouter,
monkeypatch: pytest.MonkeyPatch,
sent_again: httpx.Response,
logged_rejections: tuple[str, ...],
) -> None:
failure_log: Final = _FailureLog()
monkeypatch.setattr(litellm, "callbacks", [failure_log])
respx_mock.post(url__regex=_RETRY_URL).mock(side_effect=[_RETRY_REJECTION, sent_again])
with pytest.raises(litellm.BadRequestError) if logged_rejections else nullcontext():
await litellm.anthropic.messages.acreate(
model="azure/gpt-5.4-nano",
api_base=_RETRY_API_BASE,
api_key="fake-key",
api_version="2025-04-01-preview",
max_tokens=64,
tools=_RETRY_TOOLS,
messages=_history_with_foreign_reasoning(),
)
assert failure_log.rejections == logged_rejections
@pytest.mark.usefixtures("_chat_bridge_httpx")
@pytest.mark.parametrize(("sent_again", "logged_rejections"), _OUTCOMES_OF_THE_TURN_SENT_AGAIN)
def test_sync_messages_log_only_the_outcome_of_the_turn_sent_again(
respx_mock: respx.MockRouter,
monkeypatch: pytest.MonkeyPatch,
sent_again: httpx.Response,
logged_rejections: tuple[str, ...],
) -> None:
failure_log: Final = _FailureLog()
monkeypatch.setattr(litellm, "callbacks", [failure_log])
respx_mock.post(url__regex=_RETRY_URL).mock(side_effect=[_RETRY_REJECTION, sent_again])
with pytest.raises(litellm.BadRequestError) if logged_rejections else nullcontext():
litellm.anthropic.messages.create(
model="azure/gpt-5.4-nano",
api_base=_RETRY_API_BASE,
api_key="fake-key",
api_version="2025-04-01-preview",
max_tokens=64,
tools=_RETRY_TOOLS,
messages=_history_with_foreign_reasoning(),
)
assert failure_log.rejections == logged_rejections
@pytest.mark.asyncio
@pytest.mark.usefixtures("_chat_bridge_httpx")
@pytest.mark.parametrize(
("messages", "rejection"),
[
pytest.param(
[{"role": "user", "content": "Which city is colder?"}],
_RETRY_REJECTION,
id="no_encrypted_reasoning_to_drop",
),
pytest.param(_history_with_foreign_reasoning(), _UNRELATED_REJECTION, id="unrelated_rejection"),
pytest.param(
_history_with_foreign_reasoning(), _UNRELATED_UNVERIFIED_REJECTION, id="unrelated_unverified_rejection"
),
],
)
async def test_messages_do_not_retry_other_rejections(
respx_mock: respx.MockRouter, messages: list[dict[str, object]], rejection: httpx.Response
) -> None:
route: Final = respx_mock.post(url__regex=_RETRY_URL).mock(return_value=rejection)
with pytest.raises(litellm.BadRequestError):
await litellm.anthropic.messages.acreate(
model="azure/gpt-5.4-nano",
api_base=_RETRY_API_BASE,
api_key="fake-key",
api_version="2025-04-01-preview",
max_tokens=64,
tools=_RETRY_TOOLS,
messages=messages,
)
assert route.call_count == 1
_STRIPPED_CHAT_MODEL: Final = "azure/gpt-5.6"
_STRIPPED_CHAT_API_BASE: Final = "https://stripped-reasoning.openai.azure.com"
_STRIPPED_CHAT_URL: Final = r".*stripped-reasoning\.openai\.azure\.com/openai/deployments/gpt-5\.6/chat/completions.*"
@pytest.mark.asyncio
@pytest.mark.usefixtures("_chat_bridge_httpx")
async def test_messages_do_not_retry_when_the_translated_request_already_dropped_the_reasoning(
respx_mock: respx.MockRouter,
) -> None:
route: Final = respx_mock.post(url__regex=_STRIPPED_CHAT_URL).mock(return_value=_RETRY_REJECTION)
with pytest.raises(litellm.BadRequestError):
await litellm.anthropic.messages.acreate(
model=_STRIPPED_CHAT_MODEL,
api_base=_STRIPPED_CHAT_API_BASE,
api_key="fake-key",
max_tokens=64,
messages=_history_with_foreign_reasoning(),
)
assert route.call_count == 1
@pytest.mark.asyncio
@pytest.mark.usefixtures("_chat_bridge_httpx")
async def test_openai_thinking_on_the_chat_adapter_keeps_reasoning_its_responses_route_replays(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(litellm, "use_chat_completions_url_for_anthropic_messages", True)
route: Final = respx_mock.post(url__regex=r".*openai-thinking\.example/v1/responses").mock(
return_value=httpx.Response(200, json=_RETRY_REPLY)
)
await litellm.anthropic.messages.acreate(
model="openai/gpt-5-mini",
api_base="https://openai-thinking.example/v1",
api_key="fake-key",
max_tokens=2048,
thinking={"type": "enabled", "budget_tokens": 1024},
messages=[
{"role": "user", "content": "Which city is colder?"},
{
"role": "assistant",
"content": [
{"type": "redacted_thinking", "data": encrypted_reasoning_signature("gAAA_same_org")},
{"type": "text", "text": "Oslo."},
],
},
{"role": "user", "content": "And by how much?"},
],
)
assert route.call_count == 1
assert [item["encrypted_content"] for item in _sent_reasoning_items(route.calls[0].request)] == ["gAAA_same_org"]
def _responses_sse(events: Sequence[Mapping[str, object]]) -> bytes:
return b"".join(
f"event: {event['type']}\ndata: {json.dumps({**event, 'sequence_number': number})}\n\n".encode()
for number, event in enumerate(events)
)
def _reasoning_item(item_id: str, encrypted: str, summary: str = "") -> dict[str, object]:
return {
"type": "reasoning",
"id": item_id,
"summary": [{"type": "summary_text", "text": summary}] if summary else [],
"encrypted_content": encrypted,
}
@pytest.mark.asyncio
@pytest.mark.usefixtures("_chat_bridge_httpx")
async def test_streamed_consecutive_reasoning_items_each_become_their_own_block(
respx_mock: respx.MockRouter,
) -> None:
response: Final = {**_RETRY_REPLY, "id": "resp_two_reasoning", "status": "in_progress", "output": []}
message: Final = {
"type": "message",
"id": "msg_two_reasoning",
"status": "completed",
"role": "assistant",
"content": [{"type": "output_text", "text": "Oslo", "annotations": []}],
}
first: Final = _reasoning_item("rs_1", "gAAA_first")
second: Final = _reasoning_item("rs_2", "gAAA_second")
completed: Final = {**response, "status": "completed", "output": [first, second, message]}
events: Final = (
{"type": "response.created", "response": response},
{"type": "response.output_item.added", "output_index": 0, "item": _reasoning_item("rs_1", "")},
{"type": "response.output_item.done", "output_index": 0, "item": first},
{"type": "response.output_item.added", "output_index": 1, "item": _reasoning_item("rs_2", "")},
{"type": "response.output_item.done", "output_index": 1, "item": second},
{"type": "response.output_item.added", "output_index": 2, "item": {**message, "content": []}},
{
"type": "response.output_text.delta",
"output_index": 2,
"item_id": "msg_two_reasoning",
"content_index": 0,
"delta": "Oslo",
},
{"type": "response.output_item.done", "output_index": 2, "item": message},
{"type": "response.completed", "response": completed},
)
respx_mock.post(url__regex=_RETRY_URL).mock(
return_value=httpx.Response(200, headers={"content-type": "text/event-stream"}, content=_responses_sse(events))
)
stream: Final = await litellm.anthropic.messages.acreate(
model="azure/gpt-5.4-nano",
api_base=_RETRY_API_BASE,
api_key="fake-key",
api_version="2025-04-01-preview",
max_tokens=64,
tools=_RETRY_TOOLS,
stream=True,
messages=[{"role": "user", "content": "Which city is colder?"}],
)
frames: Final = b"".join([chunk async for chunk in stream]).decode()
starts: Final = tuple(
json.loads(line.removeprefix("data: "))["content_block"]
for line in frames.splitlines()
if line.startswith("data: ") and '"content_block_start"' in line
)
assert [(block["type"], block.get("data")) for block in starts] == [
("redacted_thinking", encrypted_reasoning_signature("gAAA_first")),
("redacted_thinking", encrypted_reasoning_signature("gAAA_second")),
("text", None),
]
def _streamed_blocks(frames: str) -> list[tuple[str, str, str | None]]:
events: Final = tuple(
json.loads(line.removeprefix("data: ")) for line in frames.splitlines() if line.startswith("data: ")
)
deltas: Final = tuple(event for event in events if event["type"] == "content_block_delta")
def block_deltas(index: int) -> tuple[Mapping[str, str], ...]:
return tuple(delta["delta"] for delta in deltas if delta["index"] == index)
return [
(
event["content_block"]["type"],
"".join(delta.get("thinking", "") for delta in block_deltas(event["index"])),
next(
(delta["signature"] for delta in block_deltas(event["index"]) if delta["type"] == "signature_delta"),
None,
),
)
for event in events
if event["type"] == "content_block_start"
]
@pytest.mark.asyncio
@pytest.mark.usefixtures("_chat_bridge_httpx")
async def test_streamed_reasoning_summaries_stay_with_their_own_encrypted_item(
respx_mock: respx.MockRouter,
) -> None:
response: Final = {**_RETRY_REPLY, "id": "resp_two_summaries", "status": "in_progress", "output": []}
message: Final = {
"type": "message",
"id": "msg_two_summaries",
"status": "completed",
"role": "assistant",
"content": [{"type": "output_text", "text": "Oslo", "annotations": []}],
}
first: Final = _reasoning_item("rs_1", "gAAA_first", "Oslo is further north.")
second: Final = _reasoning_item("rs_2", "gAAA_second", "Check the forecast.")
completed: Final = {**response, "status": "completed", "output": [first, second, message]}
def summary_delta(output_index: int, item_id: str, text: str) -> dict[str, object]:
return {
"type": "response.reasoning_summary_text.delta",
"output_index": output_index,
"item_id": item_id,
"summary_index": 0,
"delta": text,
}
events: Final = (
{"type": "response.created", "response": response},
{"type": "response.output_item.added", "output_index": 0, "item": _reasoning_item("rs_1", "")},
summary_delta(0, "rs_1", "Oslo is further north."),
{"type": "response.output_item.done", "output_index": 0, "item": first},
{"type": "response.output_item.added", "output_index": 1, "item": _reasoning_item("rs_2", "")},
summary_delta(1, "rs_2", "Check the forecast."),
{"type": "response.output_item.done", "output_index": 1, "item": second},
{"type": "response.output_item.added", "output_index": 2, "item": {**message, "content": []}},
{
"type": "response.output_text.delta",
"output_index": 2,
"item_id": "msg_two_summaries",
"content_index": 0,
"delta": "Oslo",
},
{"type": "response.output_item.done", "output_index": 2, "item": message},
{"type": "response.completed", "response": completed},
)
respx_mock.post(url__regex=_RETRY_URL).mock(
return_value=httpx.Response(200, headers={"content-type": "text/event-stream"}, content=_responses_sse(events))
)
stream: Final = await litellm.anthropic.messages.acreate(
model="azure/gpt-5.4-nano",
api_base=_RETRY_API_BASE,
api_key="fake-key",
api_version="2025-04-01-preview",
max_tokens=64,
tools=_RETRY_TOOLS,
stream=True,
messages=[{"role": "user", "content": "Which city is colder?"}],
)
frames: Final = b"".join([chunk async for chunk in stream]).decode()
assert _streamed_blocks(frames) == [
("thinking", "Oslo is further north.", encrypted_reasoning_signature("gAAA_first")),
("thinking", "Check the forecast.", encrypted_reasoning_signature("gAAA_second")),
("text", "", None),
]

379
tests/unit/llms/anthropic/pass_through/responses_adapters/test_responses_adapters_handler.py
View file

@ -1,20 +1,26 @@
import asyncio
import datetime
import json
import os
import sys
from collections.abc import Mapping
from typing import Final
from collections.abc import AsyncIterator, Mapping, Sequence
from contextlib import nullcontext
from typing import Final, cast
from unittest.mock import AsyncMock, patch
import httpx
import pytest
import respx
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../../../..")))
import litellm
from litellm.integrations.custom_logger import CustomLogger
from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
from litellm.litellm_core_utils.prompt_templates.common_utils import (
TOOL_RESULT_IMAGE_BOUNDARY,
TOOL_RESULT_IMAGE_PLACEHOLDER,
encrypted_reasoning_signature,
)
from litellm.llms.anthropic.pass_through.responses_adapters.handler import (
LiteLLMMessagesToResponsesAPIHandler,
@ -22,6 +28,7 @@ from litellm.llms.anthropic.pass_through.responses_adapters.handler import (
)
from litellm.router import Router
from litellm.types.integrations.custom_logger import CONVERTED_STREAM_KEYS, converted_stream_requested
from litellm.types.utils import StandardLoggingPayload
MESSAGES = [{"role": "user", "content": "hello"}]
CLAUDE_CODE_USER_ID = json.dumps({"device_id": "d" * 64, "account_uuid": "", "session_id": "session-abc"})
@ -684,3 +691,371 @@ def test_drop_params_drops_the_mcp_toolset_tools_of_dropped_mcp_servers(monkeypa
)
assert [tool["name"] for tool in responses_kwargs["tools"]] == ["get_weather"]
TEXT_RESPONSE: Final = {
"id": "resp_replayed",
"object": "response",
"created_at": 1,
"status": "completed",
"model": "gpt-5.6-luna",
"output": [
{
"type": "message",
"id": "msg_1",
"role": "assistant",
"status": "completed",
"content": [{"type": "output_text", "text": "done", "annotations": []}],
}
],
"parallel_tool_calls": True,
"tool_choice": "auto",
"tools": [],
"usage": {"input_tokens": 3, "output_tokens": 4, "total_tokens": 7},
}
REJECTED_REPLAYED_REASONING: Final = {
"error": {
"message": "The encrypted content gAAA_foreign could not be verified.",
"type": "invalid_request_error",
"param": None,
"code": "invalid_encrypted_content",
}
}
UNVERIFIED_ORGANIZATION: Final = {
"error": {"message": "The organization could not be verified.", "type": "invalid_request_error"}
}
def _history(*, replayed_reasoning: bool) -> list[dict[str, object]]:
reasoning: Final = (
({"type": "redacted_thinking", "data": encrypted_reasoning_signature("gAAA_foreign")},)
if replayed_reasoning
else ()
)
return [
{"role": "user", "content": "List the logs."},
{"role": "assistant", "content": [*reasoning, {"type": "tool_use", "id": "call_1", "name": "Bash", "input": {}}]},
{"role": "user", "content": [{"type": "tool_result", "tool_use_id": "call_1", "content": "app.log"}]},
]
def _replayed_reasoning_counts(upstream: respx.Route) -> list[int]:
return [
sum(item.get("type") == "reasoning" for item in json.loads(call.request.content)["input"])
for call in upstream.calls
]
def _upstream(respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, first: httpx.Response, stream: bool):
return _upstream_after(respx_mock, monkeypatch, (first,), stream)
def _upstream_after(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, failures: tuple[httpx.Response, ...], stream: bool
) -> respx.Route:
monkeypatch.setenv("OPENAI_API_KEY", "sk-replay-test")
monkeypatch.setenv("DISABLE_AIOHTTP_TRANSPORT", "True")
litellm.in_memory_llm_clients_cache.flush_cache()
served: Final = (
httpx.Response(200, headers={"Content-Type": "text/event-stream"}, content=RESPONSES_SSE_BODY)
if stream
else httpx.Response(200, json=TEXT_RESPONSE)
)
return respx_mock.post("https://api.openai.com/v1/responses").mock(side_effect=[*failures, served])
@pytest.mark.asyncio
@pytest.mark.parametrize("stream", [False, True])
async def test_openai_turn_is_sent_again_once_without_replayed_reasoning_the_backend_cannot_decrypt(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, stream: bool
) -> None:
upstream: Final = _upstream(respx_mock, monkeypatch, httpx.Response(400, json=REJECTED_REPLAYED_REASONING), stream)
response: Final = await LiteLLMMessagesToResponsesAPIHandler.async_anthropic_messages_handler(
max_tokens=64,
messages=_history(replayed_reasoning=True),
model="openai/gpt-5.6-luna",
tools=[dict(BASH_TOOL)],
stream=stream,
custom_llm_provider="openai",
)
events: Final = (
[json.loads(chunk.decode().split("data: ", 1)[1]) async for chunk in cast(AsyncIterator[bytes], response)]
if stream
else [cast(Mapping[str, object], response)]
)
assert _replayed_reasoning_counts(upstream) == [1, 0]
assert events[-1]["type"] in ("message", "message_stop")
def test_sync_openai_turn_is_sent_again_once_without_replayed_reasoning_the_backend_cannot_decrypt(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
upstream: Final = _upstream(respx_mock, monkeypatch, httpx.Response(400, json=REJECTED_REPLAYED_REASONING), False)
response: Final = LiteLLMMessagesToResponsesAPIHandler.anthropic_messages_handler(
max_tokens=64,
messages=_history(replayed_reasoning=True),
model="openai/gpt-5.6-luna",
tools=[dict(BASH_TOOL)],
custom_llm_provider="openai",
)
assert _replayed_reasoning_counts(upstream) == [1, 0]
assert cast(Mapping[str, object], response)["content"] == [{"type": "text", "text": "done"}]
@pytest.mark.asyncio
async def test_openai_turn_with_num_retries_sends_the_rejected_reasoning_only_once(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
upstream: Final = _upstream(respx_mock, monkeypatch, httpx.Response(400, json=REJECTED_REPLAYED_REASONING), False)
await litellm.anthropic.messages.acreate(
max_tokens=64,
messages=_history(replayed_reasoning=True),
model="openai/gpt-5.6-luna",
tools=[dict(BASH_TOOL)],
num_retries=2,
)
assert _replayed_reasoning_counts(upstream) == [1, 0]
def test_sync_openai_turn_with_global_num_retries_sends_the_rejected_reasoning_once_and_keeps_the_setting(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(litellm, "num_retries", 2)
upstream: Final = _upstream(respx_mock, monkeypatch, httpx.Response(400, json=REJECTED_REPLAYED_REASONING), False)
litellm.anthropic.messages.create(
max_tokens=64,
messages=_history(replayed_reasoning=True),
model="openai/gpt-5.6-luna",
tools=[dict(BASH_TOOL)],
)
assert _replayed_reasoning_counts(upstream) == [1, 0]
assert litellm.num_retries == 2
_SERVER_ERROR: Final = {
"error": {"message": "The server had an error while processing your request.", "type": "server_error"}
}
@pytest.mark.asyncio
async def test_openai_turn_retries_a_server_error_with_the_reasoning_and_drops_it_only_once_rejected(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
upstream: Final = _upstream_after(
respx_mock,
monkeypatch,
(httpx.Response(500, json=_SERVER_ERROR), httpx.Response(400, json=REJECTED_REPLAYED_REASONING)),
False,
)
response: Final = await litellm.anthropic.messages.acreate(
max_tokens=64,
messages=_history(replayed_reasoning=True),
model="openai/gpt-5.6-luna",
tools=[dict(BASH_TOOL)],
num_retries=2,
)
assert _replayed_reasoning_counts(upstream) == [1, 1, 0]
assert cast(Mapping[str, object], response)["content"] == [{"type": "text", "text": "done"}]
def test_sync_openai_turn_retries_a_server_error_with_the_reasoning_and_drops_it_only_once_rejected(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
upstream: Final = _upstream_after(
respx_mock,
monkeypatch,
(httpx.Response(500, json=_SERVER_ERROR), httpx.Response(400, json=REJECTED_REPLAYED_REASONING)),
False,
)
response: Final = litellm.anthropic.messages.create(
max_tokens=64,
messages=_history(replayed_reasoning=True),
model="openai/gpt-5.6-luna",
tools=[dict(BASH_TOOL)],
num_retries=2,
)
assert _replayed_reasoning_counts(upstream) == [1, 1, 0]
assert cast(Mapping[str, object], response)["content"] == [{"type": "text", "text": "done"}]
class _FailureLog(CustomLogger):
def __init__(self) -> None:
super().__init__()
self.rejections: tuple[str, ...] = ()
def _record(self, kwargs: Mapping[str, object]) -> None:
error: Final = str(kwargs.get("exception"))
self.rejections = (
*self.rejections,
next((phrase for phrase in ("encrypted content", "organization") if phrase in error), error),
)
def log_failure_event(
self, kwargs: Mapping[str, object], response_obj: object, start_time: object, end_time: object
) -> None:
self._record(kwargs)
async def async_log_failure_event(
self, kwargs: Mapping[str, object], response_obj: object, start_time: object, end_time: object
) -> None:
self._record(kwargs)
_OUTCOMES_OF_THE_TURN_SENT_AGAIN: Final = (
pytest.param(httpx.Response(200, json=TEXT_RESPONSE), (), id="served"),
pytest.param(httpx.Response(400, json=UNVERIFIED_ORGANIZATION), ("organization",), id="rejected"),
)
def _rejected_then(respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch, sent_again: httpx.Response) -> None:
monkeypatch.setenv("OPENAI_API_KEY", "sk-replay-test")
monkeypatch.setenv("DISABLE_AIOHTTP_TRANSPORT", "True")
litellm.in_memory_llm_clients_cache.flush_cache()
respx_mock.post("https://api.openai.com/v1/responses").mock(
side_effect=[httpx.Response(400, json=REJECTED_REPLAYED_REASONING), sent_again]
)
@pytest.mark.asyncio
@pytest.mark.parametrize(("sent_again", "logged_rejections"), _OUTCOMES_OF_THE_TURN_SENT_AGAIN)
async def test_openai_turn_logs_only_the_outcome_of_the_turn_sent_again(
respx_mock: respx.MockRouter,
monkeypatch: pytest.MonkeyPatch,
sent_again: httpx.Response,
logged_rejections: tuple[str, ...],
) -> None:
failure_log: Final = _FailureLog()
monkeypatch.setattr(litellm, "callbacks", [failure_log])
_rejected_then(respx_mock, monkeypatch, sent_again)
with pytest.raises(litellm.BadRequestError) if logged_rejections else nullcontext():
await litellm.anthropic.messages.acreate(
max_tokens=64,
messages=_history(replayed_reasoning=True),
model="openai/gpt-5.6-luna",
tools=[dict(BASH_TOOL)],
)
assert failure_log.rejections == logged_rejections
@pytest.mark.parametrize(("sent_again", "logged_rejections"), _OUTCOMES_OF_THE_TURN_SENT_AGAIN)
def test_sync_openai_turn_logs_only_the_outcome_of_the_turn_sent_again(
respx_mock: respx.MockRouter,
monkeypatch: pytest.MonkeyPatch,
sent_again: httpx.Response,
logged_rejections: tuple[str, ...],
) -> None:
failure_log: Final = _FailureLog()
monkeypatch.setattr(litellm, "callbacks", [failure_log])
_rejected_then(respx_mock, monkeypatch, sent_again)
with pytest.raises(litellm.BadRequestError) if logged_rejections else nullcontext():
litellm.anthropic.messages.create(
max_tokens=64,
messages=_history(replayed_reasoning=True),
model="openai/gpt-5.6-luna",
tools=[dict(BASH_TOOL)],
)
assert failure_log.rejections == logged_rejections
@pytest.mark.asyncio
@pytest.mark.parametrize(
"replayed_reasoning, rejection",
[
pytest.param(False, REJECTED_REPLAYED_REASONING, id="no_replayed_reasoning"),
pytest.param(True, UNVERIFIED_ORGANIZATION, id="unrelated_rejection"),
],
)
async def test_openai_turn_is_not_sent_again_for_other_rejections(
respx_mock: respx.MockRouter,
monkeypatch: pytest.MonkeyPatch,
replayed_reasoning: bool,
rejection: Mapping[str, object],
) -> None:
upstream: Final = _upstream(respx_mock, monkeypatch, httpx.Response(400, json=rejection), False)
with pytest.raises(litellm.BadRequestError):
await LiteLLMMessagesToResponsesAPIHandler.async_anthropic_messages_handler(
max_tokens=64,
messages=_history(replayed_reasoning=replayed_reasoning),
model="openai/gpt-5.6-luna",
tools=[dict(BASH_TOOL)],
custom_llm_provider="openai",
)
assert upstream.call_count == 1
class _SuccessLog(CustomLogger):
def __init__(self) -> None:
super().__init__()
self.logged_reasoning_item_ids: tuple[tuple[object, ...], ...] = ()
self.logged: Final = asyncio.Event()
async def async_log_success_event(
self, kwargs: Mapping[str, object], response_obj: object, start_time: object, end_time: object
) -> None:
payload: Final = cast(StandardLoggingPayload, kwargs["standard_logging_object"])
logged: Final = cast(Mapping[str, Sequence[Mapping[str, Mapping[str, object]]]], payload["response"])
message: Final = logged["choices"][0]["message"]
items: Final = cast(Sequence[Mapping[str, object]], message.get("reasoning_items") or ())
self.logged_reasoning_item_ids = (*self.logged_reasoning_item_ids, tuple(item["id"] for item in items))
self.logged.set()
def _reasoning_before_call(index: int) -> tuple[dict[str, object], dict[str, object]]:
return (
{"type": "reasoning", "id": f"rs_{index}", "summary": [], "encrypted_content": f"gAAA_rs_{index}"},
{
"type": "function_call",
"id": f"fc_{index}",
"status": "completed",
"arguments": "{}",
"call_id": f"call_{index}",
"name": "Bash",
},
)
@pytest.mark.asyncio
async def test_openai_turn_logs_every_reasoning_item_the_messages_client_was_served(
respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch
) -> None:
success_log: Final = _SuccessLog()
monkeypatch.setattr(litellm, "callbacks", [success_log])
monkeypatch.setenv("OPENAI_API_KEY", "sk-replay-test")
monkeypatch.setenv("DISABLE_AIOHTTP_TRANSPORT", "True")
litellm.in_memory_llm_clients_cache.flush_cache()
respx_mock.post("https://api.openai.com/v1/responses").mock(
return_value=httpx.Response(
200,
json={**TEXT_RESPONSE, "output": [*_reasoning_before_call(1), *_reasoning_before_call(2)]},
)
)
reply: Final = await litellm.anthropic.messages.acreate(
max_tokens=64,
messages=[{"role": "user", "content": "List the logs twice."}],
model="openai/gpt-5.6-luna",
tools=[dict(BASH_TOOL)],
)
GLOBAL_LOGGING_WORKER.start()
await asyncio.wait_for(success_log.logged.wait(), timeout=30.0)
assert [block["type"] for block in reply["content"]].count("redacted_thinking") == 2
assert success_log.logged_reasoning_item_ids == (("rs_1", "rs_2"),)

15
tests/unit/test_main.py
View file

@ -8,7 +8,7 @@ import logging
import os
import threading
import urllib.parse
from collections.abc import Mapping
from collections.abc import Awaitable, Callable, Mapping
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from datetime import datetime
@ -7936,6 +7936,19 @@ async def test_responses_with_retries(sync_mode):
assert mock_responses.call_args.kwargs["max_retries"] == 0
@pytest.mark.parametrize("retry_function", [acompletion_with_retries, aresponses_with_retries])
@pytest.mark.asyncio
async def test_async_retry_helpers_stop_when_the_call_is_cancelled(
retry_function: Callable[..., Awaitable[object]],
) -> None:
upstream: Final = AsyncMock(side_effect=[asyncio.CancelledError(), MagicMock()])
with pytest.raises(asyncio.CancelledError):
await retry_function(model="gpt-4o", num_retries=3, original_function=upstream)
assert upstream.await_count == 1
def test_azure_embedding_exceptions():
with pytest.raises(Exception, match="Mock error") as exc_info:
litellm.embedding(