How Dograh Handles Transport Setup for Different Telephony Providers

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 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, 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)

# 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.

# 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), 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 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, the create_transport function loads credentials, instantiates a TwilioFrameSerializer, and returns a configured FastAPIWebsocketTransport:

# 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 but utilizes a binary WebSocket serializer and JWT-based authentication:

# 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 to map provider names to ProviderSpec objects containing transport_factory callables.
• Factory Resolution: 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, 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 with an async create_transport function, a frame serializer, and a registration call in __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.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}