diff --git a/src/pipecat/runner/run.py b/src/pipecat/runner/run.py
index f3965b79c..88b518c07 100644
--- a/src/pipecat/runner/run.py
+++ b/src/pipecat/runner/run.py
@@ -19,6 +19,10 @@ 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.
+By default the runner starts a single FastAPI server that supports WebRTC, Daily,
+and telephony transports simultaneously. Clients declare which transport they want
+via the ``transport`` field in the ``/start`` request body (default: ``"webrtc"``).
+
Single transport example::
async def bot(runner_args: RunnerArguments):
@@ -55,14 +59,33 @@ Supported transports:
- WebRTC - Provides local WebRTC interface with prebuilt UI
- Telephony - Handles webhook and WebSocket connections for Twilio, Telnyx, Plivo, Exotel
+The ``/start`` endpoint accepts::
+
+ {
+ "transport": "webrtc", // "webrtc" | "daily" | "twilio" | "telnyx" |
+ // "plivo" | "exotel" — default: "webrtc"
+
+ // WebRTC-specific
+ "enableDefaultIceServers": false,
+ "body": {...},
+
+ // Daily-specific
+ "createDailyRoom": true,
+ "dailyRoomProperties": {...},
+ "dailyMeetingTokenProperties": {...},
+ "body": {...}
+ }
+
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)
+- All transports (default): ``python bot.py``
+- WebRTC only: ``python bot.py -t webrtc``
+- ESP32: ``python bot.py -t webrtc --esp32 --host 192.168.1.100``
+- Daily only: ``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)
+- WhatsApp: ``python bot.py --whatsapp``
"""
import argparse
@@ -187,7 +210,7 @@ async def _run_telephony_bot(websocket: WebSocket, args: argparse.Namespace):
def _configure_server_app(args: argparse.Namespace):
- """Configure the module-level FastAPI app with transport-specific routes."""
+ """Configure the module-level FastAPI app with routes for all transports."""
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
@@ -196,17 +219,178 @@ def _configure_server_app(args: argparse.Namespace):
allow_headers=["*"],
)
- # Set up transport-specific routes
- if args.transport == "webrtc":
- _setup_webrtc_routes(app, args)
- if args.whatsapp:
- _setup_whatsapp_routes(app, args)
- elif args.transport == "daily":
- _setup_daily_routes(app, args)
- elif args.transport in TELEPHONY_TRANSPORTS:
- _setup_telephony_routes(app, args)
- else:
- logger.warning(f"Unknown transport type: {args.transport}")
+ # Shared session store: session_id -> body data. Used by the WebRTC /start
+ # flow and the /sessions/{session_id}/... proxy routes.
+ active_sessions: dict[str, dict[str, Any]] = {}
+
+ _setup_webrtc_routes(app, args, active_sessions)
+ _setup_daily_routes(app, args)
+ _setup_telephony_routes(app, args)
+ _setup_unified_start_route(app, args, active_sessions)
+
+ if args.whatsapp:
+ _setup_whatsapp_routes(app, args)
+
+
+def _setup_unified_start_route(
+ app: FastAPI, args: argparse.Namespace, active_sessions: dict[str, dict[str, Any]]
+):
+ """Register the unified POST /start endpoint.
+
+ Handles WebRTC, Daily, and telephony transport start flows. Clients specify
+ which transport they want via the ``transport`` field in the request body.
+ When ``-t`` was passed on the command line, requests for any other transport
+ are rejected with HTTP 400.
+ """
+
+ class IceServer(TypedDict, total=False):
+ urls: str | list[str]
+
+ class IceConfig(TypedDict):
+ iceServers: list[IceServer]
+
+ class StartBotResult(TypedDict, total=False):
+ sessionId: str
+ iceConfig: IceConfig | None
+ dailyRoom: str | None
+ dailyToken: str | None
+ websocketUrl: str | None
+
+ @app.post("/start")
+ async def start_agent(request: Request):
+ """Start a bot session.
+
+ Accepts::
+
+ {
+ "transport": "webrtc", // "webrtc" | "daily" | "twilio" | "telnyx" |
+ // "plivo" | "exotel" — default: "webrtc"
+
+ // WebRTC-specific
+ "enableDefaultIceServers": false,
+ "body": {...},
+
+ // Daily-specific
+ "createDailyRoom": true,
+ "dailyRoomProperties": {...},
+ "dailyMeetingTokenProperties": {...},
+ "body": {...}
+ }
+ """
+ try:
+ request_data = await request.json()
+ logger.debug(f"Received request: {request_data}")
+ except Exception as e:
+ logger.error(f"Failed to parse request body: {e}")
+ request_data = {}
+
+ # Determine transport: explicit field → legacy Daily hint → CLI default → webrtc
+ transport = request_data.get("transport")
+ if transport is None and request_data.get("createDailyRoom", False):
+ transport = "daily"
+ if transport is None:
+ transport = args.transport or "webrtc"
+
+ # Enforce restriction when -t was explicitly set on the command line
+ if args.transport is not None and transport != args.transport:
+ raise HTTPException(
+ status_code=400,
+ detail=(
+ f"Transport '{transport}' is not allowed. "
+ f"Server is configured for '{args.transport}' only (-t {args.transport})."
+ ),
+ )
+
+ if transport == "webrtc":
+ # WebRTC: register the session; the bot starts when the WebRTC offer arrives.
+ session_id = str(uuid.uuid4())
+ active_sessions[session_id] = request_data.get("body", {})
+
+ result: StartBotResult = {"sessionId": session_id}
+ if request_data.get("enableDefaultIceServers"):
+ result["iceConfig"] = IceConfig(
+ iceServers=[IceServer(urls=["stun:stun.l.google.com:19302"])]
+ )
+ return result
+
+ elif transport == "daily":
+ create_daily_room = request_data.get("createDailyRoom", False)
+ body = request_data.get("body", {})
+ daily_room_properties_dict = request_data.get("dailyRoomProperties", None)
+ daily_token_properties_dict = request_data.get("dailyMeetingTokenProperties", None)
+
+ bot_module = _get_bot_module()
+
+ existing_room_url = os.getenv("DAILY_ROOM_URL")
+ session_id = str(uuid.uuid4())
+ result = None
+
+ if create_daily_room or existing_room_url:
+ from pipecat.runner.daily import configure
+ from pipecat.transports.daily.utils import (
+ DailyMeetingTokenProperties,
+ DailyRoomProperties,
+ )
+
+ async with aiohttp.ClientSession() as session:
+ room_properties = None
+ if daily_room_properties_dict:
+ daily_room_properties_dict.setdefault(
+ "exp", time.time() + PIPECAT_ROOM_EXP_HOURS * 3600
+ )
+ daily_room_properties_dict.setdefault("eject_at_room_exp", True)
+ try:
+ room_properties = DailyRoomProperties(**daily_room_properties_dict)
+ logger.debug(f"Using custom room properties: {room_properties}")
+ except Exception as e:
+ logger.error(f"Failed to parse dailyRoomProperties: {e}")
+
+ token_properties = None
+ if daily_token_properties_dict:
+ try:
+ token_properties = DailyMeetingTokenProperties(
+ **daily_token_properties_dict
+ )
+ logger.debug(f"Using custom token properties: {token_properties}")
+ except Exception as e:
+ logger.error(f"Failed to parse dailyMeetingTokenProperties: {e}")
+
+ room_url, token = await configure(
+ session,
+ room_exp_duration=PIPECAT_ROOM_EXP_HOURS,
+ room_properties=room_properties,
+ token_properties=token_properties,
+ )
+ runner_args = DailyRunnerArguments(
+ room_url=room_url, token=token, body=body, session_id=session_id
+ )
+ result = {
+ "dailyRoom": room_url,
+ "dailyToken": token,
+ "sessionId": session_id,
+ }
+ else:
+ runner_args = RunnerArguments(body=body, session_id=session_id)
+
+ runner_args.cli_args = args
+ asyncio.create_task(bot_module.bot(runner_args))
+ return result
+
+ elif transport in TELEPHONY_TRANSPORTS:
+ # Telephony: the bot starts when the provider connects to /ws.
+ # Return the WebSocket URL so the caller knows where to point their provider.
+ session_id = str(uuid.uuid4())
+ scheme = "wss" if args.host != "localhost" else "ws"
+ return StartBotResult(
+ sessionId=session_id,
+ websocketUrl=f"{scheme}://{args.host}:{args.port}/ws",
+ )
+
+ else:
+ raise HTTPException(
+ status_code=400,
+ detail=f"Unknown transport '{transport}'.",
+ )
def _resolve_download_path(folder: str, filename: str) -> Path:
@@ -220,7 +404,9 @@ def _resolve_download_path(folder: str, filename: str) -> Path:
return file_path
-def _setup_webrtc_routes(app: FastAPI, args: argparse.Namespace):
+def _setup_webrtc_routes(
+ app: FastAPI, args: argparse.Namespace, active_sessions: dict[str, dict[str, Any]]
+):
"""Set up WebRTC-specific routes."""
try:
from pipecat_ai_small_webrtc_prebuilt.frontend import SmallWebRTCPrebuiltUI
@@ -236,19 +422,6 @@ def _setup_webrtc_routes(app: FastAPI, args: argparse.Namespace):
logger.error(f"WebRTC transport dependencies not installed: {e}")
return
- class IceServer(TypedDict, total=False):
- urls: str | list[str]
-
- class IceConfig(TypedDict):
- iceServers: list[IceServer]
-
- class StartBotResult(TypedDict, total=False):
- sessionId: str
- iceConfig: IceConfig | None
-
- # In-memory store of active sessions: session_id -> session info
- active_sessions: dict[str, dict[str, Any]] = {}
-
# Mount the frontend
app.mount("/client", SmallWebRTCPrebuiltUI)
@@ -315,29 +488,6 @@ def _setup_webrtc_routes(app: FastAPI, args: argparse.Namespace):
await small_webrtc_handler.handle_patch_request(request)
return {"status": "success"}
- @app.post("/start")
- async def rtvi_start(request: Request):
- """Mimic Pipecat Cloud's /start endpoint."""
- # Parse the request body
- try:
- request_data = await request.json()
- logger.debug(f"Received request: {request_data}")
- except Exception as e:
- logger.error(f"Failed to parse request body: {e}")
- request_data = {}
-
- # Store session info immediately in memory, replicate the behavior expected on Pipecat Cloud
- session_id = str(uuid.uuid4())
- active_sessions[session_id] = request_data.get("body", {})
-
- result: StartBotResult = {"sessionId": session_id}
- if request_data.get("enableDefaultIceServers"):
- result["iceConfig"] = IceConfig(
- iceServers=[IceServer(urls=["stun:stun.l.google.com:19302"])]
- )
-
- return result
-
@app.api_route(
"/sessions/{session_id}/{path:path}",
methods=["GET", "POST", "PUT", "PATCH", "DELETE"],
@@ -563,12 +713,10 @@ def _setup_whatsapp_routes(app: FastAPI, args: argparse.Namespace):
def _setup_daily_routes(app: FastAPI, args: argparse.Namespace):
"""Set up Daily-specific routes."""
- @app.get("/")
+ @app.get("/daily")
async def create_room_and_start_agent():
"""Launch a Daily bot and redirect to room."""
- print("Starting bot with Daily transport and redirecting to Daily room")
-
- import aiohttp
+ logger.debug("Starting bot with Daily transport and redirecting to Daily room")
from pipecat.runner.daily import configure
@@ -584,105 +732,6 @@ def _setup_daily_routes(app: FastAPI, args: argparse.Namespace):
asyncio.create_task(bot_module.bot(runner_args))
return RedirectResponse(room_url)
- @app.post("/start")
- async def start_agent(request: Request):
- """Handler for /start endpoints.
-
- Expects POST body like::
- {
- "createDailyRoom": true,
- "dailyRoomProperties": { "start_video_off": true },
- "dailyMeetingTokenProperties": { "is_owner": true, "user_name": "Bot" },
- "body": { "custom_data": "value" }
- }
- """
- print("Starting bot with Daily transport")
-
- # Parse the request body
- try:
- request_data = await request.json()
- logger.debug(f"Received request: {request_data}")
- except Exception as e:
- logger.error(f"Failed to parse request body: {e}")
- request_data = {}
-
- create_daily_room = request_data.get("createDailyRoom", False)
- body = request_data.get("body", {})
- daily_room_properties_dict = request_data.get("dailyRoomProperties", None)
- daily_token_properties_dict = request_data.get("dailyMeetingTokenProperties", None)
-
- bot_module = _get_bot_module()
-
- existing_room_url = os.getenv("DAILY_ROOM_URL")
-
- session_id = str(uuid.uuid4())
- result = None
-
- # Configure room if:
- # 1. Explicitly requested via createDailyRoom in payload
- # 2. Using pre-configured room from DAILY_ROOM_URL env var
- if create_daily_room or existing_room_url:
- import aiohttp
-
- from pipecat.runner.daily import configure
- from pipecat.transports.daily.utils import (
- DailyMeetingTokenProperties,
- DailyRoomProperties,
- )
-
- async with aiohttp.ClientSession() as session:
- # Parse dailyRoomProperties if provided
- room_properties = None
- if daily_room_properties_dict:
- # Apply Pipecat Cloud's session policy if caller didn't override.
- daily_room_properties_dict.setdefault(
- "exp", time.time() + PIPECAT_ROOM_EXP_HOURS * 3600
- )
- daily_room_properties_dict.setdefault("eject_at_room_exp", True)
- try:
- room_properties = DailyRoomProperties(**daily_room_properties_dict)
- logger.debug(f"Using custom room properties: {room_properties}")
- except Exception as e:
- logger.error(f"Failed to parse dailyRoomProperties: {e}")
- # Continue without custom properties
-
- # Parse dailyMeetingTokenProperties if provided
- token_properties = None
- if daily_token_properties_dict:
- try:
- token_properties = DailyMeetingTokenProperties(
- **daily_token_properties_dict
- )
- logger.debug(f"Using custom token properties: {token_properties}")
- except Exception as e:
- logger.error(f"Failed to parse dailyMeetingTokenProperties: {e}")
- # Continue without custom properties
-
- room_url, token = await configure(
- session,
- room_exp_duration=PIPECAT_ROOM_EXP_HOURS,
- room_properties=room_properties,
- token_properties=token_properties,
- )
- runner_args = DailyRunnerArguments(
- room_url=room_url, token=token, body=body, session_id=session_id
- )
- result = {
- "dailyRoom": room_url,
- "dailyToken": token,
- "sessionId": session_id,
- }
- else:
- runner_args = RunnerArguments(body=body, session_id=session_id)
-
- # Update CLI args.
- runner_args.cli_args = args
-
- # Start the bot in the background
- asyncio.create_task(bot_module.bot(runner_args))
-
- return result
-
if args.dialin:
@app.post("/daily-dialin-webhook")
@@ -731,8 +780,6 @@ def _setup_daily_routes(app: FastAPI, args: argparse.Namespace):
detail="Missing required fields: From, To, callId, callDomain",
)
- import aiohttp
-
from pipecat.runner.daily import configure
from pipecat.runner.types import DailyDialinRequest, DialinSettings
@@ -801,44 +848,51 @@ def _setup_daily_routes(app: FastAPI, args: argparse.Namespace):
def _setup_telephony_routes(app: FastAPI, args: argparse.Namespace):
- """Set up telephony-specific routes."""
- # XML response templates (Exotel doesn't use XML webhooks)
- XML_TEMPLATES = {
- "twilio": f"""
+ """Set up telephony-specific routes.
+
+ The WebSocket endpoint (``/ws``) is always registered so providers can
+ connect directly. The XML webhook (``POST /``) is only registered when a
+ specific telephony transport is chosen via ``-t`` because the XML template
+ is provider-specific and requires a proxy hostname (``--proxy``).
+ """
+ if args.transport in TELEPHONY_TRANSPORTS:
+ # XML response templates (Exotel doesn't use XML webhooks)
+ XML_TEMPLATES = {
+ "twilio": f"""
""",
- "telnyx": f"""
+ "telnyx": f"""
""",
- "plivo": f"""
+ "plivo": f"""
wss://{args.proxy}/ws
""",
- }
+ }
- @app.post("/")
- async def start_call():
- """Handle telephony webhook and return XML response."""
- if args.transport == "exotel":
- # Exotel doesn't use POST webhooks - redirect to proper documentation
- logger.debug("POST Exotel endpoint - not used")
- return {
- "error": "Exotel doesn't use POST webhooks",
- "websocket_url": f"wss://{args.proxy}/ws",
- "note": "Configure the WebSocket URL above in your Exotel App Bazaar Voicebot Applet",
- }
- else:
- logger.debug(f"POST {args.transport.upper()} XML")
- xml_content = XML_TEMPLATES.get(args.transport, "")
- return HTMLResponse(content=xml_content, media_type="application/xml")
+ @app.post("/")
+ async def start_call():
+ """Handle telephony webhook and return XML response."""
+ if args.transport == "exotel":
+ # Exotel doesn't use POST webhooks - redirect to proper documentation
+ logger.debug("POST Exotel endpoint - not used")
+ return {
+ "error": "Exotel doesn't use POST webhooks",
+ "websocket_url": f"wss://{args.proxy}/ws",
+ "note": "Configure the WebSocket URL above in your Exotel App Bazaar Voicebot Applet",
+ }
+ else:
+ logger.debug(f"POST {args.transport.upper()} XML")
+ xml_content = XML_TEMPLATES.get(args.transport, "")
+ return HTMLResponse(content=xml_content, media_type="application/xml")
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
@@ -847,11 +901,6 @@ def _setup_telephony_routes(app: FastAPI, args: argparse.Namespace):
logger.debug("WebSocket connection accepted")
await _run_telephony_bot(websocket, args)
- @app.get("/")
- async def start_agent():
- """Simple status endpoint for telephony transports."""
- return {"status": f"Bot started with {args.transport}"}
-
async def _run_daily_direct(args: argparse.Namespace):
"""Run Daily bot with direct connection (no FastAPI server)."""
@@ -922,22 +971,27 @@ def runner_port() -> int:
def main(parser: argparse.ArgumentParser | None = None):
"""Start the Pipecat development runner.
- Parses command-line arguments and starts a FastAPI server configured
- for the specified transport type.
+ Parses command-line arguments and starts a FastAPI server that supports
+ WebRTC, Daily, and telephony transports simultaneously. Clients declare
+ which transport to use via the ``transport`` field in the ``/start`` body.
+
+ When ``-t`` is provided, the server restricts ``/start`` to that transport
+ only and displays transport-specific startup information.
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
+ - --host: Server host address (default: localhost)
- --port: Server port (default: 7860)
- - -t/--transport: Transport type (daily, webrtc, twilio, telnyx, plivo, exotel)
+ - -t/--transport: Restrict to a single transport and set as default for /start
+ (daily, webrtc, twilio, telnyx, plivo, exotel). Omit to support all transports.
- -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)
+ - --dialin: Enable Daily PSTN dial-in webhook handling
- --esp32: Enable SDP munging for ESP32 compatibility (requires --host with IP address)
- - --whatsapp: Ensure requried WhatsApp environment variables are present
+ - --whatsapp: Ensure required WhatsApp environment variables are present
- -v/--verbose: Increase logging verbosity
Args:
@@ -958,8 +1012,11 @@ def main(parser: argparse.ArgumentParser | None = None):
"--transport",
type=str,
choices=["daily", "webrtc", *TELEPHONY_TRANSPORTS],
- default="webrtc",
- help="Transport type",
+ default=None,
+ help=(
+ "Restrict the server to a single transport and set it as the default for /start. "
+ "Omit to support all transports simultaneously (default behaviour)."
+ ),
)
parser.add_argument("-x", "--proxy", help="Public proxy host name")
parser.add_argument(
@@ -977,7 +1034,7 @@ def main(parser: argparse.ArgumentParser | None = None):
"--dialin",
action="store_true",
default=False,
- help="Enable Daily PSTN dial-in webhook handling (requires Daily transport)",
+ help="Enable Daily PSTN dial-in webhook handling",
)
parser.add_argument(
"--esp32",
@@ -989,7 +1046,7 @@ def main(parser: argparse.ArgumentParser | None = None):
"--whatsapp",
action="store_true",
default=False,
- help="Ensure requried WhatsApp environment variables are present",
+ help="Ensure required WhatsApp environment variables are present",
)
args = parser.parse_args()
@@ -998,12 +1055,13 @@ def main(parser: argparse.ArgumentParser | None = None):
if args.proxy:
args.proxy = _validate_and_clean_proxy(args.proxy)
- # Auto-set transport to daily if --direct is used without explicit transport
- if args.direct and args.transport == "webrtc": # webrtc is the default
- args.transport = "daily"
- elif args.direct and args.transport != "daily":
- logger.error("--direct flag only works with Daily transport (-t daily)")
- return
+ # --direct implies Daily transport
+ if args.direct:
+ if args.transport is None or args.transport == "daily":
+ args.transport = "daily"
+ else:
+ logger.error("--direct flag only works with Daily transport (-t daily)")
+ return
# Validate ESP32 requirements
if args.esp32 and args.host == "localhost":
@@ -1011,7 +1069,7 @@ def main(parser: argparse.ArgumentParser | None = None):
return
# Validate dial-in requirements
- if args.dialin and args.transport != "daily":
+ if args.dialin and args.transport is not None and args.transport != "daily":
logger.error("--dialin flag only works with Daily transport (-t daily)")
return
@@ -1029,28 +1087,38 @@ def main(parser: argparse.ArgumentParser | None = None):
asyncio.run(_run_daily_direct(args))
return
- # Print startup message for server-based transports
- if args.transport == "webrtc":
- print()
+ # Print startup message
+ print()
+ if args.transport is None:
+ print("🚀 Bot ready!")
+ print(f" → WebRTC: http://{args.host}:{args.port}/client")
+ print(f" → Daily: http://{args.host}:{args.port}/daily")
+ print(f" → Telephony: ws://{args.host}:{args.port}/ws")
+ elif args.transport == "webrtc":
if args.esp32:
- print(f"🚀 Bot ready! (ESP32 mode)")
+ print("🚀 Bot ready! (ESP32 mode)")
elif args.whatsapp:
- print(f"🚀 Bot ready! (WhatsApp)")
+ print("🚀 Bot ready! (WhatsApp)")
else:
- print(f"🚀 Bot ready!")
+ print("🚀 Bot ready! (WebRTC)")
print(f" → Open http://{args.host}:{args.port}/client in your browser")
- print()
elif args.transport == "daily":
- print()
- print(f"🚀 Bot ready!")
+ print("🚀 Bot ready! (Daily)")
if args.dialin:
print(
f" → Daily dial-in webhook: http://{args.host}:{args.port}/daily-dialin-webhook"
)
print(f" → Configure this URL in your Daily phone number settings")
else:
- print(f" → Open http://{args.host}:{args.port} in your browser to start a session")
- print()
+ print(
+ f" → Open http://{args.host}:{args.port}/daily in your browser to start a session"
+ )
+ elif args.transport in TELEPHONY_TRANSPORTS:
+ print(f"🚀 Bot ready! ({args.transport.capitalize()})")
+ if args.proxy:
+ print(f" → XML webhook: http://{args.host}:{args.port}/")
+ print(f" → WebSocket: ws://{args.host}:{args.port}/ws")
+ print()
RUNNER_DOWNLOADS_FOLDER = args.folder
RUNNER_HOST = args.host