diff --git a/litellm/litellm_core_utils/realtime_streaming.py b/litellm/litellm_core_utils/realtime_streaming.py
index cef6a24aba1..5b55a5223bf 100644
--- a/litellm/litellm_core_utils/realtime_streaming.py
+++ b/litellm/litellm_core_utils/realtime_streaming.py
@@ -255,7 +255,7 @@ class RealTimeStreaming:
         provider_config: BaseRealtimeConfig | None = None,
         model: str = "",
         user_api_key_dict: object | None = None,
-        request_data: dict | None = None,
+        request_data: Mapping[str, object] | None = None,
         backend_uses_beta_protocol: bool | None = None,
         force_transcription_model: str | None = None,
         event_normalizer: RealtimeEventNormalizer | None = None,
@@ -302,7 +302,7 @@ class RealTimeStreaming:
         self.current_delta_type: ALL_DELTA_TYPES | None = None
         self.session_configuration_request: str | None = None
         self.user_api_key_dict = user_api_key_dict
-        self.request_data: dict = request_data or {}
+        self.request_data: Mapping[str, object] = request_data or {}
         # Violation counter for end_session_after_n_fails support
         self._violation_count: int = 0
         # When a text message is blocked, hold the guardrail reason so the next
@@ -1176,9 +1176,18 @@ class RealTimeStreaming:
         )
 
     def _has_realtime_guardrails(self) -> bool:
-        """Return True if any callback is registered for realtime guardrail event types."""
         from litellm.types.guardrails import GuardrailEventHooks
 
