# Source: https://docs.pipecat.ai/server/utilities/runner/guide.md

> ## Documentation Index
> Fetch the complete documentation index at: https://docs.pipecat.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Development Runner

> Unified runner for building voice AI bots with Daily, WebRTC, and telephony transports

## Overview

The Pipecat development runner provides a unified way to run voice AI bots across multiple transport types. It handles infrastructure setup - creating Daily rooms, managing WebRTC connections, and routing telephony calls.

## Installation

```bash  theme={null}
pip install pipecat-ai[runner]
```

## What is a Runner?

A runner in Pipecat refers to a "bot runner", an HTTP service that provides a gateway for spawning bots on-demand. It's the component that enables your bot to run by providing it with server infrastructure and connection details like rooms and tokens.

A bot runner typically creates transport sessions (like Daily WebRTC rooms), generates authentication tokens for both bots and users, spawns new bot processes when users request sessions, and manages bot lifecycle and cleanup. Think of it as the bridge between incoming user connections and your bot logic.

## How the Development Runner Works

The development runner operates as a FastAPI web server that automatically discovers and executes your bot code. When you start the runner, it creates the necessary web endpoints and infrastructure for your chosen transport type.

* **WebRTC connections**: It serves a built-in web interface where users can connect directly as well as an endpoint to create new WebRTC sessions
* **Daily integration**: It provides endpoints that create new rooms and tokens and redirect users to join them
* **Telephony providers**: For Twilio, it sets up webhook endpoints that handle incoming calls and establish WebSocket connections for audio streaming

The runner automatically detects which transport type you're using and configures the appropriate infrastructure. It then discovers your bot function and spawns new instances whenever users connect. This means you can focus on writing your bot logic while the runner handles all the server infrastructure, connection management, and transport-specific details.

Your bot code receives runner arguments that contain everything it needs, including Daily room URLs and tokens, WebRTC connections, or WebSocket streams for telephony. The runner abstracts away the complexity of managing these different connection types, providing a unified interface for building bots that work across multiple platforms.

## Pipecat Cloud Ready

The bot runner is designed to be cloud-ready, meaning that you can run the same bot code locally and deployed to Pipecat Cloud without any modifications. It automatically handles the differences in transport setup, providing you with the flexibility to test locally using an open source transport, like SmallWebRTCTransport, but run in production using Daily or telephony transports.

## Building with the Runner

Now let's build a practical example to see how this works. The key insight is that your bot code is structured into two parts: the core bot logic that works with any transport, and the entry point that creates the appropriate transport based on the runner arguments.

Here's the basic structure:

```python  theme={null}
# Your imports
from pipecat.runner.types import RunnerArguments
from pipecat.transports.base_transport import BaseTransport, TransportParams
from pipecat.transports.network.small_webrtc import SmallWebRTCTransport

async def run_bot(transport: BaseTransport):
    """Your core bot logic here:
    - Define services (STT, TTS, LLM)
    - Initialize messages and context
    - Create the pipeline
    - Add event handlers
    - Run the pipeline
    """

    # Your bot logic goes here
    # Define STT, LLM, TTS...
    # ...
    # Run your pipeline
    await runner.run(task)

async def bot(runner_args: RunnerArguments):
    """Main bot entry point compatible for local dev and Pipecat Cloud."""
    transport = SmallWebRTCTransport(
        params=TransportParams(
            audio_in_enabled=True,
            audio_out_enabled=True,
        ),
        webrtc_connection=runner_args.webrtc_connection,
    )

    await run_bot(transport)

if __name__ == "__main__":
    from pipecat.runner.run import main
    main()
```

The `run_bot()` function contains your actual bot logic and is transport-agnostic. The `bot()` function is the entry point that the runner calls - it creates the appropriate transport and passes it to your bot logic. This separation allows the same bot code to work across different transports.

When you run this with `python bot.py`, the development runner starts a web server and opens a browser interface at `http://localhost:7860/client`. Each time someone connects, the runner calls your `bot()` function with WebRTC runner arguments.

## Supporting Multiple Transports

To make your bot work across different platforms, you can detect the transport type from the runner arguments and create the appropriate transport. Here's how to support both Daily and WebRTC:

