Custom Telephony Provider

Prerequisites#

This guide assumes familiarity with two libraries the provider interface is built on:

Pipecat — the real-time voice/multimodal pipeline framework Menace Voice's call pipeline is built on. Its FastAPIWebsocketTransport is what streams audio between the telephony provider and Menace Voice's pipeline — you'll be constructing one of these in your transport factory. If you haven't built a Pipecat transport or frame serializer before, skim their transports guide first.
Pydantic — used for the request/response schemas that define a provider's stored credentials (config.py below). If you're new to Pydantic, its models documentation covers everything used here (BaseModel, Field, Literal discriminators).

You don't need deep expertise in either — enough to read a BaseModel and a Pipecat transport class is sufficient to follow this guide.

Overview#

A telephony provider is implemented as a self-registering package under api/services/telephony/providers/<name>/. The package contributes everything Menace Voice needs to wire the provider in — the provider class, transport factory, audio config, request/response schemas, optional HTTP routes, and the form metadata used to render its configuration UI — through a single ProviderSpec registered at import time.

Adding a new provider should not require touching the factory, the audio config, the API routes module, the run-pipeline module, or the frontend. The only edits outside the provider folder are:

One import line in api/services/telephony/providers/__init__.py
One import line in api/schemas/telephony_config.py to add the request/response classes to the TelephonyConfigRequest discriminated union

Provider Package Layout#

api/services/telephony/providers/your_provider/├── __init__.py        # Builds and registers ProviderSpec├── config.py          # Pydantic Request/Response schemas├── provider.py        # TelephonyProvider subclass├── transport.py       # Pipecat WebSocket transport factory├── serializers.py     # Frame serializer (usually re-exports from pipecat)├── routes.py          # (optional) HTTP webhook/callback handlers└── strategies.py      # (optional) Transfer/hangup strategies

Three files are required (__init__.py, config.py, provider.py, transport.py). The rest are optional and are discovered automatically when present:

routes.py — if the module exists and exports router: APIRouter, the routes module is imported lazily and mounted under /api/v1/telephony by api.routes.telephony via importlib. Providers that only stream over WebSocket (e.g. ARI) can omit it.
strategies.py — used by transports that need provider-specific call transfer/hangup logic in the frame serializer (e.g. Twilio Conference transfers).
serializers.py — typically a re-export from pipecat. Keep the file even when it's a one-line re-export so transport code imports from .serializers, giving you an obvious place to drop a custom subclass later.

The `TelephonyProvider` Interface#

Subclass TelephonyProvider in provider.py:

from api.services.telephony.base import (    CallInitiationResult,    NormalizedInboundData,    ProviderSyncResult,    TelephonyProvider,)class YourProvider(TelephonyProvider):    PROVIDER_NAME = "your_provider"    WEBHOOK_ENDPOINT = "your-provider-xml"  # path under /api/v1/telephony    def __init__(self, config: dict):        self.api_key = config.get("api_key")        self.from_numbers = config.get("from_numbers", [])    # ---------- outbound ----------    async def initiate_call(self, to_number, webhook_url, workflow_run_id=None,                            from_number=None, **kwargs) -> CallInitiationResult: ...    async def get_call_status(self, call_id) -> dict: ...    async def get_call_cost(self, call_id) -> dict: ...    async def get_available_phone_numbers(self) -> list[str]: ...    def validate_config(self) -> bool: ...    # ---------- webhooks ----------    async def verify_webhook_signature(self, url, params, signature) -> bool: ...    async def get_webhook_response(self, workflow_id, organization_id, workflow_run_id) -> str: ...    def parse_status_callback(self, data: dict) -> dict: ...    # ---------- websocket ----------    async def handle_websocket(self, websocket, workflow_id, organization_id, workflow_run_id): ...    # ---------- inbound ----------    @classmethod    def can_handle_webhook(cls, webhook_data, headers) -> bool: ...    @staticmethod    def parse_inbound_webhook(webhook_data) -> NormalizedInboundData: ...    @staticmethod    def validate_account_id(config_data, webhook_account_id) -> bool: ...    def normalize_phone_number(self, phone_number: str) -> str: ...    async def verify_inbound_signature(self, url, webhook_data, headers, body="") -> bool: ...    async def start_inbound_stream(self, *, websocket_url, workflow_run_id,                                   normalized_data, backend_endpoint): ...    @staticmethod    def generate_error_response(error_type, message) -> tuple: ...    # ---------- transfers ----------    async def transfer_call(self, destination, transfer_id, conference_name,                            timeout=30, **kwargs) -> dict: ...    def supports_transfers(self) -> bool: ...    # ---------- optional ----------    async def configure_inbound(self, address, webhook_url) -> ProviderSyncResult:        # Default returns ok=True — implement only if your provider supports        # programmatic webhook configuration (e.g. binding a number to a URL        # via API). Used to point inbound numbers at /api/v1/telephony/inbound/run.        return ProviderSyncResult(ok=True)

