Transport Utilities

Configuration and helper utilities for Daily, LiveKit, telephony, and WebRTC transports

Overview

Pipecat provides several utility modules for configuring transports and handling transport-specific operations. While the development runner handles most of these automatically, these utilities are useful for custom setups, advanced configurations, or when building your own deployment infrastructure.

Daily Configuration

The Daily utilities handle room creation, token generation, and authentication setup for Daily integration. They support both standard video/audio rooms and SIP-enabled rooms for telephony integration.

Basic Configuration

Use configure() for simple room and token setup:

import aiohttp
from pipecat.runner.daily import configure

async with aiohttp.ClientSession() as session:
    # Returns a DailyRoomConfig object
    room_config = await configure(session)

    room_url = room_config.room_url
    token = room_config.token

    # Use with DailyTransport
    transport = DailyTransport(room_url, token, "Bot Name", params=DailyParams())

The configure() function returns a DailyRoomConfig object with:

• room_url: The Daily room URL for joining
• token: Authentication token for the bot
• sip_endpoint: SIP endpoint URI (None for standard rooms)

This function:

• Uses DAILY_SAMPLE_ROOM_URL environment variable if set, otherwise creates a new room
• Creates rooms with 2-hour expiration and automatic ejection by default
• Generates an authentication token with 2-hour expiration using DAILY_API_KEY

Environment Variables

Required:

• DAILY_API_KEY: Daily API key for creating rooms and tokens

Optional:

