From 7e82a0cf493590b76d68430ea1fc3dcc0050f63e Mon Sep 17 00:00:00 2001 From: ssillerom Date: Mon, 19 Jan 2026 20:45:22 +0100 Subject: [PATCH 01/11] feature: Genesys AudioHook WebSocket protocol serializer for Pipecat --- src/pipecat/serializers/genesys.py | 945 +++++++++++++++++++++++++++++ 1 file changed, 945 insertions(+) create mode 100644 src/pipecat/serializers/genesys.py diff --git a/src/pipecat/serializers/genesys.py b/src/pipecat/serializers/genesys.py new file mode 100644 index 000000000..f768b8431 --- /dev/null +++ b/src/pipecat/serializers/genesys.py @@ -0,0 +1,945 @@ +""" +Genesys AudioHook WebSocket protocol serializer for Pipecat. + +This serializer implements the Genesys AudioHook protocol for bidirectional +audio streaming between Pipecat pipelines and Genesys Cloud contact center. + +Protocol Reference: +- https://developer.genesys.cloud/devapps/audiohook + +Audio Format: +- PCMU (μ-law) at 8kHz sample rate (preferred) +- L16 (16-bit linear PCM) at 8kHz also supported +- Mono or Stereo (external on left, internal on right) +""" + +import json +import uuid +from datetime import timedelta +from enum import Enum +from typing import Any, Awaitable, Callable, Dict, List, Optional + +from loguru import logger +from pydantic import BaseModel + +from pipecat.audio.dtmf.types import KeypadEntry +from pipecat.frames.frames import ( + AudioRawFrame, + CancelFrame, + EndFrame, + Frame, + InputAudioRawFrame, + InputDTMFFrame, + OutputTransportMessageFrame, + OutputTransportMessageUrgentFrame, + StartFrame +) +from pipecat.serializers.base_serializer import FrameSerializer +from pipecat.audio.resamplers.soxr_stream_resampler import SOXRStreamAudioResampler +from pipecat.audio.utils import ulaw_to_pcm, pcm_to_ulaw + + +class AudioHookMessageType(str, Enum): + """AudioHook protocol message types.""" + OPEN = "open" + OPENED = "opened" + CLOSE = "close" + CLOSED = "closed" + PAUSE = "pause" + RESUMED = "resumed" + PING = "ping" + PONG = "pong" + UPDATE = "update" + EVENT = "event" + ERROR = "error" + DISCONNECT = "disconnect" + + +class AudioHookChannel(str, Enum): + """AudioHook audio channel configuration.""" + EXTERNAL = "external" # Customer audio only (mono) + INTERNAL = "internal" # Agent audio only (mono) + BOTH = "both" # Stereo: external=left, internal=right + + +class AudioHookMediaFormat(str, Enum): + """Supported audio formats.""" + PCMU = "PCMU" # μ-law, 8kHz + L16 = "L16" # 16-bit linear PCM, 8kHz + + +class GenesysAudioHookSerializer(FrameSerializer): + """Serializer for Genesys AudioHook WebSocket protocol. + + This serializer handles converting between Pipecat frames and Genesys + AudioHook protocol messages. It supports: + + - Bidirectional audio streaming (PCMU at 8kHz) + - Session lifecycle management (open, close, pause) + - Probe mode for health checks + - Custom configuration passthrough + - Event messaging back to Genesys Cloud + + The AudioHook protocol uses: + - Text WebSocket frames for JSON control messages + - Binary WebSocket frames for audio data + + Example usage: + ```python + serializer = GenesysAudioHookSerializer( + session_id="abc-123", + params=GenesysAudioHookSerializer.InputParams( + channel=AudioHookChannel.BOTH, + ) + ) + + # Use with WebSocket transport + transport = WebsocketServerTransport( + serializer=serializer, + ... + ) + ``` + + Attributes: + PROTOCOL_VERSION: The AudioHook protocol version (currently "2"). + """ + + PROTOCOL_VERSION = "2" + + class InputParams(BaseModel): + """Configuration parameters for GenesysAudioHookSerializer. + + Parameters: + genesys_sample_rate: Sample rate used by Genesys (8000 Hz). + sample_rate: Optional override for pipeline input sample rate. + channel: Which audio channels to process (external, internal, both). + media_format: Audio format (PCMU or L16). + process_external: Whether to process external (customer) audio. + process_internal: Whether to process internal (agent) audio. + enable_interruption_events: Send interruption events to Genesys. + """ + + genesys_sample_rate: int = 8000 + sample_rate: Optional[int] = None + channel: AudioHookChannel = AudioHookChannel.EXTERNAL + media_format: AudioHookMediaFormat = AudioHookMediaFormat.PCMU + process_external: bool = True + process_internal: bool = False + enable_interruption_events: bool = True + + def __init__( + self, + session_id: Optional[str] = None, + params: Optional[InputParams] = None, + send_message_callback: Optional[Callable[[str], Awaitable[None]]] = None, + ): + """Initialize the GenesysAudioHookSerializer. + + Args: + session_id: The AudioHook session ID (received in open message). + params: Configuration parameters. + send_message_callback: Optional async callback to send messages directly + (bypassing the pipeline). Used for urgent messages like pong. + """ + self._params = params or GenesysAudioHookSerializer.InputParams() + self._session_id = session_id or "" + self._send_message_callback = send_message_callback + + self._genesys_sample_rate = self._params.genesys_sample_rate + self._sample_rate = 0 # Pipeline input rate, set in setup() + + # Use Pipecat's official SOXR resampler + # Only used for TTS output (16kHz → 8kHz), input goes without resampling + self._input_resampler = SOXRStreamAudioResampler() + self._output_resampler = SOXRStreamAudioResampler() + + # Protocol state + self._client_seq = 0 + self._server_seq = 0 + self._is_open = False + self._is_paused = False + self._position = timedelta(0) + + # TTS output state + self._tts_chunk_count = 0 + + # Session metadata + self._conversation_id: Optional[str] = None + self._participant: Optional[Dict[str, Any]] = None + self._custom_config: Optional[Dict[str, Any]] = None + self._media_info: Optional[List[Dict[str, Any]]] = None + self._input_variables: Optional[Dict[str, Any]] = None # Custom input from Genesys + + def set_send_message_callback(self, callback: Callable[[str], Awaitable[None]]): + """Set the callback for sending urgent messages directly. + + Args: + callback: An async function that takes a string message and sends it. + """ + self._send_message_callback = callback + + @property + def session_id(self) -> str: + """Get the current session ID.""" + return self._session_id + + @property + def conversation_id(self) -> Optional[str]: + """Get the Genesys conversation ID.""" + return self._conversation_id + + @property + def is_open(self) -> bool: + """Check if the AudioHook session is open.""" + return self._is_open + + @property + def is_paused(self) -> bool: + """Check if audio streaming is paused.""" + return self._is_paused + + @property + def participant(self) -> Optional[Dict[str, Any]]: + """Get participant info (ani, dnis, etc.) from the open message.""" + return self._participant + + @property + def input_variables(self) -> Optional[Dict[str, Any]]: + """Get custom input variables from the open message.""" + return self._input_variables + + 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 + logger.debug(f"GenesysAudioHookSerializer setup with sample_rate={self._sample_rate}") + + def reset_tts_state(self): + """Reset TTS state for a new utterance. + + NOTE: We don't reset the resampler because that causes artifacts. + The resampler maintains its state between utterances for cleaner audio. + """ + self._tts_chunk_count = 0 + + def _format_position(self, position: timedelta) -> str: + """Format a timedelta as ISO 8601 duration string. + + Args: + position: The timedelta to format. + + Returns: + ISO 8601 duration string (e.g., "PT1.5S"). + """ + total_seconds = position.total_seconds() + return f"PT{total_seconds:.3f}S" + + def _parse_position(self, position_str: str) -> timedelta: + """Parse an ISO 8601 duration string to timedelta. + + Args: + position_str: ISO 8601 duration string (e.g., "PT1.5S"). + + Returns: + Corresponding timedelta. + """ + # Simple parser for PT#S or PT#.#S format + if position_str.startswith("PT") and position_str.endswith("S"): + try: + seconds = float(position_str[2:-1]) + return timedelta(seconds=seconds) + except ValueError: + pass + return timedelta(0) + + def _next_server_seq(self) -> int: + """Get the next server sequence number.""" + self._server_seq += 1 + return self._server_seq + + def _create_message( + self, + msg_type: AudioHookMessageType, + parameters: Optional[Dict[str, Any]] = None, + include_position: bool = True, + ) -> Dict[str, Any]: + """Create a protocol message with common fields. + + Based on the Genesys AudioHook protocol, responses include: + - seq: Server's sequence number (incremented per message) + - clientseq: Echo of the client's last sequence number + + Args: + msg_type: The message type. + parameters: Optional parameters object. + include_position: Whether to include position field. + + Returns: + The message dictionary. + """ + seq = self._next_server_seq() + msg = { + "version": self.PROTOCOL_VERSION, + "type": msg_type.value, + "seq": seq, + "clientseq": self._client_seq, + "id": self._session_id, + } + + if include_position: + msg["position"] = self._format_position(self._position) + + if parameters: + msg["parameters"] = parameters + + return msg + + def create_opened_response( + self, + start_paused: bool = False, + supported_languages: Optional[List[str]] = None, + selected_language: Optional[str] = None, + ) -> str: + """Create an 'opened' response message for the client. + + This should be sent in response to an 'open' message from Genesys. + + Args: + start_paused: Whether to start the session paused. + supported_languages: List of supported language codes. + selected_language: The selected language code. + + Returns: + JSON string of the opened response message. + """ + # Build channels list based on configuration + channels = [] + if self._params.channel == AudioHookChannel.EXTERNAL: + channels = ["external"] + elif self._params.channel == AudioHookChannel.INTERNAL: + channels = ["internal"] + elif self._params.channel == AudioHookChannel.BOTH: + channels = ["external", "internal"] + + parameters = { + "startPaused": start_paused, + "media": [ + { + "type": "audio", + "format": self._params.media_format.value, + "channels": channels, + "rate": self._genesys_sample_rate, + } + ], + } + + if supported_languages: + parameters["supportedLanguages"] = supported_languages + if selected_language: + parameters["selectedLanguage"] = selected_language + + msg = self._create_message( + AudioHookMessageType.OPENED, + parameters=parameters, + include_position=False, # opened doesn't need position + ) + + self._is_open = True + logger.debug(f"AudioHook session opened: {self._session_id}") + + return json.dumps(msg) + + def create_closed_response(self) -> str: + """Create a 'closed' response message. + + This should be sent in response to a 'close' message from Genesys. + + Returns: + JSON string of the closed response message. + """ + msg = self._create_message(AudioHookMessageType.CLOSED) + + self._is_open = False + logger.debug(f"AudioHook session closed: {self._session_id}") + + return json.dumps(msg) + + def create_pong_response(self) -> str: + """Create a 'pong' response message. + + This should be sent in response to a 'ping' message from Genesys. + + Returns: + JSON string of the pong response message. + """ + msg = self._create_message(AudioHookMessageType.PONG) + return json.dumps(msg) + + def create_resumed_response(self) -> str: + """Create a 'resumed' response message. + + This should be sent in response to a 'pause' message when ready to resume. + + Returns: + JSON string of the resumed response message. + """ + msg = self._create_message(AudioHookMessageType.RESUMED) + + self._is_paused = False + logger.debug(f"AudioHook session resumed: {self._session_id}") + + return json.dumps(msg) + + def create_event_message( + self, + entity_type: str, + entity_data: Dict[str, Any], + ) -> str: + """Create an 'event' message to send data back to Genesys. + + This can be used for transcriptions, agent assist, or other events. + + Args: + entity_type: The type of entity (e.g., "transcript"). + entity_data: The entity data. + + Returns: + JSON string of the event message. + """ + parameters = { + "entities": [ + { + "type": entity_type, + **entity_data, + } + ] + } + + msg = self._create_message( + AudioHookMessageType.EVENT, + parameters=parameters, + ) + + return json.dumps(msg) + + def create_disconnect_message( + self, + reason: str = "completed", + action: str = "transfer", + output_variables: Optional[Dict[str, Any]] = None, + info: Optional[str] = None, + ) -> str: + """Create a 'disconnect' message to initiate session termination. + + Args: + reason: Disconnect reason (e.g., "completed", "error"). + action: Action to take ("transfer" to agent, "finished" if completed). + output_variables: Custom output variables to pass back to Genesys. + info: Optional additional information. + + Returns: + JSON string of the disconnect message. + """ + parameters: Dict[str, Any] = {"reason": reason} + + # Build outputVariables + out_vars = {"action": action} + if output_variables: + out_vars.update(output_variables) + parameters["outputVariables"] = out_vars + + if info: + parameters["info"] = info + + msg = self._create_message( + AudioHookMessageType.DISCONNECT, + parameters=parameters, + ) + + logger.debug(f"AudioHook disconnect: reason={reason}, action={action}") + return json.dumps(msg) + + def create_error_message( + self, + code: int, + message: str, + retryable: bool = False, + ) -> str: + """Create an 'error' message. + + Args: + code: Error code. + message: Error message. + retryable: Whether the operation can be retried. + + Returns: + JSON string of the error message. + """ + parameters = { + "code": code, + "message": message, + "retryable": retryable, + } + + msg = self._create_message( + AudioHookMessageType.ERROR, + parameters=parameters, + ) + + logger.error(f"AudioHook error: {code} - {message}") + return json.dumps(msg) + + def create_barge_in_event(self) -> str: + """Create a barge-in event message. + + This notifies Genesys Cloud that the user has interrupted the bot's + audio output. Genesys will stop any queued audio playback. + + Returns: + JSON string of the barge-in event message. + """ + msg = self._create_message( + AudioHookMessageType.EVENT, + parameters={ + "entities": [ + {"type": "barge_in", "data": {}} + ] + }, + ) + + logger.debug("🔇 Barge-in event sent to Genesys") + + return json.dumps(msg) + + def create_resume_event(self) -> str: + """Create a resume event message. + + This notifies Genesys that the bot is ready to resume after a barge-in. + Should be called after the user stops speaking following a barge-in. + + Returns: + JSON string of the resume event message. + """ + self._barge_in = False + + # Note: 'resume' might not be a standard AudioHook message type, + # but it's used in some implementations + msg = { + "version": self.PROTOCOL_VERSION, + "type": "resume", + "seq": self._next_server_seq(), + "clientseq": self._client_seq, + "id": self._session_id, + "parameters": {}, + } + + logger.debug("Resume event sent") + return json.dumps(msg) + + def create_interruption_event( + self, + reason: str = "user_speaking", + discarded_bytes: Optional[int] = None, + ) -> str: + """Create a generic interruption event message. + + This is an alternative to create_barge_in_event() that includes + more detailed information about the interruption. + + Args: + reason: Reason for interruption (e.g., "user_speaking", "dtmf"). + discarded_bytes: Number of audio bytes discarded due to interruption. + + Returns: + JSON string of the interruption event message. + """ + entity_data: Dict[str, Any] = { + "reason": reason, + "timestamp": self._format_position(self._position), + } + + if discarded_bytes is not None: + entity_data["discardedAudioBytes"] = discarded_bytes + + logger.info( + f"AudioHook interruption: reason={reason}, " + f"discarded={discarded_bytes or 0} bytes" + ) + + return self.create_event_message( + entity_type="interruption", + entity_data=entity_data, + ) + + async def serialize(self, frame: Frame) -> str | bytes | None: + """Serializes a Pipecat frame to Genesys AudioHook format. + + Handles conversion of various frame types to AudioHook messages: + - AudioRawFrame -> Binary audio data + - EndFrame/CancelFrame -> Disconnect message + - OutputTransportMessageFrame -> Pass-through JSON + + Args: + frame: The Pipecat frame to serialize. + + Returns: + Serialized data as string (JSON) or bytes (audio), or None. + """ + if isinstance(frame, (EndFrame, CancelFrame)): + return self.create_disconnect_message(reason="completed") + + elif isinstance(frame, AudioRawFrame): + if not self._is_open or self._is_paused: + return None + + data = frame.audio + + self._tts_chunk_count += 1 + + # Convert PCM to μ-law at 8kHz for Genesys + if self._params.media_format == AudioHookMediaFormat.PCMU: + serialized_data = await pcm_to_ulaw( + data, + frame.sample_rate, + self._genesys_sample_rate, + self._output_resampler, + ) + else: + # L16 format - just resample if needed + logger.warning("L16 format not yet fully implemented") + return None + + if serialized_data is None or len(serialized_data) == 0: + return None + + return bytes(serialized_data) + + elif isinstance(frame, (OutputTransportMessageFrame, OutputTransportMessageUrgentFrame)): + # Pass through custom JSON messages + return json.dumps(frame.message) + + # Ignore other frames - we don't need to process them here + return None + + async def deserialize(self, data: str | bytes) -> Frame | None: + """Deserializes Genesys AudioHook data to Pipecat frames. + + Handles: + - Binary data -> InputAudioRawFrame + - JSON text -> Protocol messages (open, close, ping, pause, etc.) + + For control messages (open, close, ping, pause), this method handles + the protocol response internally and logs the events. The application + should monitor session state via the is_open and is_paused properties. + + Args: + data: The raw WebSocket data from Genesys. + + Returns: + A Pipecat frame corresponding to the data, or None if handled internally. + """ + # Binary data = audio + if isinstance(data, bytes): + logger.debug(f"[AUDIO IN] Received {len(data)} bytes from Genesys") + return await self._deserialize_audio(data) + + # Text data = JSON control message + try: + message = json.loads(data) + except json.JSONDecodeError as e: + logger.error(f"Failed to parse AudioHook message: {e}") + return None + + return await self._handle_control_message(message) + + async def _deserialize_audio(self, data: bytes) -> Frame | None: + """Deserialize binary audio data to an InputAudioRawFrame. + + Args: + data: Raw audio bytes (PCMU or L16). + + Returns: + InputAudioRawFrame with PCM audio at pipeline sample rate. + """ + if not self._is_open or self._is_paused: + return None + + audio_data = data + original_len = len(data) + + # If Genesys sends stereo audio (BOTH channels), extract only the external channel (left) + # Stereo audio is interleaved: [L0, R0, L1, R1, ...] + if self._params.channel == AudioHookChannel.BOTH and len(data) > 0: + # For PCMU, each sample is 1 byte + # Extract only bytes at even positions (left channel = external) + audio_data = bytes(data[i] for i in range(0, len(data), 2)) + logger.debug(f"🔊 Stereo audio: {original_len} bytes → {len(audio_data)} bytes (external channel)") + + if self._params.media_format == AudioHookMediaFormat.PCMU: + # Convert μ-law at 8kHz to PCM at pipeline rate + deserialized_data = await ulaw_to_pcm( + audio_data, + self._genesys_sample_rate, + self._sample_rate, + self._input_resampler, + ) + else: + # L16 format + logger.warning("L16 format not yet fully implemented") + return None + + if deserialized_data is None or len(deserialized_data) == 0: + return None + + # Always use mono for STT - ElevenLabs expects single channel + num_channels = 1 + + audio_frame = InputAudioRawFrame( + audio=deserialized_data, + num_channels=num_channels, + sample_rate=self._sample_rate, + ) + + return audio_frame + + async def _handle_control_message(self, message: Dict[str, Any]) -> Frame | None: + """Handle a JSON control message from Genesys. + + Args: + message: Parsed JSON message. + + Returns: + Frame if the message should be passed to the pipeline, None otherwise. + """ + msg_type = message.get("type", "") + self._client_seq = message.get("seq", 0) + + # Update position if provided + if "position" in message: + self._position = self._parse_position(message["position"]) + + if msg_type == AudioHookMessageType.OPEN.value: + return await self._handle_open(message) + + elif msg_type == AudioHookMessageType.CLOSE.value: + return await self._handle_close(message) + + elif msg_type == AudioHookMessageType.PING.value: + return await self._handle_ping(message) + + elif msg_type == AudioHookMessageType.PAUSE.value: + return await self._handle_pause(message) + + elif msg_type == AudioHookMessageType.UPDATE.value: + return await self._handle_update(message) + + elif msg_type == AudioHookMessageType.ERROR.value: + return await self._handle_error(message) + + elif msg_type == "dtmf": + return await self._handle_dtmf(message) + + elif msg_type == "playback_started": + self._is_playing = True + logger.debug("Playback started (from Genesys)") + return None + + elif msg_type == "playback_completed": + self._is_playing = False + logger.debug("Playback completed (from Genesys)") + return None + + else: + logger.warning(f"Unknown AudioHook message type: {msg_type}") + return None + + async def _handle_open(self, message: Dict[str, Any]) -> Frame | None: + """Handle an 'open' message from Genesys. + + This initializes the session with metadata from Genesys Cloud. + + Args: + message: The open message. + + Returns: + None (response should be sent via create_opened_response()). + """ + self._session_id = message.get("id", str(uuid.uuid4())) + + params = message.get("parameters", {}) + self._conversation_id = params.get("conversationId") + self._participant = params.get("participant") + self._custom_config = params.get("customConfig") + self._media_info = params.get("media") # This is a list of media objects + self._input_variables = params.get("inputVariables") # Custom vars from Genesys + + # Extract media configuration if present + # media is a list like: [{"type": "audio", "format": "PCMU", "channels": ["external"], "rate": 8000}] + media_list = self._media_info + if media_list and isinstance(media_list, list) and len(media_list) > 0: + audio_media: Dict[str, Any] = media_list[0] # Get first media entry + channels = audio_media.get("channels", []) + logger.debug(f"📡 Genesys audio config: format={audio_media.get('format')}, channels={channels}, rate={audio_media.get('rate')}") + # channels is a list like ["external"] or ["external", "internal"] + if isinstance(channels, list): + if "external" in channels and "internal" in channels: + self._params.channel = AudioHookChannel.BOTH + logger.debug("📡 Stereo mode: extracting external channel") + elif "external" in channels: + self._params.channel = AudioHookChannel.EXTERNAL + logger.debug("📡 Mono mode: external channel") + elif "internal" in channels: + self._params.channel = AudioHookChannel.INTERNAL + logger.debug("📡 Mono mode: internal channel") + + # Log participant info for debugging + ani = self._participant.get("ani", "unknown") if self._participant else "unknown" + logger.info( + f"AudioHook open request: session={self._session_id}, " + f"conversation={self._conversation_id}, ani={ani}" + ) + + # Note: Application should call create_opened_response() to respond + return None + + async def _handle_close(self, message: Dict[str, Any]) -> Frame | None: + """Handle a 'close' message from Genesys. + + Args: + message: The close message. + + Returns: + EndFrame to signal the pipeline to close. + """ + params = message.get("parameters", {}) + reason = params.get("reason", "unknown") + + logger.info(f"🔴 Genesys closed the connection: {reason}") + + self._is_open = False + + # Send closed response via callback if available + if self._send_message_callback: + try: + closed_response = self.create_closed_response() + await self._send_message_callback(closed_response) + except Exception as e: + logger.error(f"Failed to send closed response: {e}") + + # Return EndFrame to close the pipeline and WebSocket + return EndFrame() + + async def _handle_ping(self, message: Dict[str, Any]) -> Frame | None: + """Handle a 'ping' message from Genesys. + + Args: + message: The ping message. + + Returns: + None if pong was sent directly via callback, otherwise + OutputTransportMessageUrgentFrame with pong response. + """ + # Create pong response + pong_response = self.create_pong_response() + + # If we have a direct callback, use it for immediate response + if self._send_message_callback: + try: + await self._send_message_callback(pong_response) + logger.debug("Pong sent directly via callback") + return None + except Exception as e: + logger.error(f"Failed to send pong via callback: {e}") + + # Fallback: return as urgent frame to be sent through pipeline + return OutputTransportMessageUrgentFrame(message=json.loads(pong_response)) + + async def _handle_pause(self, message: Dict[str, Any]) -> Frame | None: + """Handle a 'pause' message from Genesys. + + This is used when audio streaming is temporarily suspended + (e.g., during hold). + + Args: + message: The pause message. + + Returns: + None (response should be sent via create_resumed_response()). + """ + params = message.get("parameters", {}) + reason = params.get("reason", "unknown") + + logger.info(f"AudioHook pause request: reason={reason}") + + self._is_paused = True + + # Note: Application should call create_resumed_response() when ready + return None + + async def _handle_update(self, message: Dict[str, Any]) -> Frame | None: + """Handle an 'update' message from Genesys. + + Updates may include changes to participants or configuration. + + Args: + message: The update message. + + Returns: + None. + """ + params = message.get("parameters", {}) + + if "participant" in params: + self._participant = params["participant"] + + logger.debug(f"AudioHook update received: {params}") + + return None + + async def _handle_error(self, message: Dict[str, Any]) -> Frame | None: + """Handle an 'error' message from Genesys. + + Args: + message: The error message. + + Returns: + None. + """ + params = message.get("parameters", {}) + code = params.get("code", 0) + error_msg = params.get("message", "Unknown error") + + logger.error(f"AudioHook error from Genesys: {code} - {error_msg}") + + return None + + async def _handle_dtmf(self, message: Dict[str, Any]) -> Frame | None: + """Handle a 'dtmf' message from Genesys. + + DTMF (Dual-Tone Multi-Frequency) events are sent when the user + presses keys on their phone keypad. + + Args: + message: The DTMF message. + + Returns: + InputDTMFFrame with the pressed digit. + """ + params = message.get("parameters", {}) + digit = params.get("digit", "") + + if not digit: + logger.warning("DTMF message received without digit") + return None + + logger.info(f"DTMF received: {digit}") + + try: + return InputDTMFFrame(KeypadEntry(digit)) + except ValueError: + # Invalid digit + logger.warning(f"Invalid DTMF digit: {digit}") + return None From fa5da3b0becae911fcbf3a3fcf0ef22ce83a52f6 Mon Sep 17 00:00:00 2001 From: ssillerom Date: Mon, 19 Jan 2026 20:49:23 +0100 Subject: [PATCH 02/11] change comments --- src/pipecat/serializers/genesys.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/pipecat/serializers/genesys.py b/src/pipecat/serializers/genesys.py index f768b8431..11a842772 100644 --- a/src/pipecat/serializers/genesys.py +++ b/src/pipecat/serializers/genesys.py @@ -694,7 +694,7 @@ class GenesysAudioHookSerializer(FrameSerializer): if deserialized_data is None or len(deserialized_data) == 0: return None - # Always use mono for STT - ElevenLabs expects single channel + # Always use mono for STT num_channels = 1 audio_frame = InputAudioRawFrame( From a446bca72dc3af999a71856b217c8c973094e4db Mon Sep 17 00:00:00 2001 From: ssillerom Date: Sun, 25 Jan 2026 21:10:22 +0100 Subject: [PATCH 03/11] changes: added OutputTransportUrgentFrame to on closed, removed callback --- src/pipecat/serializers/genesys.py | 205 ++++++----------------------- 1 file changed, 37 insertions(+), 168 deletions(-) diff --git a/src/pipecat/serializers/genesys.py b/src/pipecat/serializers/genesys.py index 11a842772..48637653b 100644 --- a/src/pipecat/serializers/genesys.py +++ b/src/pipecat/serializers/genesys.py @@ -1,15 +1,15 @@ """ -Genesys AudioHook WebSocket protocol serializer for Pipecat. +Use with Genesys Audio Connector to connect Genesys Cloud Contact Center with Pipecat pipelines. -This serializer implements the Genesys AudioHook protocol for bidirectional -audio streaming between Pipecat pipelines and Genesys Cloud contact center. +This connector implements the Genesys AudioHook protocol for bidirectional +audio streaming between Genesys Cloud contact center and Pipecat pipelines. Protocol Reference: - https://developer.genesys.cloud/devapps/audiohook Audio Format: - PCMU (μ-law) at 8kHz sample rate (preferred) -- L16 (16-bit linear PCM) at 8kHz also supported +- L16 (16-bit linear PCM) at 8kHz also supported - Mono or Stereo (external on left, internal on right) """ @@ -17,7 +17,7 @@ import json import uuid from datetime import timedelta from enum import Enum -from typing import Any, Awaitable, Callable, Dict, List, Optional +from typing import Any, Dict, List, Optional from loguru import logger from pydantic import BaseModel @@ -32,7 +32,8 @@ from pipecat.frames.frames import ( InputDTMFFrame, OutputTransportMessageFrame, OutputTransportMessageUrgentFrame, - StartFrame + StartFrame, + InterruptionFrame ) from pipecat.serializers.base_serializer import FrameSerializer from pipecat.audio.resamplers.soxr_stream_resampler import SOXRStreamAudioResampler @@ -116,7 +117,6 @@ class GenesysAudioHookSerializer(FrameSerializer): media_format: Audio format (PCMU or L16). process_external: Whether to process external (customer) audio. process_internal: Whether to process internal (agent) audio. - enable_interruption_events: Send interruption events to Genesys. """ genesys_sample_rate: int = 8000 @@ -125,30 +125,25 @@ class GenesysAudioHookSerializer(FrameSerializer): media_format: AudioHookMediaFormat = AudioHookMediaFormat.PCMU process_external: bool = True process_internal: bool = False - enable_interruption_events: bool = True def __init__( self, session_id: Optional[str] = None, params: Optional[InputParams] = None, - send_message_callback: Optional[Callable[[str], Awaitable[None]]] = None, ): """Initialize the GenesysAudioHookSerializer. Args: session_id: The AudioHook session ID (received in open message). params: Configuration parameters. - send_message_callback: Optional async callback to send messages directly - (bypassing the pipeline). Used for urgent messages like pong. """ self._params = params or GenesysAudioHookSerializer.InputParams() self._session_id = session_id or "" - self._send_message_callback = send_message_callback self._genesys_sample_rate = self._params.genesys_sample_rate self._sample_rate = 0 # Pipeline input rate, set in setup() - # Use Pipecat's official SOXR resampler + # Use Pipecat's official resampler if needed (SOXR) # Only used for TTS output (16kHz → 8kHz), input goes without resampling self._input_resampler = SOXRStreamAudioResampler() self._output_resampler = SOXRStreamAudioResampler() @@ -160,23 +155,13 @@ class GenesysAudioHookSerializer(FrameSerializer): self._is_paused = False self._position = timedelta(0) - # TTS output state - self._tts_chunk_count = 0 - # Session metadata self._conversation_id: Optional[str] = None self._participant: Optional[Dict[str, Any]] = None self._custom_config: Optional[Dict[str, Any]] = None self._media_info: Optional[List[Dict[str, Any]]] = None self._input_variables: Optional[Dict[str, Any]] = None # Custom input from Genesys - - def set_send_message_callback(self, callback: Callable[[str], Awaitable[None]]): - """Set the callback for sending urgent messages directly. - Args: - callback: An async function that takes a string message and sends it. - """ - self._send_message_callback = callback @property def session_id(self) -> str: @@ -217,14 +202,6 @@ class GenesysAudioHookSerializer(FrameSerializer): self._sample_rate = self._params.sample_rate or frame.audio_in_sample_rate logger.debug(f"GenesysAudioHookSerializer setup with sample_rate={self._sample_rate}") - def reset_tts_state(self): - """Reset TTS state for a new utterance. - - NOTE: We don't reset the resampler because that causes artifacts. - The resampler maintains its state between utterances for cleaner audio. - """ - self._tts_chunk_count = 0 - def _format_position(self, position: timedelta) -> str: """Format a timedelta as ISO 8601 duration string. @@ -316,7 +293,7 @@ class GenesysAudioHookSerializer(FrameSerializer): JSON string of the opened response message. """ # Build channels list based on configuration - channels = [] + channels: list[str] = [] if self._params.channel == AudioHookChannel.EXTERNAL: channels = ["external"] elif self._params.channel == AudioHookChannel.INTERNAL: @@ -393,36 +370,26 @@ class GenesysAudioHookSerializer(FrameSerializer): return json.dumps(msg) - def create_event_message( - self, - entity_type: str, - entity_data: Dict[str, Any], - ) -> str: - """Create an 'event' message to send data back to Genesys. + def create_barge_in_event(self) -> str: + """Create a barge-in event message. - This can be used for transcriptions, agent assist, or other events. + This notifies Genesys Cloud that the user has interrupted the bot's + audio output. Genesys will stop any queued audio playback. - Args: - entity_type: The type of entity (e.g., "transcript"). - entity_data: The entity data. - Returns: - JSON string of the event message. + JSON string of the barge-in event message. """ - parameters = { - "entities": [ - { - "type": entity_type, - **entity_data, - } - ] - } - msg = self._create_message( AudioHookMessageType.EVENT, - parameters=parameters, + parameters={ + "entities": [ + {"type": "barge_in", "data": {}} + ] + }, ) + logger.debug("🔇 Barge-in event sent to Genesys") + return json.dumps(msg) def create_disconnect_message( @@ -492,88 +459,6 @@ class GenesysAudioHookSerializer(FrameSerializer): logger.error(f"AudioHook error: {code} - {message}") return json.dumps(msg) - def create_barge_in_event(self) -> str: - """Create a barge-in event message. - - This notifies Genesys Cloud that the user has interrupted the bot's - audio output. Genesys will stop any queued audio playback. - - Returns: - JSON string of the barge-in event message. - """ - msg = self._create_message( - AudioHookMessageType.EVENT, - parameters={ - "entities": [ - {"type": "barge_in", "data": {}} - ] - }, - ) - - logger.debug("🔇 Barge-in event sent to Genesys") - - return json.dumps(msg) - - def create_resume_event(self) -> str: - """Create a resume event message. - - This notifies Genesys that the bot is ready to resume after a barge-in. - Should be called after the user stops speaking following a barge-in. - - Returns: - JSON string of the resume event message. - """ - self._barge_in = False - - # Note: 'resume' might not be a standard AudioHook message type, - # but it's used in some implementations - msg = { - "version": self.PROTOCOL_VERSION, - "type": "resume", - "seq": self._next_server_seq(), - "clientseq": self._client_seq, - "id": self._session_id, - "parameters": {}, - } - - logger.debug("Resume event sent") - return json.dumps(msg) - - def create_interruption_event( - self, - reason: str = "user_speaking", - discarded_bytes: Optional[int] = None, - ) -> str: - """Create a generic interruption event message. - - This is an alternative to create_barge_in_event() that includes - more detailed information about the interruption. - - Args: - reason: Reason for interruption (e.g., "user_speaking", "dtmf"). - discarded_bytes: Number of audio bytes discarded due to interruption. - - Returns: - JSON string of the interruption event message. - """ - entity_data: Dict[str, Any] = { - "reason": reason, - "timestamp": self._format_position(self._position), - } - - if discarded_bytes is not None: - entity_data["discardedAudioBytes"] = discarded_bytes - - logger.info( - f"AudioHook interruption: reason={reason}, " - f"discarded={discarded_bytes or 0} bytes" - ) - - return self.create_event_message( - entity_type="interruption", - entity_data=entity_data, - ) - async def serialize(self, frame: Frame) -> str | bytes | None: """Serializes a Pipecat frame to Genesys AudioHook format. @@ -597,8 +482,6 @@ class GenesysAudioHookSerializer(FrameSerializer): data = frame.audio - self._tts_chunk_count += 1 - # Convert PCM to μ-law at 8kHz for Genesys if self._params.media_format == AudioHookMediaFormat.PCMU: serialized_data = await pcm_to_ulaw( @@ -616,6 +499,9 @@ class GenesysAudioHookSerializer(FrameSerializer): return None return bytes(serialized_data) + + elif isinstance(frame, InterruptionFrame): + return self.create_barge_in_event() elif isinstance(frame, (OutputTransportMessageFrame, OutputTransportMessageUrgentFrame)): # Pass through custom JSON messages @@ -671,7 +557,7 @@ class GenesysAudioHookSerializer(FrameSerializer): original_len = len(data) # If Genesys sends stereo audio (BOTH channels), extract only the external channel (left) - # Stereo audio is interleaved: [L0, R0, L1, R1, ...] + # Stereo audio comes interleaved: [L0, R0, L1, R1, ...] if self._params.channel == AudioHookChannel.BOTH and len(data) > 0: # For PCMU, each sample is 1 byte # Extract only bytes at even positions (left channel = external) @@ -694,7 +580,7 @@ class GenesysAudioHookSerializer(FrameSerializer): if deserialized_data is None or len(deserialized_data) == 0: return None - # Always use mono for STT + # Always use mono for STT - ElevenLabs expects single channel num_channels = 1 audio_frame = InputAudioRawFrame( @@ -704,6 +590,7 @@ class GenesysAudioHookSerializer(FrameSerializer): ) return audio_frame + async def _handle_control_message(self, message: Dict[str, Any]) -> Frame | None: """Handle a JSON control message from Genesys. @@ -735,7 +622,7 @@ class GenesysAudioHookSerializer(FrameSerializer): elif msg_type == AudioHookMessageType.UPDATE.value: return await self._handle_update(message) - + elif msg_type == AudioHookMessageType.ERROR.value: return await self._handle_error(message) @@ -743,15 +630,12 @@ class GenesysAudioHookSerializer(FrameSerializer): return await self._handle_dtmf(message) elif msg_type == "playback_started": - self._is_playing = True logger.debug("Playback started (from Genesys)") return None elif msg_type == "playback_completed": - self._is_playing = False logger.debug("Playback completed (from Genesys)") return None - else: logger.warning(f"Unknown AudioHook message type: {msg_type}") return None @@ -812,7 +696,8 @@ class GenesysAudioHookSerializer(FrameSerializer): message: The close message. Returns: - EndFrame to signal the pipeline to close. + OutputTransportMessageUrgentFrame with the closed response, + or None if no response should be sent. """ params = message.get("parameters", {}) reason = params.get("reason", "unknown") @@ -820,17 +705,11 @@ class GenesysAudioHookSerializer(FrameSerializer): logger.info(f"🔴 Genesys closed the connection: {reason}") self._is_open = False + + logger.info(f"Sending closed response to Genesys...") - # Send closed response via callback if available - if self._send_message_callback: - try: - closed_response = self.create_closed_response() - await self._send_message_callback(closed_response) - except Exception as e: - logger.error(f"Failed to send closed response: {e}") - - # Return EndFrame to close the pipeline and WebSocket - return EndFrame() + # Return as urgent frame to be sent through pipeline immediately + return OutputTransportMessageUrgentFrame(message=json.loads(self.create_closed_response())) async def _handle_ping(self, message: Dict[str, Any]) -> Frame | None: """Handle a 'ping' message from Genesys. @@ -842,20 +721,10 @@ class GenesysAudioHookSerializer(FrameSerializer): None if pong was sent directly via callback, otherwise OutputTransportMessageUrgentFrame with pong response. """ - # Create pong response - pong_response = self.create_pong_response() - # If we have a direct callback, use it for immediate response - if self._send_message_callback: - try: - await self._send_message_callback(pong_response) - logger.debug("Pong sent directly via callback") - return None - except Exception as e: - logger.error(f"Failed to send pong via callback: {e}") - - # Fallback: return as urgent frame to be sent through pipeline - return OutputTransportMessageUrgentFrame(message=json.loads(pong_response)) + logger.info(f"Sending pong response to Genesys...") + # Return as urgent frame to be sent through pipeline immediately + return OutputTransportMessageUrgentFrame(message=json.loads(self.create_pong_response())) async def _handle_pause(self, message: Dict[str, Any]) -> Frame | None: """Handle a 'pause' message from Genesys. From d7d8e93a3d116329a671aec3987d70c64ff2041a Mon Sep 17 00:00:00 2001 From: ssillerom Date: Tue, 27 Jan 2026 23:36:47 +0100 Subject: [PATCH 04/11] feature: added custom params in closed message to genesys, simplified create_* functions, simplified constructor method and simplified opened message --- src/pipecat/serializers/genesys.py | 245 +++++++++++++------ tests/test_genesys_serializer.py | 375 +++++++++++++++++++++++++++++ 2 files changed, 553 insertions(+), 67 deletions(-) create mode 100644 tests/test_genesys_serializer.py diff --git a/src/pipecat/serializers/genesys.py b/src/pipecat/serializers/genesys.py index 48637653b..678d0ef10 100644 --- a/src/pipecat/serializers/genesys.py +++ b/src/pipecat/serializers/genesys.py @@ -1,8 +1,15 @@ -""" -Use with Genesys Audio Connector to connect Genesys Cloud Contact Center with Pipecat pipelines. +"""Genesys AudioHook Serializer for Pipecat. -This connector implements the Genesys AudioHook protocol for bidirectional -audio streaming between Genesys Cloud contact center and Pipecat pipelines. +This module provides a serializer for integrating Pipecat pipelines with +Genesys Cloud Contact Center via the AudioHook protocol. + +Features: +- Bidirectional audio streaming (PCMU μ-law at 8kHz) +- Automatic protocol handshake handling (open/opened, close/closed, ping/pong) +- Input/output variables for Architect flow integration +- DTMF event support +- Barge-in (interruption) events +- Pause/resume support for hold scenarios Protocol Reference: - https://developer.genesys.cloud/devapps/audiohook @@ -10,7 +17,7 @@ Protocol Reference: Audio Format: - PCMU (μ-law) at 8kHz sample rate (preferred) - L16 (16-bit linear PCM) at 8kHz also supported -- Mono or Stereo (external on left, internal on right) +- Mono (external channel) or Stereo (external on left, internal on right) """ import json @@ -76,10 +83,11 @@ class GenesysAudioHookSerializer(FrameSerializer): AudioHook protocol messages. It supports: - Bidirectional audio streaming (PCMU at 8kHz) - - Session lifecycle management (open, close, pause) - - Probe mode for health checks - - Custom configuration passthrough - - Event messaging back to Genesys Cloud + - Automatic protocol handshake (open/opened, close/closed, ping/pong) + - Session lifecycle management with pause/resume support + - Custom input/output variables for Architect flow integration + - DTMF event handling + - Barge-in events for interruption support The AudioHook protocol uses: - Text WebSocket frames for JSON control messages @@ -88,17 +96,30 @@ class GenesysAudioHookSerializer(FrameSerializer): Example usage: ```python serializer = GenesysAudioHookSerializer( - session_id="abc-123", params=GenesysAudioHookSerializer.InputParams( - channel=AudioHookChannel.BOTH, + channel=AudioHookChannel.EXTERNAL, + supported_languages=["en-US", "es-ES"], + selected_language="en-US", ) ) - # Use with WebSocket transport - transport = WebsocketServerTransport( - serializer=serializer, - ... + # Use with FastAPI WebSocket transport + transport = FastAPIWebsocketTransport( + websocket=websocket, + params=FastAPIWebsocketParams( + audio_in_enabled=True, + audio_out_enabled=True, + serializer=serializer, + audio_out_fixed_packet_size=1600, # Important: prevents 429 rate limiting from Genesys + ), ) + + # Access call information after connection + participant = serializer.participant # ani, dnis, etc. + input_vars = serializer.input_variables # Custom vars from Architect + + # Set output variables to return to Architect + serializer.set_output_variables({"intent": "billing", "resolved": True}) ``` Attributes: @@ -110,13 +131,16 @@ class GenesysAudioHookSerializer(FrameSerializer): class InputParams(BaseModel): """Configuration parameters for GenesysAudioHookSerializer. - Parameters: - genesys_sample_rate: Sample rate used by Genesys (8000 Hz). + Attributes: + genesys_sample_rate: Sample rate used by Genesys (default: 8000 Hz). sample_rate: Optional override for pipeline input sample rate. channel: Which audio channels to process (external, internal, both). media_format: Audio format (PCMU or L16). process_external: Whether to process external (customer) audio. process_internal: Whether to process internal (agent) audio. + supported_languages: List of language codes the bot supports (e.g., ["en-US", "es-ES"]). + selected_language: Default language code to use. + start_paused: Whether to start the session in paused state. """ genesys_sample_rate: int = 8000 @@ -125,23 +149,24 @@ class GenesysAudioHookSerializer(FrameSerializer): media_format: AudioHookMediaFormat = AudioHookMediaFormat.PCMU process_external: bool = True process_internal: bool = False + supported_languages: Optional[List[str]] = None + selected_language: Optional[str] = None + start_paused: bool = False def __init__( self, - session_id: Optional[str] = None, params: Optional[InputParams] = None, ): """Initialize the GenesysAudioHookSerializer. Args: - session_id: The AudioHook session ID (received in open message). params: Configuration parameters. """ self._params = params or GenesysAudioHookSerializer.InputParams() - self._session_id = session_id or "" self._genesys_sample_rate = self._params.genesys_sample_rate self._sample_rate = 0 # Pipeline input rate, set in setup() + self._session_id = str(uuid.uuid4()) # Use Pipecat's official resampler if needed (SOXR) # Only used for TTS output (16kHz → 8kHz), input goes without resampling @@ -161,11 +186,12 @@ class GenesysAudioHookSerializer(FrameSerializer): self._custom_config: Optional[Dict[str, Any]] = None self._media_info: Optional[List[Dict[str, Any]]] = None self._input_variables: Optional[Dict[str, Any]] = None # Custom input from Genesys + self._output_variables: Optional[Dict[str, Any]] = None # Custom output to Genesys @property def session_id(self) -> str: - """Get the current session ID.""" + """Get the Genesys AudioHook session ID generated by the serializer.""" return self._session_id @property @@ -193,6 +219,34 @@ class GenesysAudioHookSerializer(FrameSerializer): """Get custom input variables from the open message.""" return self._input_variables + @property + def output_variables(self) -> Optional[Dict[str, Any]]: + """Get custom output variables to send back to Genesys.""" + return self._output_variables + + def set_output_variables(self, variables: Dict[str, Any]) -> None: + """Set custom output variables to send back to Genesys on close. + + These variables will be included in the 'closed' response when Genesys + closes the connection, making them available in the Architect flow. + + Args: + variables: Dictionary of custom variables to send to Genesys. + + Example: + ```python + # During the conversation, collect data and set it + serializer.set_output_variables({ + "intent": "billing_inquiry", + "customer_verified": True, + "summary": "Customer asked about their bill", + "transfer_to": "billing_queue" + }) + ``` + """ + self._output_variables = variables + logger.debug(f"Output variables set: {variables}") + async def setup(self, frame: StartFrame): """Sets up the serializer with pipeline configuration. @@ -279,7 +333,7 @@ class GenesysAudioHookSerializer(FrameSerializer): start_paused: bool = False, supported_languages: Optional[List[str]] = None, selected_language: Optional[str] = None, - ) -> str: + ) -> Dict[str, Any]: """Create an 'opened' response message for the client. This should be sent in response to an 'open' message from Genesys. @@ -290,10 +344,11 @@ class GenesysAudioHookSerializer(FrameSerializer): selected_language: The selected language code. Returns: - JSON string of the opened response message. + Dictionary of the opened response message. """ # Build channels list based on configuration channels: list[str] = [] + if self._params.channel == AudioHookChannel.EXTERNAL: channels = ["external"] elif self._params.channel == AudioHookChannel.INTERNAL: @@ -325,59 +380,88 @@ class GenesysAudioHookSerializer(FrameSerializer): ) self._is_open = True + logger.debug(f"AudioHook session opened: {self._session_id}") - return json.dumps(msg) + return msg - def create_closed_response(self) -> str: + def create_closed_response( + self, + output_variables: Optional[Dict[str, Any]] = None, + ) -> Dict[str, Any]: """Create a 'closed' response message. This should be sent in response to a 'close' message from Genesys. + Args: + output_variables: Optional custom variables to pass back to Genesys. + These will be available in the Architect flow after the AudioHook + action completes. + Returns: - JSON string of the closed response message. + Dictionary of the closed response message. + + Example: + ```python + # Pass custom data back to Genesys + serializer.create_closed_response( + output_variables={ + "intent": "billing_inquiry", + "customer_verified": True, + "summary": "Customer asked about their bill" + } + ) + ``` """ - msg = self._create_message(AudioHookMessageType.CLOSED) + parameters: Optional[Dict[str, Any]] = None + + if output_variables: + parameters = {"outputVariables": output_variables} + + msg = self._create_message( + AudioHookMessageType.CLOSED, + parameters=parameters, + ) self._is_open = False logger.debug(f"AudioHook session closed: {self._session_id}") - return json.dumps(msg) + return msg - def create_pong_response(self) -> str: + def create_pong_response(self) -> Dict[str, Any]: """Create a 'pong' response message. This should be sent in response to a 'ping' message from Genesys. Returns: - JSON string of the pong response message. + Dictionary of the pong response message. """ msg = self._create_message(AudioHookMessageType.PONG) - return json.dumps(msg) + return msg - def create_resumed_response(self) -> str: + def create_resumed_response(self) -> Dict[str, Any]: """Create a 'resumed' response message. This should be sent in response to a 'pause' message when ready to resume. Returns: - JSON string of the resumed response message. + Dictionary of the resumed response message. """ msg = self._create_message(AudioHookMessageType.RESUMED) self._is_paused = False logger.debug(f"AudioHook session resumed: {self._session_id}") - return json.dumps(msg) + return msg - def create_barge_in_event(self) -> str: + def create_barge_in_event(self) -> Dict[str, Any]: """Create a barge-in event message. This notifies Genesys Cloud that the user has interrupted the bot's audio output. Genesys will stop any queued audio playback. Returns: - JSON string of the barge-in event message. + Dictionary of the barge-in event message. """ msg = self._create_message( AudioHookMessageType.EVENT, @@ -390,7 +474,7 @@ class GenesysAudioHookSerializer(FrameSerializer): logger.debug("🔇 Barge-in event sent to Genesys") - return json.dumps(msg) + return msg def create_disconnect_message( self, @@ -398,7 +482,7 @@ class GenesysAudioHookSerializer(FrameSerializer): action: str = "transfer", output_variables: Optional[Dict[str, Any]] = None, info: Optional[str] = None, - ) -> str: + ) -> Dict[str, Any]: """Create a 'disconnect' message to initiate session termination. Args: @@ -408,7 +492,7 @@ class GenesysAudioHookSerializer(FrameSerializer): info: Optional additional information. Returns: - JSON string of the disconnect message. + Dictionary of the disconnect message. """ parameters: Dict[str, Any] = {"reason": reason} @@ -427,14 +511,14 @@ class GenesysAudioHookSerializer(FrameSerializer): ) logger.debug(f"AudioHook disconnect: reason={reason}, action={action}") - return json.dumps(msg) + return msg def create_error_message( self, code: int, message: str, retryable: bool = False, - ) -> str: + ) -> Dict[str, Any]: """Create an 'error' message. Args: @@ -443,7 +527,7 @@ class GenesysAudioHookSerializer(FrameSerializer): retryable: Whether the operation can be retried. Returns: - JSON string of the error message. + Dictionary of the error message. """ parameters = { "code": code, @@ -457,24 +541,26 @@ class GenesysAudioHookSerializer(FrameSerializer): ) logger.error(f"AudioHook error: {code} - {message}") - return json.dumps(msg) + return msg async def serialize(self, frame: Frame) -> str | bytes | None: """Serializes a Pipecat frame to Genesys AudioHook format. Handles conversion of various frame types to AudioHook messages: - - AudioRawFrame -> Binary audio data - - EndFrame/CancelFrame -> Disconnect message + - AudioRawFrame -> Binary PCMU audio data (resampled to 8kHz) + - EndFrame/CancelFrame -> Disconnect message (JSON) + - InterruptionFrame -> Barge-in event (JSON) - OutputTransportMessageFrame -> Pass-through JSON Args: frame: The Pipecat frame to serialize. Returns: - Serialized data as string (JSON) or bytes (audio), or None. + Serialized data as string (JSON) or bytes (audio), or None if + the frame type is not handled or session is not open. """ if isinstance(frame, (EndFrame, CancelFrame)): - return self.create_disconnect_message(reason="completed") + return json.dumps(self.create_disconnect_message(reason="completed")) elif isinstance(frame, AudioRawFrame): if not self._is_open or self._is_paused: @@ -501,7 +587,7 @@ class GenesysAudioHookSerializer(FrameSerializer): return bytes(serialized_data) elif isinstance(frame, InterruptionFrame): - return self.create_barge_in_event() + return json.dumps(self.create_barge_in_event()) elif isinstance(frame, (OutputTransportMessageFrame, OutputTransportMessageUrgentFrame)): # Pass through custom JSON messages @@ -514,18 +600,23 @@ class GenesysAudioHookSerializer(FrameSerializer): """Deserializes Genesys AudioHook data to Pipecat frames. Handles: - - Binary data -> InputAudioRawFrame - - JSON text -> Protocol messages (open, close, ping, pause, etc.) + - Binary data -> InputAudioRawFrame (converted from PCMU to PCM) + - JSON 'open' -> OutputTransportMessageUrgentFrame with 'opened' response + - JSON 'close' -> OutputTransportMessageUrgentFrame with 'closed' response + - JSON 'ping' -> OutputTransportMessageUrgentFrame with 'pong' response + - JSON 'pause' -> Sets is_paused=True, returns None + - JSON 'dtmf' -> InputDTMFFrame + - JSON 'update' -> Updates participant info, returns None + - JSON 'error' -> Logs error, returns None - For control messages (open, close, ping, pause), this method handles - the protocol response internally and logs the events. The application - should monitor session state via the is_open and is_paused properties. + Protocol responses (opened, closed, pong) are returned as urgent frames + to be sent immediately through the transport. Args: - data: The raw WebSocket data from Genesys. + data: The raw WebSocket data from Genesys (binary audio or JSON text). Returns: - A Pipecat frame corresponding to the data, or None if handled internally. + A Pipecat frame to process, or None if handled internally. """ # Binary data = audio if isinstance(data, bytes): @@ -643,13 +734,22 @@ class GenesysAudioHookSerializer(FrameSerializer): async def _handle_open(self, message: Dict[str, Any]) -> Frame | None: """Handle an 'open' message from Genesys. - This initializes the session with metadata from Genesys Cloud. + This initializes the session with metadata from Genesys Cloud and + automatically responds with an 'opened' message using the configured + InputParams (supported_languages, selected_language, start_paused). + + Extracts and stores: + - session_id: The AudioHook session identifier + - conversation_id: The Genesys conversation ID + - participant: Caller info (ani, dnis, etc.) + - input_variables: Custom variables from Architect flow + - media_info: Audio configuration from Genesys Args: - message: The open message. + message: The open message from Genesys. Returns: - None (response should be sent via create_opened_response()). + OutputTransportMessageUrgentFrame with the 'opened' response. """ self._session_id = message.get("id", str(uuid.uuid4())) @@ -686,18 +786,25 @@ class GenesysAudioHookSerializer(FrameSerializer): f"conversation={self._conversation_id}, ani={ani}" ) - # Note: Application should call create_opened_response() to respond - return None + return OutputTransportMessageUrgentFrame(message=self.create_opened_response( + start_paused=self._params.start_paused, + supported_languages=self._params.supported_languages, + selected_language=self._params.selected_language + )) async def _handle_close(self, message: Dict[str, Any]) -> Frame | None: """Handle a 'close' message from Genesys. + Automatically responds with a 'closed' message. If output_variables + were set via set_output_variables(), they will be included in the + response and made available in the Architect flow. + Args: - message: The close message. + message: The close message from Genesys. Returns: - OutputTransportMessageUrgentFrame with the closed response, - or None if no response should be sent. + OutputTransportMessageUrgentFrame with the closed response + (includes outputVariables if set). """ params = message.get("parameters", {}) reason = params.get("reason", "unknown") @@ -709,22 +816,26 @@ class GenesysAudioHookSerializer(FrameSerializer): logger.info(f"Sending closed response to Genesys...") # Return as urgent frame to be sent through pipeline immediately - return OutputTransportMessageUrgentFrame(message=json.loads(self.create_closed_response())) + # Include any output variables that were set during the session + return OutputTransportMessageUrgentFrame( + message=self.create_closed_response(output_variables=self._output_variables) + ) async def _handle_ping(self, message: Dict[str, Any]) -> Frame | None: """Handle a 'ping' message from Genesys. + Automatically responds with a 'pong' message to maintain the connection. + Args: - message: The ping message. + message: The ping message from Genesys. Returns: - None if pong was sent directly via callback, otherwise OutputTransportMessageUrgentFrame with pong response. """ logger.info(f"Sending pong response to Genesys...") # Return as urgent frame to be sent through pipeline immediately - return OutputTransportMessageUrgentFrame(message=json.loads(self.create_pong_response())) + return OutputTransportMessageUrgentFrame(message=self.create_pong_response()) async def _handle_pause(self, message: Dict[str, Any]) -> Frame | None: """Handle a 'pause' message from Genesys. diff --git a/tests/test_genesys_serializer.py b/tests/test_genesys_serializer.py new file mode 100644 index 000000000..65f624c9a --- /dev/null +++ b/tests/test_genesys_serializer.py @@ -0,0 +1,375 @@ +"""Tests for the Genesys AudioHook serializer.""" + +import json +import pytest + +from pipecat.frames.frames import InputDTMFFrame, OutputTransportMessageUrgentFrame + +from pipecat.serializers.genesys import ( + GenesysAudioHookSerializer, + AudioHookChannel +) + + +class TestGenesysAudioHookSerializer: + """Tests for GenesysAudioHookSerializer.""" + + # ==================== Initialization Tests ==================== + + def test_create_serializer_default_params(self): + """Test creating serializer with default parameters.""" + serializer = GenesysAudioHookSerializer() + + # session_id is auto-generated as UUID + assert serializer.session_id != "" + assert len(serializer.session_id) == 36 # UUID format + assert serializer.is_open is False + assert serializer.is_paused is False + + def test_create_serializer_with_custom_params(self): + """Test creating serializer with custom parameters.""" + params = GenesysAudioHookSerializer.InputParams( + channel=AudioHookChannel.BOTH, + sample_rate=16000, + supported_languages=["es-ES", "en-US"], + selected_language="es-ES", + start_paused=True, + ) + serializer = GenesysAudioHookSerializer(params=params) + + assert serializer.session_id != "" + + # ==================== Response Creation Tests ==================== + + def test_create_opened_response(self): + """Test creating an opened response message.""" + serializer = GenesysAudioHookSerializer() + + msg = serializer.create_opened_response() + + assert msg["type"] == "opened" + assert msg["version"] == "2" + assert msg["id"] == serializer.session_id + assert "parameters" in msg + assert serializer.is_open is True + + def test_create_opened_response_with_languages(self): + """Test creating an opened response with language options.""" + serializer = GenesysAudioHookSerializer() + + msg = serializer.create_opened_response( + supported_languages=["es", "en", "fr"], + selected_language="es", + ) + + assert msg["parameters"]["supportedLanguages"] == ["es", "en", "fr"] + assert msg["parameters"]["selectedLanguage"] == "es" + + def test_create_pong_response(self): + """Test creating a pong response message.""" + serializer = GenesysAudioHookSerializer() + + msg = serializer.create_pong_response() + + assert msg["type"] == "pong" + assert msg["id"] == serializer.session_id + + def test_create_closed_response(self): + """Test creating a closed response message.""" + serializer = GenesysAudioHookSerializer() + serializer._is_open = True + + msg = serializer.create_closed_response() + + assert msg["type"] == "closed" + assert serializer.is_open is False + assert "parameters" not in msg # No parameters when no output_variables + + def test_create_closed_response_with_output_variables(self): + """Test creating a closed response with custom output variables.""" + serializer = GenesysAudioHookSerializer() + serializer._is_open = True + + msg = serializer.create_closed_response( + output_variables={ + "intent": "billing_inquiry", + "customer_verified": True, + "summary": "Customer asked about their bill" + } + ) + + assert msg["type"] == "closed" + assert msg["parameters"]["outputVariables"]["intent"] == "billing_inquiry" + assert msg["parameters"]["outputVariables"]["customer_verified"] is True + assert msg["parameters"]["outputVariables"]["summary"] == "Customer asked about their bill" + + def test_create_resumed_response(self): + """Test creating a resumed response message.""" + serializer = GenesysAudioHookSerializer() + serializer._is_paused = True + + msg = serializer.create_resumed_response() + + assert msg["type"] == "resumed" + assert serializer.is_paused is False + + def test_create_disconnect_message(self): + """Test creating a disconnect message.""" + serializer = GenesysAudioHookSerializer() + + msg = serializer.create_disconnect_message( + reason="completed", + action="transfer", + ) + + assert msg["type"] == "disconnect" + assert msg["parameters"]["reason"] == "completed" + assert msg["parameters"]["outputVariables"]["action"] == "transfer" + + def test_create_disconnect_message_with_output_variables(self): + """Test creating a disconnect message with custom output variables.""" + serializer = GenesysAudioHookSerializer() + + msg = serializer.create_disconnect_message( + reason="completed", + action="finished", + output_variables={"result": "success", "code": "123"}, + ) + + assert msg["parameters"]["outputVariables"]["result"] == "success" + assert msg["parameters"]["outputVariables"]["code"] == "123" + + def test_create_error_message(self): + """Test creating an error message.""" + serializer = GenesysAudioHookSerializer() + + msg = serializer.create_error_message( + code=500, + message="Internal error", + retryable=True, + ) + + assert msg["type"] == "error" + assert msg["parameters"]["code"] == 500 + assert msg["parameters"]["message"] == "Internal error" + assert msg["parameters"]["retryable"] is True + + # ==================== Message Handling Tests ==================== + + @pytest.mark.asyncio + async def test_handle_open_message(self, sample_open_message): + """Test handling an open message returns opened frame.""" + serializer = GenesysAudioHookSerializer() + + result = await serializer.deserialize(json.dumps(sample_open_message)) + + # Now returns OutputTransportMessageUrgentFrame with opened response + assert isinstance(result, OutputTransportMessageUrgentFrame) + assert result.message["type"] == "opened" + assert serializer.session_id == "test-session-123" + assert serializer.conversation_id == "conv-456" + + @pytest.mark.asyncio + async def test_handle_open_message_extracts_participant(self, sample_open_message): + """Test that open message extracts participant info.""" + serializer = GenesysAudioHookSerializer() + + await serializer.deserialize(json.dumps(sample_open_message)) + + assert serializer.participant is not None + assert serializer.participant["ani"] == "+1234567890" + assert serializer.participant["dnis"] == "+0987654321" + + @pytest.mark.asyncio + async def test_handle_open_message_uses_params(self, sample_open_message): + """Test that open message uses InputParams for response.""" + params = GenesysAudioHookSerializer.InputParams( + supported_languages=["es-ES", "en-US"], + selected_language="es-ES", + start_paused=True, + ) + serializer = GenesysAudioHookSerializer(params=params) + + result = await serializer.deserialize(json.dumps(sample_open_message)) + + assert isinstance(result, OutputTransportMessageUrgentFrame) + assert result.message["parameters"]["supportedLanguages"] == ["es-ES", "en-US"] + assert result.message["parameters"]["selectedLanguage"] == "es-ES" + assert result.message["parameters"]["startPaused"] is True + + @pytest.mark.asyncio + async def test_handle_open_message_extracts_input_variables(self, sample_open_message_with_input_variables): + """Test that open message extracts inputVariables from Genesys.""" + serializer = GenesysAudioHookSerializer() + + await serializer.deserialize(json.dumps(sample_open_message_with_input_variables)) + + assert serializer.input_variables is not None + assert serializer.input_variables["customer_id"] == "cust-789" + assert serializer.input_variables["queue_name"] == "billing" + assert serializer.input_variables["priority"] == "high" + assert serializer.input_variables["language"] == "es-ES" + + @pytest.mark.asyncio + async def test_handle_ping_message(self, sample_ping_message): + """Test handling a ping message returns pong frame.""" + serializer = GenesysAudioHookSerializer() + + result = await serializer.deserialize(json.dumps(sample_ping_message)) + + assert isinstance(result, OutputTransportMessageUrgentFrame) + assert result.message["type"] == "pong" + + @pytest.mark.asyncio + async def test_handle_close_message(self, sample_close_message): + """Test handling a close message returns closed frame.""" + serializer = GenesysAudioHookSerializer() + serializer._is_open = True + + result = await serializer.deserialize(json.dumps(sample_close_message)) + + assert isinstance(result, OutputTransportMessageUrgentFrame) + assert result.message["type"] == "closed" + assert serializer.is_open is False + + @pytest.mark.asyncio + async def test_handle_close_message_includes_output_variables(self, sample_close_message): + """Test that close response includes output variables when set.""" + serializer = GenesysAudioHookSerializer() + serializer._is_open = True + + # Set output variables before close + serializer.set_output_variables({ + "intent": "support", + "resolved": True, + "transfer_to": "agent_queue" + }) + + result = await serializer.deserialize(json.dumps(sample_close_message)) + + assert isinstance(result, OutputTransportMessageUrgentFrame) + assert result.message["type"] == "closed" + assert result.message["parameters"]["outputVariables"]["intent"] == "support" + assert result.message["parameters"]["outputVariables"]["resolved"] is True + assert result.message["parameters"]["outputVariables"]["transfer_to"] == "agent_queue" + + # ==================== Output Variables Tests ==================== + + def test_set_output_variables(self): + """Test setting output variables.""" + serializer = GenesysAudioHookSerializer() + + assert serializer.output_variables is None + + serializer.set_output_variables({ + "intent": "billing", + "score": 0.95 + }) + + assert serializer.output_variables is not None + assert serializer.output_variables["intent"] == "billing" + assert serializer.output_variables["score"] == 0.95 + + def test_set_output_variables_overwrites(self): + """Test that setting output variables overwrites previous values.""" + serializer = GenesysAudioHookSerializer() + + serializer.set_output_variables({"first": "value"}) + serializer.set_output_variables({"second": "value"}) + + assert "first" not in serializer.output_variables + assert serializer.output_variables["second"] == "value" + + @pytest.mark.asyncio + async def test_handle_pause_message(self, sample_pause_message): + """Test handling a pause message.""" + serializer = GenesysAudioHookSerializer() + serializer._is_open = True + + result = await serializer.deserialize(json.dumps(sample_pause_message)) + + assert result is None # Pause is handled internally + assert serializer.is_paused is True + + @pytest.mark.asyncio + async def test_handle_update_message(self, sample_update_message): + """Test handling an update message.""" + serializer = GenesysAudioHookSerializer() + + result = await serializer.deserialize(json.dumps(sample_update_message)) + + assert result is None # Update is handled internally + assert serializer.participant is not None + assert serializer.participant["name"] == "John Doe" + + @pytest.mark.asyncio + async def test_handle_error_message(self, sample_error_message): + """Test handling an error message.""" + serializer = GenesysAudioHookSerializer() + + result = await serializer.deserialize(json.dumps(sample_error_message)) + + assert result is None # Error is logged but returns None + + # ==================== DTMF Tests ==================== + + @pytest.mark.asyncio + async def test_handle_dtmf_digit(self, sample_dtmf_message): + """Test handling a DTMF digit message.""" + serializer = GenesysAudioHookSerializer() + + result = await serializer.deserialize(json.dumps(sample_dtmf_message)) + + assert isinstance(result, InputDTMFFrame) + assert result.button.value == "5" + + @pytest.mark.asyncio + async def test_handle_dtmf_star(self, sample_dtmf_star_message): + """Test handling a DTMF star (*) message.""" + serializer = GenesysAudioHookSerializer() + + result = await serializer.deserialize(json.dumps(sample_dtmf_star_message)) + + assert isinstance(result, InputDTMFFrame) + assert result.button.value == "*" + + @pytest.mark.asyncio + async def test_handle_dtmf_hash(self, sample_dtmf_hash_message): + """Test handling a DTMF hash (#) message.""" + serializer = GenesysAudioHookSerializer() + + result = await serializer.deserialize(json.dumps(sample_dtmf_hash_message)) + + assert isinstance(result, InputDTMFFrame) + assert result.button.value == "#" + + @pytest.mark.asyncio + async def test_handle_dtmf_empty_digit(self): + """Test handling a DTMF message without digit.""" + serializer = GenesysAudioHookSerializer() + + dtmf_msg = { + "version": "2", + "type": "dtmf", + "seq": 6, + "id": "test-session-123", + "parameters": {}, + } + + result = await serializer.deserialize(json.dumps(dtmf_msg)) + + assert result is None # No digit provided + + # ==================== Sequence Number Tests ==================== + + def test_sequence_numbers_increment(self): + """Test that sequence numbers increment correctly.""" + serializer = GenesysAudioHookSerializer() + + response1 = serializer.create_pong_response() + response2 = serializer.create_pong_response() + response3 = serializer.create_pong_response() + + assert response1["seq"] == 1 + assert response2["seq"] == 2 + assert response3["seq"] == 3 From 9102e81cb84bf23f94814e287919343b35ed49ae Mon Sep 17 00:00:00 2001 From: ssillerom Date: Tue, 27 Jan 2026 23:39:43 +0100 Subject: [PATCH 05/11] added tests to the PR --- tests/genesys/__init__.py | 6 + tests/genesys/conftest.py | 186 ++++++++++++++++++ .../{ => genesys}/test_genesys_serializer.py | 0 3 files changed, 192 insertions(+) create mode 100644 tests/genesys/__init__.py create mode 100644 tests/genesys/conftest.py rename tests/{ => genesys}/test_genesys_serializer.py (100%) diff --git a/tests/genesys/__init__.py b/tests/genesys/__init__.py new file mode 100644 index 000000000..05982ea05 --- /dev/null +++ b/tests/genesys/__init__.py @@ -0,0 +1,6 @@ +# Copyright (c) 2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Tests for Genesys AudioHook serializer.""" diff --git a/tests/genesys/conftest.py b/tests/genesys/conftest.py new file mode 100644 index 000000000..656e815a5 --- /dev/null +++ b/tests/genesys/conftest.py @@ -0,0 +1,186 @@ +"""Pytest fixtures for Genesys AudioHook serializer tests. + +These fixtures provide sample AudioHook protocol messages for testing +the GenesysAudioHookSerializer. They are scoped to this directory only. +""" + +import pytest + + +@pytest.fixture +def sample_open_message(): + """Sample AudioHook open message from Genesys.""" + return { + "version": "2", + "type": "open", + "seq": 1, + "id": "test-session-123", + "parameters": { + "conversationId": "conv-456", + "participant": { + "ani": "+1234567890", + "dnis": "+0987654321", + }, + "media": [ + { + "type": "audio", + "format": "PCMU", + "channels": ["external"], + "rate": 8000, + } + ], + }, + } + + +@pytest.fixture +def sample_open_message_with_input_variables(): + """Sample AudioHook open message with custom inputVariables from Genesys.""" + return { + "version": "2", + "type": "open", + "seq": 1, + "id": "test-session-123", + "parameters": { + "conversationId": "conv-456", + "participant": { + "ani": "+1234567890", + "dnis": "+0987654321", + }, + "media": [ + { + "type": "audio", + "format": "PCMU", + "channels": ["external"], + "rate": 8000, + } + ], + "inputVariables": { + "customer_id": "cust-789", + "queue_name": "billing", + "priority": "high", + "language": "es-ES", + }, + }, + } + + +@pytest.fixture +def sample_ping_message(): + """Sample AudioHook ping message.""" + return { + "version": "2", + "type": "ping", + "seq": 5, + "id": "test-session-123", + "position": "PT10.5S", + } + + +@pytest.fixture +def sample_close_message(): + """Sample AudioHook close message from Genesys.""" + return { + "version": "2", + "type": "close", + "seq": 10, + "id": "test-session-123", + "position": "PT30.0S", + "parameters": { + "reason": "disconnect", + }, + } + + +@pytest.fixture +def sample_pause_message(): + """Sample AudioHook pause message.""" + return { + "version": "2", + "type": "pause", + "seq": 7, + "id": "test-session-123", + "position": "PT15.0S", + "parameters": { + "reason": "hold", + }, + } + + +@pytest.fixture +def sample_update_message(): + """Sample AudioHook update message.""" + return { + "version": "2", + "type": "update", + "seq": 8, + "id": "test-session-123", + "position": "PT20.0S", + "parameters": { + "participant": { + "ani": "+1234567890", + "dnis": "+0987654321", + "name": "John Doe", + }, + }, + } + + +@pytest.fixture +def sample_error_message(): + """Sample AudioHook error message.""" + return { + "version": "2", + "type": "error", + "seq": 9, + "id": "test-session-123", + "parameters": { + "code": 500, + "message": "Internal server error", + }, + } + + +@pytest.fixture +def sample_dtmf_message(): + """Sample AudioHook DTMF message.""" + return { + "version": "2", + "type": "dtmf", + "seq": 6, + "id": "test-session-123", + "position": "PT12.0S", + "parameters": { + "digit": "5", + }, + } + + +@pytest.fixture +def sample_dtmf_star_message(): + """Sample AudioHook DTMF message with star key.""" + return { + "version": "2", + "type": "dtmf", + "seq": 6, + "id": "test-session-123", + "position": "PT12.0S", + "parameters": { + "digit": "*", + }, + } + + +@pytest.fixture +def sample_dtmf_hash_message(): + """Sample AudioHook DTMF message with hash key.""" + return { + "version": "2", + "type": "dtmf", + "seq": 6, + "id": "test-session-123", + "position": "PT12.0S", + "parameters": { + "digit": "#", + }, + } diff --git a/tests/test_genesys_serializer.py b/tests/genesys/test_genesys_serializer.py similarity index 100% rename from tests/test_genesys_serializer.py rename to tests/genesys/test_genesys_serializer.py From 55e0d4ecc45cb2e3df2b9475cf0ce51011728ac2 Mon Sep 17 00:00:00 2001 From: ssillerom Date: Wed, 28 Jan 2026 08:59:28 +0100 Subject: [PATCH 06/11] ruff fixes done --- src/pipecat/serializers/genesys.py | 7 +++---- tests/genesys/test_genesys_serializer.py | 7 ++----- 2 files changed, 5 insertions(+), 9 deletions(-) diff --git a/src/pipecat/serializers/genesys.py b/src/pipecat/serializers/genesys.py index 678d0ef10..8b315dcdf 100644 --- a/src/pipecat/serializers/genesys.py +++ b/src/pipecat/serializers/genesys.py @@ -30,6 +30,8 @@ from loguru import logger from pydantic import BaseModel from pipecat.audio.dtmf.types import KeypadEntry +from pipecat.audio.resamplers.soxr_stream_resampler import SOXRStreamAudioResampler +from pipecat.audio.utils import pcm_to_ulaw, ulaw_to_pcm from pipecat.frames.frames import ( AudioRawFrame, CancelFrame, @@ -37,14 +39,12 @@ from pipecat.frames.frames import ( Frame, InputAudioRawFrame, InputDTMFFrame, + InterruptionFrame, OutputTransportMessageFrame, OutputTransportMessageUrgentFrame, StartFrame, - InterruptionFrame ) from pipecat.serializers.base_serializer import FrameSerializer -from pipecat.audio.resamplers.soxr_stream_resampler import SOXRStreamAudioResampler -from pipecat.audio.utils import ulaw_to_pcm, pcm_to_ulaw class AudioHookMessageType(str, Enum): @@ -832,7 +832,6 @@ class GenesysAudioHookSerializer(FrameSerializer): Returns: OutputTransportMessageUrgentFrame with pong response. """ - logger.info(f"Sending pong response to Genesys...") # Return as urgent frame to be sent through pipeline immediately return OutputTransportMessageUrgentFrame(message=self.create_pong_response()) diff --git a/tests/genesys/test_genesys_serializer.py b/tests/genesys/test_genesys_serializer.py index 65f624c9a..00942ab3c 100644 --- a/tests/genesys/test_genesys_serializer.py +++ b/tests/genesys/test_genesys_serializer.py @@ -1,14 +1,11 @@ """Tests for the Genesys AudioHook serializer.""" import json + import pytest from pipecat.frames.frames import InputDTMFFrame, OutputTransportMessageUrgentFrame - -from pipecat.serializers.genesys import ( - GenesysAudioHookSerializer, - AudioHookChannel -) +from pipecat.serializers.genesys import AudioHookChannel, GenesysAudioHookSerializer class TestGenesysAudioHookSerializer: From a4acafd3be7ab1dbda015ba103a945674ae4d62c Mon Sep 17 00:00:00 2001 From: ssillerom Date: Wed, 28 Jan 2026 10:54:26 +0100 Subject: [PATCH 07/11] feature: added event handlers in constructor and call func in each _handle_* func --- src/pipecat/serializers/genesys.py | 26 +++++++++++++++++++++++++- 1 file changed, 25 insertions(+), 1 deletion(-) diff --git a/src/pipecat/serializers/genesys.py b/src/pipecat/serializers/genesys.py index 8b315dcdf..1914eacea 100644 --- a/src/pipecat/serializers/genesys.py +++ b/src/pipecat/serializers/genesys.py @@ -187,7 +187,16 @@ class GenesysAudioHookSerializer(FrameSerializer): self._media_info: Optional[List[Dict[str, Any]]] = None self._input_variables: Optional[Dict[str, Any]] = None # Custom input from Genesys self._output_variables: Optional[Dict[str, Any]] = None # Custom output to Genesys - + + # Event handlers + self._register_event_handler("on_open") + self._register_event_handler("on_close") + self._register_event_handler("on_ping") + self._register_event_handler("on_pause") + self._register_event_handler("on_update") + self._register_event_handler("on_error") + self._register_event_handler("on_dtmf") + @property def session_id(self) -> str: @@ -786,6 +795,8 @@ class GenesysAudioHookSerializer(FrameSerializer): f"conversation={self._conversation_id}, ani={ani}" ) + await self._call_event_handler("on_open", message) + return OutputTransportMessageUrgentFrame(message=self.create_opened_response( start_paused=self._params.start_paused, supported_languages=self._params.supported_languages, @@ -814,6 +825,8 @@ class GenesysAudioHookSerializer(FrameSerializer): self._is_open = False logger.info(f"Sending closed response to Genesys...") + + await self._call_event_handler("on_close", message) # Return as urgent frame to be sent through pipeline immediately # Include any output variables that were set during the session @@ -833,6 +846,9 @@ class GenesysAudioHookSerializer(FrameSerializer): OutputTransportMessageUrgentFrame with pong response. """ logger.info(f"Sending pong response to Genesys...") + + await self._call_event_handler("on_ping", message) + # Return as urgent frame to be sent through pipeline immediately return OutputTransportMessageUrgentFrame(message=self.create_pong_response()) @@ -854,6 +870,8 @@ class GenesysAudioHookSerializer(FrameSerializer): logger.info(f"AudioHook pause request: reason={reason}") self._is_paused = True + + await self._call_event_handler("on_pause", message) # Note: Application should call create_resumed_response() when ready return None @@ -875,6 +893,8 @@ class GenesysAudioHookSerializer(FrameSerializer): self._participant = params["participant"] logger.debug(f"AudioHook update received: {params}") + + await self._call_event_handler("on_update", message) return None @@ -892,6 +912,8 @@ class GenesysAudioHookSerializer(FrameSerializer): error_msg = params.get("message", "Unknown error") logger.error(f"AudioHook error from Genesys: {code} - {error_msg}") + + await self._call_event_handler("on_error", message) return None @@ -915,6 +937,8 @@ class GenesysAudioHookSerializer(FrameSerializer): return None logger.info(f"DTMF received: {digit}") + + await self._call_event_handler("on_dtmf", message) try: return InputDTMFFrame(KeypadEntry(digit)) From ef6bbace98ec55e177642aa582e3bb02e7b82684 Mon Sep 17 00:00:00 2001 From: ssillerom Date: Wed, 28 Jan 2026 15:40:24 +0100 Subject: [PATCH 08/11] fixes: super init inhereted class to set event hanlders in the construct --- src/pipecat/serializers/genesys.py | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/src/pipecat/serializers/genesys.py b/src/pipecat/serializers/genesys.py index 1914eacea..cced94997 100644 --- a/src/pipecat/serializers/genesys.py +++ b/src/pipecat/serializers/genesys.py @@ -156,12 +156,15 @@ class GenesysAudioHookSerializer(FrameSerializer): def __init__( self, params: Optional[InputParams] = None, + **kwargs, ): """Initialize the GenesysAudioHookSerializer. Args: params: Configuration parameters. + **kwargs: Additional arguments passed to BaseObject (e.g., name). """ + super().__init__(**kwargs) self._params = params or GenesysAudioHookSerializer.InputParams() self._genesys_sample_rate = self._params.genesys_sample_rate @@ -599,8 +602,13 @@ class GenesysAudioHookSerializer(FrameSerializer): return json.dumps(self.create_barge_in_event()) elif isinstance(frame, (OutputTransportMessageFrame, OutputTransportMessageUrgentFrame)): - # Pass through custom JSON messages - return json.dumps(frame.message) + # Only pass through AudioHook protocol messages (those with "version" field) + # Filter out RTVI and other non-AudioHook messages + if isinstance(frame.message, dict) and "version" in frame.message: + return json.dumps(frame.message) + else: + # Not an AudioHook message, ignore + return None # Ignore other frames - we don't need to process them here return None From c5be67f2935e8909a4fabd3573c258038689bcbd Mon Sep 17 00:00:00 2001 From: ssillerom Date: Wed, 28 Jan 2026 17:56:21 +0100 Subject: [PATCH 09/11] fix: create disconnect message passing output vars --- src/pipecat/serializers/genesys.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/pipecat/serializers/genesys.py b/src/pipecat/serializers/genesys.py index cced94997..31e32f4dc 100644 --- a/src/pipecat/serializers/genesys.py +++ b/src/pipecat/serializers/genesys.py @@ -572,7 +572,7 @@ class GenesysAudioHookSerializer(FrameSerializer): the frame type is not handled or session is not open. """ if isinstance(frame, (EndFrame, CancelFrame)): - return json.dumps(self.create_disconnect_message(reason="completed")) + return json.dumps(self.create_disconnect_message(output_variables=self.output_variables, reason="completed")) elif isinstance(frame, AudioRawFrame): if not self._is_open or self._is_paused: From 2612fae5271b2aa2298694be557154b386e99ff3 Mon Sep 17 00:00:00 2001 From: ssillerom Date: Wed, 28 Jan 2026 18:02:51 +0100 Subject: [PATCH 10/11] ruff linting --- src/pipecat/serializers/genesys.py | 299 +++++++++++++++-------------- 1 file changed, 153 insertions(+), 146 deletions(-) diff --git a/src/pipecat/serializers/genesys.py b/src/pipecat/serializers/genesys.py index 31e32f4dc..d5d37d12b 100644 --- a/src/pipecat/serializers/genesys.py +++ b/src/pipecat/serializers/genesys.py @@ -9,14 +9,14 @@ Features: - Input/output variables for Architect flow integration - DTMF event support - Barge-in (interruption) events -- Pause/resume support for hold scenarios +- Pause/resume support for hold scenarios (optional) Protocol Reference: - https://developer.genesys.cloud/devapps/audiohook Audio Format: - PCMU (μ-law) at 8kHz sample rate (preferred) -- L16 (16-bit linear PCM) at 8kHz also supported +- L16 (16-bit linear PCM) at 8kHz also supported - Mono (external channel) or Stereo (external on left, internal on right) """ @@ -49,6 +49,7 @@ from pipecat.serializers.base_serializer import FrameSerializer class AudioHookMessageType(str, Enum): """AudioHook protocol message types.""" + OPEN = "open" OPENED = "opened" CLOSE = "close" @@ -65,15 +66,17 @@ class AudioHookMessageType(str, Enum): class AudioHookChannel(str, Enum): """AudioHook audio channel configuration.""" + EXTERNAL = "external" # Customer audio only (mono) INTERNAL = "internal" # Agent audio only (mono) - BOTH = "both" # Stereo: external=left, internal=right + BOTH = "both" # Stereo: external=left, internal=right class AudioHookMediaFormat(str, Enum): """Supported audio formats.""" + PCMU = "PCMU" # μ-law, 8kHz - L16 = "L16" # 16-bit linear PCM, 8kHz + L16 = "L16" # 16-bit linear PCM, 8kHz class GenesysAudioHookSerializer(FrameSerializer): @@ -81,7 +84,7 @@ class GenesysAudioHookSerializer(FrameSerializer): This serializer handles converting between Pipecat frames and Genesys AudioHook protocol messages. It supports: - + - Bidirectional audio streaming (PCMU at 8kHz) - Automatic protocol handshake (open/opened, close/closed, ping/pong) - Session lifecycle management with pause/resume support @@ -102,7 +105,7 @@ class GenesysAudioHookSerializer(FrameSerializer): selected_language="en-US", ) ) - + # Use with FastAPI WebSocket transport transport = FastAPIWebsocketTransport( websocket=websocket, @@ -113,11 +116,11 @@ class GenesysAudioHookSerializer(FrameSerializer): audio_out_fixed_packet_size=1600, # Important: prevents 429 rate limiting from Genesys ), ) - + # Access call information after connection participant = serializer.participant # ani, dnis, etc. input_vars = serializer.input_variables # Custom vars from Architect - + # Set output variables to return to Architect serializer.set_output_variables({"intent": "billing", "resolved": True}) ``` @@ -166,23 +169,23 @@ class GenesysAudioHookSerializer(FrameSerializer): """ super().__init__(**kwargs) self._params = params or GenesysAudioHookSerializer.InputParams() - + self._genesys_sample_rate = self._params.genesys_sample_rate self._sample_rate = 0 # Pipeline input rate, set in setup() self._session_id = str(uuid.uuid4()) - + # Use Pipecat's official resampler if needed (SOXR) # Only used for TTS output (16kHz → 8kHz), input goes without resampling self._input_resampler = SOXRStreamAudioResampler() self._output_resampler = SOXRStreamAudioResampler() - + # Protocol state self._client_seq = 0 self._server_seq = 0 self._is_open = False self._is_paused = False self._position = timedelta(0) - + # Session metadata self._conversation_id: Optional[str] = None self._participant: Optional[Dict[str, Any]] = None @@ -200,7 +203,6 @@ class GenesysAudioHookSerializer(FrameSerializer): self._register_event_handler("on_error") self._register_event_handler("on_dtmf") - @property def session_id(self) -> str: """Get the Genesys AudioHook session ID generated by the serializer.""" @@ -238,13 +240,13 @@ class GenesysAudioHookSerializer(FrameSerializer): def set_output_variables(self, variables: Dict[str, Any]) -> None: """Set custom output variables to send back to Genesys on close. - + These variables will be included in the 'closed' response when Genesys closes the connection, making them available in the Architect flow. - + Args: variables: Dictionary of custom variables to send to Genesys. - + Example: ```python # During the conversation, collect data and set it @@ -267,13 +269,13 @@ class GenesysAudioHookSerializer(FrameSerializer): """ self._sample_rate = self._params.sample_rate or frame.audio_in_sample_rate logger.debug(f"GenesysAudioHookSerializer setup with sample_rate={self._sample_rate}") - + def _format_position(self, position: timedelta) -> str: """Format a timedelta as ISO 8601 duration string. - + Args: position: The timedelta to format. - + Returns: ISO 8601 duration string (e.g., "PT1.5S"). """ @@ -282,10 +284,10 @@ class GenesysAudioHookSerializer(FrameSerializer): def _parse_position(self, position_str: str) -> timedelta: """Parse an ISO 8601 duration string to timedelta. - + Args: position_str: ISO 8601 duration string (e.g., "PT1.5S"). - + Returns: Corresponding timedelta. """ @@ -310,16 +312,16 @@ class GenesysAudioHookSerializer(FrameSerializer): include_position: bool = True, ) -> Dict[str, Any]: """Create a protocol message with common fields. - + Based on the Genesys AudioHook protocol, responses include: - seq: Server's sequence number (incremented per message) - clientseq: Echo of the client's last sequence number - + Args: msg_type: The message type. parameters: Optional parameters object. include_position: Whether to include position field. - + Returns: The message dictionary. """ @@ -331,13 +333,13 @@ class GenesysAudioHookSerializer(FrameSerializer): "clientseq": self._client_seq, "id": self._session_id, } - + if include_position: msg["position"] = self._format_position(self._position) - + if parameters: msg["parameters"] = parameters - + return msg def create_opened_response( @@ -347,27 +349,27 @@ class GenesysAudioHookSerializer(FrameSerializer): selected_language: Optional[str] = None, ) -> Dict[str, Any]: """Create an 'opened' response message for the client. - + This should be sent in response to an 'open' message from Genesys. - + Args: start_paused: Whether to start the session paused. supported_languages: List of supported language codes. selected_language: The selected language code. - + Returns: Dictionary of the opened response message. """ # Build channels list based on configuration channels: list[str] = [] - + if self._params.channel == AudioHookChannel.EXTERNAL: channels = ["external"] elif self._params.channel == AudioHookChannel.INTERNAL: channels = ["internal"] elif self._params.channel == AudioHookChannel.BOTH: channels = ["external", "internal"] - + parameters = { "startPaused": start_paused, "media": [ @@ -379,22 +381,22 @@ class GenesysAudioHookSerializer(FrameSerializer): } ], } - + if supported_languages: parameters["supportedLanguages"] = supported_languages if selected_language: parameters["selectedLanguage"] = selected_language - + msg = self._create_message( AudioHookMessageType.OPENED, parameters=parameters, include_position=False, # opened doesn't need position ) - + self._is_open = True logger.debug(f"AudioHook session opened: {self._session_id}") - + return msg def create_closed_response( @@ -402,17 +404,17 @@ class GenesysAudioHookSerializer(FrameSerializer): output_variables: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: """Create a 'closed' response message. - + This should be sent in response to a 'close' message from Genesys. - + Args: output_variables: Optional custom variables to pass back to Genesys. These will be available in the Architect flow after the AudioHook action completes. - + Returns: Dictionary of the closed response message. - + Example: ```python # Pass custom data back to Genesys @@ -426,25 +428,25 @@ class GenesysAudioHookSerializer(FrameSerializer): ``` """ parameters: Optional[Dict[str, Any]] = None - + if output_variables: parameters = {"outputVariables": output_variables} - + msg = self._create_message( AudioHookMessageType.CLOSED, parameters=parameters, ) - + self._is_open = False logger.debug(f"AudioHook session closed: {self._session_id}") - + return msg def create_pong_response(self) -> Dict[str, Any]: """Create a 'pong' response message. - + This should be sent in response to a 'ping' message from Genesys. - + Returns: Dictionary of the pong response message. """ @@ -453,39 +455,35 @@ class GenesysAudioHookSerializer(FrameSerializer): def create_resumed_response(self) -> Dict[str, Any]: """Create a 'resumed' response message. - + This should be sent in response to a 'pause' message when ready to resume. - + Returns: Dictionary of the resumed response message. """ msg = self._create_message(AudioHookMessageType.RESUMED) - + self._is_paused = False logger.debug(f"AudioHook session resumed: {self._session_id}") - + return msg def create_barge_in_event(self) -> Dict[str, Any]: """Create a barge-in event message. - + This notifies Genesys Cloud that the user has interrupted the bot's audio output. Genesys will stop any queued audio playback. - + Returns: Dictionary of the barge-in event message. """ msg = self._create_message( AudioHookMessageType.EVENT, - parameters={ - "entities": [ - {"type": "barge_in", "data": {}} - ] - }, + parameters={"entities": [{"type": "barge_in", "data": {}}]}, ) - + logger.debug("🔇 Barge-in event sent to Genesys") - + return msg def create_disconnect_message( @@ -496,32 +494,32 @@ class GenesysAudioHookSerializer(FrameSerializer): info: Optional[str] = None, ) -> Dict[str, Any]: """Create a 'disconnect' message to initiate session termination. - + Args: reason: Disconnect reason (e.g., "completed", "error"). action: Action to take ("transfer" to agent, "finished" if completed). output_variables: Custom output variables to pass back to Genesys. info: Optional additional information. - + Returns: Dictionary of the disconnect message. """ parameters: Dict[str, Any] = {"reason": reason} - + # Build outputVariables out_vars = {"action": action} if output_variables: out_vars.update(output_variables) parameters["outputVariables"] = out_vars - + if info: parameters["info"] = info - + msg = self._create_message( AudioHookMessageType.DISCONNECT, parameters=parameters, ) - + logger.debug(f"AudioHook disconnect: reason={reason}, action={action}") return msg @@ -532,12 +530,12 @@ class GenesysAudioHookSerializer(FrameSerializer): retryable: bool = False, ) -> Dict[str, Any]: """Create an 'error' message. - + Args: code: Error code. message: Error message. retryable: Whether the operation can be retried. - + Returns: Dictionary of the error message. """ @@ -546,12 +544,12 @@ class GenesysAudioHookSerializer(FrameSerializer): "message": message, "retryable": retryable, } - + msg = self._create_message( AudioHookMessageType.ERROR, parameters=parameters, ) - + logger.error(f"AudioHook error: {code} - {message}") return msg @@ -572,14 +570,18 @@ class GenesysAudioHookSerializer(FrameSerializer): the frame type is not handled or session is not open. """ if isinstance(frame, (EndFrame, CancelFrame)): - return json.dumps(self.create_disconnect_message(output_variables=self.output_variables, reason="completed")) - + return json.dumps( + self.create_disconnect_message( + output_variables=self.output_variables, reason="completed" + ) + ) + elif isinstance(frame, AudioRawFrame): if not self._is_open or self._is_paused: return None - + data = frame.audio - + # Convert PCM to μ-law at 8kHz for Genesys if self._params.media_format == AudioHookMediaFormat.PCMU: serialized_data = await pcm_to_ulaw( @@ -592,15 +594,15 @@ class GenesysAudioHookSerializer(FrameSerializer): # L16 format - just resample if needed logger.warning("L16 format not yet fully implemented") return None - + if serialized_data is None or len(serialized_data) == 0: return None - + return bytes(serialized_data) elif isinstance(frame, InterruptionFrame): return json.dumps(self.create_barge_in_event()) - + elif isinstance(frame, (OutputTransportMessageFrame, OutputTransportMessageUrgentFrame)): # Only pass through AudioHook protocol messages (those with "version" field) # Filter out RTVI and other non-AudioHook messages @@ -609,7 +611,7 @@ class GenesysAudioHookSerializer(FrameSerializer): else: # Not an AudioHook message, ignore return None - + # Ignore other frames - we don't need to process them here return None @@ -639,39 +641,41 @@ class GenesysAudioHookSerializer(FrameSerializer): if isinstance(data, bytes): logger.debug(f"[AUDIO IN] Received {len(data)} bytes from Genesys") return await self._deserialize_audio(data) - + # Text data = JSON control message try: message = json.loads(data) except json.JSONDecodeError as e: logger.error(f"Failed to parse AudioHook message: {e}") return None - + return await self._handle_control_message(message) async def _deserialize_audio(self, data: bytes) -> Frame | None: """Deserialize binary audio data to an InputAudioRawFrame. - + Args: data: Raw audio bytes (PCMU or L16). - + Returns: InputAudioRawFrame with PCM audio at pipeline sample rate. """ if not self._is_open or self._is_paused: return None - + audio_data = data original_len = len(data) - + # If Genesys sends stereo audio (BOTH channels), extract only the external channel (left) # Stereo audio comes interleaved: [L0, R0, L1, R1, ...] if self._params.channel == AudioHookChannel.BOTH and len(data) > 0: # For PCMU, each sample is 1 byte # Extract only bytes at even positions (left channel = external) audio_data = bytes(data[i] for i in range(0, len(data), 2)) - logger.debug(f"🔊 Stereo audio: {original_len} bytes → {len(audio_data)} bytes (external channel)") - + logger.debug( + f"🔊 Stereo audio: {original_len} bytes → {len(audio_data)} bytes (external channel)" + ) + if self._params.media_format == AudioHookMediaFormat.PCMU: # Convert μ-law at 8kHz to PCM at pipeline rate deserialized_data = await ulaw_to_pcm( @@ -684,63 +688,62 @@ class GenesysAudioHookSerializer(FrameSerializer): # L16 format logger.warning("L16 format not yet fully implemented") return None - + if deserialized_data is None or len(deserialized_data) == 0: return None - + # Always use mono for STT - ElevenLabs expects single channel num_channels = 1 - + audio_frame = InputAudioRawFrame( audio=deserialized_data, num_channels=num_channels, sample_rate=self._sample_rate, ) - + return audio_frame - async def _handle_control_message(self, message: Dict[str, Any]) -> Frame | None: """Handle a JSON control message from Genesys. - + Args: message: Parsed JSON message. - + Returns: Frame if the message should be passed to the pipeline, None otherwise. """ msg_type = message.get("type", "") self._client_seq = message.get("seq", 0) - + # Update position if provided if "position" in message: self._position = self._parse_position(message["position"]) - + if msg_type == AudioHookMessageType.OPEN.value: return await self._handle_open(message) - + elif msg_type == AudioHookMessageType.CLOSE.value: return await self._handle_close(message) - + elif msg_type == AudioHookMessageType.PING.value: return await self._handle_ping(message) - + elif msg_type == AudioHookMessageType.PAUSE.value: return await self._handle_pause(message) - + elif msg_type == AudioHookMessageType.UPDATE.value: return await self._handle_update(message) elif msg_type == AudioHookMessageType.ERROR.value: return await self._handle_error(message) - + elif msg_type == "dtmf": return await self._handle_dtmf(message) - + elif msg_type == "playback_started": logger.debug("Playback started (from Genesys)") return None - + elif msg_type == "playback_completed": logger.debug("Playback completed (from Genesys)") return None @@ -750,40 +753,42 @@ class GenesysAudioHookSerializer(FrameSerializer): async def _handle_open(self, message: Dict[str, Any]) -> Frame | None: """Handle an 'open' message from Genesys. - + This initializes the session with metadata from Genesys Cloud and automatically responds with an 'opened' message using the configured InputParams (supported_languages, selected_language, start_paused). - + Extracts and stores: - session_id: The AudioHook session identifier - conversation_id: The Genesys conversation ID - participant: Caller info (ani, dnis, etc.) - input_variables: Custom variables from Architect flow - media_info: Audio configuration from Genesys - + Args: message: The open message from Genesys. - + Returns: OutputTransportMessageUrgentFrame with the 'opened' response. """ self._session_id = message.get("id", str(uuid.uuid4())) - + params = message.get("parameters", {}) self._conversation_id = params.get("conversationId") self._participant = params.get("participant") self._custom_config = params.get("customConfig") self._media_info = params.get("media") # This is a list of media objects self._input_variables = params.get("inputVariables") # Custom vars from Genesys - + # Extract media configuration if present # media is a list like: [{"type": "audio", "format": "PCMU", "channels": ["external"], "rate": 8000}] media_list = self._media_info if media_list and isinstance(media_list, list) and len(media_list) > 0: audio_media: Dict[str, Any] = media_list[0] # Get first media entry channels = audio_media.get("channels", []) - logger.debug(f"📡 Genesys audio config: format={audio_media.get('format')}, channels={channels}, rate={audio_media.get('rate')}") + logger.debug( + f"📡 Genesys audio config: format={audio_media.get('format')}, channels={channels}, rate={audio_media.get('rate')}" + ) # channels is a list like ["external"] or ["external", "internal"] if isinstance(channels, list): if "external" in channels and "internal" in channels: @@ -795,47 +800,49 @@ class GenesysAudioHookSerializer(FrameSerializer): elif "internal" in channels: self._params.channel = AudioHookChannel.INTERNAL logger.debug("📡 Mono mode: internal channel") - + # Log participant info for debugging ani = self._participant.get("ani", "unknown") if self._participant else "unknown" logger.info( f"AudioHook open request: session={self._session_id}, " f"conversation={self._conversation_id}, ani={ani}" ) - + await self._call_event_handler("on_open", message) - return OutputTransportMessageUrgentFrame(message=self.create_opened_response( - start_paused=self._params.start_paused, - supported_languages=self._params.supported_languages, - selected_language=self._params.selected_language - )) + return OutputTransportMessageUrgentFrame( + message=self.create_opened_response( + start_paused=self._params.start_paused, + supported_languages=self._params.supported_languages, + selected_language=self._params.selected_language, + ) + ) async def _handle_close(self, message: Dict[str, Any]) -> Frame | None: """Handle a 'close' message from Genesys. - + Automatically responds with a 'closed' message. If output_variables were set via set_output_variables(), they will be included in the response and made available in the Architect flow. - + Args: message: The close message from Genesys. - + Returns: OutputTransportMessageUrgentFrame with the closed response (includes outputVariables if set). """ params = message.get("parameters", {}) reason = params.get("reason", "unknown") - + logger.info(f"🔴 Genesys closed the connection: {reason}") - + self._is_open = False logger.info(f"Sending closed response to Genesys...") await self._call_event_handler("on_close", message) - + # Return as urgent frame to be sent through pipeline immediately # Include any output variables that were set during the session return OutputTransportMessageUrgentFrame( @@ -844,12 +851,12 @@ class GenesysAudioHookSerializer(FrameSerializer): async def _handle_ping(self, message: Dict[str, Any]) -> Frame | None: """Handle a 'ping' message from Genesys. - + Automatically responds with a 'pong' message to maintain the connection. - + Args: message: The ping message from Genesys. - + Returns: OutputTransportMessageUrgentFrame with pong response. """ @@ -862,92 +869,92 @@ class GenesysAudioHookSerializer(FrameSerializer): async def _handle_pause(self, message: Dict[str, Any]) -> Frame | None: """Handle a 'pause' message from Genesys. - + This is used when audio streaming is temporarily suspended (e.g., during hold). - + Args: message: The pause message. - + Returns: None (response should be sent via create_resumed_response()). """ params = message.get("parameters", {}) reason = params.get("reason", "unknown") - + logger.info(f"AudioHook pause request: reason={reason}") - + self._is_paused = True await self._call_event_handler("on_pause", message) - + # Note: Application should call create_resumed_response() when ready return None async def _handle_update(self, message: Dict[str, Any]) -> Frame | None: """Handle an 'update' message from Genesys. - + Updates may include changes to participants or configuration. - + Args: message: The update message. - + Returns: None. """ params = message.get("parameters", {}) - + if "participant" in params: self._participant = params["participant"] - + logger.debug(f"AudioHook update received: {params}") await self._call_event_handler("on_update", message) - + return None async def _handle_error(self, message: Dict[str, Any]) -> Frame | None: """Handle an 'error' message from Genesys. - + Args: message: The error message. - + Returns: None. """ params = message.get("parameters", {}) code = params.get("code", 0) error_msg = params.get("message", "Unknown error") - + logger.error(f"AudioHook error from Genesys: {code} - {error_msg}") await self._call_event_handler("on_error", message) - + return None async def _handle_dtmf(self, message: Dict[str, Any]) -> Frame | None: """Handle a 'dtmf' message from Genesys. - + DTMF (Dual-Tone Multi-Frequency) events are sent when the user presses keys on their phone keypad. - + Args: message: The DTMF message. - + Returns: InputDTMFFrame with the pressed digit. """ params = message.get("parameters", {}) digit = params.get("digit", "") - + if not digit: logger.warning("DTMF message received without digit") return None - + logger.info(f"DTMF received: {digit}") await self._call_event_handler("on_dtmf", message) - + try: return InputDTMFFrame(KeypadEntry(digit)) except ValueError: From 10fb77c0e21dec0c7951f4feda7dd7a0bde71a17 Mon Sep 17 00:00:00 2001 From: ssillerom Date: Wed, 28 Jan 2026 18:07:33 +0100 Subject: [PATCH 11/11] added changelog file --- changelog/3500.added.md | 1 + 1 file changed, 1 insertion(+) create mode 100644 changelog/3500.added.md diff --git a/changelog/3500.added.md b/changelog/3500.added.md new file mode 100644 index 000000000..64881e91b --- /dev/null +++ b/changelog/3500.added.md @@ -0,0 +1 @@ +- Added new `GenesysFrameSerializer` for the Genesys AudioHook WebSocket protocol, enabling bidirectional audio streaming between Pipecat pipelines and Genesys Cloud contact center. \ No newline at end of file