""" Cloudonix implementation of the TelephonyProvider interface. """ import json import random from typing import TYPE_CHECKING, Any, Dict, List, Optional import aiohttp from fastapi import HTTPException from loguru import logger from api.db import db_client from api.enums import WorkflowRunMode from api.services.telephony.base import ( CallInitiationResult, NormalizedInboundData, ProviderSyncResult, TelephonyProvider, ) from api.utils.common import get_backend_endpoints if TYPE_CHECKING: from fastapi import WebSocket CLOUDONIX_API_BASE_URL = "https://api.cloudonix.io" class CloudonixProvider(TelephonyProvider): """ Cloudonix implementation of TelephonyProvider. Uses Bearer token authentication and is TwiML-compatible for WebSocket audio. """ PROVIDER_NAME = WorkflowRunMode.CLOUDONIX.value WEBHOOK_ENDPOINT = "twiml" # Cloudonix is TwiML-compatible def __init__(self, config: Dict[str, Any]): """ Initialize CloudonixProvider with configuration. Args: config: Dictionary containing: - bearer_token: Cloudonix API Bearer Token - domain_id: Cloudonix Domain ID - application_name: Cloudonix Voice Application name whose url is updated by ``configure_inbound`` - from_numbers: List of phone numbers to use (optional, fetched from API if not provided) """ self.bearer_token = config.get("bearer_token") self.domain_id = self._normalize_domain(config.get("domain_id")) self.application_name = config.get("application_name") self.from_numbers = config.get("from_numbers", []) # Handle both single number (string) and multiple numbers (list) if isinstance(self.from_numbers, str): self.from_numbers = [self.from_numbers] self.base_url = CLOUDONIX_API_BASE_URL @staticmethod def _normalize_domain(domain: Optional[str]) -> Optional[str]: """Ensure a Cloudonix domain is fully qualified. Cloudonix domains are always of the form ``.cloudonix.net``. Users sometimes configure or pass just ````; normalize so equality checks against stored credentials and API URLs work regardless of input form. """ if not domain: return domain domain = domain.strip() if not domain: return domain if domain.endswith(".cloudonix.net"): return domain return f"{domain}.cloudonix.net" def _get_auth_headers(self) -> Dict[str, str]: """Generate authorization headers for Cloudonix API.""" return { "Authorization": f"Bearer {self.bearer_token}", "Content-Type": "application/json", } async def initiate_call( self, to_number: str, webhook_url: str, workflow_run_id: Optional[int] = None, from_number: Optional[str] = None, **kwargs: Any, ) -> CallInitiationResult: """ Initiate an outbound call via Cloudonix. Note: webhook_url parameter is ignored for Cloudonix. Unlike Twilio/Vonage, Cloudonix embeds CXML directly in the API call rather than using webhook callbacks. """ if not self.validate_config(): raise ValueError("Cloudonix provider not properly configured") endpoint = f"{self.base_url}/calls/{self.domain_id}/application" # Use provided from_number or select a random one (REQUIRED by Cloudonix) if from_number is None: if not self.from_numbers: raise ValueError( "No phone numbers configured for Cloudonix provider. " "At least one phone number is required as 'caller-id' for outbound calls. " "Please configure phone numbers in the telephony settings." ) from_number = random.choice(self.from_numbers) logger.info( f"Selected phone number {from_number} for outbound call to {to_number}" ) workflow_id, user_id = kwargs["workflow_id"], kwargs["user_id"] # Prepare call data using Cloudonix callObject schema # Note: 'caller-id' is REQUIRED by Cloudonix API backend_endpoint, wss_backend_endpoint = await get_backend_endpoints() data: Dict[str, Any] = { "destination": to_number, "cxml": f""" """, "caller-id": from_number, # Required field } # TODO: Cloudonix status callbacks are spammy, so commenting it out. Can send it to # some persistent logging system instead of transcational database. # Add status callback if workflow_run_id provided # if workflow_run_id: # callback_url = f"{backend_endpoint}/api/v1/telephony/cloudonix/status-callback/{workflow_run_id}" # data["callback"] = callback_url # Merge any additional kwargs data.update(kwargs) # Make the API request headers = self._get_auth_headers() # Log request details (mask sensitive token) masked_headers = { k: v if k != "Authorization" else f"Bearer {self.bearer_token[:8]}..." for k, v in headers.items() } logger.info( f"[Cloudonix] Initiating outbound call:\n" f" Endpoint: {endpoint}\n" f" To: {to_number}\n" f" From: {from_number}\n" f" Workflow Run ID: {workflow_run_id}" ) logger.debug( f"[Cloudonix] Request details:\n" f" Headers: {masked_headers}\n" f" Payload: {json.dumps(data, indent=2)}" ) async with aiohttp.ClientSession() as session: async with session.post(endpoint, json=data, headers=headers) as response: response_text = await response.text() response_status = response.status # Log response logger.info( f"[Cloudonix] API Response:\n" f" HTTP Status: {response_status}\n" f" Response Body: {response_text}" ) if response_status != 200: logger.error( f"[Cloudonix] Call initiation FAILED:\n" f" HTTP Status: {response_status}\n" f" Error Details: {response_text}\n" f" Request: POST {endpoint}\n" f" Payload: {json.dumps(data, indent=2)}" ) raise HTTPException( status_code=response_status, detail=f"Failed to initiate call via Cloudonix (HTTP {response_status}): {response_text}", ) response_data = await response.json() # Extract session token (call ID) and other metadata session_token = response_data.get("token") domain_id = response_data.get("domainId") subscriber_id = response_data.get("subscriberId") if not session_token: logger.error( f"[Cloudonix] Missing session token in response:\n" f" Response: {json.dumps(response_data, indent=2)}" ) raise Exception("No session token returned from Cloudonix") logger.info( f"[Cloudonix] Call initiated successfully:\n" f" Session Token: {session_token}\n" f" Domain ID: {domain_id}\n" f" Subscriber ID: {subscriber_id}\n" f" To: {to_number}\n" f" From: {from_number}\n" f" Workflow Run ID: {workflow_run_id}" ) return CallInitiationResult( call_id=session_token, status="initiated", caller_number=from_number, provider_metadata={ "call_id": session_token, "domain_id": domain_id, "subscriber_id": subscriber_id, }, raw_response=response_data, ) async def get_call_status(self, call_id: str) -> Dict[str, Any]: """ Get the current status of a Cloudonix call (session). Args: call_id: The session token returned from call initiation """ if not self.validate_config(): raise ValueError("Cloudonix provider not properly configured") endpoint = ( f"{self.base_url}/customers/self/domains/" f"{self.domain_id}/sessions/{call_id}" ) headers = self._get_auth_headers() async with aiohttp.ClientSession() as session: async with session.get(endpoint, headers=headers) as response: if response.status != 200: error_data = await response.text() logger.error(f"Failed to get call status: {error_data}") raise Exception(f"Failed to get call status: {error_data}") return await response.json() async def get_available_phone_numbers(self) -> List[str]: """ Get list of available Cloudonix phone numbers (DNIDs). """ # If phone numbers are already configured, return them if self.from_numbers: return self.from_numbers # Otherwise, fetch from API if not self.validate_config(): raise ValueError("Cloudonix provider not properly configured") endpoint = f"{self.base_url}/customers/self/domains/{self.domain_id}/dnids" headers = self._get_auth_headers() try: async with aiohttp.ClientSession() as session: async with session.get(endpoint, headers=headers) as response: if response.status != 200: logger.warning( f"Failed to fetch DNIDs from Cloudonix: {response.status}" ) return [] dnids = await response.json() # Extract phone numbers from DNID objects # Use "source" field which contains the original phone number phone_numbers = [ dnid.get("source") or dnid.get("dnid") for dnid in dnids if dnid.get("source") or dnid.get("dnid") ] # Cache the fetched numbers self.from_numbers = phone_numbers return phone_numbers except Exception as e: logger.error(f"Exception fetching Cloudonix DNIDs: {e}") return [] def validate_config(self) -> bool: """ Validate Cloudonix configuration. """ return bool(self.bearer_token and self.domain_id) async def verify_webhook_signature( self, url: str, params: Dict[str, Any], signature: str ) -> bool: """ Dummy implementation - Cloudonix doesn't use webhook signature verification. Cloudonix embeds CXML directly in the API call during initiate_call(), so webhook endpoints are never called and signature verification is not needed. This method only exists to satisfy the abstract base class requirement. Always returns True since no actual webhook verification is performed. """ logger.warning( "verify_webhook_signature called for Cloudonix - this should not happen. " "Cloudonix embeds CXML directly in API calls and doesn't use webhook callbacks." ) return True async def get_call_cost(self, call_id: str) -> Dict[str, Any]: """ Get cost information for a completed Cloudonix call. Note: Cloudonix does not currently support call cost retrieval via API. This method returns zero cost. Args: call_id: The Cloudonix session token Returns: Dict containing cost information (all zeros for now) """ logger.info( f"Cloudonix does not support call cost retrieval - returning zero cost for call {call_id}" ) return { "cost_usd": 0.0, "duration": 0, "status": "unknown", "error": "Cloudonix does not support cost retrieval", } def parse_status_callback(self, data: Dict[str, Any]) -> Dict[str, Any]: """ Parse Cloudonix status callback data into generic format. Note: The exact format of Cloudonix status callbacks needs to be confirmed. This implementation assumes a similar structure to Twilio. """ # Map Cloudonix status values to common format # These mappings may need adjustment based on actual Cloudonix callback format status_map = { "initiated": "initiated", "ringing": "ringing", "answered": "answered", "completed": "completed", "failed": "failed", "busy": "busy", "no-answer": "no-answer", "canceled": "canceled", "error": "error", } call_status = data.get("status", "") mapped_status = status_map.get(call_status.lower(), call_status) return { "call_id": data.get("token") or data.get("session_id") or data.get("CallSid", ""), "status": mapped_status, "from_number": data.get("caller_id") or data.get("From"), "to_number": data.get("destination") or data.get("To"), "direction": data.get("direction"), "duration": data.get("duration") or data.get("CallDuration"), "extra": data, # Include all original data } async def get_webhook_response( self, workflow_id: int, user_id: int, workflow_run_id: int ) -> str: """ Dummy implementation - Cloudonix doesn't use webhook responses. Cloudonix embeds CXML directly in the API call during initiate_call(), so this webhook endpoint is never actually called. This method only exists to satisfy the abstract base class requirement. """ logger.warning( "get_webhook_response called for Cloudonix - this should not happen. " "Cloudonix embeds CXML directly in API calls." ) return """ Error: This endpoint should not be called for Cloudonix """ async def handle_websocket( self, websocket: "WebSocket", workflow_id: int, user_id: int, workflow_run_id: int, ) -> None: """ Handle Cloudonix-specific WebSocket connection. Cloudonix WebSocket is compatible with Twilio, so we use the same handler. Cloudonix sends: 1. "connected" event first 2. "start" event with streamSid and callSid 3. Then audio messages """ from api.services.pipecat.run_pipeline import run_pipeline_telephony try: # Wait for "connected" event first_msg = await websocket.receive_text() msg = json.loads(first_msg) logger.debug(f"Received first message: {msg}") if msg.get("event") != "connected": logger.error(f"Expected 'connected' event, got: {msg.get('event')}") await websocket.close(code=4400, reason="Expected connected event") return # Wait for "start" event with stream details start_msg = await websocket.receive_text() logger.debug(f"Received start message: {start_msg}") start_msg = json.loads(start_msg) if start_msg.get("event") != "start": logger.error("Expected 'start' event second") await websocket.close(code=4400, reason="Expected start event") return # Extract Twilio-compatible identifiers try: stream_sid = start_msg["start"]["streamSid"] call_sid = start_msg["start"]["callSid"] except KeyError: logger.error("Missing streamSid or callSid in start message") await websocket.close(code=4400, reason="Missing stream identifiers") return logger.debug( f"Cloudonix WebSocket connected for workflow_run {workflow_run_id} " f"stream_sid: {stream_sid} call_sid: {call_sid}" ) workflow_run = await db_client.get_workflow_run_by_id(workflow_run_id) call_id = ( workflow_run.gathered_context.get("call_id") if workflow_run and workflow_run.gathered_context else None ) if not call_id: logger.error( f"call_id not found in gathered_context for workflow run {workflow_run_id}" ) await websocket.close(code=4400, reason="Missing call_id") return await run_pipeline_telephony( websocket, provider_name=self.PROVIDER_NAME, workflow_id=workflow_id, workflow_run_id=workflow_run_id, user_id=user_id, call_id=call_id, transport_kwargs={"call_id": call_id, "stream_sid": stream_sid}, ) except Exception as e: logger.error(f"Error in Cloudonix WebSocket handler: {e}") raise async def handle_external_websocket( self, websocket: "WebSocket", *, organization_id: int, workflow_id: int, user_id: int, workflow_run_id: int, params: Dict[str, str], ) -> None: """Agent-stream entry point. ``Domain`` (domain id) is read from the query string. The bearer token comes from the stored Cloudonix telephony configuration matched by ``domain_id`` within the workflow's organization — never from the URL or stream payload. The websocket handshake (connected / start) is identical to the standard inbound flow. Before starting the pipeline we (a) require an existing Cloudonix telephony configuration for the supplied ``domain_id`` and (b) validate the call session with Cloudonix using the bearer token from that configuration. Either failure closes the socket with 4400. """ from api.services.pipecat.run_pipeline import run_pipeline_telephony domain_id = self._normalize_domain(params.get("Domain")) if not domain_id: logger.error("Cloudonix agent-stream missing required param: Domain") await websocket.close(code=4400, reason="Missing Domain query param") return config = await self._find_config_by_domain(organization_id, domain_id) if not config: logger.error( f"Cloudonix agent-stream: no telephony configuration found " f"for domain_id={domain_id}" ) await websocket.close( code=4400, reason=f"Unknown Cloudonix domain: {domain_id}" ) return bearer_token = (config.credentials or {}).get("bearer_token") if not bearer_token: logger.error( f"Cloudonix agent-stream: telephony configuration {config.id} " f"is missing bearer_token in credentials" ) await websocket.close( code=4400, reason="Cloudonix configuration missing bearer_token" ) return try: first_msg = await websocket.receive_text() msg = json.loads(first_msg) if msg.get("event") != "connected": logger.error(f"Expected 'connected' event, got: {msg.get('event')}") await websocket.close(code=4400, reason="Expected connected event") return start_msg = json.loads(await websocket.receive_text()) if start_msg.get("event") != "start": logger.error("Expected 'start' event second") await websocket.close(code=4400, reason="Expected start event") return try: stream_sid = start_msg["start"]["streamSid"] call_sid = start_msg["start"]["callSid"] call_session = start_msg["start"]["session"] except KeyError: logger.error("Missing streamSid or callSid or session in start message") await websocket.close(code=4400, reason="Missing stream identifiers") return if not await self._validate_session(domain_id, call_session, bearer_token): await websocket.close( code=4400, reason="Cloudonix session validation failed" ) return logger.info( f"Cloudonix agent-stream connected for workflow_run " f"{workflow_run_id} stream_sid={stream_sid} call_sid={call_sid} session={call_session}" f"telephony_configuration_id={config.id}" ) await run_pipeline_telephony( websocket, provider_name=self.PROVIDER_NAME, workflow_id=workflow_id, workflow_run_id=workflow_run_id, user_id=user_id, call_id=call_session, transport_kwargs={ "call_id": call_session, "stream_sid": stream_sid, "bearer_token": bearer_token, "domain_id": domain_id, }, ) except Exception as e: logger.error(f"Error in Cloudonix agent-stream handler: {e}") raise async def _validate_session( self, domain_id: str, call_session: str, bearer_token: str ) -> bool: """Confirm the session is live with Cloudonix. Hits ``GET /customers/self/domains/{domain_id}/sessions/{call_session}`` with the supplied bearer token. A 200 response means both the token is valid and the session exists. """ endpoint = f"{self.base_url}/customers/self/domains/{domain_id}/sessions/{call_session}" headers = { "Authorization": f"Bearer {bearer_token}", "Content-Type": "application/json", } try: async with aiohttp.ClientSession() as http: async with http.get(endpoint, headers=headers) as response: if response.status == 200: return True body = await response.text() logger.error( f"Cloudonix session validation failed: " f"HTTP {response.status} domain_id={domain_id} " f"call_id={call_session} body={body}" ) return False except Exception as e: logger.error( f"Cloudonix session validation error for domain_id={domain_id} " f"call_id={call_session}: {e}" ) return False async def _find_config_by_domain(self, organization_id: int, domain_id: str): """Find a Cloudonix config by its normalized ``domain_id`` within ``organization_id`` — scoped lookup so credentials from a different org can never be used.""" normalized = self._normalize_domain(domain_id) if not normalized: return None candidates = await db_client.list_telephony_configurations_by_provider( organization_id, self.PROVIDER_NAME ) for cand in candidates: if (cand.credentials or {}).get("domain_id") == normalized: return cand return None # ======== INBOUND CALL METHODS ======== @classmethod def can_handle_webhook( cls, webhook_data: Dict[str, Any], headers: Dict[str, str] ) -> bool: """ Determine if this provider can handle the incoming webhook. """ # 1: Check User-Agent header user_agent = headers.get("user-agent", "").lower() if "cloudonix" in user_agent: return True # 2: Check for Cloudonix-specific headers cloudonix_headers = [ "x-cx-apikey", "x-cx-domain", "x-cx-session", "x-cx-source", ] if any(header in headers for header in cloudonix_headers): return True # 3: Check data structure for Cloudonix-specific fields if ( "SessionData" in webhook_data and "Domain" in webhook_data and webhook_data.get("Domain", "").endswith(".cloudonix.net") ): return True # Check if AccountSid is a Cloudonix domain account_sid = webhook_data.get("AccountSid", "") if account_sid.endswith(".cloudonix.net"): return True return False @staticmethod def parse_inbound_webhook(webhook_data: Dict[str, Any]) -> NormalizedInboundData: """ Parse Cloudonix-specific inbound webhook data into normalized format. Cloudonix webhook structure includes: - CallSid: Call id - From: Caller number - To: Called number - AccountSid: Domain (e.g., "abc.cloudonix.net") - SessionData: Contains additional call info including underlying provider details """ session_data = webhook_data.get("SessionData", {}) token = session_data.get("token", "") if isinstance(session_data, dict) else "" call_id = webhook_data.get("Session") or webhook_data.get("CallSid") or token account_id = CloudonixProvider._normalize_domain( webhook_data.get("Domain") or webhook_data.get("AccountSid", "") ) # Extract underlying provider information from SessionData if available session_data = webhook_data.get("SessionData", {}) underlying_provider = None if isinstance(session_data, dict): profile = session_data.get("profile", {}) trunk_headers = profile.get("trunk-sip-headers", {}) if "Twilio-AccountSid" in trunk_headers: underlying_provider = "twilio" direction = webhook_data.get("Direction", "inbound").lower() if direction in {"inbound", "subscriber"}: direction = "inbound" return NormalizedInboundData( provider=CloudonixProvider.PROVIDER_NAME, call_id=call_id, from_number=webhook_data.get("From", ""), to_number=webhook_data.get("To", ""), direction=direction, call_status=webhook_data.get("CallStatus", "in-progress"), account_id=account_id, from_country=webhook_data.get("FromCountry"), to_country=webhook_data.get("ToCountry"), raw_data={ **webhook_data, "underlying_provider": underlying_provider, }, ) @staticmethod def validate_account_id(config_data: dict, webhook_account_id: str) -> bool: """ Validate that the account_id from webhook matches the Cloudonix configuration. For Cloudonix: - webhook_account_id is the Domain field (e.g., "test1.cloudonix.net") - config domain_id stores the same domain string """ if not webhook_account_id: return False # Get stored domain from config (stored under 'domain_id' key) stored_domain = config_data.get("domain_id") if not stored_domain: return False return CloudonixProvider._normalize_domain( webhook_account_id ) == CloudonixProvider._normalize_domain(stored_domain) async def verify_inbound_signature( self, url: str, webhook_data: Dict[str, Any], headers: Dict[str, str], body: str = "", ) -> bool: """ Verify the API key of an inbound Cloudonix webhook for security. Cloudonix uses ``x-cx-apikey`` header validation instead of signature verification. The API key from the webhook should match the bearer_token in our configuration. """ api_key = headers.get("x-cx-apikey", "") if not api_key: logger.warning("No x-cx-apikey provided in Cloudonix webhook") return False # The bearer_token in config is the same as x-cx-apikey header value if not self.bearer_token: logger.warning("No bearer_token configured for Cloudonix provider") return False # Compare the API keys is_valid = api_key == self.bearer_token if is_valid: logger.info("Cloudonix x-cx-apikey validation successful") else: logger.warning( f"Cloudonix x-cx-apikey validation failed. Expected key ending with ...{self.bearer_token[-8:] if len(self.bearer_token) > 8 else 'SHORT_KEY'}" ) return True # TODO: update this post clarification from cloudonix async def configure_inbound( self, address: str, webhook_url: Optional[str] ) -> ProviderSyncResult: """Update the ``url`` on the Cloudonix Voice Application. PATCH is partial, so we send only ``url`` and ``method=POST`` (our ``/inbound/run`` is POST-only); ``type``, ``active``, and ``profile`` are preserved as configured in the cockpit. The URL is shared across every DNID on the application — clearing is a no-op to avoid silently breaking inbound for sibling numbers. """ if webhook_url is None: logger.info( f"Cloudonix configure_inbound clear for {address}: skipping " f"application update (url is shared across all DNIDs on Voice " f"Application {self.application_name})" ) return ProviderSyncResult(ok=True) if not self.validate_config(): return ProviderSyncResult( ok=False, message="Cloudonix provider not properly configured" ) if not self.application_name: return ProviderSyncResult( ok=False, message=( "Cloudonix application_name is not configured. Set it in " "the telephony configuration so inbound webhooks can be " "synced to the right Voice Application." ), ) app_endpoint = ( f"{self.base_url}/customers/self/domains/{self.domain_id}/" f"applications/{self.application_name}" ) data = { "url": webhook_url, "method": "POST", } try: async with aiohttp.ClientSession() as session: async with session.patch( app_endpoint, json=data, headers=self._get_auth_headers() ) as response: if response.status != 200: body = await response.text() logger.error( f"Cloudonix Voice Application update failed for " f"{self.application_name} on domain " f"{self.domain_id}: {response.status} {body}" ) return ProviderSyncResult( ok=False, message=f"Cloudonix API {response.status}: {body}", ) except Exception as e: logger.error( f"Exception updating Cloudonix Voice Application " f"{self.application_name}: {e}" ) return ProviderSyncResult(ok=False, message=f"Cloudonix update failed: {e}") logger.info( f"Cloudonix url set on Voice Application {self.application_name} " f"(domain={self.domain_id}, triggered by address {address})" ) return ProviderSyncResult(ok=True) async def start_inbound_stream( self, *, websocket_url: str, workflow_run_id: int, normalized_data, backend_endpoint: str, ): """ Generate the appropriate CXML response for an inbound Cloudonix webhook. Returns CXML to connect to WebSocket, same format as outbound calls. """ from fastapi import Response # Generate CXML response (same format as outbound calls) cxml_content = f""" """ logger.info(f"Cloudonix inbound CXML response content:") logger.info(cxml_content) response = Response(content=cxml_content, media_type="application/xml") logger.info(f"Cloudonix inbound response object: {response}") logger.info(f"Response headers: {response.headers}") logger.info(f"Response media type: {response.media_type}") return response @staticmethod def generate_validation_error_response(error_type) -> tuple: """ Generate Cloudonix-specific error response for validation failures. Since Cloudonix is TwiML-compatible, we use the same XML format. """ from fastapi import Response from api.errors.telephony_errors import TELEPHONY_ERROR_MESSAGES, TelephonyError message = TELEPHONY_ERROR_MESSAGES.get( error_type, TELEPHONY_ERROR_MESSAGES[TelephonyError.GENERAL_AUTH_FAILED] ) twiml_content = f""" {message} """ return Response(content=twiml_content, media_type="application/xml") @staticmethod def generate_error_response(error_type: str, message: str) -> tuple: """ Generate a Cloudonix-specific error response. Since Cloudonix is TwiML-compatible, we use TwiML format. """ from fastapi import Response # Map error types to appropriate TwiML responses if error_type == "auth_failed": twiml = """ Authentication failed. This call cannot be processed. """ elif error_type == "not_configured": twiml = """ Service not configured. Please contact support. """ elif error_type == "invalid_number": twiml = """ Invalid phone number. This call cannot be processed. """ else: # Generic error twiml = f""" An error occurred: {message} """ return Response(content=twiml, media_type="application/xml"), "application/xml" # ======== CALL TRANSFER METHODS ======== async def transfer_call( self, destination: str, transfer_id: str, conference_name: str, timeout: int = 30, **kwargs: Any, ) -> Dict[str, Any]: """ Initiate a call transfer via Cloudonix. Uses inline CXML to put the destination into a conference when they answer, and a status callback to track the transfer outcome. 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 Returns: Dict containing transfer result information Raises: ValueError: If provider configuration is invalid Exception: If Twilio API call fails """ raise NotImplementedError("Cloudonix provider does not support call transfers") def supports_transfers(self) -> bool: """ Cloudonix does not support call transfers. Returns: False - Cloudonix provider does not support call transfers """ return False