• DAILY_SAMPLE_ROOM_URL: Use an existing room instead of creating one
• DAILY_API_URL: Override Daily API endpoint (defaults to https://api.daily.co/v1)

Room and Token Management

When no DAILY_SAMPLE_ROOM_URL is provided, rooms are created automatically with:

• 2-hour expiration from creation time
• Automatic participant ejection when the room expires
• Unique names using UUID prefixes (e.g. pipecat-uuid)

Expired rooms are automatically cleaned up by Daily, so you don’t need to manage them manually.

Tokens are generated with 2-hour expiration and include necessary permissions for bot participation. The utilities handle all the Daily REST API interactions automatically.

SIP Configuration

Daily utilities support SIP-enabled rooms for telephony integration. When SIP parameters are provided, the function creates rooms with telephony capabilities:

import aiohttp
from pipecat.runner.daily import configure

async with aiohttp.ClientSession() as session:
    # Create SIP-enabled room
    sip_config = await configure(
        session,
        sip_caller_phone="[phone]",
        sip_enable_video=False,  # Voice-only by default
        sip_num_endpoints=1,
        sip_codecs={"audio": ["OPUS"], "video": ["H264"]}
    )

    print(f"Room URL: {sip_config.room_url}")
    print(f"SIP Endpoint: {sip_config.sip_endpoint}")
    print(f"Token: {sip_config.token}")

SIP Parameters:

• sip_caller_phone: Phone number or identifier for SIP display name (enables SIP mode)
• sip_enable_video: Whether video is enabled for SIP calls (default: False)
• sip_num_endpoints: Number of allowed SIP endpoints (default: 1)
• sip_codecs: Audio/video codecs to support (optional, uses Daily defaults if not specified)

SIP-enabled rooms cannot use existing room URLs from DAILY_SAMPLE_ROOM_URL. They always create new temporary rooms with SIP configuration.

Backward Compatibility

The configure() function supports tuple unpacking:

# Tuple unpacking (legacy style)
room_url, token = await configure(session)

# Object access (recommended)
config = await configure(session)
room_url = config.room_url
token = config.token

LiveKit Configuration

LiveKit utilities manage authentication tokens, room setup, and agent permissions for LiveKit server integration.

Basic Configuration

Use configure() for standard setup:

from pipecat.runner.livekit import configure

url, token, room_name = await configure()

# Use with LiveKitTransport
transport = LiveKitTransport(url=url, token=token, room_name=room_name, params=LiveKitParams())

Configuration with Arguments

For command-line integration:

import argparse
from pipecat.runner.livekit import configure_with_args

parser = argparse.ArgumentParser()
url, token, room_name, args = await configure_with_args(parser)

Supports these command-line options:

• -r, --room: Specify LiveKit room name
• -u, --url: Specify LiveKit server URL

Token Generation

LiveKit provides two token generation functions:

generate_token(room_name, participant_name, api_key, api_secret) Creates a standard participant token for users or testing.

generate_token_with_agent(room_name, participant_name, api_key, api_secret) Creates an agent token with special permissions. Use this for your bots.

from pipecat.runner.livekit import generate_token_with_agent

# Generate agent token for your bot
agent_token = generate_token_with_agent("my-room", "Pipecat Bot", api_key, api_secret)

# Generate user token for testing
user_token = generate_token("my-room", "Test User", api_key, api_secret)

Environment Variables

Required:

• LIVEKIT_API_KEY: LiveKit API key
• LIVEKIT_API_SECRET: LiveKit API secret
• LIVEKIT_URL: LiveKit server URL
• LIVEKIT_ROOM_NAME: Default room name

All environment variables are required for LiveKit to function properly.

WebSocket and Transport Utilities

The transport utilities provide helper functions for WebSocket parsing, SDP manipulation, and transport management.

Telephony WebSocket Parsing

Use parse_telephony_websocket() to auto-detect telephony providers and extract call data:

from pipecat.runner.utils import parse_telephony_websocket

transport_type, call_data = await parse_telephony_websocket(websocket)

if transport_type == "twilio":
    stream_id = call_data["stream_id"]
    call_id = call_data["call_id"]
elif transport_type == "telnyx":
    stream_id = call_data["stream_id"]
    call_control_id = call_data["call_control_id"]
    outbound_encoding = call_data["outbound_encoding"]
elif transport_type == "plivo":
    stream_id = call_data["stream_id"]
    call_id = call_data["call_id"]
elif transport_type == "exotel":
    stream_sid=call_data["stream_id"],
    call_sid=call_data["call_id"],

The function automatically:

• Reads and parses initial WebSocket messages
• Detects provider based on message structure
• Extracts provider-specific call information
• Returns structured data for transport configuration

Transport Helper Functions

Client ID Detection:

from pipecat.runner.utils import get_transport_client_id

client_id = get_transport_client_id(transport, client)
# Returns pc_id for WebRTC or participant ID for Daily

Video Capture (Daily only):

from pipecat.runner.utils import maybe_capture_participant_camera, maybe_capture_participant_screen

# Capture participant's camera
await maybe_capture_participant_camera(transport, client, framerate=30)

# Capture participant's screen
await maybe_capture_participant_screen(transport, client, framerate=15)

These functions safely handle transport detection and only execute if the transport supports the operation.

When to Use These Utilities

Automatic Usage (Most Common)

The development runner and create_transport utility handle these automatically. Most users won’t need to call these functions directly.

Manual Usage (Advanced)

Use these utilities directly when:

Custom deployment infrastructure: Building your own bot runner or deployment system

Advanced transport configuration: Need specific room settings, token permissions, or custom authentication

Non-runner scenarios: Integrating Pipecat transports into existing applications

Testing and debugging: Need to create rooms/tokens independently for testing

Integration Example

Here’s how you might use these utilities in a custom deployment:

import aiohttp
from pipecat.runner.daily import configure
from pipecat.runner.utils import parse_telephony_websocket
from pipecat.transports.services.daily import DailyTransport, DailyParams

async def create_custom_bot_session(transport_type: str):
    if transport_type == "daily":
        async with aiohttp.ClientSession() as session:
            room_config = await configure(session)

            room_url = room_config.room_url
            token = room_config.token

            return DailyTransport(room_url, token, "Custom Bot", DailyParams())

    elif transport_type == "telephony":
        # Handle custom telephony setup
        transport_type, call_data = await parse_telephony_websocket(websocket)
        # Configure based on detected provider...

These utilities provide the building blocks for any transport configuration scenario while maintaining the same reliability and functionality as the development runner.