fix(gemini): read the thought signature back from Responses input and coerce the per-request setting
Some checks failed
LiteLLM Rust / rust-lint (push) Has been cancelled
LiteLLM Rust / rust-test (push) Has been cancelled
LiteLLM Rust / rust-wheel (push) Has been cancelled
Terraform Modules / fmt, validate, test (aws) (push) Has been cancelled
Terraform Modules / fmt, validate, test (gcp) (push) Has been cancelled
Terraform Provider / gofmt, vet, build, test (push) Has been cancelled
Terraform Provider / Provider endpoints vs proxy OpenAPI schema (push) Has been cancelled

A Responses function_call input item now keeps its provider_specific_fields, so a
client that sends the item back gets its real thought signature replayed to Gemini.

A per-request thought_signature_in_tool_call_id of "false" or 0 now keeps ids short,
and a value that is not a boolean answers 400, or is dropped under drop_params,
instead of a 500.

The e2e test now uses the sdk fixture and @meta subjects
This commit is contained in:
mateo-berri 2026-10-20 20:47:01 -07:00
parent f617727ff5
commit 8034fdbb9e
5 changed files with 275 additions and 172 deletions

View file

@ -221,6 +221,7 @@ def _handle_ollama_system_message(messages: list, prompt: str, msg_i: int) -> tu
_OLLAMA_USER_ROLES: Final = frozenset({"user", "tool", "function"})
_DECODED_JSON: Final = TypeAdapter(object)
_OPTIONAL_BOOL: Final = TypeAdapter[bool | None](bool | None)
def _ollama_bad_message(model: str, message: AllMessageValues, msg_i: int, detail: str) -> "litellm.BadRequestError":
@ -1269,7 +1270,10 @@ def _same_tool_call(tool_call_id: str, other_tool_call_id: str) -> bool:
def embeds_thought_signature_in_tool_call_id(litellm_params: Mapping[str, object]) -> bool:
return litellm_params.get("thought_signature_in_tool_call_id") is not False
try:
return _OPTIONAL_BOOL.validate_python(litellm_params.get("thought_signature_in_tool_call_id")) is not False
except ValidationError:
return True
def get_thought_signature_from_tool(tool: Mapping[str, object]) -> str | None:

View file

