diff --git a/litellm/litellm_core_utils/realtime_streaming.py b/litellm/litellm_core_utils/realtime_streaming.py
index a1a070eb5b7..e4b34852b99 100644
--- a/litellm/litellm_core_utils/realtime_streaming.py
+++ b/litellm/litellm_core_utils/realtime_streaming.py
@@ -903,6 +903,8 @@ class RealTimeStreaming:
                 self._collect_user_input_from_backend_event(cast(dict, event))
                 self.store_message(event_str)
                 await self._send_event_to_client(event, event_str)
+                if not self._has_audio_transcription_guardrails():
+                    continue
                 blocked = await self.run_realtime_guardrails(
                     cast(str, transcript),
                     item_id=cast(Optional[str], event.get("item_id")),
@@ -953,6 +955,8 @@ class RealTimeStreaming:
             if self._is_transcription_session:
                 self._capture_transcription_usage(event_obj)
                 return True
+            if not self._has_audio_transcription_guardrails():
+                return True
 
             blocked = await self.run_realtime_guardrails(
                 transcript,
diff --git a/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py b/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py
index 766befd1a99..c614ce2643b 100644
--- a/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py
+++ b/tests/test_litellm/litellm_core_utils/test_realtime_streaming.py
@@ -17,10 +17,6 @@ from litellm.litellm_core_utils.realtime_streaming import (
 )
 from litellm.llms.xai.realtime.transformation import XAIRealtimeNormalizer
 from litellm.types.guardrails import GuardrailEventHooks
-from litellm.types.llms.openai import (
-    OpenAIRealtimeStreamResponseBaseObject,
-    OpenAIRealtimeStreamSessionEvents,
-)
 
 
 def _make_transcript_event(text: str, item_id: str = "item_x") -> bytes:
@@ -809,8 +805,6 @@ async def test_transcription_captured_in_backend_to_client():
     Test that conversation.item.input_audio_transcription.completed events
     from the backend are captured as user input during the WebSocket session.
     """
-    import litellm
-
     client_ws = MagicMock()
     client_ws.send_text = AsyncMock()
 
@@ -903,10 +897,11 @@ async def test_transcription_session_captures_usage_and_skips_response_create():
 
 
 @pytest.mark.asyncio
-async def test_non_transcription_completed_event_still_triggers_response_create():
+async def test_non_transcription_completed_event_without_guardrails_skips_response_create():
     """
-    Regression guard: a normal (non-transcription) session with no guardrails must
-    keep triggering response.create on a completed transcription event.
+    Regression guard: without realtime_input_transcription guardrails, the proxy
+    must not inject response.create on transcription completion because the
+    backend's own server-VAD auto-response is still responsible for the turn.
     """
     client_ws = MagicMock()
     client_ws.send_text = AsyncMock()
@@ -932,7 +927,56 @@ async def test_non_transcription_completed_event_still_triggers_response_create(
 
     assert streaming._is_transcription_session is False
     sent_to_backend = [json.loads(c.args[0]) for c in backend_ws.send.call_args_list if c.args]
-    assert any(e.get("type") == "response.create" for e in sent_to_backend)
+    assert all(e.get("type") != "response.create" for e in sent_to_backend), (
+        f"proxy must not inject response.create without transcription guardrails, got: {sent_to_backend}"
+    )
+
+
+@pytest.mark.asyncio
+async def test_provider_path_completed_event_without_guardrails_skips_response_create():
+    client_ws = MagicMock()
+    client_ws.send_text = AsyncMock()
+
+    backend_ws = MagicMock()
+    backend_ws.send = AsyncMock()
+
+    provider_config = MagicMock()
+    provider_config.transform_realtime_response = MagicMock(
+        return_value={
+            "response": [
+                {
+                    "type": "conversation.item.input_audio_transcription.completed",
+                    "transcript": "hi",
+                    "item_id": "item_1",
+                }
+            ],
+            "current_output_item_id": None,
+            "current_response_id": None,
+            "current_delta_chunks": [],
+            "current_conversation_id": None,
+            "current_item_chunks": [],
+            "current_delta_type": None,
+            "session_configuration_request": None,
+        }
+    )
+
+    logging_obj = MagicMock()
+    logging_obj.litellm_trace_id = "trace_1"
+    logging_obj.async_success_handler = AsyncMock()
+    logging_obj.success_handler = MagicMock()
+
+    streaming = RealTimeStreaming(
+        websocket=client_ws,
+        backend_ws=backend_ws,
+        logging_obj=logging_obj,
+        provider_config=provider_config,
+        model="gemini-2.5-flash",
+    )
+    streaming._send_to_backend = AsyncMock()  # type: ignore[method-assign]
+
+    await streaming._handle_provider_config_message("{}")
+
+    assert streaming._send_to_backend.await_count == 0
 
 
 def test_client_session_update_marks_transcription_session():
@@ -1097,8 +1141,6 @@ def test_capture_transcription_usage_deduplicates_when_already_stored():
     When the event is already in messages (logged via store_message), it must not
     be appended a second time by _capture_transcription_usage.
     """
-    import litellm
-
     streaming = RealTimeStreaming(MagicMock(), MagicMock(), MagicMock())
     # Add the event type to the default logged list so _should_store_message returns True.
     streaming.logged_real_time_event_types = ["conversation.item.input_audio_transcription.completed"]