feat(gemini): support context circulation for server-side tool combination

Enables Gemini 3+ models to combine built-in tools (Google Search, etc.)
with custom functions via `include_server_side_tool_invocations=True`.
Server-side invocations are surfaced in provider_specific_fields and
automatically re-injected on subsequent turns for multi-turn coherence.

Closes #24047
This commit is contained in:
Chesars 2026-03-18 22:33:01 -03:00
parent 2405e0d400
commit 6f4b4d3c42
5 changed files with 461 additions and 2 deletions

View file

@ -54,6 +54,7 @@ response = completion(
- stream
- tools
- tool_choice
- include_server_side_tool_invocations
- functions
- response_format
- n
@ -856,7 +857,112 @@ curl -X POST 'http://0.0.0.0:4000/chat/completions' \
</TabItem>
</Tabs>
### URL Context
### Context Circulation (Server-Side Tool Combination)
Context circulation allows Gemini 3+ models to combine **built-in tools** (like Google Search) with **your custom functions** in the same request. Without it, Gemini returns an error if you try to use both.
When enabled, Gemini can execute Google Search server-side, use those results to decide whether to call your custom functions, and return the full chain of reasoning.
**How it works:**
1. You pass `include_server_side_tool_invocations=True` along with both Google Search and your function tools
2. Gemini executes server-side tools internally and returns `toolCall`/`toolResponse` parts alongside any `functionCall` parts
3. LiteLLM extracts the server-side invocations into `provider_specific_fields["server_side_tool_invocations"]`
4. On subsequent turns, include the full assistant message in your conversation history — LiteLLM re-injects the server-side parts automatically
<Tabs>
<TabItem value="sdk" label="SDK">
```python
from litellm import completion
response = completion(
model="gemini/gemini-3-flash-preview",
messages=[{"role": "user", "content": "What's the weather in Buenos Aires? If it's raining, schedule a meeting."}],
tools=[
{"type": "web_search_preview"}, # Google Search (server-side)
{
"type": "function",
"function": {
"name": "schedule_meeting",
"description": "Schedule a meeting",
"parameters": {
"type": "object",
"properties": {"reason": {"type": "string"}},
"required": ["reason"],
},
},
},
],
include_server_side_tool_invocations=True,
)
msg = response.choices[0].message
# Server-side tool results are in provider_specific_fields
psf = msg.provider_specific_fields or {}
for invocation in psf.get("server_side_tool_invocations", []):
print(invocation["tool_type"]) # e.g. "GOOGLE_SEARCH_WEB"
print(invocation["id"])
print(invocation["args"]) # e.g. {"queries": ["weather Buenos Aires"]}
print(invocation["response"]) # Search results from Google
# For multi-turn: just append the full message to history
messages.append(msg)
messages.append({"role": "user", "content": "Thanks!"})
# LiteLLM automatically re-injects the server-side parts + thought signatures
response2 = completion(
model="gemini/gemini-3-flash-preview",
messages=messages,
tools=tools,
include_server_side_tool_invocations=True,
)
```
</TabItem>
<TabItem value="proxy" label="PROXY">
1. Setup config.yaml
```yaml
model_list:
- model_name: gemini-3-flash
litellm_params:
model: gemini/gemini-3-flash-preview
api_key: os.environ/GEMINI_API_KEY
```
2. Start Proxy
```bash
$ litellm --config /path/to/config.yaml
```
3. Make Request
```bash
curl -X POST 'http://0.0.0.0:4000/chat/completions' \
-H 'Content-Type: application/json' \
-H 'Authorization: Bearer sk-1234' \
-d '{
"model": "gemini-3-flash",
"messages": [{"role": "user", "content": "What is the weather in Buenos Aires?"}],
"tools": [
{"type": "web_search_preview"},
{"type": "function", "function": {"name": "schedule_meeting", "description": "Schedule a meeting", "parameters": {"type": "object", "properties": {"reason": {"type": "string"}}}}}
],
"include_server_side_tool_invocations": true
}'
```
</TabItem>
</Tabs>
:::info
- Context circulation requires **Gemini 3+** models
- Server-side tool invocations (`toolCall`/`toolResponse`) are **not** included in `tool_calls` — they are in `provider_specific_fields["server_side_tool_invocations"]` because they were already executed by Google, not by your code
- `thought_signatures` are automatically preserved alongside server-side invocations for multi-turn coherence
:::
### URL Context
<Tabs>
<TabItem value="sdk" label="SDK">

View file

@ -540,6 +540,39 @@ def _gemini_convert_messages_with_history( # noqa: PLR0915
assistant_content.append(gemini_tool_call_part)
last_message_with_tool_calls = assistant_msg
## HANDLE SERVER-SIDE TOOL INVOCATIONS (context circulation)
_psf = assistant_msg.get("provider_specific_fields")
if isinstance(_psf, dict):
_ss_invocations = _psf.get("server_side_tool_invocations")
if isinstance(_ss_invocations, list):
for invocation in _ss_invocations:
# Re-inject toolCall part
tc_part: Dict[str, Any] = {
"toolCall": {
"toolType": invocation.get("tool_type"),
"id": invocation.get("id"),
"args": invocation.get("args"),
}
}
if "thought_signature" in invocation:
tc_part["thoughtSignature"] = invocation["thought_signature"]
assistant_content.append(tc_part) # type: ignore
# Re-inject toolResponse part if response is present
if "response" in invocation:
tr_dict: Dict[str, Any] = {
"id": invocation.get("id"),
"response": invocation.get("response"),
}
if invocation.get("tool_type"):
tr_dict["toolType"] = invocation["tool_type"]
tr_part: Dict[str, Any] = {
"toolResponse": tr_dict
}
if "thought_signature" in invocation:
tr_part["thoughtSignature"] = invocation["thought_signature"]
assistant_content.append(tr_part) # type: ignore
msg_i += 1
if assistant_content:
@ -666,6 +699,9 @@ def _transform_request_body( # noqa: PLR0915
)
tools: Optional[Tools] = optional_params.pop("tools", None)
tool_choice: Optional[ToolConfig] = optional_params.pop("tool_choice", None)
include_server_side_tool_invocations: bool = optional_params.pop(
"include_server_side_tool_invocations", False
)
safety_settings: Optional[List[SafetSettingsConfig]] = optional_params.pop(
"safety_settings", None
) # type: ignore
@ -715,6 +751,10 @@ def _transform_request_body( # noqa: PLR0915
data["tools"] = tools
if tool_choice is not None:
data["toolConfig"] = tool_choice
if include_server_side_tool_invocations:
if "toolConfig" not in data:
data["toolConfig"] = {}
data["toolConfig"]["includeServerSideToolInvocations"] = True
if safety_settings is not None:
data["safetySettings"] = safety_settings
if generation_config is not None and len(generation_config) > 0:

View file

@ -316,6 +316,7 @@ class VertexGeminiConfig(VertexAIBaseConfig, BaseConfig):
"audio",
"parallel_tool_calls",
"web_search_options",
"include_server_side_tool_invocations",
]
# Add penalty parameters only for non-preview models
@ -1119,6 +1120,8 @@ class VertexGeminiConfig(VertexAIBaseConfig, BaseConfig):
optional_params = self._add_tools_to_optional_params(
optional_params, [_tools]
)
elif param == "include_server_side_tool_invocations" and value is True:
optional_params["include_server_side_tool_invocations"] = True
if litellm.vertex_ai_safety_settings is not None:
optional_params["safety_settings"] = litellm.vertex_ai_safety_settings
@ -1360,6 +1363,67 @@ class VertexGeminiConfig(VertexAIBaseConfig, BaseConfig):
signatures.append(signature)
return signatures if signatures else None
@staticmethod
def _extract_server_side_tool_invocations(
parts: List[HttpxPartType],
) -> Optional[List[Dict[str, Any]]]:
"""Extract server-side tool invocations (toolCall/toolResponse) from parts.
These are returned by Gemini when context circulation is enabled
(includeServerSideToolInvocations=true). They represent tools executed
server-side (e.g. Google Search) and must be circulated back in
subsequent turns for multi-turn coherence.
Returns:
List of server-side invocation dicts if any found, None otherwise.
"""
invocations: List[Dict[str, Any]] = []
# Index toolCalls by id so we can pair them with responses
tool_calls_by_id: Dict[str, Dict[str, Any]] = {}
tool_responses_by_id: Dict[str, Dict[str, Any]] = {}
for part in parts:
if "toolCall" in part:
tc = part["toolCall"]
entry: Dict[str, Any] = {
"tool_type": tc.get("toolType"),
"id": tc.get("id"),
"args": tc.get("args"),
}
signature = part.get("thoughtSignature")
if signature is not None:
entry["thought_signature"] = signature
tool_calls_by_id[tc.get("id", "")] = entry
elif "toolResponse" in part:
tr = part["toolResponse"]
entry = {
"id": tr.get("id"),
"tool_type": tr.get("toolType"),
"response": tr.get("response"),
}
signature = part.get("thoughtSignature")
if signature is not None:
entry["thought_signature"] = signature
tool_responses_by_id[tr.get("id", "")] = entry
# Merge calls with their responses
for call_id, call_entry in tool_calls_by_id.items():
merged = dict(call_entry)
resp = tool_responses_by_id.pop(call_id, None)
if resp is not None:
merged["response"] = resp.get("response")
# Keep response signature if call didn't have one
if "thought_signature" not in merged and "thought_signature" in resp:
merged["thought_signature"] = resp["thought_signature"]
invocations.append(merged)
# Any orphan responses (shouldn't happen, but be safe)
for resp_id, resp_entry in tool_responses_by_id.items():
invocations.append(resp_entry)
return invocations if invocations else None
def _extract_image_response_from_parts(
self, parts: List[HttpxPartType]
) -> Optional[List[ImageURLListItem]]:
@ -2018,6 +2082,7 @@ class VertexGeminiConfig(VertexAIBaseConfig, BaseConfig):
thinking_blocks: Optional[List[ChatCompletionThinkingBlock]] = None
reasoning_content: Optional[str] = None
thought_signatures: Optional[Any] = None
server_side_tool_invocations: Optional[List[Dict[str, Any]]] = None
for idx, candidate in enumerate(_candidates):
if "content" not in candidate:
@ -2068,6 +2133,13 @@ class VertexGeminiConfig(VertexAIBaseConfig, BaseConfig):
)
)
# Extract server-side tool invocations (context circulation)
server_side_tool_invocations = (
VertexGeminiConfig._extract_server_side_tool_invocations(
parts=candidate["content"]["parts"]
)
)
if audio_response is not None:
cast(Dict[str, Any], chat_completion_message)[
"audio"
@ -2139,6 +2211,12 @@ class VertexGeminiConfig(VertexAIBaseConfig, BaseConfig):
chat_completion_message["provider_specific_fields"] = {}
chat_completion_message["provider_specific_fields"]["thought_signatures"] = thought_signatures # type: ignore
# Store server-side tool invocations in provider_specific_fields
if server_side_tool_invocations is not None:
if "provider_specific_fields" not in chat_completion_message:
chat_completion_message["provider_specific_fields"] = {}
chat_completion_message["provider_specific_fields"]["server_side_tool_invocations"] = server_side_tool_invocations # type: ignore
if isinstance(model_response, ModelResponseStream):
choice = VertexGeminiConfig._create_streaming_choice(
chat_completion_message=chat_completion_message,

View file

@ -244,8 +244,9 @@ class Tools(TypedDict, total=False):
retrieval: Retrieval
class ToolConfig(TypedDict):
class ToolConfig(TypedDict, total=False):
functionCallingConfig: FunctionCallingConfig
includeServerSideToolInvocations: bool
class TTL(TypedDict, total=False):

View file

@ -0,0 +1,234 @@
"""
Tests for Gemini context circulation (server-side tool invocations).
When includeServerSideToolInvocations=true is set, Gemini returns toolCall/toolResponse
parts for server-side tools (e.g. Google Search). These must be:
1. Extracted from the response into provider_specific_fields["server_side_tool_invocations"]
2. Re-injected as raw toolCall/toolResponse parts when converting messages back to Gemini format
3. The includeServerSideToolInvocations flag must be passed through to toolConfig
"""
import json
from typing import Any, Dict, List
from unittest.mock import MagicMock
import pytest
from litellm.llms.vertex_ai.gemini.vertex_and_google_ai_studio_gemini import (
VertexGeminiConfig,
)
from litellm.llms.vertex_ai.gemini.transformation import (
_gemini_convert_messages_with_history,
)
from litellm.types.llms.vertex_ai import HttpxPartType
# --- Response extraction tests ---
class TestExtractServerSideToolInvocations:
"""Test _extract_server_side_tool_invocations from response parts."""
def test_extracts_tool_call_and_response(self):
"""Basic case: one toolCall + one toolResponse with same id."""
parts: List[HttpxPartType] = [
{
"thoughtSignature": "sig_call_1",
"toolCall": {
"toolType": "GOOGLE_SEARCH_WEB",
"id": "abc123",
"args": {"queries": ["weather Buenos Aires"]},
},
},
{
"thoughtSignature": "sig_resp_1",
"toolResponse": {
"toolType": "GOOGLE_SEARCH_WEB",
"id": "abc123",
"response": {"weather": "Sunny, 20°C"},
},
},
{
"text": "The weather in Buenos Aires is sunny.",
"thoughtSignature": "sig_text",
},
]
result = VertexGeminiConfig._extract_server_side_tool_invocations(parts)
assert result is not None
assert len(result) == 1
assert result[0]["tool_type"] == "GOOGLE_SEARCH_WEB"
assert result[0]["id"] == "abc123"
assert result[0]["args"] == {"queries": ["weather Buenos Aires"]}
assert result[0]["response"] == {"weather": "Sunny, 20°C"}
assert result[0]["thought_signature"] == "sig_call_1"
def test_returns_none_when_no_server_side_tools(self):
"""No toolCall/toolResponse parts → returns None."""
parts: List[HttpxPartType] = [
{"text": "Hello world", "thoughtSignature": "sig1"},
{
"functionCall": {
"name": "get_weather",
"args": {"location": "Paris"},
},
"thoughtSignature": "sig2",
},
]
result = VertexGeminiConfig._extract_server_side_tool_invocations(parts)
assert result is None
def test_multiple_server_side_invocations(self):
"""Multiple toolCall/toolResponse pairs."""
parts: List[HttpxPartType] = [
{
"toolCall": {
"toolType": "GOOGLE_SEARCH_WEB",
"id": "search1",
"args": {"queries": ["query1"]},
},
"thoughtSignature": "sig1",
},
{
"toolResponse": {"toolType": "GOOGLE_SEARCH_WEB", "id": "search1", "response": "result1"},
"thoughtSignature": "sig2",
},
{
"toolCall": {
"toolType": "GOOGLE_SEARCH_WEB",
"id": "search2",
"args": {"queries": ["query2"]},
},
"thoughtSignature": "sig3",
},
{
"toolResponse": {"toolType": "GOOGLE_SEARCH_WEB", "id": "search2", "response": "result2"},
"thoughtSignature": "sig4",
},
]
result = VertexGeminiConfig._extract_server_side_tool_invocations(parts)
assert result is not None
assert len(result) == 2
assert result[0]["id"] == "search1"
assert result[0]["response"] == "result1"
assert result[1]["id"] == "search2"
assert result[1]["response"] == "result2"
def test_tool_call_without_response(self):
"""toolCall without matching toolResponse is still captured."""
parts: List[HttpxPartType] = [
{
"toolCall": {
"toolType": "CODE_EXECUTION",
"id": "exec1",
"args": {"code": "print('hello')"},
},
},
]
result = VertexGeminiConfig._extract_server_side_tool_invocations(parts)
assert result is not None
assert len(result) == 1
assert result[0]["id"] == "exec1"
assert "response" not in result[0]
# --- Input re-injection tests ---
class TestReInjectServerSideToolInvocations:
"""Test that server_side_tool_invocations are re-injected into Gemini parts."""
def test_roundtrip_single_invocation(self):
"""Server-side invocations from assistant message are converted back to Gemini parts."""
messages = [
{"role": "user", "content": "What's the weather?"},
{
"role": "assistant",
"content": "It's sunny in Buenos Aires.",
"provider_specific_fields": {
"server_side_tool_invocations": [
{
"tool_type": "GOOGLE_SEARCH_WEB",
"id": "abc123",
"args": {"queries": ["weather Buenos Aires"]},
"response": {"weather": "Sunny, 20°C"},
"thought_signature": "sig_abc",
}
]
},
},
{"role": "user", "content": "Thanks!"},
]
contents = _gemini_convert_messages_with_history(messages)
# Find the model turn
model_turn = [c for c in contents if c["role"] == "model"]
assert len(model_turn) == 1
parts = model_turn[0]["parts"]
# Should have: text part + toolCall part + toolResponse part
tool_call_parts = [p for p in parts if "toolCall" in p]
tool_response_parts = [p for p in parts if "toolResponse" in p]
assert len(tool_call_parts) == 1
assert tool_call_parts[0]["toolCall"]["toolType"] == "GOOGLE_SEARCH_WEB"
assert tool_call_parts[0]["toolCall"]["id"] == "abc123"
assert tool_call_parts[0]["toolCall"]["args"] == {"queries": ["weather Buenos Aires"]}
assert tool_call_parts[0]["thoughtSignature"] == "sig_abc"
assert len(tool_response_parts) == 1
assert tool_response_parts[0]["toolResponse"]["id"] == "abc123"
assert tool_response_parts[0]["toolResponse"]["toolType"] == "GOOGLE_SEARCH_WEB"
assert tool_response_parts[0]["toolResponse"]["response"] == {"weather": "Sunny, 20°C"}
def test_no_invocations_no_extra_parts(self):
"""Without server_side_tool_invocations, no extra parts are added."""
messages = [
{"role": "user", "content": "Hello"},
{"role": "assistant", "content": "Hi there!"},
{"role": "user", "content": "Bye"},
]
contents = _gemini_convert_messages_with_history(messages)
model_turn = [c for c in contents if c["role"] == "model"]
assert len(model_turn) == 1
parts = model_turn[0]["parts"]
assert len(parts) == 1
assert "text" in parts[0]
assert "toolCall" not in parts[0]
# --- toolConfig flag tests ---
class TestIncludeServerSideToolInvocationsConfig:
"""Test that the flag is passed through to toolConfig."""
def test_flag_added_to_tool_config(self):
"""include_server_side_tool_invocations=True should be mapped to optional_params."""
config = VertexGeminiConfig()
non_default_params = {"include_server_side_tool_invocations": True}
optional_params: Dict[str, Any] = {}
result = config.map_openai_params(
non_default_params=non_default_params,
optional_params=optional_params,
model="gemini-3-flash-preview",
drop_params=False,
)
assert result["include_server_side_tool_invocations"] is True
def test_flag_in_supported_params(self):
"""include_server_side_tool_invocations should be in supported params."""
config = VertexGeminiConfig()
supported = config.get_supported_openai_params(model="gemini-3-flash-preview")
assert "include_server_side_tool_invocations" in supported