feat: initial commit adding realtime guardrails

litellm/litellm_core_utils/realtime_streaming.py

@@ -1,11 +1,12 @@
 import asyncio
 import concurrent.futures
 import json
-from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union, cast
 
 import litellm
 from litellm._logging import verbose_logger
 from litellm.llms.base_llm.realtime.transformation import BaseRealtimeConfig
+from litellm.proxy._types import UserAPIKeyAuth
 from litellm.types.llms.openai import (
     OpenAIRealtimeEvents,
     OpenAIRealtimeOutputItemDone,
@@ -17,6 +18,13 @@ from litellm.types.realtime import ALL_DELTA_TYPES
 
 from .litellm_logging import Logging as LiteLLMLogging
 
+if TYPE_CHECKING:
+    from litellm.proxy.utils import ProxyLogging
+
+    ProxyLoggingObj = ProxyLogging
+else:
+    ProxyLoggingObj = Any
+
 if TYPE_CHECKING:
     from websockets.asyncio.client import ClientConnection
 
@@ -40,6 +48,7 @@ class RealTimeStreaming:
         websocket: Any,
         backend_ws: CLIENT_CONNECTION_CLASS,
         logging_obj: LiteLLMLogging,
+        proxy_logging_obj: Optional[Any] = None,
         provider_config: Optional[BaseRealtimeConfig] = None,
         model: str = "",
     ):
@@ -55,6 +64,7 @@ class RealTimeStreaming:
             _logged_real_time_event_types = DefaultLoggedRealTimeEventTypes
         self.logged_real_time_event_types = _logged_real_time_event_types
         self.provider_config = provider_config
+        self.proxy_logging_obj = proxy_logging_obj
         self.model = model
         self.current_delta_chunks: Optional[List[OpenAIRealtimeResponseDelta]] = None
         self.current_output_item_id: Optional[str] = None
@@ -75,7 +85,9 @@ class RealTimeStreaming:
             return True
         return False
 
-    def store_message(self, message: Union[str, bytes, OpenAIRealtimeEvents]):
+    async def store_and_check_message(
+        self, message: Union[str, bytes, OpenAIRealtimeEvents]
+    ):
         """Store message in list"""
         if isinstance(message, bytes):
             message = message.decode("utf-8")
@@ -98,6 +110,9 @@ class RealTimeStreaming:
         if self._should_store_message(message_obj):
             self.messages.append(message_obj)
 
+        if self.proxy_logging_obj:
+            await self.run_post_call_guardrails(cast(OpenAIRealtimeEvents, message_obj))
+
     def store_input(self, message: dict):
         """Store input message"""
         self.input_message = message
@@ -113,6 +128,14 @@ class RealTimeStreaming:
             ## SYNC LOGGING
             executor.submit(self.logging_obj.success_handler(self.messages))
 
+    async def run_post_call_guardrails(self, message: OpenAIRealtimeEvents):
+        if self.proxy_logging_obj:
+            await cast(ProxyLoggingObj, self.proxy_logging_obj).post_call_success_hook(
+                data={},
+                user_api_key_dict=UserAPIKeyAuth(),
+                response=message,
+            )
+
     async def backend_to_client_send_messages(self):
         import websockets
 
@@ -159,17 +182,17 @@ class RealTimeStreaming:
                         for event in transformed_response:
                             event_str = json.dumps(event)
                             ## LOGGING
-                            self.store_message(event_str)
+                            await self.store_and_check_message(event_str)
                             await self.websocket.send_text(event_str)
                     else:
                         event_str = json.dumps(transformed_response)
                         ## LOGGING
-                        self.store_message(event_str)
+                        await self.store_and_check_message(event_str)
                         await self.websocket.send_text(event_str)
 
                 else:
                     ## LOGGING
-                    self.store_message(raw_response)
+                    await self.store_and_check_message(raw_response)
                     await self.websocket.send_text(raw_response)
 
         except websockets.exceptions.ConnectionClosed as e:  # type: ignore

litellm/llms/azure/realtime/handler.py

@@ -6,13 +6,13 @@ This requires websockets, and is currently only supported on LiteLLM Proxy.
 
 from typing import Any, Optional, cast
 
+from litellm._logging import verbose_proxy_logger
 from litellm.constants import REALTIME_WEBSOCKET_MAX_MESSAGE_SIZE_BYTES
 
 from ....litellm_core_utils.litellm_logging import Logging as LiteLLMLogging
 from ....litellm_core_utils.realtime_streaming import RealTimeStreaming
 from ....llms.custom_httpx.http_handler import get_shared_realtime_ssl_context
 from ..azure import AzureChatCompletion
-from litellm._logging import verbose_proxy_logger
 
 # BACKEND_WS_URL = "ws://localhost:8080/v1/realtime?model=gpt-4o-realtime-preview-2024-10-01"
 
@@ -58,7 +58,7 @@ class AzureOpenAIRealtime(AzureChatCompletion):
 
         # Determine path based on realtime_protocol
         if realtime_protocol in ("GA", "v1"):
