Custom Telephony Provider

Overview#

Lemu 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#

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
  };
}

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