run

Pipecat development runner.

This development runner executes Pipecat bots and provides the supporting infrastructure they need - creating Daily rooms and tokens, managing WebRTC connections, and setting up telephony webhook/WebSocket infrastructure. It supports multiple transport types with a unified interface.

Install with:

pip install pipecat-ai[runner]

All bots must implement a bot(runner_args) async function as the entry point. The server automatically discovers and executes this function when connections are established.

Single transport example:

async def bot(runner_args: RunnerArguments):
    transport = DailyTransport(
        runner_args.room_url,
        runner_args.token,
        "Bot",
        DailyParams(...)
    )
    # Your bot logic here
    await run_pipeline(transport)

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

Multiple transport example:

async def bot(runner_args: RunnerArguments):
    # Type-safe transport detection
    if isinstance(runner_args, DailyRunnerArguments):
        transport = setup_daily_transport(runner_args)  # Your application code
    elif isinstance(runner_args, SmallWebRTCRunnerArguments):
        transport = setup_webrtc_transport(runner_args)  # Your application code
    elif isinstance(runner_args, WebSocketRunnerArguments):
        transport = setup_telephony_transport(runner_args)  # Your application code

    # Your bot implementation
    await run_pipeline(transport)

Supported transports:

  • Daily - Creates rooms and tokens, runs bot as participant

  • WebRTC - Provides local WebRTC interface with prebuilt UI

  • Telephony - Handles webhook and WebSocket connections for Twilio, Telnyx, Plivo, Exotel

To run locally:

  • WebRTC: python bot.py -t webrtc

  • ESP32: python bot.py -t webrtc –esp32 –host 192.168.1.100

  • Daily (server): python bot.py -t daily

  • Daily (direct, testing only): python bot.py -d

  • Telephony: python bot.py -t twilio -x your_username.ngrok.io

  • Exotel: python bot.py -t exotel (no proxy needed, but ngrok connection to HTTP 7860 is required)

pipecat.runner.run.app: fastapi.FastAPI

The FastAPI application instance.

Import this to add custom routes from other packages before calling main():

from pipecat.runner.run import app, main

@app.get("/my-route")
async def my_route():
    return {"hello": "world"}

if __name__ == "__main__":
    main()
pipecat.runner.run.runner_downloads_folder() str | None[source]

Returns the folder where files are stored for later download.

pipecat.runner.run.runner_host() str[source]

Returns the host name of this runner.

pipecat.runner.run.runner_port() int[source]

Returns the port of this runner.

pipecat.runner.run.main(parser: ArgumentParser | None = None)[source]

Start the Pipecat development runner.

Parses command-line arguments and starts a FastAPI server configured for the specified transport type.

The runner discovers and runs any bot(runner_args) function found in the calling module.

Command-line arguments:
  • –host: Server host address (default: localhost) 879

  • –port: Server port (default: 7860)

  • -t/–transport: Transport type (daily, webrtc, twilio, telnyx, plivo, exotel)

  • -x/–proxy: Public proxy hostname for telephony webhooks

  • -d/–direct: Connect directly to Daily room (automatically sets transport to daily)

  • -f/–folder: Path to downloads folder

  • –dialin: Enable Daily PSTN dial-in webhook handling (requires Daily transport)

  • –esp32: Enable SDP munging for ESP32 compatibility (requires –host with IP address)

  • –whatsapp: Ensure requried WhatsApp environment variables are present

  • -v/–verbose: Increase logging verbosity

Parameters:

parser – Optional custom argument parser. If provided, default runner arguments are added to it so bots can define their own CLI arguments. Custom arguments should not conflict with the default ones. Custom args are accessible via runner_args.cli_args.