-            path = "/openai/v1/realtime" 
+            path = "/openai/v1/realtime"
             return f"{api_base}{path}?model={model}"
         else:
             # Default to beta path for backwards compatibility
@@ -77,6 +77,7 @@ class AzureOpenAIRealtime(AzureChatCompletion):
         client: Optional[Any] = None,
         timeout: Optional[float] = None,
         realtime_protocol: Optional[str] = None,
+        proxy_logging_obj: Optional[Any] = None,
     ):
         import websockets
         from websockets.asyncio.client import ClientConnection
@@ -101,12 +102,17 @@ class AzureOpenAIRealtime(AzureChatCompletion):
                 ssl=ssl_context,
             ) as backend_ws:
                 realtime_streaming = RealTimeStreaming(
-                    websocket, cast(ClientConnection, backend_ws), logging_obj
+                    websocket,
+                    cast(ClientConnection, backend_ws),
+                    logging_obj,
+                    proxy_logging_obj,
                 )
                 await realtime_streaming.bidirectional_forward()
 
         except websockets.exceptions.InvalidStatusCode as e:  # type: ignore
             await websocket.close(code=e.status_code, reason=str(e))
         except Exception:
-            verbose_proxy_logger.exception("Error in AzureOpenAIRealtime.async_realtime")
+            verbose_proxy_logger.exception(
+                "Error in AzureOpenAIRealtime.async_realtime"
+            )
             pass

litellm/llms/custom_httpx/llm_http_handler.py

@@ -1320,6 +1320,7 @@ class BaseLLMHTTPHandler:
         Returns: (headers, complete_url, data, files)
         """
         from litellm.llms.base_llm.ocr.transformation import OCRRequestData
+
         headers = provider_config.validate_environment(
             api_key=api_key,
             api_base=api_base,
@@ -1812,9 +1813,11 @@ class BaseLLMHTTPHandler:
             Optional[litellm.types.utils.ProviderSpecificHeader],
             kwargs.get("provider_specific_header", None),
         )
-        provider_specific_headers = ProviderSpecificHeaderUtils.get_provider_specific_headers(
-            provider_specific_header=provider_specific_header,
-            custom_llm_provider=custom_llm_provider,
+        provider_specific_headers = (
+            ProviderSpecificHeaderUtils.get_provider_specific_headers(
+                provider_specific_header=provider_specific_header,
+                custom_llm_provider=custom_llm_provider,
+            )
         )
         forwarded_headers = kwargs.get("headers", None)
         # Also check for extra_headers in kwargs (from config or direct calls)
@@ -2809,12 +2812,12 @@ class BaseLLMHTTPHandler:
     ) -> tuple[Optional[str], Optional[dict]]:
         """
         Extract upload URL from initial file creation response.
-        
+
         Args:
             response: HTTP response from initial file creation request
             upload_url_location: Where to find URL ('headers' or 'body')
             upload_url_key: Key name for URL in response body (default: 'upload_url')
-            
+
         Returns:
             Tuple of (upload_url, response_data)
             - upload_url: The extracted upload URL, or None if not found
@@ -2895,7 +2898,10 @@ class BaseLLMHTTPHandler:
         else:
             sync_httpx_client = client
 
-        if isinstance(transformed_request, dict) and "initial_request" in transformed_request:
+        if (
+            isinstance(transformed_request, dict)
+            and "initial_request" in transformed_request
+        ):
             # Handle two-step uploads (TwoStepFileUploadConfig)
             # Used by providers like Manus, Google Cloud Storage
             try:
@@ -2911,24 +2917,32 @@ class BaseLLMHTTPHandler:
                 )
 
                 # Extract upload URL from response
-                upload_url, initial_response_data = self._extract_upload_url_from_response(
-                    response=initial_response,
-                    upload_url_location=transformed_request.get("upload_url_location", "headers"),
-                    upload_url_key=transformed_request.get("upload_url_key", "upload_url"),
+                upload_url, initial_response_data = (
+                    self._extract_upload_url_from_response(
+                        response=initial_response,
+                        upload_url_location=transformed_request.get(
+                            "upload_url_location", "headers"
+                        ),
+                        upload_url_key=transformed_request.get(
+                            "upload_url_key", "upload_url"
+                        ),
+                    )
                 )
 
                 if not upload_url:
                     raise ValueError("Failed to get upload URL from initial request")
 
                 # Step 2: Upload the actual file
-                upload_method = transformed_request["upload_request"].get("method", "POST").lower()
+                upload_method = (
+                    transformed_request["upload_request"].get("method", "POST").lower()
+                )
                 upload_response = getattr(sync_httpx_client, upload_method)(
                     url=upload_url,
                     headers=transformed_request["upload_request"]["headers"],
                     data=transformed_request["upload_request"]["data"],
                     timeout=timeout,
                 )
-                
+
                 # Store initial response for transformation
                 if initial_response_data:
                     litellm_params["initial_file_response"] = initial_response_data
@@ -2937,7 +2951,11 @@ class BaseLLMHTTPHandler:
                     e=e,
                     provider_config=provider_config,
                 )
