Egyedi telefonszolgáltató

Előfeltételek

Ez az útmutató feltételezi, hogy két olyan könyvtárat ismer, amelyre a szolgáltatói felület épül:

• Pipecat – a nyílt forráskódú, valós idejű voice/multimodal streaming framework, amelyre az IntraCord call pipeline-ja épül. A FastAPIWebsocketTransport streameli az audiot a telefonszolgáltató és az IntraCord pipeline-ja között; ebből hoz létre egyet a transport factoryban. Ha még nem épített Pipecat transportot vagy frame serializert, először olvassa el a transport útmutatót.
• Pydantic – a szolgáltató tárolt Credentials adatait meghatározó kérés/válasz sémákhoz használatos (config.py alább). Ha még nem ismeri a Pydanticot, annak modelldokumentációja mindent lefed, amit itt használnak (BaseModel, Field, Literal diszkriminátorok).

Nincs szükség mély szakértelemre egyikben sem – az útmutató követéséhez elég egy BaseModel és egy Pipecat transport class olvasása.

Áttekintés

A telefonszolgáltató önregisztráló package-ként valósul meg az api/services/telephony/providers/<name>/ alatt. A package mindent megad, amire az IntraCordnak szüksége van a szolgáltató bekötéséhez: provider class, transport factory, audio config, request/response schemák, opcionális HTTP route-ok és a konfigurációs UI form metadata. Mindezt egy importáláskor regisztrált ProviderSpec fogja össze.

Új szolgáltató hozzáadásához nem szükséges megérinteni a gyárat, az audiokonfigurációt, az API-útvonal-modult, a run-pipeline modult vagy a frontendet. A szolgáltató mappán kívüli egyetlen módosítások a következők:

1. Egy importsor a api/services/telephony/providers/__init__.py-ben
2. Egy import sor a api/schemas/telephony_config.py-ben a kérés/válasz osztályok hozzáadásához a TelephonyConfigRequest megkülönböztetett unióhoz

Szolgáltatói csomag elrendezése

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

Három fájl szükséges (__init__.py, config.py, provider.py, transport.py). A többi nem kötelező, és a rendszer automatikusan észleli, ha jelen van:

• routes.py — ha a modul létezik, és exportálja a router: APIRouter-t, az útvonal modult lustán importálja és a api.routes.telephony alá szereli a api.routes.telephony a importlib-n keresztül. Azok a szolgáltatók, amelyek csak WebSocketen keresztül streamelnek (pl. ARI), elhagyhatják.
• strategies.py – olyan szállítások használják, amelyek szolgáltató-specifikus hívásátvitelt/lekötési logikát igényelnek a keretsorosítóban (pl. Twilio Conference átvitelek).
• serializers.py – jellemzően pipecatből történő reexport. Tartsa meg a fájlt akkor is, ha egysoros újraexportálásról van szó, így a szállítási kód a .serializers-ből importálódik, így kézenfekvő helyet biztosít az egyéni alosztály későbbi eldobására.

A TelephonyProvider interfész

TelephonyProvider alosztály a provider.py-ben:

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)

Tekintse meg a api/services/telephony/base.py dokumentumot az egyes metódusok teljes leírásához.

!

A get_webhook_response és a handle_websocket organization_id-t kap, nem felhasználói azonosítót. Minden workflow-nak és Workflow Runnak szervezeti hatókörben kell működnie; az azonosítót közvetlenül adja tovább a run_pipeline_telephony függvénynek. A workflow tulajdonosát szándékosan nem adjuk át a szolgáltatóknak; ha a hozzárendeléshez szüksége van rá, olvassa ki a workflow rekordjából.

Megvalósítási útmutató

1. Konfigurációs sémák

Határozzon meg Pydantic modelleket a Credentials adataihoz. A provider Literal diszkriminátor teszi a sémákat helyesen a rendszerleíró adatbázis megkülönböztetett unióján keresztül.

