feat: add vonage telephony

api/services/pipecat/audio_config.py

@@ -80,7 +80,7 @@ def create_audio_config(transport_type: str) -> AudioConfig:
     """Create audio configuration based on transport type.
 
     Args:
-        transport_type: Type of transport ("webrtc", "twilio", "stasis")
+        transport_type: Type of transport ("webrtc", "twilio", "vonage", "stasis")
 
     Returns:
         AudioConfig instance with appropriate settings
@@ -93,6 +93,15 @@ def create_audio_config(transport_type: str) -> AudioConfig:
             pipeline_sample_rate=8000,  # Keep at 8kHz to avoid resampling
             buffer_size_seconds=1.0,
         )
+    elif transport_type == WorkflowRunMode.VONAGE.value:
+        # Vonage uses 16kHz Linear PCM
+        return AudioConfig(
+            transport_in_sample_rate=16000,
+            transport_out_sample_rate=16000,
+            vad_sample_rate=16000,  # Use matching VAD rate
+            pipeline_sample_rate=16000,  # Keep at 16kHz to avoid resampling
+            buffer_size_seconds=1.0,
+        )
     elif transport_type in [
         WorkflowRunMode.WEBRTC.value,
         WorkflowRunMode.SMALLWEBRTC.value,

api/services/pipecat/run_pipeline.py

@@ -4,6 +4,7 @@ from fastapi import HTTPException, WebSocket
 from loguru import logger
 
 from api.db import db_client
+from api.db.models import WorkflowModel
 from api.enums import WorkflowRunMode
 from api.services.pipecat.audio_config import AudioConfig, create_audio_config
 from api.services.pipecat.engine_pre_aggregator_processor import (
@@ -33,6 +34,7 @@ from api.services.pipecat.tracing_config import setup_pipeline_tracing
 from api.services.pipecat.transport_setup import (
     create_stasis_transport,
     create_twilio_transport,
+    create_vonage_transport,
     create_webrtc_transport,
 )
 from api.services.telephony.stasis_rtp_connection import StasisRTPConnection
@@ -70,7 +72,7 @@ async def run_pipeline_twilio(
     set_current_run_id(workflow_run_id)
 
     # Store Twilio call SID in cost_info for later cost calculation
-    cost_info = {"twilio_call_sid": call_sid}
+    cost_info = {"twilio_call_sid": call_sid, "provider": "twilio"}
     await db_client.update_workflow_run(workflow_run_id, cost_info=cost_info)
 
     # Get workflow to extract all pipeline configurations
@@ -107,6 +109,69 @@ async def run_pipeline_twilio(
     )
 
 
+async def run_pipeline_vonage(
+    websocket_client,
+    call_uuid: str,
+    workflow: WorkflowModel,
+    organization_id: int,
+    workflow_id: int,
+    workflow_run_id: int,
+    user_id: int,
+):
+    """Run pipeline for Vonage WebSocket connections.
+    
+    Vonage uses raw PCM audio over WebSocket instead of base64-encoded μ-law.
+    The audio is transmitted as binary frames at 16kHz by default.
+    """
+    logger.info(f"Starting Vonage pipeline for workflow run {workflow_run_id}")
+    set_current_run_id(workflow_run_id)
+
+    # Store Vonage call UUID in cost_info for later cost calculation
+    cost_info = {"vonage_call_uuid": call_uuid, "provider": "vonage"}
+    await db_client.update_workflow_run(workflow_run_id, cost_info=cost_info)
+
+    # Extract VAD and ambient noise config from workflow
+    vad_config = None
+    ambient_noise_config = None
+    if workflow and workflow.workflow_configurations:
+        if "vad_configuration" in workflow.workflow_configurations:
+            vad_config = workflow.workflow_configurations["vad_configuration"]
+        if "ambient_noise_configuration" in workflow.workflow_configurations:
+            ambient_noise_config = workflow.workflow_configurations["ambient_noise_configuration"]
+
+    try:
+        # Setup audio config for Vonage using the centralized config
+        audio_config = create_audio_config(WorkflowRunMode.VONAGE.value)
+
+        # Create Vonage transport
+        transport = await create_vonage_transport(
+            websocket_client,
+            call_uuid,
+            workflow_run_id,
+            audio_config,
+            organization_id,
+            vad_config,
+            ambient_noise_config,
+        )
+
+        # No special handshake needed for Vonage
+        # Audio streaming starts immediately
+
+        # Run the pipeline (same as Twilio/WebRTC)
+        await _run_pipeline(
+            transport,
+            workflow_id,
+            workflow_run_id,
+            user_id,
+            call_context_vars={},
+            audio_config=audio_config,
+        )
+
+    except Exception as e:
+        logger.error(f"Error in Vonage pipeline: {e}")
+        raise
+
+
 async def run_pipeline_smallwebrtc(
     webrtc_connection: SmallWebRTCConnection,
     workflow_id: int,

api/services/pipecat/transport_setup.py

@@ -22,6 +22,7 @@ from pipecat.audio.mixers.soundfile_mixer import SoundfileMixer
 from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams
 from pipecat.audio.vad.silero import SileroVADAnalyzer, VADParams
 from pipecat.serializers.twilio import TwilioFrameSerializer
+from pipecat.serializers.vonage import VonageFrameSerializer
 from pipecat.transports.base_transport import TransportParams
 from pipecat.transports.smallwebrtc.connection import SmallWebRTCConnection
 from pipecat.transports.smallwebrtc.transport import SmallWebRTCTransport
@@ -85,7 +86,7 @@ async def create_twilio_transport(
 
     # Fetch Twilio credentials from organization config
     config = await db_client.get_configuration(
-        organization_id, OrganizationConfigurationKey.TWILIO_CONFIGURATION.value
+        organization_id, OrganizationConfigurationKey.TELEPHONY_CONFIGURATION.value
     )
 
     if not config or not config.value:
@@ -151,6 +152,86 @@ async def create_twilio_transport(
     )
 
 
+async def create_vonage_transport(
+    websocket_client,
+    call_uuid: str,
+    workflow_run_id: int,
+    audio_config: AudioConfig,
+    organization_id: int,
+    vad_config: dict | None = None,
+    ambient_noise_config: dict | None = None,
+):
+    """Create a transport for Vonage connections"""
+
+    # Use the factory to load config from database
+    from api.services.telephony.factory import load_telephony_config
+    config = await load_telephony_config(organization_id)
+    
+    if config.get("provider") != "vonage":
+        raise ValueError(f"Expected Vonage provider, got {config.get('provider')}")
+
+    application_id = config.get("application_id")
+    private_key = config.get("private_key")
+    
+    if not application_id or not private_key:
+        raise ValueError(
+            f"Incomplete Vonage configuration for organization {organization_id}"
+        )
+
+    turn_analyzer = create_turn_analyzer(workflow_run_id, audio_config)
+
+    serializer = VonageFrameSerializer(
+        call_uuid=call_uuid,
+        application_id=application_id,
+        private_key=private_key,
+        params=VonageFrameSerializer.InputParams(
+            vonage_sample_rate=audio_config.transport_in_sample_rate,
+            sample_rate=audio_config.pipeline_sample_rate
+        )
+    )
+
+    # Important: Vonage uses binary WebSocket mode, not text
+    return FastAPIWebsocketTransport(
+        websocket=websocket_client,
+        params=FastAPIWebsocketParams(
+            audio_in_enabled=True,
+            audio_out_enabled=True,
+            audio_in_sample_rate=audio_config.transport_in_sample_rate,
+            audio_out_sample_rate=audio_config.transport_out_sample_rate,
+            vad_analyzer=(
+                SileroVADAnalyzer(
+                    params=VADParams(
+                        confidence=vad_config.get("confidence", 0.7),
+                        start_secs=vad_config.get("start_seconds", 0.4),
+                        stop_secs=vad_config.get("stop_seconds", 0.8),
+                        min_volume=vad_config.get("minimum_volume", 0.6),
+                    )
+                )
+                if vad_config
+                else SileroVADAnalyzer()
+            ),
+            audio_out_mixer=(
+                SoundfileMixer(
+                    sound_files={
+                        "office": APP_ROOT_DIR
+                        / "assets"
+                        / f"office-ambience-{audio_config.transport_out_sample_rate}-mono.wav"
+                    },
+                    default_sound="office",
+                    volume=ambient_noise_config.get("volume", 0.3),
+                )
+                if ambient_noise_config and ambient_noise_config.get("enabled", False)
+                else SilenceAudioMixer()
+            ),
+            turn_analyzer=turn_analyzer,
+            serializer=serializer,
+            audio_in_filter=RNNoiseFilter(library_path=librnnoise_path)
+            if ENABLE_RNNOISE
+            else None,
+        ),
+    )
+
+
 def create_webrtc_transport(
     webrtc_connection: SmallWebRTCConnection,
     workflow_run_id: int,

api/services/telephony/README.md

@@ -15,7 +15,7 @@ Business Logic → TelephonyProvider (Interface) → Concrete Provider (Twilio,
 ```python
 from api.services.telephony.factory import get_telephony_provider
 
-# Get provider based on environment/config
+# Get provider based on organization config
 provider = await get_telephony_provider(organization_id)
 
 # Initiate a call
@@ -35,8 +35,9 @@ telephony/
 ├── factory.py           # Provider creation and config loading
 ├── providers/
 │   ├── __init__.py
-│   └── twilio_provider.py  # Twilio implementation
-├── twilio.py           # Legacy TwilioService (backward compat)
+│   ├── twilio_provider.py  # Twilio implementation
+│   └── vonage_provider.py  # Vonage implementation
+├── twilio.py           # Legacy (removed, use factory instead)
 └── README.md           # This file
 ```
 
@@ -47,9 +48,8 @@ See the [Custom Provider Guide](https://docs.dograh.com/integrations/telephony/c
 Quick checklist:
 1. Create `providers/your_provider.py` implementing `TelephonyProvider`
 2. Update `factory.py` to include your provider
-3. Add environment variable support in `factory.py`
-4. Write unit tests
-5. Update documentation
+3. Write unit tests
+4. Update documentation
 
 ## Key Interfaces
 
@@ -76,21 +76,20 @@ class TelephonyProvider(ABC):
 
 ## Configuration Loading
 
-The `factory.py` handles configuration from two sources:
+The `factory.py` loads configuration from the database:
 
-1. **OSS Mode** (default): Environment variables
-   ```python
-   TELEPHONY_PROVIDER=twilio
-   TWILIO_ACCOUNT_SID=xxx
-   TWILIO_AUTH_TOKEN=xxx
-   TWILIO_FROM_NUMBER=+1234567890
-   ```
-
-2. **SaaS Mode**: Database configuration per organization
+**Both Saas and OSS Modes**: Database configuration via UI
    ```python
    # Loaded from organization_configuration table
-   key: "TWILIO_CONFIGURATION"
-   value: {"account_sid": "xxx", "auth_token": "xxx", "from_numbers": [...]}
+   key: "TELEPHONY_CONFIGURATION"
+   value: {
+       "provider": "twilio",  # or "vonage"
+       "account_sid": "xxx",  # for Twilio
+       "auth_token": "xxx",   # for Twilio
+       "application_id": "xxx",  # for Vonage
+       "private_key": "xxx",     # for Vonage
+       "from_numbers": [...]
+   }
    ```
 
 ## Testing
@@ -117,14 +116,15 @@ async def test_call_initiation(mock_get_provider):
 ### Integration Testing
 
 Run against actual providers in development:
-```bash
-# Set test credentials
-export TELEPHONY_PROVIDER=twilio
-export TWILIO_ACCOUNT_SID=test_sid
-export TWILIO_AUTH_TOKEN=test_token
-export TWILIO_FROM_NUMBER=+15005550006  # Twilio test number
 
-# Run integration tests
+1. Configure your provider through the UI:
+   - Navigate to Settings → Integrations → Telephony
+   - Select your provider (Twilio or Vonage)
+   - Enter test credentials
+   - Save configuration
+
+2. Run integration tests:
+```bash
 pytest tests/integration/test_telephony.py
 ```
 
@@ -155,7 +155,7 @@ await provider.initiate_call(...)
 ## Common Issues
 
 1. **Import Error**: Always import from `factory`, not directly from providers
-2. **Config Not Found**: Check environment variables or database configuration
+2. **Config Not Found**: Check database configuration via UI
 3. **Signature Verification**: Ensure auth tokens match between provider and config
 4. **WebSocket Issues**: Verify audio format compatibility (MULAW for Twilio)
 

api/services/telephony/base.py

@@ -100,4 +100,21 @@ class TelephonyProvider(ABC):
         Returns:
             Provider-specific response (e.g., TwiML for Twilio)
         """
