Add a new generic server to client message and frame type

CHANGELOG.md

@@ -5,6 +5,16 @@ All notable changes to **Pipecat** will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Added
+
+- Added a new frame, `ServerMessageFrame`, and RTVI message `RTVIServerMessage`
+  which provides a generic mechanism for sending custom messages from server to
+  client. The `ServerMessageFrame` can be processed by either a `RTVIProcessor`
+  or `RTVIObserver` and will be delivered to the client's `onServerMessage`
+  callback.
+
 ## [0.0.58] - 2025-02-26
 
 ### Added

src/pipecat/frames/frames.py

@@ -568,7 +568,8 @@ class UserStoppedSpeakingFrame(SystemFrame):
 @dataclass
 class EmulateUserStartedSpeakingFrame(SystemFrame):
     """Emitted by internal processors upstream to emulate VAD behavior when a
-    user starts speaking."""
+    user starts speaking.
+    """
 
     pass
 
@@ -576,7 +577,8 @@ class EmulateUserStartedSpeakingFrame(SystemFrame):
 @dataclass
 class EmulateUserStoppedSpeakingFrame(SystemFrame):
     """Emitted by internal processors upstream to emulate VAD behavior when a
-    user stops speaking."""
+    user stops speaking.
+    """
 
     pass
 
@@ -704,6 +706,16 @@ class VisionImageRawFrame(InputImageRawFrame):
         return f"{self.name}(pts: {pts}, text: [{self.text}], size: {self.size}, format: {self.format})"
 
 
+@dataclass
+class ServerMessageFrame(SystemFrame):
+    """A frame for sending server messages to the client."""
+
+    data: Any
+
+    def __str__(self):
+        return f"{self.name}(data: {self.data})"
+
+
 #
 # Control frames
 #

src/pipecat/processors/frameworks/rtvi.py

@@ -38,6 +38,7 @@ from pipecat.frames.frames import (
     LLMFullResponseStartFrame,
     LLMTextFrame,
     MetricsFrame,
+    ServerMessageFrame,
     StartFrame,
     SystemFrame,
     TranscriptionFrame,
@@ -375,6 +376,12 @@ class RTVIMetricsMessage(BaseModel):
     data: Mapping[str, Any]
 
 
+class RTVIServerMessage(BaseModel):
+    label: RTVIMessageLiteral = RTVI_MESSAGE_LABEL
+    type: Literal["server-message"] = "server-message"
+    data: Any
+
+
 class RTVIFrameProcessor(FrameProcessor):
     def __init__(self, direction: FrameDirection = FrameDirection.DOWNSTREAM, **kwargs):
         super().__init__(**kwargs)
@@ -710,6 +717,9 @@ class RTVIObserver(BaseObserver):
                 mark_as_seen = False
         elif isinstance(frame, MetricsFrame):
             await self._handle_metrics(frame)
+        elif isinstance(frame, ServerMessageFrame):
+            message = RTVIServerMessage(data=frame.data)
+            await self.push_transport_message_urgent(message)
 
         if mark_as_seen:
             self._frames_seen.add(frame.id)
@@ -895,6 +905,11 @@ class RTVIProcessor(FrameProcessor):
     async def handle_message(self, message: RTVIMessage):
         await self._message_queue.put(message)
 
+    async def _handle_server_message(self, frame: ServerMessageFrame):
+        """Handle server message frame by converting it to a transport message."""
+        message = RTVIServerMessage(data=frame.data)
+        await self._push_transport_message(message)
+
     async def handle_function_call(
         self,
         function_name: str,
@@ -934,6 +949,9 @@ class RTVIProcessor(FrameProcessor):
             await self.push_frame(frame, direction)
         elif isinstance(frame, TransportMessageUrgentFrame):
             await self._handle_transport_message(frame)
+        elif isinstance(frame, ServerMessageFrame):
+            await self._handle_server_message(frame)
+            await self.push_frame(frame, direction)
         # All other system frames
         elif isinstance(frame, SystemFrame):
             await self.push_frame(frame, direction)