FreeSWITCH Frame Serializer

uv add pipecat-freeswitch

from fastapi import WebSocket
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.worker import PipelineParams, PipelineWorker
from pipecat.transports.websocket.fastapi import (
    FastAPIWebsocketParams,
    FastAPIWebsocketTransport,
)
from pipecat_freeswitch import FreeSWITCHFrameSerializer


async def run_bot(websocket: WebSocket):
    await websocket.accept()
    call_uuid = (await websocket.receive_text()).strip()

    serializer = FreeSWITCHFrameSerializer(
        call_uuid=call_uuid,
        esl_host="10.0.0.5",
        esl_password="your-event-socket-password",
        params=FreeSWITCHFrameSerializer.InputParams(
            freeswitch_sample_rate=8000, output_gain_db=-12
        ),
    )
    transport = FastAPIWebsocketTransport(
        websocket=websocket,
        params=FastAPIWebsocketParams(
            serializer=serializer,
            audio_in_enabled=True,
            audio_out_enabled=True,
            audio_out_10ms_chunks=20,  # 200 ms blocks keep barge-in working
            audio_out_end_silence_secs=0,
        ),
    )

    pipeline = Pipeline(
        [
            transport.input(),
            # ... VAD, STT, LLM, TTS, etc.
            transport.output(),
        ]
    )
    worker = PipelineWorker(
        pipeline,
        params=PipelineParams(audio_in_sample_rate=8000, audio_out_sample_rate=8000),
    )

    @transport.event_handler("on_client_disconnected")
    async def on_client_disconnected(transport, client):
        await worker.cancel()
        await serializer.cleanup()  # hang up and close the event socket

    # ... run the worker