-        elif isinstance(transformed_request, dict) and "method" in transformed_request and "initial_request" not in transformed_request:
+        elif (
+            isinstance(transformed_request, dict)
+            and "method" in transformed_request
+            and "initial_request" not in transformed_request
+        ):
             # Handle pre-signed requests (e.g., from Bedrock S3 uploads)
             # Type narrowing: this is a plain dict, not TwoStepFileUploadConfig
             presigned_request = cast(Dict[str, Any], transformed_request)
@@ -2974,7 +2992,9 @@ class BaseLLMHTTPHandler:
                     timeout=timeout,
                 )
         else:
-            raise ValueError(f"Unsupported transformed_request type: {type(transformed_request)}")
+            raise ValueError(
+                f"Unsupported transformed_request type: {type(transformed_request)}"
+            )
 
         # Store the upload URL in litellm_params for the transformation method
         litellm_params_with_url = dict(litellm_params)
@@ -3021,7 +3041,10 @@ class BaseLLMHTTPHandler:
             },
         )
 
-        if isinstance(transformed_request, dict) and "initial_request" in transformed_request:
+        if (
+            isinstance(transformed_request, dict)
+            and "initial_request" in transformed_request
+        ):
             # Handle two-step uploads (TwoStepFileUploadConfig)
             # Used by providers like Manus, Google Cloud Storage
             try:
@@ -3037,24 +3060,32 @@ class BaseLLMHTTPHandler:
                 )
 
                 # Extract upload URL from response
-                upload_url, initial_response_data = self._extract_upload_url_from_response(
-                    response=initial_response,
-                    upload_url_location=transformed_request.get("upload_url_location", "headers"),
-                    upload_url_key=transformed_request.get("upload_url_key", "upload_url"),
+                upload_url, initial_response_data = (
+                    self._extract_upload_url_from_response(
+                        response=initial_response,
+                        upload_url_location=transformed_request.get(
+                            "upload_url_location", "headers"
+                        ),
+                        upload_url_key=transformed_request.get(
+                            "upload_url_key", "upload_url"
+                        ),
+                    )
                 )
 
                 if not upload_url:
                     raise ValueError("Failed to get upload URL from initial request")
 
                 # Step 2: Upload the actual file
-                upload_method = transformed_request["upload_request"].get("method", "POST").lower()
+                upload_method = (
+                    transformed_request["upload_request"].get("method", "POST").lower()
+                )
                 upload_response = await getattr(async_httpx_client, upload_method)(
                     url=upload_url,
                     headers=transformed_request["upload_request"]["headers"],
                     data=transformed_request["upload_request"]["data"],
                     timeout=timeout,
                 )
-                
+
                 # Store initial response for transformation
                 if initial_response_data:
                     litellm_params["initial_file_response"] = initial_response_data
@@ -3064,7 +3095,11 @@ class BaseLLMHTTPHandler:
                     e=e,
                     provider_config=provider_config,
                 )
-        elif isinstance(transformed_request, dict) and "method" in transformed_request and "initial_request" not in transformed_request:
+        elif (
+            isinstance(transformed_request, dict)
+            and "method" in transformed_request
+            and "initial_request" not in transformed_request
+        ):
             # Handle pre-signed requests (e.g., from Bedrock S3 uploads)
             # Type narrowing: this is a plain dict, not TwoStepFileUploadConfig
             presigned_request = cast(Dict[str, Any], transformed_request)
@@ -3099,7 +3134,9 @@ class BaseLLMHTTPHandler:
                     timeout=timeout,
                 )
         else:
-            raise ValueError(f"Unsupported transformed_request type: {type(transformed_request)}")
+            raise ValueError(
+                f"Unsupported transformed_request type: {type(transformed_request)}"
+            )
 
         return provider_config.transform_create_file_response(
             model=None,
@@ -3698,13 +3735,15 @@ class BaseLLMHTTPHandler:
             litellm_params=dict(litellm_params),
         )
 
-        url, data = responses_api_provider_config.transform_compact_response_api_request(
-            model=model,
-            input=input,
-            response_api_optional_request_params=response_api_optional_request_params,
-            api_base=api_base,
-            litellm_params=litellm_params,
-            headers=headers,
+        url, data = (
+            responses_api_provider_config.transform_compact_response_api_request(
+                model=model,
+                input=input,
+                response_api_optional_request_params=response_api_optional_request_params,
+                api_base=api_base,
+                litellm_params=litellm_params,
+                headers=headers,
+            )
         )
 
         ## LOGGING
@@ -3777,13 +3816,15 @@ class BaseLLMHTTPHandler:
             litellm_params=dict(litellm_params),
         )
 
