fix: Improve Gemini Live model provider

src/strands/experimental/bidirectional_streaming/models/gemini_live.py

@@ -33,6 +33,7 @@
     BidiImageInputEvent,
     BidiInputEvent,
     BidiInterruptionEvent,
+    BidiOutputEvent,
     BidiUsageEvent,
     BidiTextInputEvent,
     BidiTranscriptStreamEvent,
@@ -59,7 +60,7 @@ class BidiGeminiLiveModel(BidiModel):
 
     def __init__(
         self,
-        model_id: str = "models/gemini-2.0-flash-live-preview-04-09",
+        model_id: str = "gemini-2.5-flash-native-audio-preview-09-2025",
         api_key: Optional[str] = None,
         live_config: Optional[Dict[str, Any]] = None,
         **kwargs
@@ -75,7 +76,19 @@ def __init__(
         # Model configuration
         self.model_id = model_id
         self.api_key = api_key
-        self.live_config = live_config or {}
+
+        # Set default live_config with transcription enabled
+        default_config = {
+            "response_modalities": ["AUDIO"],
+            "outputAudioTranscription": {},  # Enable output transcription by default
+            "inputAudioTranscription": {}    # Enable input transcription by default
+        }
+
+        # Merge user config with defaults (user config takes precedence)
+        if live_config:
+            default_config.update(live_config)
+
+        self.live_config = default_config
 
         # Create Gemini client with proper API version
         client_kwargs = {}
@@ -161,7 +174,7 @@ async def _send_message_history(self, messages: Messages) -> None:
                 content = genai_types.Content(role=role, parts=content_parts)
                 await self.live_session.send_client_content(turns=content)
 
-    async def receive(self) -> AsyncIterable[Dict[str, Any]]:
+    async def receive(self) -> AsyncIterable[BidiOutputEvent]:
         """Receive Gemini Live API events and convert to provider-agnostic format."""
 
         # Emit connection start event
@@ -178,10 +191,9 @@ async def receive(self) -> AsyncIterable[Dict[str, Any]]:
                         if not self._active:
                             break
 
-                        # Convert to provider-agnostic format
-                        provider_event = self._convert_gemini_live_event(message)
-                        if provider_event:
-                            yield provider_event
+                        # Convert to provider-agnostic format (always returns list)
+                        for event in self._convert_gemini_live_event(message):
+                            yield event
 
                     # SDK exits receive loop after turn_complete - restart automatically
                     if self._active:
@@ -199,19 +211,22 @@ async def receive(self) -> AsyncIterable[Dict[str, Any]]:
             # Emit connection close event when exiting
             yield BidiConnectionCloseEvent(connection_id=self.connection_id, reason="complete")
 
-    def _convert_gemini_live_event(self, message: LiveServerMessage) -> Optional[Dict[str, Any]]:
+    def _convert_gemini_live_event(self, message: LiveServerMessage) -> List[BidiOutputEvent]:
         """Convert Gemini Live API events to provider-agnostic format.
 
         Handles different types of content:
         - inputTranscription: User's speech transcribed to text
         - outputTranscription: Model's audio transcribed to text
         - modelTurn text: Text response from the model
         - usageMetadata: Token usage information
+
+        Returns:
+            List of event dicts (empty list if no events to emit).
         """
         try:
             # Handle interruption first (from server_content)
             if message.server_content and message.server_content.interrupted:
-                return BidiInterruptionEvent(reason="user_speech")
+                return [BidiInterruptionEvent(reason="user_speech")]
 
             # Handle input transcription (user's speech) - emit as transcript event
             if message.server_content and message.server_content.input_transcription:
@@ -221,13 +236,13 @@ def _convert_gemini_live_event(self, message: LiveServerMessage) -> Optional[Dic
                     transcription_text = input_transcript.text
                     role = getattr(input_transcript, 'role', 'user')
                     logger.debug(f"Input transcription detected: {transcription_text}")
-                    return BidiTranscriptStreamEvent(
+                    return [BidiTranscriptStreamEvent(
                         delta={"text": transcription_text},
                         text=transcription_text,
                         role=role.lower() if isinstance(role, str) else "user",
                         is_final=True,
                         current_transcript=transcription_text
-                    )
+                    )]
 
             # Handle output transcription (model's audio) - emit as transcript event
             if message.server_content and message.server_content.output_transcription:
@@ -237,50 +252,65 @@ def _convert_gemini_live_event(self, message: LiveServerMessage) -> Optional[Dic
                     transcription_text = output_transcript.text
                     role = getattr(output_transcript, 'role', 'assistant')
                     logger.debug(f"Output transcription detected: {transcription_text}")
-                    return BidiTranscriptStreamEvent(
+                    return [BidiTranscriptStreamEvent(
                         delta={"text": transcription_text},
                         text=transcription_text,
                         role=role.lower() if isinstance(role, str) else "assistant",
                         is_final=True,
                         current_transcript=transcription_text
-                    )
-
-            # Handle text output from model
-            if message.text:
-                role = getattr(message, 'role', 'assistant')
-                logger.debug(f"Text output as transcript: {message.text}")
-                return BidiTranscriptStreamEvent(
-                    delta={"text": message.text},
-                    text=message.text,
-                    role=role.lower() if isinstance(role, str) else "assistant",
-                    is_final=True,
-                    current_transcript=message.text
-                )
+                    )]
 
             # Handle audio output using SDK's built-in data property
+            # Check this BEFORE text to avoid triggering warning on mixed content
             if message.data:
                 # Convert bytes to base64 string for JSON serializability
                 audio_b64 = base64.b64encode(message.data).decode('utf-8')
-                return BidiAudioStreamEvent(
+                return [BidiAudioStreamEvent(
                     audio=audio_b64,
                     format="pcm",
                     sample_rate=GEMINI_OUTPUT_SAMPLE_RATE,
                     channels=GEMINI_CHANNELS
-                )
+                )]
+
+            # Handle text output from model_turn (avoids warning by checking parts directly)
+            if message.server_content and message.server_content.model_turn:
+                model_turn = message.server_content.model_turn
+                if model_turn.parts:
+                    # Concatenate all text parts (Gemini may send multiple parts)
+                    text_parts = []
+                    for part in model_turn.parts:
+                        # Log all part types for debugging
+                        part_attrs = {attr: getattr(part, attr, None) for attr in dir(part) if not attr.startswith('_')}
+
+                        # Check if part has text attribute and it's not empty
+                        if hasattr(part, 'text') and part.text:
+                            text_parts.append(part.text)
+
+                    if text_parts:
+                        full_text = " ".join(text_parts)
+                        return [BidiTranscriptStreamEvent(
+                            delta={"text": full_text},
+                            text=full_text,
+                            role="assistant",
+                            is_final=True,
+                            current_transcript=full_text
+                        )]
 
-            # Handle tool calls
+            # Handle tool calls - return list to support multiple tool calls
             if message.tool_call and message.tool_call.function_calls:
+                tool_events = []
                 for func_call in message.tool_call.function_calls:
                     tool_use_event: ToolUse = {
                         "toolUseId": func_call.id,
                         "name": func_call.name,
                         "input": func_call.args or {}
                     }
-                    # Return ToolUseStreamEvent for consistency with standard agent
-                    return ToolUseStreamEvent(
+                    # Create ToolUseStreamEvent for consistency with standard agent
+                    tool_events.append(ToolUseStreamEvent(
                         delta={"toolUse": tool_use_event},
                         current_tool_use=tool_use_event
-                    )
+                    ))
+                return tool_events
 
             # Handle usage metadata
             if hasattr(message, 'usage_metadata') and message.usage_metadata:
@@ -315,22 +345,23 @@ def _convert_gemini_live_event(self, message: LiveServerMessage) -> Optional[Dic
                                     "output_tokens": detail.token_count
                                 })
 
-                return BidiUsageEvent(
+                return [BidiUsageEvent(
                     input_tokens=usage.prompt_token_count or 0,
                     output_tokens=usage.response_token_count or 0,
                     total_tokens=usage.total_token_count or 0,
                     modality_details=modality_details if modality_details else None,
                     cache_read_input_tokens=usage.cached_content_token_count if usage.cached_content_token_count else None
-                )
+                )]
 
             # Silently ignore setup_complete and generation_complete messages
-            return None
+            return []
 
         except Exception as e:
             logger.error("Error converting Gemini Live event: %s", e)
             logger.error("Message type: %s", type(message).__name__)
             logger.error("Message attributes: %s", [attr for attr in dir(message) if not attr.startswith('_')])
-            return None
+            # Return ErrorEvent in list so caller can handle it
+            return [BidiErrorEvent(error=e)]
 
     async def send(
         self,

src/strands/experimental/bidirectional_streaming/scripts/test_gemini_live.py

@@ -41,9 +41,9 @@
 from strands.experimental.bidirectional_streaming.models.gemini_live import BidiGeminiLiveModel
 
 # Configure logging - debug only for Gemini Live, info for everything else
-logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+logging.basicConfig(level=logging.WARN, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
 gemini_logger = logging.getLogger('strands.experimental.bidirectional_streaming.models.gemini_live')
-gemini_logger.setLevel(logging.DEBUG)
+gemini_logger.setLevel(logging.WARN)
 logger = logging.getLogger(__name__)
 
 
@@ -162,18 +162,18 @@ async def receive(agent, context):
             # Handle interruption events (bidirectional_interruption)
             elif event_type == "bidirectional_interruption":
                 context["interrupted"] = True
-                logger.info("Interruption detected")
+                print("⚠️  Interruption detected")
 
             # Handle transcript events (bidirectional_transcript_stream)
             elif event_type == "bidirectional_transcript_stream":
                 transcript_text = event.get("text", "")
-                transcript_source = event.get("source", "unknown")
+                transcript_role = event.get("role", "unknown")
                 is_final = event.get("is_final", False)
 
                 # Print transcripts with special formatting
-                if transcript_source == "user":
+                if transcript_role == "user":
                     print(f"🎤 User: {transcript_text}")
-                elif transcript_source == "assistant":
+                elif transcript_role == "assistant":
                     print(f"🔊 Assistant: {transcript_text}")
 
             # Handle turn complete events (bidirectional_turn_complete)
@@ -313,17 +313,10 @@ async def main(duration=180):
     # Initialize Gemini Live model with proper configuration
     logger.info("Initializing Gemini Live model with API key")
 
-    model = BidiGeminiLiveModel(
-        model_id="gemini-2.5-flash-native-audio-preview-09-2025",
-        api_key=api_key,
-        live_config={
-            "response_modalities": ["AUDIO"],
-            "output_audio_transcription": {},  # Enable output transcription
-            "input_audio_transcription": {}    # Enable input transcription
-        }
-    )
+    # Use default model and config (includes transcription enabled by default)
+    model = BidiGeminiLiveModel(api_key=api_key)
     logger.info("Gemini Live model initialized successfully")
-    print("Using Gemini Live model")
+    print("Using Gemini Live model with default config (audio output + transcription enabled)")
 
     agent = BidirectionalAgent(
         model=model,