transports: add support for sending transport messages

src/pipecat/transports/base_output.py

@@ -21,7 +21,8 @@ from pipecat.frames.frames import (
     StartFrame,
     EndFrame,
     Frame,
-    ImageRawFrame)
+    ImageRawFrame,
+    TransportMessageFrame)
 from pipecat.transports.base_transport import TransportParams
 
 from loguru import logger
@@ -60,6 +61,9 @@ class BaseOutputTransport(FrameProcessor):
 
         self._stopped_event.set()
 
+    def send_message(self, frame: TransportMessageFrame):
+        pass
+
     def write_frame_to_camera(self, frame: ImageRawFrame):
         pass
 
@@ -98,6 +102,7 @@ class BaseOutputTransport(FrameProcessor):
         return (isinstance(frame, AudioRawFrame)
                 or isinstance(frame, ImageRawFrame)
                 or isinstance(frame, SpriteFrame)
+                or isinstance(frame, TransportMessageFrame)
                 or isinstance(frame, CancelFrame)
                 or isinstance(frame, EndFrame))
 
@@ -121,6 +126,8 @@ class BaseOutputTransport(FrameProcessor):
                     self._set_camera_image(frame)
                 elif isinstance(frame, SpriteFrame) and self._params.camera_out_enabled:
                     self._set_camera_images(frame.images)
+                elif isinstance(frame, TransportMessageFrame):
+                    self.send_message(frame)
             except queue.Empty:
                 pass
             except BaseException as e:

src/pipecat/transports/services/daily.py

@@ -11,6 +11,7 @@ import threading
 import time
 import types
 
+from dataclasses import dataclass
 from functools import partial
 from typing import Any, Callable, Mapping
 
@@ -30,6 +31,7 @@ from pipecat.frames.frames import (
     InterimTranscriptionFrame,
     SpriteFrame,
     TranscriptionFrame,
+    TransportMessageFrame,
     UserImageRawFrame,
     UserImageRequestFrame)
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
@@ -51,6 +53,11 @@ except ModuleNotFoundError as e:
 VAD_RESET_PERIOD_MS = 2000
 
 
+@dataclass
+class DailyTransportMessageFrame(TransportMessageFrame):
+    participant_id: str | None = None
+
+
 class WebRTCVADAnalyzer(VADAnalyzer):
 
     def __init__(self, sample_rate=16000, num_channels=1):
@@ -169,6 +176,9 @@ class DailySession(EventHandler):
             state = self._vad_analyzer.analyze_audio(audio_frames)
         return state
 
+    def send_message(self, frame: DailyTransportMessageFrame):
+        self._client.send_app_message(frame.message, frame.participant_id)
+
     def read_raw_audio_frames(self, frame_count: int) -> bytes:
         return self._speaker.read_frames(frame_count)