# providers/your_provider/config.py
from typing import List, Literal
from pydantic import BaseModel, Field


class 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. Szállítógyár

Építse meg a Pipecat FastAPIWebsocketTransport-t az elfogadott WebSocketekhez. A Credentials adatokat mindig a load_credentials_for_transport-n keresztül töltse be, hogy a megfelelő konfigurációs sor kerüljön kiválasztásra, ha a Workflow Run telephony_configuration_id értéket tartalmaz.

# providers/your_provider/transport.py
from fastapi import WebSocket
from api.services.pipecat.audio_config import AudioConfig
from api.services.pipecat.audio_mixer import build_audio_out_mixer
from api.services.telephony.factory import load_credentials_for_transport
from pipecat.transports.websocket.fastapi import (
    FastAPIWebsocketParams,
    FastAPIWebsocketTransport,
)
from .serializers import YourProviderFrameSerializer


async 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. Útvonalak (opcionális)

Ha a szolgáltató POST-okat küld az IntraCord számára (válasz URL-cím, állapot-visszahívások, hívásmegszakítási visszahívások), tegye közzé őket egy modulszintű router-n keresztül. Az útvonalak automatikusan fel vannak szerelve a /api/v1/telephony alá.

# providers/your_provider/routes.py
from fastapi import APIRouter, Request
from 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):
    ...

Az útvonalak lustán töltődnek be a importlib-n keresztül a api.routes.telephony._mount_provider_routers-ről, így az útvonalmodul szabadon importálhat más háttérszolgáltatásokat anélkül, hogy importálási ciklusokat hozna létre a szolgáltatói betöltési időben.

4. Regisztrálja a ProviderSpec

A csomagban található __init__.py minden egyesül:

# providers/your_provider/__init__.py
from typing import Any, Dict

from api.services.pipecat.audio_config import AudioConfig
from api.services.telephony.registry import (
    ProviderSpec,
    ProviderUIField,
    ProviderUIMetadata,
    register,
)

from .config import YourProviderConfigurationRequest, YourProviderConfigurationResponse
from .provider import YourProvider
from .transport import create_transport


def _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)

A ProviderSpec mindent lefed, amire a downstream kódnak szüksége van:

Mező	Használja
name	Minden TelephonyConfiguration sorban megkülönböztetőként és WorkflowRunMode értékként tárolva
provider_cls	factory.get_default_telephony_provider, get_telephony_provider_by_id, get_telephony_provider_for_run
config_loader	factory._normalize_with_phone_numbers (a régi if/elif láncot helyettesíti)
transport_factory	run_pipeline_telephony
audio_config	create_audio_config() és run_pipeline_telephony
config_request_cls / config_response_cls	TelephonyConfigRequest diszkriminált szakszervezet
ui_metadata	GET /api/v1/organizations/telephony-providers/metadata (meghajtja az űrlap felhasználói felületét) és a _sensitive_fields maszkoló segédeszköz
account_id_credential_field	Bejövő webhook-útválasztás ugyanazon szolgáltató több konfigurációján keresztül

5. Csatlakoztassa a csomagot a rendszerleíró adatbázis importálási láncához

Adjon hozzá egy importsort a api/services/telephony/providers/__init__.py-hez:

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

6. Add hozzá a diszkriminált szakszervezethez

Adjon hozzá egy importblokkot a api/schemas/telephony_config.py-hez, hogy a kérés/válasz osztályok részt vegyenek a TelephonyConfigRequest unióban és a TelephonyConfigurationResponse alakzatban:

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

Ennyi a háttérbekötéshez.

Frontend

A konfigurációs űrlap metaadat-vezérelt. A felhasználói felület meghívja a GET /api/v1/organizations/telephony-providers/metadata-t, visszakéri a szolgáltatók listáját és azok ProviderUIField definícióit, és általánosan rendereli az egyes űrlapokat. Nincs szükség szolgáltatónkénti frontend kódra — az Ön ProviderUIMetadata deklarációja az, ami az űrlapot hajtja.