-        url, data = responses_api_provider_config.transform_compact_response_api_request(
-            model=model,
-            input=input,
-            response_api_optional_request_params=response_api_optional_request_params,
-            api_base=api_base,
-            litellm_params=litellm_params,
-            headers=headers,
+        url, data = (
+            responses_api_provider_config.transform_compact_response_api_request(
+                model=model,
+                input=input,
+                response_api_optional_request_params=response_api_optional_request_params,
+                api_base=api_base,
+                litellm_params=litellm_params,
+                headers=headers,
+            )
         )
 
         ## LOGGING
@@ -3871,9 +3912,7 @@ class BaseLLMHTTPHandler:
         )
 
         try:
-            response = sync_httpx_client.get(
-                url=url, headers=headers, params=params
-            )
+            response = sync_httpx_client.get(url=url, headers=headers, params=params)
         except Exception as e:
             raise self._handle_error(e=e, provider_config=provider_config)
 
@@ -4001,9 +4040,7 @@ class BaseLLMHTTPHandler:
         )
 
         try:
-            response = sync_httpx_client.delete(
-                url=url, headers=headers, params=params
-            )
+            response = sync_httpx_client.delete(url=url, headers=headers, params=params)
         except Exception as e:
             raise self._handle_error(e=e, provider_config=provider_config)
 
@@ -4131,9 +4168,7 @@ class BaseLLMHTTPHandler:
         )
 
         try:
-            response = sync_httpx_client.get(
-                url=url, headers=headers, params=params
-            )
+            response = sync_httpx_client.get(url=url, headers=headers, params=params)
         except Exception as e:
             raise self._handle_error(e=e, provider_config=provider_config)
 
@@ -4213,7 +4248,9 @@ class BaseLLMHTTPHandler:
         _is_async: bool = False,
         client: Optional[Union[HTTPHandler, AsyncHTTPHandler]] = None,
         timeout: Optional[Union[float, httpx.Timeout]] = None,
-    ) -> Union["HttpxBinaryResponseContent", Coroutine[Any, Any, "HttpxBinaryResponseContent"]]:
+    ) -> Union[
+        "HttpxBinaryResponseContent", Coroutine[Any, Any, "HttpxBinaryResponseContent"]
+    ]:
         """
         Retrieve file content by ID
         """
@@ -4261,9 +4298,7 @@ class BaseLLMHTTPHandler:
         )
 
         try:
-            response = sync_httpx_client.get(
-                url=url, headers=headers, params=params
-            )
+            response = sync_httpx_client.get(url=url, headers=headers, params=params)
         except Exception as e:
             raise self._handle_error(e=e, provider_config=provider_config)
 
@@ -4371,9 +4406,7 @@ class BaseLLMHTTPHandler:
         from litellm._logging import verbose_logger
         from litellm.integrations.custom_logger import CustomLogger
 
-        callbacks = litellm.callbacks + (
-            logging_obj.dynamic_success_callbacks or []
-        )
+        callbacks = litellm.callbacks + (logging_obj.dynamic_success_callbacks or [])
         tools = anthropic_messages_optional_request_params.get("tools", [])
 
         for callback in callbacks:
@@ -4396,7 +4429,9 @@ class BaseLLMHTTPHandler:
                         # Second: Execute agentic loop
                         # Add custom_llm_provider to kwargs so the agentic loop can reconstruct the full model name
                         kwargs_with_provider = kwargs.copy() if kwargs else {}
-                        kwargs_with_provider["custom_llm_provider"] = custom_llm_provider
+                        kwargs_with_provider["custom_llm_provider"] = (
+                            custom_llm_provider
+                        )
                         agentic_response = await callback.async_run_agentic_loop(
                             tools=tool_calls,
                             model=model,
@@ -4422,11 +4457,13 @@ class BaseLLMHTTPHandler:
         # 2. No agentic loop ran (LLM didn't use the tool)
         # 3. We have a non-streaming response that needs to be converted to streaming
         websearch_converted_stream = (
-            logging_obj.model_call_details.get("websearch_interception_converted_stream", False)
+            logging_obj.model_call_details.get(
+                "websearch_interception_converted_stream", False
+            )
             if logging_obj is not None
             else False
         )
-        
+
         if websearch_converted_stream:
             from typing import cast
 
@@ -4437,11 +4474,11 @@ class BaseLLMHTTPHandler:
             from litellm.types.llms.anthropic_messages.anthropic_response import (
                 AnthropicMessagesResponse,
             )
-            
+
             verbose_logger.debug(
                 "WebSearchInterception: No tool call made, converting non-streaming response to fake stream"
             )
-            
+
             # Convert the non-streaming response to a fake stream
             # The response should be an AnthropicMessagesResponse (dict)
             if isinstance(response, dict):
@@ -4450,7 +4487,7 @@ class BaseLLMHTTPHandler:
                     response=cast(AnthropicMessagesResponse, response)
                 )
                 return fake_stream
-        
+
         return None
 
     def _handle_error(
@@ -4513,6 +4550,7 @@ class BaseLLMHTTPHandler:
         model: str,
         websocket: Any,
         logging_obj: LiteLLMLoggingObj,
+        proxy_logging_obj: Optional[Any],
         provider_config: BaseRealtimeConfig,
         headers: dict,
         api_base: Optional[str] = None,
@@ -4542,6 +4580,7 @@ class BaseLLMHTTPHandler:
                     websocket,
                     cast(ClientConnection, backend_ws),
                     logging_obj,
+                    proxy_logging_obj,
                     provider_config,
                     model,
                 )
