diff --git a/litellm/llms/custom_httpx/llm_http_handler.py b/litellm/llms/custom_httpx/llm_http_handler.py index b9ada079f6b..1a73002f1b4 100644 --- a/litellm/llms/custom_httpx/llm_http_handler.py +++ b/litellm/llms/custom_httpx/llm_http_handler.py @@ -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 diff --git a/litellm/llms/gemini/realtime/transformation.py b/litellm/llms/gemini/realtime/transformation.py index 4378db06358..4e278261154 100644 --- a/litellm/llms/gemini/realtime/transformation.py +++ b/litellm/llms/gemini/realtime/transformation.py @@ -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: """ diff --git a/litellm/llms/vertex_ai/realtime/transformation.py b/litellm/llms/vertex_ai/realtime/transformation.py index 2b4746b174e..41b48d23b80 100644 --- a/litellm/llms/vertex_ai/realtime/transformation.py +++ b/litellm/llms/vertex_ai/realtime/transformation.py @@ -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 ) diff --git a/tests/test_litellm/llms/gemini/realtime/test_gemini_realtime_transformation.py b/tests/test_litellm/llms/gemini/realtime/test_gemini_realtime_transformation.py index cc0a32d2ce6..8291d62c99b 100644 --- a/tests/test_litellm/llms/gemini/realtime/test_gemini_realtime_transformation.py +++ b/tests/test_litellm/llms/gemini/realtime/test_gemini_realtime_transformation.py @@ -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