​
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.co 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:
    room_url, token = await configure(session)

    # Use with DailyTransport
    transport = DailyTransport(room_url, token, "Bot Name", params=DailyParams())
This function:
  • Uses DAILY_SAMPLE_ROOM_URL environment variable if set, otherwise creates a new room
  • Generates an authentication token using DAILY_API_KEY
  • Returns a room URL and token ready for use

​
Configuration with Arguments

For command-line integration, use configure_with_args(): 
import argparse
from pipecat.runner.daily import configure_with_args

parser = argparse.ArgumentParser()
# Your other arguments...

async with aiohttp.ClientSession() as session:
    room_url, token, args = await configure_with_args(session, parser)
This supports additional command-line options:
  • -u, --url: Specify a Daily room URL directly
  • -k, --apikey: Override the 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)

​
Token Management

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

​
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"]
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_url, token = await configure(session)
            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.