@ -381,6 +381,27 @@ def _resolve_control_options(kwargs: Mapping[str, object], model: str) -> Contro
return assert_never(control)
def _supplemental_provider_params(kwargs: Mapping[str, object], model: str) -> GenericLiteLLMParams | None:
supplemental: Final = {key: kwargs[key] for key in OPTIONAL_KWARGS_KEYS if key in kwargs}
if not supplemental:
return None
try:
return GenericLiteLLMParams.model_validate(supplemental)
except ValidationError as error:
invalid: Final = tuple(detail for detail in error.errors() if detail["loc"])
invalid_keys: Final = frozenset(str(detail["loc"][0]) for detail in invalid)
if litellm.drop_params is True or normalize_drop_params(kwargs.get("drop_params")) is True:
return GenericLiteLLMParams.model_validate(
{key: value for key, value in supplemental.items() if key not in invalid_keys}
)
raise litellm.BadRequestError(
message="; ".join(f"{detail['loc'][0]}: {detail['msg']}" for detail in invalid),
model=model,
llm_provider=None,
body={"param": ", ".join(sorted(invalid_keys))},
) from error
####### COMPLETION ENDPOINTS ################
@ -672,17 +693,15 @@ async def acompletion(
"enable_json_schema_validation": enable_json_schema_validation,
}
if custom_llm_provider is None:
_supplemental: Final = cast( # cast-ok: kwargs values are request-scoped objects
"dict[str, object]", {k: kwargs[k] for k in OPTIONAL_KWARGS_KEYS if k in kwargs}
)
request_kwargs: Final = cast("Mapping[str, object]", kwargs) # cast-ok: untyped request kwargs
_, custom_llm_provider, _, _ = get_llm_provider(
model=model,
custom_llm_provider=custom_llm_provider,
api_base=cast( # cast-ok: api_base arrives as a str in the request kwargs
"str | None", kwargs.get("api_base")
"str | None", request_kwargs.get("api_base")
)
or base_url,
litellm_params=(GenericLiteLLMParams.model_validate(_supplemental) if _supplemental else None),
litellm_params=_supplemental_provider_params(request_kwargs, model),
)
fallbacks = fallbacks or litellm.model_fallbacks
@ -5657,16 +5676,15 @@ def completion(
if deployment_id is not None: # azure llms
model = deployment_id
custom_llm_provider = "azure"
_supplemental_provider_params: Final = {k: kwargs[k] for k in OPTIONAL_KWARGS_KEYS if k in kwargs}
routing_kwargs: Final = cast("Mapping[str, object]", kwargs) # cast-ok: kwargs is the untyped request dict
model, custom_llm_provider, dynamic_api_key, api_base = get_llm_provider(
model=model,
custom_llm_provider=custom_llm_provider,
api_base=api_base,
api_key=api_key,
litellm_params=(
GenericLiteLLMParams.model_validate(_supplemental_provider_params)
if _supplemental_provider_params
else None
litellm_params=_supplemental_provider_params(
routing_kwargs,
cast(str, model), # cast-ok: model is the alias-resolved request model name
),
)
@ -5680,7 +5698,6 @@ def completion(
cast(object, model_info) # cast-ok: model_info is the untyped dict kwarg
),
)
routing_kwargs: Final = cast("Mapping[str, object]", kwargs) # cast-ok: kwargs is the untyped request dict
drop_unsupported_reasoning: Final = (
litellm.drop_params is True or normalize_drop_params(routing_kwargs.get("drop_params")) is True
)

View file

@ -45,7 +45,7 @@ from litellm.constants import (
REDACTED_TOOL_CALL_ARGUMENTS_PLACEHOLDER,
)
from litellm.integrations.anthropic_cache_control_hook import MAX_CACHE_CONTROL_BLOCKS, AnthropicCacheControlHook
from litellm.litellm_core_utils.core_helpers import drops_unsupported_params, normalize_drop_params
from litellm.litellm_core_utils.core_helpers import as_str_mapping, drops_unsupported_params, normalize_drop_params
from litellm.litellm_core_utils.get_supported_openai_params import (
get_supported_openai_params,
)
@ -252,8 +252,9 @@ class _ResponsesToolCallItem(Protocol):
def get(self, key: str, /) -> object: ...
class _MarkedToolCallChunk(ChatCompletionToolCallChunk):
class _ReplayedToolCallChunk(ChatCompletionToolCallChunk, total=False):
cache_control: ReadOnly[ChatCompletionCachedContent]
provider_specific_fields: ReadOnly[Mapping[str, object]]
class _MarkedToolMessage(ChatCompletionToolMessage):
@ -2231,7 +2232,7 @@ class LiteLLMCompletionResponsesConfig:
tool_call: Final = (
ChatCompletionToolCallChunk(id=tool_call_id, type="function", function=tool_call_function, index=0)
if function_call_cache_control is None
else _MarkedToolCallChunk(
else _ReplayedToolCallChunk(
id=tool_call_id,
type="function",
function=tool_call_function,
@ -2239,9 +2240,17 @@ class LiteLLMCompletionResponsesConfig:
cache_control=function_call_cache_control,
)
)
provider_specific_fields: Final = as_str_mapping(function_call.get("provider_specific_fields"))
replayed_tool_call: Final = (
tool_call
if provider_specific_fields is None
else cast( # cast-ok: the same tool call plus the provider_specific_fields the client sent back
_ReplayedToolCallChunk, {**tool_call, "provider_specific_fields": provider_specific_fields}
)
)
# Create an assistant message with the tool call
return [ChatCompletionResponseMessage(tool_calls=[tool_call], role="assistant", content=None)]
return [ChatCompletionResponseMessage(tool_calls=[replayed_tool_call], role="assistant", content=None)]
@staticmethod
def _resolve_file_id(item: Mapping[str, object]) -> object:

View file

@ -1,26 +1,27 @@
from __future__ import annotations
import re
from itertools import chain
from collections.abc import Iterator
from typing import Final
import pytest
from e2e_config import unique_marker
from e2e_http import Success, unwrap
from e2e_metadata import Capability, Domain, Mode, Provider, Route, Subject, meta
from lifecycle import ResourceManager
from models import (
ChatAssistantTurn,
ChatBody,
ChatMessage,
ChatResponse,
ChatTool,
ChatToolFunction,
ChatToolResultTurn,
LiteLLMParamsBody,
ToolCall,
from llm_translation.sdk_clients import NO_PROXY_CACHE, SdkClients
from models import LiteLLMParamsBody
from openai.types.chat import (
ChatCompletion,
ChatCompletionAssistantMessageParam,
ChatCompletionChunk,
ChatCompletionMessageFunctionToolCallParam,
ChatCompletionToolMessageParam,
ChatCompletionToolParam,
ChatCompletionUserMessageParam,
)
from openai.types.chat.chat_completion_chunk import ChoiceDeltaToolCall
from openai.types.chat.chat_completion_message_function_tool_call import ChatCompletionMessageFunctionToolCall
from passthrough_client import PassthroughClient
from pydantic import BaseModel
pytestmark = pytest.mark.e2e
@ -28,33 +29,25 @@ BACKEND_MODEL: Final = "gemini/gemini-3.5-flash"
GEMINI_API_KEY: Final = "os.environ/GEMINI_API_KEY"
# bedrock-agentcore HarnessToolUseId shape in botocore 1.43.6, read 2026-10-20
AGENTCORE_TOOL_USE_ID: Final = re.compile(r"[a-zA-Z0-9_-]{1,64}")
WEATHER_TOOL: Final = ChatTool(
function=ChatToolFunction(
name="get_weather",
description="Get the weather for a city",
parameters={
"type": "object",
"properties": {"city": {"type": "string"}},
"required": ["city"],
},
WEATHER_TOOL: Final[ChatCompletionToolParam] = {
"type": "function",
"function": {
"name": "get_weather",
"description": "Get the weather for a city",
"parameters": {"type": "object", "properties": {"city": {"type": "string"}}, "required": ["city"]},
},
}
def _subject(mode: Mode) -> Subject:
return Subject(
domain=Domain.LLM_TRANSLATION,
route=Route.CHAT_COMPLETIONS,
providers=(Provider.GEMINI,),
models=(BACKEND_MODEL,),
mode=mode,
capabilities=(Capability.FUNCTION_CALLING, Capability.TOOL_CHOICE),
)
)
class _StreamToolCall(BaseModel):
id: str | None = None
class _StreamDelta(BaseModel):
tool_calls: tuple[_StreamToolCall, ...] = ()
class _StreamChoice(BaseModel):
delta: _StreamDelta
class _StreamChunk(BaseModel):
choices: tuple[_StreamChoice, ...] = ()
def _register_model(client: PassthroughClient, resources: ResourceManager) -> str:
@ -71,44 +64,42 @@ def _register_model(client: PassthroughClient, resources: ResourceManager) -> st
return model_name
def _tool_request(model: str, *, stream: bool = False) -> ChatBody:
return ChatBody(
model=model,
messages=(
ChatMessage(
role="user",
content=f"Use get_weather for San Francisco and return its result. {unique_marker()}",
),
),
tools=(WEATHER_TOOL,),
tool_choice="required",
max_tokens=128,
stream=stream,
def _user_message() -> ChatCompletionUserMessageParam:
return {"role": "user", "content": f"Use get_weather for San Francisco and return its result. {unique_marker()}"}
def _tool_calls(completion: ChatCompletion) -> tuple[ChatCompletionMessageFunctionToolCall, ...]:
assert completion.choices, f"Gemini returned no choices: {completion.model_dump_json()[:400]}"
calls: Final = tuple(
call
for call in completion.choices[0].message.tool_calls or ()
if isinstance(call, ChatCompletionMessageFunctionToolCall)
)
def _tool_calls(response: ChatResponse) -> tuple[ToolCall, ...]:
assert response.choices, f"Gemini returned no choices: {response}"
message: Final = response.choices[0].message
calls: Final = tuple(message.tool_calls or ()) if message else ()
assert calls, f"Gemini returned no tool calls for tool_choice='required': {response}"
assert calls, f"Gemini returned no tool calls for tool_choice='required': {completion.model_dump_json()[:400]}"
return calls
def _tool_call_id(call: ToolCall) -> str:
assert call.id, f"Gemini returned a tool call without an id: {call}"
return call.id
def _assert_agentcore_accepts_ids(ids: tuple[str, ...]) -> None:
rejected: Final = tuple(call_id for call_id in ids if not AGENTCORE_TOOL_USE_ID.fullmatch(call_id))
assert not rejected, f"tool call id(s) AgentCore would reject: {tuple(call_id[:80] for call_id in rejected)}"
def _stream_tool_calls(events: list[str]) -> tuple[_StreamToolCall, ...]:
chunks: Final = tuple(_StreamChunk.model_validate_json(event) for event in events)
choices: Final = chain.from_iterable(chunk.choices for chunk in chunks)
return tuple(chain.from_iterable(choice.delta.tool_calls for choice in choices))
def _streamed_tool_calls(chunks: tuple[ChatCompletionChunk, ...]) -> Iterator[ChoiceDeltaToolCall]:
for chunk in chunks:
for choice in chunk.choices:
yield from choice.delta.tool_calls or ()
def _short_id_tool_call(call: ChatCompletionMessageFunctionToolCall) -> ChatCompletionMessageFunctionToolCallParam:
return {
"id": call.id,
"type": "function",
"function": {"name": call.function.name, "arguments": call.function.arguments},
}
def _tool_result(call: ChatCompletionMessageFunctionToolCall) -> ChatCompletionToolMessageParam:
return {"role": "tool", "tool_call_id": call.id, "content": '{"temperature": 21, "condition": "sunny"}'}
class TestGeminiToolCallIdLength:
@ -116,37 +107,44 @@ class TestGeminiToolCallIdLength:
"llm.chat_completions.gemini.tool_use.nonstream.works",
exercised_on=["chat_completions"],
)
@meta(_subject(Mode.NONSTREAM))
def test_non_streaming_tool_call_id_fits_agentcore(
self, client: PassthroughClient, resources: ResourceManager
self, client: PassthroughClient, resources: ResourceManager, sdk: SdkClients
) -> None:
model: Final = _register_model(client, resources)
response: Final = unwrap(
client.proxy.chat(resources.key(), _tool_request(model))
completion: Final = sdk.openai(resources.key()).chat.completions.create(
model=model,
messages=[_user_message()],
tools=[WEATHER_TOOL],
tool_choice="required",
max_tokens=128,
extra_body=NO_PROXY_CACHE,
)
calls: Final = _tool_calls(response)
ids: Final = tuple(_tool_call_id(call) for call in calls)
_assert_agentcore_accepts_ids(ids)
_assert_agentcore_accepts_ids(tuple(call.id for call in _tool_calls(completion)))
@pytest.mark.covers(
"llm.chat_completions.gemini.tool_use.stream.works",
exercised_on=["chat_completions"],
)
@meta(_subject(Mode.STREAM))
def test_streaming_tool_call_id_fits_agentcore(
self, client: PassthroughClient, resources: ResourceManager
self, client: PassthroughClient, resources: ResourceManager, sdk: SdkClients
) -> None:
model: Final = _register_model(client, resources)
result: Final = client.proxy.chat_stream(
resources.key(), _tool_request(model, stream=True)
chunks: Final[tuple[ChatCompletionChunk, ...]] = tuple(
sdk.openai(resources.key()).chat.completions.create(
model=model,
messages=[_user_message()],
tools=[WEATHER_TOOL],
tool_choice="required",
max_tokens=128,
stream=True,
extra_body=NO_PROXY_CACHE,
)
)
assert result.ok, f"Gemini chat stream failed: {result}"
assert result.is_streaming, f"Gemini response was not SSE: {result.content_type!r}"
assert result.stream_error is None, f"Gemini stream contained an error: {result.stream_error}"
calls: Final = _stream_tool_calls(result.stream_events)
assert calls, f"Gemini stream returned no tool call deltas: {result.stream_events[:5]}"
ids: Final = tuple(call.id for call in calls if call.id)
assert ids, f"Gemini stream tool call deltas contained no ids: {calls}"
ids: Final = tuple(call.id for call in _streamed_tool_calls(chunks) if call.id)
assert ids, f"Gemini stream returned no tool call ids in {len(chunks)} chunks"
_assert_agentcore_accepts_ids(ids)
@ -154,49 +152,36 @@ class TestGeminiToolCallIdLength:
"llm.chat_completions.gemini.multi_turn.nonstream.works",
exercised_on=["chat_completions"],
)
@meta(_subject(Mode.NONSTREAM))
def test_tool_call_round_trip_succeeds(
self, client: PassthroughClient, resources: ResourceManager
self, client: PassthroughClient, resources: ResourceManager, sdk: SdkClients
) -> None:
model: Final = _register_model(client, resources)
key: Final = resources.key()
first_request: Final = _tool_request(model)
first_response: Final = unwrap(client.proxy.chat(key, first_request))
calls: Final = _tool_calls(first_response)
first_message: Final = first_response.choices[0].message
assert first_message is not None, f"Gemini returned no assistant message: {first_response}"
ids: Final = tuple(_tool_call_id(call) for call in calls)
openai_client: Final = sdk.openai(resources.key())
question: Final = _user_message()
first: Final = openai_client.chat.completions.create(
model=model,
messages=[question],
tools=[WEATHER_TOOL],
tool_choice="required",
max_tokens=128,
extra_body=NO_PROXY_CACHE,
)
calls: Final = _tool_calls(first)
assistant: Final[ChatCompletionAssistantMessageParam] = {
"role": "assistant",
"tool_calls": [_short_id_tool_call(call) for call in calls],
}
second_result: Final = client.proxy.chat(
key,
ChatBody(
model=model,
messages=(
*first_request.messages,
ChatAssistantTurn(
content=first_message.content,
reasoning_content=first_message.reasoning_content,
tool_calls=first_message.tool_calls,
),
*tuple(
ChatToolResultTurn(
tool_call_id=call_id,
content='{"temperature": 21, "condition": "sunny"}',
)
for call_id in ids
),
),
tools=(WEATHER_TOOL,),
tool_choice="none",
max_tokens=128,
),
)
assert isinstance(second_result, Success), f"Gemini turn 2 failed: {second_result}"
assert second_result.status_code == 200, (
f"Gemini turn 2 returned HTTP {second_result.status_code}, expected 200"
)
second_response: Final = second_result.data
assert second_response.choices, f"Gemini turn 2 returned no choices: {second_response}"
answer: Final = second_response.choices[0].message
assert answer is not None and answer.content and answer.content.strip(), (
f"Gemini turn 2 returned an empty answer: {second_response}"
second: Final = openai_client.chat.completions.create(
model=model,
messages=[question, assistant, *(_tool_result(call) for call in calls)],
tools=[WEATHER_TOOL],
tool_choice="none",
max_tokens=128,
extra_body=NO_PROXY_CACHE,
)
assert second.choices, f"Gemini turn 2 returned no choices: {second.model_dump_json()[:400]}"
answer: Final = second.choices[0].message.content
assert answer and answer.strip(), f"Gemini turn 2 returned an empty answer: {second.model_dump_json()[:400]}"

View file

@ -11,13 +11,14 @@ enable_preview_features=True to be enabled.
import json
import re
from collections.abc import Mapping
from collections.abc import Mapping, Sequence
from itertools import chain
from typing import Final
import httpx
import pytest
import respx
from typing_extensions import ReadOnly, TypedDict
import litellm
from litellm.litellm_core_utils.prompt_templates.factory import (
@ -339,6 +340,37 @@ def _gemini_response(native_id: str | None, stream: bool) -> httpx.Response:
return httpx.Response(200, json=body)
def _gemini_answer() -> httpx.Response:
return httpx.Response(
200,
json={
"candidates": [
{"content": {"role": "model", "parts": [{"text": "Sunny"}]}, "finishReason": "STOP", "index": 0}
],
"usageMetadata": {"promptTokenCount": 9, "candidatesTokenCount": 1, "totalTokenCount": 10},
},
)
class _GeminiTurn(TypedDict):
role: ReadOnly[str]
parts: ReadOnly[Sequence[Mapping[str, object]]]
class _GeminiRequestBody(TypedDict):
contents: ReadOnly[Sequence[_GeminiTurn]]
def _is_bounded(tool_call_id: object) -> bool:
return isinstance(tool_call_id, str) and BOUNDED_TOOL_CALL_ID.fullmatch(tool_call_id) is not None
def _replayed_signatures(sent_body: _GeminiRequestBody) -> list[object]:
model_turns: Final = [turn for turn in sent_body["contents"] if turn["role"] == "model"]
model_parts: Final = chain.from_iterable(turn["parts"] for turn in model_turns)
return [part["thoughtSignature"] for part in model_parts if "function_call" in part]
def _returned_tool_calls(
native_id: str | None, stream: bool, request_params: Mapping[str, object]
) -> tuple[list[Mapping[str, object]], Mapping[str, object]]:
@ -369,7 +401,7 @@ def test_tool_call_id_stays_bounded_when_thought_signature_kept_out_of_it(stream
tool_call_ids: Final = [call["id"] for call in tool_calls if call["id"]]
assert len(tool_call_ids) == 1
assert BOUNDED_TOOL_CALL_ID.fullmatch(tool_call_ids[0])
assert _is_bounded(tool_call_ids[0])
assert native_id is None or tool_call_ids[0] == native_id
assert tool_calls[0]["provider_specific_fields"]["thought_signature"] == LONG_SIGNATURE
assert "thought_signature_in_tool_call_id" not in json.dumps(sent_body)
@ -400,17 +432,7 @@ def test_bounded_tool_call_id_round_trips_to_gemini(client_keeps_provider_fields
}
with respx.mock:
route: Final = respx.post(url__regex=GEMINI_URL_PATTERN).mock(
return_value=httpx.Response(
200,
json={
"candidates": [
{"content": {"role": "model", "parts": [{"text": "Sunny"}]}, "finishReason": "STOP", "index": 0}
],
"usageMetadata": {"promptTokenCount": 9, "candidatesTokenCount": 1, "totalTokenCount": 10},
},
)
)
route: Final = respx.post(url__regex=GEMINI_URL_PATTERN).mock(return_value=_gemini_answer())
litellm.completion(
model="gemini/gemini-3.5-flash",
messages=[
@ -424,9 +446,7 @@ def test_bounded_tool_call_id_round_trips_to_gemini(client_keeps_provider_fields
)
sent_body: Final = json.loads(route.calls.last.request.content)
model_turns: Final = [turn for turn in sent_body["contents"] if turn["role"] == "model"]
model_parts: Final = chain.from_iterable(turn["parts"] for turn in model_turns)
replayed_signatures: Final = [part["thoughtSignature"] for part in model_parts if "function_call" in part]
replayed_signatures: Final = _replayed_signatures(sent_body)
assert len(replayed_signatures) == 1
assert (replayed_signatures[0] == LONG_SIGNATURE) is client_keeps_provider_fields
assert replayed_signatures[0]
@ -435,17 +455,7 @@ def test_bounded_tool_call_id_round_trips_to_gemini(client_keeps_provider_fields
def test_tool_result_with_bare_id_matches_call_whose_id_the_client_re_signed():
bare_id: Final = "call_601401"
with respx.mock:
route: Final = respx.post(url__regex=GEMINI_URL_PATTERN).mock(
return_value=httpx.Response(
200,
json={
"candidates": [
{"content": {"role": "model", "parts": [{"text": "Sunny"}]}, "finishReason": "STOP", "index": 0}
],
"usageMetadata": {"promptTokenCount": 9, "candidatesTokenCount": 1, "totalTokenCount": 10},
},
)
)
route: Final = respx.post(url__regex=GEMINI_URL_PATTERN).mock(return_value=_gemini_answer())
litellm.completion(
model="gemini/gemini-3.5-flash",
messages=[
@ -514,7 +524,7 @@ async def test_messages_tool_use_keeps_thought_signature_kept_out_of_its_id(
tool_uses: Final = await _messages_tool_use_blocks(stream)
assert len(tool_uses) == 1
assert BOUNDED_TOOL_CALL_ID.fullmatch(tool_uses[0]["id"])
assert _is_bounded(tool_uses[0]["id"])
assert tool_uses[0]["provider_specific_fields"]["signature"] == LONG_SIGNATURE
@ -545,10 +555,88 @@ def test_responses_function_call_keeps_thought_signature_kept_out_of_its_id(stre
function_calls: Final = _responses_function_call_items(stream)
assert len(function_calls) == 1
assert BOUNDED_TOOL_CALL_ID.fullmatch(function_calls[0].call_id)
assert _is_bounded(function_calls[0].call_id)
assert function_calls[0].provider_specific_fields["thought_signature"] == LONG_SIGNATURE
@pytest.mark.parametrize("client_keeps_provider_fields", [True, False])
def test_responses_function_call_replays_its_thought_signature_to_gemini(client_keeps_provider_fields: bool):
function_call: Final = _responses_function_call_items(False)[0]
replayed_item: Final = {
"type": "function_call",
"call_id": function_call.call_id,
"name": function_call.name,
"arguments": function_call.arguments,
**(
{"provider_specific_fields": function_call.provider_specific_fields}
if client_keeps_provider_fields
else {}
),
}
with respx.mock:
route: Final = respx.post(url__regex=GEMINI_URL_PATTERN).mock(return_value=_gemini_answer())
litellm.responses(
model="gemini/gemini-3.5-flash",
input=[
{"role": "user", "content": USER_TURN["content"]},
replayed_item,
{"type": "function_call_output", "call_id": function_call.call_id, "output": '{"temp_c": 21}'},
],
tools=[RESPONSES_WEATHER_TOOL],
api_key="test-gemini-key",
thought_signature_in_tool_call_id=False,
)
sent_body: Final = json.loads(route.calls.last.request.content)
replayed_signatures: Final = _replayed_signatures(sent_body)
assert len(replayed_signatures) == 1
assert (replayed_signatures[0] == LONG_SIGNATURE) is client_keeps_provider_fields
assert replayed_signatures[0]
@pytest.mark.parametrize("requested_value", ["false", "False", 0])
def test_request_setting_given_as_a_string_or_number_keeps_ids_bounded(requested_value: object):
tool_calls, _ = _returned_tool_calls(None, False, {"thought_signature_in_tool_call_id": requested_value})
assert _is_bounded(tool_calls[0]["id"])
assert tool_calls[0]["provider_specific_fields"]["thought_signature"] == LONG_SIGNATURE
async def _complete(use_async: bool, **request_params: object) -> litellm.ModelResponse:
request: Final = {
"model": "gemini/gemini-3.5-flash",
"messages": [USER_TURN],
"tools": [WEATHER_TOOL],
"api_key": "test-gemini-key",
**request_params,
}
return await litellm.acompletion(**request) if use_async else litellm.completion(**request)
@pytest.mark.parametrize("use_async", [False, True])
async def test_request_setting_that_is_not_a_boolean_answers_400(use_async: bool):
with respx.mock(assert_all_called=False):
respx.post(url__regex=GEMINI_URL_PATTERN).mock(return_value=_gemini_response(None, False))
with pytest.raises(litellm.BadRequestError, match="thought_signature_in_tool_call_id") as raised:
await _complete(use_async, thought_signature_in_tool_call_id="maybe")
assert raised.value.status_code == 400
@pytest.mark.parametrize("use_async", [False, True])
async def test_request_setting_that_is_not_a_boolean_is_dropped_under_drop_params(
use_async: bool, monkeypatch: pytest.MonkeyPatch
):
monkeypatch.setattr(litellm, "disable_aiohttp_transport", True)
with respx.mock:
respx.post(url__regex=GEMINI_URL_PATTERN).mock(return_value=_gemini_response(None, False))
response: Final = await _complete(use_async, thought_signature_in_tool_call_id="maybe", drop_params=True)
tool_call_id: Final = response.choices[0].message.tool_calls[0].id
assert get_thought_signature_from_tool({"id": tool_call_id}) == LONG_SIGNATURE
@pytest.mark.parametrize("configured_value", ["false", "False", 0, False])
def test_deployment_setting_given_as_a_string_or_number_keeps_ids_bounded(configured_value: object):
router: Final = litellm.Router(
@ -567,7 +655,7 @@ def test_deployment_setting_given_as_a_string_or_number_keeps_ids_bounded(config
respx.post(url__regex=GEMINI_URL_PATTERN).mock(return_value=_gemini_response(None, False))
response: Final = router.completion(model="gemini-agentcore", messages=[USER_TURN], tools=[WEATHER_TOOL])
assert BOUNDED_TOOL_CALL_ID.fullmatch(response.choices[0].message.tool_calls[0].id)
assert _is_bounded(response.choices[0].message.tool_calls[0].id)
def test_deployment_setting_rejects_a_value_that_is_not_a_boolean():