From e96595fe59a41109c9f722f2999ad7127584386f Mon Sep 17 00:00:00 2001 From: Varun Pratap Singh Date: Mon, 12 Jan 2026 17:50:38 +0530 Subject: [PATCH 1/3] feat: update FastAPI WebSocket transport and add Vonage serializer --- src/pipecat/serializers/vonage.py | 182 ++++++++++++++++++++ src/pipecat/transports/websocket/fastapi.py | 30 ++++ 2 files changed, 212 insertions(+) create mode 100644 src/pipecat/serializers/vonage.py diff --git a/src/pipecat/serializers/vonage.py b/src/pipecat/serializers/vonage.py new file mode 100644 index 000000000..9de1cc038 --- /dev/null +++ b/src/pipecat/serializers/vonage.py @@ -0,0 +1,182 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Vonage Audio Connector WebSocket serializer for Pipecat.""" + +import json +from typing import Optional + +from loguru import logger +from pydantic import BaseModel + +from pipecat.audio.dtmf.types import KeypadEntry +from pipecat.audio.utils import create_stream_resampler +from pipecat.frames.frames import ( + AudioRawFrame, + Frame, + InputAudioRawFrame, + InputDTMFFrame, + InterruptionFrame, + OutputTransportMessageFrame, + OutputTransportMessageUrgentFrame, + StartFrame, +) +from pipecat.serializers.base_serializer import FrameSerializer + + +class VonageFrameSerializer(FrameSerializer): + """Serializer for Vonage Video API Audio Connector WebSocket protocol. + + This serializer converts between Pipecat frames and the Vonage Audio Connector + WebSocket streaming protocol. + + Note: + Ref docs: + https://developer.vonage.com/en/video/guides/audio-connector + """ + + class InputParams(BaseModel): + """Configuration parameters for VonageFrameSerializer. + + Parameters: + vonage_sample_rate: Sample rate used by Vonage, defaults to 16000 Hz. + Common values: 8000, 16000, 24000 Hz. + sample_rate: Optional override for pipeline input sample rate. + """ + + vonage_sample_rate: int = 16000 + sample_rate: Optional[int] = None + + def __init__(self, params: Optional[InputParams] = None): + """Initialize the VonageFrameSerializer. + + Args: + params: Configuration parameters. + """ + self._params = params or VonageFrameSerializer.InputParams() + + self._vonage_sample_rate = self._params.vonage_sample_rate + self._sample_rate = 0 # Pipeline input rate + + self._input_resampler = create_stream_resampler() + self._output_resampler = create_stream_resampler() + + async def setup(self, frame: StartFrame): + """Sets up the serializer with pipeline configuration. + + Args: + frame: The StartFrame containing pipeline configuration. + """ + self._sample_rate = self._params.sample_rate or frame.audio_in_sample_rate + + async def serialize(self, frame: Frame) -> str | bytes | None: + """Serializes a Pipecat frame to Vonage WebSocket format. + + Handles conversion of various frame types to Vonage WebSocket messages. + + Args: + frame: The Pipecat frame to serialize. + + Returns: + Serialized data as string (JSON commands) or bytes (audio), or None if the frame isn't handled. + """ + if isinstance(frame, InterruptionFrame): + # Clear the audio buffer to stop playback immediately + answer = {"action": "clear"} + return json.dumps(answer) + elif isinstance(frame, AudioRawFrame): + data = frame.audio + + # Output: Convert PCM at frame's rate to Vonage's sample rate (16-bit linear PCM) + serialized_data = await self._output_resampler.resample( + data, frame.sample_rate, self._vonage_sample_rate + ) + if serialized_data is None or len(serialized_data) == 0: + # Ignoring in case we don't have audio + return None + + # Vonage expects raw binary PCM data (not base64 encoded) + return serialized_data + elif isinstance(frame, (OutputTransportMessageFrame, OutputTransportMessageUrgentFrame)): + # Allow sending custom JSON commands (e.g., notify) + return json.dumps(frame.message) + + return None + + async def deserialize(self, data: str | bytes) -> Frame | None: + """Deserializes Vonage WebSocket data to Pipecat frames. + + Handles conversion of Vonage events to appropriate Pipecat frames. + - Binary messages contain audio data (16-bit linear PCM) + - Text messages contain JSON events (websocket:connected, websocket:cleared, dtmf, etc.) + + Args: + data: The raw WebSocket data from Vonage. + + Returns: + A Pipecat frame corresponding to the Vonage event, or None if unhandled. + """ + # Check if this is binary audio data + if isinstance(data, bytes): + # Binary message = audio data (16-bit linear PCM) + payload = data + + # Input: Convert Vonage's PCM audio to pipeline sample rate + deserialized_data = await self._input_resampler.resample( + payload, + self._vonage_sample_rate, + self._sample_rate, + ) + if deserialized_data is None or len(deserialized_data) == 0: + # Ignoring in case we don't have audio + return None + + audio_frame = InputAudioRawFrame( + audio=deserialized_data, + num_channels=1, # Vonage uses mono audio + sample_rate=self._sample_rate, # Use the configured pipeline input rate + ) + return audio_frame + else: + # Text message = JSON event + try: + message = json.loads(data) + event = message.get("event") + + # Handle different event types + if event == "websocket:connected": + logger.debug( + f"Vonage WebSocket connected: content-type={message.get('content-type')}" + ) + return None + elif event == "websocket:cleared": + logger.debug("Vonage audio buffer cleared") + return None + elif event == "websocket:notify": + logger.debug(f"Vonage notify event: {message.get('payload')}") + return None + elif event == "websocket:dtmf": + # Handle DTMF input + # Vonage may send digit in different formats, try both + digit = message.get("digit") or message.get("dtmf", {}).get("digit") + if digit is None: + logger.warning(f"DTMF event received but no digit found: {message}") + return None + + digit = str(digit) + logger.debug(f"Received DTMF digit: {digit}") + try: + return InputDTMFFrame(KeypadEntry(digit)) + except ValueError: + logger.warning(f"Invalid DTMF digit received: {digit}") + return None + else: + logger.debug(f"Vonage event: {event}") + return None + + except json.JSONDecodeError: + logger.warning(f"Failed to parse JSON message from Vonage: {data}") + return None diff --git a/src/pipecat/transports/websocket/fastapi.py b/src/pipecat/transports/websocket/fastapi.py index 1bcc59e8b..cffacd932 100644 --- a/src/pipecat/transports/websocket/fastapi.py +++ b/src/pipecat/transports/websocket/fastapi.py @@ -56,11 +56,14 @@ class FastAPIWebsocketParams(TransportParams): add_wav_header: Whether to add WAV headers to audio frames. serializer: Frame serializer for encoding/decoding messages. session_timeout: Session timeout in seconds, None for no timeout. + audio_packet_bytes: Optional fixed-size packetization for raw PCM audio payloads. + Useful when the remote WebSocket media endpoint requires strict audio framing. """ add_wav_header: bool = False serializer: Optional[FrameSerializer] = None session_timeout: Optional[int] = None + audio_packet_bytes: Optional[int] = None class FastAPIWebsocketCallbacks(BaseModel): @@ -360,6 +363,14 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport): self._send_interval = 0 self._next_send_time = 0 + # Buffer for optional protocol-level audio packetization. + # Some serializers may emit arbitrarily sized raw PCM payloads, while + # certain downstream transports or media endpoints require audio to be + # sent in fixed-size frames. When `params.audio_packet_bytes` is set, + # this buffer accumulates outgoing audio until a full packet can be + # emitted, preserving any remainder for subsequent sends. + self._audio_send_buffer = bytearray() + # Whether we have seen a StartFrame already. self._initialized = False @@ -417,6 +428,10 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport): await super().process_frame(frame, direction) if isinstance(frame, InterruptionFrame): + # Drop any partially buffered audio to avoid replaying stale PCM + if self._params.audio_packet_bytes: + self._audio_send_buffer.clear() + await self._write_frame(frame) self._next_send_time = 0 @@ -480,6 +495,21 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport): try: payload = await self._params.serializer.serialize(frame) if payload: + # Optional protocol-level audio packetization: + # If a downstream WebSocket media endpoint requires fixed-size PCM frames, + # configure params.audio_packet_bytes (e.g. 640 for 20ms @ 16kHz PCM16 mono). + packet_bytes = self._params.audio_packet_bytes + + if packet_bytes and isinstance(payload, (bytes, bytearray)): + self._audio_send_buffer.extend(bytes(payload)) + + # Send only full frames; keep remainder for the next call. + while len(self._audio_send_buffer) >= packet_bytes: + chunk = bytes(self._audio_send_buffer[:packet_bytes]) + del self._audio_send_buffer[:packet_bytes] + await self._client.send(chunk) + return + await self._client.send(payload) except Exception as e: logger.error(f"{self} exception sending data: {e.__class__.__name__} ({e})") From 14a115f37203510309531def9600bd6216fc04ac Mon Sep 17 00:00:00 2001 From: Varun Pratap Singh Date: Mon, 12 Jan 2026 18:12:27 +0530 Subject: [PATCH 2/3] changelog: add fragments for PR #3410 --- changelog/3410.added.md | 1 + changelog/3410.changed.md | 1 + 2 files changed, 2 insertions(+) create mode 100644 changelog/3410.added.md create mode 100644 changelog/3410.changed.md diff --git a/changelog/3410.added.md b/changelog/3410.added.md new file mode 100644 index 000000000..094532343 --- /dev/null +++ b/changelog/3410.added.md @@ -0,0 +1 @@ +- Added `VonageFrameSerializer` for the Vonage Video API Audio Connector WebSocket protocol. diff --git a/changelog/3410.changed.md b/changelog/3410.changed.md new file mode 100644 index 000000000..0be207c65 --- /dev/null +++ b/changelog/3410.changed.md @@ -0,0 +1 @@ +- Enhanced `FastAPIWebsocketTransport` with optional protocol-level audio packetization to support media endpoints requiring strict framing and real-time pacing. From 3e982f7a4a9804774600e9f135e9a36cb2217f8b Mon Sep 17 00:00:00 2001 From: Varun Pratap Singh Date: Mon, 12 Jan 2026 22:11:39 +0530 Subject: [PATCH 3/3] refactor: rename audio_packet_bytes to fixed_audio_packet_size --- changelog/3410.changed.md | 1 + src/pipecat/transports/websocket/fastapi.py | 12 ++++++------ 2 files changed, 7 insertions(+), 6 deletions(-) diff --git a/changelog/3410.changed.md b/changelog/3410.changed.md index 0be207c65..f58ff546a 100644 --- a/changelog/3410.changed.md +++ b/changelog/3410.changed.md @@ -1 +1,2 @@ - Enhanced `FastAPIWebsocketTransport` with optional protocol-level audio packetization to support media endpoints requiring strict framing and real-time pacing. +- Renamed `audio_packet_bytes` to `fixed_audio_packet_size` for clearer audio framing semantics. diff --git a/src/pipecat/transports/websocket/fastapi.py b/src/pipecat/transports/websocket/fastapi.py index cffacd932..e1d02ac00 100644 --- a/src/pipecat/transports/websocket/fastapi.py +++ b/src/pipecat/transports/websocket/fastapi.py @@ -56,14 +56,14 @@ class FastAPIWebsocketParams(TransportParams): add_wav_header: Whether to add WAV headers to audio frames. serializer: Frame serializer for encoding/decoding messages. session_timeout: Session timeout in seconds, None for no timeout. - audio_packet_bytes: Optional fixed-size packetization for raw PCM audio payloads. + fixed_audio_packet_size: Optional fixed-size packetization for raw PCM audio payloads. Useful when the remote WebSocket media endpoint requires strict audio framing. """ add_wav_header: bool = False serializer: Optional[FrameSerializer] = None session_timeout: Optional[int] = None - audio_packet_bytes: Optional[int] = None + fixed_audio_packet_size: Optional[int] = None class FastAPIWebsocketCallbacks(BaseModel): @@ -366,7 +366,7 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport): # Buffer for optional protocol-level audio packetization. # Some serializers may emit arbitrarily sized raw PCM payloads, while # certain downstream transports or media endpoints require audio to be - # sent in fixed-size frames. When `params.audio_packet_bytes` is set, + # sent in fixed-size frames. When `params.fixed_audio_packet_size` is set, # this buffer accumulates outgoing audio until a full packet can be # emitted, preserving any remainder for subsequent sends. self._audio_send_buffer = bytearray() @@ -429,7 +429,7 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport): if isinstance(frame, InterruptionFrame): # Drop any partially buffered audio to avoid replaying stale PCM - if self._params.audio_packet_bytes: + if self._params.fixed_audio_packet_size: self._audio_send_buffer.clear() await self._write_frame(frame) @@ -497,8 +497,8 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport): if payload: # Optional protocol-level audio packetization: # If a downstream WebSocket media endpoint requires fixed-size PCM frames, - # configure params.audio_packet_bytes (e.g. 640 for 20ms @ 16kHz PCM16 mono). - packet_bytes = self._params.audio_packet_bytes + # configure params.fixed_audio_packet_size (e.g. 640 for 20ms @ 16kHz PCM16 mono). + packet_bytes = self._params.fixed_audio_packet_size if packet_bytes and isinstance(payload, (bytes, bytearray)): self._audio_send_buffer.extend(bytes(payload))