See api/services/telephony/base.py for the full docstrings on each method.

Warning

get_webhook_response and handle_websocket receive an organization_id, not a user id. It is the tenant that every workflow and workflow-run lookup must be scoped by — pass it straight through to run_pipeline_telephony. The workflow owner is deliberately not handed to providers; read it off the workflow row if you need it for attribution.

Implementation Guide#

1. Configuration schemas#

Define Pydantic models for the credential payload. The provider Literal discriminator is what makes the schemas dispatch correctly through the registry's discriminated union.

# providers/your_provider/config.pyfrom typing import List, Literalfrom pydantic import BaseModel, Fieldclass YourProviderConfigurationRequest(BaseModel):    provider: Literal["your_provider"] = Field(default="your_provider")    api_key: str = Field(..., description="Your Provider API key")    api_secret: str = Field(..., description="Your Provider API secret")    from_numbers: List[str] = Field(default_factory=list)class YourProviderConfigurationResponse(BaseModel):    provider: Literal["your_provider"] = Field(default="your_provider")    api_key: str        # masked when returned    api_secret: str     # masked when returned    from_numbers: List[str]

2. Transport factory#

Build the Pipecat FastAPIWebsocketTransport for accepted WebSockets. Always load credentials through load_credentials_for_transport so the right config row is picked when the workflow run carries a telephony_configuration_id (multi-config orgs).

# providers/your_provider/transport.pyfrom fastapi import WebSocketfrom api.services.pipecat.audio_config import AudioConfigfrom api.services.pipecat.audio_mixer import build_audio_out_mixerfrom api.services.telephony.factory import load_credentials_for_transportfrom pipecat.transports.websocket.fastapi import (    FastAPIWebsocketParams,    FastAPIWebsocketTransport,)from .serializers import YourProviderFrameSerializerasync def create_transport(    websocket: WebSocket,    workflow_run_id: int,    audio_config: AudioConfig,    organization_id: int,    *,    vad_config: dict | None = None,    ambient_noise_config: dict | None = None,    telephony_configuration_id: int | None = None,    # provider-specific kwargs (forwarded by run_pipeline_telephony as **transport_kwargs)    stream_id: str,    call_id: str,):    config = await load_credentials_for_transport(        organization_id, telephony_configuration_id,        expected_provider="your_provider",    )    serializer = YourProviderFrameSerializer(        stream_id=stream_id,        call_id=call_id,        api_key=config["api_key"],    )    mixer = await build_audio_out_mixer(        audio_config.transport_out_sample_rate, ambient_noise_config    )    return FastAPIWebsocketTransport(        websocket=websocket,        params=FastAPIWebsocketParams(            audio_in_enabled=True,            audio_out_enabled=True,            audio_in_sample_rate=audio_config.transport_in_sample_rate,            audio_out_sample_rate=audio_config.transport_out_sample_rate,            audio_out_mixer=mixer,            serializer=serializer,        ),    )

3. Routes (optional)#

