""" Asterisk ARI (Asterisk REST Interface) implementation of the TelephonyProvider interface. Uses ARI REST API to originate calls into a Stasis application. The ARI WebSocket event listener runs as a separate process (ari_manager.py). """ import json from typing import TYPE_CHECKING, Any, Dict, List, Optional from urllib.parse import urlparse import aiohttp from fastapi import HTTPException from loguru import logger from pipecat.utils.enums import EndTaskReason from api.db import db_client from api.enums import TelephonyCallStatus, WorkflowRunMode from api.services.telephony.base import ( CallInitiationResult, NormalizedInboundData, ProviderSyncResult, TelephonyProvider, ) from api.services.telephony.providers.ari.external_pbx import create_adapter if TYPE_CHECKING: from fastapi import WebSocket class ARIProvider(TelephonyProvider): """ Asterisk ARI implementation of TelephonyProvider. Uses ARI REST API for call control and relies on a separate ari_manager process for WebSocket event listening. """ PROVIDER_NAME = WorkflowRunMode.ARI.value WEBHOOK_ENDPOINT = None # ARI uses WebSocket events, not webhooks def __init__(self, config: Dict[str, Any]): """ Initialize ARIProvider with configuration. Args: config: Dictionary containing: - ari_endpoint: ARI base URL (e.g., http://asterisk:8088) - app_name: ARI username (ari.conf section name) - app_password: ARI user password - stasis_app_name: Stasis application to originate into - from_numbers: List of SIP extensions/numbers (optional) ``app_name`` authenticates to Asterisk; ``stasis_app_name`` names the dialplan application calls are placed into. Asterisk treats these as unrelated namespaces, and Dograh generates the second one so two configurations can never claim the same application. Configurations written before the split have only ``app_name`` and use it for both. """ self.ari_endpoint = config.get("ari_endpoint", "").rstrip("/") self.app_name = config.get("app_name", "") self.app_password = config.get("app_password", "") self.stasis_app_name = config.get("stasis_app_name") or self.app_name self.from_numbers = config.get("from_numbers", []) self.default_from_number = config.get("default_from_number") self.external_pbx_adapter = create_adapter(config.get("external_pbx")) if isinstance(self.from_numbers, str): self.from_numbers = [self.from_numbers] self.base_url = f"{self.ari_endpoint}/ari" def _get_auth(self) -> aiohttp.BasicAuth: """Generate BasicAuth for ARI API requests.""" return aiohttp.BasicAuth(self.app_name, self.app_password) 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 ARI. Creates a channel in Asterisk using the ARI channels endpoint. The channel is placed into the Stasis application where the ari_manager will receive the StasisStart event. """ if not self.validate_config(): raise ValueError("ARI provider not properly configured") endpoint = f"{self.base_url}/channels" # Build the SIP endpoint string # to_number can be a SIP URI or extension if to_number.startswith("SIP/") or to_number.startswith("PJSIP/"): sip_endpoint = to_number else: # Default to PJSIP technology sip_endpoint = f"PJSIP/{to_number}" # Prepare channel creation data params = { "endpoint": sip_endpoint, "app": self.stasis_app_name, "appArgs": ",".join( filter( None, [ f"workflow_run_id={workflow_run_id}", f"workflow_id={kwargs.get('workflow_id', '')}", ], ) ), } # ARI omits callerId entirely when nothing is requested (the trunk # decides), so only fall back to the config's default — never random. from_number = from_number or self.default_from_number if from_number: params["callerId"] = from_number logger.info( f"[ARI] Initiating call to {sip_endpoint} " f"via app={self.stasis_app_name}, workflow_run_id={workflow_run_id}" ) async with aiohttp.ClientSession() as session: async with session.post( endpoint, params=params, auth=self._get_auth(), ) as response: response_text = await response.text() if response.status != 200: raise HTTPException( status_code=response.status, detail=f"Failed to create ARI channel: {response_text}", ) response_data = json.loads(response_text) channel_id = response_data.get("id", "") logger.info( f"[ARI] Channel created: {channel_id} " f"state={response_data.get('state')}" ) return CallInitiationResult( call_id=channel_id, status=response_data.get("state", "created"), caller_number=from_number, provider_metadata={ "call_id": channel_id, "channel_name": response_data.get("name", ""), }, raw_response=response_data, ) async def get_call_status(self, call_id: str) -> Dict[str, Any]: """Get channel status from ARI.""" if not self.validate_config(): raise ValueError("ARI provider not properly configured") endpoint = f"{self.base_url}/channels/{call_id}" async with aiohttp.ClientSession() as session: async with session.get(endpoint, auth=self._get_auth()) as response: if response.status != 200: error_data = await response.text() raise Exception(f"Failed to get channel status: {error_data}") return await response.json() async def get_available_phone_numbers(self) -> List[str]: """Return configured extensions/numbers.""" return self.from_numbers def validate_config(self) -> bool: """Validate ARI configuration.""" return bool(self.ari_endpoint and self.app_name and self.app_password) async def validate_phone_number(self, address: str) -> ProviderSyncResult: """Accept PBX-managed caller IDs and extensions. ARI exposes channel endpoints and accepts ``callerId`` during channel origination, but it has no carrier-number ownership resource. The PBX dialplan and outbound trunk remain authoritative for these addresses. """ return ProviderSyncResult(ok=True) async def verify_webhook_signature( self, url: str, params: Dict[str, Any], signature: str ) -> bool: """ARI does not use webhook signatures - events come via WebSocket.""" return True async def get_webhook_response( self, workflow_id: int, organization_id: int, workflow_run_id: int ) -> str: """ARI does not use webhook responses - call control is via REST API.""" logger.warning( "get_webhook_response called for ARI - this should not happen. " "ARI uses REST API for call control, not webhooks." ) return "" async def get_call_cost(self, call_id: str) -> Dict[str, Any]: """ARI/Asterisk does not provide call cost information.""" return { "cost_usd": 0.0, "duration": 0, "status": "unknown", "error": "ARI does not support cost retrieval", } def parse_status_callback(self, data: Dict[str, Any]) -> Dict[str, Any]: """ Parse ARI event data into generic status callback format. ARI events come from the WebSocket listener, not HTTP callbacks. """ # Map ARI channel states to common status format state_map = { "Up": TelephonyCallStatus.ANSWERED, "Down": TelephonyCallStatus.COMPLETED, "Ringing": TelephonyCallStatus.RINGING, "Ring": TelephonyCallStatus.RINGING, "Busy": TelephonyCallStatus.BUSY, "Unavailable": TelephonyCallStatus.FAILED, } channel_state = data.get("channel", {}).get("state", "") event_type = data.get("type", "") # Determine status from event type if event_type == "StasisStart": status = TelephonyCallStatus.ANSWERED elif event_type == "StasisEnd": status = TelephonyCallStatus.COMPLETED elif event_type == "ChannelDestroyed": status = TelephonyCallStatus.COMPLETED else: status = state_map.get(channel_state, channel_state.lower()) channel = data.get("channel", {}) return { "call_id": channel.get("id", ""), "status": status, "from_number": channel.get("caller", {}).get("number"), "to_number": channel.get("dialplan", {}).get("exten"), "direction": None, "duration": None, "extra": data, } async def handle_websocket( self, websocket: "WebSocket", workflow_id: int, organization_id: int, workflow_run_id: int, ) -> None: """ Handle WebSocket connection from ARI externalMedia channel. Unlike Twilio (which sends "connected" and "start" JSON messages), Asterisk chan_websocket starts streaming audio immediately. """ from api.services.pipecat.run_pipeline import run_pipeline_telephony # Get channel_id from workflow run context workflow_run = await db_client.get_workflow_run( workflow_run_id, organization_id=organization_id ) channel_id = "" if workflow_run and workflow_run.gathered_context: channel_id = workflow_run.gathered_context.get("call_id", "") logger.info( f"[ARI] Starting pipeline for workflow_run {workflow_run_id}, channel={channel_id}" ) await run_pipeline_telephony( websocket, provider_name=self.PROVIDER_NAME, workflow_id=workflow_id, workflow_run_id=workflow_run_id, organization_id=organization_id, call_id=channel_id, transport_kwargs={"channel_id": channel_id}, ) # ======== INBOUND CALL METHODS ======== @classmethod def can_handle_webhook( cls, webhook_data: Dict[str, Any], headers: Dict[str, str] ) -> bool: """ ARI does not use HTTP webhooks for inbound calls. Inbound calls are received via the ARI WebSocket event listener. """ return False @staticmethod def parse_inbound_webhook(webhook_data: Dict[str, Any]) -> NormalizedInboundData: """Parse ARI event data into normalized inbound format.""" channel = webhook_data.get("channel", {}) caller = channel.get("caller", {}) connected = channel.get("connected", {}) return NormalizedInboundData( provider=ARIProvider.PROVIDER_NAME, call_id=channel.get("id", ""), from_number=caller.get("number", ""), to_number=channel.get("dialplan", {}).get("exten", ""), direction="inbound", call_status=channel.get("state", ""), account_id=None, raw_data=webhook_data, ) @staticmethod def validate_account_id(config_data: dict, webhook_account_id: str) -> bool: """ARI doesn't use account IDs for validation.""" return True async def verify_inbound_signature( self, url: str, webhook_data: Dict[str, Any], headers: Dict[str, str], body: str = "", ) -> bool: """ARI authenticates via WebSocket connection credentials, not signatures.""" return True async def start_inbound_stream( self, *, websocket_url: str, workflow_run_id: int, normalized_data, backend_endpoint: str, ): """ARI does not generate HTTP responses for inbound calls.""" from fastapi import Response return Response(content="", status_code=204) @staticmethod def generate_error_response(error_type: str, message: str) -> tuple: """Generate a generic JSON error response.""" from fastapi import Response return Response( content=json.dumps({"error": error_type, "message": message}), media_type="application/json", ) @staticmethod def generate_validation_error_response(error_type) -> tuple: """Generate JSON error response for validation failures.""" 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] ) return Response( content=json.dumps({"error": str(error_type), "message": message}), media_type="application/json", ) # ======== CALL TRANSFER METHODS ======== def supports_transfers(self) -> bool: """ARI supports call transfers via bridge manipulation.""" return True async def transfer_external_pbx_call( self, *, identity: Dict[str, Any], destination: str, field_updates: Optional[Dict[str, str]] = None, disposition: Optional[str] = None, ) -> Optional[Dict[str, Any]]: """Delegate a PBX-owned customer leg to the configured adapter.""" adapter = self.external_pbx_adapter if adapter is None or not identity: return None identity_type = identity.get("type") or identity.get("provider") if identity_type != adapter.type: logger.warning( "[ARI External PBX] Captured identity does not match configured " f"adapter: identity={identity_type!r} adapter={adapter.type!r}" ) return { "status": "failed", "action": "external_pbx_transfer", "message": "The external PBX call identity is invalid.", "reason": "external_pbx_identity_mismatch", } # The outcome is known here by construction -- this method exists to # perform a transfer -- and that is what lets it be recorded before the # PBX pulls the customer off our leg. It cannot be read out of # gathered_context, which the caller resolves `field_updates` from while # the call is still in progress, before the transfer has been stamped. # The caller passes `disposition` so the code written here is the same # one the organization's mapping puts on the run; falling back to the # raw reason covers callers with no mapping to consult. update_fields = { **adapter.disposition_fields( disposition or EndTaskReason.CALL_TRANSFERRED.value ), **(field_updates or {}), } update_result = None if update_fields: update_result = await adapter.update_fields(identity, update_fields) if not update_result.ok: logger.warning( "[ARI External PBX] Field update failed; continuing transfer " f"adapter={adapter.type} message={update_result.message}" ) transfer_result = await adapter.transfer(identity, destination) return { "status": "success" if transfer_result.ok else "failed", "action": "external_pbx_transfer", "message": ( "Transferring your call now." if transfer_result.ok else "I'm sorry, I couldn't complete the transfer." ), "reason": None if transfer_result.ok else "external_pbx_transfer_failed", "field_update_ok": update_result.ok if update_result else None, } async def transfer_call( self, destination: str, transfer_id: str, conference_name: str, timeout: int = 30, **kwargs: Any, ) -> Dict[str, Any]: """Initiate ARI call transfer by creating an outbound channel to the destination. This method creates the destination channel and returns immediately. The transfer process completes asynchronously - success/failure is determined by ARI events and communicated through the transfer event system. Args: destination: Destination phone number (SIP endpoint) transfer_id: Unique identifier for this transfer attempt conference_name: Conference name (unused in ARI, kept for interface compatibility) timeout: Transfer timeout in seconds **kwargs: Additional arguments Returns: Dict containing: - call_sid: Destination channel ID - status: "initiated" - provider: "ari" - raw_response: Full ARI channel creation response """ if not self.validate_config(): raise ValueError("ARI provider not properly configured") logger.info( f"[ARI Transfer] Initiating transfer {transfer_id} to {destination} " f"(timeout: {timeout}s)" ) from api.services.telephony.call_transfer_manager import ( get_call_transfer_manager, ) # Get call transfer manager for event correlation mapping call_transfer_manager = await get_call_transfer_manager() # Build SIP endpoint if destination.startswith("SIP/") or destination.startswith("PJSIP/"): sip_endpoint = destination else: sip_endpoint = f"PJSIP/{destination}" # Build transfer appArgs for event correlation app_args = f"transfer,{transfer_id}" try: endpoint = f"{self.base_url}/channels" params = { "endpoint": sip_endpoint, "app": self.stasis_app_name, "appArgs": app_args, "timeout": timeout, # Keep timeout for transfer calls } async with aiohttp.ClientSession() as session: async with session.post( endpoint, params=params, auth=self._get_auth(), ) as response: response_text = await response.text() if response.status != 200: error_msg = f"ARI channel creation failed: {response.status} {response_text}" logger.error(f"[ARI Transfer] {error_msg}") await call_transfer_manager.remove_transfer_context(transfer_id) raise Exception(error_msg) result = json.loads(response_text) destination_channel_id = result.get("id", "") if not destination_channel_id: logger.error( f"[ARI Transfer] Failed to get channel ID from response: {result}" ) await call_transfer_manager.remove_transfer_context(transfer_id) raise Exception("Failed to create destination channel") # Store transfer channel mapping for event correlation await call_transfer_manager.store_transfer_channel_mapping( destination_channel_id, transfer_id ) logger.info( f"[ARI Transfer] Originated destination channel {destination_channel_id} " f"for transfer {transfer_id}" ) return { "call_sid": destination_channel_id, "status": "initiated", "provider": self.PROVIDER_NAME, "raw_response": result, } except Exception as e: logger.error( f"[ARI Transfer] Failed to originate call transfer destination channel: {e}" ) await call_transfer_manager.remove_transfer_context(transfer_id) raise # ======== ARI-SPECIFIC METHODS ======== async def hangup_channel(self, channel_id: str, reason: str = "normal") -> bool: """Hang up an ARI channel.""" endpoint = f"{self.base_url}/channels/{channel_id}" params = {"reason_code": reason} try: async with aiohttp.ClientSession() as session: async with session.delete( endpoint, params=params, auth=self._get_auth() ) as response: if response.status in (200, 204): logger.info(f"[ARI] Channel {channel_id} hung up") return True else: error = await response.text() logger.error( f"[ARI] Failed to hangup channel {channel_id}: {error}" ) return False except Exception as e: logger.error(f"[ARI] Exception hanging up channel {channel_id}: {e}") return False async def answer_channel(self, channel_id: str) -> bool: """Answer an ARI channel.""" endpoint = f"{self.base_url}/channels/{channel_id}/answer" try: async with aiohttp.ClientSession() as session: async with session.post(endpoint, auth=self._get_auth()) as response: if response.status in (200, 204): logger.info(f"[ARI] Channel {channel_id} answered") return True else: error = await response.text() logger.error( f"[ARI] Failed to answer channel {channel_id}: {error}" ) return False except Exception as e: logger.error(f"[ARI] Exception answering channel {channel_id}: {e}") return False def get_ws_url(self) -> str: """Get the ARI WebSocket URL for event listening.""" parsed = urlparse(self.ari_endpoint) ws_scheme = "wss" if parsed.scheme == "https" else "ws" return ( f"{ws_scheme}://{parsed.netloc}/ari/events" f"?api_key={self.app_name}:{self.app_password}" f"&app={self.stasis_app_name}" f"&subscribeAll=true" )