Telephony Integration

Overview#

Lemu AI's telephony integration system provides a unified interface for connecting with various telephony providers. This abstraction layer allows you to easily switch between providers without changing your application logic.

Supported Providers#

Architecture#

The telephony integration system uses a provider abstraction pattern that ensures consistency across different services:

# All providers implement this interface
class TelephonyProvider(ABC):
    async def initiate_call(to_number: str, webhook_url: str, workflow_run_id: Optional[int] = None, **kwargs) -> CallInitiationResult
    async def get_call_status(call_id: str) -> Dict[str, Any]
    async def get_available_phone_numbers() -> List[str]
    def validate_config() -> bool
    async def verify_webhook_signature(url: str, params: Dict, signature: str) -> bool
    async def get_webhook_response(workflow_id: int, user_id: int, workflow_run_id: int) -> str
    async def get_call_cost(call_id: str) -> Dict[str, Any]
    def parse_status_callback(data: Dict[str, Any]) -> Dict[str, Any]
    async def handle_websocket(websocket: WebSocket, workflow_id: int, user_id: int, workflow_run_id: int) -> None

Configuration#

Lemu AI uses database configuration for all telephony providers. Configure providers through the web interface:

1. Go to Configuration in the sidebar → Telephony tab
2. Select your provider (Twilio or Vonage)
3. Watch the provider-specific video tutorial for setup guidance
4. Enter your credentials
5. Save configuration
6. Test with a phone call

Common Features#

The telephony integration in Lemu AI supports:

• Outbound Calls: Initiate calls to any phone number
• Call Status Tracking: Monitor call lifecycle events (initiated, ringing, answered, completed, failed)
• WebSocket Streaming: Real-time audio streaming for voice agents
• Webhook Authentication: Secure webhook signature verification

Code Usage#

Here's how to use the telephony provider in your code:

from api.services.telephony.factory import get_telephony_provider
 
# Get provider based on organization configuration
provider = await get_telephony_provider(organization_id)
 
# Initiate a call
result = await provider.initiate_call(
    to_number="+1234567890",
    webhook_url="https://your-domain.com/webhook",
    workflow_run_id=123
)
 
# Check call status
status = await provider.get_call_status(result.call_id)
 
# Get call cost after completion
cost_info = await provider.get_call_cost(result.call_id)

API Endpoints#

The telephony system exposes these unified endpoints:

Implementation Status#

• Twilio: ✅ Fully implemented and tested
• Vonage: ✅ Fully implemented with 16kHz audio support
• Custom Providers: The abstraction layer supports adding new providers by implementing the TelephonyProvider interface

Troubleshooting#

Next Steps#

• Set up your first telephony provider
• Build a custom provider integration
• Configure webhooks and callbacks