pipecat-asterisk

A Pipecat community integration for Asterisk. PyPI package This repository provides a transport and frame serializer to connect your Asterisk with Pipecat pipelines.

Features

• AsteriskWebsocketTransport: Handles raw audio streaming and lifecycle events natively with Asterisk.
• AsteriskFrameSerializer: Serializer to translate Asterisk websocket JSON or plain-text payloads and raw(audio) payloads into Pipecat frames.
• Flow Control: Built-in logic to manage buffer utilization between the Pipecat application and Asterisk.
• AsteriskCommandFrame: If the frame with an arbitrary command is received by the transport, it will be sent to the Asterisk websocket channel. The List of supported commands

Installation

uv add pipecat-asterisk

(Or use pip install pipecat-asterisk if you are using pip for dependency management)

Usage

Understanding sampling rates:: The transport and the serializer are designed to autodetect and handle sampling rate mismatch, but it's a very good idea to have them equal to the slin sample rate on the Asterisk side. Otherwise, we will have double resampling that might introduce some artifacts, CPU load, and latency. Please ensure that the audio_out_sample_rate and audio_in_sample_rate parameters in the PipelineParams are set to match the sampling rate of the Asterisk channel. Max reasonable sampling rate for Asterisk is 48kHz in case of using Opus codec, if you don't use Opus codec, you can safely use 24kHz, it's more than enough for HD codes like G.722, and if you use G.711 (typical for PSTN), 8kHz is enough.

Here is a basic example of how to integrate the Asterisk WebSocket transport into a Pipecat pipeline:

from fastapi import FastAPI, WebSocket
from pipecat.pipeline.worker import PipelineWorker, PipelineParams
from pipecat.workers.runner import WorkerRunner
from pipecat_asterisk import AsteriskWebsocketTransport

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()
    
    # Initialize the Asterisk Transport
    ws_transport = AsteriskWebsocketTransport(websocket=websocket)
    
    # Build your Pipecat pipeline
    pipeline = Pipeline([
        ws_transport.input(),
        # ... other pipeline components (VAD, LLM, TTS, etc.)
        ws_transport.output(),
    ])
    
    # Add the pipeline to a worker and run it
    worker = PipelineWorker(
        pipeline,
        params=PipelineParams(
            audio_out_sample_rate=48000, # It's a very good idea to have them == slin sample rate on asterisk side
            audio_in_sample_rate=48000,  # Otherwise, we will have double resampling that might introduce some artifacts, cpu-load and latency
        )
    )
    ...
    # Run the worker
    runner = WorkerRunner(handle_sigint=False)
    await runner.add_workers(worker)
    await runner.run()
    ...

Running the Example

An example Gemini-based voice bot is provided in examples/pipecat_asterisk/.

1. Configure Asterisk

The examples/pipecat_asterisk/ directory includes a docker-compose.yml and Asterisk configuration files in etc/ to easily spin up a local Asterisk testing environment. After the Docker container is running you can connect any sip client to localhost:5060 with the credentials specified in etc/asterisk/pjsip.conf (user: 1, password: 1). There are a few extensions configured in etc/asterisk/extensions.conf that you can use to test the bot, every extension represents a respective sampling rate.

exten = 8,1,Dial(WebSocket/pipecat/c(slin))
exten = 12,1,Dial(WebSocket/pipecat/c(slin12))
exten = 16,1,Dial(WebSocket/pipecat/c(slin16))
exten = 24,1,Dial(WebSocket/pipecat/c(slin24))
...

To run the Asterisk server with the provided configuration:

cd examples/pipecat_asterisk
docker-compose up -d

2. Set API Keys

The example uses Google's Gemini for conversational AI. Create a .env file or export your key directly:

export GOOGLE_API_KEY="your-google-api-key"

3. Run the application

Run the WebSocket server:

uv sync
uv run examples/pipecat_asterisk/ws_server.py

In case you want to estimate audio quality with respect to different sampling rates, you can run:

cd examples/pipecat_asterisk
uv run long_media_test.py

The script emulates generating audio chunks from TTS service, without need to pay for a real TTS service. The directory contains some audio samples with different sampling rates, but your can use your own files. Save them as raw audio in slin (16-bit PCM, mono) format.

Compatibility

• Tested with Pipecat v1.8.1
• Requires Python 3.12+

Internal Architecture

The transport and the serializer are designed to work with slin encoded audio, because Asterisk natively supports all the flavors of slin and Pipecat's audio frames require to be slin-encoded. If you need to use a different codec, you can transcode on the Asterisk side, it's computationally more efficient and simplifies the transport and serializer logic.

Serializer and the transport implementation are based on the Asterisk websocket channel documentation.

The transport supports flow control logic to manage the buffer utilization between the Pipecat application and Asterisk. This ensures that we don't overwhelm the Asterisk server with too much data at once, while also ensuring that we send data as soon as there is capacity in the remote buffer. The output transport create an instance of flow controller and adds serialized (and resampled if needed) audio frames to the flow controller. The flow controller then decides when to send data to Asterisk based on the current buffer utilization and the amount of data in the local buffer. The flow control logic is as follows:

Flow control logic

flowchart TD
    C{Remote buffer utilization < <br>low water}
    C -->|yes| D{There are bytes in local buffer}
    C -->|no| H{Remote buffer utilization > <br>high water}
    H -->|yes| E[Skip]
    D -->|yes| F[Send up to MAX_WS_SEND <br> bytes from local buffer]
    D -->|no| E
    H -->|no| I{Do we have more than <br>MIN_BATCH bytes in local <br>buffer}
    I -->|yes| F
    I -->|no| E