+        pass
+
+    @abstractmethod
+    async def get_call_cost(self, call_id: str) -> Dict[str, Any]:
+        """
+        Get cost information for a completed call.
+        
+        Args:
+            call_id: Provider-specific call identifier (SID for Twilio, UUID for Vonage)
+            
+        Returns:
+            Dict containing:
+                - cost_usd: The cost in USD as float
+                - duration: Call duration in seconds
+                - status: Call completion status
+                - raw_response: Full provider response for debugging
+        """
         pass

api/services/telephony/factory.py

@@ -12,86 +12,74 @@ from api.db import db_client
 from api.enums import OrganizationConfigurationKey
 from api.services.telephony.base import TelephonyProvider
 from api.services.telephony.providers.twilio_provider import TwilioProvider
+from api.services.telephony.providers.vonage_provider import VonageProvider
 
 
-async def load_telephony_config(organization_id: Optional[int] = None) -> Dict[str, Any]:
+async def load_telephony_config(organization_id: int) -> Dict[str, Any]:
     """
-    Load telephony configuration from appropriate source.
+    Load telephony configuration from database.
     
     Args:
-        organization_id: Organization ID for database config (SaaS mode)
-                        None for environment config (OSS mode)
+        organization_id: Organization ID for database config
     
     Returns:
         Configuration dictionary with provider type and credentials
+    
+    Raises:
+        ValueError: If no configuration found for the organization
     """
