diff --git a/src/google/adk/agents/live_request_queue.py b/src/google/adk/agents/live_request_queue.py
index c9c7b82c685..2241b053622 100644
--- a/src/google/adk/agents/live_request_queue.py
+++ b/src/google/adk/agents/live_request_queue.py
@@ -53,6 +53,14 @@ class LiveRequest(BaseModel):
   activity_end: Optional[types.ActivityEnd] = None
   """If set, signal the end of user activity to the model.
 
+  When multiple fields are set, they are processed by priority (highest first):
+  activity_start > activity_end > blob > content. state_delta, if set, is always
+  applied regardless of the other fields.
+  """
+  audio_stream_end: Optional[bool] = None
+  """If set, signal the end of the audio stream to force flush buffered audio
+  when VAD (voice activity detection) is enabled.
+
   When multiple fields are set, they are processed by priority (highest first):
   activity_start > activity_end > blob > content. state_delta, if set, is always
   applied regardless of the other fields.
@@ -92,6 +100,14 @@ def send_activity_end(self) -> None:
     """Sends an activity end signal to mark the end of user input."""
     self._queue.put_nowait(LiveRequest(activity_end=types.ActivityEnd()))
 
+  def send_audio_stream_end(self) -> None:
+    """Sends an audio stream end signal to force flush buffered audio.
+
+    Use this to signal the end of an audio stream when VAD is enabled,
+    which causes the model to process any buffered audio immediately.
+    """
+    self._queue.put_nowait(LiveRequest(audio_stream_end=True))
+
   def send(self, req: LiveRequest) -> None:
     self._queue.put_nowait(req)
 
diff --git a/src/google/adk/flows/llm_flows/base_llm_flow.py b/src/google/adk/flows/llm_flows/base_llm_flow.py
index f215895c994..4485cb46148 100644
--- a/src/google/adk/flows/llm_flows/base_llm_flow.py
+++ b/src/google/adk/flows/llm_flows/base_llm_flow.py
@@ -827,6 +827,8 @@ async def _send_to_model(
         await llm_connection.send_realtime(types.ActivityStart())
       elif live_request.activity_end:
         await llm_connection.send_realtime(types.ActivityEnd())
+      elif live_request.audio_stream_end:
+        await llm_connection.send_realtime(True)
       elif live_request.blob:
         # Cache input audio chunks before flushing
         self.audio_cache_manager.cache_audio(
diff --git a/src/google/adk/models/gemini_llm_connection.py b/src/google/adk/models/gemini_llm_connection.py
index 66534b39fdc..717623bb8c7 100644
--- a/src/google/adk/models/gemini_llm_connection.py
+++ b/src/google/adk/models/gemini_llm_connection.py
@@ -29,7 +29,7 @@
 
 logger = logging.getLogger('google_adk.' + __name__)
 
-RealtimeInput = Union[types.Blob, types.ActivityStart, types.ActivityEnd]
+RealtimeInput = Union[types.Blob, types.ActivityStart, types.ActivityEnd, bool]
 from typing import TYPE_CHECKING
 
 if TYPE_CHECKING:
@@ -173,6 +173,9 @@ async def send_realtime(self, input: RealtimeInput) -> None:
     elif isinstance(input, types.ActivityEnd):
       logger.debug('Sending LLM activity end signal.')
       await self._gemini_session.send_realtime_input(activity_end=input)
+    elif isinstance(input, bool) and input:
+      logger.debug('Sending LLM audio stream end signal.')
+      await self._gemini_session.send_realtime_input(audio_stream_end=True)
     else:
       raise ValueError('Unsupported input type: %s' % type(input))
 
diff --git a/tests/unittests/models/test_gemini_llm_connection.py b/tests/unittests/models/test_gemini_llm_connection.py
index 5bded80a266..6d5beee3847 100644
--- a/tests/unittests/models/test_gemini_llm_connection.py
+++ b/tests/unittests/models/test_gemini_llm_connection.py
@@ -89,6 +89,18 @@ async def test_send_realtime_audio_uses_audio_channel_for_live_translate(
   )
 
 
+@pytest.mark.asyncio
+async def test_send_realtime_audio_stream_end(
+    gemini_connection, mock_gemini_session
+):
+  """Test send_realtime with audio_stream_end=True sends the flush signal."""
+  await gemini_connection.send_realtime(True)
+
+  mock_gemini_session.send_realtime_input.assert_called_once_with(
+      audio_stream_end=True
+  )
+
+
 @pytest.mark.asyncio
 async def test_send_history(gemini_connection, mock_gemini_session):
   """Test send_history method."""