Add Gemini Model Provider

src/strands/experimental/bidirectional_streaming/agent/agent.py

@@ -21,8 +21,7 @@
 from ....types.content import Messages
 from ..event_loop.bidirectional_event_loop import start_bidirectional_connection, stop_bidirectional_connection
 from ..models.bidirectional_model import BidirectionalModel
-from ..types.bidirectional_streaming import AudioInputEvent, BidirectionalStreamEvent
-
+from ..types.bidirectional_streaming import AudioInputEvent, BidirectionalStreamEvent, ImageInputEvent
 
 logger = logging.getLogger(__name__)
 
@@ -81,18 +80,16 @@ async def start(self) -> None:
 
         logger.debug("Conversation start - initializing session")
         self._session = await start_bidirectional_connection(self)
-        logger.debug("Conversation ready")
-
-    async def send(self, input_data: str | AudioInputEvent) -> None:
-        """Send input to the model (text or audio).
-
-        Unified method for sending both text and audio input to the model during
-        an active conversation session. User input is automatically added to
-        conversation history for complete message tracking.
-
+
+    async def send(self, input_data: str | AudioInputEvent | ImageInputEvent) -> None:
+        """Send input to the model (text, audio, or image).
+
+        Unified method for sending text, audio, and image input to the model during
+        an active conversation session.
+
         Args:
-            input_data: Either a string for text input or AudioInputEvent for audio input.
-
+            input_data: String for text, AudioInputEvent for audio, or ImageInputEvent for images.
+            
         Raises:
             ValueError: If no active session or invalid input type.
         """
@@ -107,10 +104,14 @@ async def send(self, input_data: str | AudioInputEvent) -> None:
         elif isinstance(input_data, dict) and "audioData" in input_data:
             # Handle audio input
             await self._session.model_session.send_audio_content(input_data)
+        elif isinstance(input_data, dict) and "imageData" in input_data:
+            # Handle image input (ImageInputEvent)
+            await self._session.model_session.send_image_content(input_data)
         else:
             raise ValueError(
-                "Input must be either a string (text) or AudioInputEvent "
-                "(dict with audioData, format, sampleRate, channels)"
+                "Input must be either a string (text), AudioInputEvent "
+                "(dict with audioData, format, sampleRate, channels), or ImageInputEvent "
+                "(dict with imageData, mimeType, encoding)"
             )
 
     async def receive(self) -> AsyncIterable[BidirectionalStreamEvent]:

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

@@ -1,6 +1,14 @@
 """Bidirectional model interfaces and implementations."""
 
 from .bidirectional_model import BidirectionalModel, BidirectionalModelSession
+from .gemini_live import GeminiLiveBidirectionalModel, GeminiLiveSession
 from .novasonic import NovaSonicBidirectionalModel, NovaSonicSession
 
-__all__ = ["BidirectionalModel", "BidirectionalModelSession", "NovaSonicBidirectionalModel", "NovaSonicSession"]
+__all__ = [
+    "BidirectionalModel",
+    "BidirectionalModelSession",
+    "GeminiLiveBidirectionalModel",
+    "GeminiLiveSession",
+    "NovaSonicBidirectionalModel",
+    "NovaSonicSession",
+]

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

@@ -17,7 +17,7 @@
 
 from ....types.content import Messages
 from ....types.tools import ToolSpec
-from ..types.bidirectional_streaming import AudioInputEvent, BidirectionalStreamEvent
+from ..types.bidirectional_streaming import AudioInputEvent, BidirectionalStreamEvent, ImageInputEvent
 
 logger = logging.getLogger(__name__)
 
@@ -48,6 +48,15 @@ async def send_audio_content(self, audio_input: AudioInputEvent) -> None:
         """
         raise NotImplementedError
 
+    @abc.abstractmethod
+    async def send_image_content(self, image_input: ImageInputEvent) -> None:
+        """Send image content to the model during an active connection.
+
+        Handles image encoding and provider-specific formatting while presenting
+        a simple ImageInputEvent interface.
+        """
+        raise NotImplementedError
+
     @abc.abstractmethod
     async def send_text_content(self, text: str, **kwargs) -> None:
         """Send text content to the model during ongoing generation.