​
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:See the Frame Serializers documentation for more details.
​
session_timeout
int | None
default:"None"
Session timeout in seconds. If set, triggers timeout event when no activity is detected

​
Audio Configuration

​
audio_in_enabled
bool
default:false
Enable audio input from the WebRTC client
​
audio_in_passthrough
bool
default:"False"
When enabled, incoming audio frames are pushed downstream
​
audio_out_enabled
bool
default:"False"
Enable audio output capabilities
​
audio_out_sample_rate
int
default:"None"
Audio output sample rate in Hz
​
audio_out_channels
int
default:"1"
Number of audio output channels

​
Voice Activity Detection (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_in_enabled=True,
            audio_out_enabled=True,
            vad_analyzer=SileroVADAnalyzer(),
            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