# How Dograh Handles Transport Setup for Different Telephony Providers

> Dograh simplifies transport setup for telephony providers using its provider-registry pattern. Learn how Dograh builds FastAPIWebsocketTransport for seamless integration.

- Repository: [Dograh/dograh](https://github.com/dograh-hq/dograh)
- Tags: how-to-guide
- Published: 2026-05-18

---

**Dograh abstracts telephony provider transport creation behind a provider-registry pattern where each vendor registers a `ProviderSpec` containing a `transport_factory` that builds a fully-configured `FastAPIWebsocketTransport`.**

Dograh is an open-source conversational AI platform built on Pipecat that routes real-time audio between telephony services and large language models. When handling inbound calls from carriers like Twilio, Vonage, or Telnyx, the system must initialize the correct WebSocket transport with provider-specific serializers and credentials. This article examines how Dograh's transport setup handles different telephony provider transports through a modular registry architecture that decouples vendor-specific logic from core audio pipeline code.

## The Provider Registry Architecture

Dograh implements a plugin-style registry in [`api/services/telephony/registry.py`](https://github.com/dograh-hq/dograh/blob/main/api/services/telephony/registry.py) that maps provider names to their implementation specifications. Each telephony vendor registers a **ProviderSpec** dataclass containing the factory function responsible for constructing that provider's transport layer.

### Registering Provider Specifications

Every provider package calls `register()` from its [`__init__.py`](https://github.com/dograh-hq/dograh/blob/main/__init__.py), supplying a `ProviderSpec` that records:

- The provider class (`provider_cls`)
- A `config_loader` for normalizing stored credentials
- The **`transport_factory`** async callable that instantiates the transport
- The native audio sample rate (`transport_sample_rate`)

```python

# api/services/telephony/providers/twilio/__init__.py

from .transport import create_transport
from .provider import TwilioProvider
from .config import TwilioConfigRequest, TwilioConfigResponse

register(
    ProviderSpec(
        name="twilio",
        provider_cls=TwilioProvider,
        config_loader=load_twilio_config,
        transport_factory=create_transport,          # ← key hook

        transport_sample_rate=8000,
        config_request_cls=TwilioConfigRequest,
        config_response_cls=TwilioConfigResponse,
        ui_metadata=twilio_ui_metadata,
    )
)

```

This registration pattern applies uniformly across Twilio, Vonage, Telnyx, Plivo, Vobiz, and Cloudonix providers.

## Resolving the Correct Transport Factory

When a workflow run requires a transport, `api.services.telephony.factory` resolves the appropriate configuration and retrieves the registered `transport_factory`. The factory validates that the stored credentials match the expected provider before invoking the vendor-specific builder.

```python

# factory.get_telephony_provider_for_run(...)

spec = registry.get(provider_name)                # e.g. "twilio"

transport = await spec.transport_factory(...)

```

The factory also enforces credential validation via `load_credentials_for_transport` (lines 70-95 in [`factory.py`](https://github.com/dograh-hq/dograh/blob/main/factory.py)), ensuring that a Twilio configuration cannot be mistakenly used with a Vonage transport factory.

## Implementing Provider-Specific Transport Factories

Each provider implements its own `create_transport` function in a dedicated [`transport.py`](https://github.com/dograh-hq/dograh/blob/main/transport.py) module. These factories share a common signature but encapsulate vendor-specific serialization logic and authentication parameters.

### The transport_factory Contract

The `transport_factory` is an async callable that accepts:

- `websocket`: The FastAPI `WebSocket` connection
- `workflow_run_id`: The unique run identifier
- `audio_config`: `AudioConfig` containing sample rates for transport and pipeline
- `organization_id`: The tenant identifier for credential lookup
- Provider-specific identifiers (e.g., `call_sid` for Twilio, `call_uuid` for Vonage)

### Twilio Transport Implementation

In [`api/services/telephony/providers/twilio/transport.py`](https://github.com/dograh-hq/dograh/blob/main/api/services/telephony/providers/twilio/transport.py), the `create_transport` function loads credentials, instantiates a `TwilioFrameSerializer`, and returns a configured `FastAPIWebsocketTransport`:

```python

# api/services/telephony/providers/twilio/transport.py

async def create_transport(
    websocket: WebSocket,
    workflow_run_id: int,
    audio_config: AudioConfig,
    organization_id: int,
    *,
    ambient_noise_config: dict | None = None,
    telephony_configuration_id: int | None = None,
    is_realtime: bool = False,
    stream_sid: str,
    call_sid: str,
):
    # 1️⃣ Load credentials

    config = await load_credentials_for_transport(
        organization_id, telephony_configuration_id, expected_provider="twilio"
    )
    # 2️⃣ Build serializer with provider-specific secrets

    serializer = TwilioFrameSerializer(
        stream_sid=stream_sid,
        call_sid=call_sid,
        account_sid=config["account_sid"],
        auth_token=config["auth_token"],
        transfer_strategy=TwilioConferenceStrategy(),
        hangup_strategy=TwilioHangupStrategy(),
    )
    # 3️⃣ Build optional audio mixer

    mixer = await build_audio_out_mixer(
        audio_config.transport_out_sample_rate, ambient_noise_config
    )
    # 4️⃣ Return the transport

    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,
            **realtime_param_overrides(is_realtime),
        ),
    )

```

### Vonage and Binary Transports

Vonage follows an identical pattern in [`api/services/telephony/providers/vonage/transport.py`](https://github.com/dograh-hq/dograh/blob/main/api/services/telephony/providers/vonage/transport.py) but utilizes a binary WebSocket serializer and JWT-based authentication:

```python

# api/services/telephony/providers/vonage/transport.py

config = await load_credentials_for_transport(
    organization_id, telephony_configuration_id, expected_provider="vonage"
)
serializer = VonageFrameSerializer(
    call_uuid=call_uuid,
    application_id=config["application_id"],
    private_key=config["private_key"],
    params=VonageFrameSerializer.InputParams(
        vonage_sample_rate=audio_config.transport_in_sample_rate,
        sample_rate=audio_config.pipeline_sample_rate,
    ),
)

# … FastAPIWebsocketTransport returned similarly

```

Telnyx, Plivo, Vobiz, and Cloudonix implement equivalent transport factories with their respective credential fields and frame serializers.

## Runtime Transport Initialization Flow

When a call connects, the runtime flow executes as follows:

1. The WebSocket endpoint (`/api/v1/telephony/<provider>/events/...`) extracts the provider name from the URL path.
2. `api.routes.telephony` retrieves the spec via `registry.get(name)`.
3. The endpoint passes the active `WebSocket` and run identifiers to `spec.transport_factory`.
4. The resulting `FastAPIWebsocketTransport` attaches to the `PipecatEngine`, forming the **upstream → transport → TTS → LLM → transport → downstream** audio pipeline.

This architecture enables Dograh to onboard new telephony vendors by implementing only a serializer, optional transfer/hangup strategies, and a thin `create_transport` wrapper—no modifications required in core routing or pipeline logic.

## Summary

- **Provider Registry**: Dograh uses [`api/services/telephony/registry.py`](https://github.com/dograh-hq/dograh/blob/main/api/services/telephony/registry.py) to map provider names to `ProviderSpec` objects containing `transport_factory` callables.
- **Factory Resolution**: [`api/services/telephony/factory.py`](https://github.com/dograh-hq/dograh/blob/main/api/services/telephony/factory.py) validates credentials and retrieves the correct factory via `registry.get(provider_name)`.
- **Modular Transport Creation**: Each provider implements `create_transport` in its own [`transport.py`](https://github.com/dograh-hq/dograh/blob/main/transport.py), handling credential loading, frame serialization, and `FastAPIWebsocketTransport` instantiation.
- **Uniform Interface**: All transports conform to the Pipecat `FastAPIWebsocketTransport` interface, ensuring consistent audio pipeline behavior regardless of the underlying telephony service.

## Frequently Asked Questions

### How does Dograh support a new telephony provider?

Add a new provider package under `api/services/telephony/providers/` containing a [`transport.py`](https://github.com/dograh-hq/dograh/blob/main/transport.py) with an async `create_transport` function, a frame serializer, and a registration call in [`__init__.py`](https://github.com/dograh-hq/dograh/blob/main/__init__.py) that supplies a `ProviderSpec` with `transport_factory` set to your creation function. No changes to core routing code are required.

### What is the role of the transport_factory in Dograh?

The `transport_factory` is an async callable registered in `ProviderSpec` that constructs a fully-configured `FastAPIWebsocketTransport` instance. It encapsulates provider-specific logic for credential retrieval, frame serialization, and audio parameter configuration while exposing a uniform interface to the Pipecat pipeline.

### How does Dograh validate telephony credentials at runtime?

The `load_credentials_for_transport` function (called within each `create_transport` implementation) fetches stored configuration and validates that the `expected_provider` matches the provider name associated with the credentials, preventing cross-provider configuration errors.

### Can transports be created manually outside the factory?

Yes. You can import `create_transport` directly from provider modules (e.g., `from api.services.telephony.providers.twilio.transport import create_transport`) and invoke it with explicit parameters for testing, custom tooling, or bypassing the registry resolution logic when you already know the target provider.