​
Overview

FastAPIWebsocketTransport provides WebSocket support for FastAPI web applications, enabling real-time audio communication. It supports bidirectional audio streams and voice activity detection (VAD).

FastAPIWebsocketTransport is best suited for server-side applications and prototyping client/server apps.

For client/server production applications, we strongly recommend using a WebRTC-based transport for robust network and media handling.

​
Installation

To use FastAPIWebsocketTransport, install the required dependencies:

pip install pipecat-ai[websocket]

​
Configuration

​
Constructor Parameters

​
websocket
WebSocket
required

FastAPI WebSocket connection instance

​
params
FastAPIWebsocketParams
required

Transport configuration parameters

​
FastAPIWebsocketParams Configuration

​
add_wav_header
bool
default:
"False"

Add WAV header to audio frames

​
serializer
FrameSerializer
required

Frame serializer for WebSocket messages. Common options include: - ProtobufFrameSerializer() - TwilioFrameSerializer(stream_sid)

​
session_timeout
int | None
default:
"None"

Session timeout in seconds. If set, triggers timeout event when no activity is detected

​
Audio Output Configuration

​
audio_out_enabled
bool
default:
"False"

Enable audio output capabilities

​
audio_out_sample_rate
int
default:
"16000"

Audio output sample rate in Hz

​
audio_out_channels
int
default:
"1"

Number of audio output channels

​
Voice Activity Detection (VAD)

​
vad_enabled
bool
default:
"False"

Enable voice activity detection

​
vad_audio_passthrough
bool
default:
"False"

Pass through audio during VAD

​
vad_analyzer
VADAnalyzer | None
default:
"None"

Voice Activity Detection analyzer. You can set this to either SileroVADAnalyzer() or WebRTCVADAnalyzer(). SileroVADAnalyzer is the recommended option. Learn more about the SileroVADAnalyzer.

​
Basic Usage

from fastapi import FastAPI, WebSocket
from pipecat.transports.network.fastapi_websocket import (
    FastAPIWebsocketTransport,
    FastAPIWebsocketParams
)
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.pipeline.pipeline import Pipeline
from pipecat.serializers.twilio import TwilioFrameSerializer

app = FastAPI()

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    await websocket.accept()

    # Configure transport
    transport = FastAPIWebsocketTransport(
        websocket=websocket,
        params=FastAPIWebsocketParams(
            audio_out_enabled=True,
            vad_enabled=True,
            vad_analyzer=SileroVADAnalyzer(),
            vad_audio_passthrough=True,
            serializer=TwilioFrameSerializer(),
        )
    )

    # Use in pipeline
    pipeline = Pipeline([
        transport.input(),    # Handle incoming audio
        stt,                  # Speech-to-text
        llm,                  # Language model
        tts,                  # Text-to-speech
        transport.output()    # Handle outgoing audio
    ])

    # Run pipeline
    task = PipelineTask(pipeline)
    await PipelineRunner().run(task)

Check out the Twilio Chatbot example to see how to use the FastAPI transport in a phone application.

​
Event Callbacks

FastAPIWebsocketTransport provides callbacks for handling client connection events. Register callbacks using the @transport.event_handler() decorator.

​
Connection Events

​
on_client_connected
async callback

Called when a client connects to the WebSocket endpoint.

Parameters:

  • transport: The FastAPIWebsocketTransport instance
  • client: FastAPI WebSocket connection object
@transport.event_handler("on_client_connected")
async def on_client_connected(transport, client):
    logger.info("Client connected")
    # Initialize conversation
    await task.queue_frames([LLMMessagesFrame(initial_messages)])
​
on_client_disconnected
async callback

Called when a client disconnects from the WebSocket endpoint.

Parameters:

  • transport: The FastAPIWebsocketTransport instance
  • client: FastAPI WebSocket connection object
@transport.event_handler("on_client_disconnected")
async def on_client_disconnected(transport, client):
    logger.info("Client disconnected")
    await task.queue_frames([EndFrame()])
​
on_session_timeout
async callback

Called when a session times out (if session_timeout is configured).

Parameters:

  • transport: The FastAPIWebsocketTransport instance
  • client: FastAPI WebSocket connection object
@transport.event_handler("on_session_timeout")
async def on_session_timeout(transport, client):
    logger.info("Session timeout")
    # Handle timeout (e.g., send message, close connection)

​
Frame Types

​
Input Frames

​
InputAudioRawFrame
Frame

Raw audio data from the WebSocket client

​
Output Frames

​
OutputAudioRawFrame
Frame

Audio data to be sent to the WebSocket client

​
Notes

  • Integrates with FastAPI web applications
  • Supports real-time audio communication
  • Handles WebSocket connection management
  • Provides voice activity detection
  • Supports session timeouts
  • All callbacks are asynchronous
  • Compatible with various frame serializers
Daily WebRTCWebSocket Server