From c7dc2e886f42cdf945fa5f4012e66d792b53b38c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 25 Sep 2025 12:07:31 -0700 Subject: [PATCH 1/5] frames: use InputTransportMessageFrame instead of InputTransportMessageUrgentFrame By default, input frames are already urgent. --- CHANGELOG.md | 5 +++ src/pipecat/frames/frames.py | 43 ++++++++++++++----- src/pipecat/processors/frameworks/rtvi.py | 5 ++- src/pipecat/serializers/protobuf.py | 3 +- src/pipecat/transports/base_output.py | 5 +-- src/pipecat/transports/daily/transport.py | 31 +++++++++++-- .../transports/smallwebrtc/transport.py | 4 +- 7 files changed, 74 insertions(+), 22 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d77524a06..70716937b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -32,6 +32,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Fixed an issue where local SmartTurn was not being ran in a separate thread. +### Deprecated + +- `InputTransportMessageUrgentFrame` is deprecated, use + `InputTransportMessageFrame` instead. + ## [0.0.86] - 2025-09-24 ### Added diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index ffef7c99c..f075a6896 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -1106,20 +1106,43 @@ class TransportMessageUrgentFrame(SystemFrame): @dataclass -class InputTransportMessageUrgentFrame(TransportMessageUrgentFrame): +class InputTransportMessageFrame(SystemFrame): """Frame for transport messages received from external sources. - This frame wraps incoming transport messages to distinguish them from outgoing - urgent transport messages (TransportMessageUrgentFrame), preventing infinite - message loops in the transport layer. It inherits the message payload from - TransportMessageFrame while marking the message as having been received - rather than generated locally. - - Used by transport implementations to properly handle bidirectional message - flow without creating feedback loops. + Parameters: + message: The urgent transport message payload. """ - pass + message: Any + + def __str__(self): + return f"{self.name}(message: {self.message})" + + +@dataclass +class InputTransportMessageUrgentFrame(InputTransportMessageFrame): + """Frame for transport messages received from external sources. + + .. deprecated:: 0.0.87 + This frame is deprecated and will be removed in a future version. + Instead, use `InputTransportMessageFrame`. + + Parameters: + message: The urgent transport message payload. + """ + + def __post_init__(self): + super().__post_init__() + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "InputTransportMessageUrgentFrame is deprecated and will be removed in a future version. " + "Instead, use InputTransportMessageFrame.", + DeprecationWarning, + stacklevel=2, + ) @dataclass diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index 9389d819d..3202d66b0 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -42,6 +42,7 @@ from pipecat.frames.frames import ( Frame, FunctionCallResultFrame, InputAudioRawFrame, + InputTransportMessageUrgentFrame, InterimTranscriptionFrame, LLMConfigureOutputFrame, LLMContextFrame, @@ -1418,7 +1419,7 @@ class RTVIProcessor(FrameProcessor): elif isinstance(frame, ErrorFrame): await self._send_error_frame(frame) await self.push_frame(frame, direction) - elif isinstance(frame, TransportMessageUrgentFrame): + elif isinstance(frame, InputTransportMessageUrgentFrame): await self._handle_transport_message(frame) # All other system frames elif isinstance(frame, SystemFrame): @@ -1481,7 +1482,7 @@ class RTVIProcessor(FrameProcessor): await self._handle_message(message) self._message_queue.task_done() - async def _handle_transport_message(self, frame: TransportMessageUrgentFrame): + async def _handle_transport_message(self, frame: InputTransportMessageUrgentFrame): """Handle an incoming transport message frame.""" try: transport_message = frame.message diff --git a/src/pipecat/serializers/protobuf.py b/src/pipecat/serializers/protobuf.py index 867fa0674..8a70bc7eb 100644 --- a/src/pipecat/serializers/protobuf.py +++ b/src/pipecat/serializers/protobuf.py @@ -15,6 +15,7 @@ import pipecat.frames.protobufs.frames_pb2 as frame_protos from pipecat.frames.frames import ( Frame, InputAudioRawFrame, + InputTransportMessageFrame, OutputAudioRawFrame, TextFrame, TranscriptionFrame, @@ -138,7 +139,7 @@ class ProtobufFrameSerializer(FrameSerializer): if class_name == MessageFrame: try: msg = json.loads(args_dict["data"]) - instance = TransportMessageUrgentFrame(message=msg) + instance = InputTransportMessageFrame(message=msg) logger.debug(f"ProtobufFrameSerializer: Transport message {instance}") except Exception as e: logger.error(f"Error parsing MessageFrame data: {e}") diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index 9d1b2b3bb..e51d444e3 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -29,7 +29,6 @@ from pipecat.frames.frames import ( CancelFrame, EndFrame, Frame, - InputTransportMessageUrgentFrame, InterruptionFrame, MixerControlFrame, OutputAudioRawFrame, @@ -307,9 +306,7 @@ class BaseOutputTransport(FrameProcessor): elif isinstance(frame, InterruptionFrame): await self.push_frame(frame, direction) await self._handle_frame(frame) - elif isinstance(frame, TransportMessageUrgentFrame) and not isinstance( - frame, InputTransportMessageUrgentFrame - ): + elif isinstance(frame, TransportMessageUrgentFrame): await self.send_message(frame) elif isinstance(frame, OutputDTMFUrgentFrame): await self.write_dtmf(frame) diff --git a/src/pipecat/transports/daily/transport.py b/src/pipecat/transports/daily/transport.py index f61a799e8..1ffb1b231 100644 --- a/src/pipecat/transports/daily/transport.py +++ b/src/pipecat/transports/daily/transport.py @@ -30,7 +30,7 @@ from pipecat.frames.frames import ( ErrorFrame, Frame, InputAudioRawFrame, - InputTransportMessageUrgentFrame, + InputTransportMessageFrame, InterimTranscriptionFrame, OutputAudioRawFrame, OutputImageRawFrame, @@ -96,7 +96,7 @@ class DailyTransportMessageUrgentFrame(TransportMessageUrgentFrame): @dataclass -class DailyInputTransportMessageUrgentFrame(InputTransportMessageUrgentFrame): +class DailyInputTransportMessageFrame(InputTransportMessageFrame): """Frame for input urgent transport messages in Daily calls. Parameters: @@ -106,6 +106,31 @@ class DailyInputTransportMessageUrgentFrame(InputTransportMessageUrgentFrame): participant_id: Optional[str] = None +class DailyInputTransportMessageUrgentFrame(DailyInputTransportMessageFrame): + """Frame for input urgent transport messages in Daily calls. + + .. deprecated:: 0.0.87 + This frame is deprecated and will be removed in a future version. + Instead, use `DailyInputTransportMessageFrame`. + + Parameters: + participant_id: Optional ID of the participant this message is for/from. + """ + + def __post_init__(self): + super().__post_init__() + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "DailyInputTransportMessageUrgentFrame is deprecated and will be removed in a future version. " + "Instead, use DailyInputTransportMessageFrame.", + DeprecationWarning, + stacklevel=2, + ) + + @dataclass class DailyUpdateRemoteParticipantsFrame(ControlFrame): """Frame to update remote participants in Daily calls. @@ -1621,7 +1646,7 @@ class DailyInputTransport(BaseInputTransport): message: The message data to send. sender: ID of the message sender. """ - frame = DailyInputTransportMessageUrgentFrame(message=message, participant_id=sender) + frame = DailyInputTransportMessageFrame(message=message, participant_id=sender) await self.push_frame(frame) # diff --git a/src/pipecat/transports/smallwebrtc/transport.py b/src/pipecat/transports/smallwebrtc/transport.py index 4b2437be1..17b492d77 100644 --- a/src/pipecat/transports/smallwebrtc/transport.py +++ b/src/pipecat/transports/smallwebrtc/transport.py @@ -26,7 +26,7 @@ from pipecat.frames.frames import ( EndFrame, Frame, InputAudioRawFrame, - InputTransportMessageUrgentFrame, + InputTransportMessageFrame, OutputAudioRawFrame, OutputImageRawFrame, SpriteFrame, @@ -683,7 +683,7 @@ class SmallWebRTCInputTransport(BaseInputTransport): message: The application message to process. """ logger.debug(f"Received app message inside SmallWebRTCInputTransport {message}") - frame = InputTransportMessageUrgentFrame(message=message) + frame = InputTransportMessageFrame(message=message) await self.push_frame(frame) # Add this method similar to DailyInputTransport.request_participant_image From 4dc1e15a99854efe3ae364daa3ed85160c1c681a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 25 Sep 2025 13:04:31 -0700 Subject: [PATCH 2/5] frames: use OutputTransportMessage(Urgent)Frame instead of TransportMessage(Urgent)Frame --- CHANGELOG.md | 4 + src/pipecat/frames/frames.py | 82 +++++++++++++++---- src/pipecat/processors/frameworks/rtvi.py | 6 +- src/pipecat/serializers/exotel.py | 6 +- src/pipecat/serializers/plivo.py | 6 +- src/pipecat/serializers/protobuf.py | 8 +- src/pipecat/serializers/twilio.py | 6 +- src/pipecat/transports/base_output.py | 12 +-- src/pipecat/transports/daily/transport.py | 16 ++-- src/pipecat/transports/livekit/transport.py | 12 +-- .../transports/smallwebrtc/transport.py | 12 ++- src/pipecat/transports/tavus/transport.py | 14 ++-- src/pipecat/transports/websocket/client.py | 8 +- src/pipecat/transports/websocket/fastapi.py | 8 +- src/pipecat/transports/websocket/server.py | 8 +- tests/test_frame_processor.py | 6 +- 16 files changed, 147 insertions(+), 67 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 70716937b..c7868b186 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -34,6 +34,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Deprecated +- `TransportMessageFrame` and `TransportMessageUrgentFrame` are deprecated, use + `OutputTransportMessageFrame` and `OutputTransportMessageUrgentFrame` + respectively. + - `InputTransportMessageUrgentFrame` is deprecated, use `InputTransportMessageFrame` instead. diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index f075a6896..2a9651b83 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -672,7 +672,7 @@ class TTSSpeakFrame(DataFrame): @dataclass -class TransportMessageFrame(DataFrame): +class OutputTransportMessageFrame(DataFrame): """Frame containing transport-specific message data. Parameters: @@ -685,6 +685,32 @@ class TransportMessageFrame(DataFrame): return f"{self.name}(message: {self.message})" +@dataclass +class TransportMessageFrame(OutputTransportMessageFrame): + """Frame containing transport-specific message data. + + .. deprecated:: 0.0.87 + This frame is deprecated and will be removed in a future version. + Instead, use `OutputTransportMessageFrame`. + + Parameters: + message: The transport message payload. + """ + + def __post_init__(self): + super().__post_init__() + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "TransportMessageFrame is deprecated and will be removed in a future version. " + "Instead, use OutputTransportMessageFrame.", + DeprecationWarning, + stacklevel=2, + ) + + @dataclass class DTMFFrame: """Base class for DTMF (Dual-Tone Multi-Frequency) keypad frames. @@ -1091,20 +1117,6 @@ class STTMuteFrame(SystemFrame): mute: bool -@dataclass -class TransportMessageUrgentFrame(SystemFrame): - """Frame for urgent transport messages that need immediate processing. - - Parameters: - message: The urgent transport message payload. - """ - - message: Any - - def __str__(self): - return f"{self.name}(message: {self.message})" - - @dataclass class InputTransportMessageFrame(SystemFrame): """Frame for transport messages received from external sources. @@ -1145,6 +1157,46 @@ class InputTransportMessageUrgentFrame(InputTransportMessageFrame): ) +@dataclass +class OutputTransportMessageUrgentFrame(SystemFrame): + """Frame for urgent transport messages that need to be sent immediately. + + Parameters: + message: The urgent transport message payload. + """ + + message: Any + + def __str__(self): + return f"{self.name}(message: {self.message})" + + +@dataclass +class TransportMessageUrgentFrame(OutputTransportMessageUrgentFrame): + """Frame for urgent transport messages that need to be sent immediately. + + .. deprecated:: 0.0.87 + This frame is deprecated and will be removed in a future version. + Instead, use `OutputTransportMessageUrgentFrame`. + + Parameters: + message: The urgent transport message payload. + """ + + def __post_init__(self): + super().__post_init__() + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "TransportMessageUrgentFrame is deprecated and will be removed in a future version. " + "Instead, use OutputTransportMessageFrame.", + DeprecationWarning, + stacklevel=2, + ) + + @dataclass class UserImageRequestFrame(SystemFrame): """Frame requesting an image from a specific user. diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index 3202d66b0..4d1f8aeae 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -51,10 +51,10 @@ from pipecat.frames.frames import ( LLMMessagesAppendFrame, LLMTextFrame, MetricsFrame, + OutputTransportMessageUrgentFrame, StartFrame, SystemFrame, TranscriptionFrame, - TransportMessageUrgentFrame, TTSAudioRawFrame, TTSStartedFrame, TTSStoppedFrame, @@ -1346,7 +1346,9 @@ class RTVIProcessor(FrameProcessor): async def push_transport_message(self, model: BaseModel, exclude_none: bool = True): """Push a transport message frame.""" - frame = TransportMessageUrgentFrame(message=model.model_dump(exclude_none=exclude_none)) + frame = OutputTransportMessageUrgentFrame( + message=model.model_dump(exclude_none=exclude_none) + ) await self.push_frame(frame) async def handle_message(self, message: RTVIMessage): diff --git a/src/pipecat/serializers/exotel.py b/src/pipecat/serializers/exotel.py index 1a5859211..3922e5a17 100644 --- a/src/pipecat/serializers/exotel.py +++ b/src/pipecat/serializers/exotel.py @@ -21,9 +21,9 @@ from pipecat.frames.frames import ( InputAudioRawFrame, InputDTMFFrame, InterruptionFrame, + OutputTransportMessageFrame, + OutputTransportMessageUrgentFrame, StartFrame, - TransportMessageFrame, - TransportMessageUrgentFrame, ) from pipecat.serializers.base_serializer import FrameSerializer, FrameSerializerType @@ -121,7 +121,7 @@ class ExotelFrameSerializer(FrameSerializer): } return json.dumps(answer) - elif isinstance(frame, (TransportMessageFrame, TransportMessageUrgentFrame)): + elif isinstance(frame, (OutputTransportMessageFrame, OutputTransportMessageUrgentFrame)): return json.dumps(frame.message) return None diff --git a/src/pipecat/serializers/plivo.py b/src/pipecat/serializers/plivo.py index 519e0893a..51ae11151 100644 --- a/src/pipecat/serializers/plivo.py +++ b/src/pipecat/serializers/plivo.py @@ -23,9 +23,9 @@ from pipecat.frames.frames import ( InputAudioRawFrame, InputDTMFFrame, InterruptionFrame, + OutputTransportMessageFrame, + OutputTransportMessageUrgentFrame, StartFrame, - TransportMessageFrame, - TransportMessageUrgentFrame, ) from pipecat.serializers.base_serializer import FrameSerializer, FrameSerializerType @@ -148,7 +148,7 @@ class PlivoFrameSerializer(FrameSerializer): } return json.dumps(answer) - elif isinstance(frame, (TransportMessageFrame, TransportMessageUrgentFrame)): + elif isinstance(frame, (OutputTransportMessageFrame, OutputTransportMessageUrgentFrame)): return json.dumps(frame.message) # Return None for unhandled frames diff --git a/src/pipecat/serializers/protobuf.py b/src/pipecat/serializers/protobuf.py index 8a70bc7eb..803de0575 100644 --- a/src/pipecat/serializers/protobuf.py +++ b/src/pipecat/serializers/protobuf.py @@ -17,10 +17,10 @@ from pipecat.frames.frames import ( InputAudioRawFrame, InputTransportMessageFrame, OutputAudioRawFrame, + OutputTransportMessageFrame, + OutputTransportMessageUrgentFrame, TextFrame, TranscriptionFrame, - TransportMessageFrame, - TransportMessageUrgentFrame, ) from pipecat.serializers.base_serializer import FrameSerializer, FrameSerializerType @@ -83,7 +83,7 @@ class ProtobufFrameSerializer(FrameSerializer): Serialized frame as bytes, or None if frame type is not serializable. """ # Wrapping this messages as a JSONFrame to send - if isinstance(frame, (TransportMessageFrame, TransportMessageUrgentFrame)): + if isinstance(frame, (OutputTransportMessageFrame, OutputTransportMessageUrgentFrame)): frame = MessageFrame( data=json.dumps(frame.message), ) @@ -135,7 +135,7 @@ class ProtobufFrameSerializer(FrameSerializer): if "pts" in args_dict: del args_dict["pts"] - # Special handling for MessageFrame -> TransportMessageUrgentFrame + # Special handling for MessageFrame -> OutputTransportMessageUrgentFrame if class_name == MessageFrame: try: msg = json.loads(args_dict["data"]) diff --git a/src/pipecat/serializers/twilio.py b/src/pipecat/serializers/twilio.py index aaeb07ec3..d61b29295 100644 --- a/src/pipecat/serializers/twilio.py +++ b/src/pipecat/serializers/twilio.py @@ -23,9 +23,9 @@ from pipecat.frames.frames import ( InputAudioRawFrame, InputDTMFFrame, InterruptionFrame, + OutputTransportMessageFrame, + OutputTransportMessageUrgentFrame, StartFrame, - TransportMessageFrame, - TransportMessageUrgentFrame, ) from pipecat.serializers.base_serializer import FrameSerializer, FrameSerializerType @@ -175,7 +175,7 @@ class TwilioFrameSerializer(FrameSerializer): } return json.dumps(answer) - elif isinstance(frame, (TransportMessageFrame, TransportMessageUrgentFrame)): + elif isinstance(frame, (OutputTransportMessageFrame, OutputTransportMessageUrgentFrame)): return json.dumps(frame.message) # Return None for unhandled frames diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index e51d444e3..9d4630100 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -35,13 +35,13 @@ from pipecat.frames.frames import ( OutputDTMFFrame, OutputDTMFUrgentFrame, OutputImageRawFrame, + OutputTransportMessageFrame, + OutputTransportMessageUrgentFrame, OutputTransportReadyFrame, SpeechOutputAudioRawFrame, SpriteFrame, StartFrame, SystemFrame, - TransportMessageFrame, - TransportMessageUrgentFrame, TTSAudioRawFrame, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor @@ -177,7 +177,9 @@ class BaseOutputTransport(FrameProcessor): # Sending a frame indicating that the output transport is ready and able to receive frames. await self.push_frame(OutputTransportReadyFrame(), FrameDirection.UPSTREAM) - async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame): + async def send_message( + self, frame: OutputTransportMessageFrame | OutputTransportMessageUrgentFrame + ): """Send a transport message. Args: @@ -306,7 +308,7 @@ class BaseOutputTransport(FrameProcessor): elif isinstance(frame, InterruptionFrame): await self.push_frame(frame, direction) await self._handle_frame(frame) - elif isinstance(frame, TransportMessageUrgentFrame): + elif isinstance(frame, OutputTransportMessageUrgentFrame): await self.send_message(frame) elif isinstance(frame, OutputDTMFUrgentFrame): await self.write_dtmf(frame) @@ -643,7 +645,7 @@ class BaseOutputTransport(FrameProcessor): await self._set_video_image(frame) elif isinstance(frame, SpriteFrame): await self._set_video_images(frame.images) - elif isinstance(frame, TransportMessageFrame): + elif isinstance(frame, OutputTransportMessageFrame): await self._transport.send_message(frame) elif isinstance(frame, OutputDTMFFrame): await self._transport.write_dtmf(frame) diff --git a/src/pipecat/transports/daily/transport.py b/src/pipecat/transports/daily/transport.py index 1ffb1b231..787216341 100644 --- a/src/pipecat/transports/daily/transport.py +++ b/src/pipecat/transports/daily/transport.py @@ -34,11 +34,11 @@ from pipecat.frames.frames import ( InterimTranscriptionFrame, OutputAudioRawFrame, OutputImageRawFrame, + OutputTransportMessageFrame, + OutputTransportMessageUrgentFrame, SpriteFrame, StartFrame, TranscriptionFrame, - TransportMessageFrame, - TransportMessageUrgentFrame, UserAudioRawFrame, UserImageRawFrame, UserImageRequestFrame, @@ -74,7 +74,7 @@ VAD_RESET_PERIOD_MS = 2000 @dataclass -class DailyTransportMessageFrame(TransportMessageFrame): +class DailyTransportMessageFrame(OutputTransportMessageFrame): """Frame for transport messages in Daily calls. Parameters: @@ -85,7 +85,7 @@ class DailyTransportMessageFrame(TransportMessageFrame): @dataclass -class DailyTransportMessageUrgentFrame(TransportMessageUrgentFrame): +class DailyTransportMessageUrgentFrame(OutputTransportMessageUrgentFrame): """Frame for urgent transport messages in Daily calls. Parameters: @@ -499,7 +499,9 @@ class DailyTransportClient(EventHandler): """ return self._out_sample_rate - async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame): + async def send_message( + self, frame: OutputTransportMessageFrame | OutputTransportMessageUrgentFrame + ): """Send an application message to participants. Args: @@ -1868,7 +1870,9 @@ class DailyOutputTransport(BaseOutputTransport): if isinstance(frame, DailyUpdateRemoteParticipantsFrame): await self._client.update_remote_participants(frame.remote_participants) - async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame): + async def send_message( + self, frame: OutputTransportMessageFrame | OutputTransportMessageUrgentFrame + ): """Send a transport message to participants. Args: diff --git a/src/pipecat/transports/livekit/transport.py b/src/pipecat/transports/livekit/transport.py index 334d48ecb..cb06464f0 100644 --- a/src/pipecat/transports/livekit/transport.py +++ b/src/pipecat/transports/livekit/transport.py @@ -29,9 +29,9 @@ from pipecat.frames.frames import ( OutputAudioRawFrame, OutputDTMFFrame, OutputDTMFUrgentFrame, + OutputTransportMessageFrame, + OutputTransportMessageUrgentFrame, StartFrame, - TransportMessageFrame, - TransportMessageUrgentFrame, UserAudioRawFrame, UserImageRawFrame, ) @@ -68,7 +68,7 @@ DTMF_CODE_MAP = { @dataclass -class LiveKitTransportMessageFrame(TransportMessageFrame): +class LiveKitTransportMessageFrame(OutputTransportMessageFrame): """Frame for transport messages in LiveKit rooms. Parameters: @@ -79,7 +79,7 @@ class LiveKitTransportMessageFrame(TransportMessageFrame): @dataclass -class LiveKitTransportMessageUrgentFrame(TransportMessageUrgentFrame): +class LiveKitTransportMessageUrgentFrame(OutputTransportMessageUrgentFrame): """Frame for urgent transport messages in LiveKit rooms. Parameters: @@ -836,7 +836,9 @@ class LiveKitOutputTransport(BaseOutputTransport): await super().cleanup() await self._transport.cleanup() - async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame): + async def send_message( + self, frame: OutputTransportMessageFrame | OutputTransportMessageUrgentFrame + ): """Send a transport message to participants. Args: diff --git a/src/pipecat/transports/smallwebrtc/transport.py b/src/pipecat/transports/smallwebrtc/transport.py index 17b492d77..cdefa976f 100644 --- a/src/pipecat/transports/smallwebrtc/transport.py +++ b/src/pipecat/transports/smallwebrtc/transport.py @@ -29,10 +29,10 @@ from pipecat.frames.frames import ( InputTransportMessageFrame, OutputAudioRawFrame, OutputImageRawFrame, + OutputTransportMessageFrame, + OutputTransportMessageUrgentFrame, SpriteFrame, StartFrame, - TransportMessageFrame, - TransportMessageUrgentFrame, UserImageRawFrame, UserImageRequestFrame, ) @@ -461,7 +461,9 @@ class SmallWebRTCClient: await self._webrtc_connection.disconnect() await self._handle_peer_disconnected() - async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame): + async def send_message( + self, frame: OutputTransportMessageFrame | OutputTransportMessageUrgentFrame + ): """Send an application message through the WebRTC connection. Args: @@ -820,7 +822,9 @@ class SmallWebRTCOutputTransport(BaseOutputTransport): await super().cancel(frame) await self._client.disconnect() - async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame): + async def send_message( + self, frame: OutputTransportMessageFrame | OutputTransportMessageUrgentFrame + ): """Send a transport message through the WebRTC connection. Args: diff --git a/src/pipecat/transports/tavus/transport.py b/src/pipecat/transports/tavus/transport.py index f6da826fd..40ffeeed7 100644 --- a/src/pipecat/transports/tavus/transport.py +++ b/src/pipecat/transports/tavus/transport.py @@ -27,9 +27,9 @@ from pipecat.frames.frames import ( InputAudioRawFrame, InterruptionFrame, OutputAudioRawFrame, + OutputTransportMessageFrame, + OutputTransportMessageUrgentFrame, StartFrame, - TransportMessageFrame, - TransportMessageUrgentFrame, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, FrameProcessorSetup from pipecat.transports.base_input import BaseInputTransport @@ -345,7 +345,9 @@ class TavusTransportClient: participant_id, callback, audio_source, sample_rate, callback_interval_ms ) - async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame): + async def send_message( + self, frame: OutputTransportMessageFrame | OutputTransportMessageUrgentFrame + ): """Send a message to participants. Args: @@ -373,7 +375,7 @@ class TavusTransportClient: async def send_interrupt_message(self) -> None: """Send an interrupt message to the conversation.""" - transport_frame = TransportMessageUrgentFrame( + transport_frame = OutputTransportMessageUrgentFrame( message={ "message_type": "conversation", "event_type": "conversation.interrupt", @@ -605,7 +607,9 @@ class TavusOutputTransport(BaseOutputTransport): await super().cancel(frame) await self._client.stop() - async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame): + async def send_message( + self, frame: OutputTransportMessageFrame | OutputTransportMessageUrgentFrame + ): """Send a message to participants. Args: diff --git a/src/pipecat/transports/websocket/client.py b/src/pipecat/transports/websocket/client.py index c18c0d22a..966b78e00 100644 --- a/src/pipecat/transports/websocket/client.py +++ b/src/pipecat/transports/websocket/client.py @@ -28,9 +28,9 @@ from pipecat.frames.frames import ( Frame, InputAudioRawFrame, OutputAudioRawFrame, + OutputTransportMessageFrame, + OutputTransportMessageUrgentFrame, StartFrame, - TransportMessageFrame, - TransportMessageUrgentFrame, ) from pipecat.processors.frame_processor import FrameProcessorSetup from pipecat.serializers.base_serializer import FrameSerializer @@ -385,7 +385,9 @@ class WebsocketClientOutputTransport(BaseOutputTransport): await super().cleanup() await self._transport.cleanup() - async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame): + async def send_message( + self, frame: OutputTransportMessageFrame | OutputTransportMessageUrgentFrame + ): """Send a transport message through the WebSocket. Args: diff --git a/src/pipecat/transports/websocket/fastapi.py b/src/pipecat/transports/websocket/fastapi.py index 72654204a..28d08abcd 100644 --- a/src/pipecat/transports/websocket/fastapi.py +++ b/src/pipecat/transports/websocket/fastapi.py @@ -28,9 +28,9 @@ from pipecat.frames.frames import ( InputAudioRawFrame, InterruptionFrame, OutputAudioRawFrame, + OutputTransportMessageFrame, + OutputTransportMessageUrgentFrame, StartFrame, - TransportMessageFrame, - TransportMessageUrgentFrame, ) from pipecat.processors.frame_processor import FrameDirection from pipecat.serializers.base_serializer import FrameSerializer, FrameSerializerType @@ -402,7 +402,9 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport): await self._write_frame(frame) self._next_send_time = 0 - async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame): + async def send_message( + self, frame: OutputTransportMessageFrame | OutputTransportMessageUrgentFrame + ): """Send a transport message frame. Args: diff --git a/src/pipecat/transports/websocket/server.py b/src/pipecat/transports/websocket/server.py index 095407422..94743ce66 100644 --- a/src/pipecat/transports/websocket/server.py +++ b/src/pipecat/transports/websocket/server.py @@ -27,9 +27,9 @@ from pipecat.frames.frames import ( InputAudioRawFrame, InterruptionFrame, OutputAudioRawFrame, + OutputTransportMessageFrame, + OutputTransportMessageUrgentFrame, StartFrame, - TransportMessageFrame, - TransportMessageUrgentFrame, ) from pipecat.processors.frame_processor import FrameDirection from pipecat.serializers.base_serializer import FrameSerializer @@ -338,7 +338,9 @@ class WebsocketServerOutputTransport(BaseOutputTransport): await self._write_frame(frame) self._next_send_time = 0 - async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame): + async def send_message( + self, frame: OutputTransportMessageFrame | OutputTransportMessageUrgentFrame + ): """Send a transport message frame to the client. Args: diff --git a/tests/test_frame_processor.py b/tests/test_frame_processor.py index f514c94b0..d0072e5fb 100644 --- a/tests/test_frame_processor.py +++ b/tests/test_frame_processor.py @@ -11,8 +11,8 @@ from pipecat.frames.frames import ( EndFrame, Frame, InterruptionFrame, + OutputTransportMessageUrgentFrame, TextFrame, - TransportMessageUrgentFrame, ) from pipecat.pipeline.pipeline import Pipeline from pipecat.processors.filters.identity_filter import IdentityFilter @@ -81,7 +81,7 @@ class TestFrameProcessor(unittest.IsolatedAsyncioTestCase): if isinstance(frame, TextFrame): await self.push_interruption_task_frame_and_wait() - await self.push_frame(TransportMessageUrgentFrame(message=frame.text)) + await self.push_frame(OutputTransportMessageUrgentFrame(message=frame.text)) else: await self.push_frame(frame, direction) @@ -101,7 +101,7 @@ class TestFrameProcessor(unittest.IsolatedAsyncioTestCase): expected_down_frames = [ InterruptionFrame, InterruptionFrame, - TransportMessageUrgentFrame, + OutputTransportMessageUrgentFrame, EndFrame, ] await run_test( From 089e703e1f2c6011b9655108fe74895edb0ff1af Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 25 Sep 2025 14:22:34 -0700 Subject: [PATCH 3/5] LiveKitTransport: deprecate LiveKitTransportMessage(Urgent)Frame --- CHANGELOG.md | 6 ++- src/pipecat/transports/livekit/transport.py | 60 ++++++++++++++++++--- 2 files changed, 59 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c7868b186..7a0918b0b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -34,9 +34,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Deprecated +- `LiveKitTransportMessageFrame` and `LiveKitTransportMessageUrgentFrame` are + deprecated, use `LiveKitOutputTransportMessageFrame` and + `LiveKitOutputTransportMessageUrgentFrame` respectively instead. + - `TransportMessageFrame` and `TransportMessageUrgentFrame` are deprecated, use `OutputTransportMessageFrame` and `OutputTransportMessageUrgentFrame` - respectively. + respectively instead. - `InputTransportMessageUrgentFrame` is deprecated, use `InputTransportMessageFrame` instead. diff --git a/src/pipecat/transports/livekit/transport.py b/src/pipecat/transports/livekit/transport.py index cb06464f0..f1b8d9088 100644 --- a/src/pipecat/transports/livekit/transport.py +++ b/src/pipecat/transports/livekit/transport.py @@ -68,7 +68,7 @@ DTMF_CODE_MAP = { @dataclass -class LiveKitTransportMessageFrame(OutputTransportMessageFrame): +class LiveKitOutputTransportMessageFrame(OutputTransportMessageFrame): """Frame for transport messages in LiveKit rooms. Parameters: @@ -79,7 +79,7 @@ class LiveKitTransportMessageFrame(OutputTransportMessageFrame): @dataclass -class LiveKitTransportMessageUrgentFrame(OutputTransportMessageUrgentFrame): +class LiveKitOutputTransportMessageUrgentFrame(OutputTransportMessageUrgentFrame): """Frame for urgent transport messages in LiveKit rooms. Parameters: @@ -89,6 +89,50 @@ class LiveKitTransportMessageUrgentFrame(OutputTransportMessageUrgentFrame): participant_id: Optional[str] = None +@dataclass +class LiveKitTransportMessageFrame(LiveKitOutputTransportMessageFrame): + """Frame for transport messages in LiveKit rooms. + + Parameters: + participant_id: Optional ID of the participant this message is for/from. + """ + + def __post_init__(self): + super().__post_init__() + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "LiveKitTransportMessageFrame is deprecated and will be removed in a future version. " + "Instead, use LiveKitOutputTransportMessageFrame.", + DeprecationWarning, + stacklevel=2, + ) + + +@dataclass +class LiveKitTransportMessageUrgentFrame(LiveKitOutputTransportMessageUrgentFrame): + """Frame for urgent transport messages in LiveKit rooms. + + Parameters: + participant_id: Optional ID of the participant this message is for/from. + """ + + def __post_init__(self): + super().__post_init__() + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "LiveKitTransportMessageUrgentFrame is deprecated and will be removed in a future version. " + "Instead, use LiveKitOutputTransportMessageUrgentFrame.", + DeprecationWarning, + stacklevel=2, + ) + + class LiveKitParams(TransportParams): """Configuration parameters for LiveKit transport. @@ -677,7 +721,7 @@ class LiveKitInputTransport(BaseInputTransport): message: The message data to send. sender: ID of the message sender. """ - frame = LiveKitTransportMessageUrgentFrame(message=message, participant_id=sender) + frame = LiveKitOutputTransportMessageUrgentFrame(message=message, participant_id=sender) await self.push_frame(frame) async def _audio_in_task_handler(self): @@ -848,7 +892,9 @@ class LiveKitOutputTransport(BaseOutputTransport): if isinstance(message, dict): # fix message encoding for dict-like messages, e.g. RTVI messages. message = json.dumps(message, ensure_ascii=False) - if isinstance(frame, (LiveKitTransportMessageFrame, LiveKitTransportMessageUrgentFrame)): + if isinstance( + frame, (LiveKitOutputTransportMessageFrame, LiveKitOutputTransportMessageUrgentFrame) + ): await self._client.send_data(message.encode(), frame.participant_id) else: await self._client.send_data(message.encode()) @@ -1107,7 +1153,9 @@ class LiveKitTransport(BaseTransport): participant_id: Optional specific participant to send to. """ if self._output: - frame = LiveKitTransportMessageFrame(message=message, participant_id=participant_id) + frame = LiveKitOutputTransportMessageFrame( + message=message, participant_id=participant_id + ) await self._output.send_message(frame) async def send_message_urgent(self, message: str, participant_id: Optional[str] = None): @@ -1118,7 +1166,7 @@ class LiveKitTransport(BaseTransport): participant_id: Optional specific participant to send to. """ if self._output: - frame = LiveKitTransportMessageUrgentFrame( + frame = LiveKitOutputTransportMessageUrgentFrame( message=message, participant_id=participant_id ) await self._output.send_message(frame) From be562cedfc01d9f94f7ae34a689076eb4ca2e493 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 25 Sep 2025 14:23:39 -0700 Subject: [PATCH 4/5] DailyTransport: deprecate DailyTransportMessage(Urgent)Frame --- CHANGELOG.md | 4 ++ .../16-gpu-container-local-bot.py | 12 ++-- src/pipecat/transports/daily/transport.py | 60 ++++++++++++++++++- 3 files changed, 69 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 7a0918b0b..0a50cc8ba 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -34,6 +34,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Deprecated +- `DailyTransportMessageFrame` and `DailyTransportMessageUrgentFrame` are + deprecated, use `DailyOutputTransportMessageFrame` and + `DailyOutputTransportMessageUrgentFrame` respectively instead. + - `LiveKitTransportMessageFrame` and `LiveKitTransportMessageUrgentFrame` are deprecated, use `LiveKitOutputTransportMessageFrame` and `LiveKitOutputTransportMessageUrgentFrame` respectively instead. diff --git a/examples/foundational/16-gpu-container-local-bot.py b/examples/foundational/16-gpu-container-local-bot.py index 89316e63c..90ff271c6 100644 --- a/examples/foundational/16-gpu-container-local-bot.py +++ b/examples/foundational/16-gpu-container-local-bot.py @@ -26,7 +26,11 @@ from pipecat.services.deepgram.stt import DeepgramSTTService from pipecat.services.deepgram.tts import DeepgramTTSService from pipecat.services.openai.llm import OpenAILLMService from pipecat.transports.base_transport import BaseTransport, TransportParams -from pipecat.transports.daily.transport import DailyParams, DailyTransportMessageFrame +from pipecat.transports.daily.transport import ( + DailyOutputTransportMessageFrame, + DailyOutputTransportMessageUrgentFrame, + DailyParams, +) from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams load_dotenv(override=True) @@ -128,14 +132,14 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): logger.debug(f"Received latency ping app message: {message}") ts = message["latency-ping"]["ts"] # Send immediately - transport.output().send_message( - DailyTransportMessageFrame( + await task.queue_frame( + DailyOutputTransportMessageUrgentFrame( message={"latency-pong-msg-handler": {"ts": ts}}, participant_id=sender ) ) # And push to the pipeline for the Daily transport.output to send await task.queue_frame( - DailyTransportMessageFrame( + DailyOutputTransportMessageFrame( message={"latency-pong-pipeline-delivery": {"ts": ts}}, participant_id=sender, ) diff --git a/src/pipecat/transports/daily/transport.py b/src/pipecat/transports/daily/transport.py index 787216341..7f6b21ee2 100644 --- a/src/pipecat/transports/daily/transport.py +++ b/src/pipecat/transports/daily/transport.py @@ -74,7 +74,7 @@ VAD_RESET_PERIOD_MS = 2000 @dataclass -class DailyTransportMessageFrame(OutputTransportMessageFrame): +class DailyOutputTransportMessageFrame(OutputTransportMessageFrame): """Frame for transport messages in Daily calls. Parameters: @@ -85,7 +85,7 @@ class DailyTransportMessageFrame(OutputTransportMessageFrame): @dataclass -class DailyTransportMessageUrgentFrame(OutputTransportMessageUrgentFrame): +class DailyOutputTransportMessageUrgentFrame(OutputTransportMessageUrgentFrame): """Frame for urgent transport messages in Daily calls. Parameters: @@ -95,6 +95,58 @@ class DailyTransportMessageUrgentFrame(OutputTransportMessageUrgentFrame): participant_id: Optional[str] = None +@dataclass +class DailyTransportMessageFrame(DailyOutputTransportMessageFrame): + """Frame for transport messages in Daily calls. + + .. deprecated:: 0.0.87 + This frame is deprecated and will be removed in a future version. + Instead, use `DailyOutputTransportMessageFrame`. + + Parameters: + participant_id: Optional ID of the participant this message is for/from. + """ + + def __post_init__(self): + super().__post_init__() + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "DailyTransportMessageFrame is deprecated and will be removed in a future version. " + "Instead, use DailyOutputTransportMessageFrame.", + DeprecationWarning, + stacklevel=2, + ) + + +@dataclass +class DailyTransportMessageUrgentFrame(DailyOutputTransportMessageUrgentFrame): + """Frame for urgent transport messages in Daily calls. + + .. deprecated:: 0.0.87 + This frame is deprecated and will be removed in a future version. + Instead, use `DailyOutputTransportMessageUrgentFrame`. + + Parameters: + participant_id: Optional ID of the participant this message is for/from. + """ + + def __post_init__(self): + super().__post_init__() + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "DailyTransportMessageUrgentFrame is deprecated and will be removed in a future version. " + "Instead, use DailyOutputTransportMessageUrgentFrame.", + DeprecationWarning, + stacklevel=2, + ) + + @dataclass class DailyInputTransportMessageFrame(InputTransportMessageFrame): """Frame for input urgent transport messages in Daily calls. @@ -511,7 +563,9 @@ class DailyTransportClient(EventHandler): return participant_id = None - if isinstance(frame, (DailyTransportMessageFrame, DailyTransportMessageUrgentFrame)): + if isinstance( + frame, (DailyOutputTransportMessageFrame, DailyOutputTransportMessageUrgentFrame) + ): participant_id = frame.participant_id future = self._get_event_loop().create_future() From ad507ce23da59f93acaf821838582d53423bb4b8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 1 Oct 2025 15:29:30 -0700 Subject: [PATCH 5/5] FrameLogger: it's fine to print transport messages --- src/pipecat/processors/logger.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/pipecat/processors/logger.py b/src/pipecat/processors/logger.py index acc85c40b..887ce71fe 100644 --- a/src/pipecat/processors/logger.py +++ b/src/pipecat/processors/logger.py @@ -15,7 +15,7 @@ from pipecat.frames.frames import ( Frame, InputAudioRawFrame, OutputAudioRawFrame, - TransportMessageFrame, + UserSpeakingFrame, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor @@ -36,9 +36,9 @@ class FrameLogger(FrameProcessor): color: Optional[str] = None, ignored_frame_types: Tuple[Type[Frame], ...] = ( BotSpeakingFrame, + UserSpeakingFrame, InputAudioRawFrame, OutputAudioRawFrame, - TransportMessageFrame, ), ): """Initialize the frame logger.