SmallWebRTCTransport

A lightweight WebRTC transport for peer-to-peer audio and video communication in Pipecat

Overview

SmallWebRTCTransport enables peer-to-peer (“serverless”) WebRTC connections between clients and your Pipecat application. It implements bidirectional audio, video and data channels using WebRTC for real-time communication. This transport is open source and self-contained, with no dependencies on any other infrastructure.

SmallWebRTCTransport is the default transport for the Pipecat examples and starter kits. It is heavily tested and can be used in production.

For detailed notes on how to decide between using the SmallWebRTCTransport or other WebRTC transports like the DailyTransport, see this post.

SmallWebRTCTransport API

Transport methods and configuration options

SmallWebRTCConnection API

Connection management and signaling methods

Example Code

Working example with voice agent implementation

Official WebRTC Docs

Learn more about WebRTC!

Installation

To use SmallWebRTCTransport, install the required dependencies:

pip install "pipecat-ai[webrtc]"

No API keys are required since this is a peer-to-peer transport implementation.

For production deployments across different networks, you may need to configure STUN/TURN servers for NAT traversal.

Frames

Input

• InputAudioRawFrame - Audio data from WebRTC peer
• UserImageRawFrame - Video frames from peer’s camera
• TransportMessageUrgentFrame - Application messages from peer

Output

• OutputAudioRawFrame - Audio data to WebRTC peer
• OutputImageRawFrame - Video frames to peer
• TransportMessageFrame - Application messages to peer
• TransportMessageUrgentFrame - Urgent messages to peer

Key Features

• Serverless Architecture: Direct peer-to-peer connections with no intermediate servers
• Bidirectional Media: Full-duplex audio and video streaming
• Data Channels: Application messaging and signaling support
• Production Ready: Heavily tested and used in Pipecat examples
• ICE Support: Configurable STUN/TURN servers for NAT traversal

Usage Example

SmallWebRTCTransport requires two components working together: a signaling server to handle the WebRTC handshake, and your Pipecat bot that processes the media streams. Unlike other transports, you cannot use SmallWebRTCTransport without implementing both parts.

Using the Development Runner (Recommended)

The easiest way to get started is using Pipecat’s development runner, which handles all the server infrastructure automatically:

from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.pipeline.pipeline import Pipeline
from pipecat.runner.types import RunnerArguments, SmallWebRTCRunnerArguments
from pipecat.transports.base_transport import TransportParams
from pipecat.transports.network.small_webrtc import SmallWebRTCTransport

async def run_bot(transport):
    """Your core bot logic - works with any transport."""
    # Your services (STT, LLM, TTS, etc.)
    # ...

    # Create pipeline
    pipeline = Pipeline([
        transport.input(),              # Receive audio from client
        stt,                            # Convert speech to text
        context_aggregator.user(),      # Add user messages to context
        llm,                            # Process text with LLM
        tts,                            # Convert text to speech
        transport.output(),             # Send audio responses to client
        context_aggregator.assistant(), # Add assistant responses to context
    ])

    # Event handlers
    @transport.event_handler("on_client_connected")
    async def on_client_connected(transport, client):
        # Start conversation when client connects
        await task.queue_frames([context_aggregator.user().get_context_frame()])

    @transport.event_handler("on_client_disconnected")
    async def on_client_disconnected(transport, client):
        await task.cancel()

    # Run the pipeline
    # ...

# The `bot()` function is the entry point for the development runner.
async def bot(runner_args: RunnerArguments):
    """Entry point called by the development runner."""

    # RunnerArguments are types that can be used to configure the transport.
    # The SmallWebRTCRunnerArguments contains the WebRTC connection
    # parameters needed to establish the connection.
    if isinstance(runner_args, SmallWebRTCRunnerArguments):
        transport = SmallWebRTCTransport(
            webrtc_connection=runner_args.webrtc_connection,
            params=TransportParams(
                audio_in_enabled=True,
                audio_out_enabled=True,
                vad_analyzer=SileroVADAnalyzer(),
            ),
        )
        await run_bot(transport)