-    if organization_id:
-        # SaaS mode: Load from database
-        logger.debug(f"Loading telephony config from database for org {organization_id}")
-        
-        # TODO: Use TELEPHONY_CONFIGURATION
-        twilio_config = await db_client.get_configuration(
+    if not organization_id:
+        raise ValueError("Organization ID is required to load telephony configuration")
+    
+    logger.debug(f"Loading telephony config from database for org {organization_id}")
+    
+    # Try new key first
+    config = await db_client.get_configuration(
+        organization_id,
+        OrganizationConfigurationKey.TELEPHONY_CONFIGURATION.value,
+    )
+    
+    # Fallback to old key for backward compatibility
+    if not config:
+        config = await db_client.get_configuration(
             organization_id,
             OrganizationConfigurationKey.TWILIO_CONFIGURATION.value,
         )
-        
-        if twilio_config and twilio_config.value:
-            # TODO: Get provider from config
-            return {
-                "provider": "twilio",
-                "account_sid": twilio_config.value.get("account_sid"),
-                "auth_token": twilio_config.value.get("auth_token"),
-                "from_numbers": twilio_config.value.get("from_numbers", [])
-            }
-        
-        raise ValueError(f"No telephony configuration found for organization {organization_id}")
     
-    else:
-        # OSS mode: Load from environment variables
-        logger.debug("Loading telephony config from environment variables")
-        
-        provider = os.getenv("TELEPHONY_PROVIDER", "twilio").lower()
+    if config and config.value:
+        # Simple single-provider format
+        provider = config.value.get("provider", "twilio")
         
         if provider == "twilio":
-            # Load Twilio config from env
-            account_sid = os.getenv("TWILIO_ACCOUNT_SID")
-            auth_token = os.getenv("TWILIO_AUTH_TOKEN")
-            from_number = os.getenv("TWILIO_FROM_NUMBER")
-            
-            if not all([account_sid, auth_token, from_number]):
-                raise ValueError(
-                    "Missing Twilio configuration. Please set TWILIO_ACCOUNT_SID, "
-                    "TWILIO_AUTH_TOKEN, and TWILIO_FROM_NUMBER environment variables."
-                )
-            
             return {
                 "provider": "twilio",
-                "account_sid": account_sid,
-                "auth_token": auth_token,
-                "from_numbers": [from_number] if isinstance(from_number, str) else from_number
+                "account_sid": config.value.get("account_sid"),
+                "auth_token": config.value.get("auth_token"),
+                "from_numbers": config.value.get("from_numbers", [])
+            }
+        elif provider == "vonage":
+            return {
+                "provider": "vonage",
+                "application_id": config.value.get("application_id"),
+                "private_key": config.value.get("private_key"),
+                "api_key": config.value.get("api_key"),
+                "api_secret": config.value.get("api_secret"),
+                "from_numbers": config.value.get("from_numbers", [])
             }
-        
-        # Future providers can be added here
-        # elif provider == "vonage":
-        #     return {
-        #         "provider": "vonage",
-        #         "api_key": os.getenv("VONAGE_API_KEY"),
-        #         "api_secret": os.getenv("VONAGE_API_SECRET"),
-        #         "from_numbers": [os.getenv("VONAGE_FROM_NUMBER")]
-        #     }
-        
         else:
-            raise ValueError(f"Unknown telephony provider: {provider}")
+            raise ValueError(f"Unknown provider in config: {provider}")
+    
+    raise ValueError(f"No telephony configuration found for organization {organization_id}")
 
 
 async def get_telephony_provider(
-    organization_id: Optional[int] = None
+    organization_id: int
 ) -> TelephonyProvider:
     """
     Factory function to create telephony providers.
     
     Args:
-        organization_id: Organization ID for SaaS mode (optional)
+        organization_id: Organization ID (required)
         
     Returns:
         Configured telephony provider instance
@@ -110,9 +98,10 @@ async def get_telephony_provider(
     if provider_type == "twilio":
         return TwilioProvider(config)
     
+    elif provider_type == "vonage":
+        return VonageProvider(config)
+    
     # Future providers can be added here
-    # elif provider_type == "vonage":
-    #     return VonageProvider(config)
     # elif provider_type == "plivo":
     #     return PlivoProvider(config)
     

api/services/telephony/providers/twilio_provider.py

@@ -128,6 +128,7 @@ class TwilioProvider(TelephonyProvider):
         Verify Twilio webhook signature for security.
         """
         if not self.auth_token:
+            logger.error("No auth token available for webhook signature verification")
             return False
         
         validator = RequestValidator(self.auth_token)
@@ -148,4 +149,56 @@ class TwilioProvider(TelephonyProvider):
     </Connect>
     <Pause length="40"/>
 </Response>"""
-        return twiml_content
+        return twiml_content
+
+    async def get_call_cost(self, call_id: str) -> Dict[str, Any]:
+        """
+        Get cost information for a completed Twilio call.
+        
+        Args:
+            call_id: The Twilio Call SID
+            
+        Returns:
+            Dict containing cost information
+        """
+        endpoint = f"{self.base_url}/Calls/{call_id}.json"
+        
+        try:
+            async with aiohttp.ClientSession() as session:
+                auth = aiohttp.BasicAuth(self.account_sid, self.auth_token)
+                async with session.get(endpoint, auth=auth) as response:
+                    if response.status != 200:
+                        error_data = await response.json()
+                        logger.error(f"Failed to get Twilio call cost: {error_data}")
+                        return {
+                            "cost_usd": 0.0,
+                            "duration": 0,
+                            "status": "error",
+                            "error": str(error_data)
+                        }
+                    
+                    call_data = await response.json()
+                    
+                    # Twilio returns price as a negative string (e.g., "-0.0085")
+                    price_str = call_data.get("price", "0")
+                    cost_usd = abs(float(price_str)) if price_str else 0.0
+                    
+                    # Duration is in seconds as a string
+                    duration = int(call_data.get("duration", "0"))
+                    
+                    return {
+                        "cost_usd": cost_usd,
+                        "duration": duration,
+                        "status": call_data.get("status", "unknown"),
+                        "price_unit": call_data.get("price_unit", "USD"),
+                        "raw_response": call_data
+                    }
+                    
+        except Exception as e:
+            logger.error(f"Exception fetching Twilio call cost: {e}")
+            return {
+                "cost_usd": 0.0,
+                "duration": 0,
+                "status": "error",
+                "error": str(e)
+            }

api/services/telephony/providers/vonage_provider.py

@@ -0,0 +1,274 @@
+"""
+Vonage (Nexmo) implementation of the TelephonyProvider interface.
+"""
+import json
+import random
+import time
+from typing import Any, Dict, List, Optional
+
+import aiohttp
+import jwt
+from loguru import logger
+
+from api.services.telephony.base import TelephonyProvider
+from api.utils.tunnel import TunnelURLProvider
+
+
+class VonageProvider(TelephonyProvider):
+    """
+    Vonage implementation of TelephonyProvider.
+    Uses JWT authentication and NCCO for call control.
+    """
+
+    def __init__(self, config: Dict[str, Any]):
+        """
+        Initialize VonageProvider with configuration.
+        
+        Args:
+            config: Dictionary containing:
+                - api_key: Vonage API Key
+                - api_secret: Vonage API Secret
+                - application_id: Vonage Application ID
+                - private_key: Private key for JWT generation
+                - from_numbers: List of phone numbers to use
+        """
+        self.api_key = config.get("api_key")
+        self.api_secret = config.get("api_secret")
+        self.application_id = config.get("application_id")
+        self.private_key = config.get("private_key")
+        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 = "https://api.nexmo.com"
+
+    def _generate_jwt(self) -> str:
+        """Generate JWT token for Vonage API authentication."""
+        if not self.application_id or not self.private_key:
+            raise ValueError("Application ID and private key required for JWT generation")
+        
+        claims = {
+            "application_id": self.application_id,
+            "iat": int(time.time()),
+            "exp": int(time.time()) + 3600,  # 1 hour expiry
+            "jti": str(time.time())  # Unique token ID
+        }
+        
+        return jwt.encode(claims, self.private_key, algorithm="RS256")
+
+    async def initiate_call(
+        self,
+        to_number: str,
+        webhook_url: str,
+        workflow_run_id: Optional[int] = None,
+        **kwargs: Any,
+    ) -> Dict[str, Any]:
+        """
+        Initiate an outbound call via Vonage Voice API.
+        """
+        if not self.validate_config():
+            raise ValueError("Vonage provider not properly configured")
+        
+        endpoint = f"{self.base_url}/v1/calls"
+        
+        # Select a random phone number
+        from_number = random.choice(self.from_numbers)
+        # Remove + prefix for Vonage
+        from_number = from_number.replace("+", "")
+        to_number = to_number.replace("+", "")
+        
+        logger.info(f"Selected phone number {from_number} for outbound call")
+        
+        # Prepare call data
+        data = {
+            "to": [{
+                "type": "phone",
+                "number": to_number
+            }],
+            "from": {
+                "type": "phone",
+                "number": from_number
+            },
+            "answer_url": [webhook_url],
+            "answer_method": "GET"
+        }
+        
+        # Add event webhook if workflow_run_id provided
+        if workflow_run_id:
+            backend_endpoint = await TunnelURLProvider.get_tunnel_url()
+            event_url = f"https://{backend_endpoint}/api/v1/telephony/events/{workflow_run_id}"
+            data.update({
+                "event_url": [event_url],
+                "event_method": "POST"
+            })
+        
+        # Add any additional kwargs
+        data.update(kwargs)
+        
+        # Generate JWT token
+        token = self._generate_jwt()
+        headers = {
+            "Authorization": f"Bearer {token}",
+            "Content-Type": "application/json"
+        }
+        
+        # Make the API request
+        async with aiohttp.ClientSession() as session:
+            async with session.post(
+                endpoint, 
+                json=data,  # Use json parameter for proper encoding
+                headers=headers
+            ) as response:
+                response_data = await response.json()
+                
+                if response.status != 201:
+                    raise Exception(f"Failed to initiate call: {response_data}")
+                
+                return response_data
+
+    async def get_call_status(self, call_id: str) -> Dict[str, Any]:
+        """
+        Get the current status of a Vonage call.
+        """
+        if not self.validate_config():
+            raise ValueError("Vonage provider not properly configured")
+        
+        endpoint = f"{self.base_url}/v1/calls/{call_id}"
+        
+        # Generate JWT token
+        token = self._generate_jwt()
+        headers = {
+            "Authorization": f"Bearer {token}"
+        }
+        
+        async with aiohttp.ClientSession() as session:
+            async with session.get(endpoint, headers=headers) as response:
+                if response.status != 200:
+                    error_data = await response.json()
+                    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 Vonage phone numbers.
+        """
+        return self.from_numbers
+
+    def validate_config(self) -> bool:
+        """
+        Validate Vonage configuration.
+        """
+        return bool(
+            self.application_id and 
+            self.private_key and 
+            self.from_numbers
+        )
+
+    async def verify_webhook_signature(
+        self, url: str, params: Dict[str, Any], signature: str
+    ) -> bool:
+        """
+        Verify Vonage webhook signature for security.
+        Vonage uses JWT for webhook signatures.
+        """
+        if not self.api_secret:
+            logger.error("No API secret available for webhook signature verification")
+            return False
+        
+        try:
+            # Vonage sends JWT in Authorization header
+            # Verify the JWT signature
+            decoded = jwt.decode(
+                signature, 
+                self.api_secret, 
+                algorithms=["HS256"],
+                options={"verify_signature": True}
+            )
+            return True
+        except jwt.InvalidTokenError:
+            return False
+
+    async def get_webhook_response(
+        self, workflow_id: int, user_id: int, workflow_run_id: int
+    ) -> str:
+        """
+        Generate NCCO response for starting a call session.
+        NCCO (Nexmo Call Control Objects) is JSON-based, unlike TwiML which is XML.
+        """
+        backend_endpoint = await TunnelURLProvider.get_tunnel_url()
+        
+        # NCCO for WebSocket connection
+        ncco = [
+            {
+                "action": "connect",
+                "endpoint": [{
+                    "type": "websocket",
+                    "uri": f"wss://{backend_endpoint}/api/v1/telephony/ws/{workflow_id}/{user_id}/{workflow_run_id}",
+                    "content-type": "audio/l16;rate=16000",  # 16kHz Linear PCM
+                    "headers": {}
+                }]
+            }
+        ]
+        
+        # Return JSON instead of XML
+        return json.dumps(ncco)
+
+    async def get_call_cost(self, call_id: str) -> Dict[str, Any]:
+        """
+        Get cost information for a completed Vonage call.
+        
+        Args:
+            call_id: The Vonage Call UUID
+            
+        Returns:
+            Dict containing cost information
+        """
+        headers = self._get_auth_headers()
+        endpoint = f"https://api.nexmo.com/v1/calls/{call_id}"
+        
+        try:
+            async with aiohttp.ClientSession() as session:
+                async with session.get(endpoint, headers=headers) as response:
+                    if response.status != 200:
+                        error_data = await response.json()
+                        logger.error(f"Failed to get Vonage call cost: {error_data}")
+                        return {
+                            "cost_usd": 0.0,
+                            "duration": 0,
+                            "status": "error",
+                            "error": str(error_data)
+                        }
+                    
+                    call_data = await response.json()
+                    
+                    # Vonage returns price and rate
+                    # Price is the total cost, rate is the per-minute rate
+                    price = float(call_data.get("price", 0))
+                    cost_usd = price  # Vonage returns positive values
+                    
+                    # Duration is in seconds
+                    duration = int(call_data.get("duration", 0))
+                    
+                    # Get the call status
+                    status = call_data.get("status", "unknown")
+                    
+                    return {
+                        "cost_usd": cost_usd,
+                        "duration": duration,
+                        "status": status,
+                        "price_unit": "USD",  # Vonage uses USD by default
+                        "rate": call_data.get("rate", 0),  # Per-minute rate
+                        "raw_response": call_data
+                    }
+                    
+        except Exception as e:
+            logger.error(f"Exception fetching Vonage call cost: {e}")
+            return {
+                "cost_usd": 0.0,
+                "duration": 0,
+                "status": "error",
+                "error": str(e)
+            }

api/services/telephony/twilio.py

@@ -1,196 +0,0 @@
-# TODO: Remove this file after migrating workflow_run_cost.py to use telephony abstraction
-# Deprecated - use api/services/telephony/providers/twilio_provider.py instead
-
-import random
-from typing import Any, Dict, List, Optional
-from urllib.parse import urlencode
-
-import aiohttp
-from loguru import logger
-from pydantic import ValidationError
-from twilio.request_validator import RequestValidator
-
-from api.db import db_client
-from api.enums import OrganizationConfigurationKey
-from api.utils.tunnel import TunnelURLProvider
-
-
-class TwilioService:
-    """Service for interacting with Twilio API."""
-
-    def __init__(self, organization_id: int):
-        """Initialize TwilioService with organization_id."""
-        self.organization_id = organization_id
-        self.account_sid = None
-        self.auth_token = None
-        self.from_numbers = []
-        self.base_url = None
-
-    async def _ensure_credentials(self):
-        """Load credentials from organization configuration."""
-        if self.account_sid and self.auth_token:
-            return
-
-        # Fetch from organization config only - no env var fallback
-        config = await db_client.get_configuration(
-            self.organization_id,
-            OrganizationConfigurationKey.TWILIO_CONFIGURATION.value,
-        )
-
-        if not config or not config.value:
-            raise ValidationError(
-                "Twilio credentials not configured for this organization. "
-                "Please configure telephony settings."
-            )
-
-        self.account_sid = config.value.get("account_sid")
-        self.auth_token = config.value.get("auth_token")
-        self.from_numbers = config.value.get("from_numbers", [])
-
-        if not self.account_sid or not self.auth_token or not self.from_numbers:
-            raise ValidationError(
-                "Incomplete Twilio configuration. Please update telephony settings."
-            )
-
-        self.base_url = f"https://api.twilio.com/2010-04-01/Accounts/{self.account_sid}"
-
-    async def get_organization_phone_numbers(self) -> List[str]:
-        """
-        Get the list of Twilio phone numbers configured for the organization.
-
-        Returns:
-            List of phone numbers
-        """
-        await self._ensure_credentials()
-        return self.from_numbers
-
-    async def initiate_call(
-        self,
-        to_number: str,
-        url_args: Dict[str, Any] = {},
-        workflow_run_id: Optional[int] = None,
-        **kwargs: Any,
-    ) -> Dict[str, Any]:
-        """
-        Initiates a Twilio call using the Calls API.
-
-        Args:
-            to_number: The destination phone number
-            url_args: Dictionary of URL parameters to append to the base URL
-            workflow_run_id: The workflow run ID for tracking callbacks
-            **kwargs: Additional parameters to pass to the Twilio API
-
-        Returns:
-            Dict containing the Twilio API response
-        """
-        await self._ensure_credentials()
-
-        endpoint = f"{self.base_url}/Calls.json"
-
-        # Get tunnel URL at runtime
-        backend_endpoint = await TunnelURLProvider.get_tunnel_url()
-
-        # Construct the URL with parameters if any
-        url: str = f"https://{backend_endpoint}/api/v1/twilio/twiml"
-        if url_args:
-            query_string = urlencode(url_args)
-            url = f"{url}?{query_string}"
-
-        logger.debug(f"Initiating call with URL: {url}")
-
-        # Get phone numbers for organization and select one randomly
-        phone_numbers = await self.get_organization_phone_numbers()
-        from_number = random.choice(phone_numbers)
-        logger.info(
-            f"Selected phone number {from_number} from {len(phone_numbers)} "
-            f"available numbers for org {self.organization_id}"
-        )
-
-        # Prepare call data
-        data = {"To": to_number, "From": from_number, "Url": url}
-
-        # Add status callback configuration if workflow_run_id is provided
-        if workflow_run_id:
-            callback_url = f"https://{backend_endpoint}/api/v1/twilio/status-callback/{workflow_run_id}"
-            data.update(
-                {
-                    "StatusCallback": callback_url,
-                    "StatusCallbackEvent": [
-                        "initiated",
-                        "ringing",
-                        "answered",
-                        "completed",
-                    ],
-                    "StatusCallbackMethod": "POST",
-                }
-            )
-
-        # Add any additional kwargs
-        data.update(kwargs)
-
-        # Make the API request
-        async with aiohttp.ClientSession() as session:
-            auth = aiohttp.BasicAuth(self.account_sid, self.auth_token)
-            async with session.post(endpoint, data=data, auth=auth) as response:
-                if response.status != 201:
-                    error_data = await response.json()
-                    raise Exception(f"Failed to initiate call: {error_data}")
-
-                return await response.json()
-
-    async def get_start_call_twiml(
-        self, workflow_id: int, user_id: int, workflow_run_id: int
-    ) -> str:
-        # Get tunnel URL at runtime
-        backend_endpoint = await TunnelURLProvider.get_tunnel_url()
-
-        twiml_content = f"""<?xml version="1.0" encoding="UTF-8"?>
-<Response>
-    <Connect>
-        <Stream url="wss://{backend_endpoint}/api/v1/twilio/ws/{workflow_id}/{user_id}/{workflow_run_id}"></Stream>
-    </Connect>
-    <Pause length="40"/>
-</Response>"""
-        return twiml_content
-
-    async def get_call(self, call_sid: str) -> Dict[str, Any]:
-        """
-        Retrieves information about a specific call.
-
-        Args:
-            call_sid: The SID of the call to retrieve
-
-        Returns:
-            Dict containing the call information
-        """
-        await self._ensure_credentials()
-
-        endpoint = f"{self.base_url}/Calls/{call_sid}.json"
-
-        async with aiohttp.ClientSession() as session:
-            auth = aiohttp.BasicAuth(self.account_sid, self.auth_token)
-            async with session.get(endpoint, auth=auth) as response:
-                if response.status != 200:
-                    error_data = await response.json()
-                    raise Exception(f"Failed to get call: {error_data}")
-
-                return await response.json()
-
-    async def verify_signature(
-        self, url: str, params: Dict[str, Any], signature: str
-    ) -> bool:
-        """
-        Verify Twilio request signature using official Twilio SDK.
-
-        Args:
-            url: The full URL of the webhook
-            params: The POST parameters (form data) as a dictionary
-            signature: The X-Twilio-Signature header value
-
-        Returns:
-            bool: True if signature is valid, False otherwise
-        """
-        await self._ensure_credentials()
-
-        validator = RequestValidator(self.auth_token)
-        return validator.validate(url, params, signature)