scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-13 23:11:40 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 101

fix(responses): preserve hosted web search calls (#40828)

Browse source 
* fix(responses): preserve hosted web search calls

Co-Authored-By: Claude Code <noreply@anthropic.com>
(cherry picked from commit 09183b3346)

* chore: remove unrelated generated schema documentation changes

(cherry picked from commit ca6a860757)

* fix(responses): preserve hosted search context during replay

(cherry picked from commit 425f1e9b3a)

* chore: regenerate dashboard API types

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

---------

Co-authored-by: Tin Chi Lo <tin@berri.ai>
Co-authored-by: Claude Code <noreply@anthropic.com>
Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
devin-ai-integration[bot] 2026-09-12 09:52:19 -07:00 • committed by GitHub
parent 1c61c2606e
commit eddfb5fb20
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
10 changed files with 853 additions and 47 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -44,6 +44,7 @@ from litellm.types.llms.openai import (
from litellm.types.responses.main import (
OutputCodeInterpreterCall,
build_code_interpreter_log_outputs,
build_web_search_call,
)
from litellm.types.utils import (
Delta,
@ -649,6 +650,7 @@ class ModelResponseIterator:
# Accumulate web_search_tool_result blocks for multi-turn reconstruction
# See: https://github.com/BerriAI/litellm/issues/17737
self.web_search_results: list[dict[str, object]] = []
self._web_search_calls: dict[str, object] = {} # mutable-ok: provider call state by id
# Accumulate compaction blocks for multi-turn reconstruction
self.compaction_blocks: list[dict[str, object]] = []
@ -822,6 +824,19 @@ class ModelResponseIterator:
return content_block_start
def _web_search_call_snapshot(self) -> dict[str, object]:
return dict(self._web_search_calls) # mutable-ok: stream payload snapshot
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:
return
self._web_search_calls[tool_use_id] = build_web_search_call(
tool_id=tool_use_id,
tool_input=self._server_tool_inputs.get(tool_use_id, {}), # mutable-ok: empty provider input
result=result,
)
def _build_code_interpreter_results(self) -> list:
"""Convert accumulated tool_results to OutputCodeInterpreterCall objects.
@ -923,6 +938,14 @@ class ModelResponseIterator:
self._current_server_tool_id = content_block_start["content_block"]["id"]
tool_input: Final = content_block_start["content_block"].get("input", {})
self._server_tool_inputs[self._current_server_tool_id] = tool_input
if _stream_tool_name == "web_search":
self._web_search_calls[self._current_server_tool_id] = build_web_search_call(
self._current_server_tool_id,
tool_input,
{"content": []}, # mutable-ok: no provider result yet
status="in_progress",
)
provider_specific_fields["web_search_calls"] = self._web_search_call_snapshot()
# Include caller information if present (for programmatic tool calling)
if "caller" in content_block_start["content_block"]:
caller_data: Final = content_block_start["content_block"]["caller"]
@ -957,7 +980,9 @@ class ModelResponseIterator:
# The full content comes in content_block_start, not in deltas
# See: https://github.com/BerriAI/litellm/issues/17737
self.web_search_results.append(content_block_start["content_block"])
self._complete_web_search_call(content_block_start["content_block"])
provider_specific_fields["web_search_results"] = self.web_search_results
provider_specific_fields["web_search_calls"] = self._web_search_call_snapshot()
elif content_type == "web_fetch_tool_result":
# Capture web_fetch_tool_result for multi-turn reconstruction
# The full content comes in content_block_start, not in deltas

34
litellm/llms/anthropic/chat/transformation.py
View file

@ -70,6 +70,7 @@ from litellm.types.llms.openai import (
from litellm.types.responses.main import (
OutputCodeInterpreterCall,
build_code_interpreter_log_outputs,
build_web_search_call,
)
from litellm.types.utils import (
CacheCreationTokenDetails,
@ -2464,6 +2465,35 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
)
return code_interpreter_results
def _build_web_search_calls(
self,
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 = { # mutable-ok: indexes provider server inputs
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)
}
return [ # mutable-ok: provider-neutral response items
build_web_search_call(
tool_id=tool_use_id,
tool_input=inputs.get(tool_use_id, {}), # mutable-ok: empty provider input
result=result,
)
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
]
def _build_provider_specific_fields(
self,
completion_response: dict,
@ -2485,6 +2515,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,
)
if tool_results is not None:
provider_specific_fields["tool_results"] = tool_results

132
litellm/responses/litellm_completion_transformation/streaming_iterator.py
View file

@ -40,6 +40,9 @@ from litellm.types.llms.openai import (
ResponsesAPIResponse,
ResponsesAPIStreamEvents,
ResponsesAPIStreamingResponse,
WebSearchCallCompletedEvent,
WebSearchCallInProgressEvent,
WebSearchCallSearchingEvent,
)
from litellm.types.utils import Delta as ChatCompletionDelta
from litellm.types.utils import (
@ -135,6 +138,8 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
self._namespace_tool_names = LiteLLMCompletionResponsesConfig.namespace_tool_name_map(
self.responses_api_request.get("tools")
)
self._web_search_calls: dict[str, object] = {} # mutable-ok: latest call by provider id
self._queued_web_search_call_ids: set[str] = set() # mutable-ok: emitted call ids
def _get_or_assign_tool_output_index(self, call_id: str) -> int:
existing: Final = self._tool_output_index_by_call_id.get(call_id)
@ -172,6 +177,43 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
return delta.content or delta.function_call or delta.tool_calls or chunk.choices[0].finish_reason is not None
def _reserve_web_search_indexes(self, provider_fields: object) -> None:
if not isinstance(provider_fields, dict):
return
calls: Final = provider_fields.get("web_search_calls")
items: Final = calls.values() if isinstance(calls, dict) else calls if isinstance(calls, list) else ()
for item in items:
try:
call_id = item.id.removeprefix("ws_")
status = item.status
except AttributeError:
call_id = str(item.get("id", "")).removeprefix("ws_") if isinstance(item, dict) else ""
status = item.get("status") if isinstance(item, dict) else None
if call_id:
output_index = self._get_or_assign_tool_output_index(call_id)
self._web_search_calls[call_id] = item
if status == "in_progress":
self._pending_tool_events = [ # mutable-ok: replaces speculative function events
event
for event in self._pending_tool_events
if getattr(event, "output_index", None) != output_index
]
def _tool_call_id(self, tool_call: object) -> str:
index: Final = self._normalize_tool_call_index(tool_call)
call_id_raw: Final = tool_call.get("id") if isinstance(tool_call, dict) else getattr(tool_call, "id", None)
if call_id_raw:
call_id: Final = str(call_id_raw)
if index is not None:
existing: Final = self._tool_call_id_by_index.get(index)
if existing is not None and existing != call_id:
self._ambiguous_tool_call_indexes.add(index)
self._tool_call_id_by_index[index] = call_id
return call_id
if index is None or index in self._ambiguous_tool_call_indexes:
return ""
return self._tool_call_id_by_index.get(index, "")
def _queue_tool_call_delta_events(self, tool_calls: object) -> None:
"""
Convert chat-completions streaming `tool_calls` deltas into Responses API streaming events.
@ -187,28 +229,11 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
return
for tc in tool_calls:
tc_index = self._normalize_tool_call_index(tc)
call_id_raw = tc.get("id") if isinstance(tc, dict) else getattr(tc, "id", None)
call_id = ""
if call_id_raw:
call_id = str(call_id_raw)
if tc_index is not None:
existing_call_id = self._tool_call_id_by_index.get(tc_index)
if existing_call_id is not None and existing_call_id != call_id:
# Reusing the same index for multiple call_ids is ambiguous for id-less deltas.
# Guard against silent misrouting by disabling index fallback for this index.
self._ambiguous_tool_call_indexes.add(tc_index)
self._tool_call_id_by_index[tc_index] = call_id
elif tc_index is not None:
if tc_index in self._ambiguous_tool_call_indexes:
continue
mapped_call_id = self._tool_call_id_by_index.get(tc_index)
if mapped_call_id:
call_id = mapped_call_id
call_id = self._tool_call_id(tc)
if not call_id:
continue
if call_id in self._web_search_calls:
continue
fn = tc.get("function") if isinstance(tc, dict) else getattr(tc, "function", None)
fn_name = ""
@ -220,7 +245,6 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
fn_name = str(getattr(fn, "name", "") or "")
fn_args_delta = serialize_tool_call_arguments(getattr(fn, "arguments", ""))
tool_name, tool_namespace = self._responses_namespace_tool_call_fields(fn_name)
output_index = self._get_or_assign_tool_output_index(call_id)
if call_id not in self._tool_args_by_call_id:
@ -292,9 +316,15 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
fn_name = str(getattr(fn, "name", "") or "")
fn_args = serialize_tool_call_arguments(getattr(fn, "arguments", ""))
tool_name, tool_namespace = self._responses_namespace_tool_call_fields(fn_name)
web_search_call = self._web_search_calls.get(call_id)
if web_search_call is not None:
if call_id not in self._queued_web_search_call_ids:
self._queue_web_search_events(call_id, web_search_call)
self._queued_web_search_call_ids.add(call_id)
continue
# Track if this is a new tool call that wasn't streamed
is_new_tool_call = call_id not in self._tool_args_by_call_id
is_new_tool_call = call_id not in self._tool_item_id_by_call_id
# If we never sent output_item.added for this call_id, emit it now.
if is_new_tool_call:
@ -359,6 +389,49 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
)
self._pending_tool_events.append(item_done_event)
def _queue_web_search_events(self, call_id: str, web_search_call: object) -> None:
from openai.types.responses import ResponseFunctionWebSearch
item: Final = (
web_search_call
if isinstance(web_search_call, ResponseFunctionWebSearch)
else ResponseFunctionWebSearch.model_validate(web_search_call)
)
output_index: Final = self._get_or_assign_tool_output_index(call_id)
self._sequence_number += 1
added: Final = OutputItemAddedEvent(
type=ResponsesAPIStreamEvents.OUTPUT_ITEM_ADDED,
output_index=output_index,
item=BaseLiteLLMOpenAIResponseObject(
**{ # mutable-ok: BaseLiteLLM object accepts dynamic item fields
"id": item.id,
"type": item.type,
"status": "in_progress",
"action": None,
}
),
)
added.__dict__["sequence_number"] = self._sequence_number
self._pending_tool_events.append(added)
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),
):
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
self._pending_tool_events.append(event)
self._sequence_number += 1
self._pending_tool_events.append(
OutputItemDoneEvent(
type=ResponsesAPIStreamEvents.OUTPUT_ITEM_DONE,
output_index=output_index,
sequence_number=self._sequence_number,
item=BaseLiteLLMOpenAIResponseObject(**item.model_dump()),
)
)
def _adopt_response_id_from_chunk(self, chunk: ModelResponseStream) -> None:
if self._cached_response_id is not None:
return
@ -915,8 +988,6 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
chunk = await self.litellm_custom_stream_wrapper.__anext__()
if chunk is not None:
chunk = cast(ModelResponseStream, chunk)
self._ensure_output_item_for_chunk(chunk)
# Accumulate provider_specific_fields from chunk and delta
for src in (
getattr(chunk, "provider_specific_fields", None),
getattr(
@ -927,6 +998,8 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
):
if src and isinstance(src, dict):
self._merge_provider_specific_fields(src)
self._reserve_web_search_indexes(src)
self._ensure_output_item_for_chunk(chunk)
# Proceed to transformation
self.collected_chat_completion_chunks.append(
self._snapshot_chunk_for_stream_chunk_builder(chunk)
@ -1021,8 +1094,6 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
raise StopIteration
else:
chunk = self.litellm_custom_stream_wrapper.__next__()
self._ensure_output_item_for_chunk(chunk)
# Accumulate provider_specific_fields from chunk and delta
for src in (
getattr(chunk, "provider_specific_fields", None),
getattr(
@ -1033,6 +1104,8 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
):
if src and isinstance(src, dict):
self._merge_provider_specific_fields(src)
self._reserve_web_search_indexes(src)
self._ensure_output_item_for_chunk(chunk)
# Always snapshot before returning any pending events so that
# finish_reason (e.g. content_filter) is captured even when
# _ensure_output_item_for_chunk queues events on the same chunk.
@ -1168,7 +1241,12 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator):
"message",
self._cached_item_id,
)
return _output_items_with_id(message_aligned, "reasoning", self._cached_reasoning_item_id)
reasoning_aligned: Final = _output_items_with_id(
message_aligned,
"reasoning",
self._cached_reasoning_item_id,
)
return reasoning_aligned
def _emit_response_completed_event(self, litellm_model_response: ModelResponse) -> ResponseCompletedEvent | None:
if litellm_model_response:

98
litellm/responses/litellm_completion_transformation/transformation.py
View file

@ -25,7 +25,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
from openai.types.responses import ResponseFunctionToolCall, ResponseFunctionWebSearch
from openai.types.responses.response_create_params import ResponseInputParam
from openai.types.responses.tool_choice_custom_param import ToolChoiceCustomParam
from openai.types.responses.tool_choice_function_param import ToolChoiceFunctionParam
@ -45,6 +45,7 @@ from litellm.responses.litellm_completion_transformation.session_handler import
)
from litellm.types.llms.openai import (
AllMessageValues,
ChatCompletionAssistantMessage,
ChatCompletionImageObject,
ChatCompletionImageUrlObject,
ChatCompletionRedactedThinkingBlock,
@ -635,6 +636,7 @@ class LiteLLMCompletionResponsesConfig:
merged_assistant = LiteLLMCompletionResponsesConfig._merged_trailing_assistant_message(
messages=messages,
chat_completion_messages=chat_completion_messages,
hosted_search=_input.get("type") == "web_search_call",
)
if merged_assistant is not None:
messages[-1] = merged_assistant
@ -807,29 +809,44 @@ class LiteLLMCompletionResponsesConfig:
chat_completion_messages: Sequence[
AllMessageValues | GenericChatCompletionMessage | ChatCompletionResponseMessage
],
) -> ChatCompletionResponseMessage | None:
"""Fold an assistant content message into a directly preceding assistant
tool_calls message. Providers like DeepSeek and Anthropic require tool
results immediately after the tool_calls message, so an assistant message
between them is rejected."""
hosted_search: bool = False,
) -> ChatCompletionAssistantMessage | None:
"""Keep replayed search context on the assistant turn so client tool results
still immediately follow the assistant that requested them."""
if not messages or len(chat_completion_messages) != 1:
return None
last_message = messages[-1]
new_message = chat_completion_messages[0]
if not isinstance(last_message, dict):
if not isinstance(messages[-1], dict):
return None
last_message: Final = _STR_KEY_DICT_ADAPTER.validate_python(messages[-1])
new_message: Final = _STR_KEY_DICT_ADAPTER.validate_python(chat_completion_messages[0])
if last_message.get("role") != "assistant" or new_message.get("role") != "assistant":
return None
if not last_message.get("tool_calls") or last_message.get("content") or new_message.get("tool_calls"):
if not (last_message.get("tool_calls") or hosted_search) or new_message.get("tool_calls"):
return None
new_content = new_message.get("content")
new_content: Final = new_message.get("content")
if new_content is None:
return None
previous_content: Final = last_message.get("content")
content: Final = (
new_content
if not previous_content
else [ # mutable-ok: outbound chat content uses JSON arrays
block
for value in (previous_content, new_content)
for block in (
(ChatCompletionTextObject(type="text", text=value),)
if isinstance(value, str)
else _OBJECT_LIST_ADAPTER.validate_python(value)
)
]
)
merged: Final = { # mutable-ok: json.dumps rejects MappingProxyType in outbound chat messages
**last_message,
"content": new_content,
"content": content,
}
return cast(ChatCompletionResponseMessage, merged) # cast-ok: TypedDict spread widens to dict[str, object]
return cast( # cast-ok: preserves the assistant fields and content blocks
ChatCompletionAssistantMessage, merged
)
@staticmethod
def _deduplicate_tool_call_output_messages(
@ -1252,6 +1269,14 @@ class LiteLLMCompletionResponsesConfig:
- ResponseReasoningItemParam
- ItemReference
"""
if input_item.get("type") == "web_search_call":
search: Final = ResponseFunctionWebSearch.model_validate(input_item)
return [ # mutable-ok: input conversion returns chat message lists
GenericChatCompletionMessage(
role="assistant",
content="Hosted web search: " + search.model_dump_json(exclude_none=True),
)
]
if LiteLLMCompletionResponsesConfig._is_input_item_tool_call_output(input_item):
# handle executed tool call results
return (
@ -1438,7 +1463,6 @@ class LiteLLMCompletionResponsesConfig:
return input_item.get("type") in [
"function_call_output",
"custom_tool_call_output",
"web_search_call",
"computer_call_output",
"tool_result", # Anthropic/MCP format
]
@ -2041,7 +2065,7 @@ class LiteLLMCompletionResponsesConfig:
def transform_chat_completion_tools_to_responses_tools(
chat_completion_response: ModelResponse,
responses_api_request: ResponsesAPIOptionalRequestParams | None = None,
) -> list[ResponseFunctionToolCall | CustomToolCallOutputItem]:
) -> list[ResponseFunctionToolCall | ResponseFunctionWebSearch | CustomToolCallOutputItem]:
"""
Transform a Chat Completion tools into a Responses API tools.
@ -2064,7 +2088,12 @@ class LiteLLMCompletionResponsesConfig:
custom_tool_names: Final = extract_custom_tool_names(request_tools)
namespace_tool_names: Final = LiteLLMCompletionResponsesConfig.namespace_tool_name_map(request_tools)
responses_tools: Final[list[ResponseFunctionToolCall | CustomToolCallOutputItem]] = []
web_search_calls: Final = LiteLLMCompletionResponsesConfig._web_search_calls_by_call_id(
chat_completion_response
)
responses_tools: Final[
list[ResponseFunctionToolCall | ResponseFunctionWebSearch | CustomToolCallOutputItem]
] = [] # mutable-ok: preserves provider tool-call order
for tool in all_chat_completion_tools:
if tool.type == "function":
function_definition = tool.function
@ -2072,8 +2101,10 @@ class LiteLLMCompletionResponsesConfig:
tool_id = tool.id or ""
tool_arguments = serialize_tool_call_arguments(function_definition.get("arguments"))
# Check if this is a custom tool
if is_custom_tool_call(tool_name, custom_tool_names):
web_search_call = web_search_calls.get(tool_id)
if web_search_call is not None:
responses_tools.append(web_search_call)
elif is_custom_tool_call(tool_name, custom_tool_names):
# Build custom_tool_call output item
input_str = unwrap_custom_tool_arguments(tool_arguments)
custom_item = CustomToolCallOutputItem(
@ -2128,6 +2159,35 @@ class LiteLLMCompletionResponsesConfig:
responses_tools.append(output_tool_call)
return responses_tools
@staticmethod
def _web_search_calls_by_call_id(
chat_completion_response: ModelResponse,
) -> Mapping[str, ResponseFunctionWebSearch]:
calls: Final[dict[str, ResponseFunctionWebSearch]] = {} # mutable-ok: indexes provider-built calls
for choice in chat_completion_response.choices:
provider_fields = getattr(choice.message, "provider_specific_fields", None)
if not isinstance(provider_fields, Mapping):
continue
web_search_calls = provider_fields.get("web_search_calls")
items = (
web_search_calls.values()
if isinstance(web_search_calls, Mapping)
else web_search_calls
if isinstance(web_search_calls, Sequence)
else ()
)
for item in items:
try:
call = (
item
if isinstance(item, ResponseFunctionWebSearch)
else ResponseFunctionWebSearch.model_validate(item)
)
except (TypeError, ValueError):
continue
calls[call.id.removeprefix("ws_")] = call
return MappingProxyType(calls)
@staticmethod
def _map_chat_completion_finish_reason_to_responses_status(
finish_reason: str | None,
@ -2326,6 +2386,7 @@ class LiteLLMCompletionResponsesConfig:
| OutputFunctionToolCall
| OutputImageGenerationCall
| ResponseFunctionToolCall
| ResponseFunctionWebSearch
| CustomToolCallOutputItem
]:
responses_output: list[
@ -2334,6 +2395,7 @@ class LiteLLMCompletionResponsesConfig:
| OutputFunctionToolCall
| OutputImageGenerationCall
| ResponseFunctionToolCall
| ResponseFunctionWebSearch
| CustomToolCallOutputItem
] = []

2
litellm/types/llms/openai.py
View file

@ -61,6 +61,7 @@ from openai.types.responses.response_create_params import (
ToolParam,
)
from openai.types.responses.response_function_tool_call import ResponseFunctionToolCall
from openai.types.responses.response_function_web_search import ResponseFunctionWebSearch
from pydantic import (
BaseModel,
ConfigDict,
@ -1358,6 +1359,7 @@ class ResponsesAPIResponse(BaseLiteLLMOpenAIResponseObject):
| OutputFunctionToolCall
| OutputImageGenerationCall
| ResponseFunctionToolCall
| ResponseFunctionWebSearch
| CustomToolCallOutputItem
]
)

32
litellm/types/responses/main.py
View file

@ -1,6 +1,8 @@
from collections.abc import Mapping, Sequence
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 pydantic import PrivateAttr
from typing_extensions import Any, TypedDict
@ -39,6 +41,36 @@ class OutputFunctionToolCall(BaseLiteLLMOpenAIResponseObject):
phase: Phase = None
def build_web_search_call(
tool_id: str,
tool_input: object,
result: object,
status: Literal["in_progress", "searching", "completed", "failed"] | None = None,
) -> ResponseFunctionWebSearch:
query: Final = tool_input.get("query", "") if isinstance(tool_input, Mapping) else ""
content: Final = result.get("content") if isinstance(result, Mapping) else None
result_items: Final = content if isinstance(content, Sequence) and not isinstance(content, (str, bytes)) else ()
sources: Final = [ # mutable-ok: official SDK expects a source list
ActionSearchSource(type="url", url=url)
for item in result_items
if isinstance(item, Mapping)
and item.get("type") == "web_search_result"
and isinstance((url := item.get("url")), str)
]
failed: Final = isinstance(content, Mapping) and content.get("type") == "web_search_tool_result_error"
return ResponseFunctionWebSearch(
id=f"ws_{tool_id}",
type="web_search_call",
status=status or ("failed" if failed else "completed"),
action={ # mutable-ok: official SDK expects an action mapping
"type": "search",
"query": query if isinstance(query, str) else "",
"queries": [query] if isinstance(query, str) and query else [], # mutable-ok: SDK list field
"sources": sources,
},
)
class OutputImageGenerationCall(BaseLiteLLMOpenAIResponseObject):
"""An image generation call output"""

46
tests/test_litellm/llms/anthropic/chat/test_anthropic_chat_handler.py
View file

@ -1382,6 +1382,52 @@ def test_current_content_block_type_tracking():
assert iterator.current_content_block_type is None
def test_web_search_calls_are_cumulative_through_incomplete_search():
iterator = ModelResponseIterator(None, sync_stream=True)
first_start = iterator.chunk_parser(
{
"type": "content_block_start",
"index": 0,
"content_block": {
"type": "server_tool_use",
"id": "srvtoolu_A",
"name": "web_search",
"input": {"query": "a"},
},
}
)
first_result = iterator.chunk_parser(
{
"type": "content_block_start",
"index": 1,
"content_block": {
"type": "web_search_tool_result",
"tool_use_id": "srvtoolu_A",
"content": [],
},
}
)
second_start = iterator.chunk_parser(
{
"type": "content_block_start",
"index": 2,
"content_block": {
"type": "server_tool_use",
"id": "srvtoolu_B",
"name": "web_search",
"input": {"query": "b"},
},
}
)
assert list(first_start.choices[0].delta.provider_specific_fields["web_search_calls"]) == ["srvtoolu_A"]
assert first_result.choices[0].delta.provider_specific_fields["web_search_calls"]["srvtoolu_A"].status == "completed"
calls = second_start.choices[0].delta.provider_specific_fields["web_search_calls"]
assert list(calls) == ["srvtoolu_A", "srvtoolu_B"]
assert calls["srvtoolu_A"].status == "completed"
assert calls["srvtoolu_B"].status == "in_progress"
def test_web_search_tool_result_captured_in_provider_specific_fields():
"""
Test that web_search_tool_result content is captured in provider_specific_fields.

320
tests/test_litellm/responses/litellm_completion_transformation/test_litellm_completion_responses.py
View file

@ -1,13 +1,23 @@
import json
from typing import Final
from copy import deepcopy
from typing import Final, Literal
import pytest
from openai.types.responses.response_function_web_search import (
ActionFind,
ActionOpenPage,
ActionSearch,
ActionSearchSource,
ResponseFunctionWebSearch,
)
import litellm
from litellm.litellm_core_utils.prompt_templates.factory import anthropic_messages_pt
from litellm.responses.litellm_completion_transformation.transformation import (
TOOL_CALLS_CACHE,
LiteLLMCompletionResponsesConfig,
)
from litellm.types.responses.main import build_web_search_call
from litellm.types.utils import (
ChatCompletionMessageToolCall,
Choices,
@ -3513,6 +3523,8 @@ class TestEnsureOutputItemContentPartAdded:
iterator._custom_tool_names = set()
iterator.responses_api_request = {}
iterator._namespace_tool_names = LiteLLMCompletionResponsesConfig.namespace_tool_name_map(None)
iterator._web_search_calls = {}
iterator._queued_web_search_call_ids = set()
return iterator
def _make_text_chunk(self):
@ -4017,6 +4029,211 @@ def test_function_call_tool_id_falls_back_to_unique_id_for_degenerate_call_id():
assert convert(openai)["id"] == "call_tokyo"
class TestHostedWebSearchReplay:
def test_emitted_hosted_search_output_round_trips_with_client_tool_result(self) -> None:
search_result: Final = {
"type": "web_search_tool_result",
"tool_use_id": "srvtoolu_round_trip_search",
"content": [{"type": "web_search_result", "url": "https://example.com/forecast"}],
}
search: Final = build_web_search_call(
tool_id="srvtoolu_round_trip_search", tool_input={"query": "Paris forecast"}, result=search_result
)
message: Final = Message(
role="assistant",
content="I found a forecast source.",
tool_calls=[
ChatCompletionMessageToolCall(
id="srvtoolu_round_trip_search",
type="function",
function=Function(name="web_search", arguments='{"query":"Paris forecast"}'),
),
ChatCompletionMessageToolCall(
id="call_round_trip_weather",
type="function",
function=Function(name="get_weather", arguments='{"city":"Paris"}'),
),
],
provider_specific_fields={"web_search_calls": [search], "web_search_results": [search_result]},
)
response: Final = LiteLLMCompletionResponsesConfig.transform_chat_completion_response_to_responses_api_response(
request_input="Find a forecast source and check the weather in Paris.",
responses_api_request={
"tools": [
{"type": "web_search"},
{"type": "function", "name": "get_weather", "parameters": {"type": "object"}},
]
},
chat_completion_response=_bridged_chat_completion_response(
choices=[Choices(index=0, finish_reason="tool_calls", message=message)]
),
)
assert [item for item in response.output if item.type == "web_search_call"] == [search]
assert [item.call_id for item in response.output if item.type == "function_call"] == ["call_round_trip_weather"]
history: Final = [
{"role": "user", "content": "Find a forecast source and check the weather in Paris."},
*(item.model_dump(exclude_none=True) for item in response.output),
{"type": "function_call_output", "call_id": "call_round_trip_weather", "output": "Paris is sunny."},
]
original: Final = deepcopy(history)
messages: Final = LiteLLMCompletionResponsesConfig.transform_responses_api_input_to_messages(
input=history, responses_api_request={}
)
assert [item.get("role") for item in messages] == ["user", "assistant", "tool"]
assistant: Final = messages[1]
assert [call["id"] for call in assistant["tool_calls"]] == ["call_round_trip_weather"]
assert [call["function"]["name"] for call in assistant["tool_calls"]] == ["get_weather"]
content: Final = assistant["content"]
assert isinstance(content, list)
text_parts: Final = tuple(block["text"] for block in content if block.get("type") == "text")
assert text_parts[0] == "I found a forecast source."
replayed_searches: Final = tuple(
ResponseFunctionWebSearch.model_validate_json(text[text.index("{"):])
for text in text_parts
if "web_search_call" in text
)
assert replayed_searches == (search,)
assert messages[2]["tool_call_id"] == "call_round_trip_weather"
assert messages[2]["content"] == "Paris is sunny."
assert history == original
@pytest.mark.parametrize(
"action",
(
ActionSearch(
type="search",
query="hosted search history",
queries=["hosted search history", "search replay"],
sources=[ActionSearchSource(type="url", url="https://example.com/search-result")],
),
ActionOpenPage(type="open_page", url="https://example.com/opened-page"),
ActionFind(type="find_in_page", url="https://example.com/find-page", pattern="search history"),
),
ids=("search", "open_page", "find"),
)
@pytest.mark.parametrize("status", ("completed", "failed"))
def test_replays_typed_search_action_without_client_tool_call(
self,
action: ActionSearch | ActionOpenPage | ActionFind,
status: Literal["completed", "failed"],
) -> None:
search: Final = ResponseFunctionWebSearch(
id="ws_replayed_search", type="web_search_call", status=status, action=action
)
input_item: Final = search.model_dump(exclude_none=True)
original: Final = deepcopy(input_item)
messages: Final = LiteLLMCompletionResponsesConfig._transform_responses_api_input_item_to_chat_completion_message(
input_item=input_item
)
assert len(messages) == 1
assert messages[0]["role"] == "assistant"
assert not messages[0].get("tool_calls")
content: Final = messages[0].get("content")
assert isinstance(content, str)
replayed: Final = ResponseFunctionWebSearch.model_validate_json(content[content.index("{"):])
assert replayed == search
assert input_item == original
@pytest.mark.parametrize("order", ((0, 1, 2, 3), (1, 0, 3, 2), (1, 3, 0, 2)))
@pytest.mark.parametrize("modify_params", (False, True))
@pytest.mark.parametrize("structured_content", (False, True))
def test_search_replay_preserves_client_tool_result_adjacency(
self,
monkeypatch: pytest.MonkeyPatch,
order: tuple[int, int, int, int],
modify_params: bool,
structured_content: bool,
) -> None:
monkeypatch.setattr(litellm, "modify_params", modify_params)
searches: Final = tuple(
ResponseFunctionWebSearch(
id=f"ws_search_{index}",
type="web_search_call",
status="completed",
action=ActionSearch(
type="search",
query=f"search query {index}",
queries=[f"search query {index}"],
sources=[ActionSearchSource(type="url", url=f"https://example.com/result-{index}")],
),
)
for index in (1, 2)
)
replay_items: Final = (
{
"type": "function_call",
"name": "get_weather",
"call_id": "call_weather",
"arguments": '{"city":"Paris"}',
},
searches[0].model_dump(exclude_none=True),
{"type": "function_call", "name": "get_time", "call_id": "call_time", "arguments": "{}"},
searches[1].model_dump(exclude_none=True),
)
history: Final = [
{"role": "user", "content": "Research the forecast and call get_weather."},
{
"role": "assistant",
"content": [{"type": "output_text", "text": "I will check the forecast."}]
if structured_content
else "I will check the forecast.",
},
*(replay_items[index] for index in order),
{"role": "assistant", "content": [{"type": "output_text", "text": "I found two sources."}]},
{"type": "function_call_output", "call_id": "call_weather", "output": "Paris is sunny."},
{"type": "function_call_output", "call_id": "call_time", "output": "12:00"},
]
original: Final = deepcopy(history)
messages: Final = LiteLLMCompletionResponsesConfig.transform_responses_api_input_to_messages(
input=history, responses_api_request={}
)
assert [message.get("role") for message in messages] == ["user", "assistant", "tool", "tool"]
assistant: Final = messages[1]
assert [call["id"] for call in assistant["tool_calls"]] == ["call_weather", "call_time"]
assert [call["function"]["name"] for call in assistant["tool_calls"]] == ["get_weather", "get_time"]
assert messages[2]["tool_call_id"] == "call_weather"
assert messages[2]["content"] == "Paris is sunny."
assert messages[3]["tool_call_id"] == "call_time"
assert messages[3]["content"] == "12:00"
content: Final = assistant["content"]
assert isinstance(content, list)
text_parts: Final = tuple(block["text"] for block in content if block.get("type") == "text")
assert text_parts[0] == "I will check the forecast."
assert text_parts[-1] == "I found two sources."
replayed_searches: Final = tuple(
ResponseFunctionWebSearch.model_validate_json(text[text.index("{"):])
for text in text_parts
if "web_search_call" in text
)
assert replayed_searches == searches
assert history == original
provider_messages: Final = anthropic_messages_pt(
messages=messages, model="claude-fable-5-1", llm_provider="anthropic"
)
assert [message["role"] for message in provider_messages] == ["user", "assistant", "user"]
assistant_blocks: Final = provider_messages[1]["content"]
result_blocks: Final = provider_messages[2]["content"]
assert [block["id"] for block in assistant_blocks if block.get("type") == "tool_use"] == [
"call_weather", "call_time"
]
assert [block["tool_use_id"] for block in result_blocks if block.get("type") == "tool_result"] == [
"call_weather", "call_time"
]
assert [block["content"] for block in result_blocks if block.get("type") == "tool_result"] == [
"Paris is sunny.", "12:00"
]
assert [block["text"] for block in assistant_blocks if block.get("type") == "text"] == list(text_parts)
assert history == original
BRIDGED_CHAT_COMPLETION_ID = "chatcmpl-dfa2da3a-1586-4ff7-b64e-f59c692a5d11"
@ -4058,6 +4275,105 @@ class TestBridgedOutputItemIdPrefixes:
chat_completion_response=chat_completion_response,
)
@pytest.mark.parametrize(
"tool_type,result_kind,expected_sources",
[
(
"web_search",
"valid",
{"srvtoolu_01Search": ["https://example.com/one"], "srvtoolu_02Search": ["https://example.com/two"]},
),
(
"web_search_preview",
"valid",
{"srvtoolu_01Search": ["https://example.com/one"], "srvtoolu_02Search": ["https://example.com/two"]},
),
("function", "valid", {}),
("web_search", "unpaired", {"srvtoolu_01Search": ["https://example.com/one"]}),
("web_search", "web_fetch", {"srvtoolu_02Search": ["https://example.com/two"]}),
("web_search", "error", {"srvtoolu_01Search": [], "srvtoolu_02Search": ["https://example.com/two"]}),
],
)
def test_anthropic_web_search_output_mapping(self, tool_type, result_kind, expected_sources):
call_ids: Final = ("srvtoolu_01Search", "srvtoolu_02Search")
valid_results: Final = (
{
"type": "web_search_tool_result",
"tool_use_id": call_ids[0],
"content": [{"type": "web_search_result", "url": "https://example.com/one"}],
},
{
"type": "web_search_tool_result",
"tool_use_id": call_ids[1],
"content": [{"type": "web_search_result", "url": "https://example.com/two"}],
},
)
first_result: Final = (
{**valid_results[0], "type": "web_fetch_tool_result"}
if result_kind == "web_fetch"
else {**valid_results[0], "content": {"type": "web_search_tool_result_error", "error_code": "unavailable"}}
if result_kind == "error"
else valid_results[0]
)
results: Final = (first_result,) if result_kind == "unpaired" else (first_result, valid_results[1])
message: Final = Message(
role="assistant",
content="answer",
tool_calls=[
ChatCompletionMessageToolCall(
id=call_id,
type="function",
function=Function(name="web_search", arguments=json.dumps({"query": query})),
)
for call_id, query in zip(call_ids, ("one", "two"), strict=True)
]
+ [
ChatCompletionMessageToolCall(
id="toolu_regular",
type="function",
function=Function(name="get_weather", arguments='{"city":"Paris"}'),
)
],
provider_specific_fields={
"web_search_results": results,
"web_search_calls": [
build_web_search_call(
tool_id=result["tool_use_id"],
tool_input={"query": "one" if result["tool_use_id"].endswith("01Search") else "two"},
result=result,
)
for result in results
if tool_type != "function" and result["type"] == "web_search_tool_result"
],
},
)
request_tools: Final = (
[{"type": "function", "name": "web_search", "parameters": {"type": "object"}}]
if tool_type == "function"
else [{"type": tool_type}]
)
response: Final = LiteLLMCompletionResponsesConfig.transform_chat_completion_response_to_responses_api_response(
request_input="search",
responses_api_request={"tools": request_tools},
chat_completion_response=_bridged_chat_completion_response(
choices=[Choices(index=0, finish_reason="stop", message=message)]
),
)
search_items: Final = {
item.id.removeprefix("ws_"): item for item in response.output if item.type == "web_search_call"
}
function_ids: Final = {item.call_id for item in response.output if item.type == "function_call"}
assert set(search_items) == set(expected_sources)
assert function_ids == set(call_ids).difference(expected_sources) | {"toolu_regular"}
assert [item.content[0].text for item in response.output if item.type == "message"] == ["answer"]
for call_id, item in search_items.items():
assert item.status == ("failed" if result_kind == "error" and call_id.endswith("01Search") else "completed")
assert item.action.type == "search"
assert item.action.query == ("one" if call_id.endswith("01Search") else "two")
assert item.action.queries == [item.action.query]
assert [source.url for source in item.action.sources] == expected_sources[call_id]
def test_message_item_id_uses_msg_prefix(self):
response = self._transform(_bridged_chat_completion_response())

209
tests/test_litellm/responses/litellm_completion_transformation/test_streaming_iterator_transformation.py
View file

@ -21,6 +21,7 @@ from litellm.responses.litellm_completion_transformation.streaming_iterator impo
)
from litellm.responses.utils import ResponsesAPIRequestUtils
from litellm.types.llms.openai import ResponsesAPIStreamEvents
from litellm.types.responses.main import build_web_search_call
from litellm.types.utils import (
Delta,
ModelResponse,
@ -140,6 +141,214 @@ def test_tool_call_delta_is_emitted_as_responses_events():
assert len(evt2.delta) <= 10 # Chunks are max 10 characters
@pytest.mark.asyncio
@pytest.mark.parametrize("sync_mode", [True, False])
@pytest.mark.parametrize(
"tool_type,result_kind,expected_sources",
[
(
"web_search",
"valid",
{"srvtoolu_01Search": ["https://example.com/one"], "srvtoolu_02Search": ["https://example.com/two"]},
),
(
"web_search_preview",
"valid",
{"srvtoolu_01Search": ["https://example.com/one"], "srvtoolu_02Search": ["https://example.com/two"]},
),
("function", "valid", {}),
("web_search", "unpaired", {"srvtoolu_01Search": ["https://example.com/one"]}),
("web_search", "web_fetch", {"srvtoolu_02Search": ["https://example.com/two"]}),
("web_search", "error", {"srvtoolu_01Search": [], "srvtoolu_02Search": ["https://example.com/two"]}),
],
)
async def test_web_search_stream_preserves_hosted_and_client_calls(sync_mode, tool_type, result_kind, expected_sources):
call_ids: Final = ("srvtoolu_01Search", "srvtoolu_02Search")
valid_results: Final = (
{
"type": "web_search_tool_result",
"tool_use_id": call_ids[0],
"content": [{"type": "web_search_result", "url": "https://example.com/one"}],
},
{
"type": "web_search_tool_result",
"tool_use_id": call_ids[1],
"content": [{"type": "web_search_result", "url": "https://example.com/two"}],
},
)
first_result: Final = (
{**valid_results[0], "type": "web_fetch_tool_result"}
if result_kind == "web_fetch"
else {**valid_results[0], "content": {"type": "web_search_tool_result_error", "error_code": "unavailable"}}
if result_kind == "error"
else valid_results[0]
)
results: Final = [first_result] if result_kind == "unpaired" else [first_result, valid_results[1]]
deltas: Final = (
Delta(
role="assistant",
content=None,
tool_calls=[
{"index": 0, "id": call_ids[0], "type": "function", "function": {"name": "web_search", "arguments": ""}}
],
provider_specific_fields={
"web_search_calls": [
build_web_search_call(
call_ids[0],
{},
{"content": []},
status="in_progress",
)
]
if tool_type != "function" and result_kind != "web_fetch"
else [],
},
),
Delta(
content=None,
tool_calls=[
{
"index": 1,
"id": "toolu_regular",
"type": "function",
"function": {"name": "get_weather", "arguments": '{"city":"Paris"}'},
}
],
),
Delta(content=None, tool_calls=[{"index": 0, "function": {"arguments": '{"query":'}}]),
Delta(content=None, tool_calls=[{"index": 0, "function": {"arguments": '"one"}'}}]),
Delta(
content=None,
provider_specific_fields={
"web_search_results": [first_result],
"web_search_calls": [
build_web_search_call(call_ids[0], {"query": "one"}, first_result)
]
if tool_type != "function" and first_result["type"] == "web_search_tool_result"
else [],
},
),
Delta(
content=None,
provider_specific_fields={
"web_search_results": results,
"web_search_calls": [
build_web_search_call(
result["tool_use_id"],
{"query": "one" if result["tool_use_id"].endswith("01Search") else "two"},
result,
)
for result in results
if tool_type != "function" and result["type"] == "web_search_tool_result"
],
},
),
Delta(
content="answer",
tool_calls=[
{
"index": 2,
"id": call_ids[1],
"type": "function",
"function": {"name": "web_search", "arguments": '{"query":"two"}'},
}
],
),
)
chunks: Final = tuple(
ModelResponseStream(
id=CHAT_COMPLETION_ID,
created=1748575031,
model="claude-fable-5-1",
object="chat.completion.chunk",
choices=[
StreamingChoices(index=0, delta=delta, finish_reason="stop" if index == len(deltas) - 1 else None)
],
)
for index, delta in enumerate(deltas)
)
request_tools: Final = (
[{"type": "function", "name": "web_search", "parameters": {"type": "object"}}]
if tool_type == "function"
else [{"type": tool_type}]
)
iterator: Final = LiteLLMCompletionStreamingIterator(
model="claude-fable-5-1",
litellm_custom_stream_wrapper=_FakeStreamWrapper(chunks),
request_input="search",
responses_api_request={"tools": request_tools},
custom_llm_provider="anthropic",
)
events: Final = (
[event.model_dump(exclude_none=True) for event in iterator]
if sync_mode
else [event.model_dump(exclude_none=True) async for event in iterator]
)
completed: Final = events[-1]
search_items: Final = {
item["id"].removeprefix("ws_"): item
for item in completed["response"]["output"]
if item["type"] == "web_search_call"
}
function_items: Final = {
item["call_id"]: item for item in completed["response"]["output"] if item["type"] == "function_call"
}
function_events: Final = [event for event in events if "function_call_arguments" in event["type"]]
expected_functions: Final = set(call_ids).difference(expected_sources) | {"toolu_regular"}
search_indexes: Final = {
event["output_index"] for event in events if event["type"] == "response.web_search_call.completed"
}
completed_indexes: Final = {item["id"]: index for index, item in enumerate(completed["response"]["output"])}
assert completed["type"] == "response.completed"
assert [item["content"][0]["text"] for item in completed["response"]["output"] if item["type"] == "message"] == [
"answer"
]
assert set(search_items) == set(expected_sources)
assert set(function_items) == expected_functions
assert {event["item_id"] for event in function_events} == {item["id"] for item in function_items.values()}
assert len(search_indexes) == len(expected_sources)
for call_id, item in search_items.items():
search_events = [
event for event in events if event.get("item_id", event.get("item", {}).get("id")) == item["id"]
]
assert [event["type"] for event in search_events] == [
"response.output_item.added",
"response.web_search_call.in_progress",
"response.web_search_call.searching",
"response.web_search_call.completed",
"response.output_item.done",
]
assert {event["output_index"] for event in search_events} == {completed_indexes[item["id"]]}
assert search_events[0]["item"]["status"] == "in_progress"
assert search_events[-1]["item"] == item
assert item["status"] == (
"failed" if result_kind == "error" and call_id.endswith("01Search") else "completed"
)
assert item["action"]["type"] == "search"
assert item["action"]["query"] == ("one" if call_id.endswith("01Search") else "two")
assert item["action"]["queries"] == [item["action"]["query"]]
assert [source["url"] for source in item["action"]["sources"]] == expected_sources[call_id]
for call_id, item in function_items.items():
argument_deltas = [
event["delta"]
for event in function_events
if event["item_id"] == item["id"] and event["type"].endswith(".delta")
]
assert json.loads("".join(argument_deltas)) == json.loads(item["arguments"])
assert json.loads(item["arguments"]) == (
{"city": "Paris"}
if call_id == "toolu_regular"
else {"query": "one" if call_id.endswith("01Search") else "two"}
)
assert any(
event["type"] == "response.output_item.done"
and event.get("item") == item
and event["output_index"] == completed_indexes[item["id"]]
for event in events
)
def test_tool_calls_present_only_in_final_response_are_emitted_before_completed():
iterator = LiteLLMCompletionStreamingIterator(
model="test-model",

2
tests/test_litellm/responses/test_custom_tool_call.py
View file

@ -374,6 +374,8 @@ class TestTransformationCustomTools:
"srvtoolu_01ServerCall",
"toolu_01CustomCall",
]
assert result[1].type == "function_call"
assert result[1].name == "web_search"
def test_transform_mixed_tool_calls(self):
"""Test transformation with both custom and regular tool calls."""