if __name__ == "__main__":
    # Run the bot using the development runner
    from pipecat.runner.run import main
    main()

Run your bot with:

python bot.py --transport webrtc
# Opens http://localhost:7860/client

The development runner automatically:

• Creates a FastAPI server with WebRTC signaling endpoints
• Serves a built-in web interface for testing
• Handles WebRTC connection management and cleanup
• Spawns your bot for each new connection

The development runner is the recommended approach for development and can also be used in production with Pipecat Cloud.

Manual Server Implementation

For advanced use cases or custom deployment requirements, you can implement the server infrastructure manually:

1. Signaling Server (Required)

First, create a web server to handle WebRTC connection establishment. This server receives offers from clients and creates the WebRTC connections that your bot will use:

import asyncio
from contextlib import asynccontextmanager
from typing import Dict

from fastapi import BackgroundTasks, FastAPI
from fastapi.responses import FileResponse
from pipecat.transports.network.webrtc_connection import IceServer, SmallWebRTCConnection

app = FastAPI()

# Store active WebRTC connections by their unique ID
connections: Dict[str, SmallWebRTCConnection] = {}

# Configure ICE servers for NAT traversal
ice_servers = [
    IceServer(urls="stun:stun.l.google.com:19302"),
]

@app.post("/api/offer")
async def handle_offer(request: dict, background_tasks: BackgroundTasks):
    """Handle WebRTC offer from client and return SDP answer."""
    pc_id = request.get("pc_id")

    if pc_id and pc_id in connections:
        # Handle reconnections
        webrtc_connection = connections[pc_id]
        await webrtc_connection.renegotiate(
            sdp=request["sdp"],
            type=request["type"]
        )
    else:
        # Create new WebRTC connection
        webrtc_connection = SmallWebRTCConnection(ice_servers=ice_servers)
        await webrtc_connection.initialize(
            sdp=request["sdp"],
            type=request["type"]
        )

        # Clean up when client disconnects
        @webrtc_connection.event_handler("closed")
        async def on_closed(connection):
            connections.pop(connection.pc_id, None)

        # Start bot for this connection (defined below)
        background_tasks.add_task(run_bot, webrtc_connection)

    answer = webrtc_connection.get_answer()
    connections[answer["pc_id"]] = webrtc_connection
    return answer

# Run with: uvicorn server:app --host 0.0.0.0 --port 7860

2. Pipecat Bot Implementation

Next, implement your bot function that receives the WebRTC connection from the server and creates the transport:

from pipecat.audio.vad.silero import SileroVADAnalyzer
from pipecat.pipeline.pipeline import Pipeline
from pipecat.transports.base_transport import TransportParams
from pipecat.transports.network.small_webrtc import SmallWebRTCTransport

async def run_bot(webrtc_connection):
    """Run the Pipecat bot for a specific WebRTC connection."""
    # Create transport using the connection from the server
    transport = SmallWebRTCTransport(
        webrtc_connection=webrtc_connection,
        params=TransportParams(
            audio_in_enabled=True,   # Accept audio from client
            audio_out_enabled=True,  # Send audio to client
            vad_analyzer=SileroVADAnalyzer(),
        ),
    )

    # Your services (STT, LLM, TTS, etc.)
    # ...

    # Create pipeline
    pipeline = Pipeline([
        transport.input(),              # Receive audio from client
        stt,                            # Convert speech to text
        context_aggregator.user(),      # Add user messages to context
        llm,                            # Process text with LLM
        tts,                            # Convert text to speech
        transport.output(),             # Send audio responses to client
        context_aggregator.assistant(), # Add assistant responses to context
    ])

    # Handle connection events
    @transport.event_handler("on_client_connected")
    async def on_client_connected(transport, client):
        # Start conversation when client connects
        await task.queue_frames([context_aggregator.user().get_context_frame()])

    @transport.event_handler("on_client_disconnected")
    async def on_client_disconnected(transport, client):
        await task.cancel()

    # Run the pipeline
    # ...

How It Works Together