+        if any(
+            isinstance(bucket, Mapping) and "_guardrail_pipelines" in bucket
+            for bucket in (self.request_data.get("metadata"), self.request_data.get("litellm_metadata"))
+        ):
+            from litellm.proxy.utils import pipeline_managed_guardrail_names
+
+            if pipeline_managed_guardrail_names(self.request_data, "pre_call") or pipeline_managed_guardrail_names(
+                self.request_data, "post_call"
+            ):
+                return True
         return self._has_realtime_guardrails_for_event_hooks(
             [
                 GuardrailEventHooks.realtime_input_transcription,
diff --git a/litellm/realtime_api/main.py b/litellm/realtime_api/main.py
index de807867bed..2c946a3dd91 100644
--- a/litellm/realtime_api/main.py
+++ b/litellm/realtime_api/main.py
@@ -10,6 +10,7 @@ from typing import TYPE_CHECKING, Any, Final, Literal, NoReturn, cast
 from typing_extensions import ReadOnly, TypedDict, assert_never
 
 import httpx
+from pydantic import TypeAdapter
 
 import litellm
 from litellm.constants import (
@@ -66,6 +67,7 @@ vertex_llm_base: Final = VertexBase()
 base_llm_http_handler = BaseLLMHTTPHandler()
 _EMPTY_MODEL_PARAMS: Final[Mapping[str, object]] = MappingProxyType({})
 _EMPTY_AUTH_HEADERS: Final[Mapping[str, str]] = MappingProxyType({})
+_METADATA_ADAPTER: Final = TypeAdapter(Mapping[str, object])
 
 
 def _model_params_with_stored_credentials(model_params: Mapping[str, object]) -> Mapping[str, object]:
@@ -104,13 +106,11 @@ def _with_resolved_session_model(session: dict[str, object], model_name: str) ->
     return {**session, "model": model_name}
 
 
-def _build_litellm_metadata(kwargs: dict) -> dict[str, object]:
-    """Build the litellm_metadata dict for guardrail checking (internal only, not forwarded to provider)."""
-    metadata: Final[dict[str, object]] = {**(kwargs.get("litellm_metadata") or {})}
-    guardrails: Final = (kwargs.get("metadata") or {}).get("guardrails") or kwargs.get("guardrails") or []
-    if guardrails:
-        metadata["guardrails"] = guardrails
-    return metadata
+def _build_litellm_metadata(kwargs: Mapping[str, object]) -> dict[str, object]:
+    request_metadata: Final = _METADATA_ADAPTER.validate_python(kwargs.get("metadata") or {})
+    internal_metadata: Final = _METADATA_ADAPTER.validate_python(kwargs.get("litellm_metadata") or {})
+    guardrails: Final = request_metadata.get("guardrails") or kwargs.get("guardrails")
+    return {**request_metadata, **internal_metadata, **({"guardrails": guardrails} if guardrails else {})}
 
 
 def _get_realtime_http_provider_config(
diff --git a/tests/unit/litellm_core_utils/test_realtime_streaming.py b/tests/unit/litellm_core_utils/test_realtime_streaming.py
index b896da11c3c..d9fab04b7ae 100644
--- a/tests/unit/litellm_core_utils/test_realtime_streaming.py
+++ b/tests/unit/litellm_core_utils/test_realtime_streaming.py
@@ -3,6 +3,7 @@ import base64
 import json
 from collections.abc import Coroutine, Mapping, Sequence
 from dataclasses import dataclass
+from datetime import UTC, datetime
 from typing import Final, Literal
 from unittest.mock import AsyncMock, MagicMock, patch
 
@@ -19,6 +20,7 @@ from litellm.constants import (
     REALTIME_SESSION_SUCCESS_LOGGED_KEY,
 )
 from litellm.integrations.custom_guardrail import CustomGuardrail
+from litellm.litellm_core_utils.litellm_logging import Logging
 from litellm.litellm_core_utils.realtime_streaming import (
     RealTimeStreaming,
     client_sent_openai_beta_realtime_header,
@@ -29,10 +31,14 @@ from litellm.llms.xai.realtime.transformation import XAIRealtimeNormalizer
 from litellm.llms.xai.realtime.usage_meter import XAIRealtimeUsageMeter
 
 from litellm.proxy._types import UserAPIKeyAuth
+from litellm.proxy.litellm_pre_call_utils import _apply_resolved_guardrails_to_metadata
+from litellm.realtime_api.main import _build_litellm_metadata
 from litellm.types.guardrails import GuardrailEventHooks
 from litellm.types.llms.xai import RealtimeMeteredUsage
 from litellm.types.realtime import RealtimeResponseTransformInput, RealtimeResponseTypedDict
 from litellm.types.utils import GenericGuardrailAPIInputs
+from litellm.types.proxy.policy_engine import Policy, PolicyCondition, PolicyMatchContext
+from litellm.types.proxy.policy_engine.pipeline_types import GuardrailPipeline, PipelineStep
 
 
 def _mock_logging() -> MagicMock:
@@ -3280,6 +3286,79 @@ async def test_translation_rejects_applicable_guardrails_before_forwarding_audio
         backend.send.assert_awaited_once_with(input_event)
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("metadata_bucket", ("metadata", "litellm_metadata"))
+@pytest.mark.parametrize("mode", ("pre_call", "post_call"))
+@pytest.mark.parametrize("selected,condition_matches", ((True, True), (True, False), (False, True)))
+async def test_translation_rejects_selected_pipeline_guardrails_before_forwarding_audio(
+    monkeypatch: pytest.MonkeyPatch,
+    metadata_bucket: str,
+    mode: Literal["pre_call", "post_call"],
+    selected: bool,
+    condition_matches: bool,
+) -> None:
+    guardrail: Final = CustomGuardrail(
+        guardrail_name="speech-policy", event_hook=GuardrailEventHooks(mode), default_on=False
+    )
+    monkeypatch.setattr(litellm, "callbacks", [guardrail])
+    request_data: Final[dict[str, object]] = {"metadata": {}, "litellm_metadata": {"user_id": "qa-user"}}
+    _apply_resolved_guardrails_to_metadata(
+        data=request_data,
+        metadata_variable_name=metadata_bucket,
+        context=PolicyMatchContext(model="qa-translation"),
+        policy_names=["translation-policy"] if selected else [],
+        policies={
+            "translation-policy": Policy(
+                condition=PolicyCondition(model="qa-translation" if condition_matches else "unrelated-model"),
+                pipeline=GuardrailPipeline(mode=mode, steps=[PipelineStep(guardrail="speech-policy", on_fail="block")]),
+            )
+        },
+    )
+    input_event: Final = json.dumps(
+        {"type": "session.input_audio_buffer.append", "audio": base64.b64encode(bytes(4800)).decode()}
+    )
+    output_event: Final = {"type": "session.output_audio.delta", "delta": base64.b64encode(bytes(4800)).decode()}
+    client: Final = _ga_client_ws()
+    client.receive_text = AsyncMock(side_effect=[input_event, ConnectionClosed(None, None)])
+    client.send_text = AsyncMock()
+    client.close = AsyncMock()
+    backend: Final = MagicMock()
+    backend.send = AsyncMock()
+    backend.recv = AsyncMock(side_effect=[json.dumps(output_event), ConnectionClosed(None, None)])
+    logging_obj: Final = Logging(
+        model="gpt-realtime-translate",
+        messages=[],
+        stream=True,
+        call_type="_arealtime",
+        start_time=datetime(2026, 1, 1, tzinfo=UTC),
+        litellm_call_id="pipeline-rejection",
+        function_id="pipeline-rejection",
+    )
+    worker: Final = _InlineLoggingWorker()
+    streaming: Final = RealTimeStreaming(
+        websocket=client,
+        backend_ws=backend,
+        logging_obj=logging_obj,
+        translation_session=True,
+        request_data={"litellm_metadata": _build_litellm_metadata(request_data)},
+        logging_worker=worker,
+    )
+
+    await streaming.bidirectional_forward()
+    await worker.drain()
+
+    messages: Final = tuple(json.loads(call.args[0]) for call in client.send_text.await_args_list)
+    if selected and condition_matches:
+        assert messages[0]["error"]["type"] == "guardrail_error"
+        client.close.assert_awaited_once_with(
+            code=1008, reason="Translation sessions cannot enforce configured realtime guardrails"
+        )
+        backend.send.assert_not_awaited()
+    else:
+        assert output_event in messages
+        backend.send.assert_awaited_once_with(input_event)
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize("retain_close", (False, True))
 @pytest.mark.parametrize("reported_input,expected_input", [(None, 2.0), (0.0, 0.0), (0.25, 0.25)])