If your provider POSTs webhooks to Menace Voice (answer URL, status callbacks, hangup callbacks), expose them through a module-level router. The routes are auto-mounted under /api/v1/telephony.

# providers/your_provider/routes.pyfrom fastapi import APIRouter, Requestfrom api.services.telephony.status_processor import (    StatusCallbackRequest,    _process_status_update,)router = APIRouter()@router.post("/your-provider/status-callback/{workflow_run_id}")async def status_callback(workflow_run_id: int, request: Request):    ...

Routes are loaded lazily via importlib from api.routes.telephony._mount_provider_routers, so your route module can freely import other backend services without creating import cycles at provider-class load time.

4. Register the `ProviderSpec`#

The package's __init__.py is where everything comes together:

# providers/your_provider/__init__.pyfrom typing import Any, Dictfrom api.services.pipecat.audio_config import AudioConfigfrom api.services.telephony.registry import (    ProviderSpec,    ProviderUIField,    ProviderUIMetadata,    register,)from .config import YourProviderConfigurationRequest, YourProviderConfigurationResponsefrom .provider import YourProviderfrom .transport import create_transportdef _config_loader(value: Dict[str, Any]) -> Dict[str, Any]:    """Normalize the stored credentials dict into the constructor shape."""    return {        "provider": "your_provider",        "api_key": value.get("api_key"),        "api_secret": value.get("api_secret"),        "from_numbers": value.get("from_numbers", []),    }_AUDIO_CONFIG = AudioConfig(    transport_in_sample_rate=8000,    transport_out_sample_rate=8000,    vad_sample_rate=8000,    pipeline_sample_rate=8000,    buffer_size_seconds=5.0,)_UI_METADATA = ProviderUIMetadata(    display_name="Your Provider",    docs_url="https://docs.your-provider.com",    fields=[        ProviderUIField(name="api_key",    label="API Key",    type="text",     sensitive=True),        ProviderUIField(name="api_secret", label="API Secret", type="password", sensitive=True),        ProviderUIField(            name="from_numbers", label="Phone Numbers", type="string-array",            description="E.164-formatted phone numbers used for outbound calls",        ),    ],)SPEC = ProviderSpec(    name="your_provider",    provider_cls=YourProvider,    config_loader=_config_loader,    transport_factory=create_transport,    audio_config=_AUDIO_CONFIG,    config_request_cls=YourProviderConfigurationRequest,    config_response_cls=YourProviderConfigurationResponse,    ui_metadata=_UI_METADATA,    # Credential field that uniquely identifies the provider account.    # Used to disambiguate inbound webhooks across multiple configs of the    # same provider. Empty string for providers without an account-id concept.    account_id_credential_field="api_key",)register(SPEC)

ProviderSpec covers everything downstream code needs:

Field	Used by
`name`	Stored as the discriminator on every `TelephonyConfiguration` row and as the `WorkflowRunMode` value
`provider_cls`	`factory.get_default_telephony_provider`, `get_telephony_provider_by_id`, `get_telephony_provider_for_run`
`config_loader`	`factory._normalize_with_phone_numbers` (replaces the old if/elif chain)
`transport_factory`	`run_pipeline_telephony`
`audio_config`	`create_audio_config()` and `run_pipeline_telephony`
`config_request_cls` / `config_response_cls`	`TelephonyConfigRequest` discriminated union
`ui_metadata`	`GET /api/v1/organizations/telephony-providers/metadata` (drives the form UI) and the `_sensitive_fields` masking helper
`account_id_credential_field`	Inbound webhook routing across multiple configs of the same provider

5. Wire the package into the registry import chain#

Add one import line to api/services/telephony/providers/__init__.py:

