From 6c75d6f24a5b7cde4921721727fb53a98d1088bf Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 18 Dec 2025 22:32:57 -0500 Subject: [PATCH] Add VonageFrameSerializer for Vonage Voice API --- changelog/3265.added.md | 1 + src/pipecat/serializers/vonage.py | 194 ++++++++++++++++++++++++++++++ 2 files changed, 195 insertions(+) create mode 100644 changelog/3265.added.md create mode 100644 src/pipecat/serializers/vonage.py diff --git a/changelog/3265.added.md b/changelog/3265.added.md new file mode 100644 index 000000000..0c2cad463 --- /dev/null +++ b/changelog/3265.added.md @@ -0,0 +1 @@ +- Added `VonageFrameSerializer` in support of Vonage's Voice API over websockets. diff --git a/src/pipecat/serializers/vonage.py b/src/pipecat/serializers/vonage.py new file mode 100644 index 000000000..3f0cb8a17 --- /dev/null +++ b/src/pipecat/serializers/vonage.py @@ -0,0 +1,194 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Vonage Voice API 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, FrameSerializerType + + +class VonageFrameSerializer(FrameSerializer): + """Serializer for Vonage Voice API WebSocket protocol. + + This serializer handles converting between Pipecat frames and Vonage's WebSocket + voice streaming protocol. It supports audio conversion (16-bit linear PCM), + control commands like clear and notify, and DTMF input handling. + + The Vonage Voice API uses: + - Binary messages for audio data (16-bit linear PCM) + - Text messages (JSON) for control commands, events, and DTMF + + Note: Ref docs for WebSocket API: + https://developer.vonage.com/en/voice/voice-api/guides/websockets + """ + + 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() + + @property + def type(self) -> FrameSerializerType: + """Gets the serializer type. + + Returns: + The serializer type (MIXED for Vonage - supports both binary and text). + """ + return FrameSerializerType.MIXED + + 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: + 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.warning(f"DTMF event received but no digit found: {message}") + 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