@@ -5084,7 +5123,7 @@ class BaseLLMHTTPHandler:
             model=model,
             litellm_params=litellm_params,
         )
-        
+
         if extra_headers:
             headers.update(extra_headers)
 
@@ -5094,13 +5133,15 @@ class BaseLLMHTTPHandler:
             litellm_params=dict(litellm_params),
         )
 
-        data, files, api_base = video_generation_provider_config.transform_video_create_request(
-            model=model,
-            prompt=prompt,
-            video_create_optional_request_params=video_generation_optional_request_params,
-            litellm_params=litellm_params,
-            headers=headers,
-            api_base=api_base,
+        data, files, api_base = (
+            video_generation_provider_config.transform_video_create_request(
+                model=model,
+                prompt=prompt,
+                video_create_optional_request_params=video_generation_optional_request_params,
+                litellm_params=litellm_params,
+                headers=headers,
+                api_base=api_base,
+            )
         )
 
         ## LOGGING
@@ -5195,13 +5236,15 @@ class BaseLLMHTTPHandler:
             litellm_params=dict(litellm_params),
         )
 
-        data, files, api_base = video_generation_provider_config.transform_video_create_request(
-            model=model,
-            prompt=prompt,
-            api_base=api_base,
-            video_create_optional_request_params=video_generation_optional_request_params,
-            litellm_params=litellm_params,
-            headers=headers,
+        data, files, api_base = (
+            video_generation_provider_config.transform_video_create_request(
+                model=model,
+                prompt=prompt,
+                api_base=api_base,
+                video_create_optional_request_params=video_generation_optional_request_params,
+                litellm_params=litellm_params,
+                headers=headers,
+            )
         )
 
         ## LOGGING
@@ -5216,7 +5259,7 @@ class BaseLLMHTTPHandler:
         )
 
         try:
-            #Use JSON when no files, otherwise use form data with files
+            # Use JSON when no files, otherwise use form data with files
             if files is None or len(files) == 0:
                 response = await async_httpx_client.post(
                     url=api_base,
@@ -5865,11 +5908,13 @@ class BaseLLMHTTPHandler:
         )
 
         # Transform the request using the provider config
-        url, data = video_status_provider_config.transform_video_status_retrieve_request(
-            video_id=video_id,
-            api_base=api_base,
-            litellm_params=litellm_params,
-            headers=headers,
+        url, data = (
+            video_status_provider_config.transform_video_status_retrieve_request(
+                video_id=video_id,
+                api_base=api_base,
+                litellm_params=litellm_params,
+                headers=headers,
+            )
         )
 
         ## LOGGING
@@ -5899,10 +5944,12 @@ class BaseLLMHTTPHandler:
                     headers=headers,
                 )
 
-            return video_status_provider_config.transform_video_status_retrieve_response(
-                raw_response=response,
-                logging_obj=logging_obj,
-                custom_llm_provider=custom_llm_provider,
+            return (
+                video_status_provider_config.transform_video_status_retrieve_response(
+                    raw_response=response,
+                    logging_obj=logging_obj,
+                    custom_llm_provider=custom_llm_provider,
+                )
             )
 
         except Exception as e:
@@ -5952,11 +5999,13 @@ class BaseLLMHTTPHandler:
         )
 
         # Transform the request using the provider config
-        url, data = video_status_provider_config.transform_video_status_retrieve_request(
-            video_id=video_id,
-            api_base=api_base,
-            litellm_params=litellm_params,
-            headers=headers,
+        url, data = (
+            video_status_provider_config.transform_video_status_retrieve_request(
+                video_id=video_id,
+                api_base=api_base,
+                litellm_params=litellm_params,
+                headers=headers,
+            )
         )
 
         ## LOGGING
@@ -5985,10 +6034,12 @@ class BaseLLMHTTPHandler:
                     url=url,
                     headers=headers,
                 )
-            return video_status_provider_config.transform_video_status_retrieve_response(
-                raw_response=response,
-                logging_obj=logging_obj,
-                custom_llm_provider=custom_llm_provider,
+            return (
+                video_status_provider_config.transform_video_status_retrieve_response(
+                    raw_response=response,
+                    logging_obj=logging_obj,
+                    custom_llm_provider=custom_llm_provider,
+                )
             )
 
         except Exception as e:
@@ -5996,7 +6047,7 @@ class BaseLLMHTTPHandler:
                 e=e,
                 provider_config=video_status_provider_config,
             )
-    
+
     ###### CONTAINER HANDLER ######
     def container_create_handler(
         self,
@@ -6036,7 +6087,7 @@ class BaseLLMHTTPHandler:
             headers=extra_headers or {},
             api_key=litellm_params.get("api_key", None),
         )
