From ba312d0dfaabbe166595f0a5393106abb7ac448f Mon Sep 17 00:00:00 2001 From: Nir Simionovich Date: Wed, 15 Jul 2026 12:21:40 +0300 Subject: [PATCH] Cloudonix Transfer Feature (#542) * Add support for foreground debugging * Add support for Cloudonix call transfers * Improve the customer/agent conference experience with less annoying sounds. * Update remote_up.sh Co-authored-by: cubic-dev-ai[bot] <191113872+cubic-dev-ai[bot]@users.noreply.github.com> * Resolve a small redundant code segment from cubic * Resolve an issue with callbacks not providing the correct experience for failed originated calls * Yet a small fix * Remove stale code * Remove the beeps on transfer * Remove unrelated remote_up.sh changes * Update pipecat submodule to main --------- Co-authored-by: Nir Simionovich Co-authored-by: cubic-dev-ai[bot] <191113872+cubic-dev-ai[bot]@users.noreply.github.com> Co-authored-by: Abhishek Kumar --- .../telephony/providers/cloudonix/provider.py | 111 +++++++++++++---- .../telephony/providers/cloudonix/routes.py | 91 ++++++++++++++ .../providers/cloudonix/strategies.py | 117 +++++++++++++++++- .../providers/cloudonix/transport.py | 3 +- pipecat | 2 +- 5 files changed, 298 insertions(+), 26 deletions(-) diff --git a/api/services/telephony/providers/cloudonix/provider.py b/api/services/telephony/providers/cloudonix/provider.py index d3e97466..9051f506 100644 --- a/api/services/telephony/providers/cloudonix/provider.py +++ b/api/services/telephony/providers/cloudonix/provider.py @@ -1051,6 +1051,18 @@ class CloudonixProvider(TelephonyProvider): return Response(content=twiml, media_type="application/xml"), "application/xml" # ======== CALL TRANSFER METHODS ======== + @staticmethod + def _conference_join_cxml(conference_name: str, callback_url: str) -> str: + """CXML the destination leg runs once it answers: join the conference.""" + return ( + '' + "" + "You have answered a transfer call. Connecting you now." + "" + f'{conference_name}' + "" + "" + ) async def transfer_call( self, @@ -1060,33 +1072,86 @@ class CloudonixProvider(TelephonyProvider): timeout: int = 30, **kwargs: Any, ) -> Dict[str, Any]: + """Dial the transfer destination into a conference via Cloudonix. + + Places an outbound call whose inline CXML joins ``conference_name`` when + the destination answers, and sets the call object's ``callback`` to the + Cloudonix transfer-result route so the destination's session-status + transitions (``connected`` / terminal) drive transfer completion. The + original caller leg is later forked into the same conference by + ``CloudonixConferenceStrategy``. + + Supports both PSTN numbers and SIP URIs as ``destination``. + + Returns a dict with the destination leg's ``call_sid`` (session token). """ - Initiate a call transfer via Cloudonix. + if not self.validate_config(): + raise ValueError("Cloudonix provider not properly configured") - Uses inline CXML to put the destination into a conference when they answer, - and a status callback to track the transfer outcome. + if not self.from_numbers: + raise ValueError( + "No phone numbers configured for Cloudonix provider; a caller-id " + "is required to place the transfer call." + ) + from_number = random.choice(self.from_numbers) - Args: - destination: The destination phone number (E.164 format) - transfer_id: Unique identifier for tracking this transfer - conference_name: Name of the conference to join the destination into - timeout: Transfer timeout in seconds - **kwargs: Additional Twilio-specific parameters + backend_endpoint, _ = await get_backend_endpoints() + callback_url = ( + f"{backend_endpoint}/api/v1/telephony/cloudonix/transfer-result/{transfer_id}" + ) - Returns: - Dict containing transfer result information + endpoint = f"{self.base_url}/calls/{self.domain_id}/application" + data: Dict[str, Any] = { + "destination": destination, + "caller-id": from_number, + "cxml": self._conference_join_cxml(conference_name, callback_url), + "callback": callback_url, + "timeout": timeout, + } - Raises: - ValueError: If provider configuration is invalid - Exception: If Twilio API call fails - """ - raise NotImplementedError("Cloudonix provider does not support call transfers") + data.update(kwargs) + headers = self._get_auth_headers() + masked_destination = ( + f"***{destination[-4:]}" if len(destination) > 4 else "***" + ) + logger.info( + f"[Cloudonix Transfer] Dialing {masked_destination} into conference " + f"{conference_name} (transfer_id={transfer_id})" + ) + + async with aiohttp.ClientSession() as session: + async with session.post(endpoint, json=data, headers=headers) as response: + response_text = await response.text() + if response.status != 200: + logger.error( + f"[Cloudonix Transfer] Dial failed: HTTP {response.status}, " + f"body: {response_text}" + ) + raise Exception( + f"Cloudonix transfer dial failed (HTTP {response.status}): " + f"{response_text}" + ) + + response_data = await response.json() + session_token = response_data.get("token") + if not session_token: + raise Exception( + "No session token returned from Cloudonix transfer dial" + ) + + logger.info( + f"[Cloudonix Transfer] Destination leg initiated " + f"(token={session_token}, transfer_id={transfer_id})" + ) + return { + "call_sid": session_token, + "status": response_data.get("status", "initiated"), + "provider": self.PROVIDER_NAME, + "from_number": from_number, + "to_number": destination, + "raw_response": response_data, + } def supports_transfers(self) -> bool: - """ - Cloudonix does not support call transfers. - - Returns: - False - Cloudonix provider does not support call transfers - """ - return False + """Cloudonix supports conference-based call transfers.""" + return True diff --git a/api/services/telephony/providers/cloudonix/routes.py b/api/services/telephony/providers/cloudonix/routes.py index 80e7037a..7a6852f1 100644 --- a/api/services/telephony/providers/cloudonix/routes.py +++ b/api/services/telephony/providers/cloudonix/routes.py @@ -11,15 +11,106 @@ from loguru import logger from pipecat.utils.run_context import set_current_run_id from api.db import db_client +from api.services.telephony.call_transfer_manager import get_call_transfer_manager from api.services.telephony.factory import get_telephony_provider_for_run from api.services.telephony.providers.cloudonix.provider import CloudonixProvider from api.services.telephony.status_processor import ( StatusCallbackRequest, _process_status_update, ) +from api.services.telephony.transfer_event_protocol import ( + TransferEvent, + TransferEventType, +) router = APIRouter() +# Cloudonix session statuses that terminate a transfer without an answer. +_CLOUDONIX_TRANSFER_FAILURE_STATUSES = { + "busy", + "noanswer", + "cancel", + "nocredit", + "error", + "congestion", + "failed", +} + + +@router.post("/cloudonix/transfer-result/{transfer_id}") +async def handle_cloudonix_transfer_result(transfer_id: str, request: Request): + """Drive transfer completion from the destination leg's session status. + + ``CloudonixProvider.transfer_call`` sets this URL as the outbound call + object's ``callback``. Cloudonix POSTs session-status notifications here; + a ``connected`` status means the destination answered (publish + DESTINATION_ANSWERED so the shared handler forks the caller into the + conference), while terminal non-answer statuses publish TRANSFER_FAILED. + Intermediate statuses (ringing/processing) are acked without publishing. + """ + content_type = request.headers.get("content-type", "") + if "application/json" in content_type: + data = await request.json() + else: + data = dict(await request.form()) + + # Cloudonix session notifications may arrive as a single object or a + # one-element list (see session-update webhook payloads). + if isinstance(data, list): + data = data[0] if data else {} + + status = str(data.get("StatusCallbackEvent", "")).lower() + destination_token = data.get("Session", "") + + logger.info( + f"[Cloudonix Transfer] transfer_id={transfer_id} status={status} " + f"token={destination_token}" + ) + + call_transfer_manager = await get_call_transfer_manager() + transfer_context = await call_transfer_manager.get_transfer_context(transfer_id) + if not transfer_context: + logger.warning( + f"[Cloudonix Transfer] No transfer context for {transfer_id}; ignoring" + ) + return {"status": "ignored", "reason": "unknown_transfer"} + + original_call_sid = transfer_context.original_call_sid + conference_name = transfer_context.conference_name + + if (status == "participant-join"): + event = TransferEvent( + type=TransferEventType.DESTINATION_ANSWERED, + transfer_id=transfer_id, + original_call_sid=original_call_sid or "", + transfer_call_sid=destination_token, + conference_name=conference_name, + message="Great! The destination answered. Connecting you now.", + status="success", + action="destination_answered", + ) + elif status in _CLOUDONIX_TRANSFER_FAILURE_STATUSES: + event = TransferEvent( + type=TransferEventType.TRANSFER_FAILED, + transfer_id=transfer_id, + original_call_sid=original_call_sid or "", + transfer_call_sid=destination_token, + conference_name=conference_name, + message="The transfer call could not be completed.", + status="transfer_failed", + action="transfer_failed", + reason=status, + ) + else: + logger.info( + f"[Cloudonix Transfer] Intermediate status {status} for {transfer_id}, " + "waiting" + ) + return {"status": "pending"} + + await call_transfer_manager.publish_transfer_event(event) + return {"status": "completed"} + @router.post("/cloudonix/status-callback/{workflow_run_id}") async def handle_cloudonix_status_callback( diff --git a/api/services/telephony/providers/cloudonix/strategies.py b/api/services/telephony/providers/cloudonix/strategies.py index b64cf6da..32ae389e 100644 --- a/api/services/telephony/providers/cloudonix/strategies.py +++ b/api/services/telephony/providers/cloudonix/strategies.py @@ -3,11 +3,126 @@ from typing import Any, Dict from loguru import logger -from pipecat.serializers.call_strategies import HangupStrategy +from pipecat.serializers.call_strategies import HangupStrategy, TransferStrategy from api.services.telephony.providers.cloudonix.provider import CLOUDONIX_API_BASE_URL +class CloudonixConferenceStrategy(TransferStrategy): + """Conference-based call transfer for Cloudonix. + + Moves the original caller leg into the transfer conference by forking its + live session onto new CXML. Cloudonix has no live-CXML push equivalent to + Twilio's call-update; ``POST /calls/{domain}/sessions/{token}/fork`` is the + primitive that re-runs CXML on a connected session. The destination leg was + already dialed into the conference by ``CloudonixProvider.transfer_call``. + + The fork MUST target the Cloudonix session token, which is carried on + ``TransferContext.original_call_sid`` (the media ``callSid`` will not + resolve the session). + """ + + async def execute_transfer(self, context: Dict[str, Any]) -> bool: + import aiohttp + + transfer_context = None + try: + # call_sid here is the serializer's session token (remapped from + # Cloudonix call_id); use it only to locate the transfer context. + call_sid = context.get("call_sid") or context.get("call_id") + domain_id = context.get("account_sid") or context.get("domain_id") + bearer_token = context.get("auth_token") or context.get("bearer_token") + + transfer_context = await self._find_transfer_context_for_call(call_sid) + if not transfer_context: + logger.error( + f"[Cloudonix Transfer] No active transfer context for call {call_sid}" + ) + return False + + if not domain_id or not bearer_token: + logger.error( + "[Cloudonix Transfer] Missing domain_id or bearer_token in context" + ) + await self._cleanup_transfer_context(transfer_context.transfer_id) + return False + + # Always fork the session token, never the media callSid. + session_token = transfer_context.original_call_sid + conference_name = transfer_context.conference_name + + endpoint = ( + f"{CLOUDONIX_API_BASE_URL}/calls/{domain_id}/sessions/" + f"{session_token}/application" + ) + caller_cxml = ( + '' + "" + f'{conference_name}' + "" + ) + payload = {"cxml": caller_cxml} + headers = { + "Authorization": f"Bearer {bearer_token}", + "Content-Type": "application/json", + } + + logger.info( + f"[Cloudonix Transfer] Switching session {session_token} into " + f"conference {conference_name}" + ) + + async with aiohttp.ClientSession() as session: + async with session.post( + endpoint, json=payload, headers=headers + ) as response: + body = await response.text() + if response.status in (200, 202): + logger.info( + f"[Cloudonix Transfer] Session {session_token} joined " + f"conference {conference_name} (HTTP {response.status})" + ) + await self._cleanup_transfer_context( + transfer_context.transfer_id + ) + return True + logger.error( + f"[Cloudonix Transfer] Switch Voice Application failed for session " + f"{session_token}: HTTP {response.status}, body: {body}" + ) + await self._cleanup_transfer_context(transfer_context.transfer_id) + return False + + except Exception as e: + logger.error(f"[Cloudonix Transfer] Failed to transfer call: {e}") + if transfer_context: + await self._cleanup_transfer_context(transfer_context.transfer_id) + return False + + async def _find_transfer_context_for_call(self, call_sid: str): + try: + from api.services.telephony.call_transfer_manager import ( + get_call_transfer_manager, + ) + + manager = await get_call_transfer_manager() + return await manager.find_transfer_context_for_call(call_sid) + except Exception as e: + logger.error(f"[Cloudonix Transfer] Error finding transfer context: {e}") + return None + + async def _cleanup_transfer_context(self, transfer_id: str): + try: + from api.services.telephony.call_transfer_manager import ( + get_call_transfer_manager, + ) + + manager = await get_call_transfer_manager() + await manager.remove_transfer_context(transfer_id) + except Exception as e: + logger.error(f"[Cloudonix Transfer] Error cleaning up transfer context: {e}") + + class CloudonixHangupStrategy(HangupStrategy): """Implements hangup for Cloudonix calls.""" diff --git a/api/services/telephony/providers/cloudonix/transport.py b/api/services/telephony/providers/cloudonix/transport.py index c38c9753..6c2729fd 100644 --- a/api/services/telephony/providers/cloudonix/transport.py +++ b/api/services/telephony/providers/cloudonix/transport.py @@ -12,7 +12,7 @@ from api.services.pipecat.transport_params import realtime_param_overrides from api.services.telephony.factory import load_credentials_for_transport from .serializers import CloudonixFrameSerializer -from .strategies import CloudonixHangupStrategy +from .strategies import CloudonixConferenceStrategy, CloudonixHangupStrategy async def create_transport( @@ -55,6 +55,7 @@ async def create_transport( domain_id=domain_id, bearer_token=bearer_token, hangup_strategy=CloudonixHangupStrategy(), + transfer_strategy=CloudonixConferenceStrategy(), ) mixer = await build_audio_out_mixer( diff --git a/pipecat b/pipecat index 695f9322..f4feab0c 160000 --- a/pipecat +++ b/pipecat @@ -1 +1 @@ -Subproject commit 695f9322f870484d8507a63a7343d23725f2fb3a +Subproject commit f4feab0c59129702b3a6e6f818f62229cc663f1a