Merge pull request #1303 from pipecat-ai/mb/add-server-to-client-msg

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` is processed by the `RTVIObserver` and will
+  be delivered to the client's `onServerMessage` callback or `ServerMessage`
+  event.
+
 ## [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)