-        
+
         # Add Content-Type header for JSON requests
         headers["Content-Type"] = "application/json"
 
@@ -6086,7 +6137,7 @@ class BaseLLMHTTPHandler:
                 e=e,
                 provider_config=container_provider_config,
             )
-    
+
     async def async_container_create_handler(
         self,
         name: str,
@@ -6112,7 +6163,7 @@ class BaseLLMHTTPHandler:
             headers=extra_headers or {},
             api_key=litellm_params.get("api_key", None),
         )
-        
+
         # Add Content-Type header for JSON requests
         headers["Content-Type"] = "application/json"
 
@@ -6162,7 +6213,7 @@ class BaseLLMHTTPHandler:
                 e=e,
                 provider_config=container_provider_config,
             )
-    
+
     def container_list_handler(
         self,
         container_provider_config: "BaseContainerConfig",
@@ -6254,7 +6305,7 @@ class BaseLLMHTTPHandler:
                 e=e,
                 provider_config=container_provider_config,
             )
-    
+
     async def async_container_list_handler(
         self,
         container_provider_config: "BaseContainerConfig",
@@ -6331,7 +6382,7 @@ class BaseLLMHTTPHandler:
                 e=e,
                 provider_config=container_provider_config,
             )
-    
+
     def container_retrieve_handler(
         self,
         container_id: str,
@@ -6387,7 +6438,7 @@ class BaseLLMHTTPHandler:
             litellm_params=litellm_params,
             headers=headers,
         )
-        
+
         # Add any extra query parameters
         if extra_query:
             params.update(extra_query)
@@ -6421,7 +6472,7 @@ class BaseLLMHTTPHandler:
                 e=e,
                 provider_config=container_provider_config,
             )
-    
+
     async def async_container_retrieve_handler(
         self,
         container_id: str,
@@ -6464,7 +6515,7 @@ class BaseLLMHTTPHandler:
             litellm_params=litellm_params,
             headers=headers,
         )
-        
+
         # Add any extra query parameters
         if extra_query:
             params.update(extra_query)
@@ -6498,7 +6549,7 @@ class BaseLLMHTTPHandler:
                 e=e,
                 provider_config=container_provider_config,
             )
-    
+
     def container_delete_handler(
         self,
         container_id: str,
@@ -6554,7 +6605,7 @@ class BaseLLMHTTPHandler:
             litellm_params=litellm_params,
             headers=headers,
         )
-        
+
         # Add any extra query parameters
         if extra_query:
             params.update(extra_query)
@@ -6588,7 +6639,7 @@ class BaseLLMHTTPHandler:
                 e=e,
                 provider_config=container_provider_config,
             )
-    
+
     async def async_container_delete_handler(
         self,
         container_id: str,
@@ -6631,7 +6682,7 @@ class BaseLLMHTTPHandler:
             litellm_params=litellm_params,
             headers=headers,
         )
-        
+
         # Add any extra query parameters
         if extra_query:
             params.update(extra_query)
@@ -6680,7 +6731,9 @@ class BaseLLMHTTPHandler:
         timeout: Union[float, httpx.Timeout] = 600,
         _is_async: bool = False,
         client: Optional[Union[HTTPHandler, AsyncHTTPHandler]] = None,
-    ) -> Union["ContainerFileListResponse", Coroutine[Any, Any, "ContainerFileListResponse"]]:
+    ) -> Union[
+        "ContainerFileListResponse", Coroutine[Any, Any, "ContainerFileListResponse"]
+    ]:
         if _is_async:
             return self.async_container_file_list_handler(
                 container_id=container_id,
@@ -6887,12 +6940,14 @@ class BaseLLMHTTPHandler:
         )
 
         # Transform the request using the provider config
-        url, params = container_provider_config.transform_container_file_content_request(
-            container_id=container_id,
-            file_id=file_id,
-            api_base=api_base,
-            litellm_params=litellm_params,
-            headers=headers,
+        url, params = (
+            container_provider_config.transform_container_file_content_request(
+                container_id=container_id,
+                file_id=file_id,
+                api_base=api_base,
+                litellm_params=litellm_params,
+                headers=headers,
+            )
         )
 
         ## LOGGING
@@ -6960,12 +7015,14 @@ class BaseLLMHTTPHandler:
         )
 
         # Transform the request using the provider config
-        url, params = container_provider_config.transform_container_file_content_request(
-            container_id=container_id,
-            file_id=file_id,
-            api_base=api_base,
-            litellm_params=litellm_params,
-            headers=headers,
+        url, params = (
+            container_provider_config.transform_container_file_content_request(
+                container_id=container_id,
+                file_id=file_id,
+                api_base=api_base,
+                litellm_params=litellm_params,
+                headers=headers,
+            )
         )
 
         ## LOGGING
@@ -7034,7 +7091,9 @@ class BaseLLMHTTPHandler:
         )
 
         # Check if provider has async transform method