```python  theme={null}
from pipecat.runner.types import DailyRunnerArguments, RunnerArguments, SmallWebRTCRunnerArguments

async def bot(runner_args: RunnerArguments):
    """Main bot entry point compatible with Pipecat Cloud."""

    transport = None

    if isinstance(runner_args, DailyRunnerArguments):
        from pipecat.transports.daily.transport import DailyParams, DailyTransport

        transport = DailyTransport(
            runner_args.room_url,
            runner_args.token,
            "Pipecat Bot",
            params=DailyParams(
                audio_in_enabled=True,
                audio_out_enabled=True,
            ),
        )

    elif isinstance(runner_args, SmallWebRTCRunnerArguments):
        from pipecat.transports.base_transport import TransportParams
        from pipecat.transports.network.small_webrtc import SmallWebRTCTransport

        transport = SmallWebRTCTransport(
            params=TransportParams(
                audio_in_enabled=True,
                audio_out_enabled=True,
            ),
            webrtc_connection=runner_args.webrtc_connection,
        )

    else:
        logger.error(f"Unsupported runner arguments type: {type(runner_args)}")
        return

    if transport is None:
        logger.error("Failed to create transport")
        return

    await run_bot(transport)

if __name__ == "__main__":
    from pipecat.runner.run import main
    main()
```

Now you can run your bot with different transports:

```bash  theme={null}
python bot.py -t webrtc  # Uses SmallWebRTCRunnerArguments
python bot.py -t daily   # Uses DailyRunnerArguments
```

### Understanding Runner Arguments

Runner arguments are how the runner communicates transport-specific information to your bot. The runner determines which transport to use based on the command-line arguments, then creates the appropriate runner arguments:

* **`DailyRunnerArguments`**: Contains `room_url`, `token` (Optional), `body` (Optional) for joining Daily rooms
* **`SmallWebRTCRunnerArguments`**: Contains `webrtc_connection` for local WebRTC sessions
* **`WebSocketRunnerArguments`**: Contains `websocket` for telephony connections

All runner arguments also include:

* **`handle_sigint`**: Whether the bot should handle SIGINT (Ctrl+C) signals (managed automatically)
* **`handle_sigterm`**: Whether the bot should handle SIGTERM signals (managed automatically)

<Warning>
  `handle_sigint` and `handle_sigterm` are development only features and cannot
  be used when deploying to Pipecat Cloud.
</Warning>

The runner handles all the complex setup - creating Daily rooms, generating tokens, establishing WebSocket connections - and provides your bot with everything it needs through these runner arguments.

Notice how we use lazy imports (`from pipecat.transports.daily.transport import ...`) inside the conditional blocks. This ensures that transport-specific dependencies are only required when that transport is actually used, making your bot more portable.

<Info>
  `RunnerArguments` is the base class for all runner arguments. It provides a
  common interface for the runner to pass transport-specific information to your
  bot.
</Info>

## Environment Detection

When building bots that work both locally and in production, you often need to detect the execution environment to enable different features. The development runner sets the `ENV` environment variable to help with this:

```python  theme={null}
import os

async def bot(runner_args: RunnerArguments):
    # Check if running in local development environment
    is_local = os.environ.get("ENV") == "local"

    # Enable production features only when deployed
    if not is_local:
        from pipecat.audio.filters.krisp_viva_filter import KrispVivaFilter
        krisp_filter = KrispVivaFilter()
    else:
        krisp_filter = None

    transport = DailyTransport(
        runner_args.room_url,
        runner_args.token,
        "Pipecat Bot",
        params=DailyParams(
            audio_in_filter=krisp_filter,  # Krisp VIVA filter only in production
            audio_in_enabled=True,
            audio_out_enabled=True,
        ),
    )
```

### Environment Values

The development runner automatically sets environment variables based on how your bot is running:

* **Local development**: `ENV=local` (set by the development runner)
* **Production/Cloud deployment**: `ENV` is not set or has a different value

This allows you to easily customize behavior between development and production environments:

## All Supported Transports

The development runner supports six transport types, each designed for different use cases:

### WebRTC (`-t webrtc`)

Local WebRTC connections with a built-in browser interface. Perfect for development and testing.

```bash  theme={null}
python bot.py -t webrtc
# Opens http://localhost:7860/client

# ESP32 compatibility mode
python bot.py -t webrtc --esp32 --host 192.168.1.100
```

**Runner Arguments**: `SmallWebRTCRunnerArguments`

* `webrtc_connection`: Pre-configured WebRTC peer connection

### Daily (`-t daily`)

Integration with Daily for production video conferencing with rooms, participant management, and Pipecat client compatibility.

