RTVIProcessor: add create_rtvi_observer()

src/pipecat/processors/frameworks/rtvi.py

@@ -1413,6 +1413,18 @@ class RTVIProcessor(FrameProcessor):
 
         self._registered_services[service.name] = service
 
+    def create_rtvi_observer(self, *, params: Optional[RTVIObserverParams] = None, **kwargs):
+        """Creates a new RTVI Observer.
+
+        Args:
+            params: Settings to enable/disable specific messages.
+            **kwargs: Additional arguments passed to the observer.
+
+        Returns:
+            A new RTVI observer.
+        """
+        return RTVIObserver(self, params=params, **kwargs)
+
     async def set_client_ready(self):
         """Mark the client as ready and trigger the ready event."""
         self._client_ready = True

src/pipecat/services/google/rtvi.py

@@ -4,7 +4,7 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
-"""Google RTVI integration models and observer implementation.
+"""Google RTVI processor and observer implementation.
 
 This module provides integration with Google's services through the RTVI framework,
 including models for search responses and an observer for handling Google-specific
@@ -16,7 +16,7 @@ from typing import List, Literal, Optional
 from pydantic import BaseModel
 
 from pipecat.observers.base_observer import FramePushed
-from pipecat.processors.frameworks.rtvi import RTVIObserver, RTVIProcessor
+from pipecat.processors.frameworks.rtvi import RTVIObserver, RTVIObserverParams, RTVIProcessor
 from pipecat.services.google.frames import LLMSearchOrigin, LLMSearchResponseFrame
 
 
@@ -86,4 +86,23 @@ class GoogleRTVIObserver(RTVIObserver):
                 rendered_content=frame.rendered_content,
             )
         )
-        await self.push_transport_message_urgent(message)
+        await self.send_rtvi_message(message)
+
+
+class GoogleRTVIProcessor(RTVIProcessor):
+    """RTVI processor for Google service integration.
+
+    Creates a specific Google RTVI Observer.
+    """
+
+    def create_rtvi_observer(self, *, params: Optional[RTVIObserverParams] = None, **kwargs):
+        """Creates a new RTVI Observer.
+
+        Args:
+            params: Settings to enable/disable specific messages.
+            **kwargs: Additional arguments passed to the observer.
+
+        Returns:
+            A new RTVI observer.
+        """
+        return GoogleRTVIObserver(self)