-        if hasattr(vector_store_provider_config, "atransform_search_vector_store_request"):
+        if hasattr(
+            vector_store_provider_config, "atransform_search_vector_store_request"
+        ):
             (
                 url,
                 request_body,
@@ -8709,29 +8768,29 @@ class BaseLLMHTTPHandler:
     ) -> tuple[Optional[Dict], Optional[list]]:
         """
         Helper to prepare multipart/form-data request for skills API.
-        
+
         Args:
             request_body: Request body containing files and other fields
             headers: Request headers
-            
+
         Returns:
             Tuple of (data_dict, files_list) for multipart request, or (None, None) if no files
         """
         if "files" not in request_body or not request_body["files"]:
             return None, None
-            
+
         # Remove content-type header if present - httpx will set it automatically for multipart
         if "content-type" in headers:
             del headers["content-type"]
-        
+
         # Prepare files for multipart upload
         files = []
         for file_obj in request_body["files"]:
             files.append(("files[]", file_obj))
-        
+
         # Prepare data (non-file fields)
         data = {k: v for k, v in request_body.items() if k != "files"}
-        
+
         return data, files
 
     def create_skill_handler(
@@ -8787,7 +8846,7 @@ class BaseLLMHTTPHandler:
             data, files = self._prepare_skill_multipart_request(
                 request_body=request_body, headers=headers
             )
-            
+
             if files is not None:
                 response = sync_httpx_client.post(
                     url=url, headers=headers, data=data, files=files, timeout=timeout
@@ -8847,7 +8906,7 @@ class BaseLLMHTTPHandler:
             data, files = self._prepare_skill_multipart_request(
                 request_body=request_body, headers=headers
             )
-            
+
             if files is not None:
                 response = await async_httpx_client.post(
                     url=url, headers=headers, data=data, files=files, timeout=timeout
@@ -9071,9 +9130,7 @@ class BaseLLMHTTPHandler:
         )
 
         try:
-            response = await async_httpx_client.get(
-                url=url, headers=headers
-            )
+            response = await async_httpx_client.get(url=url, headers=headers)
         except Exception as e:
             raise self._handle_error(
                 e=e,

litellm/llms/openai/realtime/handler.py

@@ -42,9 +42,11 @@ class OpenAIRealtime(OpenAIChatCompletion):
         client: Optional[Any] = None,
         timeout: Optional[float] = None,
         query_params: Optional[RealtimeQueryParams] = None,
+        proxy_logging_obj: Optional[Any] = None,
     ):
         import websockets
         from websockets.asyncio.client import ClientConnection
+
         if api_base is None:
             api_base = "https://api.openai.com/"
         if api_key is None:
@@ -58,7 +60,9 @@ class OpenAIRealtime(OpenAIChatCompletion):
         try:
             # Only use SSL context for secure websocket connections (wss://)
             # websockets library doesn't accept ssl argument for ws:// URIs
-            ssl_context = None if url.startswith("ws://") else get_shared_realtime_ssl_context()
+            ssl_context = (
+                None if url.startswith("ws://") else get_shared_realtime_ssl_context()
+            )
             # Log a masked request preview consistent with other endpoints.
             logging_obj.pre_call(
                 input=None,
@@ -82,7 +86,10 @@ class OpenAIRealtime(OpenAIChatCompletion):
                 ssl=ssl_context,
             ) as backend_ws:
                 realtime_streaming = RealTimeStreaming(
-                    websocket, cast(ClientConnection, backend_ws), logging_obj
+                    websocket,
+                    cast(ClientConnection, backend_ws),
+                    logging_obj,
+                    proxy_logging_obj,
                 )
                 await realtime_streaming.bidirectional_forward()
 

litellm/proxy/_new_secret_config.yaml

@@ -13,4 +13,21 @@ model_list:
   - model_name: gpt-4.1-mini
     litellm_params:
       model: openai/gpt-4.1-mini
+  - model_name: openai-gpt-4o-realtime-audio
+    litellm_params:
+      model: openai/gpt-realtime
+      api_key: os.environ/OPENAI_API_KEY
+    model_info:
+      mode: realtime
 
+
+
+guardrails:
+  - guardrail_name: "keyword-filter"
+    litellm_params:
+      guardrail: litellm_content_filter
+      mode: "post_call"
+      blocked_words:
+        - keyword: "lincoln"
+          action: "BLOCK"
+          description: "Do not talk about Lincoln"

litellm/proxy/proxy_server.py

@@ -6517,7 +6517,7 @@ async def realtime_websocket_endpoint(
         RealtimeQueryParams, dict(_realtime_query_params_template(model, intent))
     )
 
-    data = {
+    data: Dict = {
         "model": model,
         "websocket": websocket,
         "query_params": query_params,  # Only explicit params
@@ -6559,6 +6559,11 @@ async def realtime_websocket_endpoint(
             model=model,
             route_type="_arealtime",
         )
+
+        ## POST-CALL GUARDRAILS ##
+        # PASS post-call guardrails to the route request for realtime requests
+        data["proxy_logging_obj"] = proxy_logging_obj
+
         llm_call = await route_request(
             data=data,
             route_type="_arealtime",

litellm/realtime_api/main.py

@@ -3,8 +3,8 @@
 from typing import Any, Optional, cast
 
 import litellm
-from litellm.litellm_core_utils.get_llm_provider_logic import get_llm_provider
 from litellm.constants import REALTIME_WEBSOCKET_MAX_MESSAGE_SIZE_BYTES
+from litellm.litellm_core_utils.get_llm_provider_logic import get_llm_provider
 from litellm.llms.base_llm.realtime.transformation import BaseRealtimeConfig
 from litellm.llms.custom_httpx.llm_http_handler import BaseLLMHTTPHandler
 from litellm.secret_managers.main import get_secret_str
@@ -16,9 +16,9 @@ from litellm.utils import ProviderConfigManager
 from ..litellm_core_utils.get_litellm_params import get_litellm_params
 from ..litellm_core_utils.litellm_logging import Logging as LiteLLMLogging
 from ..llms.azure.realtime.handler import AzureOpenAIRealtime
+from ..llms.custom_httpx.http_handler import get_shared_realtime_ssl_context
 from ..llms.openai.realtime.handler import OpenAIRealtime
 from ..utils import client as wrapper_client
-from ..llms.custom_httpx.http_handler import get_shared_realtime_ssl_context
 
 azure_realtime = AzureOpenAIRealtime()
 openai_realtime = OpenAIRealtime()
@@ -50,6 +50,7 @@ async def _arealtime(
     if extra_headers is not None:
         headers.update(extra_headers)
     litellm_logging_obj: LiteLLMLogging = kwargs.get("litellm_logging_obj")  # type: ignore
+    proxy_logging_obj: LiteLLMLogging = kwargs.pop("proxy_logging_obj", None)  # type: ignore
     user = kwargs.get("user", None)
     litellm_params = GenericLiteLLMParams(**kwargs)
 
@@ -85,6 +86,7 @@ async def _arealtime(
             websocket=websocket,
             logging_obj=litellm_logging_obj,
             provider_config=provider_config,
+            proxy_logging_obj=proxy_logging_obj,
             api_base=api_base,
             api_key=api_key,
             client=client,
@@ -106,16 +108,9 @@ async def _arealtime(
             or get_secret_str("AZURE_API_KEY")
         )
 
-        api_version = (
-            api_version
-            or litellm_params.api_version
-            or "2024-10-01-preview"
-        )
-        
-        realtime_protocol = (
-            kwargs.get("realtime_protocol")
-            or "beta"
-        )
+        api_version = api_version or litellm_params.api_version or "2024-10-01-preview"
+
+        realtime_protocol = kwargs.get("realtime_protocol") or "beta"
         await azure_realtime.async_realtime(
             model=model,
             websocket=websocket,
@@ -127,6 +122,7 @@ async def _arealtime(
             timeout=timeout,
             logging_obj=litellm_logging_obj,
             realtime_protocol=realtime_protocol,
+            proxy_logging_obj=proxy_logging_obj,
         )
     elif _custom_llm_provider == "openai":
         api_base = (
@@ -152,6 +148,7 @@ async def _arealtime(
             client=None,
             timeout=timeout,
             query_params=query_params,
+            proxy_logging_obj=proxy_logging_obj,
         )
     else:
         raise ValueError(f"Unsupported model: {model}")
@@ -193,7 +190,8 @@ async def _realtime_health_check(
         )
     elif custom_llm_provider == "openai":
         url = openai_realtime._construct_url(
-            api_base=api_base or "https://api.openai.com/", query_params={"model": model}
+            api_base=api_base or "https://api.openai.com/",
+            query_params={"model": model},
         )
     else:
         raise ValueError(f"Unsupported model: {model}")

litellm/types/utils.py

@@ -51,6 +51,7 @@ from .llms.openai import (
     OpenAIChatCompletionChunk,
     OpenAIChatCompletionFinishReason,
     OpenAIFileObject,
+    OpenAIRealtimeEvents,
     OpenAIRealtimeStreamList,
     ResponsesAPIResponse,
     WebSearchOptions,
@@ -2637,7 +2638,9 @@ class CostBreakdown(TypedDict, total=False):
     )
     total_cost: float  # Total cost (input + output + tool usage)
     tool_usage_cost: float  # Cost of usage of built-in tools
-    additional_costs: Dict[str, float]  # Free-form additional costs (e.g., {"azure_model_router_flat_cost": 0.00014})
+    additional_costs: Dict[
+        str, float
+    ]  # Free-form additional costs (e.g., {"azure_model_router_flat_cost": 0.00014})
     original_cost: float  # Cost before discount (optional)
     discount_percent: float  # Discount percentage applied (e.g., 0.05 = 5%) (optional)
     discount_amount: float  # Discount amount in USD (optional)
@@ -3350,6 +3353,7 @@ LLMResponseTypes = Union[
     LiteLLMFineTuningJob,
     AnthropicMessagesResponse,
     ResponsesAPIResponse,
+    OpenAIRealtimeEvents,
 ]