​
Overview

WebsocketServerTransport provides a WebSocket server implementation for real-time audio communication. It supports bidirectional audio streams and voice activity detection (VAD).

WebsocketServerTransport 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 WebsocketServerTransport, install the required dependencies:

pip install pipecat-ai[websocket]

​
Configuration

​
Constructor Parameters

​
host
str
default:
"localhost"

Host address to bind the WebSocket server

​
port
int
default:
"8765"

Port number for the WebSocket server

​
params
WebsocketServerParams
default:
"WebsocketServerParams()"

Transport configuration parameters

​
WebsocketServerParams Configuration

​
add_wav_header
bool
default:
"False"

Add WAV header to audio frames

​
serializer
FrameSerializer
default:
"ProtobufFrameSerializer()"

Frame serializer for WebSocket messages

​
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 pipecat.transports.network.websocket_server import (
    WebsocketServerTransport,
    WebsocketServerParams
)
from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.pipeline.pipeline import Pipeline

# Configure transport
transport = WebsocketServerTransport(
    host="localhost",
    port=8765,
    params=WebsocketServerParams(
        audio_out_sample_rate=16000,
        audio_out_enabled=True,
        add_wav_header=True,
        vad_enabled=True,
        vad_analyzer=SileroVADAnalyzer(),
        vad_audio_passthrough=True,
        session_timeout=180  # 3 minutes
    )
)

# 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
])

Check out the Websocket Server example to see how to use this transport in a pipeline.

​
Event Callbacks

WebsocketServerTransport 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 server.

Parameters:

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

Called when a client disconnects from the WebSocket server.

Parameters:

  • transport: The WebsocketServerTransport instance
  • client: WebSocket client connection object
@transport.event_handler("on_client_disconnected")
async def on_client_disconnected(transport, client):
    logger.info(f"Client disconnected: {client.remote_address}")
​
on_session_timeout
async callback

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

Parameters:

  • transport: The WebsocketServerTransport instance
  • client: WebSocket client connection object
@transport.event_handler("on_session_timeout")
async def on_session_timeout(transport, client):
    logger.info(f"Session timeout for client: {client.remote_address}")
    # 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

  • Supports real-time audio communication
  • Best suited for server-side applications
  • Handles WebSocket connection management
  • Provides voice activity detection
  • Supports session timeouts
  • Single client per server (new connections replace existing ones)
  • All callbacks are asynchronous
FastAPI WebSocketAssemblyAI