feat: add amd callback

api/routes/telephony.py

@@ -6,8 +6,7 @@ Consolidated from split modules for easier maintenance.
 import json
 import uuid
 from datetime import UTC, datetime
-from typing import Optional
-
+from typing import Optional, Any, Dict
 from fastapi import (
     APIRouter,
     Depends,
@@ -49,6 +48,7 @@ from api.utils.telephony_helper import (
     normalize_webhook_data,
     numbers_match,
     parse_webhook_request,
+    parse_cloudonix_amd_callback
 )
 from pipecat.utils.run_context import set_current_run_id
 
@@ -1182,13 +1182,12 @@ async def handle_cloudonix_status_callback(
         logger.warning(f"Workflow run {workflow_run_id} not found for status callback")
         return {"status": "ignored", "reason": "workflow_run_not_found"}
 
-    # Get workflow and provider
-    workflow = await db_client.get_workflow_by_id(workflow_run.workflow_id)
-    if not workflow:
+    # Get workflow from workflow_run and get the provider
+    if not workflow_run.workflow:
         logger.warning(f"Workflow {workflow_run.workflow_id} not found")
         return {"status": "ignored", "reason": "workflow_not_found"}
 
-    provider = await get_telephony_provider(workflow.organization_id)
+    provider = await get_telephony_provider(workflow_run.workflow.organization_id)
 
     # Parse the callback data into generic format
     parsed_data = provider.parse_status_callback(callback_data)
@@ -1209,6 +1208,71 @@ async def handle_cloudonix_status_callback(
 
     return {"status": "success"}
 
+@router.post("/cloudonix/amd-callback/{workflow_run_id}")
+async def handle_cloudonix_amd_callback(
+    workflow_run_id: int,
+    request: Request,
+):
+    """Handle Cloudonix-specific status callbacks.
+
+    Cloudonix sends call status updates to the callback URL specified during call initiation.
+    """
+    set_current_run_id(workflow_run_id)
+    # Parse callback data - determine if JSON or form data
+    content_type = request.headers.get("content-type", "")
+
+    if "application/json" in content_type:
+        callback_data = await request.json()
+    else:
+        # Assume form data (like Twilio)
+        form_data = await request.form()
+        callback_data = dict(form_data)
+
+    logger.info(
+        f"[run {workflow_run_id}] Received Cloudonix AMD status callback: {json.dumps(callback_data)}"
+    )
+    
+    call_id = callback_data["CallSid"]
+    answered_by = callback_data["AnsweredBy"]
+    logger.info(
+        f"[run {workflow_run_id}] returning from the AMD callback for call answered-by: {answered_by}"
+)
+
+    if answered_by in ["human", "unknown"]:
+        logger.info(
+        f"[run {workflow_run_id}] returning from the AMD callback for call answered-by: {answered_by}"
+)
+        return {"status": answered_by}
+    
+    logger.info(
+        f"[run {workflow_run_id}] proceeding with call hang up and workflow run update for call answered-by: {answered_by}"
+    )
+    workflow_run = await db_client.get_workflow_run_by_call_id(call_id)
+    if not workflow_run:
+        logger.warning(f"No workflow run found for Cloudonix call_id: {call_id}")
+        return {"status": "ignored", "reason": "workflow_run_not_found"}
+    organization_id = workflow_run.workflow.organization_id
+    provider = await get_telephony_provider(organization_id)
+
+    # Parse the AMD callback data to persist in workflow_run
+    parsed_data = parse_cloudonix_amd_callback(callback_data)
+
+    # Create status update for AMD detection
+    status_update = StatusCallbackRequest(
+        call_id=parsed_data["call_id"],
+        status=f"amd_{parsed_data['answered_by']}",
+        extra_data=parsed_data,
+    )
+
+    # Process the status update
+    await _process_status_update(workflow_run_id, status_update)
+
+    logger.info(
+        f"[run {workflow_run_id}] Call answered by: {answered_by}, hang up call with call_id: {call_id}"
+    )
+    await provider.hang_up(call_id)
+    return {"status": answered_by}
+
 
 @router.post("/vobiz/hangup-callback/workflow/{workflow_id}")
 async def handle_vobiz_hangup_callback_by_workflow(

api/services/telephony/base.py

@@ -352,3 +352,17 @@ class TelephonyProvider(ABC):
             True if provider supports call transfers, False otherwise
         """
         pass
+
+    @abstractmethod
+    async def hang_up(self, call_id: str, **kwargs: Any) -> Dict[str, Any]:
+        """
+        Terminate an active call.
+
+        Args:
+            call_id: Provider-specific call identifier
+            **kwargs: Provider-specific additional parameters
+
+        Returns:
+            Dict containing hangup response (format varies by provider)
+        """
+        pass

api/services/telephony/providers/ari_provider.py

@@ -418,3 +418,8 @@ class ARIProvider(TelephonyProvider):
             f"&app={self.app_name}"
             f"&subscribeAll=true"
         )
+
+    async def hang_up(self, call_id: str, **kwargs: Any) -> Dict[str, Any]:
+        """Terminate an ARI call."""
+        # TODO: Implement ARI call termination
+        return {"status": "not_implemented"}

api/services/telephony/providers/cloudonix_provider.py

@@ -106,6 +106,16 @@ class CloudonixProvider(TelephonyProvider):
             "caller-id": from_number,  # Required field
         }
 
+        # Enable and process AMD
+        data["machineDetection"] = "DetectMessageEnd"
+        data["asyncAmd"] = True
+        data["machineDetectionTimeout"] = 30
+        data["machineDetectionSpeechThreshold"] = 2500
+        data["machineDetectionSpeechEndThreshold"] = 6000
+        data["machineDetectionSilenceTimeout"] = 2500
+        data["asyncAmdStatusCallback"] = f"{backend_endpoint}/api/v1/telephony/cloudonix/amd-callback/{workflow_run_id}"
+        data["asyncAmdStatusCallbackMethod"]= "POST"
+
         # 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
@@ -682,6 +692,42 @@ class CloudonixProvider(TelephonyProvider):
 
         return Response(content=twiml, media_type="application/xml"), "application/xml"
 
+    async def hang_up(self, call_id: str, **kwargs: Any) -> Dict[str, Any]:
+
+        # Construct the DELETE session endpoint
+        # Using "self" as customer-id as per Cloudonix documentation
+        endpoint = (
+            f"{self.base_url}/customers/self/domains/{self.domain_id}/sessions/{call_id}"
+        )
+
+        # Prepare headers with Bearer token authentication
+        headers = {
+            "Authorization": f"Bearer {self.bearer_token}",
+            "Content-Type": "application/json",
+        }
+
+        logger.info(f"Terminating Cloudonix call {call_id} via DELETE {endpoint}")
+
+        # Make the DELETE request to terminate the session
+        async with aiohttp.ClientSession() as session:
+            async with session.delete(endpoint, headers=headers) as response:
+                status = response.status
+                response_text = await response.text()
+
+                if status in (200, 204, 404):
+                    # 200/204: Success
+                    # 404: Session already terminated (acceptable)
+                    logger.info(
+                        f"Successfully terminated Cloudonix session {call_id} "
+                        f"(HTTP {status}), Response: {response_text}"
+                    )
+                else:
+                    logger.warning(
+                        f"Unexpected response terminating Cloudonix session {call_id}: "
+                        f"HTTP {status}, Response: {response_text}"
+                    )        
+        return {"status": "success"}
+
     # ======== CALL TRANSFER METHODS ========
 
     async def transfer_call(

api/services/telephony/providers/twilio_provider.py

@@ -587,3 +587,8 @@ class TwilioProvider(TelephonyProvider):
             True - Twilio provider supports call transfers
         """
         return True
+
+    async def hang_up(self, call_id: str, **kwargs: Any) -> Dict[str, Any]:
+        """Terminate a Twilio call."""
+        # TODO: Implement Twilio call termination
+        return {"status": "not_implemented"}

api/services/telephony/providers/vobiz_provider.py

@@ -560,3 +560,8 @@ class VobizProvider(TelephonyProvider):
             False - Vobiz provider does not support call transfers
         """
         return False
+
+    async def hang_up(self, call_id: str, **kwargs: Any) -> Dict[str, Any]:
+        """Terminate a Vobiz call."""
+        # TODO: Implement Vobiz call termination
+        return {"status": "not_implemented"}

api/services/telephony/providers/vonage_provider.py

@@ -511,3 +511,8 @@ class VonageProvider(TelephonyProvider):
             False - Vonage provider does not support call transfers
         """
         return False
+
+    async def hang_up(self, call_id: str, **kwargs: Any) -> Dict[str, Any]:
+        """Terminate a Vonage call."""
+        # TODO: Implement Vonage call termination
+        return {"status": "not_implemented"}

api/utils/telephony_helper.py

@@ -231,3 +231,30 @@ def get_countries_for_code(dialing_code: str) -> list[str]:
         return []
 
     return [country for country, code in COUNTRY_CODES.items() if code == dialing_code]
+
+
+def parse_cloudonix_amd_callback(data: dict) -> dict:
+    """
+    Parse Cloudonix AMD callback data into generic format.
+    
+    Note: This is Cloudonix-specific and not part of the generic provider interface
+    as AMD callbacks are currently only supported by Cloudonix.
+    
+    Args:
+        data: Raw AMD callback data from Cloudonix
+        
+    Returns:
+        Dict with parsed AMD information
+    """
+    return {
+        "call_id": data.get("CallSid", ""),
+        "session": data.get("Session", ""),
+        "answered_by": data.get("AnsweredBy", ""),
+        "from_number": data.get("From", ""),
+        "to_number": data.get("To", ""),
+        "call_status": data.get("CallStatus", ""),
+        "domain": data.get("Domain", ""),
+        "direction": data.get("Direction", ""),
+        "account_sid": data.get("AccountSid", ""),
+        "api_version": data.get("ApiVersion", "")
+    }