Skip to main content

Overview

Dograh AI’s telephony abstraction layer allows you to integrate any telephony service by implementing the TelephonyProvider interface.

Provider Interface

All telephony providers must implement this abstract base class:
from abc import ABC, abstractmethod
from typing import Any, Dict, List, Optional

class TelephonyProvider(ABC):
    """Abstract base class for telephony providers."""
    
    @abstractmethod
    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."""
        pass
    
    @abstractmethod
    async def get_call_status(self, call_id: str) -> Dict[str, Any]:
        """Get current status of a call."""
        pass
    
    @abstractmethod
    async def get_available_phone_numbers(self) -> List[str]:
        """Get list of available phone numbers."""
        pass
    
    @abstractmethod
    def validate_config(self) -> bool:
        """Validate provider configuration."""
        pass
    
    @abstractmethod
    async def verify_webhook_signature(
        self, url: str, params: Dict[str, Any], signature: str
    ) -> bool:
        """Verify webhook signature for security."""
        pass
    
    @abstractmethod
    async def get_webhook_response(
        self, workflow_id: int, user_id: int, workflow_run_id: int
    ) -> str:
        """Generate initial webhook response."""
        pass
    
    async def get_call_cost(self, call_id: str) -> Dict[str, Any]:
        """Get cost information for a completed call."""
        pass

Implementation Guide

1. Create Your Provider

Create a new file in api/services/telephony/providers/:
# api/services/telephony/providers/your_provider.py

from typing import Any, Dict, List, Optional
from api.services.telephony.base import TelephonyProvider

class YourProvider(TelephonyProvider):
    """Your custom telephony provider implementation."""
    
    def __init__(self, config: Dict[str, Any]):
        """Initialize with configuration dictionary."""
        # Extract your provider-specific configuration
        self.api_key = config.get("api_key")
        self.api_secret = config.get("api_secret")
        self.from_number = config.get("from_numbers", [""])[0]
    
    def validate_config(self) -> bool:
        """Check if all required configuration is present."""
        return bool(self.api_key and self.api_secret and self.from_number)
    
    async def initiate_call(
        self,
        to_number: str,
        webhook_url: str,
        workflow_run_id: Optional[int] = None,
        **kwargs: Any
    ) -> Dict[str, Any]:
        """Start an outbound call using your provider's API."""
        # Implement your provider's call initiation logic
        pass
    
    # Implement other required methods...

2. Register in Factory

Update api/services/telephony/factory.py to include your provider:
from api.services.telephony.providers.your_provider import YourProvider

async def get_telephony_provider(
    organization_id: int
) -> TelephonyProvider:
    """Factory function to get appropriate telephony provider."""
    
    config = await load_telephony_config(organization_id)
    provider_type = config.get("provider", "twilio")
    
    if provider_type == "twilio":
        return TwilioProvider(config)
    elif provider_type == "vonage":
        return VonageProvider(config)
    elif provider_type == "your_provider":
        return YourProvider(config)
    else:
        raise ValueError(f"Unknown telephony provider: {provider_type}")

3. Add Configuration Support

Update the configuration loader in factory.py to handle your provider’s database configuration:
# In load_telephony_config function
if provider == "your_provider":
    return {
        "provider": "your_provider",
        "api_key": config.value.get("api_key"),
        "api_secret": config.value.get("api_secret"),
        "from_numbers": config.value.get("from_numbers", [])
    }
The configuration will be stored in the database under the TELEPHONY_CONFIGURATION key in the organization_configuration table and managed through the web interface.

Audio Format Considerations

Different providers use different audio formats:
  • Twilio: 8kHz μ-law (MULAW) encoded in Base64
  • Vonage: 16kHz Linear PCM as binary frames
Your provider may differ, so ensure proper audio format conversion in your WebSocket handler and configure the audio pipeline accordingly.

Testing

Create unit tests for your provider:
# tests/test_your_provider.py

import pytest
from api.services.telephony.providers.your_provider import YourProvider

@pytest.mark.asyncio
async def test_validate_config():
    config = {
        "api_key": "test_key",
        "api_secret": "test_secret",
        "from_numbers": ["+1234567890"]
    }
    provider = YourProvider(config)
    assert provider.validate_config() is True

Best Practices

  1. Error Handling: Implement robust error handling with meaningful messages
  2. Logging: Use loguru.logger for consistent logging
  3. Async Operations: All I/O operations should be async
  4. Configuration Validation: Validate config on initialization
  5. Security: Always verify webhook signatures

Reference Implementations

See these provider implementations for complete examples:
  • Twilio: api/services/telephony/providers/twilio_provider.py - Basic authentication, XML (TwiML) responses
  • Vonage: api/services/telephony/providers/vonage_provider.py - JWT authentication, JSON (NCCO) responses
Other providers like Plivo, Telnyx, or custom SIP providers can be implemented following the same pattern. These are not included out-of-the-box but can be easily added by implementing the TelephonyProvider interface.

UI Implementation Guide

To integrate your new provider into the frontend, you’ll need to update the configuration form and the workflow header.

1. Update Configuration Page

Modify src/app/configure-telephony/page.tsx to include your provider’s form fields. A. Update Interface Add your provider’s specific configuration fields to the TelephonyConfigForm interface:
interface TelephonyConfigForm {
  provider: string;
  // ... existing fields
  
  // Your Provider Fields
  your_provider_api_key?: string;
  your_provider_secret?: string;
}
B. Add to Dropdown Add your provider to the Select component options:
<SelectContent>
  <SelectItem value="twilio">Twilio</SelectItem>
  <SelectItem value="vonage">Vonage</SelectItem>
  <SelectItem value="your_provider">Your Provider</SelectItem>
</SelectContent>
C. Add Form Fields Render your provider’s fields conditionally:
{selectedProvider === "your_provider" && (
  <>
    <div className="space-y-2">
      <Label htmlFor="your_provider_api_key">API Key</Label>
      <Input
        id="your_provider_api_key"
        {...register("your_provider_api_key", { 
          required: selectedProvider === "your_provider" 
        })}
      />
    </div>
    {/* Add other fields similarly */}
  </>
)}
D. Handle Submission Update the onSubmit function to format the request correctly:
// Inside onSubmit function
if (data.provider === "your_provider") {
  requestBody = {
    provider: "your_provider",
    api_key: data.your_provider_api_key,
    // ... other fields
  };
}

2. Enable Call Button

Update src/app/workflow/[workflowId]/components/WorkflowHeader.tsx to enable the “Phone Call” button when your provider is configured.
// In handlePhoneCallClick function
if (
  configResponse.error || 
  (!configResponse.data?.twilio && 
   !configResponse.data?.vonage && 
   !configResponse.data?.your_provider) // Add this check
) {
    setConfigureDialogOpen(true);
    return;
}

3. Update API Client

After updating the backend and frontend, regenerate the API client to ensure types are synced:
npm run generate-client