1. Client connects to your server at /api/offer with WebRTC offer
2. Server creates SmallWebRTCConnection and initializes it with the offer
3. Server starts run_bot() in background with the connection
4. Bot creates SmallWebRTCTransport using the connection
5. Media flows directly between client and bot via WebRTC peer connection

This architecture gives you the benefits of peer-to-peer WebRTC (low latency, no media servers) while maintaining a simple server for connection management.

How to connect with SmallWebRTCTransport

For client connections, you have two options:

1. Pipecat Client SDK (Recommended): Use the JavaScript SDK for easy integration
2. Custom WebRTC Client: Implement WebRTC signaling manually for advanced use cases

The server setup is shown in the Usage Example above.

Examples

To see a complete implementation, check out the following examples:

Video Transform

Demonstrates real-time video processing using WebRTC transport

Voice Agent

Implements a voice assistant using WebRTC for audio communication

Media Handling

Audio

Audio is processed in 20ms chunks by default. The transport handles audio format conversion and resampling as needed:

• Input audio is processed at 16kHz (mono) to be compatible with speech recognition services
• Output audio can be configured to match your application’s requirements, but it must be mono, 16-bit PCM audio

Video

Video is streamed using RGB format by default. The transport provides:

• Frame conversion between different color formats (RGB, YUV, etc.)
• Configurable resolution and framerate

WebRTC ICE Servers Configuration

When implementing WebRTC in your project, STUN (Session Traversal Utilities for NAT) and TURN (Traversal Using Relays around NAT) servers are usually needed in cases where users are behind routers or firewalls.

In local networks (e.g., testing within the same home or office network), you usually don’t need to configure STUN or TURN servers. In such cases, WebRTC can often directly establish peer-to-peer connections without needing to traverse NAT or firewalls.

What are STUN and TURN Servers?

• STUN Server: Helps clients discover their public IP address and port when they’re behind a NAT (Network Address Translation) device (like a router). This allows WebRTC to attempt direct peer-to-peer communication by providing the public-facing IP and port.

• TURN Server: Used as a fallback when direct peer-to-peer communication isn’t possible due to strict NATs or firewalls blocking connections. The TURN server relays media traffic between peers.

Why are ICE Servers Important?

ICE (Interactive Connectivity Establishment) is a framework used by WebRTC to handle network traversal and NAT issues. The iceServers configuration provides a list of STUN and TURN servers that WebRTC uses to find the best way to connect two peers.

Advanced Configuration

ICE Servers

For better connectivity, especially when testing across different networks, you can provide STUN servers:

webrtc_connection = SmallWebRTCConnection(
    ice_servers=["stun:stun.l.google.com:19302", "stun:stun1.l.google.com:19302"]
)

You can also use IceServer objects for more advanced configuration:

from pipecat.transports.network.webrtc_connection import IceServer

webrtc_connection = SmallWebRTCConnection(
    ice_servers=[
        IceServer(urls="stun:stun.l.google.com:19302"),
        IceServer(
            urls="turn:turn.example.com:3478",
            username="username",
            credential="password"
        )
    ]
)

ESP32 Compatibility

For ESP32 WebRTC connections, use the development runner with ESP32 mode:

python bot.py -t webrtc --esp32 --host 192.168.1.100

This enables SDP munging required for ESP32 WebRTC compatibility.

Troubleshooting

If clients have trouble connecting or streaming:

1. Check browser console for WebRTC errors
2. Ensure you’re using HTTPS in production (required for WebRTC)
3. For testing across networks, consider using Daily which provides TURN servers
4. Verify browser permissions for camera and microphone
5. Try the development runner first to isolate connection issues

Additional Notes

• HTTPS Requirement: WebRTC requires HTTPS in production environments
• Browser Compatibility: Modern browsers support WebRTC natively
• Firewall Considerations: May need TURN servers for restrictive network environments
• Latency: Direct peer-to-peer connections typically provide lower latency than server-mediated solutions
• Scalability: Each connection is independent; consider connection pooling for high-traffic applications
• Development: Use the development runner for rapid prototyping and testing

For comparing with other transport options, see the transport comparison guide.