from api.services.telephony.providers import (  # noqa: F401  -- side effects    ari,    cloudonix,    plivo,    telnyx,    twilio,    vobiz,    vonage,    your_provider,        # ← add this)

6. Add to the discriminated union#

Add one import block to api/schemas/telephony_config.py so the request/response classes participate in the TelephonyConfigRequest union and the TelephonyConfigurationResponse shape:

from api.services.telephony.providers.your_provider.config import (    YourProviderConfigurationRequest,    YourProviderConfigurationResponse,)TelephonyConfigRequest = Annotated[    Union[        # ...existing entries...        YourProviderConfigurationRequest,    ],    Field(discriminator="provider"),]class TelephonyConfigurationResponse(BaseModel):    # ...existing entries...    your_provider: Optional[YourProviderConfigurationResponse] = None

That's it for backend wiring.

Frontend#

The configuration form is metadata-driven. The UI calls GET /api/v1/organizations/telephony-providers/metadata, gets back the list of providers and their ProviderUIField definitions, and renders each form generically. No per-provider frontend code is needed — your ProviderUIMetadata declaration is what drives the form.

If you add a new field type that the existing renderer doesn't support (e.g. a file upload), extend the renderer in ui/src/app/(authenticated)/telephony-configurations/. The supported ProviderUIField.type values today are text, password, textarea, string-array, and number.

Audio Format Considerations#

Each provider declares its wire format through its AudioConfig. Common shapes:

Twilio / Plivo: 8 kHz μ-law, base64-encoded JSON frames
Vonage: 16 kHz Linear PCM as binary frames
Asterisk ARI: 8 kHz Linear PCM via externalMedia

The pipeline sample rate is capped at 16 kHz to satisfy VAD, which only accepts 8000 Hz or 16000 Hz; transports handle resampling between the wire format and the pipeline's internal rate.

Testing#

# api/tests/telephony/test_your_provider.pyimport pytestfrom api.services.telephony.providers.your_provider import YourProvider@pytest.mark.asyncioasync def test_validate_config():    provider = YourProvider({        "api_key": "test_key",        "api_secret": "test_secret",        "from_numbers": ["+1234567890"],    })    assert provider.validate_config() is True

For end-to-end testing, save your provider through the telephony-configurations UI and trigger a test call from a workflow.

Best Practices#

Trust the registry — never import another provider's class directly; resolve through the factory helpers (get_default_telephony_provider, get_telephony_provider_by_id, etc.).
Sensitive fields — mark every credential field sensitive=True in ProviderUIMetadata. The save endpoint masks these on read and preserves the original when the client re-submits a masked value.
Inbound signature verification — always validate inbound webhook signatures in verify_inbound_signature, and fail closed. If your provider signs every webhook (most do), a missing signature header means the request did not come from the provider: return False, exactly as providers/twilio/ and providers/plivo/ do. Never return True as a placeholder — these handlers are reachable by anyone who can guess a workflow_run_id.
Transports load credentials lazily — call load_credentials_for_transport with the telephony_configuration_id from the workflow run. Don't read the org's default config from transport.py.
Logging — use loguru.logger.

Reference Implementations#

Provider	Notable for
`providers/twilio/`	Full-featured: outbound, inbound, conference transfers, status callbacks, custom strategies
`providers/plivo/`	Recently-added reference; mirrors Twilio's shape with multi-callback signatures
`providers/vonage/`	JWT auth, 16 kHz Linear PCM, NCCO responses
`providers/cloudonix/`	SIP-based, custom call strategies
`providers/telnyx/`	Call-control style: REST-driven inbound answer flow rather than markup response
`providers/ari/`	Minimal example — no `routes.py`, no inbound webhook verification, WebSocket-only

Note

Use ARI as the smallest viable example when your provider doesn't expose HTTP webhooks, and Twilio as the reference when it does.

Prerequisites#

Overview#

Provider Package Layout#

The TelephonyProvider Interface#

Implementation Guide#

1. Configuration schemas#

2. Transport factory#

3. Routes (optional)#

4. Register the ProviderSpec#

5. Wire the package into the registry import chain#

6. Add to the discriminated union#

Frontend#

Audio Format Considerations#

Testing#

Best Practices#

Reference Implementations#

The `TelephonyProvider` Interface#

4. Register the `ProviderSpec`#