```bash  theme={null}
python bot.py -t daily
# Opens http://localhost:7860 with room creation interface
# Also provides POST /start endpoint for Pipecat clients

# Direct connection for testing (bypasses web server)
python bot.py -d

# Enable PSTN dial-in webhook handling
python bot.py -t daily --dialin
```

**Runner Arguments**: `DailyRunnerArguments`

* `room_url`: Daily room URL to join
* `token`: Authentication token for the room
* `body`: Request data from /start endpoint (dict) or dial-in webhook data

**Available Endpoints**:

* `GET /`: Web interface for creating rooms
* `POST /start`: RTVI-compatible endpoint for programmatic access
* `POST /daily-dialin-webhook`: PSTN dial-in webhook handler (requires `--dialin` flag)

### Telephony (`-t twilio|telnyx|plivo|exotel`)

Phone call integration through telephony providers. Requires a public webhook endpoint and provider credentials.

```bash  theme={null}
# Set up environment variables first
export TWILIO_ACCOUNT_SID=your_account_sid
export TWILIO_AUTH_TOKEN=your_auth_token

# Then run with public proxy
python bot.py -t twilio -x yourproxy.ngrok.io
```

**Environment Variables**:

* Twilio: `TWILIO_ACCOUNT_SID`, `TWILIO_AUTH_TOKEN`
* Telnyx: `TELNYX_API_KEY`
* Plivo: `PLIVO_AUTH_ID`, `PLIVO_AUTH_TOKEN`
* Exotel: None required (no hang-up functionality available)

Environment variables are optional, but when provided will attempt to hang up the call when the session ends.

**Runner Arguments**: `WebSocketRunnerArguments`

* `websocket`: WebSocket connection for audio streaming

The runner automatically detects the telephony provider from incoming WebSocket messages and configures the appropriate serializers and audio settings. Your bot receives a pre-configured FastAPI WebSocket connection ready for telephony audio streaming.

## Command Line Options

The development runner accepts several command-line arguments to customize its behavior:

```bash  theme={null}
python bot.py [OPTIONS]

Options:
  --host TEXT          Server host address (default: localhost)
  --port INTEGER       Server port (default: 7860)
  -t, --transport      Transport type: daily, webrtc, twilio, telnyx, plivo, exotel (default: webrtc)
  -x, --proxy TEXT     Public proxy hostname for telephony webhooks (required for telephony)
  --esp32              Enable SDP munging for ESP32 WebRTC compatibility
  -d, --direct         Connect directly to Daily room for testing (automatically sets transport to daily)
  --dialin             Enable Daily PSTN dial-in webhook handling (requires Daily transport)
  -v, --verbose        Increase logging verbosity
```

### Key Arguments

**`--transport` / `-t`**: Determines which transport infrastructure to set up

* `webrtc`: Local WebRTC with browser interface
* `daily`: Daily.co integration with room management
* `twilio`, `telnyx`, `plivo`, `exotel`: Telephony provider integration

