diff --git a/litellm/litellm_core_utils/prompt_templates/factory.py b/litellm/litellm_core_utils/prompt_templates/factory.py index eebf6b73b44..560d756ab9a 100644 --- a/litellm/litellm_core_utils/prompt_templates/factory.py +++ b/litellm/litellm_core_utils/prompt_templates/factory.py @@ -2617,6 +2617,7 @@ def anthropic_messages_pt( unique_tool_ids: set[str] = set() assistant_content: list[AnthropicMessagesAssistantMessageValues] = [] + answered_tool_ids = answered_tool_call_ids(messages, msg_i) ## MERGE CONSECUTIVE ASSISTANT CONTENT ## while msg_i < len(messages) and messages[msg_i]["role"] == "assistant": assistant_content_block: ChatCompletionAssistantMessage = messages[msg_i] @@ -2670,7 +2671,7 @@ def anthropic_messages_pt( assistant_tool_calls, web_search_results=_web_search_results_tc, tool_results=_tool_results_tc, - answered_tool_ids=answered_tool_call_ids(messages, msg_i), + answered_tool_ids=answered_tool_ids, ) # Group tool invoke results into (server_tool_use, result) pairs @@ -2854,7 +2855,7 @@ def anthropic_messages_pt( assistant_tool_calls, web_search_results=_web_search_results, tool_results=_tool_results, - answered_tool_ids=answered_tool_call_ids(messages, msg_i), + answered_tool_ids=answered_tool_ids, ) # Prevent "tool_use ids must be unique" errors by filtering duplicates diff --git a/litellm/llms/openai/responses/guardrail_translation/handler.py b/litellm/llms/openai/responses/guardrail_translation/handler.py index 61e0127f5b1..e5cfbd4d28b 100644 --- a/litellm/llms/openai/responses/guardrail_translation/handler.py +++ b/litellm/llms/openai/responses/guardrail_translation/handler.py @@ -63,6 +63,7 @@ from litellm.llms.base_llm.guardrail_translation.utils import ( unappliable_request_rewrite, ) from litellm.llms.openai.responses.guardrail_translation.tool_merge import merge_guardrailed_tools +from litellm.responses.litellm_completion_transformation.custom_tools import without_malformed_tool_flags from litellm.responses.litellm_completion_transformation.transformation import ( LiteLLMCompletionResponsesConfig, ) @@ -567,8 +568,9 @@ class OpenAIResponsesHandler(BaseTranslation): original_tools: Final[tuple[Mapping[str, object], ...]] = ( tuple(raw_tools) if isinstance(raw_tools, list) else () ) + scanned_tools: Final = tuple(without_malformed_tool_flags(tool) for tool in original_tools) flattened_tool_groups: Final = tuple( - form.chat_tools for form in LiteLLMCompletionResponsesConfig.responses_tools_to_chat_forms(original_tools) + form.chat_tools for form in LiteLLMCompletionResponsesConfig.responses_tools_to_chat_forms(scanned_tools) ) extracted: Final = self._extract_guardrail_inputs( data, input_data, flattened_tool_groups, skip_system=skip_system diff --git a/litellm/responses/litellm_completion_transformation/custom_tools.py b/litellm/responses/litellm_completion_transformation/custom_tools.py index 3492bc99af6..a69e3d28b56 100644 --- a/litellm/responses/litellm_completion_transformation/custom_tools.py +++ b/litellm/responses/litellm_completion_transformation/custom_tools.py @@ -196,18 +196,30 @@ def _is_well_formed_tool_flag(key: str, value: object) -> bool: return True -_NAMESPACE_MEMBERS_ADAPTER: Final = TypeAdapter[list[dict[str, object]]](list[dict[str, object]]) +_NAMESPACE_MEMBERS_ADAPTER: Final = TypeAdapter[list[object]](list[object]) +_TOOL_ADAPTER: Final = TypeAdapter[dict[str, object]](dict[str, object]) + + +def _with_well_formed_flags(tool: Mapping[str, object]) -> Mapping[str, object]: + return {key: value for key, value in tool.items() if _is_well_formed_tool_flag(key, value)} + + +def _member_with_well_formed_flags(member: object) -> object: + try: + return _with_well_formed_flags(_TOOL_ADAPTER.validate_python(member)) + except ValidationError: + return member def without_malformed_tool_flags(tool: Mapping[str, object]) -> Mapping[str, object]: - kept: Final = {key: value for key, value in tool.items() if _is_well_formed_tool_flag(key, value)} + kept: Final = _with_well_formed_flags(tool) if tool.get("type") != "namespace": return kept try: members: Final = _NAMESPACE_MEMBERS_ADAPTER.validate_python(tool.get("tools")) except ValidationError: return kept - return {**kept, "tools": [without_malformed_tool_flags(member) for member in members]} + return {**kept, "tools": [_member_with_well_formed_flags(member) for member in members]} def function_chat_tool( diff --git a/litellm/responses/litellm_completion_transformation/tool_search.py b/litellm/responses/litellm_completion_transformation/tool_search.py index dd1048eea3e..e17aec939fc 100644 --- a/litellm/responses/litellm_completion_transformation/tool_search.py +++ b/litellm/responses/litellm_completion_transformation/tool_search.py @@ -1,4 +1,4 @@ -from collections.abc import Mapping, Sequence +from collections.abc import Iterable, Iterator, Mapping, Sequence, Set from typing import Final, Literal, TypeAlias from pydantic import ConfigDict, TypeAdapter, ValidationError @@ -7,7 +7,7 @@ from litellm.litellm_core_utils.prompt_templates.factory import is_anthropic_too from litellm.types.llms.anthropic import ANTHROPIC_TOOL_SEARCH_TOOL_TYPES, AnthropicToolSearchToolRegex from litellm.types.llms.base import LiteLLMBaseModel from litellm.types.responses.main import OutputToolSearchCall -from litellm.types.utils import ChatCompletionMessageToolCall, LlmProviders, Message +from litellm.types.utils import LlmProviders, Message from .custom_tools import openai_shaped_tool_call_item_id @@ -16,6 +16,7 @@ __all__ = [ "ANTHROPIC_TOOL_SEARCH_TOOL_TYPE", "CompletionRoute", "ToolSearchCallStatus", + "answers_tool_search", "anthropic_tool_search_chat_tool", "build_tool_search_call", "has_server_tool_search", @@ -24,6 +25,8 @@ __all__ = [ "is_anthropic_tool_search_use", "is_server_tool_search_tool", "provider_builds_anthropic_request", + "tool_search_call_ids", + "without_tool_search_calls", "without_tool_search_uses", ] @@ -64,6 +67,7 @@ def is_anthropic_tool_search_chat_tool(tool: Mapping[str, object]) -> bool: def provider_builds_anthropic_request(model: str, custom_llm_provider: str | None) -> bool: from litellm.litellm_core_utils.get_llm_provider_logic import get_llm_provider from litellm.llms.anthropic.chat.transformation import AnthropicConfig + from litellm.llms.openai.chat.gpt_transformation import OpenAIGPTConfig from litellm.utils import ProviderConfigManager _, provider, _, _ = get_llm_provider(model=model, custom_llm_provider=custom_llm_provider) @@ -72,7 +76,7 @@ def provider_builds_anthropic_request(model: str, custom_llm_provider: str | Non except ValueError: return False chat_config: Final = ProviderConfigManager.get_provider_chat_config(model=model, provider=llm_provider) - return isinstance(chat_config, AnthropicConfig) + return isinstance(chat_config, AnthropicConfig) and not isinstance(chat_config, OpenAIGPTConfig) class CompletionRoute(LiteLLMBaseModel): @@ -82,16 +86,75 @@ class CompletionRoute(LiteLLMBaseModel): custom_llm_provider: str | None = None +class _ToolCallFunctionFields(LiteLLMBaseModel): + model_config = ConfigDict(frozen=True, from_attributes=True) + + name: str | None = None + + +class _ToolCallFields(LiteLLMBaseModel): + model_config = ConfigDict(frozen=True, from_attributes=True) + + id: str | None = None + function: _ToolCallFunctionFields | None = None + + +class _SessionMessageFields(LiteLLMBaseModel): + model_config = ConfigDict(frozen=True, from_attributes=True) + + role: str | None = None + tool_calls: tuple[object, ...] | None = None + tool_call_id: str | None = None + + +def _session_message_fields(message: object) -> _SessionMessageFields: + try: + return _SessionMessageFields.model_validate(message) + except ValidationError: + return _SessionMessageFields() + + +def _tool_search_call_id(tool_call: object) -> str | None: + try: + parsed: Final = _ToolCallFields.model_validate(tool_call) + except ValidationError: + return None + tool_id: Final = parsed.id or "" + tool_name: Final = (parsed.function.name if parsed.function is not None else None) or "" + return tool_id if is_anthropic_tool_search_use(tool_id, tool_name) else None + + def _is_tool_search_call(tool_call: object) -> bool: - return isinstance(tool_call, ChatCompletionMessageToolCall) and is_anthropic_tool_search_use( - tool_call.id or "", tool_call.function.name or "" - ) + return _tool_search_call_id(tool_call) is not None + + +def _assistant_tool_calls(message: object) -> tuple[object, ...]: + fields: Final = _session_message_fields(message) + return (fields.tool_calls or ()) if fields.role == "assistant" else () def holds_tool_search_use(message: object) -> bool: - return isinstance(message, Message) and any( - _is_tool_search_call(tool_call) for tool_call in message.tool_calls or () - ) + return any(_is_tool_search_call(tool_call) for tool_call in _assistant_tool_calls(message)) + + +def _tool_search_call_ids_of(messages: Iterable[object]) -> Iterator[str]: + for message in messages: + for tool_call in _assistant_tool_calls(message): + if (tool_id := _tool_search_call_id(tool_call)) is not None: + yield tool_id + + +def tool_search_call_ids(messages: Iterable[object]) -> frozenset[str]: + return frozenset(_tool_search_call_ids_of(messages)) + + +def answers_tool_search(message: object, search_call_ids: Set[str]) -> bool: + fields: Final = _session_message_fields(message) + return fields.role == "tool" and fields.tool_call_id in search_call_ids + + +def without_tool_search_calls(message: object) -> tuple[object, ...]: + return tuple(tool_call for tool_call in _assistant_tool_calls(message) if not _is_tool_search_call(tool_call)) def without_tool_search_uses(message: Message) -> Message | None: diff --git a/litellm/responses/litellm_completion_transformation/transformation.py b/litellm/responses/litellm_completion_transformation/transformation.py index 7cdcc963191..ff14649ab15 100644 --- a/litellm/responses/litellm_completion_transformation/transformation.py +++ b/litellm/responses/litellm_completion_transformation/transformation.py @@ -120,6 +120,7 @@ from .custom_tools import ( from .reasoning_items import decode_thinking_blocks, encode_thinking_blocks, mint_reasoning_item_id from .tool_search import ( CompletionRoute, + answers_tool_search, anthropic_tool_search_chat_tool, build_tool_search_call, has_server_tool_search, @@ -128,6 +129,8 @@ from .tool_search import ( is_anthropic_tool_search_use, is_server_tool_search_tool, provider_builds_anthropic_request, + tool_search_call_ids, + without_tool_search_calls, without_tool_search_uses, ) @@ -258,8 +261,24 @@ SessionMessage: TypeAlias = ( ) -def _without_stored_tool_search_uses(message: SessionMessage) -> SessionMessage | None: - return without_tool_search_uses(message) if isinstance(message, Message) else message +def _replayed_assistant_without_tool_search_uses(message: Mapping[str, object]) -> SessionMessage | None: + kept_calls: Final = without_tool_search_calls(message) + if not kept_calls and not message.get("content"): + return None + without_calls: Final = {key: value for key, value in message.items() if key != "tool_calls"} + return cast( # cast-ok: the replayed assistant message minus its search calls keeps every other field as stored + SessionMessage, {**without_calls, "tool_calls": list(kept_calls)} if kept_calls else without_calls + ) + + +def _without_stored_tool_search_uses(message: SessionMessage, search_call_ids: frozenset[str]) -> SessionMessage | None: + if isinstance(message, Message): + return without_tool_search_uses(message) + if answers_tool_search(message, search_call_ids): + return None + if isinstance(message, Mapping) and holds_tool_search_use(message): + return _replayed_assistant_without_tool_search_uses(message) + return message class ChatCompletionSession(TypedDict, total=False): @@ -410,6 +429,8 @@ class LiteLLMCompletionResponsesConfig: if not has_server_tool_search(tools): return tools + if any(is_anthropic_tool_search_chat_tool(tool) for tool in tools or ()): + return [tool for tool in tools or () if not is_server_tool_search_tool(tool)] route_runs_tool_search: Final = supports_tool_search( model=model, custom_llm_provider=custom_llm_provider ) and provider_builds_anthropic_request(model, custom_llm_provider) @@ -658,7 +679,12 @@ class LiteLLMCompletionResponsesConfig: ) -> tuple[SessionMessage, ...]: if provider_builds_anthropic_request(route.model, route.custom_llm_provider): return tuple(messages) - return tuple(kept for message in messages if (kept := _without_stored_tool_search_uses(message)) is not None) + search_call_ids: Final = tool_search_call_ids(messages) + return tuple( + kept + for message in messages + if (kept := _without_stored_tool_search_uses(message, search_call_ids)) is not None + ) @staticmethod async def async_responses_api_session_handler( @@ -2130,8 +2156,9 @@ class LiteLLMCompletionResponsesConfig: chat_completion_tool["cache_control"] = tool.get("cache_control") if validated_defer_loading(tool.get("defer_loading")): chat_completion_tool["defer_loading"] = True - if tool.get("allowed_callers"): - chat_completion_tool["allowed_callers"] = tool.get("allowed_callers") + function_allowed_callers: Final = validated_allowed_callers(tool.get("allowed_callers")) + if function_allowed_callers: + chat_completion_tool["allowed_callers"] = function_allowed_callers if tool.get("input_examples"): chat_completion_tool["input_examples"] = tool.get("input_examples") if tool.get("eager_input_streaming") is not None: diff --git a/litellm/types/llms/openai.py b/litellm/types/llms/openai.py index 33986d80df3..493ce762cd6 100644 --- a/litellm/types/llms/openai.py +++ b/litellm/types/llms/openai.py @@ -1426,6 +1426,17 @@ One of: completed, failed, in_progress, cancelled, queued, or incomplete. """ +_NULL_DROPPED_OUTPUT_ITEM_KEYS: Final = frozenset( + ( + ("reasoning", "status"), + ("reasoning", "content"), + ("reasoning", "encrypted_content"), + ("tool_search_call", "created_by"), + ("tool_search_output", "created_by"), + ) +) + + class ResponsesAPIResponse(BaseLiteLLMOpenAIResponseObject): id: str created_at: int @@ -1515,8 +1526,12 @@ class ResponsesAPIResponse(BaseLiteLLMOpenAIResponseObject): return serialized return [ ( - {k: v for k, v in item.items() if v is not None or k not in ("status", "content", "encrypted_content")} - if isinstance(item, dict) and item.get("type") == "reasoning" + { + k: v + for k, v in item.items() + if v is not None or (item.get("type"), k) not in _NULL_DROPPED_OUTPUT_ITEM_KEYS + } + if isinstance(item, dict) else item ) for item in serialized diff --git a/tests/integration/providers/test_responses_bridge_tool_search_chaos.py b/tests/integration/providers/test_responses_bridge_tool_search_chaos.py index 6ca8c9f0f75..2a3b98c3659 100644 --- a/tests/integration/providers/test_responses_bridge_tool_search_chaos.py +++ b/tests/integration/providers/test_responses_bridge_tool_search_chaos.py @@ -1,25 +1,27 @@ import asyncio +import itertools import json import re import socket import threading import uuid -from collections.abc import Iterator, Mapping, Sequence +from collections.abc import Callable, Iterator, Mapping, Sequence from dataclasses import dataclass from pathlib import Path from queue import SimpleQueue -from types import MappingProxyType -from typing import Final, Literal +from types import EllipsisType, MappingProxyType +from typing import Final, Literal, TypeAlias from urllib.parse import unquote, urlsplit import httpx import psutil import pytest import yaml +from _pytest.mark.structures import ParameterSet from integration._support import claude_code as cc from integration._support import responses_vendor as rv -from integration._support.anthropic_thinking import EVENT_STREAM, Event, aws_chunks, sse_chunks -from integration._support.client import Gateway, eventually, gateway_from_environment +from integration._support.anthropic_thinking import BEDROCK_MODEL, EVENT_STREAM, Event, aws_chunks, sse_chunks +from integration._support.client import Gateway, Scenario, eventually, gateway_from_environment from integration._support.database import read_rows from integration._support.process import OwnedProxy, graceful_stop_seconds, owned_proxy_process from integration._support.wire import Reply, Request, Wire, wire_server @@ -27,6 +29,7 @@ from pydantic import JsonValue _CLAUDE: Final = cc.OPUS _BEDROCK_CLAUDE: Final = "anthropic.claude-sonnet-5-5" +_MISTRAL: Final = "mistral-large-latest" _CONFIG_MODEL: Final = "responses-bridge-tool-search-chaos" _ANTHROPIC_BASE: Final = "http://api.anthropic.com" _REGEX_TOOL_TYPE: Final = "tool_search_tool_regex_20251119" @@ -43,7 +46,15 @@ _WEATHER_PARAMETERS: Final[dict[str, JsonValue]] = { "required": ["city"], } +_ORDER_PARAMETERS: Final[dict[str, JsonValue]] = { + "type": "object", + "properties": {"order_id": {"type": "string"}}, + "required": ["order_id"], +} + Endpoint = Literal["responses", "chat", "messages"] +StoredShape: TypeAlias = Literal["tool_use", "search_only", "search_and_text"] +OffAnthropicRoute: TypeAlias = Literal["mistral", "bedrock_converse"] @dataclass(frozen=True, slots=True) @@ -635,3 +646,365 @@ def test_bedrock_invoke_pass_through_relays_the_search_frames_untouched(owned_ri seconds=70, ) assert str(rows[0]["status"]) == "success", rows + + +def _server_tool_use(marker: str) -> Event: + return { + "type": "server_tool_use", + "id": f"srvtoolu_{marker}", + "name": _REGEX_TOOL_NAME, + "input": json.loads(_SEARCH_ARGUMENTS), + } + + +def _stored_turn(marker: str, shape: StoredShape) -> dict[str, JsonValue]: + match shape: + case "tool_use": + return _message(marker) + case "search_only": + return { + **_message(marker), + "content": [_server_tool_use(marker), _search_result(marker)], + "stop_reason": "pause_turn", + } + case "search_and_text": + return { + **_message(marker), + "content": [ + _server_tool_use(marker), + _search_result(marker), + {"type": "text", "text": f"found marker-{marker}"}, + ], + "stop_reason": "end_turn", + } + + +def _stored_turn_vendor(shape: StoredShape) -> Callable[[Request], Reply]: + def respond(request: Request) -> Reply: + if request.method != "POST": + return Reply() + marker: Final = rv.newest_marker(request.body.decode()) or "health" + return Reply(body=json.dumps(_stored_turn(marker, shape)).encode()) + + return respond + + +def _tool_call_ids(message: Mapping[str, JsonValue]) -> frozenset[str]: + return frozenset(str(call["id"]) for call in rv.ITEMS.validate_python(message.get("tool_calls") or [])) + + +def _answer_ids_after(messages: Sequence[Mapping[str, JsonValue]], index: int) -> frozenset[str]: + answers: Final = itertools.takewhile(lambda message: message.get("role") == "tool", messages[index + 1 :]) + return frozenset(str(message.get("tool_call_id")) for message in answers) + + +def _opening_call_ids(messages: Sequence[Mapping[str, JsonValue]], index: int) -> frozenset[str]: + opener: Final = next((message for message in reversed(messages[:index]) if message.get("role") != "tool"), None) + return frozenset() if opener is None else _tool_call_ids(opener) + + +def _unmatched_tool_ids(messages: Sequence[Mapping[str, JsonValue]]) -> Iterator[str]: + for index, message in enumerate(messages): + if message.get("role") == "assistant": + yield from sorted(_tool_call_ids(message) - _answer_ids_after(messages, index)) + if message.get("role") == "tool" and str(message.get("tool_call_id")) not in _opening_call_ids(messages, index): + yield str(message.get("tool_call_id")) + + +def _chat_answer(marker: str, stream: bool) -> Reply: + head: Final[Mapping[str, JsonValue]] = {"id": f"cmpl-{marker}", "created": 1, "model": _MISTRAL} + if not stream: + message: Final = {"role": "assistant", "content": f"answered marker-{marker}"} + completion: Final = { + **head, + "object": "chat.completion", + "choices": [{"index": 0, "message": message, "finish_reason": "stop"}], + "usage": rv.CHAT_USAGE, + } + return Reply(body=json.dumps(completion).encode()) + delta: Final[dict[str, JsonValue]] = {"role": "assistant", "content": f"answered marker-{marker}"} + opened: Final[dict[str, JsonValue]] = {"index": 0, "delta": delta, "finish_reason": None} + return Reply( + content_type="text/event-stream", + chunks=( + rv.chat_sse({**head, "object": "chat.completion.chunk", "choices": [opened]}), + rv.chat_sse( + { + **head, + "object": "chat.completion.chunk", + "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}], + "usage": rv.CHAT_USAGE, + } + ), + b"data: [DONE]\n\n", + ), + ) + + +def _strict_mistral(request: Request) -> Reply: + if request.method != "POST": + return Reply() + marker: Final = rv.newest_marker(request.body.decode()) or "health" + if request.target != "/v1/chat/completions": + return rv.error(404, f"unexpected target {request.target}", "not_found") + if "defer_loading" in request.body.decode(): + return rv.error(422, f"tools: defer_loading: Extra inputs are not permitted marker-{marker}", "extra_forbidden") + body: Final = rv.JSON_OBJECT.validate_json(request.body) + unmatched: Final = tuple(_unmatched_tool_ids(rv.ITEMS.validate_python(body.get("messages") or []))) + if unmatched: + return rv.error( + 400, + f"Not the same number of function calls and responses: {', '.join(unmatched)} marker-{marker}", + "invalid_request_message_order", + ) + return _chat_answer(marker, body.get("stream") is True) + + +def _converse_ids(message: Mapping[str, JsonValue], block_key: str) -> frozenset[str]: + return frozenset( + str(rv.JSON_OBJECT.validate_python(block[block_key])["toolUseId"]) + for block in rv.ITEMS.validate_python(message.get("content") or []) + if block_key in block + ) + + +def _unpaired_converse_turns(messages: Sequence[Mapping[str, JsonValue]]) -> tuple[int, ...]: + return tuple( + index + 1 + for index, (previous, current) in enumerate(zip(messages, messages[1:])) + if _converse_ids(previous, "toolUse") != _converse_ids(current, "toolResult") + ) + + +def _strict_converse(request: Request) -> Reply: + if request.method != "POST": + return Reply() + marker: Final = rv.newest_marker(request.body.decode()) or "health" + if unquote(request.target) != f"/model/{BEDROCK_MODEL}/converse": + return rv.error(404, f"unexpected target {request.target}", "not_found") + body: Final = rv.JSON_OBJECT.validate_json(request.body) + unpaired: Final = _unpaired_converse_turns(rv.ITEMS.validate_python(body.get("messages") or [])) + if unpaired: + refusal: Final = ( + f"Expected toolResult blocks at messages.{unpaired[0]}.content for the toolUse ids of the previous turn " + f"marker-{marker}" + ) + return Reply(status=400, body=json.dumps({"message": refusal}).encode()) + answer: Final = { + "output": {"message": {"role": "assistant", "content": [{"text": f"answered marker-{marker}"}]}}, + "stopReason": "end_turn", + "usage": {"inputTokens": 11, "outputTokens": 4, "totalTokens": 15}, + "metrics": {"latencyMs": 1}, + } + return Reply(body=json.dumps(answer).encode()) + + +def _strict_vendor(route: OffAnthropicRoute) -> Callable[[Request], Reply]: + match route: + case "mistral": + return _strict_mistral + case "bedrock_converse": + return _strict_converse + + +def _off_anthropic_deployment(scenario: Scenario, route: OffAnthropicRoute, wire_url: str) -> str: + match route: + case "mistral": + return scenario.model(model=f"mistral/{_MISTRAL}", api_base=wire_url, api_key="synthetic-mistral-key") + case "bedrock_converse": + return scenario.model( + model=f"bedrock/converse/{BEDROCK_MODEL}", + aws_access_key_id="AKIASCRIPTEDPROVIDER", + aws_secret_access_key="scripted-secret", + aws_region_name="us-east-1", + aws_bedrock_runtime_endpoint=wire_url, + ) + + +def _stored_first_turn(proxy: Gateway, model: str, marker: str) -> str: + first: Final = proxy.request("POST", "/v1/responses", _body(model, _Call("responses", False, marker))) + assert first.status_code == 200, first.text + payload: Final = rv.JSON_OBJECT.validate_json(first.text) + types: Final = [item["type"] for item in rv.ITEMS.validate_python(payload["output"])] + assert "tool_search_call" in types, types + response_id: Final = str(payload["id"]) + eventually( + lambda: read_rows('SELECT request_id FROM "LiteLLM_SpendLogs" WHERE model_group=%s', (model,)), + lambda rows: any(rv.same_response(str(row["request_id"]), response_id) for row in rows), + seconds=70, + ) + return response_id + + +def _follow_up_input(marker: str, shape: StoredShape) -> list[JsonValue]: + if shape == "tool_use": + return [{"type": "function_call_output", "call_id": f"toolu_{marker}", "output": f"22C marker-{marker}"}] + return [{"role": "user", "content": f"continue marker-{marker}"}] + + +def _chat_turn(message: Mapping[str, JsonValue]) -> tuple[str, tuple[str, ...]]: + if message.get("role") == "tool": + return "tool", (str(message.get("tool_call_id")),) + return str(message["role"]), tuple(sorted(_tool_call_ids(message))) + + +def _expected_chat_turns(marker: str, shape: StoredShape) -> tuple[tuple[str, tuple[str, ...]], ...]: + match shape: + case "tool_use": + return (("user", ()), ("assistant", (f"toolu_{marker}",)), ("tool", (f"toolu_{marker}",))) + case "search_only": + return (("user", ()), ("user", ())) + case "search_and_text": + return (("user", ()), ("assistant", ()), ("user", ())) + + +def _assert_converse_follow_up(messages: Sequence[Mapping[str, JsonValue]], marker: str) -> None: + assert [message["role"] for message in messages] == ["user", "assistant", "user"], messages + assert _converse_ids(messages[1], "toolUse") == {f"toolu_{marker}"}, messages + assert _converse_ids(messages[2], "toolResult") == {f"toolu_{marker}"}, messages + + +def _off_anthropic_cells() -> tuple[ParameterSet, ...]: + cases: Final[tuple[tuple[OffAnthropicRoute, StoredShape, bool], ...]] = ( + ("mistral", "search_only", False), + ("mistral", "search_and_text", False), + ("mistral", "search_and_text", True), + ("bedrock_converse", "tool_use", False), + ) + return tuple( + pytest.param(route, shape, stream, id=f"{route}-{shape}-{'stream' if stream else 'sync'}") + for route, shape, stream in cases + ) + + +@pytest.mark.timeout(_OWNED_PROXY_CELL_SECONDS) +@pytest.mark.parametrize(("route", "shape", "stream"), _off_anthropic_cells()) +def test_a_follow_up_off_anthropic_drops_the_stored_search_call_and_is_served( + owned_rig: _OwnedRig, route: OffAnthropicRoute, shape: StoredShape, stream: bool +) -> None: + marker: Final = uuid.uuid4().hex + proxy: Final = owned_rig.proxy.gateway + with ( + wire_server(_stored_turn_vendor(shape)) as claude_wire, + wire_server(_strict_vendor(route)) as follow_up_wire, + proxy.scenario() as scenario, + ): + claude: Final = scenario.model( + model=f"anthropic/{_CLAUDE}", api_base=claude_wire.url, api_key=cc.ANTHROPIC_API_KEY + ) + target: Final = _off_anthropic_deployment(scenario, route, follow_up_wire.url) + response_id: Final = _stored_first_turn(proxy, claude, marker) + follow_up: Final = proxy.request( + "POST", + "/v1/responses", + { + **_body(target, _Call("responses", stream, marker)), + "previous_response_id": response_id, + "input": _follow_up_input(marker, shape), + }, + ) + assert follow_up.status_code == 200, follow_up.text + assert f"answered marker-{marker}" in follow_up.text, follow_up.text + (request,) = _marked_posts(follow_up_wire.drain(), marker) + sent: Final = request.body.decode() + assert "srvtoolu_" not in sent and "tool_search" not in sent and "defer_loading" not in sent, sent + body: Final = rv.JSON_OBJECT.validate_json(request.body) + messages: Final = rv.ITEMS.validate_python(body["messages"]) + if route == "bedrock_converse": + _assert_converse_follow_up(messages, marker) + return + assert tuple(_chat_turn(message) for message in messages) == _expected_chat_turns(marker, shape), messages + tools: Final = rv.ITEMS.validate_python(body["tools"]) + assert [rv.JSON_OBJECT.validate_python(tool["function"])["name"] for tool in tools] == ["get_weather"], tools + if shape == "search_and_text": + assert f"found marker-{marker}" in json.dumps(messages[1]), messages + + +@pytest.mark.timeout(_OWNED_PROXY_CELL_SECONDS) +def test_a_follow_up_on_anthropic_that_answers_the_stored_search_call_replays_it_with_its_answer( + owned_rig: _OwnedRig, +) -> None: + marker: Final = uuid.uuid4().hex + proxy: Final = owned_rig.proxy.gateway + with wire_server(_stored_turn_vendor("tool_use")) as claude_wire, proxy.scenario() as scenario: + claude: Final = scenario.model( + model=f"anthropic/{_CLAUDE}", api_base=claude_wire.url, api_key=cc.ANTHROPIC_API_KEY + ) + response_id: Final = _stored_first_turn(proxy, claude, marker) + answers: Final[list[JsonValue]] = [ + {"type": "function_call_output", "call_id": f"srvtoolu_{marker}", "output": f"get_weather marker-{marker}"}, + {"type": "function_call_output", "call_id": f"toolu_{marker}", "output": f"22C marker-{marker}"}, + ] + follow_up: Final = proxy.request( + "POST", + "/v1/responses", + {**_body(claude, _Call("responses", False, marker)), "previous_response_id": response_id, "input": answers}, + ) + assert follow_up.status_code == 200, follow_up.text + opening, replay = _marked_posts(claude_wire.drain(), marker) + _assert_forwarded_with_tool_search(opening) + _assert_forwarded_with_tool_search(replay) + messages: Final = rv.ITEMS.validate_python(rv.JSON_OBJECT.validate_json(replay.body)["messages"]) + stored_turn: Final = next(turn for turn in messages if turn["role"] == "assistant") + blocks: Final = rv.ITEMS.validate_python(stored_turn["content"]) + assert [(block.get("id"), block.get("name")) for block in blocks if block["type"] != "text"] == [ + (f"srvtoolu_{marker}", _REGEX_TOOL_NAME), + (f"toolu_{marker}", "get_weather"), + ], blocks + assert sorted(str(block["tool_use_id"]) for block in _tool_results(messages)) == [ + f"srvtoolu_{marker}", + f"toolu_{marker}", + ], messages + + +@pytest.fixture(scope="module") +def drop_params_proxy(tmp_path_factory: pytest.TempPathFactory) -> Iterator[OwnedProxy]: + directory: Final = tmp_path_factory.mktemp("tool-search-drop-params") + config: Final = rv.JSON_OBJECT.validate_python( + yaml.safe_load(Path("tests/integration/proxy_config.yaml").read_text()) + ) + settings: Final = rv.JSON_OBJECT.validate_python(config["litellm_settings"]) + path: Final = directory / "responses-bridge-tool-search-drop-params.yaml" + path.write_text(yaml.safe_dump({**config, "litellm_settings": {**settings, "drop_params": True}})) + with ( + gateway_from_environment() as gateway, + owned_proxy_process(gateway, directory, {}, config=path, workers=2) as owned, + ): + yield owned + + +@pytest.mark.timeout(_OWNED_PROXY_CELL_SECONDS) +@pytest.mark.parametrize("request_drop_params", [..., False], ids=["absent", "false"]) +def test_a_global_drop_params_drops_a_malformed_defer_loading_and_serves_the_search_call( + drop_params_proxy: OwnedProxy, request_drop_params: bool | EllipsisType +) -> None: + marker: Final = uuid.uuid4().hex + proxy: Final = drop_params_proxy.gateway + with wire_server(_Vendor().respond) as wire, proxy.scenario() as scenario: + model: Final = scenario.model(model=f"anthropic/{_CLAUDE}", api_base=wire.url, api_key=cc.ANTHROPIC_API_KEY) + opening: Final = _body(model, _Call("responses", False, marker)) + lookup: Final[dict[str, JsonValue]] = { + "type": "function", + "name": "lookup_order", + "description": "Look up an order", + "parameters": _ORDER_PARAMETERS, + "defer_loading": "true", + } + flag: Final[Mapping[str, JsonValue]] = ( + {} if isinstance(request_drop_params, EllipsisType) else {"drop_params": request_drop_params} + ) + response: Final = proxy.request( + "POST", + "/v1/responses", + {**opening, "tools": [*rv.ITEMS.validate_python(opening["tools"]), lookup], **flag}, + ) + _assert_carries_the_search_call( + _Served(_Call("responses", False, marker), response.status_code, response.text, "") + ) + (request,) = _marked_posts(wire.drain(), marker) + tools: Final = rv.ITEMS.validate_python(rv.JSON_OBJECT.validate_json(request.body)["tools"]) + assert tools[0] == dict(_REGEX_TOOL), tools + assert [(tool["name"], tool.get("defer_loading")) for tool in tools[1:]] == [ + ("get_weather", True), + ("lookup_order", None), + ], tools diff --git a/tests/integration/providers/test_responses_bridge_tool_search_undeferred_wire.py b/tests/integration/providers/test_responses_bridge_tool_search_undeferred_wire.py index af8765f4b7f..c43194100cc 100644 --- a/tests/integration/providers/test_responses_bridge_tool_search_undeferred_wire.py +++ b/tests/integration/providers/test_responses_bridge_tool_search_undeferred_wire.py @@ -92,6 +92,10 @@ def _loaded_tool() -> dict[str, JsonValue]: return {"type": "function", "name": "send_email", "description": "Send an email", "parameters": _EMAIL_PARAMETERS} +def _custom_tool() -> dict[str, JsonValue]: + return {"type": "custom", "name": "run_sql", "description": "Run a SQL query", "defer_loading": True} + + def _namespace_tool() -> dict[str, JsonValue]: return { "type": "namespace", @@ -480,6 +484,32 @@ def test_a_native_anthropic_search_tool_reaches_the_wire_untouched_and_keeps_eve ], sent +@pytest.mark.parametrize("stream", [False, True], ids=["sync", "stream"]) +@pytest.mark.parametrize("route", ["anthropic", "mistral"]) +def test_a_deferred_custom_tool_keeps_its_deferral_only_while_a_search_tool_goes_out( + gateway: Gateway, route: Route, stream: bool +) -> None: + marker: Final = uuid.uuid4().hex + with wire_server(_vendor(route)) as wire, gateway.scenario() as scenario: + model: Final = _deployment(scenario, route, wire.url) + tools: Final = (_tool_search(), _custom_tool(), _loaded_tool()) + response: Final = _raw(gateway, _responses_body(model, marker, tools, stream)) + _assert_served(response, marker, stream) + request, body = _only_request(wire) + _assert_target(route, stream, request, body) + if route == "mistral": + assert _wire_tool_names(route, body) == ["run_sql", "send_email"], body + assert "defer_loading" not in request.body.decode(), body + assert "tool_search" not in request.body.decode(), body + return + sent: Final = _ITEMS.validate_python(body["tools"]) + assert sent[0] == {"type": _REGEX_TOOL_TYPE, "name": _REGEX_TOOL_NAME}, sent + assert [(tool["name"], tool.get("defer_loading")) for tool in sent[1:]] == [ + ("run_sql", True), + ("send_email", None), + ], sent + + def test_bedrock_converse_skips_a_native_search_tool_and_serves_the_deferred_tools(gateway: Gateway) -> None: call: Final = _Call("bedrock_converse", False, uuid.uuid4().hex) with wire_server(_converse) as wire, gateway.scenario() as scenario: @@ -512,6 +542,108 @@ def test_a_non_boolean_defer_loading_is_refused_with_400_on_a_route_without_tool _assert_undeferred_wire(follow_up, request, body) +def _text_parts(marker: str) -> tuple[str, ...]: + return ("part one ", "part two ", f"marker-{marker}") + + +def _split_claude_events(marker: str) -> tuple[Event, ...]: + opened: Final = {**_claude_message(marker), "content": [], "stop_reason": None} + deltas: Final[tuple[Event, ...]] = tuple( + {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": part}} + for part in _text_parts(marker) + ) + return ( + {"type": "message_start", "message": opened}, + {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}}, + *deltas, + {"type": "content_block_stop", "index": 0}, + { + "type": "message_delta", + "delta": {"stop_reason": "end_turn", "stop_sequence": None}, + "usage": {"output_tokens": 7}, + }, + {"type": "message_stop"}, + ) + + +def _split_chat_chunks(marker: str) -> tuple[bytes, ...]: + head: Final[Mapping[str, JsonValue]] = { + "id": f"cmpl-{marker}", + "object": "chat.completion.chunk", + "created": 1, + "model": _MISTRAL, + } + parts: Final = tuple( + rv.chat_sse( + { + **head, + "choices": [{"index": 0, "delta": {"role": "assistant", "content": part}, "finish_reason": None}], + } + ) + for part in _text_parts(marker) + ) + return ( + *parts, + rv.chat_sse( + { + **head, + "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}], + "usage": {"prompt_tokens": 5, "completion_tokens": 3, "total_tokens": 8}, + } + ), + b"data: [DONE]\n\n", + ) + + +def _split_vendor(route: Route) -> Callable[[Request], Reply]: + def respond(request: Request) -> Reply: + if request.method != "POST": + return Reply() + marker: Final = rv.newest_marker(request.body.decode()) or "health" + if route == "mistral": + return Reply(content_type="text/event-stream", chunks=_split_chat_chunks(marker)) + return Reply(content_type="text/event-stream", chunks=sse_chunks(_split_claude_events(marker))) + + return respond + + +def _frames(text: str) -> tuple[dict[str, JsonValue], ...]: + return tuple(_JSON_OBJECT.validate_json(line[6:]) for line in text.splitlines() if line.startswith("data: {")) + + +def _message_texts(items: Sequence[Mapping[str, JsonValue]]) -> tuple[str, ...]: + return tuple( + "".join(str(part["text"]) for part in _ITEMS.validate_python(item["content"])) + for item in items + if item["type"] == "message" + ) + + +@pytest.mark.parametrize("route", ["anthropic", "mistral"]) +def test_a_streamed_answer_closes_its_message_item_with_every_streamed_delta(gateway: Gateway, route: Route) -> None: + marker: Final = uuid.uuid4().hex + with wire_server(_split_vendor(route)) as wire, gateway.scenario() as scenario: + model: Final = _deployment(scenario, route, wire.url) + tools: Final = (_tool_search(), _function_tool(), _loaded_tool()) + response: Final = _raw(gateway, _responses_body(model, marker, tools, True)) + assert response.status_code == 200, response.text + frames: Final = _frames(response.text) + expected: Final = "".join(_text_parts(marker)) + deltas: Final = [str(frame["delta"]) for frame in frames if frame.get("type") == "response.output_text.delta"] + assert "".join(deltas) == expected, frames + done: Final = [ + _JSON_OBJECT.validate_python(frame["item"]) + for frame in frames + if frame.get("type") == "response.output_item.done" + ] + assert _message_texts(done) == (expected,), done + (completed,) = [frame for frame in frames if frame.get("type") == "response.completed"] + output: Final = _ITEMS.validate_python(_JSON_OBJECT.validate_python(completed["response"])["output"]) + assert _message_texts(output) == (expected,), output + request, body = _only_request(wire) + _assert_target(route, True, request, body) + + async def _send(client: httpx.AsyncClient, key: str, model: str, call: _Call) -> _Served: body: Final = _responses_body(model, call.marker, (_function_tool(), _namespace_tool()), call.stream) async with client.stream( diff --git a/tests/integration/providers/test_responses_bridge_tool_search_wire.py b/tests/integration/providers/test_responses_bridge_tool_search_wire.py index 054c2e136b9..dbae068e861 100644 --- a/tests/integration/providers/test_responses_bridge_tool_search_wire.py +++ b/tests/integration/providers/test_responses_bridge_tool_search_wire.py @@ -1,26 +1,29 @@ import asyncio +import itertools import json import os import uuid from collections.abc import Iterator, Mapping, Sequence from dataclasses import dataclass from types import EllipsisType, MappingProxyType -from typing import Final, Literal +from typing import Final, Literal, TypeAlias from urllib.parse import unquote import anthropic import httpx import openai import pytest +from _pytest.mark.structures import ParameterSet from anthropic.types.beta import BetaToolUnionParam from integration._support import claude_code as cc from integration._support import responses_vendor as rv from integration._support.anthropic_thinking import BEDROCK_MODEL, EVENT_STREAM, MODEL, Event, aws_chunks, sse_chunks -from integration._support.client import Gateway, Scenario, eventually +from integration._support.client import Gateway, Scenario, eventually, object_value from integration._support.database import read_rows from integration._support.vertex import service_account_json from integration._support.wire import Reply, Request, Wire, wire_server from integration.providers.test_bedrock_auth_wire import RESPONSE as CONVERSE_RESPONSE +from openai.types.responses import ResponseToolSearchCall from pydantic import JsonValue, TypeAdapter from redis import Redis @@ -65,8 +68,26 @@ _GEMINI_REPLY: Final[dict[str, JsonValue]] = { "modelVersion": _GEMINI, } +_ORDER_PARAMETERS: Final[dict[str, JsonValue]] = { + "type": "object", + "properties": {"order_id": {"type": "string"}}, + "required": ["order_id"], +} +_MALFORMED_FLAGS: Final[Mapping[str, JsonValue]] = MappingProxyType( + {"defer_loading": "true", "allowed_callers": "direct"} +) +_WELL_FORMED_FLAGS: Final[Mapping[str, JsonValue]] = MappingProxyType( + {"defer_loading": True, "allowed_callers": ["direct"]} +) +_FLAG_REFUSALS: Final[Mapping[str, str]] = MappingProxyType( + {"defer_loading": "defer_loading must be a boolean", "allowed_callers": "allowed_callers must be a list of strings"} +) + _Client = Literal["sdk", "async_sdk", "raw"] Provider = Literal["anthropic", "bedrock_invoke", "vertex_ai"] +FlaggedKind: TypeAlias = Literal["function", "custom", "namespace_member"] +ToolFlag: TypeAlias = Literal["defer_loading", "allowed_callers"] +DropSource: TypeAlias = Literal["off", "request", "deployment", "null", "request_string"] @dataclass(frozen=True, slots=True) @@ -606,6 +627,35 @@ def test_a_non_streaming_tool_search_call_item_carries_only_its_own_keys(gateway _assert_claude_wire(provider, False, request, body, _DEFERRED_NAMES) +@pytest.mark.parametrize("provider", ["anthropic", "bedrock_invoke"]) +def test_a_streamed_tool_search_call_item_carries_only_its_own_keys(gateway: Gateway, provider: Provider) -> None: + marker: Final = uuid.uuid4().hex + with wire_server(_Vendor().respond) as wire, gateway.scenario() as scenario: + model: Final = _deployment(scenario, provider, wire.url, gateway.upstream_url) + tools: Final = (_tool_search(), _function_tool(), _namespace_tool()) + answer: Final = _raw_answer(gateway, _responses_body(model, marker, tools, True)) + _assert_search_call_output(answer.payload, marker) + _assert_search_call_events(answer.events, marker, "completed") + streamed: Final = [_item_of(event) for event in _item_events(answer.events, "tool_search_call")] + (completed,) = [item for item in _tool_items(answer.payload) if item["type"] == "tool_search_call"] + assert [frozenset(item) for item in (*streamed, completed)] == [_TOOL_SEARCH_CALL_KEYS] * 3, streamed + request, body = _only_request(wire) + _assert_claude_wire(provider, True, request, body, _DEFERRED_NAMES) + + +def _schema_properties(schemas: Mapping[str, JsonValue], name: str) -> frozenset[str]: + return frozenset(object_value(object_value(schemas[name])["properties"])) + + +def test_the_openapi_schema_keeps_the_sdk_tool_search_call_beside_the_proxy_output_item(gateway: Gateway) -> None: + schemas: Final = object_value(object_value(gateway.get("/openapi.json")["components"])["schemas"]) + proxy_item: Final = _schema_properties(schemas, "OutputToolSearchCall") + sdk_item: Final = _schema_properties(schemas, "ResponseToolSearchCall") + assert proxy_item == _TOOL_SEARCH_CALL_KEYS, schemas.get("OutputToolSearchCall") + assert sdk_item == frozenset(ResponseToolSearchCall.model_fields), schemas.get("ResponseToolSearchCall") + assert [name for name in schemas if name.endswith("ToolSearchCall") and "__" in name] == [], sorted(schemas) + + def test_search_arguments_that_never_parse_stream_an_incomplete_tool_search_call(gateway: Gateway) -> None: marker: Final = uuid.uuid4().hex with wire_server(_Vendor(search_arguments=_TRUNCATED_ARGUMENTS).respond) as wire, gateway.scenario() as scenario: @@ -700,6 +750,24 @@ def test_a_cache_hit_still_carries_the_tool_search_call(gateway: Gateway) -> Non assert sorted(str(row["cache_hit"]) == "True" for row in rows) == [False, True], rows +def test_a_cache_hit_tool_search_call_item_carries_only_its_own_keys(gateway: Gateway) -> None: + marker: Final = uuid.uuid4().hex + with wire_server(_Vendor().respond) as wire, gateway.scenario() as scenario: + model: Final = _deployment(scenario, "anthropic", wire.url, gateway.upstream_url) + body: Final = _responses_body(model, marker, (_tool_search(), _function_tool()), False) + first: Final = _raw(gateway, "/v1/responses", body) + assert first.status_code == 200, first.text + assert len(wire.drain()) == 1 + _await_cached(f"srvtoolu_{marker}") + hit: Final = _raw(gateway, "/v1/responses", body) + assert hit.status_code == 200, hit.text + assert wire.drain() == (), hit.text + payload: Final = _JSON_OBJECT.validate_json(hit.content) + _assert_search_call_output(payload, marker) + (search,) = [item for item in _tool_items(payload) if item["type"] == "tool_search_call"] + assert frozenset(search) == _TOOL_SEARCH_CALL_KEYS, search + + def test_model_info_reports_tool_search_support_for_the_claude_deployment(gateway: Gateway) -> None: with wire_server(_Vendor().respond) as wire, gateway.scenario() as scenario: model: Final = _deployment(scenario, "anthropic", wire.url, gateway.upstream_url) @@ -964,6 +1032,114 @@ def test_a_malformed_allowed_callers_on_a_namespace_member_is_refused_with_400( assert wire.drain() == (), refused.text +def _tool_flags(flag: ToolFlag) -> Mapping[str, JsonValue]: + return {key: _MALFORMED_FLAGS[key] if key == flag else value for key, value in _WELL_FORMED_FLAGS.items()} + + +def _flagged_tool(kind: FlaggedKind, flag: ToolFlag) -> dict[str, JsonValue]: + match kind: + case "function": + return { + "type": "function", + "name": "lookup_order", + "description": "Look up an order", + "parameters": _ORDER_PARAMETERS, + **_tool_flags(flag), + } + case "custom": + return {"type": "custom", "name": "run_sql", "description": "Run a SQL query", **_tool_flags(flag)} + case "namespace_member": + return _namespace_tool(**_tool_flags(flag)) + + +def _flagged_wire_name(kind: FlaggedKind) -> str: + match kind: + case "function": + return "lookup_order" + case "custom": + return "run_sql" + case "namespace_member": + return "finance__get_stock_price" + + +def _request_drop_params(source: DropSource) -> Mapping[str, JsonValue]: + match source: + case "off" | "deployment": + return {} + case "request": + return {"drop_params": True} + case "null": + return {"drop_params": None} + case "request_string": + return {"drop_params": "true"} + + +def _drop_params_deployment(scenario: Scenario, source: DropSource, wire_url: str) -> str: + if source == "deployment": + return scenario.model( + model=f"anthropic/{_CLAUDE}", api_base=wire_url, api_key=cc.ANTHROPIC_API_KEY, drop_params=True + ) + return scenario.model(model=f"anthropic/{_CLAUDE}", api_base=wire_url, api_key=cc.ANTHROPIC_API_KEY) + + +def _drop_params_cells() -> tuple[ParameterSet, ...]: + kinds: Final[tuple[FlaggedKind, ...]] = ("function", "custom", "namespace_member") + flags: Final[tuple[ToolFlag, ...]] = ("defer_loading", "allowed_callers") + sources: Final[tuple[DropSource, ...]] = ("off", "request", "deployment") + grid: Final = tuple( + pytest.param(kind, flag, source, id=f"{kind}-{flag}-{source}") + for kind, flag, source in itertools.product(kinds, flags, sources) + ) + edges: Final = tuple( + pytest.param("function", "defer_loading", source, id=f"function-defer_loading-{source}") + for source in ("null", "request_string") + ) + return (*grid, *edges) + + +def _assert_flag_dropped_on_wire(body: Mapping[str, JsonValue], kind: FlaggedKind, flag: ToolFlag) -> None: + tools: Final = _ITEMS.validate_python(body["tools"]) + assert tools[0] == dict(_REGEX_TOOL), tools + (flagged,) = [tool for tool in tools if tool.get("name") == _flagged_wire_name(kind)] + assert flag not in flagged, flagged + kept: Final = "allowed_callers" if flag == "defer_loading" else "defer_loading" + assert flagged[kept] == _WELL_FORMED_FLAGS[kept], flagged + (weather,) = [tool for tool in tools if tool.get("name") == "get_weather"] + assert weather.get("defer_loading") is True, weather + + +@pytest.mark.parametrize(("kind", "flag", "source"), _drop_params_cells()) +def test_drop_params_drops_a_malformed_tool_flag_and_without_it_the_request_is_refused_with_400( + gateway: Gateway, kind: FlaggedKind, flag: ToolFlag, source: DropSource +) -> None: + marker: Final = uuid.uuid4().hex + with wire_server(_Vendor().respond) as wire, gateway.scenario() as scenario: + model: Final = _drop_params_deployment(scenario, source, wire.url) + tools: Final = (_tool_search(), _function_tool(), _flagged_tool(kind, flag)) + response: Final = _raw( + gateway, + "/v1/responses", + {**_responses_body(model, marker, tools, False), **_CACHE_BUST, **_request_drop_params(source)}, + ) + if source in ("off", "null"): + assert response.status_code == 400, response.text + assert "litellm.BadRequestError" in response.text, response.text + assert _FLAG_REFUSALS[flag] in response.text, response.text + assert wire.drain() == (), response.text + follow_up: Final = uuid.uuid4().hex + served: Final = _raw_answer( + gateway, _responses_body(model, follow_up, (_tool_search(), _function_tool()), False) + ) + _assert_search_call_output(served.payload, follow_up) + assert len(wire.drain()) == 1 + return + assert response.status_code == 200, response.text + _assert_search_call_output(_JSON_OBJECT.validate_json(response.content), marker) + request, body = _only_request(wire) + assert request.target == "/v1/messages", request.target + _assert_flag_dropped_on_wire(body, kind, flag) + + def test_duplicate_tool_search_tools_reach_claude_and_its_refusal_reaches_the_caller(gateway: Gateway) -> None: marker: Final = uuid.uuid4().hex with wire_server(_Vendor().respond) as wire, gateway.scenario() as scenario: diff --git a/tests/unit/litellm_core_utils/prompt_templates/test_litellm_core_utils_prompt_templates_factory.py b/tests/unit/litellm_core_utils/prompt_templates/test_litellm_core_utils_prompt_templates_factory.py index 90dbad52036..14fea46516e 100644 --- a/tests/unit/litellm_core_utils/prompt_templates/test_litellm_core_utils_prompt_templates_factory.py +++ b/tests/unit/litellm_core_utils/prompt_templates/test_litellm_core_utils_prompt_templates_factory.py @@ -3,6 +3,7 @@ import json import logging import os import re +import time from typing import Final, List from unittest.mock import MagicMock, patch @@ -7176,3 +7177,24 @@ def test_anthropic_messages_pt_keeps_a_replayed_tool_search_call_the_caller_answ "srvtoolu_01Search", "toolu_01Weather", ] + + +def _merged_assistant_run_cost(run_length: int) -> float: + messages: Final = [ + {"role": "user", "content": "Plan the trip"}, + *({"role": "assistant", "content": "Working on it", "tool_calls": []} for _ in range(run_length)), + {"role": "user", "content": "Go on"}, + ] + + def once() -> float: + started: Final = time.perf_counter() + anthropic_messages_pt(messages=messages, model="claude-sonnet-5-5", llm_provider="anthropic") + return time.perf_counter() - started + + return min(once() for _ in range(3)) + + +def test_anthropic_messages_pt_merges_a_long_assistant_run_in_linear_time(): + growth: Final = _merged_assistant_run_cost(4000) / _merged_assistant_run_cost(1000) + + assert growth < 8.0, f"cost grew {growth:.1f}x for 4x the assistant run" diff --git a/tests/unit/llms/openai/responses/test_openai_responses_guardrail_handler.py b/tests/unit/llms/openai/responses/test_openai_responses_guardrail_handler.py index 8bd242308f7..d7d4c1b3aae 100644 --- a/tests/unit/llms/openai/responses/test_openai_responses_guardrail_handler.py +++ b/tests/unit/llms/openai/responses/test_openai_responses_guardrail_handler.py @@ -3197,6 +3197,25 @@ class TestOpenAIResponsesHandlerNamespaceTools: assert result["tools"] == expected_tools + @pytest.mark.asyncio + async def test_scanning_tools_with_malformed_flags_leaves_them_for_the_route_to_judge(self): + data: Final = _codex_request("hi") + data["tools"][0]["defer_loading"] = "false" + data["tools"][1]["tools"][0]["allowed_callers"] = "direct" + data["tools"][2]["defer_loading"] = 1 + expected_tools: Final = copy.deepcopy(data["tools"]) + guardrail: Final = MockRecordingGuardrail(guardrail_name="test") + + result: Final = await OpenAIResponsesHandler().process_input_messages(data, guardrail) + + assert result["tools"] == expected_tools + assert [tool["function"]["name"] for tool in guardrail.seen_inputs[0]["tools"]] == [ + "get_weather", + "mcp__confluence__confluence_get_page", + "mcp__confluence__confluence_search", + "apply_patch", + ] + @pytest.mark.asyncio async def test_appending_guardrail_keeps_namespace_and_adds_tool(self): data = _codex_request("hi") diff --git a/tests/unit/responses/litellm_completion_transformation/test_anthropic_responses_bridge.py b/tests/unit/responses/litellm_completion_transformation/test_anthropic_responses_bridge.py index 45d3f8996bf..6086a3f6013 100644 --- a/tests/unit/responses/litellm_completion_transformation/test_anthropic_responses_bridge.py +++ b/tests/unit/responses/litellm_completion_transformation/test_anthropic_responses_bridge.py @@ -1,5 +1,6 @@ import json from collections.abc import Mapping, Sequence +from itertools import chain from typing import Final from unittest.mock import AsyncMock, MagicMock, patch @@ -341,7 +342,13 @@ def _search_turn(tool_calls: Sequence[Mapping[str, object]]) -> Mapping[str, obj async def _follow_up_messages( tool_calls: Sequence[Mapping[str, object]], model: str, custom_llm_provider: str ) -> Sequence[object]: - with patch("litellm.proxy.proxy_server.prisma_client", _FakePrismaClient([_search_turn(tool_calls)])): + return await _follow_up_messages_after(_search_turn(tool_calls), model, custom_llm_provider) + + +async def _follow_up_messages_after( + spend_log: Mapping[str, object], model: str, custom_llm_provider: str +) -> Sequence[object]: + with patch("litellm.proxy.proxy_server.prisma_client", _FakePrismaClient([spend_log])): request: Final = await LiteLLMCompletionResponsesConfig.async_responses_api_session_handler( previous_response_id="chatcmpl-search-turn", litellm_completion_request={ @@ -365,6 +372,7 @@ def _replayed_tool_call_ids(messages: Sequence[object]) -> list[str]: [ ("mistral/mistral-small-latest", "mistral", ["toolu_weather"]), ("gemini/gemini-2.5-flash", "gemini", ["toolu_weather"]), + ("databricks/databricks-claude-sonnet-4-5", "databricks", ["toolu_weather"]), (_MODEL, "anthropic", ["srvtoolu_search", "toolu_weather"]), ], ) @@ -387,6 +395,89 @@ async def test_previous_response_id_drops_a_claude_turn_that_held_only_a_search_ assert [message["role"] for message in messages] == ["user", "user"] +def _replayed_search_turn(search_answered: bool) -> Mapping[str, object]: + search_result: Final = ( + [{"type": "function_call_output", "call_id": "srvtoolu_search", "output": "get_weather"}] + if search_answered + else [] + ) + return { + **_search_turn([]), + "proxy_server_request": { + "model": _MODEL, + "input": [ + {"role": "user", "content": "What is the weather in Tokyo?"}, + { + "type": "function_call", + "call_id": "srvtoolu_search", + "name": "tool_search_tool_regex", + "arguments": '{"pattern": "weather"}', + }, + *search_result, + { + "type": "function_call", + "call_id": "toolu_weather", + "name": "get_weather", + "arguments": '{"city": "Tokyo"}', + }, + {"type": "function_call_output", "call_id": "toolu_weather", "output": '{"temp_c": 18}'}, + ], + }, + "response": { + "id": "chatcmpl-search-turn", + "object": "chat.completion", + "created": 0, + "model": _MODEL, + "choices": [ + { + "index": 0, + "finish_reason": "stop", + "message": {"role": "assistant", "content": "It is 18C in Tokyo."}, + } + ], + }, + } + + +def _tool_call_ids_of(message: object) -> Sequence[str]: + if isinstance(message, Message): + return [tool_call.id or "" for tool_call in message.tool_calls or ()] + if isinstance(message, dict): + return [tool_call["id"] for tool_call in message.get("tool_calls") or ()] + return [] + + +def _history_tool_ids(messages: Sequence[object]) -> tuple[Sequence[str], Sequence[str]]: + called: Final = list(chain.from_iterable(map(_tool_call_ids_of, messages))) + answered: Final = [ + message["tool_call_id"] for message in messages if isinstance(message, dict) and message["role"] == "tool" + ] + return called, answered + + +@pytest.mark.asyncio +@pytest.mark.parametrize("search_answered", [False, True]) +@pytest.mark.parametrize( + ("model", "custom_llm_provider", "expect_search"), + [("mistral/mistral-small-latest", "mistral", False), (_MODEL, "anthropic", True)], +) +async def test_previous_response_id_drops_a_client_replayed_search_call_only_off_anthropic( + model: str, custom_llm_provider: str, expect_search: bool, search_answered: bool +) -> None: + """ + A client can replay Claude's search call as a ``function_call`` input item with its ``srvtoolu_`` id. A later + turn sent elsewhere must not carry it, which OpenAI-compatible providers answer 400 "did not have response messages" + """ + messages: Final = await _follow_up_messages_after( + _replayed_search_turn(search_answered), model, custom_llm_provider + ) + + called, answered = _history_tool_ids(messages) + search_ids: Final = ["srvtoolu_search"] if expect_search else [] + assert called == [*search_ids, "toolu_weather"] + assert answered == [*(search_ids if search_answered else []), "toolu_weather"] + + def _anthropic_sse(*events: Mapping[str, object]) -> bytes: return b"".join( f"event: {event['type']}\ndata: {json.dumps(event)}\n\n".encode() for event in events diff --git a/tests/unit/responses/litellm_completion_transformation/test_litellm_completion_responses.py b/tests/unit/responses/litellm_completion_transformation/test_litellm_completion_responses.py index b81da5cba64..d0d75c6a33c 100644 --- a/tests/unit/responses/litellm_completion_transformation/test_litellm_completion_responses.py +++ b/tests/unit/responses/litellm_completion_transformation/test_litellm_completion_responses.py @@ -1,4 +1,5 @@ import json +from collections.abc import Mapping from copy import deepcopy from typing import Final, Literal @@ -1853,6 +1854,27 @@ class TestToolTransformation: assert chat_tools[0] == {"type": "tool_search_tool_bm25_20251119", "name": "tool_search_tool_bm25"} assert [tool["defer_loading"] for tool in chat_tools[1:]] == [True, True] + @pytest.mark.parametrize( + "native_search_tool", + [ + {"type": "tool_search_tool_regex_20251119", "name": "tool_search_tool_regex"}, + {"type": "tool_search_tool_bm25_20251119", "name": "tool_search_tool_bm25"}, + ], + ) + def test_server_tool_search_beside_a_native_search_tool_sends_only_the_native_one( + self, native_search_tool: Mapping[str, str] + ) -> None: + result: Final = LiteLLMCompletionResponsesConfig.transform_responses_api_request_to_chat_completion_request( + model="anthropic/claude-sonnet-5-5", + input="What is the weather in Paris right now?", + responses_api_request={"tools": [native_search_tool, *_deferred_tools_with_server_tool_search()]}, + custom_llm_provider="anthropic", + ) + + chat_tools: Final = result["tools"] + assert [tool for tool in chat_tools if tool["type"] != "function"] == [native_search_tool] + assert [tool["defer_loading"] for tool in chat_tools if tool["type"] == "function"] == [True, True] + def test_native_search_tool_passes_through_routes_without_tool_search(self) -> None: model: Final = "azure_ai/claude-sonnet-5-5" assert litellm.supports_tool_search(model, "azure_ai") is False @@ -1937,11 +1959,18 @@ class TestToolTransformation: transform: Final = LiteLLMCompletionResponsesConfig.transform_responses_api_request_to_chat_completion_request tools: Final = [ {"type": "tool_search", "execution": "server"}, - {"type": "function", "name": "get_weather", "parameters": {"type": "object"}, "defer_loading": "false"}, + { + "type": "function", + "name": "get_weather", + "parameters": {"type": "object"}, + "defer_loading": "false", + "allowed_callers": "direct", + }, { "type": "namespace", "name": "finance", "tools": [ + "not a tool", { "type": "function", "name": "get_stock_price", @@ -1985,6 +2014,15 @@ class TestToolTransformation: } assert all("allowed_callers" not in tool for tool in function_tools) + def test_top_level_function_allowed_callers_must_be_a_list_of_strings(self) -> None: + transform: Final = LiteLLMCompletionResponsesConfig.transform_responses_api_tools_to_chat_completion_tools + weather_tool: Final = {"type": "function", "name": "get_weather", "parameters": {"type": "object"}} + + with pytest.raises(litellm.BadRequestError, match="allowed_callers must be a list of strings"): + transform(tools=[{**weather_tool, "allowed_callers": "direct"}]) + converted, _ = transform(tools=[{**weather_tool, "allowed_callers": ["direct"]}]) + assert converted[0]["allowed_callers"] == ["direct"] + def test_custom_tools_stay_deferred_when_tool_search_goes_out(self) -> None: result: Final = LiteLLMCompletionResponsesConfig.transform_responses_api_request_to_chat_completion_request( model="anthropic/claude-sonnet-5-5", diff --git a/tests/unit/types/llms/test_types_llms_openai.py b/tests/unit/types/llms/test_types_llms_openai.py index 871e4cc8cfb..e90d48b64af 100644 --- a/tests/unit/types/llms/test_types_llms_openai.py +++ b/tests/unit/types/llms/test_types_llms_openai.py @@ -668,3 +668,21 @@ def test_openai_video_object_hidden_params_public_accessor_preserves_identity() video.hidden_params = replacement assert video._hidden_params is replacement assert video.hidden_params["public_key"] == "visible" + + +@pytest.mark.parametrize("extra_keys", ({}, {"created_by": "server"}), ids=("bridge-item", "openai-item")) +def test_rebuilding_a_dumped_responses_answer_keeps_the_tool_search_call_keys(extra_keys: dict[str, str]) -> None: + search_call: Final = { + "id": "tsc_1", + "arguments": {"query": "weather"}, + "call_id": None, + "execution": "server", + "status": "completed", + "type": "tool_search_call", + **extra_keys, + } + answer: Final = ResponsesAPIResponse(id="resp_1", created_at=1, output=[search_call]) + + rebuilt: Final = ResponsesAPIResponse(**answer.model_dump()) + + assert rebuilt.model_dump()["output"] == [search_call]