Add text input handling to Gemini multimodal service

- Updated `RealtimeInput` to include an optional `text` parameter.
- Introduced `TextInputMessage` class for encapsulating text input data.
- Implemented `_send_user_text` method to send text input to the Gemini Live API.
- Enhanced message processing to support text input alongside media chunks.

src/pipecat/services/gemini_multimodal_live/events.py

@@ -114,13 +114,15 @@ class RealtimeInputConfig(BaseModel):
 
 
 class RealtimeInput(BaseModel):
-    """Contains realtime input media chunks.
+    """Contains realtime input media chunks and text.
 
     Parameters:
         mediaChunks: List of media chunks for realtime processing.
+        text: Text for realtime processing.
     """
 
-    mediaChunks: List[MediaChunk]
+    mediaChunks: Optional[List[MediaChunk]] = None
+    text: Optional[str] = None
 
 
 class ClientContent(BaseModel):
@@ -190,6 +192,24 @@ class VideoInputMessage(BaseModel):
         )
 
 
+class TextInputMessage(BaseModel):
+    """Message containing text input data."""
+
+    realtimeInput: RealtimeInput
+
+    @classmethod
+    def from_text(cls, text: str) -> "TextInputMessage":
+        """Create a text input message from a string.
+
+        Args:
+            text: The text to send.
+
+        Returns:
+            A TextInputMessage instance.
+        """
+        return cls(realtimeInput=RealtimeInput(text=text))
+
+
 class ClientContentMessage(BaseModel):
     """Message containing client content for the API.
 

src/pipecat/services/gemini_multimodal_live/gemini.py

@@ -733,6 +733,8 @@ class GeminiMultimodalLiveLLMService(LLMService):
                 # Support just one tool call per context frame for now
                 tool_result_message = context.messages[-1]
                 await self._tool_result(tool_result_message)
+        elif isinstance(frame, LLMTextFrame):
+            await self._send_user_text(frame.text)
         elif isinstance(frame, InputAudioRawFrame):
             await self._send_user_audio(frame)
             await self.push_frame(frame, direction)
@@ -964,6 +966,17 @@ class GeminiMultimodalLiveLLMService(LLMService):
             length = int((frame.sample_rate * frame.num_channels * 2) * 0.5)
             self._user_audio_buffer = self._user_audio_buffer[-length:]
 
+    async def _send_user_text(self, text: str):
+        """Send user text to Gemini Live API."""
+        logger.debug(f"Sending text to Gemini: {text}")
+        evt = events.TextInputMessage.from_text(text)
+        await self.send_client_event(evt)
+        # After sending text, we need to signal that the turn is complete.
+        evt = events.ClientContentMessage.model_validate(
+            {"clientContent": {"turnComplete": True}}
+        )
+        await self.send_client_event(evt)
+
     async def _send_user_video(self, frame):
         """Send user video frame to Gemini Live API."""
         if self._video_input_paused: