fix(anthropic): map pause_turn and return refusal explanations on chat and responses (#49192)

* fix(anthropic): map pause_turn and return refusal explanations on chat and responses

pause_turn now maps to finish_reason length (Python and Rust) and keeps the raw
stop_reason in provider_specific_fields. Responses reports a paused turn as
incomplete with max_output_tokens and shows the unrun server search as an
in_progress web_search_call instead of a client function_call.

A refusal's stop_details.explanation now fills message.refusal and the final
stream delta.refusal on chat, and becomes a refusal content part on Responses,
streamed or not.

* fix(anthropic): let a chat client resend a paused turn so Claude continues it

* fix(anthropic): replay a paused chat turn without its thinking blocks, and drop its unrun search before a new message

Chat cannot say where Claude wrote each thinking block in a paused web search turn, and
Anthropic answers 400 when one moves but accepts the turn without them. A new message after
a paused turn also 400s on a server_tool_use with no result, so that unrun search is dropped.

* fix(responses): send the streamed refusal part's type and text in content_part.added

* fix(anthropic): keep pause_turn and Claude refusal explanations out of the Messages adapter's text

A paused Claude turn now ends on stop_reason pause_turn instead of max_tokens, streamed or not. A Claude refusal keeps its explanation in stop_details with Claude's category and no longer repeats it as a text block, so a guardrail mask on a streamed refusal is delivered. A paused turn sent back with list content drops its thinking blocks like string content does

* fix(anthropic): replay a paused turn with one unrun search and no thinking on chat and Responses

* fix(responses): report a paused Claude turn as completed so Codex and the Agents SDK keep going

* fix(responses): mark a paused turn's message item completed

* fix(responses): mark every item of a paused turn completed

* fix(responses): keep a refusal's explanation when a refused turn is sent back

A refusal part in a Responses input message now goes to the model as text, so the assistant turn is no longer dropped from the history. The integration test now expects a paused Claude turn on Responses to report completed, and the paused-turn test helper is typed

* fix(responses): scan refusal parts in Responses guardrails

---------

Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
This commit is contained in:
paveln83 2026-10-31 02:53:59 -07:00 • committed by GitHub
parent 281bf4e48b
commit 27fbcd9895
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
33 changed files with 1885 additions and 301 deletions

View file

@ -13,6 +13,7 @@ const FINISH_REASONS: &[(&str, &str)] = &[
("max_tokens", "length"),
("refusal", "content_filter"),
("compaction", "length"),
("pause_turn", "length"),
("model_context_window_exceeded", "length"),
("guardrail_intervened", "content_filter"),
("content_filtered", "content_filter"),
@ -75,6 +76,7 @@ mod tests {
#[case::end_turn("end_turn", "stop")]
#[case::stop_sequence("stop_sequence", "stop")]
#[case::max_tokens("max_tokens", "length")]
#[case::pause_turn("pause_turn", "length")]
#[case::model_context_window_exceeded("model_context_window_exceeded", "length")]
#[case::refusal("refusal", "content_filter")]
#[case::guardrail_intervened("guardrail_intervened", "content_filter")]

View file

@ -311,6 +311,7 @@ _FINISH_REASON_MAP: Final[dict[str, OpenAIChatCompletionFinishReason]] = {
"tool_use": "tool_calls",
"refusal": "content_filter",
"compaction": "length",
"pause_turn": "length",
"model_context_window_exceeded": "length",
# Cohere
"COMPLETE": "stop",

View file

@ -26,6 +26,7 @@ from litellm.litellm_core_utils.core_helpers import as_str_mapping, attribute_or
from litellm.litellm_core_utils.prompt_templates.mid_conversation_system import (
anthropic_system_messages,
find_anthropic_server_tool_result,
is_anthropic_server_tool_use,
is_anthropic_tool_search_use,
message_field,
parts_of,
@ -2059,6 +2060,54 @@ def _placed_and_paired(
_AnthropicToolInvokeItem: TypeAlias = AnthropicMessagesToolUseParam | dict[str, Any]
def _is_server_tool_result(block: Mapping[str, object]) -> bool:
block_type: Final = block.get("type")
return isinstance(block_type, str) and block_type.endswith("_tool_result")
def _split_unrun_server_tool_uses(
groups: Sequence[Sequence[_AnthropicToolInvokeItem]],
) -> tuple[tuple[Sequence[_AnthropicToolInvokeItem], ...], tuple[_AnthropicToolInvokeItem, ...]]:
completed: Final = tuple(group for group in groups if any(_is_server_tool_result(block) for block in group))
unrun: Final = tuple(group[0] for group in groups if not any(_is_server_tool_result(block) for block in group))
return completed, unrun
def _block_id(block: Mapping[str, object]) -> object:
return block.get("id")
def _unrun_server_tool_use_ids(blocks: Sequence[Mapping[str, object]]) -> frozenset[object]:
result_ids: Final = frozenset(block.get("tool_use_id") for block in blocks if _is_server_tool_result(block))
return frozenset(
_block_id(block)
for block in blocks
if block.get("type") == "server_tool_use" and _block_id(block) not in result_ids
)
def _last_block_id_among(blocks: Sequence[Mapping[str, object]], ids: frozenset[object]) -> object:
return next(block_id for block in reversed(blocks) if (block_id := _block_id(block)) in ids)
def _paused_turn_replay(
tool_invokes: Sequence[_AnthropicToolInvokeItem],
thinking_blocks: Sequence[ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock],
*,
is_final_message: bool,
) -> tuple[
tuple[_AnthropicToolInvokeItem, ...],
tuple[ChatCompletionThinkingBlock | ChatCompletionRedactedThinkingBlock, ...],
]:
unrun_ids: Final = _unrun_server_tool_use_ids(tool_invokes)
if not unrun_ids:
return tuple(tool_invokes), tuple(thinking_blocks)
dropped_ids: Final = (
unrun_ids - frozenset((_last_block_id_among(tool_invokes, unrun_ids),)) if is_final_message else unrun_ids
)
return tuple(block for block in tool_invokes if _block_id(block) not in dropped_ids), ()
def convert_to_anthropic_tool_invoke(
tool_calls: list[ChatCompletionAssistantToolCall],
web_search_results: Sequence[object] | None = None,
@ -2141,6 +2190,15 @@ def convert_to_anthropic_tool_invoke(
anthropic_tool_invoke.append(server_tool_result)
elif is_anthropic_tool_search_use(tool_id, tool_name) and tool_id not in answered_tool_ids:
continue
elif is_anthropic_server_tool_use(tool_id) and tool_id not in answered_tool_ids:
anthropic_tool_invoke.append(
{
"type": "server_tool_use",
"id": tool_id,
"name": tool_name,
"input": tool_input,
}
)
else:
sanitized_tool_id = sanitize_anthropic_tool_use_id(tool_id)
_anthropic_tool_use_param = AnthropicMessagesToolUseParam(
@ -3109,6 +3167,30 @@ def anthropic_messages_pt(
_has_server_tool_calls = True
break
# Get web_search_results and tool_results from provider_specific_fields
# for server_tool_use reconstruction.
# Fixes: https://github.com/BerriAI/litellm/issues/17737
_provider_specific_fields_raw = assistant_content_block.get("provider_specific_fields")
_provider_specific_fields: dict[str, Any] = (
cast(dict[str, Any], _provider_specific_fields_raw)
if isinstance(_provider_specific_fields_raw, dict)
else {}
)
tool_invoke_results = (
convert_to_anthropic_tool_invoke(
assistant_tool_calls,
web_search_results=_provider_specific_fields.get("web_search_results"),
tool_results=_provider_specific_fields.get("tool_results"),
answered_tool_ids=client_answered_tool_ids,
)
if assistant_tool_calls is not None
else ()
)
turn_paused = bool(_unrun_server_tool_use_ids(tool_invoke_results))
replayed_tool_invokes, replayed_thinking_blocks = _paused_turn_replay(
tool_invoke_results, thinking_blocks or (), is_final_message=msg_i == len(messages) - 1
)
if (
thinking_blocks is not None
and _has_server_tool_calls
@ -3121,26 +3203,12 @@ def anthropic_messages_pt(
# We must preserve this interleaved order because Anthropic
# verifies thinking block signatures based on position.
# Build the tool call groups (server_tool_use + its result)
_provider_specific_fields_raw_tc = assistant_content_block.get("provider_specific_fields")
_provider_specific_fields_tc: dict[str, Any] = {}
if isinstance(_provider_specific_fields_raw_tc, dict):
_provider_specific_fields_tc = cast(dict[str, Any], _provider_specific_fields_raw_tc)
_web_search_results_tc = _provider_specific_fields_tc.get("web_search_results")
_tool_results_tc = _provider_specific_fields_tc.get("tool_results")
tool_invoke_results = convert_to_anthropic_tool_invoke(
assistant_tool_calls,
web_search_results=_web_search_results_tc,
tool_results=_tool_results_tc,
answered_tool_ids=client_answered_tool_ids,
)
# Group tool invoke results into (server_tool_use, result) pairs
# and separate regular tool_use blocks
server_tool_groups: list[list[_AnthropicToolInvokeItem]] = []
regular_tool_uses: list[_AnthropicToolInvokeItem] = []
_current_group: list[_AnthropicToolInvokeItem] = []
for item in tool_invoke_results:
for item in replayed_tool_invokes:
item_type = item.get("type", "") if isinstance(item, dict) else getattr(item, "type", "")
if item_type == "server_tool_use":
if _current_group:
@ -3154,6 +3222,9 @@ def anthropic_messages_pt(
_current_group.append(item)
if _current_group:
server_tool_groups.append(_current_group)
completed_server_tool_groups, pending_server_tool_uses = _split_unrun_server_tool_uses(
server_tool_groups
)
# Build the text block if content is a non-empty string
text_element = None
@ -3177,15 +3248,15 @@ def anthropic_messages_pt(
# Any remaining groups (after all thinking blocks) go after.
tb_idx = 0
grp_idx = 0
num_tb = len(thinking_blocks) if thinking_blocks else 0
num_grp = len(server_tool_groups)
num_tb = len(replayed_thinking_blocks)
num_grp = len(completed_server_tool_groups)
while tb_idx < num_tb or grp_idx < num_grp:
if tb_idx < num_tb and grp_idx < num_grp:
# Emit thinking block then its tool group
assistant_content.append(thinking_blocks[tb_idx])
assistant_content.append(replayed_thinking_blocks[tb_idx])
tb_idx += 1
for block in server_tool_groups[grp_idx]:
for block in completed_server_tool_groups[grp_idx]:
item_id = block.get("id") if isinstance(block, dict) else getattr(block, "id", None)
if item_id and item_id in unique_tool_ids:
continue
@ -3195,11 +3266,11 @@ def anthropic_messages_pt(
grp_idx += 1
elif tb_idx < num_tb:
# More thinking blocks than tool groups - emit before text
assistant_content.append(thinking_blocks[tb_idx])
assistant_content.append(replayed_thinking_blocks[tb_idx])
tb_idx += 1
else:
# More tool groups than thinking blocks - emit remaining
for block in server_tool_groups[grp_idx]:
for block in completed_server_tool_groups[grp_idx]:
item_id = block.get("id") if isinstance(block, dict) else getattr(block, "id", None)
if item_id and item_id in unique_tool_ids:
continue
@ -3212,8 +3283,8 @@ def anthropic_messages_pt(
if text_element is not None:
assistant_content.append(text_element)
# Add regular (non-server) tool calls at the end
for item in regular_tool_uses:
# Add regular tool calls, then a paused turn's unrun server tool call, at the end
for item in (*regular_tool_uses, *pending_server_tool_uses):
item_id = item.get("id") if isinstance(item, dict) else getattr(item, "id", None)
if item_id and item_id in unique_tool_ids:
continue
@ -3233,18 +3304,18 @@ def anthropic_messages_pt(
)
_content_list = assistant_content_block.get("content") if _content_is_list else None
if (
thinking_blocks is not None and not _content_is_list
): # IMPORTANT: ADD THIS FIRST, ELSE ANTHROPIC WILL RAISE AN ERROR
assistant_content.extend(thinking_blocks)
if not _content_is_list: # IMPORTANT: ADD THIS FIRST, ELSE ANTHROPIC WILL RAISE AN ERROR
assistant_content.extend(replayed_thinking_blocks)
if _content_is_list and _content_list is not None:
for m in _with_missing_thinking_blocks(_content_list, thinking_blocks or ()):
for m in _with_missing_thinking_blocks(_content_list, replayed_thinking_blocks):
if not isinstance(m, dict):
continue
# handle thinking blocks
if (thinking_block := _as_thinking_block(m)) is not None:
assistant_content.extend(
_replayable_thinking_blocks([thinking_block], drop_params=drop_params)
()
if turn_paused
else _replayable_thinking_blocks([thinking_block], drop_params=drop_params)
)
continue
text_block = cast(str, m.get("text", ""))
@ -3284,25 +3355,9 @@ def anthropic_messages_pt(
assistant_content.append(_anthropic_text_content_element)
if assistant_tool_calls is not None: # support assistant tool invoke conversion
# Get web_search_results and tool_results from provider_specific_fields
# for server_tool_use reconstruction.
# Fixes: https://github.com/BerriAI/litellm/issues/17737
_provider_specific_fields_raw = assistant_content_block.get("provider_specific_fields")
_provider_specific_fields: dict[str, Any] = {}
if isinstance(_provider_specific_fields_raw, dict):
_provider_specific_fields = cast(dict[str, Any], _provider_specific_fields_raw)
_web_search_results = _provider_specific_fields.get("web_search_results")
_tool_results = _provider_specific_fields.get("tool_results")
tool_invoke_results = convert_to_anthropic_tool_invoke(
assistant_tool_calls,
web_search_results=_web_search_results,
tool_results=_tool_results,
answered_tool_ids=client_answered_tool_ids,
)
# Prevent "tool_use ids must be unique" errors by filtering duplicates
# This can happen when merging history that already contains the tool calls
for item in tool_invoke_results:
for item in replayed_tool_invokes:
# tool_use items are typically dicts, but handle objects just in case
item_id = item.get("id") if isinstance(item, dict) else getattr(item, "id", None)

View file

@ -286,6 +286,10 @@ def _separate_thinking_blocks_render(message: object) -> bool:
return any(_thinking_block_renders(block) for block in parts_of(message_field(message, "thinking_blocks")))
def is_anthropic_server_tool_use(tool_id: str) -> bool:
return tool_id.startswith(ANTHROPIC_SERVER_TOOL_USE_ID_PREFIX)
def is_anthropic_tool_search_use(tool_id: str, tool_name: str) -> bool:
return tool_id.startswith(ANTHROPIC_SERVER_TOOL_USE_ID_PREFIX) and tool_name in ANTHROPIC_TOOL_SEARCH_TOOL_NAMES

View file

@ -1509,11 +1509,20 @@ def _delta_reasoning_items(delta: object) -> Sequence[object]:
return cast(Sequence[object], items) if isinstance(items, list) else () # cast-ok: items forwarded verbatim
def _chunk_reasoning_items(chunk: object) -> Sequence[object]:
def _chunk_first_delta(chunk: object) -> object:
choices: Final = _field_of(chunk, "choices")
if not isinstance(choices, list) or not choices:
return ()
return _delta_reasoning_items(_field_of(cast(Sequence[object], choices)[0], "delta")) # cast-ok: untyped chunk
return None
return _field_of(cast(Sequence[object], choices)[0], "delta") # cast-ok: untyped chunk
def _chunk_reasoning_items(chunk: object) -> Sequence[object]:
return _delta_reasoning_items(_chunk_first_delta(chunk))
def _chunk_refusal(chunk: object) -> str:
refusal: Final = _field_of(_chunk_first_delta(chunk), "refusal")
return refusal if isinstance(refusal, str) else ""
_NON_TEXT_DELTA_FIELDS: Final = (
@ -1526,6 +1535,7 @@ _NON_TEXT_DELTA_FIELDS: Final = (
"audio",
"images",
"provider_specific_fields",
"refusal",
)
@ -1674,6 +1684,9 @@ def _assembled_message_fields(processor: ChunkProcessor, chunks: Sequence) -> It
yield "reasoning_content", processor.get_combined_reasoning_content(reasoning_chunks)
typed_chunks: Final = cast(Sequence[object], chunks) # cast-ok: chunks are untyped upstream
combined_refusal: Final = "".join(_chunk_refusal(chunk) for chunk in typed_chunks)
if combined_refusal:
yield "refusal", combined_refusal
reasoning_items: Final = tuple(chain.from_iterable(_chunk_reasoning_items(chunk) for chunk in typed_chunks))
if reasoning_items:
yield "reasoning_items", list(reasoning_items)

View file

@ -70,7 +70,12 @@ from litellm.types.utils import (
)
from ...base import BaseLLM
from ..common_utils import AnthropicError, process_anthropic_headers, refusal_provider_specific_fields
from ..common_utils import (
AnthropicError,
process_anthropic_headers,
refusal_params,
stop_reason_provider_specific_fields,
)
from .transformation import ANTHROPIC_TOOL_NAME_REVERSE_MAP_KEY, AnthropicConfig, JsonModeRequest
if TYPE_CHECKING:
@ -926,6 +931,17 @@ class ModelResponseIterator:
def _web_search_call_snapshot(self) -> dict[str, object]:
return dict(self._web_search_calls)
def _refresh_pending_web_search_call(self, tool_use_id: str) -> bool:
if tool_use_id not in self._web_search_calls:
return False
self._web_search_calls[tool_use_id] = build_web_search_call(
tool_use_id,
self._server_tool_inputs.get(tool_use_id, {}),
{"content": []},
status="in_progress",
)
return True
def _complete_web_search_call(self, result: dict[str, object]) -> None:
tool_use_id: Final = result.get("tool_use_id")
if not isinstance(tool_use_id, str) or tool_use_id not in self._web_search_calls:
@ -1128,6 +1144,8 @@ class ModelResponseIterator:
self._server_tool_inputs[self._current_server_tool_id] = json.loads(args)
except (json.JSONDecodeError, TypeError):
pass
if self._refresh_pending_web_search_call(self._current_server_tool_id):
provider_specific_fields["web_search_calls"] = self._web_search_call_snapshot()
self._current_server_tool_id = None
if self.is_response_format_tool:
text = self._flush_response_format_buffer()
@ -1140,8 +1158,8 @@ class ModelResponseIterator:
# These are automatically handled by Anthropic API, we just pass them through
pass
elif type_chunk == "message_delta":
finish_reason, usage, container, refusal_fields = self._handle_message_delta(chunk)
provider_specific_fields.update(refusal_fields)
finish_reason, usage, container, stop_reason_fields = self._handle_message_delta(chunk)
provider_specific_fields.update(stop_reason_fields)
text = self._release_answer_close() + self._drain_split_answer_blocks()
if container:
provider_specific_fields["container"] = container
@ -1190,6 +1208,7 @@ class ModelResponseIterator:
)
text, tool_use = self._handle_json_mode_chunk(text=text, tool_use=tool_use)
refusal_fields: Final = refusal_params(provider_specific_fields)
returned_chunk: Final = ModelResponseStream(
choices=[
@ -1201,6 +1220,7 @@ class ModelResponseIterator:
provider_specific_fields=(provider_specific_fields if provider_specific_fields else None),
thinking_blocks=(thinking_blocks if thinking_blocks else None),
reasoning_content=reasoning_content,
**refusal_fields,
),
finish_reason=finish_reason,
)
@ -1371,7 +1391,7 @@ class ModelResponseIterator:
self._handle_usage(anthropic_usage_chunk=message_delta["usage"]) if "usage" in message_delta else None
)
container: Final = message_delta["delta"].get("container")
return finish_reason, usage, container, refusal_provider_specific_fields(message_delta["delta"])
return finish_reason, usage, container, stop_reason_provider_specific_fields(message_delta["delta"])
def _handle_accumulated_json_chunk(self, data_str: str, is_final: bool = False) -> ModelResponseStream | None:
"""

View file

@ -120,13 +120,14 @@ from litellm.utils import (
)
from ..common_utils import (
REFUSAL_STOP_FIELDS,
STOP_REASON_FIELDS,
AnthropicError,
AnthropicModelInfo,
eager_input_streaming_flag,
process_anthropic_headers,
refusal_provider_specific_fields,
refusal_params,
requires_native_compaction_beta,
stop_reason_provider_specific_fields,
strip_advisor_blocks_from_messages,
tools_marked_for_a_tool_config_point,
)
@ -167,6 +168,8 @@ _DEFER_LOADING_ADAPTER: Final = TypeAdapter(bool)
_TOOL_FIELDS_ADAPTER: Final = TypeAdapter(Mapping[str, object])
_TOOLS_ADAPTER: Final = TypeAdapter(list[object])
_LITELLM_PARAMS_ADAPTER: Final = TypeAdapter(Mapping[str, object])
_RESPONSE_BLOCKS: Final = TypeAdapter(tuple[Mapping[str, object], ...])
_PENDING_WEB_SEARCH_RESULT: Final = MappingProxyType({"content": ()})
_REQUEST_PARAMS_ADAPTER: Final = TypeAdapter(Mapping[str, object])
_NATIVE_OUTPUT_STRING_FORMATS: Final = frozenset(
("date-time", "time", "date", "duration", "email", "hostname", "uri", "ipv4", "ipv6", "uuid")
@ -2919,28 +2922,24 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
web_search_results: Sequence[object],
completion_response: Mapping[str, object],
) -> list[object]:
content: Final = completion_response.get("content")
blocks: Final = content if isinstance(content, Sequence) else ()
inputs: Final = {
call_id: tool_input
for block in blocks
if isinstance(block, Mapping)
and block.get("type") == "server_tool_use"
and block.get("name") == "web_search"
and isinstance((call_id := block.get("id")), str)
and isinstance((tool_input := block.get("input")), Mapping)
blocks: Final = _RESPONSE_BLOCKS.validate_python(completion_response.get("content") or ())
results_by_id: Final = {
tool_use_id: result
for result in _RESPONSE_BLOCKS.validate_python(web_search_results)
if result.get("type") == "web_search_tool_result"
and isinstance((tool_use_id := result.get("tool_use_id")), str)
}
return [
build_web_search_call(
tool_id=tool_use_id,
tool_input=inputs.get(tool_use_id, {}),
result=result,
tool_id=call_id,
tool_input=block.get("input"),
result=results_by_id.get(call_id, _PENDING_WEB_SEARCH_RESULT),
status=None if call_id in results_by_id else "in_progress",
)
for result in web_search_results
if isinstance(result, dict)
and result.get("type") == "web_search_tool_result"
and isinstance((tool_use_id := result.get("tool_use_id")), str)
and tool_use_id in inputs
for block in blocks
if block.get("type") == "server_tool_use"
and block.get("name") == "web_search"
and isinstance((call_id := block.get("id")), str)
]
def _build_provider_specific_fields(
@ -2964,10 +2963,10 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
if web_search_results is not None:
provider_specific_fields["web_search_results"] = web_search_results
provider_specific_fields["web_search_calls"] = self._build_web_search_calls(
web_search_results,
completion_response,
)
web_search_calls: Final = self._build_web_search_calls(web_search_results or (), completion_response)
if web_search_results is not None or web_search_calls:
provider_specific_fields["web_search_calls"] = web_search_calls
if tool_results is not None:
provider_specific_fields["tool_results"] = tool_results
@ -2989,10 +2988,10 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
if compaction_blocks is not None:
provider_specific_fields["compaction_blocks"] = compaction_blocks
refusal_fields: Final = refusal_provider_specific_fields(
REFUSAL_STOP_FIELDS.validate_python(completion_response)
stop_reason_fields: Final = stop_reason_provider_specific_fields(
STOP_REASON_FIELDS.validate_python(completion_response)
)
return {**provider_specific_fields, **refusal_fields}
return {**provider_specific_fields, **stop_reason_fields}
def transform_parsed_response(
self,
@ -3067,22 +3066,26 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
if json_extra_content:
merged_text = merged_text + json_extra_content if merged_text else json_extra_content
refusal_fields: Final = refusal_params(STOP_REASON_FIELDS.validate_python(completion_response))
_message = litellm.Message(
tool_calls=tool_calls_for_message,
content=merged_text or None,
provider_specific_fields=provider_specific_fields,
thinking_blocks=thinking_blocks,
reasoning_content=reasoning_content,
**refusal_fields,
)
_message.provider_specific_fields = provider_specific_fields
if json_mode_message is not None:
refusal_fields: Final = refusal_provider_specific_fields(
REFUSAL_STOP_FIELDS.validate_python(completion_response)
stop_reason_fields: Final = stop_reason_provider_specific_fields(
STOP_REASON_FIELDS.validate_python(completion_response)
)
completion_response["stop_reason"] = "stop"
if refusal_fields:
json_mode_message.provider_specific_fields = dict(refusal_fields)
if stop_reason_fields:
json_mode_message.provider_specific_fields = dict(stop_reason_fields)
if "refusal" in refusal_fields:
json_mode_message.refusal = refusal_fields["refusal"]
if thinking_blocks:
json_mode_message.thinking_blocks = thinking_blocks
json_mode_message.reasoning_content = reasoning_content

View file

@ -2518,7 +2518,7 @@ def create_anthropic_model_list_response(
}
class RefusalStopFields(TypedDict, total=False):
class StopReasonFields(TypedDict, total=False):
stop_reason: ReadOnly[object]
stop_details: ReadOnly[object]
@ -2535,16 +2535,38 @@ def _dict_or_empty(value: object) -> object:
return value if _is_dict(value) else _no_stop_fields()
REFUSAL_STOP_FIELDS: Final[TypeAdapter[RefusalStopFields]] = TypeAdapter(
Annotated[RefusalStopFields, BeforeValidator(_dict_or_empty)]
STOP_REASON_FIELDS: Final[TypeAdapter[StopReasonFields]] = TypeAdapter(
Annotated[StopReasonFields, BeforeValidator(_dict_or_empty)]
)
_STOP_REASONS_KEPT_RAW: Final = frozenset({"refusal", "pause_turn"})
def refusal_provider_specific_fields(message: Mapping[str, object]) -> Mapping[str, object]:
def stop_reason_provider_specific_fields(message: Mapping[str, object]) -> Mapping[str, object]:
stop_reason: Final = message.get("stop_reason")
stop_details: Final = message.get("stop_details")
if stop_details is not None:
return {"stop_reason": stop_reason, "stop_details": stop_details}
if stop_reason == "refusal":
if stop_reason in _STOP_REASONS_KEPT_RAW:
return {"stop_reason": stop_reason}
return {}
class _RefusalStopDetails(TypedDict, total=False):
explanation: ReadOnly[object]
_REFUSAL_STOP_DETAILS: Final[TypeAdapter[_RefusalStopDetails]] = TypeAdapter(
Annotated[_RefusalStopDetails, BeforeValidator(_dict_or_empty)]
)
class RefusalParams(TypedDict, total=False):
refusal: ReadOnly[str]
def refusal_params(message: Mapping[str, object]) -> RefusalParams:
if message.get("stop_reason") != "refusal":
return {}
explanation: Final = _REFUSAL_STOP_DETAILS.validate_python(message.get("stop_details")).get("explanation")
return {"refusal": explanation} if isinstance(explanation, str) and explanation else {}

View file

@ -33,6 +33,7 @@ from litellm.llms.anthropic.messages.safeguards import StreamedSafeguardResults
from litellm.llms.anthropic.messages.utils import anthropic_sse_frame
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.types.llms.anthropic import (
AnthropicStopDetails,
AppliedEdit,
CompactionBlock,
ContentBlockDelta,
@ -284,9 +285,7 @@ class _CombinedChunkSplitter:
@staticmethod
def _is_combined(chunk: "ModelResponseStream") -> bool:
"""True if ``chunk`` carries response content AND a finish_reason."""
from litellm.llms.anthropic.messages.native.utils import (
openai_chat_refusal_text,
)
from litellm.llms.anthropic.messages.native.utils import refusal_reply_text
choices: Final = _optional_attr_sequence(chunk, "choices")
if not choices:
@ -302,7 +301,7 @@ class _CombinedChunkSplitter:
or _optional_attr(delta, "tool_calls")
or _optional_attr(delta, "reasoning_content")
or _optional_attr(delta, "thinking_blocks")
or openai_chat_refusal_text(delta)
or refusal_reply_text(delta)
)
_PAYLOAD_FIELD_GROUPS: "tuple[tuple[str, ...], ...]" = (
@ -659,6 +658,8 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
self.compaction_block = compaction_block
self.iterations_usage = iterations_usage
self._refusal_text: str = ""
self._paused_turn: bool = False
self._claude_refusal_stop_details: AnthropicStopDetails | None = None
self.sent_compaction_block: bool = False
# Per-phase flags so the compaction block's start/delta/stop events
# are emitted (and the public state machine is advanced) in
@ -953,7 +954,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
current_content_block_index=self.current_content_block_index,
applied_edits=(self.applied_edits if is_final_chunk and not will_merge_into_held else None),
)
processed_chunk = self._with_refusal_stop_details(processed_chunk)
processed_chunk = self._with_stop_reason_fields(processed_chunk)
# Check if this is a usage chunk and we have a held stop_reason chunk
if will_merge_into_held:
@ -1188,7 +1189,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
current_content_block_index=self.current_content_block_index,
applied_edits=(self.applied_edits if is_final_chunk and not will_merge_into_held else None),
)
processed_chunk = self._with_refusal_stop_details(processed_chunk)
processed_chunk = self._with_stop_reason_fields(processed_chunk)
# Check if this is a usage chunk and we have a held stop_reason chunk
if will_merge_into_held:
@ -1406,14 +1407,22 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
def _increment_content_block_index(self):
self.current_content_block_index += 1
def _with_refusal_stop_details(
def _with_stop_reason_fields(
self,
processed_chunk: ContentBlockDelta | MessageBlockDelta,
) -> ContentBlockDelta | MessageBlockDelta:
if processed_chunk["type"] != "message_delta" or not self._refusal_text:
if processed_chunk["type"] != "message_delta":
return processed_chunk
delta: Final = processed_chunk["delta"]
if delta.get("stop_reason") == "max_tokens":
if self._paused_turn:
paused_delta: Final[MessageDelta] = {**delta, "stop_reason": "pause_turn"}
paused_chunk: Final[MessageBlockDelta] = {**processed_chunk, "delta": paused_delta}
return paused_chunk
if delta.get("stop_reason") == "refusal" and self._claude_refusal_stop_details is not None:
claude_refusal_delta: Final[MessageDelta] = {**delta, "stop_details": self._claude_refusal_stop_details}
claude_refusal_chunk: Final[MessageBlockDelta] = {**processed_chunk, "delta": claude_refusal_delta}
return claude_refusal_chunk
if not self._refusal_text or delta.get("stop_reason") == "max_tokens":
return processed_chunk
from litellm.llms.anthropic.messages.native.utils import (
refusal_stop_details,
@ -1470,9 +1479,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
@staticmethod
def _is_blank_delta(chunk: "ModelResponseStream") -> bool:
from litellm.llms.anthropic.common_utils import is_empty_unsigned_thinking_block
from litellm.llms.anthropic.messages.native.utils import (
openai_chat_refusal_text,
)
from litellm.llms.anthropic.messages.native.utils import refusal_reply_text
choice: Final = chunk.choices[0]
if choice.finish_reason is not None:
@ -1482,7 +1489,7 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
return False
if getattr(delta, "content", None):
return False
if openai_chat_refusal_text(delta):
if refusal_reply_text(delta):
return False
if getattr(delta, "reasoning_content", None):
return False
@ -1497,6 +1504,18 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
return False
return True
def _record_anthropic_stop_fields(self, delta: object) -> None:
from litellm.llms.anthropic.messages.native.utils import (
anthropic_refusal_stop_details,
anthropic_stop_reason,
)
if anthropic_stop_reason(delta) == "pause_turn":
self._paused_turn = True
claude_refusal_stop_details: Final = anthropic_refusal_stop_details(delta)
if claude_refusal_stop_details is not None:
self._claude_refusal_stop_details = claude_refusal_stop_details
def _should_start_new_content_block(self, chunk: "ModelResponseStream") -> bool:
"""
Determine if we should start a new content block based on the processed chunk.
@ -1507,16 +1526,15 @@ class AnthropicStreamWrapper(AdapterCompletionStreamWrapper):
- Different content types in the response
- Specific markers in the content
"""
from litellm.llms.anthropic.messages.native.utils import (
openai_chat_refusal_text,
)
from litellm.llms.anthropic.messages.native.utils import refusal_reply_text
from .transformation import LiteLLMAnthropicMessagesAdapter
self._record_anthropic_stop_fields(chunk.choices[0].delta)
if chunk.choices[0].finish_reason is not None:
return False
refusal_text: Final = openai_chat_refusal_text(chunk.choices[0].delta)
refusal_text: Final = refusal_reply_text(chunk.choices[0].delta)
if refusal_text is not None:
self._refusal_text = self._refusal_text + refusal_text

View file

@ -150,7 +150,9 @@ from litellm.llms.anthropic.messages.native.mid_conversation_system import (
is_system_role_message,
)
from litellm.llms.anthropic.messages.native.utils import (
openai_chat_refusal_text,
anthropic_refusal_stop_details,
anthropic_stop_reason,
refusal_reply_text,
refusal_stop_details,
)
from litellm.types.llms.anthropic import (
@ -1467,7 +1469,7 @@ class LiteLLMAnthropicMessagesAdapter:
new_content.append(
AnthropicResponseContentBlockText(type="text", text=choice.message.content).model_dump()
)
if (refusal_text := openai_chat_refusal_text(choice.message)) is not None:
if (refusal_text := refusal_reply_text(choice.message)) is not None:
new_content.append(AnthropicResponseContentBlockText(type="text", text=refusal_text).model_dump())
# Handle tool calls (in parallel to text content)
if choice.message.tool_calls is not None and len(choice.message.tool_calls) > 0:
@ -1634,7 +1636,15 @@ class LiteLLMAnthropicMessagesAdapter:
tool_name_mapping=tool_name_mapping,
)
refusal_text: Final = next(
(text for choice in response.choices if (text := openai_chat_refusal_text(choice.message)) is not None),
(text for choice in response.choices if (text := refusal_reply_text(choice.message)) is not None),
None,
)
claude_refusal_stop_details: Final = next(
(
details
for choice in response.choices
if (details := anthropic_refusal_stop_details(choice.message)) is not None
),
None,
)
@ -1643,8 +1653,10 @@ class LiteLLMAnthropicMessagesAdapter:
## extract finish reason
openai_finish_reason: Final = response.choices[0].finish_reason if response.choices else "stop"
translated_finish_reason: Final = self._translate_openai_finish_reason_to_anthropic(
openai_finish_reason=openai_finish_reason
translated_finish_reason: Final = (
"pause_turn"
if response.choices and anthropic_stop_reason(response.choices[0].message) == "pause_turn"
else self._translate_openai_finish_reason_to_anthropic(openai_finish_reason=openai_finish_reason)
)
anthropic_finish_reason: Final = (
"compaction"
@ -1686,7 +1698,11 @@ class LiteLLMAnthropicMessagesAdapter:
usage=anthropic_usage,
content=anthropic_content,
stop_reason=anthropic_finish_reason,
stop_details=(refusal_stop_details(refusal_text) if anthropic_finish_reason == "refusal" else None),
stop_details=(
(claude_refusal_stop_details or refusal_stop_details(refusal_text))
if anthropic_finish_reason == "refusal"
else None
),
)
applied_edits: Final = polyfill_result.applied_edits_for_response() if polyfill_result else None
@ -1727,7 +1743,7 @@ class LiteLLMAnthropicMessagesAdapter:
"signature": thought_sig,
}
return "tool_use", cast("ContentBlockContentBlockDict", tool_block)
elif (choice.delta.content is not None and len(choice.delta.content) > 0) or openai_chat_refusal_text(
elif (choice.delta.content is not None and len(choice.delta.content) > 0) or refusal_reply_text(
choice.delta
) is not None:
return "text", TextBlock(type="text", text="")
@ -1810,7 +1826,7 @@ class LiteLLMAnthropicMessagesAdapter:
return "thinking_delta", ContentThinkingBlockDelta(type="thinking_delta", thinking=reasoning_content)
else:
refusal_text: Final = "".join(
refusal for choice in choices if (refusal := openai_chat_refusal_text(choice.delta)) is not None
refusal for choice in choices if (refusal := refusal_reply_text(choice.delta)) is not None
)
return "text_delta", ContentTextBlockDelta(type="text_delta", text=text + refusal_text)

View file

@ -2,7 +2,7 @@ from collections.abc import Iterable, Mapping, Sequence
from functools import lru_cache
from typing import TYPE_CHECKING, Final, cast, get_type_hints
from pydantic import JsonValue, TypeAdapter, ValidationError
from pydantic import ConfigDict, JsonValue, TypeAdapter, ValidationError, with_config
from litellm.integrations.anthropic_cache_control_hook import AnthropicCacheControlHook
from litellm.litellm_core_utils.core_helpers import as_str_mapping
@ -162,6 +162,34 @@ def openai_chat_refusal_text(message_or_delta: object) -> str | None:
return _mapping_str_field(getattr(message_or_delta, "provider_specific_fields", None), "refusal")
def anthropic_stop_reason(message_or_delta: object) -> str | None:
return _mapping_str_field(getattr(message_or_delta, "provider_specific_fields", None), "stop_reason")
def refusal_reply_text(message_or_delta: object) -> str | None:
if anthropic_stop_reason(message_or_delta) == "refusal":
return None
return openai_chat_refusal_text(message_or_delta)
@with_config(ConfigDict(extra="allow"))
class _AnthropicRefusalStopDetails(AnthropicStopDetails, total=False):
pass
_ANTHROPIC_REFUSAL_STOP_DETAILS: Final = TypeAdapter(_AnthropicRefusalStopDetails)
def anthropic_refusal_stop_details(message_or_delta: object) -> AnthropicStopDetails | None:
provider_specific_fields: Final = getattr(message_or_delta, "provider_specific_fields", None)
if _mapping_str_field(provider_specific_fields, "stop_reason") != "refusal":
return None
try:
return _ANTHROPIC_REFUSAL_STOP_DETAILS.validate_python(_mapping_field(provider_specific_fields, "stop_details"))
except ValidationError:
return refusal_stop_details(openai_chat_refusal_text(message_or_delta))
def _responses_message_refusal_text(item: object) -> str | None:
from openai.types.responses import ResponseOutputMessage, ResponseOutputRefusal

View file

@ -38,6 +38,7 @@ from itertools import accumulate, chain, groupby, repeat
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, NamedTuple, Protocol, TypeVar, Union, cast
from openai.types.responses.response_output_refusal import ResponseOutputRefusal
from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError
from typing_extensions import ReadOnly, TypedDict
@ -277,6 +278,7 @@ _TOOL_CALL_PAYLOAD_EVENT_TYPES: Final = _TOOL_CALL_PAYLOAD_DELTA_EVENT_TYPES | f
)
_OUTPUT_ITEM_EVENT_TYPES: Final = frozenset({"response.output_item.added", "response.output_item.done"})
_OUTPUT_TEXT_EVENT_TYPES: Final = frozenset({"response.output_text.delta", "response.output_text.done"})
_TEXT_DELTA_EVENT_TYPES: Final = frozenset({"response.output_text.delta", "response.refusal.delta"})
_TOOL_OUTPUT_TEXT_PART_TYPES: Final = frozenset({"input_text", "output_text", "text"})
_PATCHABLE_ITEM_FIELDS: Final[Mapping[str, str]] = MappingProxyType(
{"function_call_output": "output", "custom_tool_call_output": "output", "message": "content"}
@ -537,6 +539,14 @@ def _scanned_text_field(item: Mapping[str, object]) -> str:
return "output" if item.get("type") in TOOL_CALL_OUTPUT_ITEM_TYPES else "content"
def _content_part_text_field(part: object) -> str:
return "refusal" if stream_item_field(part, "type") == "refusal" else "text"
def _has_writable_field(part: object, field: str) -> bool:
return isinstance(part, dict) or hasattr(part, field)
def _tool_output_part_text(part: object) -> str | None:
if not isinstance(part, dict):
return None
@ -993,7 +1003,7 @@ class OpenAIResponsesHandler(BaseTranslation):
for content_idx, content_item in enumerate(content):
if isinstance(content_item, dict):
# Extract text
text_str = content_item.get("text", None)
text_str = content_item.get(_content_part_text_field(content_item), None)
if text_str is not None:
texts_to_check.append(text_str)
task_mappings.append((msg_idx, int(content_idx)))
@ -1042,8 +1052,7 @@ class OpenAIResponsesHandler(BaseTranslation):
elif isinstance(content, list):
# Replace specific text item in list content
if isinstance(messages[msg_idx][field][content_idx_optional], dict):
messages[msg_idx][field][content_idx_optional]["text"] = guardrail_response
self._write_content_part_text(content[content_idx_optional], guardrail_response)
async def process_output_response(
self,
@ -1380,7 +1389,8 @@ class OpenAIResponsesHandler(BaseTranslation):
if isinstance(output_index, int) and isinstance(content_index, int)
)
if len(placeable_positions) != len(scanned_positions) or any(
stream_item_field(event, "type") not in _OUTPUT_TEXT_EVENT_TYPES for event in scanned_events
stream_item_field(event, "type") not in _OUTPUT_TEXT_EVENT_TYPES | _TEXT_DELTA_EVENT_TYPES
for event in scanned_events
):
from litellm.proxy.policy_engine.pipeline_executor import UndeliverableStreamRewrite
@ -1401,6 +1411,12 @@ class OpenAIResponsesHandler(BaseTranslation):
else:
setattr(event, field, value)
@staticmethod
def _write_content_part_text(part: object, text: str) -> None:
field: Final = _content_part_text_field(part)
if _has_writable_field(part, field):
OpenAIResponsesHandler._write_event_field(part, field, text)
@staticmethod
def _write_compaction_summary(item: object, summary: str) -> None:
encrypted_content: Final = stream_item_field(item, "encrypted_content")
@ -1416,11 +1432,12 @@ class OpenAIResponsesHandler(BaseTranslation):
) -> None:
"""Sync pre-completion stream events with the rewritten completed
response, keyed by ``(output_index, content_index)``: the first
``output_text.delta`` for a rewritten item carries the full rewritten
text and the rest are blanked, while ``output_text.done``,
``content_part.done``, and ``output_item.done`` events carry the full
rewritten text, so every event a client may read agrees with the
rewritten ``response.completed`` payload."""
``output_text.delta`` or ``refusal.delta`` for a rewritten item carries
the full rewritten text and the rest are blanked, while
``output_text.done``, ``refusal.done``, ``content_part.done``, and
``output_item.done`` events carry the full rewritten text, so every
event a client may read agrees with the rewritten ``response.completed``
payload."""
delta_replacements: Final = MappingProxyType(
{position: chain((rewritten,), repeat("")) for position, rewritten in rewrites_by_position.items()}
)
@ -1436,14 +1453,14 @@ class OpenAIResponsesHandler(BaseTranslation):
if not isinstance(output_index, int) or not isinstance(content_index, int):
continue
position = (output_index, content_index)
if event_type == "response.output_text.delta" and position in delta_replacements:
if event_type in _TEXT_DELTA_EVENT_TYPES and position in delta_replacements:
self._write_event_field(event, "delta", next(delta_replacements[position]))
elif event_type == "response.output_text.done" and position in rewrites_by_position:
self._write_event_field(event, "text", rewrites_by_position[position])
elif event_type == "response.refusal.done" and position in rewrites_by_position:
self._write_event_field(event, "refusal", rewrites_by_position[position])
elif event_type == "response.content_part.done" and position in rewrites_by_position:
part = event.get("part")
if isinstance(part, dict) or hasattr(part, "text"):
self._write_event_field(part, "text", rewrites_by_position[position])
self._write_content_part_text(event.get("part"), rewrites_by_position[position])
@staticmethod
def _sync_output_item_done_event(
@ -1459,7 +1476,7 @@ class OpenAIResponsesHandler(BaseTranslation):
for (item_idx, content_idx), rewritten in rewrites_by_position.items():
if item_idx != output_index or not 0 <= content_idx < len(content):
continue
OpenAIResponsesHandler._write_event_field(content[content_idx], "text", rewritten)
OpenAIResponsesHandler._write_content_part_text(content[content_idx], rewritten)
def _deliver_ended_stream_tool_call_rewrites(
self,
@ -1814,7 +1831,7 @@ class OpenAIResponsesHandler(BaseTranslation):
return
# Handle both GenericResponseOutputItem and dict
content: list[OutputText] | list[dict] | None = None
content: list[OutputText | ResponseOutputRefusal] | list[dict[str, object]] | None = None
if isinstance(output_item, BaseModel):
try:
output_item_dump: Final = output_item.model_dump()
@ -1837,17 +1854,9 @@ class OpenAIResponsesHandler(BaseTranslation):
verbose_proxy_logger.debug("OpenAI Responses API: Processing output item: %s", output_item)
# Iterate through content items (list of OutputText objects)
for content_idx, content_item in enumerate(content):
# Handle both OutputText objects and dicts
if isinstance(content_item, OutputText):
text_content = content_item.text
elif isinstance(content_item, dict):
text_content = content_item.get("text")
else:
continue
if text_content:
text_content = stream_item_field(content_item, _content_part_text_field(content_item))
if isinstance(text_content, str) and text_content:
texts_to_check.append(text_content)
task_mappings.append((output_idx, int(content_idx)))
@ -1887,33 +1896,24 @@ class OpenAIResponsesHandler(BaseTranslation):
# Handle both GenericResponseOutputItem, BaseModel, and dict
if isinstance(output_item, GenericResponseOutputItem):
if output_item.content and content_idx < len(output_item.content):
content_item = output_item.content[content_idx]
if isinstance(content_item, OutputText):
content_item.text = guardrail_response
elif isinstance(content_item, dict):
content_item["text"] = guardrail_response
self._write_content_part_text(output_item.content[content_idx], guardrail_response)
elif isinstance(output_item, BaseModel):
# Handle other Pydantic models by converting to GenericResponseOutputItem
try:
generic_item = GenericResponseOutputItem.model_validate(output_item.model_dump())
if generic_item.content and content_idx < len(generic_item.content):
content_item = generic_item.content[content_idx]
if isinstance(content_item, OutputText):
content_item.text = guardrail_response
# Update the original response output
if hasattr(output_item, "content") and output_item.content:
original_content = output_item.content[content_idx]
if hasattr(original_content, "text"):
original_content.text = guardrail_response
if (
generic_item.content
and content_idx < len(generic_item.content)
and hasattr(output_item, "content")
and output_item.content
):
self._write_content_part_text(output_item.content[content_idx], guardrail_response)
except Exception:
pass
elif isinstance(output_item, dict):
content = output_item.get("content", [])
if content and content_idx < len(content):
if isinstance(content[content_idx], dict):
content[content_idx]["text"] = guardrail_response
elif hasattr(content[content_idx], "text"):
content[content_idx].text = guardrail_response
self._write_content_part_text(content[content_idx], guardrail_response)
def build_block_sse_chunks(
self,

View file

@ -24268,7 +24268,14 @@
"properties": {
"content": {
"items": {
"$ref": "#/components/schemas/OutputText"
"anyOf": [
{
"$ref": "#/components/schemas/OutputText"
},
{
"$ref": "#/components/schemas/ResponseOutputRefusal"
}
]
},
"title": "Content",
"type": "array"

View file

@ -9,6 +9,7 @@ import litellm
from litellm.litellm_core_utils.core_helpers import as_str_mapping, attribute_items
from litellm.litellm_core_utils.hidden_params import get_or_create_hidden_params, served_from_cache
from litellm.litellm_core_utils.litellm_logging import BilledRoundsSpend
from litellm.llms.anthropic.messages.native.utils import openai_chat_refusal_text
from litellm.main import stream_chunk_builder
from litellm.responses.litellm_completion_transformation.compaction_items import compaction_output_item
from litellm.responses.litellm_completion_transformation.custom_tools import (
@ -35,6 +36,7 @@ from litellm.types.llms.openai import (
ContentPartAddedEvent,
ContentPartDoneEvent,
ContentPartDonePartOutputText,
ContentPartDonePartRefusal,
FunctionCallArgumentsDeltaEvent,
FunctionCallArgumentsDoneEvent,
OutputItemAddedEvent,
@ -45,6 +47,8 @@ from litellm.types.llms.openai import (
ReasoningSummaryPartDoneEvent,
ReasoningSummaryTextDeltaEvent,
ReasoningSummaryTextDoneEvent,
RefusalDeltaEvent,
RefusalDoneEvent,
ResponseCompletedEvent,
ResponseCreatedEvent,
ResponseIncompleteEvent,
@ -110,6 +114,7 @@ def _delta_carries_output(delta: ChatCompletionDelta) -> bool:
delta.function_call,
getattr(delta, "images", None),
getattr(delta, "audio", None),
openai_chat_refusal_text(delta),
)
return any(fields) or _delta_has_signed_thinking_block(delta)
@ -119,6 +124,13 @@ def _delta_carries_reasoning_and_answer(delta: ChatCompletionDelta) -> bool:
return has_reasoning and bool(delta.content or delta.tool_calls)
_WEB_SEARCH_PROGRESS_EVENTS: Final = (
(ResponsesAPIStreamEvents.WEB_SEARCH_CALL_IN_PROGRESS, WebSearchCallInProgressEvent),
(ResponsesAPIStreamEvents.WEB_SEARCH_CALL_SEARCHING, WebSearchCallSearchingEvent),
(ResponsesAPIStreamEvents.WEB_SEARCH_CALL_COMPLETED, WebSearchCallCompletedEvent),
)
class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
"""
Async iterator for processing streaming responses from the Responses API.
@ -152,6 +164,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
self.sent_output_text_done_event: bool = False
self.sent_output_content_part_done_event: bool = False
self.sent_output_item_done_event: bool = False
self._sent_refusal_events: bool = False
self._annotation_index: int = 0
self._annotations_awaiting_message: tuple[ChatCompletionAnnotation, ...] = ()
self.litellm_model_response: ModelResponse | TextCompletionResponse | None = None
@ -592,11 +605,10 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
}
),
)
for event_type, event_class in (
(ResponsesAPIStreamEvents.WEB_SEARCH_CALL_IN_PROGRESS, WebSearchCallInProgressEvent),
(ResponsesAPIStreamEvents.WEB_SEARCH_CALL_SEARCHING, WebSearchCallSearchingEvent),
(ResponsesAPIStreamEvents.WEB_SEARCH_CALL_COMPLETED, WebSearchCallCompletedEvent),
):
progress_events: Final = (
_WEB_SEARCH_PROGRESS_EVENTS[:1] if item.status == "in_progress" else _WEB_SEARCH_PROGRESS_EVENTS
)
for event_type, event_class in progress_events:
self._sequence_number += 1
event = event_class(type=event_type, output_index=output_index, item_id=item.id)
event.__dict__["sequence_number"] = self._sequence_number
@ -798,9 +810,14 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
return event
def _queue_message_item_added_events(self) -> None:
if not self.sent_message_item_added_event:
self._queue_message_item_added_event()
self.sent_content_part_added_event = True
self._pending_response_events.append(self.create_content_part_added_event())
def _queue_message_item_added_event(self) -> None:
item_id: Final = self._message_item_id()
self.sent_message_item_added_event = True
self.sent_content_part_added_event = True
self._sequence_number += 1
event: Final = OutputItemAddedEvent(
type=ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED,
@ -817,7 +834,47 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
)
event.__dict__["sequence_number"] = self._sequence_number
self._pending_response_events.append(event)
self._pending_response_events.append(self.create_content_part_added_event())
def _queue_refusal_events(self, refusal: str) -> None:
item_id: Final = self._message_item_id()
output_index: Final = self._message_index()
content_index: Final = 1 if self.sent_content_part_added_event else 0
events: Final = (
ContentPartAddedEvent(
type=ResponsesAPIStreamEvents.CONTENT_PART_ADDED,
item_id=item_id,
output_index=output_index,
content_index=content_index,
part=ContentPartDonePartRefusal(type="refusal", refusal=""),
),
RefusalDeltaEvent(
type=ResponsesAPIStreamEvents.REFUSAL_DELTA,
item_id=item_id,
output_index=output_index,
content_index=content_index,
delta=refusal,
),
RefusalDoneEvent(
type=ResponsesAPIStreamEvents.REFUSAL_DONE,
item_id=item_id,
output_index=output_index,
content_index=content_index,
refusal=refusal,
),
ContentPartDoneEvent(
type=ResponsesAPIStreamEvents.CONTENT_PART_DONE,
item_id=item_id,
output_index=output_index,
content_index=content_index,
part=ContentPartDonePartRefusal(type="refusal", refusal=refusal),
),
)
first_sequence_number: Final = self._sequence_number + 1
self._sequence_number += len(events)
self._pending_response_events.extend(
event.model_copy(update={"sequence_number": sequence_number})
for sequence_number, event in enumerate(events, start=first_sequence_number)
)
def _merge_provider_specific_fields(self, src: dict) -> None:
"""Merge provider_specific_fields using last-value-wins for lists.
@ -974,14 +1031,22 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
if self._cached_item_id is None:
self._cached_item_id = f"msg_{uuid.uuid4()}"
text: Final = litellm_complete_object.choices[0].message.content or ""
annotations = getattr(litellm_complete_object.choices[0].message, "annotations", None)
message: Final = litellm_complete_object.choices[0].message
text: Final = message.content or ""
annotations = getattr(message, "annotations", None)
response_annotations: Final = (
LiteLLMCompletionResponsesConfig.transform_chat_completion_annotations_to_response_output_annotations(
annotations=annotations
)
)
text_parts: Final = (
({"type": "output_text", "text": text, "annotations": response_annotations},)
if self.sent_content_part_added_event
else ()
)
refusal: Final = openai_chat_refusal_text(message)
refusal_parts: Final = ({"type": "refusal", "refusal": refusal},) if refusal is not None else ()
return OutputItemDoneEvent(
type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE,
output_index=self._message_index(),
@ -992,13 +1057,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
"status": "completed",
"type": "message",
"role": "assistant",
"content": [
{
"type": "output_text",
"text": text,
"annotations": response_annotations,
}
],
"content": [*text_parts, *refusal_parts],
}
),
)
@ -1050,22 +1109,33 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
def return_default_done_events(
self, litellm_complete_object: ModelResponse
) -> BaseLiteLLMOpenAIResponseObject | None:
message: Final = litellm_complete_object.choices[0].message
refusal: Final = openai_chat_refusal_text(message)
if self.sent_message_item_added_event is False:
final_content: Final = litellm_complete_object.choices[0].message.content or ""
if not final_content:
final_content: Final = message.content or ""
if not final_content and refusal is None:
self.sent_output_text_done_event = True
self.sent_output_content_part_done_event = True
self.sent_output_item_done_event = True
return None
self._queue_message_item_added_events()
if final_content:
self._queue_message_item_added_events()
else:
self._queue_message_item_added_event()
return self._pending_response_events.pop(0)
if self.sent_output_text_done_event is False:
if self.sent_output_text_done_event is False and self.sent_content_part_added_event:
self.sent_output_text_done_event = True
return self.create_output_text_done_event(litellm_complete_object)
if self.sent_output_content_part_done_event is False:
if self.sent_output_content_part_done_event is False and self.sent_content_part_added_event:
self.sent_output_content_part_done_event = True
return self.create_output_content_part_done_event(litellm_complete_object)
if refusal is not None and not self._sent_refusal_events:
self._sent_refusal_events = True
self._queue_refusal_events(refusal)
return self._pending_response_events.pop(0)
if self.sent_output_item_done_event is False:
self.sent_output_text_done_event = True
self.sent_output_content_part_done_event = True
self.sent_output_item_done_event = True
return self.create_output_item_done_event(litellm_complete_object)
return None
@ -1169,7 +1239,9 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
# DO NOT create message item
return
# Default: message
if openai_chat_refusal_text(delta) is not None and not delta.content:
self._queue_message_item_added_event()
return
self._queue_message_item_added_events()
return
@ -1413,7 +1485,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
# Priority 2: Handle text deltas
delta_content: Final = self._get_delta_string_from_streaming_choices(chunk.choices)
if delta_content:
if not self.sent_message_item_added_event:
if not self.sent_content_part_added_event:
self._queue_message_item_added_events()
self._hold_tool_calls_behind_text(getattr(chunk.choices[0].delta, "tool_calls", None))
self._sequence_number += 1

View file

@ -8,7 +8,7 @@ import uuid
from collections.abc import Awaitable, Callable, Iterable, Iterator, Mapping, Sequence
from dataclasses import dataclass
from functools import partial
from itertools import chain, groupby
from itertools import accumulate, chain, groupby
from types import MappingProxyType
from typing import (
TYPE_CHECKING,
@ -28,7 +28,7 @@ from openai.types.chat.chat_completion_named_tool_choice_param import (
from openai.types.chat.chat_completion_named_tool_choice_param import (
Function as NamedToolChoiceFunction,
)
from openai.types.responses import ResponseFunctionToolCall, ResponseFunctionWebSearch
from openai.types.responses import ResponseFunctionToolCall, ResponseFunctionWebSearch, ResponseOutputRefusal
from openai.types.responses.response_compaction_item import ResponseCompactionItem
from openai.types.responses.response_create_params import ResponseInputParam
from openai.types.responses.tool_choice_custom_param import ToolChoiceCustomParam
@ -65,6 +65,7 @@ from litellm.litellm_core_utils.hidden_params import (
)
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.llms.anthropic.common_utils import supports_anthropic_cache_control
from litellm.llms.anthropic.messages.native.utils import anthropic_stop_reason, openai_chat_refusal_text
from litellm.llms.bedrock.base_aws_llm import BaseAWSLLM
from litellm.llms.bedrock.common_utils import bedrock_declared_cache_point_support
from litellm.llms.openrouter.chat.transformation import openrouter_model_supports_cache_control
@ -446,6 +447,32 @@ def _input_item_role(input_item: Mapping[str, object]) -> str:
return cast(str, input_item.get("role") or "user") # cast-ok: client-supplied role forwarded verbatim, unvalidated
def _starts_a_turn(input_item: Mapping[str, object]) -> bool:
item_type: Final = input_item.get("type")
if item_type in TOOL_CALL_OUTPUT_ITEM_TYPES:
return True
return item_type in (None, "message") and input_item.get("role") != "assistant"
def _is_unrun_hosted_search(input_item: Mapping[str, object]) -> bool:
return input_item.get("type") == "web_search_call" and input_item.get("status") == "in_progress"
def _without_encrypted_content(input_item: Mapping[str, object]) -> Mapping[str, object]:
return {key: value for key, value in input_item.items() if key != "encrypted_content"}
def _without_paused_turn_thinking(input_items: Sequence[Mapping[str, object]]) -> tuple[Mapping[str, object], ...]:
turns: Final = tuple(accumulate(int(_starts_a_turn(item)) for item in input_items))
paused_turns: Final = frozenset(
turn for turn, item in zip(turns, input_items, strict=True) if _is_unrun_hosted_search(item)
)
return tuple(
_without_encrypted_content(item) if turn in paused_turns and item.get("type") == "reasoning" else item
for turn, item in zip(turns, input_items, strict=True)
)
SessionMessage: TypeAlias = (
AllMessageValues
| GenericChatCompletionMessage
@ -1416,7 +1443,7 @@ class LiteLLMCompletionResponsesConfig:
messages.append(ChatCompletionUserMessage(role="user", content=input))
elif isinstance(input, list):
existing_tool_call_ids: Final[set[str]] = set()
for _input in input:
for _input in _without_paused_turn_thinking(input):
chat_completion_messages = (
LiteLLMCompletionResponsesConfig._transform_responses_api_input_item_to_chat_completion_message(
input_item=_input,
@ -2528,45 +2555,55 @@ class LiteLLMCompletionResponsesConfig:
for item in content:
if isinstance(item, str):
content_list.append(item)
elif isinstance(item, dict):
if (
media := decode_responses_media(
fields := _validate_responses_media(_STR_KEY_DICT_ADAPTER, item)
elif (
isinstance(item, dict)
and (
part
:= LiteLLMCompletionResponsesConfig._transform_responses_api_content_part_to_chat_completion_part(
_validate_responses_media(_STR_KEY_DICT_ADAPTER, item)
)
) is not None:
content_list.append(dict(media))
elif fields.get("type") == "input_image":
content_list.append(dict(_responses_image_part(fields)))
elif fields.get("type") == "input_file":
content_list.append(dict(_responses_file_part(fields)))
elif fields.get("type") == "input_audio":
content_list.append(dict(_validate_responses_media(_AUDIO_PART_ADAPTER, fields)))
elif fields.get("type") == "encrypted_content":
encrypted_content = fields.get("encrypted_content")
if encrypted_content is not None:
content_list.append(
OpenAIChatCompletionTextObject(type="text", text=str(encrypted_content))
)
else:
# Skip text blocks with None text to avoid downstream errors
text_value = fields.get("text")
if text_value is None:
continue
content_block: dict[str, object] = {
"type": LiteLLMCompletionResponsesConfig._get_chat_completion_request_content_type(
content_type
if isinstance(content_type := fields.get("type"), str) and content_type
else "text"
),
"text": text_value,
}
if "cache_control" in fields:
content_block["cache_control"] = fields["cache_control"]
content_list.append(content_block)
)
is not None
):
content_list.append(part)
return content_list
else:
raise ValueError(f"Invalid content type: {type(content)}")
@staticmethod
def _transform_responses_api_content_part_to_chat_completion_part(
fields: dict[str, object],
) -> dict[str, object] | None:
if (media := decode_responses_media(fields)) is not None:
return dict(media)
part_type: Final = fields.get("type")
if part_type == "input_image":
return dict(_responses_image_part(fields))
if part_type == "input_file":
return dict(_responses_file_part(fields))
if part_type == "input_audio":
return dict(_validate_responses_media(_AUDIO_PART_ADAPTER, fields))
if part_type == "refusal":
refusal: Final = fields.get("refusal")
return dict(OpenAIChatCompletionTextObject(type="text", text=refusal)) if isinstance(refusal, str) else None
if part_type == "encrypted_content":
encrypted_content: Final = fields.get("encrypted_content")
return (
None
if encrypted_content is None
else dict(OpenAIChatCompletionTextObject(type="text", text=str(encrypted_content)))
)
text_value: Final = fields.get("text")
if text_value is None:
return None
return {
"type": LiteLLMCompletionResponsesConfig._get_chat_completion_request_content_type(
part_type if isinstance(part_type, str) and part_type else "text"
),
"text": text_value,
**({"cache_control": fields["cache_control"]} if "cache_control" in fields else {}),
}
@staticmethod
def _get_chat_completion_request_content_type(
content_type: str,
@ -3045,6 +3082,18 @@ class LiteLLMCompletionResponsesConfig:
# Default to completed for unknown finish reasons
return "completed"
@staticmethod
def _is_paused_turn(choice: Choices) -> bool:
return anthropic_stop_reason(choice.message) == "pause_turn"
@staticmethod
def _responses_status_for_choice(choice: Choices) -> ResponsesAPIStatus:
if LiteLLMCompletionResponsesConfig._is_paused_turn(choice):
return "completed"
return LiteLLMCompletionResponsesConfig._map_chat_completion_finish_reason_to_responses_status(
choice.finish_reason
)
@staticmethod
def _incomplete_details_for_finish_reason(
finish_reason: str | None,
@ -3181,19 +3230,21 @@ class LiteLLMCompletionResponsesConfig:
"""
if isinstance(chat_completion_response, dict):
chat_completion_response = ModelResponse(**chat_completion_response)
# Get finish_reason from the first choice to determine overall status
finish_reason: str | None = None
choices: Final[list[Choices]] = getattr(chat_completion_response, "choices", [])
if choices and len(choices) > 0:
finish_reason = choices[0].finish_reason
finish_reason: Final = choices[0].finish_reason if choices else None
paused_turn: Final = bool(choices) and LiteLLMCompletionResponsesConfig._is_paused_turn(choices[0])
echoed: Final = _echoable_request_params(responses_api_request)
chat_usage: Final = getattr(chat_completion_response, "usage", None)
incomplete_details: Final = LiteLLMCompletionResponsesConfig._incomplete_details_for_finish_reason(
finish_reason=finish_reason,
existing=getattr(chat_completion_response, "incomplete_details", None),
completion_tokens=getattr(chat_usage, "completion_tokens", None),
max_output_tokens=echoed.get("max_output_tokens"),
incomplete_details: Final = (
None
if paused_turn
else LiteLLMCompletionResponsesConfig._incomplete_details_for_finish_reason(
finish_reason=finish_reason,
existing=getattr(chat_completion_response, "incomplete_details", None),
completion_tokens=getattr(chat_usage, "completion_tokens", None),
max_output_tokens=echoed.get("max_output_tokens"),
)
)
responses_api_response: Final[ResponsesAPIResponse] = ResponsesAPIResponse(
@ -3220,8 +3271,8 @@ class LiteLLMCompletionResponsesConfig:
max_output_tokens=echoed.get("max_output_tokens"),
previous_response_id=echoed.get("previous_response_id"),
reasoning=echoed.get("reasoning"),
status=LiteLLMCompletionResponsesConfig._map_chat_completion_finish_reason_to_responses_status(
finish_reason
status=(
LiteLLMCompletionResponsesConfig._responses_status_for_choice(choices[0]) if choices else "incomplete"
),
text=echoed.get("text") or {},
truncation=echoed.get("truncation"),
@ -3367,9 +3418,7 @@ class LiteLLMCompletionResponsesConfig:
GenericResponseOutputItem(
type="reasoning",
id=mint_reasoning_item_id(),
status=LiteLLMCompletionResponsesConfig._map_chat_completion_finish_reason_to_responses_status(
choice.finish_reason
),
status=LiteLLMCompletionResponsesConfig._responses_status_for_choice(choice),
role="assistant",
content=[
OutputText(
@ -3492,24 +3541,34 @@ class LiteLLMCompletionResponsesConfig:
choice=choice,
)
message_output_items.extend(image_generation_items)
elif choice.message.content is not None or getattr(choice.message, "annotations", None):
elif content := LiteLLMCompletionResponsesConfig._message_output_content(choice.message):
message_output_items.append(
GenericResponseOutputItem(
type="message",
id=f"msg_{uuid.uuid4()}",
status=LiteLLMCompletionResponsesConfig._map_chat_completion_finish_reason_to_responses_status(
choice.finish_reason
),
status=LiteLLMCompletionResponsesConfig._responses_status_for_choice(choice),
role=choice.message.role,
content=[
LiteLLMCompletionResponsesConfig._transform_chat_message_to_response_output_text(
choice.message
)
],
content=list(content),
)
)
return message_output_items
@staticmethod
def _message_output_content(message: Message) -> tuple[OutputText | ResponseOutputRefusal, ...]:
refusal: Final = openai_chat_refusal_text(message)
refusal_parts: Final = (ResponseOutputRefusal(type="refusal", refusal=refusal),) if refusal is not None else ()
has_text_part: Final = (
bool(message.content)
or bool(getattr(message, "annotations", None))
or (message.content is not None and not refusal_parts)
)
text_parts: Final = (
(LiteLLMCompletionResponsesConfig._transform_chat_message_to_response_output_text(message),)
if has_text_part
else ()
)
return (*text_parts, *refusal_parts)
@staticmethod
def _transform_responses_api_outputs_to_chat_completion_messages(
responses_api_output: ResponsesAPIResponse,

View file

@ -751,7 +751,9 @@ class AnthropicOutputTokensDetails(LiteLLMBaseModel):
thinking_tokens: int | None = None
AnthropicFinishReason = Literal["end_turn", "max_tokens", "stop_sequence", "tool_use", "refusal", "compaction"]
AnthropicFinishReason = Literal[
"end_turn", "max_tokens", "stop_sequence", "tool_use", "refusal", "pause_turn", "compaction"
]
class AnthropicResponse(LiteLLMBaseModel):

View file

@ -98,7 +98,7 @@ class AnthropicMessagesResponse(TypedDict, total=False):
model: str | None # This represents the Model type from Anthropic
role: Literal["assistant"] | None
stop_reason: ReadOnly[
Literal["end_turn", "max_tokens", "stop_sequence", "tool_use", "refusal", "compaction"] | None
Literal["end_turn", "max_tokens", "stop_sequence", "tool_use", "refusal", "compaction", "pause_turn"] | None
]
stop_details: NotRequired[ReadOnly[AnthropicStopDetails | None]]
stop_sequence: str | None

View file

@ -70,6 +70,7 @@ from pydantic import (
FieldSerializationInfo,
NonNegativeInt,
PrivateAttr,
SerializeAsAny,
SerializerFunctionWrapHandler,
field_serializer,
field_validator,
@ -1855,7 +1856,7 @@ class ContentPartAddedEvent(BaseLiteLLMOpenAIResponseObject):
item_id: str
output_index: int
content_index: int
part: BaseLiteLLMOpenAIResponseObject
part: SerializeAsAny[BaseLiteLLMOpenAIResponseObject]
class ContentPartDonePartOutputText(BaseLiteLLMOpenAIResponseObject):

View file

@ -4,6 +4,7 @@ from typing import Final, Literal, Optional, Union
from openai.types.responses.response_function_tool_call import ResponseFunctionToolCall
from openai.types.responses.response_function_web_search import ActionSearchSource, ResponseFunctionWebSearch
from openai.types.responses.response_output_refusal import ResponseOutputRefusal
from pydantic import ConfigDict, PrivateAttr
from typing_extensions import TypedDict
@ -154,7 +155,7 @@ class GenericResponseOutputItem(BaseLiteLLMOpenAIResponseObject):
id: str
status: str # "completed", "in_progress", etc.
role: str # "assistant", "user", etc.
content: list[OutputText]
content: list[OutputText | ResponseOutputRefusal]
phase: Phase = None

View file

@ -9,6 +9,15 @@ from integration._support.wire import Reply, Request, wire_server
from pydantic import JsonValue
MODEL: Final = "claude-haiku-4-5"
TEXT: Final = {"type": "text", "text": "alpha bravo"}
EXPLANATION: Final = "This request was declined under the usage policy."
PENDING_SEARCH: Final = {
"type": "server_tool_use",
"id": "srvtoolu_pending",
"name": "web_search",
"input": {"query": "next city"},
}
TERMINAL_EVENTS: Final = ("response.completed", "response.incomplete", "response.failed")
def _sse(event: Mapping[str, JsonValue]) -> bytes:
@ -21,36 +30,64 @@ def _message(identity: str, stop_reason: str | None) -> dict[str, JsonValue]:
"type": "message",
"role": "assistant",
"model": MODEL,
"content": [{"type": "text", "text": "alpha bravo"}],
"content": [TEXT],
"stop_reason": stop_reason,
"stop_sequence": None,
"usage": {"input_tokens": 199878, "output_tokens": 12799},
}
def _stream(identity: str, stop_reason: str) -> tuple[bytes, ...]:
def _block_events(index: int, block: Mapping[str, JsonValue]) -> tuple[bytes, ...]:
if block["type"] == "text":
return (
_sse({"type": "content_block_start", "index": index, "content_block": {"type": "text", "text": ""}}),
_sse(
{
"type": "content_block_delta",
"index": index,
"delta": {"type": "text_delta", "text": block["text"]},
}
),
_sse({"type": "content_block_stop", "index": index}),
)
return (
_sse({"type": "message_start", "message": {**_message(identity, None), "content": []}}),
_sse({"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}}),
_sse({"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": "alpha bravo"}}),
_sse({"type": "content_block_stop", "index": 0}),
_sse({"type": "content_block_start", "index": index, "content_block": {**block, "input": {}}}),
_sse(
{
"type": "message_delta",
"delta": {"stop_reason": stop_reason, "stop_sequence": None},
"usage": {"output_tokens": 12799},
"type": "content_block_delta",
"index": index,
"delta": {"type": "input_json_delta", "partial_json": json.dumps(block["input"])},
}
),
_sse({"type": "message_stop"}),
_sse({"type": "content_block_stop", "index": index}),
)
def _anthropic(identity: str, stop_reason: str):
def _turn(
identity: str,
stop_reason: str,
content: tuple[Mapping[str, JsonValue], ...] = (TEXT,),
stop_details: Mapping[str, JsonValue] | None = None,
):
stop: Final = {"stop_reason": stop_reason, "stop_sequence": None}
details: Final = {"stop_details": stop_details} if stop_details is not None else {}
message: Final = {**_message(identity, stop_reason), "content": list(content), **details}
stream: Final = (
_sse({"type": "message_start", "message": {**_message(identity, None), "content": []}}),
*(
event
for index, block in enumerate(content)
for event in _block_events(index, block) # comprehension-ok: block -> events flatten
),
_sse({"type": "message_delta", "delta": {**stop, **details}, "usage": {"output_tokens": 12799}}),
_sse({"type": "message_stop"}),
)
def respond(request: Request) -> Reply:
assert request.method == "POST" and request.target == "/v1/messages", request.target
if json.loads(request.body).get("stream"):
return Reply(content_type="text/event-stream", chunks=_stream(identity, stop_reason))
return Reply(body=json.dumps(_message(identity, stop_reason)).encode())
return Reply(content_type="text/event-stream", chunks=stream)
return Reply(body=json.dumps(message).encode())
return respond
@ -67,45 +104,70 @@ def _events(gateway: Gateway, path: str, body: Mapping[str, JsonValue]) -> tuple
)
def _chat_finish_reason(gateway: Gateway, model: str, *, stream: bool) -> JsonValue:
body: Final = {"model": model, "max_tokens": 64000, "messages": [{"role": "user", "content": "copy the list"}]}
def _chat_choices(gateway: Gateway, body: Mapping[str, JsonValue], *, stream: bool) -> tuple[dict[str, JsonValue], ...]:
if stream:
reasons: Final = tuple(
choice["finish_reason"]
return tuple(
choice
for chunk in _events(gateway, "/v1/chat/completions", body)
for choice in chunk["choices"] # comprehension-ok: chunk -> choices flatten
if choice.get("finish_reason")
)
assert len(reasons) == 1, reasons
return reasons[0]
response: Final = gateway.request("POST", "/v1/chat/completions", body)
assert response.status_code == 200, response.text
return response.json()["choices"][0]["finish_reason"]
return tuple(response.json()["choices"])
def _responses_status(gateway: Gateway, model: str, *, stream: bool) -> JsonValue:
body: Final = {"model": model, "max_output_tokens": 64000, "input": "copy the list"}
def _chat_body(model: str) -> dict[str, JsonValue]:
return {"model": model, "max_tokens": 64000, "messages": [{"role": "user", "content": "copy the list"}]}
def _chat_finish_reason(gateway: Gateway, model: str, *, stream: bool) -> JsonValue:
reasons: Final = tuple(
choice["finish_reason"]
for choice in _chat_choices(gateway, _chat_body(model), stream=stream)
if choice.get("finish_reason")
)
assert len(reasons) == 1, reasons
return reasons[0]
def _chat_messages(gateway: Gateway, model: str, *, stream: bool) -> tuple[dict[str, JsonValue], ...]:
return tuple(
choice.get("delta") or choice.get("message") or {}
for choice in _chat_choices(gateway, _chat_body(model), stream=stream)
)
def _responses_body(model: str) -> dict[str, JsonValue]:
return {"model": model, "max_output_tokens": 64000, "input": "copy the list", "tools": [{"type": "web_search"}]}
def _responses_result(gateway: Gateway, model: str, *, stream: bool) -> dict[str, JsonValue]:
if stream:
terminal: Final = tuple(
event["response"]
for event in _events(gateway, "/v1/responses", body)
if event["type"] in ("response.completed", "response.incomplete", "response.failed")
for event in _events(gateway, "/v1/responses", _responses_body(model))
if event["type"] in TERMINAL_EVENTS
)
assert len(terminal) == 1, terminal
return terminal[0]["status"]
response: Final = gateway.request("POST", "/v1/responses", body)
return terminal[0]
response: Final = gateway.request("POST", "/v1/responses", _responses_body(model))
assert response.status_code == 200, response.text
return response.json()["status"]
return response.json()
@pytest.mark.parametrize("stream", [False, True], ids=["json", "stream"])
@pytest.mark.parametrize(
("stop_reason", "finish_reason"),
[("model_context_window_exceeded", "length"), ("max_tokens", "length"), ("end_turn", "stop")],
[
("model_context_window_exceeded", "length"),
("max_tokens", "length"),
("pause_turn", "length"),
("end_turn", "stop"),
],
)
def test_chat_maps_anthropic_stop_reason(gateway: Gateway, stop_reason: str, finish_reason: str, stream: bool) -> None:
identity: Final = uuid.uuid4().hex
with wire_server(_anthropic(identity, stop_reason)) as wire, gateway.scenario() as scenario:
with wire_server(_turn(identity, stop_reason)) as wire, gateway.scenario() as scenario:
model: Final = scenario.model(model=f"anthropic/{MODEL}", api_base=wire.url, api_key="scripted-anthropic-key")
assert _chat_finish_reason(gateway, model, stream=stream) == finish_reason
@ -113,10 +175,81 @@ def test_chat_maps_anthropic_stop_reason(gateway: Gateway, stop_reason: str, fin
@pytest.mark.parametrize("stream", [False, True], ids=["json", "stream"])
@pytest.mark.parametrize(
("stop_reason", "status"),
[("model_context_window_exceeded", "incomplete"), ("max_tokens", "incomplete"), ("end_turn", "completed")],
[
("model_context_window_exceeded", "incomplete"),
("max_tokens", "incomplete"),
("pause_turn", "completed"),
("end_turn", "completed"),
],
)
def test_responses_maps_anthropic_stop_reason(gateway: Gateway, stop_reason: str, status: str, stream: bool) -> None:
identity: Final = uuid.uuid4().hex
with wire_server(_anthropic(identity, stop_reason)) as wire, gateway.scenario() as scenario:
with wire_server(_turn(identity, stop_reason)) as wire, gateway.scenario() as scenario:
model: Final = scenario.model(model=f"anthropic/{MODEL}", api_base=wire.url, api_key="scripted-anthropic-key")
assert _responses_status(gateway, model, stream=stream) == status
assert _responses_result(gateway, model, stream=stream)["status"] == status
@pytest.mark.parametrize("stream", [False, True], ids=["json", "stream"])
def test_chat_keeps_pause_turn_so_the_client_knows_to_resend_the_turn(gateway: Gateway, stream: bool) -> None:
identity: Final = uuid.uuid4().hex
paused: Final = _turn(identity, "pause_turn", (PENDING_SEARCH,))
with wire_server(paused) as wire, gateway.scenario() as scenario:
model: Final = scenario.model(model=f"anthropic/{MODEL}", api_base=wire.url, api_key="scripted-anthropic-key")
stop_reasons: Final = tuple(
(message.get("provider_specific_fields") or {}).get("stop_reason")
for message in _chat_messages(gateway, model, stream=stream)
)
assert "pause_turn" in stop_reasons, stop_reasons
@pytest.mark.parametrize("stream", [False, True], ids=["json", "stream"])
def test_responses_list_a_paused_turn_unrun_search_as_an_in_progress_web_search(gateway: Gateway, stream: bool) -> None:
identity: Final = uuid.uuid4().hex
paused: Final = _turn(identity, "pause_turn", (PENDING_SEARCH,))
with wire_server(paused) as wire, gateway.scenario() as scenario:
model: Final = scenario.model(model=f"anthropic/{MODEL}", api_base=wire.url, api_key="scripted-anthropic-key")
result: Final = _responses_result(gateway, model, stream=stream)
assert (result["status"], result.get("incomplete_details")) == ("completed", None)
assert [(item["type"], item["status"], item["action"]["query"]) for item in result["output"]] == [
("web_search_call", "in_progress", PENDING_SEARCH["input"]["query"])
]
_REFUSAL: Final = {"type": "refusal", "category": "cyber", "explanation": EXPLANATION}
@pytest.mark.parametrize("stream", [False, True], ids=["json", "stream"])
def test_chat_returns_the_refusal_explanation_as_the_message_refusal(gateway: Gateway, stream: bool) -> None:
identity: Final = uuid.uuid4().hex
with wire_server(_turn(identity, "refusal", (), _REFUSAL)) as wire, gateway.scenario() as scenario:
model: Final = scenario.model(model=f"anthropic/{MODEL}", api_base=wire.url, api_key="scripted-anthropic-key")
refusal: Final = "".join(
message["refusal"] for message in _chat_messages(gateway, model, stream=stream) if message.get("refusal")
)
assert refusal == EXPLANATION
@pytest.mark.parametrize("stream", [False, True], ids=["json", "stream"])
@pytest.mark.parametrize(
("content", "expected_parts"),
[
((), [("refusal", EXPLANATION)]),
((TEXT,), [("output_text", TEXT["text"]), ("refusal", EXPLANATION)]),
],
ids=["refusal_only", "answer_then_refusal"],
)
def test_responses_return_the_refusal_explanation_as_a_refusal_part(
gateway: Gateway,
content: tuple[Mapping[str, JsonValue], ...],
expected_parts: list[tuple[str, str]],
stream: bool,
) -> None:
identity: Final = uuid.uuid4().hex
with wire_server(_turn(identity, "refusal", content, _REFUSAL)) as wire, gateway.scenario() as scenario:
model: Final = scenario.model(model=f"anthropic/{MODEL}", api_base=wire.url, api_key="scripted-anthropic-key")
result: Final = _responses_result(gateway, model, stream=stream)
assert (result["status"], result["incomplete_details"]) == ("incomplete", {"reason": "content_filter"})
assert [
[(part["type"], part.get("text") or part.get("refusal")) for part in item["content"]]
for item in result["output"]
] == [expected_parts]

View file

@ -5383,8 +5383,9 @@ def test_convert_gemini_tool_call_result_answers_tool_reference_only_result():
def test_convert_to_anthropic_tool_invoke_degrades_unpaired_server_tool_use():
"""A replayed srvtoolu_ call whose server tool result is not available
(e.g. the Responses bridge replays items without provider_specific_fields)
must become a plain client tool_use so the client's tool_result can pair
with it. A dangling server_tool_use makes Anthropic 400 the request with
and that the client answered with its own tool message must become a plain
client tool_use so the client's tool_result can pair with it. A dangling
server_tool_use makes Anthropic 400 the request with
"unexpected `tool_use_id` found in `tool_result` blocks"."""
from litellm.litellm_core_utils.prompt_templates.factory import convert_to_anthropic_tool_invoke
@ -5398,6 +5399,7 @@ def test_convert_to_anthropic_tool_invoke_degrades_unpaired_server_tool_use():
],
web_search_results=None,
tool_results=None,
answered_tool_ids=frozenset({"srvtoolu_01Unpaired"}),
)
assert result == [
@ -7435,17 +7437,11 @@ def test_convert_to_anthropic_tool_invoke_sanitizes_invalid_ids():
def test_convert_to_anthropic_tool_invoke_server_tool():
"""
Test that a server tool call (srvtoolu_) with no stored result is replayed
as a regular tool_use block.
A server_tool_use block is only valid when paired with its result block, so
an unpaired one must degrade to tool_use for Anthropic to accept the replay.
A paired call still becomes server_tool_use, covered by
test_convert_to_anthropic_tool_invoke_with_web_search_results.
Context: https://github.com/BerriAI/litellm/issues/17737 (original
server_tool_use reconstruction) and LIT-6622 / PR #39144 (unpaired calls
degrade instead of 400ing at Anthropic).
A server tool call (srvtoolu_) with no stored result and no client answer is
the unrun call a pause_turn response ends on, so it is replayed as a
server_tool_use for Anthropic to run when the client resends the turn.
An answered one degrades to tool_use, covered by
test_convert_to_anthropic_tool_invoke_degrades_unpaired_server_tool_use.
"""
tool_calls = [
{
@ -7460,11 +7456,14 @@ def test_convert_to_anthropic_tool_invoke_server_tool():
result = convert_to_anthropic_tool_invoke(tool_calls)
assert len(result) == 1
assert result[0]["type"] == "tool_use"
assert result[0]["id"] == "srvtoolu_01ABC123"
assert result[0]["name"] == "web_search"
assert result[0]["input"] == {"query": "elephant weight"}
assert result == [
{
"type": "server_tool_use",
"id": "srvtoolu_01ABC123",
"name": "web_search",
"input": {"query": "elephant weight"},
}
]
def test_convert_to_anthropic_tool_invoke_with_web_search_results():
@ -8231,6 +8230,170 @@ def test_anthropic_messages_pt_interleave_more_thinking_than_tool_groups():
], f"Expected order but got: {types}"
def _paused_web_search_turn(
thinking_blocks: list[dict[str, str]] | None, content: str | list[dict[str, str]] = "Step 1 is Tokyo."
) -> dict[str, object]:
assistant_message: Final[dict[str, object]] = {
"role": "assistant",
"content": content,
"tool_calls": [
{
"id": "srvtoolu_01Done",
"type": "function",
"function": {"name": "web_search", "arguments": '{"query": "Tokyo sister cities"}'},
},
{
"id": "srvtoolu_02Done",
"type": "function",
"function": {"name": "web_search", "arguments": '{"query": "Adana sister cities"}'},
},
{
"id": "srvtoolu_01Paused",
"type": "function",
"function": {"name": "web_search", "arguments": '{"query": "Adelaide sister cities"}'},
},
],
"provider_specific_fields": {
"stop_reason": "pause_turn",
"web_search_results": [
{
"type": "web_search_tool_result",
"tool_use_id": tool_use_id,
"content": [
{"type": "web_search_result", "url": "https://example.com", "title": "City", "snippet": "x"}
],
}
for tool_use_id in ("srvtoolu_01Done", "srvtoolu_02Done")
],
},
}
if thinking_blocks is None:
return assistant_message
return {**assistant_message, "thinking_blocks": thinking_blocks}
_PAUSED_TURN_THINKING_CASES = pytest.mark.parametrize(
"thinking_blocks",
[
None,
[{"type": "thinking", "thinking": "Search first", "signature": "sig_1"}],
[
{"type": "thinking", "thinking": "Search first", "signature": "sig_1"},
{"type": "thinking", "thinking": "Search again", "signature": "sig_2"},
{"type": "thinking", "thinking": "Search once more", "signature": "sig_3"},
{"type": "thinking", "thinking": "Wrap up later", "signature": "sig_4"},
],
],
ids=["without_thinking", "fewer_thinking_than_searches", "more_thinking_than_searches"],
)
_PAUSED_TURN_CONTENT_CASES = pytest.mark.parametrize(
"content",
[
"Step 1 is Tokyo.",
[{"type": "text", "text": "Step 1 is Tokyo."}],
[
{"type": "thinking", "thinking": "Search first", "signature": "sig_1"},
{"type": "text", "text": "Step 1 is Tokyo."},
],
],
ids=["string_content", "text_parts", "text_parts_with_inline_thinking"],
)
@_PAUSED_TURN_THINKING_CASES
@_PAUSED_TURN_CONTENT_CASES
def test_anthropic_messages_pt_paused_turn_ends_on_unrun_server_tool_use(thinking_blocks, content):
"""A pause_turn response resent as-is must end on its unrun server_tool_use:
Anthropic reads a turn ending on text as prefill, which newer models reject,
and a tool_use without a tool_result is a 400. Its thinking blocks are dropped,
since chat cannot say where Claude wrote them and Anthropic rejects moved ones."""
result = anthropic_messages_pt(
[
{"role": "user", "content": "Play the sister city chain game."},
_paused_web_search_turn(thinking_blocks, content),
],
model="claude-sonnet-4-5",
llm_provider="anthropic",
)
replayed = result[-1]["content"]
assert replayed[-1] == {
"type": "server_tool_use",
"id": "srvtoolu_01Paused",
"name": "web_search",
"input": {"query": "Adelaide sister cities"},
}
assert [block["id"] for block in replayed if block["type"] == "server_tool_use"] == [
"srvtoolu_01Done",
"srvtoolu_02Done",
"srvtoolu_01Paused",
]
assert [block["type"] for block in replayed if block["type"] in ("tool_use", "thinking")] == []
assert [block.get("tool_use_id") for block in replayed if block["type"] == "web_search_tool_result"] == [
"srvtoolu_01Done",
"srvtoolu_02Done",
]
@_PAUSED_TURN_THINKING_CASES
@_PAUSED_TURN_CONTENT_CASES
def test_anthropic_messages_pt_paused_turn_before_new_user_message_drops_unrun_search(thinking_blocks, content):
"""A client may answer a paused turn with a new user message instead of resending it.
Anthropic then rejects a server_tool_use without its result, so the unrun search is dropped."""
result = anthropic_messages_pt(
[
{"role": "user", "content": "Play the sister city chain game."},
_paused_web_search_turn(thinking_blocks, content),
{"role": "user", "content": "Please continue."},
],
model="claude-sonnet-4-5",
llm_provider="anthropic",
)
assert [message["role"] for message in result] == ["user", "assistant", "user"]
replayed = result[1]["content"]
assert [block["id"] for block in replayed if block["type"] == "server_tool_use"] == [
"srvtoolu_01Done",
"srvtoolu_02Done",
]
assert [block["type"] for block in replayed if block["type"] in ("tool_use", "thinking")] == []
assert [block.get("tool_use_id") for block in replayed if block["type"] == "web_search_tool_result"] == [
"srvtoolu_01Done",
"srvtoolu_02Done",
]
assert {"type": "text", "text": "Step 1 is Tokyo."} in replayed
@_PAUSED_TURN_THINKING_CASES
@_PAUSED_TURN_CONTENT_CASES
def test_anthropic_messages_pt_paused_turn_resent_without_search_results_runs_only_its_last_search(
thinking_blocks, content
):
paused_turn = _paused_web_search_turn(thinking_blocks, content)
result = anthropic_messages_pt(
[
{"role": "user", "content": "Play the sister city chain game."},
{key: value for key, value in paused_turn.items() if key != "provider_specific_fields"},
],
model="claude-sonnet-4-5",
llm_provider="anthropic",
)
replayed = result[-1]["content"]
assert [block["id"] for block in replayed if block["type"] in ("server_tool_use", "tool_use")] == [
"srvtoolu_01Paused"
]
assert replayed[-1] == {
"type": "server_tool_use",
"id": "srvtoolu_01Paused",
"name": "web_search",
"input": {"query": "Adelaide sister cities"},
}
assert [block["type"] for block in replayed if block["type"] == "thinking"] == []
def test_anthropic_messages_pt_list_content_with_thinking_preserves_order():
"""
Test that when assistant content is already a list containing interleaved

View file

@ -265,6 +265,7 @@ class TestMapFinishReasonAnthropic:
("max_tokens", "length"),
("tool_use", "tool_calls"),
("compaction", "length"),
("pause_turn", "length"),
("model_context_window_exceeded", "length"),
("content_filtered", "content_filter"),
],

View file

@ -3651,6 +3651,51 @@ def test_streamed_refusal_stop_details_reach_the_assembled_message():
assert provider_fields["stop_details"] == stop_details
def test_streamed_refusal_explanation_reaches_the_client_as_the_message_refusal():
stop_details: Final = ANTHROPIC_DOCS_REFUSAL_MESSAGE["stop_details"]
assembled: Final = _assemble_stream(_message_stream_events("refusal", stop_details))
assert assembled.choices[0].message.refusal == stop_details["explanation"]
def test_streamed_refusal_without_an_explanation_sets_no_message_refusal():
assembled: Final = _assemble_stream(
_message_stream_events("refusal", {"type": "refusal", "category": "cyber", "explanation": None})
)
assert assembled.choices[0].message.get("refusal") is None
def _paused_search_stream_events() -> tuple[dict, ...]:
message_start, message_delta, message_stop = _message_stream_events("pause_turn", None)
return (
message_start,
{
"type": "content_block_start",
"index": 0,
"content_block": {"type": "server_tool_use", "id": "srvtoolu_pending", "name": "web_search", "input": {}},
},
{
"type": "content_block_delta",
"index": 0,
"delta": {"type": "input_json_delta", "partial_json": '{"query": "next city"}'},
},
{"type": "content_block_stop", "index": 0},
{**message_delta, "delta": {"stop_reason": "pause_turn", "stop_sequence": None}},
message_stop,
)
def test_streamed_pause_turn_is_an_unfinished_turn_listing_the_query_it_will_run():
assembled: Final = _assemble_stream(_paused_search_stream_events())
assert assembled.choices[0].finish_reason == "length"
provider_fields: Final = assembled.choices[0].message.provider_specific_fields
assert provider_fields["stop_reason"] == "pause_turn"
pending: Final = provider_fields["web_search_calls"]["srvtoolu_pending"]
assert (pending.status, pending.action.query) == ("in_progress", "next city")
def test_streamed_end_turn_carries_no_stop_details():
assembled: Final = _assemble_stream(_message_stream_events("end_turn", None))

View file

@ -9715,6 +9715,75 @@ def test_end_turn_response_carries_no_stop_details():
assert "stop_reason" not in provider_fields
def test_refusal_explanation_reaches_the_client_as_the_message_refusal():
result: Final = _transform_message_body(ANTHROPIC_DOCS_REFUSAL_MESSAGE)
assert result.choices[0].message.refusal == ANTHROPIC_DOCS_REFUSAL_MESSAGE["stop_details"]["explanation"]
@pytest.mark.parametrize(
"stop_details",
[None, {"type": "refusal", "category": "cyber", "explanation": None}, {"type": "refusal", "explanation": ""}],
ids=["no_stop_details", "null_explanation", "empty_explanation"],
)
def test_refusal_without_an_explanation_sets_no_message_refusal(stop_details: dict | None):
result: Final = _transform_message_body(_refusal_message_body("refusal", stop_details))
assert result.choices[0].message.get("refusal") is None
_RAN_SEARCH: Final = {
"type": "server_tool_use",
"id": "srvtoolu_ran",
"name": "web_search",
"input": {"query": "tokyo"},
}
_RAN_SEARCH_RESULT: Final = {
"type": "web_search_tool_result",
"tool_use_id": "srvtoolu_ran",
"content": [
{"type": "web_search_result", "url": "https://tokyo.example", "title": "Tokyo", "encrypted_content": "e"}
],
}
_PENDING_SEARCH: Final = {
"type": "server_tool_use",
"id": "srvtoolu_pending",
"name": "web_search",
"input": {"query": "next city"},
}
def _paused_turn(*content: dict) -> dict:
return {**_refusal_message_body("pause_turn", None), "content": list(content)}
def test_pause_turn_is_an_unfinished_turn_that_keeps_its_native_stop_reason():
result: Final = _transform_message_body(_paused_turn(_RAN_SEARCH, _RAN_SEARCH_RESULT, _PENDING_SEARCH))
assert result.choices[0].finish_reason == "length"
assert result.choices[0].message.provider_specific_fields["stop_reason"] == "pause_turn"
@pytest.mark.parametrize(
("content", "expected_calls"),
[
(
(_RAN_SEARCH, _RAN_SEARCH_RESULT, _PENDING_SEARCH),
[("ws_srvtoolu_ran", "completed", "tokyo"), ("ws_srvtoolu_pending", "in_progress", "next city")],
),
((_PENDING_SEARCH,), [("ws_srvtoolu_pending", "in_progress", "next city")]),
],
ids=["after_a_finished_search", "before_any_search_result"],
)
def test_pause_turn_lists_the_search_it_has_not_run_yet_as_in_progress(
content: tuple[dict, ...], expected_calls: list[tuple[str, str, str]]
):
result: Final = _transform_message_body(_paused_turn(*content))
calls: Final = result.choices[0].message.provider_specific_fields["web_search_calls"]
assert [(call.id, call.status, call.action.query) for call in calls] == expected_calls
def _json_mode_message_body(stop_reason: str, stop_details: dict | None) -> dict:
return {
**_refusal_message_body(stop_reason, stop_details),
@ -9741,6 +9810,7 @@ def test_json_mode_refusal_keeps_stop_details_on_the_message():
message: Final = result.choices[0].message
assert json.loads(message.content) == {"answer": "Hello"}
assert message.provider_specific_fields == {"stop_reason": "refusal", "stop_details": stop_details}
assert message.refusal == stop_details["explanation"]
def test_json_mode_tool_use_turn_adds_no_stop_fields_to_the_message():

View file

@ -126,6 +126,57 @@ def test_translate_chat_length_takes_precedence_over_refusal():
assert result.get("stop_details") is None
@pytest.mark.parametrize(
("provider_specific_fields", "expected_stop_reason"),
[({"stop_reason": "pause_turn"}, "pause_turn"), (None, "max_tokens")],
ids=["paused_claude_turn", "truncated_turn"],
)
def test_translate_chat_length_keeps_a_paused_claude_turn_unfinished(
provider_specific_fields: dict[str, object] | None, expected_stop_reason: str
):
"""A paused Claude turn reaches chat as ``length``, which a /v1/messages client would read as a cut-off reply."""
message: Final = Message(content="Searching", role="assistant", provider_specific_fields=provider_specific_fields)
response: Final = ModelResponse(choices=[Choices(index=0, finish_reason="length", message=message)], usage=Usage())
result: Final = LiteLLMAnthropicMessagesAdapter().translate_openai_response_to_anthropic(response)
assert result["stop_reason"] == expected_stop_reason
assert result["content"] == [{"type": "text", "text": "Searching"}]
@pytest.mark.parametrize(
("stop_details", "expected_stop_details"),
[
(
{"type": "refusal", "category": "cyber", "explanation": "I can't help with that."},
{"type": "refusal", "category": "cyber", "explanation": "I can't help with that."},
),
(
{"type": "refusal", "category": 7, "explanation": "I can't help with that."},
{"type": "refusal", "category": None, "explanation": "I can't help with that."},
),
],
ids=["claude_stop_details", "unreadable_stop_details"],
)
def test_translate_claude_refusal_keeps_the_explanation_out_of_the_reply_text(stop_details, expected_stop_details):
"""Claude's refusal explanation is API metadata, so replaying it as a text block puts words in Claude's mouth."""
message: Final = Message(
content="Partial answer",
role="assistant",
refusal=stop_details["explanation"],
provider_specific_fields={"stop_reason": "refusal", "stop_details": stop_details},
)
response: Final = ModelResponse(
choices=[Choices(index=0, finish_reason="content_filter", message=message)], usage=Usage()
)
result: Final = LiteLLMAnthropicMessagesAdapter().translate_openai_response_to_anthropic(response)
assert result["content"] == [{"type": "text", "text": "Partial answer"}]
assert result["stop_reason"] == "refusal"
assert result.get("stop_details") == expected_stop_details
def test_translate_chat_content_filter_to_anthropic_response():
response = ModelResponse(
id="chatcmpl-content-filter",

View file

@ -241,6 +241,57 @@ async def test_streaming_chat_length_takes_precedence_over_refusal(async_mode: b
assert "stop_details" not in message_delta["delta"]
CLAUDE_REFUSAL_STOP_DETAILS: Final = MappingProxyType(
{"type": "refusal", "category": "cyber", "explanation": "I can't help with that."}
)
def _claude_chat_stream(
stop_fields: Mapping[str, object], finish_reason: str, refusal: str | None = None
) -> list[MagicMock]:
"""The chunks the chat stream hands the adapter for a Claude turn: the stop fields
ride an empty delta ahead of the chunk that carries the finish_reason."""
return [
_make_chunk(Delta(content="Partial answer")),
_make_chunk(Delta(content="", refusal=refusal, provider_specific_fields=dict(stop_fields))),
_make_chunk(Delta(content=None), finish_reason=finish_reason),
]
@pytest.mark.parametrize("async_mode", [False, True])
@pytest.mark.asyncio
async def test_streaming_claude_refusal_keeps_the_explanation_in_stop_details_only(async_mode: bool):
"""Claude's refusal explanation is API metadata, so streaming it as text puts words in Claude's mouth."""
chunks: Final = _claude_chat_stream(
{"stop_reason": "refusal", "stop_details": dict(CLAUDE_REFUSAL_STOP_DETAILS)},
"content_filter",
refusal=CLAUDE_REFUSAL_STOP_DETAILS["explanation"],
)
stream: Final = _AsyncStream(chunks) if async_mode else iter(chunks)
wrapper: Final = AnthropicStreamWrapper(completion_stream=stream, model="claude-model")
events: Final = await _drain_async(wrapper) if async_mode else _drain_sync(wrapper)
assert _text_deltas(events) == ["Partial answer"]
message_delta: Final = next(event for event in events if event["type"] == "message_delta")
assert message_delta["delta"] == {"stop_reason": "refusal", "stop_details": dict(CLAUDE_REFUSAL_STOP_DETAILS)}
@pytest.mark.parametrize("async_mode", [False, True])
@pytest.mark.asyncio
async def test_streaming_paused_claude_turn_ends_on_pause_turn(async_mode: bool):
"""A paused Claude turn reaches chat as ``length``, which a /v1/messages client would read as a cut-off reply."""
chunks: Final = _claude_chat_stream({"stop_reason": "pause_turn"}, "length")
stream: Final = _AsyncStream(chunks) if async_mode else iter(chunks)
wrapper: Final = AnthropicStreamWrapper(completion_stream=stream, model="claude-model")
events: Final = await _drain_async(wrapper) if async_mode else _drain_sync(wrapper)
assert _text_deltas(events) == ["Partial answer"]
message_delta: Final = next(event for event in events if event["type"] == "message_delta")
assert message_delta["delta"] == {"stop_reason": "pause_turn"}
def _input_json_deltas(events: List[dict]) -> List[str]:
return [
e["delta"]["partial_json"]

View file

@ -25,6 +25,9 @@ from openai.types.responses import (
ResponseCustomToolCallInputDeltaEvent,
ResponseCustomToolCallInputDoneEvent,
ResponseFunctionToolCall,
ResponseOutputMessage,
ResponseOutputRefusal,
ResponseOutputText,
)
from litellm.integrations.custom_guardrail import CustomGuardrail
@ -39,7 +42,16 @@ from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
from litellm.proxy.guardrails.guardrail_hooks.bedrock_guardrails import BedrockGuardrail
from litellm.proxy.guardrails.guardrail_hooks.litellm_content_filter.content_filter import ContentFilterGuardrail
from litellm.types.guardrails import ContentFilterAction, ContentFilterPattern
from litellm.types.llms.openai import ChatCompletionToolCallChunk
from litellm.types.llms.openai import (
ChatCompletionToolCallChunk,
ContentPartDoneEvent,
ContentPartDonePartRefusal,
OutputItemDoneEvent,
RefusalDeltaEvent,
RefusalDoneEvent,
ResponseIncompleteEvent,
ResponsesAPIStreamEvents,
)
from litellm.responses.litellm_completion_transformation.compaction_items import compaction_summary
from litellm.responses.litellm_completion_transformation.transformation import (
LiteLLMCompletionResponsesConfig,
@ -5017,3 +5029,183 @@ class TestCompactionOutputSummaryReachesOutputGuardrails:
)
assert ended_key(self.SUMMARY) != ended_key("The user lives elsewhere.")
_REFUSAL: Final = "I can't share where Jane lives."
_MASKED_REFUSAL: Final = "I can't share where [NAME] lives."
class NameMaskingGuardrail(CustomGuardrail):
def __init__(self, guardrail_name: str) -> None:
super().__init__(guardrail_name=guardrail_name)
self.seen_texts: list[list[str]] = []
async def apply_guardrail(
self,
inputs: GenericGuardrailAPIInputs,
request_data: dict[str, object],
input_type: Literal["request", "response"],
logging_obj: LiteLLMLoggingObj | None = None,
) -> GenericGuardrailAPIInputs:
texts: Final = list(inputs.get("texts", []))
self.seen_texts.append(texts)
return {**inputs, "texts": [text.replace("Jane", "[NAME]") for text in texts]}
def _refused_message(*leading_text: str) -> GenericResponseOutputItem:
return GenericResponseOutputItem(
type="message",
id="msg_123",
status="incomplete",
role="assistant",
content=[
*(OutputText(type="output_text", text=text, annotations=[]) for text in leading_text),
ResponseOutputRefusal(type="refusal", refusal=_REFUSAL),
],
)
def _refused_response(*leading_text: str) -> ResponsesAPIResponse:
return ResponsesAPIResponse(
id="resp_123",
created_at=1234567890,
model="claude-opus-5-5",
object="response",
status="incomplete",
output=[_refused_message(*leading_text)],
)
class TestRefusalPartsReachTextGuardrails:
@pytest.mark.asyncio
async def test_input_refusal_part_is_scanned_and_rewritten(self) -> None:
guardrail: Final = NameMaskingGuardrail(guardrail_name="test")
data: Final = {
"model": "claude-opus-5-5",
"input": [
{"role": "user", "content": "Where does Jane live?"},
{"type": "message", "role": "assistant", "content": [{"type": "refusal", "refusal": _REFUSAL}]},
],
}
result: Final = await OpenAIResponsesHandler().process_input_messages(data, guardrail)
assert guardrail.seen_texts == [["Where does Jane live?", _REFUSAL]]
assert result["input"][1]["content"] == [{"type": "refusal", "refusal": _MASKED_REFUSAL}]
@pytest.mark.asyncio
@pytest.mark.parametrize(
("leading_text", "expected_parts"),
[
((), [("refusal", _MASKED_REFUSAL)]),
(("Jane asked first.",), [("output_text", "[NAME] asked first."), ("refusal", _MASKED_REFUSAL)]),
],
ids=["refusal_only", "text_then_refusal"],
)
async def test_output_refusal_part_is_scanned_and_rewritten(
self, leading_text: tuple[str, ...], expected_parts: list[tuple[str, str]]
) -> None:
guardrail: Final = NameMaskingGuardrail(guardrail_name="test")
result: Final = await OpenAIResponsesHandler().process_output_response(
_refused_response(*leading_text), guardrail
)
assert guardrail.seen_texts == [[*leading_text, _REFUSAL]]
assert [
(part["type"], part.get("text") or part.get("refusal")) for part in result.model_dump()["output"][0]["content"]
] == expected_parts
@pytest.mark.asyncio
async def test_output_refusal_part_in_an_openai_message_is_rewritten(self) -> None:
guardrail: Final = NameMaskingGuardrail(guardrail_name="test")
openai_message: Final = ResponseOutputMessage(
type="message",
id="msg_123",
role="assistant",
status="incomplete",
content=[
ResponseOutputText(type="output_text", text="Jane asked first.", annotations=[]),
ResponseOutputRefusal(type="refusal", refusal=_REFUSAL),
],
)
result: Final = await OpenAIResponsesHandler().process_output_response(
_refused_response().model_copy(update={"output": [openai_message]}), guardrail
)
assert guardrail.seen_texts == [["Jane asked first.", _REFUSAL]]
assert [
(part["type"], part.get("text") or part.get("refusal")) for part in result.model_dump()["output"][0]["content"]
] == [("output_text", "[NAME] asked first."), ("refusal", _MASKED_REFUSAL)]
@pytest.mark.asyncio
async def test_streamed_refusal_events_carry_the_rewrite(self) -> None:
position: Final = {"item_id": "msg_123", "output_index": 0, "content_index": 0}
events: Final = [
RefusalDeltaEvent(type=ResponsesAPIStreamEvents.REFUSAL_DELTA, delta=_REFUSAL, **position),
RefusalDoneEvent(type=ResponsesAPIStreamEvents.REFUSAL_DONE, refusal=_REFUSAL, **position),
ContentPartDoneEvent(
type=ResponsesAPIStreamEvents.CONTENT_PART_DONE,
part=ContentPartDonePartRefusal(type="refusal", refusal=_REFUSAL),
**position,
),
OutputItemDoneEvent(type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE, output_index=0, item=_refused_message()),
ResponseIncompleteEvent(type=ResponsesAPIStreamEvents.RESPONSE_INCOMPLETE, response=_refused_response()),
]
await OpenAIResponsesHandler().process_output_streaming_response(
responses_so_far=events,
guardrail_to_apply=NameMaskingGuardrail(guardrail_name="test"),
litellm_logging_obj=None,
deliver_ended_stream_rewrites=True,
)
assert [json.loads(event.model_dump_json()) for event in events] == [
json.loads(event.model_dump_json())
for event in (
RefusalDeltaEvent(type=ResponsesAPIStreamEvents.REFUSAL_DELTA, delta=_MASKED_REFUSAL, **position),
RefusalDoneEvent(type=ResponsesAPIStreamEvents.REFUSAL_DONE, refusal=_MASKED_REFUSAL, **position),
ContentPartDoneEvent(
type=ResponsesAPIStreamEvents.CONTENT_PART_DONE,
part=ContentPartDonePartRefusal(type="refusal", refusal=_MASKED_REFUSAL),
**position,
),
OutputItemDoneEvent(
type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE,
output_index=0,
item=_refused_message().model_copy(
update={"content": [ResponseOutputRefusal(type="refusal", refusal=_MASKED_REFUSAL)]}
),
),
ResponseIncompleteEvent(
type=ResponsesAPIStreamEvents.RESPONSE_INCOMPLETE,
response=_refused_response().model_copy(
update={
"output": [
_refused_message().model_copy(
update={"content": [ResponseOutputRefusal(type="refusal", refusal=_MASKED_REFUSAL)]}
)
]
}
),
),
)
]
@pytest.mark.asyncio
async def test_cut_refusal_stream_carries_the_rewrite(self) -> None:
position: Final = {"item_id": "msg_123", "output_index": 0, "content_index": 0}
events: Final = [
RefusalDeltaEvent(type=ResponsesAPIStreamEvents.REFUSAL_DELTA, delta=_REFUSAL, **position),
RefusalDoneEvent(type=ResponsesAPIStreamEvents.REFUSAL_DONE, refusal=_REFUSAL, **position),
]
await OpenAIResponsesHandler().process_output_streaming_response(
responses_so_far=events,
guardrail_to_apply=NameMaskingGuardrail(guardrail_name="test"),
litellm_logging_obj=None,
deliver_ended_stream_rewrites=True,
)
assert (events[0].delta, events[1].refusal) == (_MASKED_REFUSAL, _MASKED_REFUSAL)

View file

@ -120,8 +120,9 @@ REFUSAL_STOP_DETAILS: Final = {"type": "refusal", "category": "cyber", "explanat
[
("end_turn", {"safeguard_results": SAFEGUARD_RESULTS}),
("refusal", {"stop_details": REFUSAL_STOP_DETAILS}),
("pause_turn", {}),
],
ids=["safeguard_results", "refusal_stop_details"],
ids=["safeguard_results", "refusal_stop_details", "pause_turn"],
)
def test_rewritten_stream_keeps_the_upstream_message_delta_fields(stop_reason: str, delta_fields: dict[str, object]):
"""A guardrail that rewrites a streamed response assembles it into the OpenAI shape, which has no

View file

@ -1817,6 +1817,94 @@ async def test_streaming_step_passes_a_tool_only_stream_the_legacy_hook_left_alo
assert not any("discarded" in record.getMessage() for record in caplog.records)
class _EmailMaskingLegacyHook(CustomGuardrail):
def __init__(self):
super().__init__(guardrail_name="masker", event_hook="post_call", default_on=True)
async def async_post_call_success_hook(self, data, user_api_key_dict, response):
return {
**response,
"content": [
{**block, "text": block["text"].replace("bob@example.com", "[EMAIL]")}
if block["type"] == "text"
else block
for block in response["content"]
],
}
def _sse_frame(event, data):
return f"event: {event}\ndata: {json.dumps(data)}\n\n".encode()
def _claude_refusal_sse_stream(text):
return [
_sse_frame(
"message_start",
{
"type": "message_start",
"message": {
"id": "msg_refused",
"type": "message",
"role": "assistant",
"model": "claude-model",
"content": [],
"stop_reason": None,
"stop_sequence": None,
"usage": {"input_tokens": 3, "output_tokens": 1},
},
},
),
_sse_frame(
"content_block_start",
{"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}},
),
_sse_frame(
"content_block_delta",
{"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": text}},
),
_sse_frame("content_block_stop", {"type": "content_block_stop", "index": 0}),
_sse_frame(
"message_delta",
{
"type": "message_delta",
"delta": {
"stop_reason": "refusal",
"stop_sequence": None,
"stop_details": {"type": "refusal", "category": "cyber", "explanation": "I can't help with that."},
},
"usage": {"output_tokens": 2},
},
),
_sse_frame("message_stop", {"type": "message_stop"}),
]
def _streamed_texts(chunks):
payloads = [json.loads(chunk.decode().split("data: ", 1)[1]) for chunk in chunks]
return [
payload["delta"]["text"]
for payload in payloads
if payload["type"] == "content_block_delta" and payload["delta"]["type"] == "text_delta"
]
@pytest.mark.asyncio
async def test_streaming_step_delivers_a_legacy_mask_of_a_streamed_claude_refusal(monkeypatch, caplog):
from litellm.llms.anthropic.chat.guardrail_translation.handler import AnthropicMessagesHandler
chunks = _claude_refusal_sse_stream("Partial answer for bob@example.com")
with caplog.at_level(logging.WARNING, logger="LiteLLM Proxy"):
result = await _run_legacy_streaming_step(
monkeypatch, _EmailMaskingLegacyHook(), chunks, translation=AnthropicMessagesHandler()
)
assert result.terminal_action == "allow"
assert not any("discarded" in record.getMessage() for record in caplog.records)
assert _streamed_texts(chunks) == ["Partial answer for [EMAIL]"]
class _NoHooksGuardrail(CustomGuardrail):
pass

View file

@ -21,8 +21,9 @@ from litellm.responses.litellm_completion_transformation.session_handler import
from litellm.responses.litellm_completion_transformation.transformation import (
LiteLLMCompletionResponsesConfig,
)
from litellm.types.llms.openai import ResponsesAPIStreamEvents
from litellm.types.llms.openai import ResponsesAPIResponse, ResponsesAPIStreamEvents
from litellm.types.utils import Message, ModelResponse
from tests.unit.llms.anthropic.anthropic_docs_refusal import ANTHROPIC_DOCS_REFUSAL_MESSAGE
from tests.unit.proxy.conftest import httpx_transport
@ -1597,6 +1598,386 @@ async def test_previous_response_id_keeps_the_tools_an_earlier_codex_tool_search
assert prisma_client.db.queries == 1
_REFUSAL_EXPLANATION: Final = ANTHROPIC_DOCS_REFUSAL_MESSAGE["stop_details"]["explanation"]
_ANSWER_BEFORE_REFUSAL: Final = {"type": "text", "text": "Here is the start."}
def _part_texts(item: BaseModel) -> list[tuple[str, str]]:
return [(part["type"], part.get("text") or part.get("refusal")) for part in item.model_dump()["content"]]
@pytest.mark.parametrize(
("content", "expected_parts"),
[
((), [("refusal", _REFUSAL_EXPLANATION)]),
(
(_ANSWER_BEFORE_REFUSAL,),
[("output_text", _ANSWER_BEFORE_REFUSAL["text"]), ("refusal", _REFUSAL_EXPLANATION)],
),
],
ids=["refusal_only", "answer_then_refusal"],
)
def test_responses_over_anthropic_return_the_refusal_explanation_as_a_refusal_part(
respx_mock: respx.MockRouter, content: tuple[Mapping[str, object], ...], expected_parts: list[tuple[str, str]]
) -> None:
respx_mock.post(_ANTHROPIC_MESSAGES_URL).mock(
return_value=httpx.Response(200, json={**ANTHROPIC_DOCS_REFUSAL_MESSAGE, "content": list(content)})
)
response: Final = litellm.responses(model=_MODEL, input="Write the exploit.", api_key="mock_api_key")
assert (response.status, response.incomplete_details.reason) == ("incomplete", "content_filter")
assert [_part_texts(item) for item in response.output] == [expected_parts]
def test_a_refused_responses_turn_sent_back_keeps_its_explanation_in_the_history(
respx_mock: respx.MockRouter,
) -> None:
route: Final = respx_mock.post(_ANTHROPIC_MESSAGES_URL).mock(
return_value=httpx.Response(200, json={**ANTHROPIC_DOCS_REFUSAL_MESSAGE, "content": []})
)
refused: Final = litellm.responses(model=_MODEL, input="Write the exploit.", api_key="mock_api_key")
litellm.responses(
model=_MODEL,
input=[
{"role": "user", "content": "Write the exploit."},
*(item.model_dump(exclude_none=True) for item in refused.output),
{"role": "user", "content": "Then explain what SFT is."},
],
api_key="mock_api_key",
)
sent: Final = json.loads(route.calls.last.request.content)
assert [(message["role"], message["content"]) for message in sent["messages"][1:2]] == [
("assistant", [{"type": "text", "text": _REFUSAL_EXPLANATION}])
]
def _refusal_stream(*content: Mapping[str, object]) -> bytes:
text_events: Final = tuple(
chain.from_iterable(
(
{"type": "content_block_start", "index": index, "content_block": {"type": "text", "text": ""}},
{"type": "content_block_delta", "index": index, "delta": {"type": "text_delta", "text": block["text"]}},
{"type": "content_block_stop", "index": index},
)
for index, block in enumerate(content)
)
)
return _anthropic_sse(
{
"type": "message_start",
"message": {**ANTHROPIC_DOCS_REFUSAL_MESSAGE, "content": [], "stop_reason": None, "stop_details": None},
},
*text_events,
{
"type": "message_delta",
"delta": {
"stop_reason": "refusal",
"stop_sequence": None,
"stop_details": ANTHROPIC_DOCS_REFUSAL_MESSAGE["stop_details"],
},
"usage": {"output_tokens": 0},
},
{"type": "message_stop"},
)
_REFUSAL_PART_EVENTS: Final = (
ResponsesAPIStreamEvents.CONTENT_PART_ADDED,
ResponsesAPIStreamEvents.REFUSAL_DELTA,
ResponsesAPIStreamEvents.REFUSAL_DONE,
ResponsesAPIStreamEvents.CONTENT_PART_DONE,
)
_TEXT_PART_EVENTS: Final = (
ResponsesAPIStreamEvents.CONTENT_PART_ADDED,
ResponsesAPIStreamEvents.OUTPUT_TEXT_DELTA,
ResponsesAPIStreamEvents.OUTPUT_TEXT_DONE,
ResponsesAPIStreamEvents.CONTENT_PART_DONE,
)
@pytest.mark.parametrize(
("content", "part_events", "expected_parts"),
[
((), _REFUSAL_PART_EVENTS, [("refusal", _REFUSAL_EXPLANATION)]),
(
(_ANSWER_BEFORE_REFUSAL,),
(*_TEXT_PART_EVENTS, *_REFUSAL_PART_EVENTS),
[("output_text", _ANSWER_BEFORE_REFUSAL["text"]), ("refusal", _REFUSAL_EXPLANATION)],
),
],
ids=["refusal_only", "answer_then_refusal"],
)
def test_streamed_responses_over_anthropic_stream_the_refusal_explanation_as_a_refusal_part(
respx_mock: respx.MockRouter,
content: tuple[Mapping[str, object], ...],
part_events: tuple[ResponsesAPIStreamEvents, ...],
expected_parts: list[tuple[str, str]],
) -> None:
respx_mock.post(_ANTHROPIC_MESSAGES_URL).mock(
return_value=httpx.Response(
200, headers={"content-type": "text/event-stream"}, content=_refusal_stream(*content)
)
)
events: Final = tuple(
litellm.responses(model=_MODEL, input="Write the exploit.", stream=True, api_key="mock_api_key")
)
refusal_done: Final = next(event for event in events if event.type == ResponsesAPIStreamEvents.REFUSAL_DONE)
item_done: Final = next(event for event in events if event.type == ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE)
finished: Final = events[-1]
assert tuple(event.type for event in events) == (
ResponsesAPIStreamEvents.RESPONSE_CREATED,
ResponsesAPIStreamEvents.RESPONSE_IN_PROGRESS,
ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED,
*part_events,
ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE,
ResponsesAPIStreamEvents.RESPONSE_INCOMPLETE,
)
assert (refusal_done.content_index, refusal_done.refusal) == (len(expected_parts) - 1, _REFUSAL_EXPLANATION)
assert _part_texts(item_done.item) == expected_parts
assert finished.response.incomplete_details.reason == "content_filter"
assert [_part_texts(item) for item in finished.response.output] == [expected_parts]
def test_streamed_responses_over_anthropic_give_a_refusal_after_thinking_its_own_message(
respx_mock: respx.MockRouter,
) -> None:
respx_mock.post(_ANTHROPIC_MESSAGES_URL).mock(
return_value=httpx.Response(
200,
headers={"content-type": "text/event-stream"},
content=_anthropic_sse(
{
"type": "message_start",
"message": {
**ANTHROPIC_DOCS_REFUSAL_MESSAGE,
"content": [],
"stop_reason": None,
"stop_details": None,
},
},
{"type": "content_block_start", "index": 0, "content_block": {"type": "thinking", "thinking": ""}},
{"type": "content_block_delta", "index": 0, "delta": {"type": "thinking_delta", "thinking": "Hmm."}},
{"type": "content_block_delta", "index": 0, "delta": {"type": "signature_delta", "signature": "sig"}},
{"type": "content_block_stop", "index": 0},
{
"type": "message_delta",
"delta": {
"stop_reason": "refusal",
"stop_sequence": None,
"stop_details": ANTHROPIC_DOCS_REFUSAL_MESSAGE["stop_details"],
},
"usage": {"output_tokens": 3},
},
{"type": "message_stop"},
),
)
)
events: Final = tuple(
litellm.responses(
model=_MODEL, input="Write the exploit.", reasoning={"effort": "low"}, stream=True, api_key="mock_api_key"
)
)
refusal_done: Final = next(event for event in events if event.type == ResponsesAPIStreamEvents.REFUSAL_DONE)
finished: Final = events[-1]
assert (refusal_done.output_index, refusal_done.content_index, refusal_done.refusal) == (
1,
0,
_REFUSAL_EXPLANATION,
)
assert [item.type for item in finished.response.output] == ["reasoning", "message"]
assert _part_texts(finished.response.output[1]) == [("refusal", _REFUSAL_EXPLANATION)]
_RAN_SEARCH: Final = {
"type": "server_tool_use",
"id": "srvtoolu_ran",
"name": "web_search",
"input": {"query": "tokyo"},
}
_RAN_SEARCH_RESULT: Final = {
"type": "web_search_tool_result",
"tool_use_id": "srvtoolu_ran",
"content": [
{"type": "web_search_result", "url": "https://tokyo.example", "title": "Tokyo", "encrypted_content": "e"}
],
}
_THINKING_BEFORE_SEARCHES: Final = {"type": "thinking", "thinking": "Find each sister city in turn.", "signature": "sig"}
_TEXT_BEFORE_PENDING_SEARCH: Final = {"type": "text", "text": "Tokyo pairs with Paris. Checking Paris next."}
_PENDING_SEARCH: Final = {
"type": "server_tool_use",
"id": "srvtoolu_pending",
"name": "web_search",
"input": {"query": "next city"},
}
_EXPECTED_PAUSED_OUTPUT: Final = [
("web_search_call", "completed", "tokyo"),
("web_search_call", "in_progress", "next city"),
]
def _search_calls(output: Sequence[BaseModel]) -> list[tuple[str, str, str]]:
return [(item.type, item.status, item.action.query) for item in output if item.type == "web_search_call"]
def _reasoning_and_message_statuses(output: Sequence[BaseModel]) -> tuple[list[str], list[str]]:
return (
[item.status for item in output if item.type == "reasoning"],
[item.status for item in output if item.type == "message"],
)
def _last_item(output: Sequence[BaseModel]) -> tuple[str, str]:
return output[-1].type, output[-1].status
def _status_and_incomplete_details(response: ResponsesAPIResponse) -> tuple[str | None, Mapping[str, object] | None]:
details: Final = response.incomplete_details
return response.status, details.model_dump() if details is not None else None
_PAUSED_OR_CUT: Final = pytest.mark.parametrize(
("stop_reason", "expected_status", "expected_incomplete_details"),
[
("pause_turn", "completed", None),
("max_tokens", "incomplete", {"reason": "max_output_tokens"}),
],
ids=["paused", "max_tokens"],
)
@_PAUSED_OR_CUT
def test_responses_over_anthropic_report_a_paused_turn_as_completed_with_its_pending_search(
respx_mock: respx.MockRouter,
stop_reason: str,
expected_status: str,
expected_incomplete_details: Mapping[str, object] | None,
) -> None:
respx_mock.post(_ANTHROPIC_MESSAGES_URL).mock(
return_value=httpx.Response(
200,
json={
**ANTHROPIC_DOCS_REFUSAL_MESSAGE,
"content": [
_THINKING_BEFORE_SEARCHES,
_RAN_SEARCH,
_RAN_SEARCH_RESULT,
_TEXT_BEFORE_PENDING_SEARCH,
_PENDING_SEARCH,
],
"stop_reason": stop_reason,
"stop_details": None,
},
)
)
response: Final = litellm.responses(
model=_MODEL, input="Chain sister cities.", tools=[{"type": "web_search"}], api_key="mock_api_key"
)
assert _status_and_incomplete_details(response) == (expected_status, expected_incomplete_details)
assert _search_calls(response.output) == _EXPECTED_PAUSED_OUTPUT
assert _reasoning_and_message_statuses(response.output) == ([expected_status], [expected_status])
assert _last_item(response.output) == ("web_search_call", "in_progress")
def _server_tool_use_events(index: int, block: Mapping[str, object]) -> tuple[Mapping[str, object], ...]:
return (
{"type": "content_block_start", "index": index, "content_block": {**block, "input": {}}},
{
"type": "content_block_delta",
"index": index,
"delta": {"type": "input_json_delta", "partial_json": json.dumps(block["input"])},
},
{"type": "content_block_stop", "index": index},
)
@_PAUSED_OR_CUT
def test_streamed_responses_over_anthropic_complete_a_paused_turn_with_its_pending_search_in_progress(
respx_mock: respx.MockRouter,
stop_reason: str,
expected_status: str,
expected_incomplete_details: Mapping[str, object] | None,
) -> None:
respx_mock.post(_ANTHROPIC_MESSAGES_URL).mock(
return_value=httpx.Response(
200,
headers={"content-type": "text/event-stream"},
content=_anthropic_sse(
{
"type": "message_start",
"message": {**ANTHROPIC_DOCS_REFUSAL_MESSAGE, "content": [], "stop_reason": None},
},
{"type": "content_block_start", "index": 0, "content_block": {"type": "thinking", "thinking": ""}},
{
"type": "content_block_delta",
"index": 0,
"delta": {"type": "thinking_delta", "thinking": _THINKING_BEFORE_SEARCHES["thinking"]},
},
{
"type": "content_block_delta",
"index": 0,
"delta": {"type": "signature_delta", "signature": _THINKING_BEFORE_SEARCHES["signature"]},
},
{"type": "content_block_stop", "index": 0},
*_server_tool_use_events(1, _RAN_SEARCH),
{"type": "content_block_start", "index": 2, "content_block": _RAN_SEARCH_RESULT},
{"type": "content_block_stop", "index": 2},
{"type": "content_block_start", "index": 3, "content_block": {"type": "text", "text": ""}},
{
"type": "content_block_delta",
"index": 3,
"delta": {"type": "text_delta", "text": _TEXT_BEFORE_PENDING_SEARCH["text"]},
},
{"type": "content_block_stop", "index": 3},
*_server_tool_use_events(4, _PENDING_SEARCH),
{
"type": "message_delta",
"delta": {"stop_reason": stop_reason, "stop_sequence": None},
"usage": {"output_tokens": 5},
},
{"type": "message_stop"},
),
)
)
events: Final = tuple(
litellm.responses(
model=_MODEL,
input="Chain sister cities.",
tools=[{"type": "web_search"}],
stream=True,
api_key="mock_api_key",
)
)
pending_item_id: Final = f"ws_{_PENDING_SEARCH['id']}"
pending_events: Final = [
event.type
for event in events
if getattr(event, "item_id", None) == pending_item_id
or getattr(getattr(event, "item", None), "id", None) == pending_item_id
]
finished: Final = events[-1]
assert pending_events == [
ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED,
ResponsesAPIStreamEvents.WEB_SEARCH_CALL_IN_PROGRESS,
ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE,
]
assert finished.type == f"response.{expected_status}"
assert _status_and_incomplete_details(finished.response) == (expected_status, expected_incomplete_details)
assert _search_calls(finished.response.output) == _EXPECTED_PAUSED_OUTPUT
assert _reasoning_and_message_statuses(finished.response.output) == ([expected_status], [expected_status])
assert _last_item(finished.response.output) == ("web_search_call", "in_progress")
async def _codex_follow_up_body(**kwargs: object) -> Mapping[str, object]:
anthropic: Final = _RecordingRawAnthropicMessages()
with patch("litellm.proxy.proxy_server.prisma_client", _FakePrismaClient([_SEARCH_TURN, _CALENDAR_TURN])):

View file

@ -4339,6 +4339,7 @@ class TestEnsureOutputItemContentPartAdded:
iterator = LiteLLMCompletionStreamingIterator.__new__(LiteLLMCompletionStreamingIterator)
iterator.sent_output_item_added_event = False
iterator.sent_message_item_added_event = False
iterator.sent_content_part_added_event = False
iterator._sequence_number = 0
iterator._cached_item_id = None
@ -6816,6 +6817,51 @@ def _reasoning_item(item_id: str, thinking: Mapping[str, str]) -> Mapping[str, o
return {"type": "reasoning", "id": item_id, "summary": [], "encrypted_content": encode_thinking_blocks([thinking])}
def _hosted_search(search_id: str, status: str) -> Mapping[str, object]:
return {
"type": "web_search_call",
"id": search_id,
"status": status,
"action": {"type": "search", "query": search_id, "queries": [search_id], "sources": []},
}
@pytest.mark.parametrize(
("last_search_status", "replayed_thinking"),
[
pytest.param("in_progress", None, id="paused_turn"),
pytest.param("completed", [_ANSWERED_THINKING], id="finished_turn"),
],
)
def test_a_new_message_after_a_paused_search_turn_replays_only_that_turn_without_its_thinking(
last_search_status: str, replayed_thinking: list[Mapping[str, str]] | None
) -> None:
tokyo_thinking: Final = {"type": "thinking", "thinking": "Need Tokyo weather.", "signature": "sig_tokyo"}
messages: Final = LiteLLMCompletionResponsesConfig.transform_responses_api_request_to_chat_completion_request(
model="anthropic/claude-sonnet-5-5",
input=cast(
ResponseInputParam,
[
{"role": "user", "content": "Plan a Tokyo trip."},
_reasoning_item("rs_tokyo", tokyo_thinking),
{"type": "message", "role": "assistant", "content": [{"type": "output_text", "text": "Tokyo is sunny."}]},
_hosted_search("ws_srvtoolu_01Tokyo", "completed"),
{"role": "user", "content": "Now Osaka."},
_reasoning_item("rs_osaka", _ANSWERED_THINKING),
_CHECKING_OSAKA,
_hosted_search("ws_srvtoolu_02Osaka", "completed"),
_hosted_search("ws_srvtoolu_03Osaka", last_search_status),
{"role": "user", "content": "Please continue."},
],
),
responses_api_request=ResponsesAPIOptionalRequestParams(),
)["messages"]
assert [message["role"] for message in messages] == ["user", "assistant", "user", "assistant", "user"]
assert messages[1].get("thinking_blocks") == [tokyo_thinking]
assert messages[3].get("thinking_blocks") == replayed_thinking
def test_summary_replay_keeps_only_the_reasoning_of_the_turn_whose_call_a_later_output_answers() -> None:
tokyo_thinking: Final = {"type": "thinking", "thinking": "Need Tokyo weather.", "signature": "sig_tokyo"}
messages: Final = LiteLLMCompletionResponsesConfig.transform_responses_api_request_to_chat_completion_request(

View file

@ -1988,3 +1988,41 @@ def test_the_partial_response_billed_on_disconnect_carries_the_logging_objects_b
partial.response.usage.input_tokens,
partial.response.usage.output_tokens,
), "the bill is priced on the partial's own usage"
def _refusal_chunk(refusal: str, finish_reason: str | None = None) -> ModelResponseStream:
return ModelResponseStream(
id=CHAT_COMPLETION_ID,
created=1748575031,
model="gpt-5.5",
object="chat.completion.chunk",
choices=[
StreamingChoices(
index=0,
delta=Delta(role="assistant", content=None, refusal=refusal),
finish_reason=finish_reason,
)
],
)
@pytest.mark.parametrize("sync_mode", [True, False])
@pytest.mark.asyncio
async def test_streamed_refusal_deltas_become_one_refusal_part_of_the_message(sync_mode: bool):
iterator: Final = _build_iterator(
[_refusal_chunk("I'm sorry, "), _refusal_chunk("I can't help with that."), _chunk("", finish_reason="stop")]
)
events: Final = await _collect_events(iterator, sync_mode)
parts_added: Final = [
event.model_dump()["part"] for event in events if event.type == ResponsesAPIStreamEvents.CONTENT_PART_ADDED
]
refusal_done: Final = [event for event in events if event.type == ResponsesAPIStreamEvents.REFUSAL_DONE]
message_done: Final = [event.item for event in events if event.type == ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE]
finished: Final = events[-1]
expected_content: Final = [{"type": "refusal", "refusal": "I'm sorry, I can't help with that."}]
assert parts_added == [{"type": "refusal", "refusal": ""}]
assert [event.refusal for event in refusal_done] == [expected_content[0]["refusal"]]
assert [item.model_dump()["content"] for item in message_done] == [expected_content]
assert [item.model_dump()["content"] for item in finished.response.output] == [expected_content]

View file

@ -33337,7 +33337,7 @@ export interface components {
*/
GenericResponseOutputItem: {
/** Content */
content: components["schemas"]["OutputText"][];
content: (components["schemas"]["OutputText"] | components["schemas"]["ResponseOutputRefusal"])[];
/** Id */
id: string;
/** Phase */