Ha olyan új mezőtípust ad hozzá, amelyet a meglévő megjelenítő nem támogat (pl. fájlfeltöltés), akkor bővítse ki a megjelenítőt a ui/src/app/(authenticated)/telephony-configurations/ mezőben. A támogatott ProviderUIField.type értékek ma a text, password, textarea, string-array és number.

Hangformátummal kapcsolatos megfontolások

Minden szolgáltató a AudioConfig-n keresztül deklarálja vezetékes formátumát. Általános formák:

• Twilio / Plivo: 8 kHz μ-törvény, base64 kódolású JSON-keretek
• Vonage: 16 kHz-es lineáris PCM bináris keretként
• Csillag ARI: 8 kHz Lineáris PCM external media kapcsolaton keresztül

A pipeline mintavételi frekvenciája 16 kHz-re van korlátozva, hogy megfeleljen a VAD, amely csak 8000 Hz-et vagy 16000 Hz-et fogad el; továbbítja az újramintavételezést a vezetékformátum és a pipeline belső sebessége között.

Tesztelés

# api/tests/telephony/test_your_provider.py
import pytest
from api.services.telephony.providers.your_provider import YourProvider


@pytest.mark.asyncio
async def test_validate_config():
    provider = YourProvider({
        "api_key": "test_key",
        "api_secret": "test_secret",
        "from_numbers": ["+1234567890"],
    })
    assert provider.validate_config() is True

Az end-to-end teszteléshez mentse el a szolgáltatót a Telefónia beállításai felületen, majd indítson teszthívást egy workflow-ból.

Legjobb gyakorlatok

1. Használja a provider registryt – soha ne importálja közvetlenül más szolgáltató osztályát; a feloldást mindig a factory helpereken keresztül végezze (get_default_telephony_provider, get_telephony_provider_by_id stb.).
2. Érzékeny mezők — jelöljön meg minden Credentials mezőt sensitive=True értékkel a ProviderUIMetadata mezőben. A mentési endpoint maszkolja ezeket olvasáskor, és megőrzi az eredetit, amikor az ügyfél újra elküld egy maszkolt értéket.
3. Ellenőrizze a bejövő aláírásokat – a verify_inbound_signature mindig ellenőrizze a webhook aláírását, és hiba esetén utasítsa el a kérést. Ha a szolgáltató minden webhookot aláír, a hiányzó aláírásfejléc azt jelenti, hogy a kérés nem tőle érkezett: adjon vissza False értéket, ahogy a providers/twilio/ és a providers/plivo/ is teszi. Ne használjon helyőrző return True értéket; ezek a handlerek bárki számára elérhetők, aki kitalál egy workflow_run_id-t.
4. A Transportok késleltetve töltik be a Credentials adatokat – a load_credentials_for_transport függvényt a Workflow Run telephony_configuration_id értékével hívja meg. Ne olvassa be a szervezet alapértelmezett konfigurációját a transport.py fájlból.
5. Naplózás — használja a loguru.logger-t.

Referencia megvalósítások

Szolgáltató	Nevezetes a
providers/twilio/	Teljes funkcionalitású: kimenő, bejövő, konferencia átvitel, állapot visszahívások, egyéni stratégiák
providers/plivo/	Nemrég hozzáadott hivatkozás; tükrözi Twilio alakját többszörös visszahívási aláírásokkal
providers/vonage/	JWT auth, 16 kHz Lineáris PCM, NCCO válaszok
providers/cloudonix/	SIP-alapú, egyedi hívási stratégiák
providers/telnyx/	Hívásvezérlési stílus: REST által vezérelt bejövő válaszfolyamat jelölési válasz helyett
providers/ari/	Minimális példa – nincs routes.py, nincs bejövő webhook-ellenőrzés, csak WebSocket

i

Használja az ARI-t a legkisebb használható példaként, ha a szolgáltató nem teszi közzé a HTTP-t webhooks, és a Twilio referenciaként szolgál.