feat: Enable telephony for OSS (#21)

* fix: fix tooltip bug

* feat: add Twilio with CloudFlare configuration

* chore: update Tella Video

Author: a6kme

api/enums.py

@@ -62,7 +62,7 @@ class OrganizationConfigurationKey(Enum):
     DISPOSITION_CODE_MAPPING = "DISPOSITION_CODE_MAPPING"
     DISPOSITION_MESSAGE_TEMPLATE = "DISPOSITION_MESSAGE_TEMPLATE"
     CONCURRENT_CALL_LIMIT = "CONCURRENT_CALL_LIMIT"
-    TWILIO_PHONE_NUMBERS = "TWILIO_PHONE_NUMBERS"
+    TWILIO_CONFIGURATION = "TWILIO_CONFIGURATION"
 
 
 class WorkflowStatus(Enum):

api/routes/campaign.py

@@ -168,10 +168,10 @@ async def start_campaign(
     user: UserModel = Depends(get_user),
 ) -> CampaignResponse:
     """Start campaign execution"""
-    # Check if organization has TWILIO_PHONE_NUMBERS configured
+    # Check if organization has TWILIO_CONFIGURATION configured
     twilio_config = await db_client.get_configuration(
         user.selected_organization_id,
-        OrganizationConfigurationKey.TWILIO_PHONE_NUMBERS.value,
+        OrganizationConfigurationKey.TWILIO_CONFIGURATION.value,
     )
 
     if (
@@ -280,10 +280,10 @@ async def resume_campaign(
     user: UserModel = Depends(get_user),
 ) -> CampaignResponse:
     """Resume a paused campaign"""
-    # Check if organization has TWILIO_PHONE_NUMBERS configured
+    # Check if organization has TWILIO_CONFIGURATION configured
     twilio_config = await db_client.get_configuration(
         user.selected_organization_id,
-        OrganizationConfigurationKey.TWILIO_PHONE_NUMBERS.value,
+        OrganizationConfigurationKey.TWILIO_CONFIGURATION.value,
     )
 
     if (

api/routes/main.py

@@ -4,6 +4,7 @@
 from api.routes.campaign import router as campaign_router
 from api.routes.integration import router as integration_router
 from api.routes.looptalk import router as looptalk_router
+from api.routes.organization import router as organization_router
 from api.routes.organization_usage import router as organization_usage_router
 from api.routes.reports import router as reports_router
 from api.routes.rtc_offer import router as rtc_offer_router
@@ -27,6 +28,7 @@
 router.include_router(user_router)
 router.include_router(campaign_router)
 router.include_router(integration_router)
+router.include_router(organization_router)
 router.include_router(s3_router)
 router.include_router(service_keys_router)
 router.include_router(looptalk_router)

api/routes/organization.py

@@ -0,0 +1,85 @@
+from fastapi import APIRouter, Depends, HTTPException
+
+from api.db import db_client
+from api.db.models import UserModel
+from api.enums import OrganizationConfigurationKey
+from api.schemas.telephony_config import (
+    TelephonyConfigurationResponse,
+    TwilioConfigurationRequest,
+    TwilioConfigurationResponse,
+)
+from api.services.auth.depends import get_user
+from api.services.configuration.masking import is_mask_of, mask_key
+
+router = APIRouter(prefix="/organizations", tags=["organizations"])
+
+
+@router.get("/telephony-config", response_model=TelephonyConfigurationResponse)
+async def get_telephony_configuration(user: UserModel = Depends(get_user)):
+    """Get telephony configuration for the user's organization with masked sensitive fields."""
+    if not user.selected_organization_id:
+        raise HTTPException(status_code=400, detail="No organization selected")
+
+    config = await db_client.get_configuration(
+        user.selected_organization_id,
+        OrganizationConfigurationKey.TWILIO_CONFIGURATION.value,
+    )
+
+    if not config or not config.value:
+        return TelephonyConfigurationResponse(twilio=None)
+
+    # Mask sensitive fields (account_sid and auth_token) before returning
+    account_sid = config.value.get("account_sid", "")
+    auth_token = config.value.get("auth_token", "")
+
+    return TelephonyConfigurationResponse(
+        twilio=TwilioConfigurationResponse(
+            provider="twilio",
+            account_sid=mask_key(account_sid) if account_sid else "",
+            auth_token=mask_key(auth_token) if auth_token else "",
+            from_numbers=config.value.get("from_numbers", []),
+        )
+    )
+
+
+@router.post("/telephony-config")
+async def save_telephony_configuration(
+    request: TwilioConfigurationRequest, user: UserModel = Depends(get_user)
+):
+    """Save telephony configuration for the user's organization."""
+    if not user.selected_organization_id:
+        raise HTTPException(status_code=400, detail="No organization selected")
+
+    # Fetch existing configuration to handle masked values
+    existing_config = await db_client.get_configuration(
+        user.selected_organization_id,
+        OrganizationConfigurationKey.TWILIO_CONFIGURATION.value,
+    )
+
+    # Build new configuration
+    config_value = {
+        "provider": request.provider,
+        "account_sid": request.account_sid,
+        "auth_token": request.auth_token,
+        "from_numbers": request.from_numbers,
+    }
+
+    # If incoming values are masked (same as stored masked value), keep the original
+    if existing_config and existing_config.value:
+        # Check if account_sid is unchanged (masked value matches)
+        stored_account_sid = existing_config.value.get("account_sid", "")
+        if stored_account_sid and is_mask_of(request.account_sid, stored_account_sid):
+            config_value["account_sid"] = stored_account_sid  # Keep original
+
+        # Check if auth_token is unchanged (masked value matches)
+        stored_auth_token = existing_config.value.get("auth_token", "")
+        if stored_auth_token and is_mask_of(request.auth_token, stored_auth_token):
+            config_value["auth_token"] = stored_auth_token  # Keep original
+
+    await db_client.upsert_configuration(
+        user.selected_organization_id,
+        OrganizationConfigurationKey.TWILIO_CONFIGURATION.value,
+        config_value,
+    )
+
+    return {"message": "Telephony configuration saved successfully"}

api/routes/twilio.py

@@ -5,7 +5,6 @@
 
 from fastapi import APIRouter, Depends, Form, Header, HTTPException, Request, WebSocket
 from loguru import logger
-from pipecat.utils.context import set_current_run_id
 from pydantic import BaseModel
 from starlette.responses import HTMLResponse
 
@@ -19,6 +18,7 @@
 )
 from api.services.pipecat.run_pipeline import run_pipeline_twilio
 from api.services.telephony.twilio import TwilioService
+from pipecat.utils.context import set_current_run_id
 
 router = APIRouter(prefix="/twilio")
 
@@ -45,20 +45,16 @@ class TwilioStatusCallbackRequest(BaseModel):
 async def initiate_call(
     request: InitiateCallRequest, user: UserModel = Depends(get_user)
 ):
-    # Check if organization has TWILIO_PHONE_NUMBERS configured
+    # Check if organization has TWILIO_CONFIGURATION configured
     twilio_config = await db_client.get_configuration(
         user.selected_organization_id,
-        OrganizationConfigurationKey.TWILIO_PHONE_NUMBERS.value,
+        OrganizationConfigurationKey.TWILIO_CONFIGURATION.value,
     )
 
-    if (
-        not twilio_config
-        or not twilio_config.value
-        or not twilio_config.value.get("value")
-    ):
+    if not twilio_config or not twilio_config.value:
         raise HTTPException(
-            status_code=401,
-            detail="Your organisation is not allowed to make phone call. Contact founders@dograh.com for further support.",
+            status_code=400,
+            detail="telephony_not_configured",  # Special error code
         )
 
     user_configuration = await db_client.get_user_configurations(user.id)
@@ -84,15 +80,16 @@ async def initiate_call(
         workflow_run_name = workflow_run.name
 
     if user_configuration.test_phone_number:
-        await TwilioService().initiate_call(
+        twilio_service = TwilioService(user.selected_organization_id)
+        await twilio_service.initiate_call(
             to_number=user_configuration.test_phone_number,
             url_args={
                 "workflow_id": request.workflow_id,
                 "user_id": user.id,
                 "workflow_run_id": workflow_run_id,
+                "organization_id": user.selected_organization_id,
             },
             workflow_run_id=workflow_run_id,
-            organization_id=user.selected_organization_id,
         )
         return {
             "message": f"Call initiated successfully with run name {workflow_run_name}"
@@ -102,8 +99,10 @@ async def initiate_call(
 
 
 @router.post("/twiml", include_in_schema=False)
-async def start_call(workflow_id: int, user_id: int, workflow_run_id: int):
-    twiml_content = await TwilioService().get_start_call_twiml(
+async def start_call(
+    workflow_id: int, user_id: int, workflow_run_id: int, organization_id: int
+):
+    twiml_content = await TwilioService(organization_id).get_start_call_twiml(
         workflow_id, user_id, workflow_run_id
     )
     return HTMLResponse(content=twiml_content, media_type="application/xml")

api/schemas/telephony_config.py

@@ -0,0 +1,29 @@
+from typing import List
+
+from pydantic import BaseModel, Field
+
+
+class TwilioConfigurationRequest(BaseModel):
+    """Request schema for Twilio configuration."""
+
+    provider: str = Field(default="twilio")
+    account_sid: str = Field(..., description="Twilio Account SID")
+    auth_token: str = Field(..., description="Twilio Auth Token")
+    from_numbers: List[str] = Field(
+        ..., min_length=1, description="List of Twilio phone numbers"
+    )
+
+
+class TwilioConfigurationResponse(BaseModel):
+    """Response schema for Twilio configuration with masked sensitive fields."""
+
+    provider: str
+    account_sid: str  # Masked (e.g., "****************def0")
+    auth_token: str  # Masked (e.g., "****************abc1")
+    from_numbers: List[str]
+
+
+class TelephonyConfigurationResponse(BaseModel):
+    """Top-level telephony configuration response."""
+
+    twilio: TwilioConfigurationResponse | None = None

api/services/campaign/call_dispatcher.py

@@ -16,15 +16,11 @@ class CampaignCallDispatcher:
     """Manages rate-limited and concurrent-limited call dispatching"""
 
     def __init__(self):
-        self._twilio_service = None
         self.default_concurrent_limit = 20
 
-    @property
-    def twilio_service(self):
-        """Lazy initialization of TwilioService"""
-        if self._twilio_service is None:
-            self._twilio_service = TwilioService()
-        return self._twilio_service
+    def get_twilio_service(self, organization_id: int) -> TwilioService:
+        """Get TwilioService instance for specific organization"""
+        return TwilioService(organization_id)
 
     async def get_org_concurrent_limit(self, organization_id: int) -> int:
         """Get the concurrent call limit for an organization."""
@@ -225,15 +221,16 @@ async def dispatch_call(
 
         # Initiate call via Twilio
         try:
-            call_result = await self.twilio_service.initiate_call(
+            twilio_service = self.get_twilio_service(campaign.organization_id)
+            call_result = await twilio_service.initiate_call(
                 to_number=phone_number,
                 workflow_run_id=workflow_run.id,
-                organization_id=campaign.organization_id,
                 url_args={
                     "workflow_id": campaign.workflow_id,
                     "user_id": campaign.created_by,
                     "workflow_run_id": workflow_run.id,
                     "campaign_id": campaign.id,
+                    "organization_id": campaign.organization_id,
                 },
             )
 

api/services/pipecat/run_pipeline.py

@@ -88,12 +88,13 @@ async def run_pipeline_twilio(
     # Create audio configuration for Twilio
     audio_config = create_audio_config(WorkflowRunMode.TWILIO.value)
 
-    transport = create_twilio_transport(
+    transport = await create_twilio_transport(
         websocket_client,
         stream_sid,
         call_sid,
         workflow_run_id,
         audio_config,
+        workflow.organization_id,
         vad_config,
         ambient_noise_config,
     )

api/services/pipecat/transport_setup.py

@@ -3,6 +3,8 @@
 from fastapi import WebSocket
 
 from api.constants import APP_ROOT_DIR, ENABLE_RNNOISE, ENABLE_SMART_TURN
+from api.db import db_client
+from api.enums import OrganizationConfigurationKey
 from api.services.looptalk.internal_transport import InternalTransport
 from api.services.pipecat.audio_config import AudioConfig
 from api.services.smart_turn.websocket_smart_turn import (
@@ -69,23 +71,43 @@ def create_turn_analyzer(workflow_run_id: int, audio_config: AudioConfig):
     return None
 
 
-def create_twilio_transport(
+async def create_twilio_transport(
     websocket_client: WebSocket,
     stream_sid: str,
     call_sid: 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 Twilio connections"""
+
+    # Fetch Twilio credentials from organization config
+    config = await db_client.get_configuration(
+        organization_id, OrganizationConfigurationKey.TWILIO_CONFIGURATION.value
+    )
+
+    if not config or not config.value:
+        raise ValueError(
+            f"Twilio credentials not configured for organization {organization_id}"
+        )
+
+    account_sid = config.value.get("account_sid")
+    auth_token = config.value.get("auth_token")
+
+    if not account_sid or not auth_token:
+        raise ValueError(
+            f"Incomplete Twilio configuration for organization {organization_id}"
+        )
+
     turn_analyzer = create_turn_analyzer(workflow_run_id, audio_config)
 
     serializer = TwilioFrameSerializer(
         stream_sid=stream_sid,
         call_sid=call_sid,
-        account_sid=os.environ["TWILIO_ACCOUNT_SID"],
-        auth_token=os.environ["TWILIO_AUTH_TOKEN"],
+        account_sid=account_sid,
+        auth_token=auth_token,
     )
 
     return FastAPIWebsocketTransport(