**`--proxy` / `-x`**: Required for most telephony transports (Twilio, Telnyx, Plivo). This should be a publicly accessible hostname (like `yourbot.ngrok.io`) that can receive webhooks from telephony providers. The runner automatically strips protocol prefixes (http\://, https\://) if provided. Not required for Exotel, which uses direct WebSocket connections.

**`--direct` / `-d`**: Special mode for Daily that bypasses the web server and connects your bot directly to a Daily room. Useful for quick testing but not recommended for production use.

**`--dialin`**: Enables the `/daily-dialin-webhook` endpoint for handling Daily PSTN dial-in calls. Only works with Daily transport (`-t daily`). This endpoint receives webhook data from Daily when a phone call dials into your configured phone number, creates a SIP-enabled room, and spawns your bot.

**`--esp32`**: Enables SDP (Session Description Protocol) modifications needed for ESP32 WebRTC compatibility. Must be used with a specific IP address via `--host`.

### Environment Variables

Different transports require different environment variables:

**Daily**:

* `DAILY_API_KEY`: Daily API key for creating rooms and tokens (required for dial-in)
* `DAILY_SAMPLE_ROOM_URL` (Optional): Existing room URL to use
* `DAILY_API_URL` (Optional): Daily API URL (defaults to [https://api.daily.co/v1](https://api.daily.co/v1))

**Telephony**:

* `TWILIO_ACCOUNT_SID`, `TWILIO_AUTH_TOKEN`: Twilio credentials
* `PLIVO_AUTH_ID`, `PLIVO_AUTH_TOKEN`: Plivo credentials
* `TELNYX_API_KEY`: Telnyx API key

The runner automatically uses these environment variables when creating transport sessions and authentication tokens.

## Simplifying with the Transport Utility

While the manual approach gives you full control, the `create_transport` utility provides a much cleaner way to handle multiple transports. Instead of writing conditional logic for each transport type, you define transport configurations upfront and let the utility handle the selection:

```python  theme={null}
from pipecat.runner.utils import create_transport

# Define transport configurations using factory functions
transport_params = {
    "daily": lambda: DailyParams(
        audio_in_enabled=True,
        audio_out_enabled=True,
    ),
    "webrtc": lambda: TransportParams(
        audio_in_enabled=True,
        audio_out_enabled=True,
    ),
    "twilio": lambda: FastAPIWebsocketParams(
        audio_in_enabled=True,
        audio_out_enabled=True,
        # add_wav_header and serializer handled automatically
    ),
}

async def bot(runner_args):
    """Simplified bot entry point using the transport utility."""
    transport = await create_transport(runner_args, transport_params)
    await run_bot(transport)
```

The utility automatically:

* Detects the telephony provider from WebSocket messages
* Configures the appropriate serializer (Twilio, Telnyx, Plivo, or Exotel)
* Sets up authentication using environment variables
* Handles WebRTC and Daily transport creation

Now your bot supports all six transport types with just two lines of code in the bot() function.

## RTVI Integration

The development runner provides RTVI (Real-Time Voice Interface) compatible endpoints for the Pipecat client SDKs. This allows you to use the Pipecat client libraries locally during development as well as when deployed to Pipecat Cloud.

### The `/start` Endpoint

For Daily transports, the runner automatically creates a `/start` POST endpoint that:

1. **Creates Daily rooms and tokens** using your `DAILY_API_KEY`
2. **Spawns bot instances** with request `body` data
3. **Returns connection details** in RTVI-compatible format

```bash  theme={null}
# Start the Daily runner
python bot.py -t daily

# The /start endpoint is now available at:
# POST http://localhost:7860/start
```

**Request Format:**

```json  theme={null}
{
  "createDailyRoom": true,
  "dailyRoomProperties": {
    "start_video_off": true,
    "start_audio_off": false
  },
  "body": {
    "custom_data": "your_value",
    "user_id": "user123"
  }
}
```

<Note>`dailyRoomProperties` are not yet handled by the runner.</Note>

**Response Format:**

```json  theme={null}
{
  "dailyRoom": "https://domain.daily.co/room-name",
  "dailyToken": "eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9..."
}
```

### Accessing Request Data in Your Bot

The `body` field from the `/start` request is passed to your bot through `DailyRunnerArguments.body`:

```python  theme={null}
async def bot(runner_args: DailyRunnerArguments):
    # Access custom data from the /start request
    custom_data = runner_args.body.get("custom_data")
    user_id = runner_args.body.get("user_id")

    print(f"Bot started for user: {user_id}")
    print(f"Custom data: {custom_data}")

    # Your bot logic here
    transport = DailyTransport(runner_args.room_url, runner_args.token, "Bot")
    await run_bot(transport)
```

### RTVI Client Example

You can use RTVI client libraries to connect to your local development runner:

```javascript  theme={null}
import { PipecatClient } from "@pipecat-ai/client-js";
import { DailyTransport } from "@pipecat-ai/daily-transport";

const client = new PipecatClient({
  transport: new DailyTransport(),
  enableMic: true,
  enableCam: false,
});

// Start a session with custom data
await client.connect({ endpoint: "http://localhost:7860/start" });
```

<Info>
  The `/start` endpoint is only available for Daily transports (`-t daily`).
  WebRTC and telephony transports use different connection methods.
</Info>

## Daily Dial-In Webhook

The development runner can handle Daily PSTN dial-in webhooks when started with the `--dialin` flag. This allows you to test phone call integrations locally before deploying to production.

### Enabling Dial-In Support

```bash  theme={null}
python bot.py -t daily --dialin
# Webhook endpoint available at:
# POST http://localhost:7860/daily-dialin-webhook
```

### How It Works

When a phone call dials into your Daily phone number:

1. **Daily sends a webhook** to your configured endpoint with call details
2. **The runner creates a SIP-enabled room** with appropriate configuration
3. **Your bot is spawned** with dial-in context in `runner_args.body`
4. **The call is connected** to the room where your bot is waiting

### Webhook Payload

Daily sends the following data to the `/daily-dialin-webhook` endpoint:

```json  theme={null}
{
  "From": "+15551234567",
  "To": "+15559876543",
  "callId": "uuid-call-id",
  "callDomain": "uuid-call-domain",
  "sipHeaders": {}
}
```

### Accessing Dial-In Data in Your Bot

The dial-in webhook data is passed to your bot through `DailyRunnerArguments.body`. You need to parse it and configure the Daily transport with dial-in settings:

```python  theme={null}
from pipecat.runner.types import DailyDialinRequest, RunnerArguments
from pipecat.transports.daily.transport import DailyDialinSettings, DailyParams, DailyTransport

async def bot(runner_args: RunnerArguments):
    # Parse the dial-in request from the runner
    request = DailyDialinRequest.model_validate(runner_args.body)

    # Configure Daily transport with dial-in settings
    daily_dialin_settings = DailyDialinSettings(
        call_id=request.dialin_settings.call_id,
        call_domain=request.dialin_settings.call_domain,
    )

    transport = DailyTransport(
        runner_args.room_url,
        runner_args.token,
        "Daily PSTN Dial-in Bot",
        params=DailyParams(
            api_key=request.daily_api_key,
            api_url=request.daily_api_url,
            dialin_settings=daily_dialin_settings,
            audio_in_enabled=True,
            audio_out_enabled=True,
        ),
    )

    # Log caller information if available
    if request.dialin_settings.From:
        print(f"Handling call from: {request.dialin_settings.From}")

    # Your bot logic here
    await run_bot(transport)
```

### Configuring Daily Phone Numbers

To test dial-in locally, you need to:

1. **Configure a Daily phone number** in your Daily dashboard
2. **Set the webhook URL** to your public endpoint (e.g., via ngrok)
3. **Ensure `DAILY_API_KEY` is set** in your environment

```bash  theme={null}
# Example with ngrok
ngrok http 7860

# Configure webhook URL in Daily dashboard:
# https://your-subdomain.ngrok.io/daily-dialin-webhook
```

<Note>
  The `--dialin` flag requires the Daily transport (`-t daily`) and a valid
  `DAILY_API_KEY` environment variable.
</Note>

## Quick Reference

| Transport     | Command                                             | Access                                                       | Environment Variables                               |
| ------------- | --------------------------------------------------- | ------------------------------------------------------------ | --------------------------------------------------- |
| WebRTC        | `python bot.py`                                     | [http://localhost:7860/client](http://localhost:7860/client) | None                                                |
| Daily         | `python bot.py -t daily`                            | [http://localhost:7860](http://localhost:7860)               | `DAILY_API_KEY`, `DAILY_SAMPLE_ROOM_URL` (Optional) |
| Daily Direct  | `python bot.py -d`                                  | Direct connection                                            | `DAILY_API_KEY`, `DAILY_SAMPLE_ROOM_URL` (Optional) |
| Daily Dial-In | `python bot.py -t daily --dialin`                   | Phone calls via Daily PSTN                                   | `DAILY_API_KEY` (Required)                          |
| Twilio        | `python bot.py -t twilio -x proxy.ngrok.io`         | Phone calls                                                  | `TWILIO_ACCOUNT_SID`, `TWILIO_AUTH_TOKEN`           |
| Telnyx        | `python bot.py -t telnyx -x proxy.ngrok.io`         | Phone calls                                                  | `TELNYX_API_KEY`                                    |
| Plivo         | `python bot.py -t plivo -x proxy.ngrok.io`          | Phone calls                                                  | `PLIVO_AUTH_ID`, `PLIVO_AUTH_TOKEN`                 |
| Exotel        | `python bot.py -t exotel`                           | Phone calls                                                  | None                                                |
| ESP32 WebRTC  | `python bot.py -t webrtc --esp32 --host <ESP32_IP>` | ESP32 WebRTC connection                                      | None                                                |

## Examples

For practical examples of using the development runner with different transports, check out the following:

<Card title="Runner Examples" icon="code" href="https://github.com/pipecat-ai/pipecat-examples/tree/main/runner-examples">
  Explore the examples for different ways to use the development runner with
  various transports.
</Card>