wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

frames: use OutputTransportMessage(Urgent)Frame instead of TransportMessage(Urgent)Frame

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2025-09-25 13:04:31 -07:00
parent c7dc2e886f
commit 4dc1e15a99
16 changed files with 147 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
CHANGELOG.md
View File

@@ -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.

82
src/pipecat/frames/frames.py
View File

@@ -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.

6
src/pipecat/processors/frameworks/rtvi.py
View File

@@ -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):

6
src/pipecat/serializers/exotel.py
View File

@@ -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

6
src/pipecat/serializers/plivo.py
View File

@@ -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

8
src/pipecat/serializers/protobuf.py
View File

@@ -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"])

6
src/pipecat/serializers/twilio.py
View File

@@ -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

12
src/pipecat/transports/base_output.py
View File

@@ -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)

16
src/pipecat/transports/daily/transport.py
View File

@@ -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:

12
src/pipecat/transports/livekit/transport.py
View File

@@ -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:

12
src/pipecat/transports/smallwebrtc/transport.py
View File

@@ -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:

14
src/pipecat/transports/tavus/transport.py
View File

@@ -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:

8
src/pipecat/transports/websocket/client.py
View File

@@ -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:

8
src/pipecat/transports/websocket/fastapi.py
View File

@@ -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:

8
src/pipecat/transports/websocket/server.py
View File

@@ -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:

6
tests/test_frame_processor.py
View File

@@ -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(