Собственный провайдер телефонии

api/services/telephony/providers/your_provider/
├── __init__.py        # Собирает и регистрирует ProviderSpec
├── config.py          # Pydantic-схемы Request/Response
├── provider.py         # Подкласс TelephonyProvider
├── transport.py        # Фабрика WebSocket-транспорта Pipecat
├── serializers.py       # Сериализатор фреймов (обычно реэкспорт из pipecat)
├── routes.py           # (опционально) обработчики HTTP-вебхуков/колбэков
└── strategies.py        # (опционально) стратегии перевода звонка/завершения звонка

from api.services.telephony.base import (
    CallInitiationResult,
    NormalizedInboundData,
    ProviderSyncResult,
    TelephonyProvider,
)


class YourProvider(TelephonyProvider):
    PROVIDER_NAME = "your_provider"
    WEBHOOK_ENDPOINT = "your-provider-xml"  # путь под /api/v1/telephony

    def __init__(self, config: dict):
        self.api_key = config.get("api_key")
        self.from_numbers = config.get("from_numbers", [])

    # ---------- исходящие звонки ----------
    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: ...

    # ---------- вебхуки ----------
    async def verify_webhook_signature(self, url, params, signature) -> bool: ...
    async def get_webhook_response(self, workflow_id, user_id, workflow_run_id) -> str: ...
    def parse_status_callback(self, data: dict) -> dict: ...

    # ---------- websocket ----------
    async def handle_websocket(self, websocket, workflow_id, user_id, workflow_run_id): ...

    # ---------- входящие звонки ----------
    @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: ...

    # ---------- переводы звонка ----------
    async def transfer_call(self, destination, transfer_id, conference_name,
                            timeout=30, **kwargs) -> dict: ...
    def supports_transfers(self) -> bool: ...

    # ---------- опционально ----------
    async def configure_inbound(self, address, webhook_url) -> ProviderSyncResult:
        # По умолчанию возвращает ok=True — реализуйте, только если ваш провайдер
        # поддерживает программную настройку вебхука (например, привязку номера
        # к URL через API). Используется, чтобы направить входящие номера на
        # /api/v1/telephony/inbound/run.
        return ProviderSyncResult(ok=True)

# 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        # маскируется при возврате
    api_secret: str     # маскируется при возврате
    from_numbers: List[str]

# 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,
    # специфичные для провайдера kwargs (передаются run_pipeline_telephony как **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,
        ),
    )

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

# 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]:
    """Приводит сохраненный словарь учётных данных к форме конструктора."""
    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, используемые для исходящих звонков",
        ),
    ],
)


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,
    # Поле учётных данных, однозначно идентифицирующее аккаунт провайдера.
    # Используется для различения входящих вебхуков между несколькими
    # конфигурациями одного и того же провайдера. Пустая строка для
    # провайдеров без концепции account-id.
    account_id_credential_field="api_key",
)

register(SPEC)

from api.services.telephony.providers import (  # noqa: F401  -- side effects
    ari,
    your_provider,        # ← добавьте это
)

from api.services.telephony.providers.your_provider.config import (
    YourProviderConfigurationRequest,
    YourProviderConfigurationResponse,
)

TelephonyConfigRequest = Annotated[
    Union[
        # ...существующие записи...
        YourProviderConfigurationRequest,
    ],
    Field(discriminator="provider"),
]


class TelephonyConfigurationResponse(BaseModel):
    # ...существующие записи...
    your_provider: Optional[YourProviderConfigurationResponse] = None

# 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