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.
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
- Error Handling: Implement robust error handling with meaningful messages
- Logging: Use
loguru.logger for consistent logging
- Async Operations: All I/O operations should be async
- Configuration Validation: Validate config on initialization
- 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
};
}
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: