fix(responses): validate function allowed_callers, send one search tool, and keep replays clean off Anthropic

A function tool whose allowed_callers is not a list of strings now answers 400
instead of 500. The guardrail scan reads tools without malformed flags, so the
route decides between a 400 and a drop under drop_params. When the caller sends
a native Anthropic search tool beside tool_search, only the native one goes out.
Databricks Claude no longer counts as an Anthropic request route.

A previous_response_id follow-up off Anthropic also drops a search call the
client replayed as a function_call with a srvtoolu_ id, and its output. The
answered tool id scan in the Anthropic message factory runs once per assistant
run instead of once per message. A null created_by on tool_search_call and
tool_search_output output items is left out of the serialized response, so a
cache hit returns the same keys as the first call.
This commit is contained in:
mateo-berri 2026-10-13 13:26:57 -07:00
parent ba49b2026d
commit 67fe21b3a2
14 changed files with 1020 additions and 31 deletions

View file

@ -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

View file

@ -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

View file

@ -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(

View file

@ -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:

View file

@ -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:

View file

@ -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

View file

@ -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

View file

@ -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(

View file

@ -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:

View file

@ -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"

View file

@ -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")

View file

@ -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

View file

@ -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",

View file

@ -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]