mirror of
https://github.com/dograh-hq/dograh.git
synced 2026-06-07 07:55:16 +02:00
* feat: add api trigger node for agent runs * feat: add webhook node * Execute webhook nodes post workflow run * Add hint to go to API keys |
||
|---|---|---|
| .. | ||
| providers | ||
| __init__.py | ||
| ari_client.py | ||
| ari_client_manager.py | ||
| ari_client_singleton.py | ||
| ari_manager.py | ||
| ari_manager_connection.py | ||
| base.py | ||
| factory.py | ||
| README.md | ||
| stasis_event_protocol.py | ||
| stasis_rtp_client.py | ||
| stasis_rtp_connection.py | ||
| stasis_rtp_serializer.py | ||
| stasis_rtp_transport.py | ||
| test_asyncari_ping.py | ||
| test_real_ping.py | ||
| worker_event_subscriber.py | ||
Telephony Provider Implementation
This module implements the telephony provider abstraction for Dograh AI. For user-facing documentation, see the Mintlify docs.
Architecture
Business Logic → TelephonyProvider (Interface) → Concrete Provider (Twilio, Vonage, etc.)
Developer Quick Reference
Using the Provider in Code
from api.services.telephony.factory import get_telephony_provider
# Get provider based on organization config
provider = await get_telephony_provider(organization_id)
# Initiate a call
result = await provider.initiate_call(
to_number="+1987654321",
webhook_url="https://your-app.com/webhook",
workflow_run_id=123
)
File Structure
telephony/
├── __init__.py
├── base.py # Abstract TelephonyProvider interface
├── factory.py # Provider creation and config loading
├── providers/
│ ├── __init__.py
│ ├── twilio_provider.py # Twilio implementation
│ └── vonage_provider.py # Vonage implementation
├── twilio.py # Legacy (removed, use factory instead)
└── README.md # This file
Implementing a New Provider
See the Custom Provider Guide in the documentation for detailed implementation instructions.
Quick checklist:
- Create
providers/your_provider.pyimplementingTelephonyProvider - Update
factory.pyto include your provider - Write unit tests
- Update documentation
Key Interfaces
class TelephonyProvider(ABC):
@abstractmethod
async def initiate_call(self, to_number: str, webhook_url: str, workflow_run_id: Optional[int] = None, **kwargs: Any) -> Dict[str, Any]
@abstractmethod
async def get_call_status(self, call_id: str) -> Dict[str, Any]
@abstractmethod
async def get_available_phone_numbers(self) -> List[str]
@abstractmethod
def validate_config(self) -> bool
@abstractmethod
async def verify_webhook_signature(self, url: str, params: Dict[str, Any], signature: str) -> bool
@abstractmethod
async def get_webhook_response(self, workflow_id: int, user_id: int, workflow_run_id: int) -> str
Configuration Loading
The factory.py loads configuration from the database:
Both Saas and OSS Modes: Database configuration via UI
# Loaded from organization_configuration table
key: "TELEPHONY_CONFIGURATION"
value: {
"provider": "twilio", # or "vonage"
"account_sid": "xxx", # for Twilio
"auth_token": "xxx", # for Twilio
"application_id": "xxx", # for Vonage
"private_key": "xxx", # for Vonage
"from_numbers": [...]
}
Testing
Unit Testing with Mock Provider
class MockProvider(TelephonyProvider):
async def initiate_call(self, to_number, webhook_url, **kwargs):
return {"call_id": "mock_123", "status": "initiated"}
async def get_call_status(self, call_id):
return {"call_id": call_id, "status": "completed"}
# Implement other required methods...
# In tests
@patch('api.services.telephony.factory.get_telephony_provider')
async def test_call_initiation(mock_get_provider):
mock_get_provider.return_value = MockProvider()
# Test your business logic
Integration Testing
Run against actual providers in development:
-
Configure your provider through the UI:
- Navigate to Settings → Integrations → Telephony
- Select your provider (Twilio or Vonage)
- Enter test credentials
- Save configuration
-
Run integration tests:
pytest tests/integration/test_telephony.py
Migration Notes
From Direct TwilioService Usage
Old code:
from api.services.telephony.twilio import TwilioService
service = TwilioService(org_id)
await service.initiate_call(...)
New code:
from api.services.telephony.factory import get_telephony_provider
provider = await get_telephony_provider(org_id)
await provider.initiate_call(...)
Backward Compatibility
- Old
/api/v1/twilio/*endpoints still work (redirect to/api/v1/telephony/*) TwilioServiceclass remains for legacy code- Database configuration key
TWILIO_CONFIGURATIONunchanged
Common Issues
- Import Error: Always import from
factory, not directly from providers - Config Not Found: Check database configuration via UI
- Signature Verification: Ensure auth tokens match between provider and config
- WebSocket Issues: Verify audio format compatibility (MULAW for Twilio)