scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-05 02:41:56 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 49

Add tool calling support for gemini and vertex ai live api

Browse source
This commit is contained in:
Sameer Kankute 2026-04-27 16:56:38 +05:30
parent 2b8b614120
commit d3e375443a
No known key found for this signature in database
4 changed files with 441 additions and 48 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

16
litellm/llms/custom_httpx/llm_http_handler.py
View file

@ -5155,6 +5155,22 @@ class BaseLLMHTTPHandler:
)
if _session_config:
realtime_streaming.session_configuration_request = _session_config
# For Gemini/Vertex AI: if no initial setup sent (waiting for client's session.update),
# send synthetic session.created to unblock client
if not provider_config.requires_session_configuration() and hasattr(
provider_config, 'transform_session_created_event'
):
synthetic_session = provider_config.transform_session_created_event(
model=model,
logging_session_id=logging_obj.litellm_trace_id,
session_configuration_request=None,
)
await websocket.send_text(json.dumps(synthetic_session))
verbose_logger.debug(
"Sent synthetic session.created to client to unblock connection"
)
await realtime_streaming.bidirectional_forward()
except websockets.exceptions.InvalidStatusCode as e: # type: ignore

208
litellm/llms/gemini/realtime/transformation.py
View file

@ -40,6 +40,7 @@ from litellm.types.llms.openai import (
OpenAIRealtimeStreamSession,
OpenAIRealtimeStreamSessionEvents,
OpenAIRealtimeTurnDetection,
ResponsesAPIStreamEvents,
)
from litellm.types.llms.vertex_ai import (
GeminiResponseModalities,
@ -56,11 +57,12 @@ from litellm.utils import get_empty_usage
from ..common_utils import encode_unserializable_types, get_api_key_from_env
MAP_GEMINI_FIELD_TO_OPENAI_EVENT: Dict[str, OpenAIRealtimeEventTypes] = {
MAP_GEMINI_FIELD_TO_OPENAI_EVENT: Dict[str, Union[OpenAIRealtimeEventTypes, ResponsesAPIStreamEvents]] = {
"setupComplete": OpenAIRealtimeEventTypes.SESSION_CREATED,
"serverContent.generationComplete": OpenAIRealtimeEventTypes.RESPONSE_TEXT_DONE,
"serverContent.turnComplete": OpenAIRealtimeEventTypes.RESPONSE_DONE,
"serverContent.interrupted": OpenAIRealtimeEventTypes.RESPONSE_DONE,
"toolCall": ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DONE,
}
@ -190,7 +192,8 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
)
vertex_gemini_config = VertexGeminiConfig()
optional_params["generationConfig"]["tools"] = (
# Tools should be at the top level of setup, not inside generationConfig
optional_params["tools"] = (
vertex_gemini_config._map_function(
value=value, optional_params=optional_params
)
@ -214,6 +217,107 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
optional_params.pop("generationConfig")
return optional_params
def _handle_session_update(
self,
json_message: dict,
model: str,
session_configuration_request: Optional[str],
) -> List[str]:
"""
Handle session.update by sending setup to Gemini.
Only sends setup on the FIRST session.update (when session_configuration_request is None).
Subsequent session.update messages are ignored because Gemini doesn't support dynamic updates.
"""
if session_configuration_request is None:
# First session.update - send the setup with all configuration
client_session_configuration_request = self.map_openai_params(
optional_params={}, non_default_params=json_message["session"]
)
client_session_configuration_request["model"] = f"models/{model}"
gemini_setup_msg = json.dumps({"setup": client_session_configuration_request})
verbose_logger.debug(
f"Gemini Realtime: Sending initial setup with tools to backend"
)
return [gemini_setup_msg]
else:
# Subsequent session.update - ignore (Gemini doesn't support dynamic updates)
verbose_logger.debug(
"Gemini Realtime: Ignoring session.update (setup already sent)"
)
return []
def _handle_conversation_item(self, json_message: dict) -> List[str]:
"""
Handle conversation.item.create for user text or function call output.
Converts OpenAI format to Gemini's clientContent (for user text) or
toolResponse (for function outputs).
"""
item = json_message.get("item", {})
item_type = item.get("type")
# Handle function call output (tool response)
if item_type == "function_call_output":
return self._handle_function_call_output(item)
# Handle regular text content
return self._handle_user_text_content(item)
def _handle_function_call_output(self, item: dict) -> List[str]:
"""Transform function_call_output to Gemini toolResponse format."""
call_id = item.get("call_id", "")
output = item.get("output", "{}")
verbose_logger.debug(f"Gemini Realtime: Transforming function_call_output for call_id={call_id}")
# Parse the output to get the result
try:
output_dict = json.loads(output) if isinstance(output, str) else output
except json.JSONDecodeError:
output_dict = {"result": output}
# Build Gemini toolResponse format
tool_response_message = {
"toolResponse": {
"functionResponses": [
{
"id": call_id,
"response": output_dict,
}
]
}
}
return [json.dumps(tool_response_message)]
def _handle_user_text_content(self, item: dict) -> List[str]:
"""Transform user text content to Gemini clientContent format."""
content_list = item.get("content", [])
text_parts = [
c.get("text", "")
for c in content_list
if isinstance(c, dict) and c.get("type") == "input_text"
]
text = " ".join(filter(None, text_parts))
if not text:
return []
# Build clientContent message with turns (proper Gemini Live API format)
client_content_message = {
"clientContent": {
"turns": [
{
"role": "user",
"parts": [{"text": text}]
}
],
"turnComplete": True
}
}
return [json.dumps(client_content_message)]
def transform_realtime_request(
self,
message: str,
@ -233,14 +337,11 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
messages: List[str] = []
msg_type = json_message.get("type")
## HANDLE SESSION UPDATE — translate to Gemini setup; no realtime_input needed ##
## HANDLE SESSION UPDATE — translate to Gemini setup ##
if msg_type == "session.update":
client_session_configuration_request = self.map_openai_params(
optional_params={}, non_default_params=json_message["session"]
return self._handle_session_update(
json_message, model, session_configuration_request
)
client_session_configuration_request["model"] = f"models/{model}"
messages.append(json.dumps({"setup": client_session_configuration_request}))
return messages
## HANDLE response.create — Gemini responds automatically; nothing to forward ##
if msg_type == "response.create":
@ -251,38 +352,30 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
realtime_input_dict["audio"] = HttpxBlobType(
mimeType=self.get_audio_mime_type(), data=json_message["audio"]
)
## HANDLE conversation.item.create — extract actual user text ##
## HANDLE conversation.item.create — extract user text or function call output ##
elif msg_type == "conversation.item.create":
item = json_message.get("item", {})
content_list = item.get("content", [])
text_parts = [
c.get("text", "")
for c in content_list
if isinstance(c, dict) and c.get("type") == "input_text"
]
text = " ".join(filter(None, text_parts))
if not text:
return []
realtime_input_dict["text"] = text
return self._handle_conversation_item(json_message)
## HANDLE INPUT AUDIO BUFFER - use realtimeInput for audio streaming ##
elif msg_type == "input_audio_buffer.append":
realtime_input_dict["audio"] = HttpxBlobType(
mimeType=self.get_audio_mime_type(), data=json_message["audio"]
)
realtime_input_dict = cast(
BidiGenerateContentRealtimeInput,
encode_unserializable_types(cast(Dict[str, object], realtime_input_dict)),
)
gemini_msg = json.dumps({"realtimeInput": realtime_input_dict})
verbose_logger.debug("Gemini Realtime: Sending audio realtimeInput to backend")
messages.append(gemini_msg)
return messages
else:
# Unknown/unsupported OpenAI event type — drop silently rather than
# forwarding raw JSON as text input to the model.
return []
if len(realtime_input_dict) != 1:
raise ValueError(
f"Only one argument can be set, got {len(realtime_input_dict)}:"
f" {list(realtime_input_dict.keys())}"
)
realtime_input_dict = cast(
BidiGenerateContentRealtimeInput,
encode_unserializable_types(cast(Dict[str, object], realtime_input_dict)),
)
messages.append(json.dumps({"realtime_input": realtime_input_dict}))
return messages
def transform_session_created_event(
self,
model: str,
@ -576,6 +669,32 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
returned_items.append(response_output_item_done)
return returned_items
def transform_tool_call_events(
self,
tool_call_message: dict,
) -> List[Dict[str, Any]]:
"""
Transform Gemini toolCall message to OpenAI function call events.
Converts Gemini's functionCalls format to OpenAI's response.function_call_arguments.done events.
"""
function_calls = tool_call_message.get("functionCalls", [])
verbose_logger.debug(
f"Gemini Realtime: Transforming {len(function_calls)} tool call(s) to OpenAI format"
)
return [
{
"type": "response.function_call_arguments.done",
"event_id": f"event_{uuid.uuid4()}",
"call_id": fc.get("id", ""),
"name": fc.get("name", ""),
"arguments": json.dumps(fc.get("args", {})),
}
for fc in function_calls
]
@staticmethod
def get_nested_value(obj: dict, path: str) -> Any:
keys = path.split(".")
@ -811,10 +930,10 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
value: dict,
current_delta_type: Optional[ALL_DELTA_TYPES],
json_message: dict,
) -> OpenAIRealtimeEventTypes:
) -> Union[OpenAIRealtimeEventTypes, ResponsesAPIStreamEvents]:
model_turn_event = value.get("modelTurn")
generation_complete_event = value.get("generationComplete")
openai_event: Optional[OpenAIRealtimeEventTypes] = None
openai_event: Optional[Union[OpenAIRealtimeEventTypes, ResponsesAPIStreamEvents]] = None
if model_turn_event: # check if model turn event
openai_event = self.map_model_turn_event(model_turn_event)
elif generation_complete_event:
@ -854,6 +973,8 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
message_str = str(message)
raise ValueError(f"Invalid JSON message: {message_str}")
verbose_logger.debug(f"Realtime Response Transform: Gemini message={json.dumps(json_message)[:500]}")
logging_session_id = logging_obj.litellm_trace_id
current_output_item_id = realtime_response_transform_input[
@ -949,6 +1070,10 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
)
session_configuration_request = json.dumps(transformed_message)
returned_message.append(transformed_message)
elif openai_event == ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DONE:
# Handle toolCall from Gemini
tool_call_events = self.transform_tool_call_events(value)
returned_message.extend(tool_call_events)
elif openai_event == OpenAIRealtimeEventTypes.RESPONSE_DONE:
transformed_response_done_event = self.transform_response_done_event(
message=BidiGenerateContentServerMessage(**json_message), # type: ignore
@ -993,6 +1118,12 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
transformed_message=returned_message,
current_item_chunks=current_item_chunks,
)
# Log the transformed events
for msg in returned_message:
event_type = msg.get("type") if isinstance(msg, dict) else "unknown"
verbose_logger.debug(f"Realtime Response Transform: OpenAI event={event_type}, data={json.dumps(msg)[:500] if isinstance(msg, dict) else str(msg)[:500]}")
return {
"response": returned_message,
"current_output_item_id": current_output_item_id,
@ -1005,7 +1136,10 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
}
def requires_session_configuration(self) -> bool:
return True
# Return False so we DON'T auto-send setup on connection
# Instead, setup will be sent when client sends session.update
# This allows us to include tools, instructions, etc. in the FIRST setup
return False
def session_configuration_request(self, model: str) -> str:
"""

65
litellm/llms/vertex_ai/realtime/transformation.py
View file

@ -14,6 +14,7 @@ Auth: OAuth2 Bearer token (not an API key).
import json
from typing import List, Optional
from litellm import verbose_logger
from litellm.llms.gemini.realtime.transformation import GeminiRealtimeConfig
@ -138,6 +139,34 @@ class VertexAIRealtimeConfig(GeminiRealtimeConfig):
# Request translation
# ------------------------------------------------------------------
def _build_vertex_ai_setup_config(self, model: str, session_params: dict) -> dict:
"""Build Vertex AI setup configuration with proper model path and defaults."""
setup_config = self.map_openai_params(
optional_params={}, non_default_params=session_params
)
# Use full Vertex AI model path
setup_config["model"] = (
f"projects/{self._project}"
f"/locations/{self._location}"
f"/publishers/google/models/{model}"
)
# Add Vertex AI specific defaults if not provided
setup_config.setdefault(
"realtimeInputConfig",
{
"automaticActivityDetection": {
"disabled": False,
"silenceDurationMs": 800,
}
},
)
setup_config.setdefault("inputAudioTranscription", {})
setup_config.setdefault("outputAudioTranscription", {})
return setup_config
def transform_realtime_request(
self,
message: str,
@ -146,19 +175,33 @@ class VertexAIRealtimeConfig(GeminiRealtimeConfig):
) -> List[str]:
"""
Translate OpenAI realtime client messages to Vertex AI format.
``session.update`` is intentionally ignored (returns []) because
Vertex AI only accepts a single ``setup`` message at the start of
the connection — sending a second one causes a 1007 close error.
The initial setup (sent automatically before bidirectional_forward)
already includes AUDIO modality and server VAD, so there is nothing
more to configure.
Handles session.update by sending setup with proper Vertex AI model path.
"""
json_message = json.loads(message)
if json_message.get("type") == "session.update":
# Do not forward as a second setup — Vertex AI rejects it.
return []
msg_type = json_message.get("type")
# Handle session.update with Vertex AI specific model path
if msg_type == "session.update":
if session_configuration_request is None:
# First session.update - send the setup with Vertex AI configuration
setup_config = self._build_vertex_ai_setup_config(
model, json_message["session"]
)
gemini_setup_msg = json.dumps({"setup": setup_config})
verbose_logger.debug(
"Vertex AI Realtime: Sending initial setup with tools to backend"
)
return [gemini_setup_msg]
else:
# Subsequent session.update - ignore
verbose_logger.debug(
"Vertex AI Realtime: Ignoring session.update (setup already sent)"
)
return []
# For other message types, use parent's logic
return super().transform_realtime_request(
message, model, session_configuration_request
)

200
tests/test_litellm/llms/gemini/realtime/test_gemini_realtime_transformation.py
View file

@ -242,3 +242,203 @@ def test_gemini_3_1_flash_live_preview_model_cost_map_entry():
assert info.get("max_output_tokens") == 65536
assert "video" in info.get("supported_modalities", [])
assert info.get("supports_function_calling") is True
def test_gemini_realtime_tool_call_transformation():
"""Test transformation of Gemini toolCall to OpenAI function_call_arguments.done format."""
config = GeminiRealtimeConfig()
# Gemini toolCall message format
gemini_tool_call = {
"toolCall": {
"functionCalls": [
{
"id": "call_123",
"name": "get_weather",
"args": {"location": "San Francisco", "unit": "fahrenheit"},
}
]
}
}
gemini_tool_call_str = json.dumps(gemini_tool_call)
logging_obj = MagicMock()
logging_obj.litellm_trace_id = "test-trace-123"
# Transform the toolCall message
result = config.transform_realtime_response(
gemini_tool_call_str,
"gemini-2.5-flash",
logging_obj,
realtime_response_transform_input={
"session_configuration_request": None,
"current_output_item_id": "item_123",
"current_response_id": "resp_123",
"current_conversation_id": None,
"current_delta_chunks": [],
"current_item_chunks": [],
"current_delta_type": None,
},
)
print("Tool call transformation result:", json.dumps(result, indent=2))
# Verify the transformation
responses = result["response"]
assert len(responses) > 0, "Expected at least one response event"
# Find the function_call_arguments.done event
function_call_event = None
for event in responses:
if event.get("type") == "response.function_call_arguments.done":
function_call_event = event
break
assert function_call_event is not None, "Expected function_call_arguments.done event"
assert function_call_event["call_id"] == "call_123"
assert function_call_event["name"] == "get_weather"
# Verify arguments are properly serialized as JSON string
args = json.loads(function_call_event["arguments"])
assert args["location"] == "San Francisco"
assert args["unit"] == "fahrenheit"
def test_gemini_realtime_session_update_with_tools():
"""Test transformation of OpenAI session.update with tools to Gemini setup format."""
config = GeminiRealtimeConfig()
# OpenAI format session update with tools
session_update = {
"type": "session.update",
"session": {
"instructions": "You are a helpful assistant with weather tools.",
"temperature": 0.7,
"max_response_output_tokens": 1024,
"modalities": ["audio"],
"tools": [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Get the current weather for a location.",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "The city name",
},
"unit": {
"type": "string",
"enum": ["fahrenheit", "celsius"],
},
},
"required": ["location"],
},
},
}
],
},
}
# Transform to Gemini format (first session.update, so setup should be sent)
messages = config.transform_realtime_request(
json.dumps(session_update), "gemini-2.5-flash", session_configuration_request=None
)
assert len(messages) == 1, "Expected one setup message"
gemini_setup = json.loads(messages[0])
assert "setup" in gemini_setup
setup_config = gemini_setup["setup"]
# Verify tools are at top level, not in generationConfig
assert "tools" in setup_config
assert "tools" not in setup_config.get("generationConfig", {})
# Verify tool structure matches Gemini format
tools = setup_config["tools"]
assert len(tools) == 1
assert "function_declarations" in tools[0]
function_decl = tools[0]["function_declarations"][0]
assert function_decl["name"] == "get_weather"
assert "Get the current weather" in function_decl["description"]
assert "parameters" in function_decl
def test_gemini_realtime_function_call_output_transformation():
"""Test transformation of OpenAI function_call_output to Gemini toolResponse format."""
config = GeminiRealtimeConfig()
# OpenAI format function call output
function_output = {
"type": "conversation.item.create",
"item": {
"type": "function_call_output",
"call_id": "call_123",
"output": json.dumps({
"location": "San Francisco",
"temperature": 72,
"unit": "fahrenheit",
"conditions": "sunny",
}),
},
}
# Transform to Gemini format
messages = config.transform_realtime_request(
json.dumps(function_output), "gemini-2.5-flash", session_configuration_request="existing"
)
assert len(messages) == 1, "Expected one toolResponse message"
gemini_response = json.loads(messages[0])
assert "toolResponse" in gemini_response
tool_response = gemini_response["toolResponse"]
assert "functionResponses" in tool_response
assert len(tool_response["functionResponses"]) == 1
func_response = tool_response["functionResponses"][0]
assert func_response["id"] == "call_123"
assert "response" in func_response
assert func_response["response"]["temperature"] == 72
assert func_response["response"]["conditions"] == "sunny"
def test_gemini_realtime_user_text_transformation():
"""Test transformation of OpenAI user message to Gemini clientContent format."""
config = GeminiRealtimeConfig()
# OpenAI format user message
user_message = {
"type": "conversation.item.create",
"item": {
"type": "message",
"role": "user",
"content": [{"type": "input_text", "text": "What's the weather in London?"}],
},
}
# Transform to Gemini format
messages = config.transform_realtime_request(
json.dumps(user_message), "gemini-2.5-flash", session_configuration_request="existing"
)
assert len(messages) == 1, "Expected one clientContent message"
gemini_message = json.loads(messages[0])
assert "clientContent" in gemini_message
client_content = gemini_message["clientContent"]
assert "turns" in client_content
assert len(client_content["turns"]) == 1
turn = client_content["turns"][0]
assert turn["role"] == "user"
assert len(turn["parts"]) == 1
assert turn["parts"][0]["text"] == "What's the weather in London?"
assert client_content["turnComplete"] is True