diff --git a/src/pipecat/runner/__init__.py b/src/pipecat/runner/__init__.py index e69de29bb..1bd2e75e9 100644 --- a/src/pipecat/runner/__init__.py +++ b/src/pipecat/runner/__init__.py @@ -0,0 +1 @@ +"""Pipecat runner package for local and cloud bot execution.""" diff --git a/src/pipecat/runner/cloud.py b/src/pipecat/runner/cloud.py index 55dbe1916..dfcad6be2 100644 --- a/src/pipecat/runner/cloud.py +++ b/src/pipecat/runner/cloud.py @@ -4,7 +4,63 @@ # SPDX-License-Identifier: BSD 2-Clause License # -"""Cloud-compatible development server - simplified without subprocesses.""" +"""Pipecat Cloud-compatible development server for running Pipecat bots. + +This module provides a FastAPI-based development server that can run bots +structured for Pipecat Cloud deployment. It supports multiple transport types +and handles room/token management automatically. + +All bots must implement a `bot(session_args)` async function as the entry point. +The server automatically discovers and executes this function when connections +are established. + +Bot function signature:: + + async def bot(session_args): + # session_args contains transport-specific connection information + + # For Daily: session_args.room_url, session_args.token, session_args.body + # For WebRTC: session_args.webrtc_connection + # For Telephony: session_args.websocket + + # Create transport based on session_args attributes + if hasattr(session_args, 'room_url'): + # Daily transport setup + transport = DailyTransport(session_args.room_url, session_args.token, session_args.body) + elif hasattr(session_args, 'webrtc_connection'): + # WebRTC transport setup + transport = SmallWebRTCTransport(..., webrtc_connection=session_args.webrtc_connection) + + # Run your bot logic + await run_bot_logic(transport) + +Supported transports: + +- Daily - Creates rooms and tokens, runs bot as participant +- LiveKit - Creates rooms and tokens, runs bot as agent +- WebRTC - Provides local WebRTC interface with prebuilt UI +- Telephony - Handles webhook and WebSocket connections for Twilio, Telnyx, Plivo + +Example:: + + async def bot(session_args): + # Detect transport type from session_args + if hasattr(session_args, "room_url"): + # Daily or LiveKit + transport = create_daily_transport(session_args) + elif hasattr(session_args, "webrtc_connection"): + # WebRTC + transport = create_webrtc_transport(session_args) + + # Your bot implementation + await run_pipeline(transport) + + if __name__ == "__main__": + from pipecat.runner.cloud import main + main() + +Then run: `python bot.py -t daily` or `python bot.py -t webrtc` +""" import argparse import asyncio @@ -20,14 +76,13 @@ from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import RedirectResponse from loguru import logger -# Import the common transport utility functions -from .transport_utilities import setup_websocket_routes +from pipecat.runner.utils import setup_websocket_routes load_dotenv(override=True) os.environ["LOCAL_RUN"] = "1" -def get_bot_module(): +def _get_bot_module(): """Get the bot module from the calling script.""" import importlib.util @@ -68,12 +123,12 @@ def get_bot_module(): ) -async def run_bot_directly(transport_type: str, **kwargs): +async def _run_bot_directly(transport_type: str, **kwargs): """Run a bot directly in the same process - no subprocess needed.""" if transport_type == "webrtc": if "webrtc_connection" in kwargs: # Direct WebRTC connection - bot_module = get_bot_module() + bot_module = _get_bot_module() class WebRTCSessionArgs: def __init__(self, webrtc_connection): @@ -88,7 +143,7 @@ async def run_bot_directly(transport_type: str, **kwargs): elif transport_type in ["twilio", "telnyx", "plivo"]: if "websocket" in kwargs: # Direct WebSocket connection - bot_module = get_bot_module() + bot_module = _get_bot_module() class WebSocketSessionArgs: def __init__(self, transport_type, websocket, call_info): @@ -104,7 +159,7 @@ async def run_bot_directly(transport_type: str, **kwargs): await bot_module.bot(session_args) -def create_server_app(transport_type: str, host: str = "0.0.0.0", proxy: str = None): +def _create_server_app(transport_type: str, host: str = "0.0.0.0", proxy: str = None): """Create FastAPI app with transport-specific routes.""" app = FastAPI() @@ -163,13 +218,13 @@ def create_server_app(transport_type: str, host: str = "0.0.0.0", proxy: str = N # Run bot directly background_tasks.add_task( - run_bot_directly, "webrtc", webrtc_connection=pipecat_connection + _run_bot_directly, "webrtc", webrtc_connection=pipecat_connection ) answer = pipecat_connection.get_answer() if host and host != "0.0.0.0": - from .transport_utilities import smallwebrtc_sdp_munging + from pipecat.runner.utils import smallwebrtc_sdp_munging answer["sdp"] = smallwebrtc_sdp_munging(answer["sdp"], host) @@ -189,7 +244,7 @@ def create_server_app(transport_type: str, host: str = "0.0.0.0", proxy: str = N app.router.lifespan_context = lifespan elif transport_type in ["twilio", "telnyx", "plivo"]: - setup_websocket_routes(app, run_bot_directly, transport_type, proxy) + setup_websocket_routes(app, _run_bot_directly, transport_type, proxy) # Add general routes @app.get("/") @@ -201,13 +256,13 @@ def create_server_app(transport_type: str, host: str = "0.0.0.0", proxy: str = N # Create Daily room and start bot import aiohttp - from .daily_runner import configure + from pipecat.runner.daily import configure async with aiohttp.ClientSession() as session: room_url, token = await configure(session) # Start the bot in the background to join the room - bot_module = get_bot_module() + bot_module = _get_bot_module() class DailySessionArgs: def __init__(self, room_url, token): @@ -226,12 +281,12 @@ def create_server_app(transport_type: str, host: str = "0.0.0.0", proxy: str = N elif transport_type == "livekit": # Create LiveKit room and start bot - from .livekit_runner import configure + from pipecat.runner.livekit import configure url, token, room_name = await configure() # Start the bot in the background to join the room - bot_module = get_bot_module() + bot_module = _get_bot_module() class LiveKitSessionArgs: def __init__(self, url, token, room_name): @@ -262,13 +317,13 @@ def create_server_app(transport_type: str, host: str = "0.0.0.0", proxy: str = N if transport_type == "daily": import aiohttp - from .daily_runner import configure + from pipecat.runner.daily import configure async with aiohttp.ClientSession() as session: room_url, token = await configure(session) # Start the bot in the background - bot_module = get_bot_module() + bot_module = _get_bot_module() class DailySessionArgs: def __init__(self, room_url, token): @@ -294,7 +349,20 @@ def create_server_app(transport_type: str, host: str = "0.0.0.0", proxy: str = N def main(): - """Main entry point for cloud-compatible server.""" + """Start the cloud-compatible development server. + + Parses command-line arguments and starts a FastAPI server configured + for the specified transport type. The server will discover and run + any bot() function found in the current directory. + + Command-line arguments: + --host: Server host address (default: 0.0.0.0) + --port: Server port (default: 7860) + -t/--transport: Transport type (daily, livekit, webrtc, twilio, telnyx, plivo) + -x/--proxy: Public proxy hostname for telephony webhooks + + The bot file must contain a `bot(session_args)` function as the entry point. + """ parser = argparse.ArgumentParser(description="Pipecat Cloud-Compatible Development Server") parser.add_argument("--host", type=str, default="0.0.0.0", help="Host address") parser.add_argument("--port", type=int, default=7860, help="Port number") @@ -311,7 +379,7 @@ def main(): args = parser.parse_args() # Create the app with transport-specific setup - app = create_server_app(args.transport, args.host, args.proxy) + app = _create_server_app(args.transport, args.host, args.proxy) # Run the server uvicorn.run(app, host=args.host, port=args.port) diff --git a/src/pipecat/runner/daily.py b/src/pipecat/runner/daily.py index d8d6acd81..6f045e2c9 100644 --- a/src/pipecat/runner/daily.py +++ b/src/pipecat/runner/daily.py @@ -4,7 +4,30 @@ # SPDX-License-Identifier: BSD 2-Clause License # -"""Daily.co room configuration utilities for Pipecat examples.""" +"""Daily room and token configuration utilities. + +This module provides helper functions for creating and configuring Daily rooms +and authentication tokens. It handles both command-line argument parsing and +environment variable configuration. + +The module supports creating temporary rooms for development or using existing +rooms specified via arguments or environment variables. + +Required environment variables: + +- DAILY_API_KEY - Daily API key for room/token creation +- DAILY_SAMPLE_ROOM_URL (optional) - Existing room URL to use +- DAILY_SAMPLE_ROOM_TOKEN (optional) - Existing token to use + +Example:: + + import aiohttp + from pipecat.runner.daily import configure + + async with aiohttp.ClientSession() as session: + room_url, token = await configure(session) + # Use room_url and token with DailyTransport +""" import argparse import os @@ -16,7 +39,7 @@ from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper async def configure(aiohttp_session: aiohttp.ClientSession): - """Configure Daily.co room URL and token from arguments or environment. + """Configure Daily room URL and token from arguments or environment. Args: aiohttp_session: HTTP session for making API requests. @@ -34,7 +57,7 @@ async def configure(aiohttp_session: aiohttp.ClientSession): async def configure_with_args( aiohttp_session: aiohttp.ClientSession, parser: Optional[argparse.ArgumentParser] = None ): - """Configure Daily.co room with command-line argument parsing. + """Configure Daily room with command-line argument parsing. Args: aiohttp_session: HTTP session for making API requests. diff --git a/src/pipecat/runner/livekit.py b/src/pipecat/runner/livekit.py index 344f65d52..f7639cd60 100644 --- a/src/pipecat/runner/livekit.py +++ b/src/pipecat/runner/livekit.py @@ -4,7 +4,29 @@ # SPDX-License-Identifier: BSD 2-Clause License # -"""LiveKit room configuration utilities for Pipecat applications.""" +"""LiveKit room and token configuration utilities. + +This module provides helper functions for creating and configuring LiveKit +rooms and authentication tokens. It handles JWT token generation with +appropriate grants for both regular participants and AI agents. + +The module supports creating tokens for development and testing, with +automatic agent detection for proper room permissions. + +Required environment variables: + +- LIVEKIT_API_KEY - LiveKit API key +- LIVEKIT_API_SECRET - LiveKit API secret +- LIVEKIT_URL - LiveKit server URL +- LIVEKIT_ROOM_NAME - Room name to join + +Example:: + + from pipecat.runner.livekit import configure + + url, token, room_name = await configure() + # Use with LiveKitTransport +""" import argparse import os diff --git a/src/pipecat/runner/local.py b/src/pipecat/runner/local.py index 94df439a4..b83c5234e 100644 --- a/src/pipecat/runner/local.py +++ b/src/pipecat/runner/local.py @@ -4,29 +4,60 @@ # SPDX-License-Identifier: BSD 2-Clause License # -"""Direct execution runner for local-only examples.""" +"""Local-only development runner for simple Pipecat examples. + +This module provides a simplified runner for local development and testing. +It supports direct function execution without requiring the structured `bot()` +function pattern needed for cloud deployment. + +Supported transports: + +- Daily - Uses environment variables or arguments for room/token +- LiveKit - Uses environment variables or arguments for connection +- WebRTC - Provides local WebRTC interface with prebuilt UI +- Telephony - Handles webhook and WebSocket connections for Twilio, Telnyx, Plivo + +This runner is ideal for quick prototypes, examples, and bots that will only +run locally. For cloud-deployable bots, use `pipecat.runner.cloud` instead. + +Example:: + + async def run_bot(transport, args, handle_sigint): + # Your bot implementation + pass + + if __name__ == "__main__": + from pipecat.runner.local import main + + transport_params = { + "webrtc": lambda: TransportParams(...) + } + + main(run_bot, transport_params=transport_params) + +Then run: `python bot.py -t webrtc` +""" import argparse import asyncio import os import sys +from contextlib import asynccontextmanager from typing import Callable, Dict, Mapping, Optional import aiohttp import uvicorn from dotenv import load_dotenv from fastapi import BackgroundTasks, FastAPI -from fastapi.concurrency import asynccontextmanager from fastapi.responses import RedirectResponse from loguru import logger -# Import the common transport utility functions -from .transport_utilities import setup_websocket_routes +from pipecat.runner.utils import setup_websocket_routes load_dotenv(override=True) -def run_webrtc( +def _run_webrtc( run: Callable, args: argparse.Namespace, transport_params: Mapping[str, Callable] = {} ): """Run using WebRTC transport with FastAPI server.""" @@ -91,7 +122,7 @@ def run_webrtc( answer = pipecat_connection.get_answer() if args.esp32 and args.host: - from .transport_utilities import smallwebrtc_sdp_munging + from pipecat.runner.utils import smallwebrtc_sdp_munging answer["sdp"] = smallwebrtc_sdp_munging(answer["sdp"], args.host) @@ -112,16 +143,13 @@ def run_webrtc( uvicorn.run(app, host=args.host, port=args.port) -def run_twilio( +def _run_twilio( run: Callable, args: argparse.Namespace, transport_params: Mapping[str, Callable] = {} ): """Run using Twilio transport with FastAPI WebSocket server.""" try: from pipecat.serializers.twilio import TwilioFrameSerializer - from pipecat.transports.network.fastapi_websocket import ( - FastAPIWebsocketParams, - FastAPIWebsocketTransport, - ) + from pipecat.transports.network.fastapi_websocket import FastAPIWebsocketTransport except ImportError as e: logger.error(f"Twilio transport dependencies not installed.") logger.debug(f"Import error: {e}") @@ -157,16 +185,13 @@ def run_twilio( uvicorn.run(app, host=args.host, port=args.port) -def run_telnyx( +def _run_telnyx( run: Callable, args: argparse.Namespace, transport_params: Mapping[str, Callable] = {} ): """Run using Telnyx transport with FastAPI WebSocket server.""" try: from pipecat.serializers.telnyx import TelnyxFrameSerializer - from pipecat.transports.network.fastapi_websocket import ( - FastAPIWebsocketParams, - FastAPIWebsocketTransport, - ) + from pipecat.transports.network.fastapi_websocket import FastAPIWebsocketTransport except ImportError as e: logger.error(f"Telnyx transport dependencies not installed.") logger.debug(f"Import error: {e}") @@ -202,16 +227,13 @@ def run_telnyx( uvicorn.run(app, host=args.host, port=args.port) -def run_plivo( +def _run_plivo( run: Callable, args: argparse.Namespace, transport_params: Mapping[str, Callable] = {} ): """Run using Plivo transport with FastAPI WebSocket server.""" try: from pipecat.serializers.plivo import PlivoFrameSerializer - from pipecat.transports.network.fastapi_websocket import ( - FastAPIWebsocketParams, - FastAPIWebsocketTransport, - ) + from pipecat.transports.network.fastapi_websocket import FastAPIWebsocketTransport except ImportError as e: logger.error(f"Plivo transport dependencies not installed.") logger.debug(f"Import error: {e}") @@ -245,10 +267,10 @@ def run_plivo( uvicorn.run(app, host=args.host, port=args.port) -def run_daily( +def _run_daily( run: Callable, args: argparse.Namespace, transport_params: Mapping[str, Callable] = {} ): - """Run using Daily.co transport.""" + """Run using Daily transport.""" try: from pipecat.runner.daily import configure from pipecat.transports.services.daily import DailyParams, DailyTransport @@ -269,7 +291,7 @@ def run_daily( asyncio.run(run_daily_impl()) -def run_livekit( +def _run_livekit( run: Callable, args: argparse.Namespace, transport_params: Mapping[str, Callable] = {} ): """Run using LiveKit transport.""" @@ -292,7 +314,7 @@ def run_livekit( asyncio.run(run_livekit_impl()) -def run_main( +def _run_main( run: Callable, args: argparse.Namespace, transport_params: Mapping[str, Callable] = {} ): """Run the application with the specified transport type.""" @@ -302,17 +324,17 @@ def run_main( match args.transport: case "daily": - run_daily(run, args, transport_params) + _run_daily(run, args, transport_params) case "livekit": - run_livekit(run, args, transport_params) + _run_livekit(run, args, transport_params) case "plivo": - run_plivo(run, args, transport_params) + _run_plivo(run, args, transport_params) case "telnyx": - run_telnyx(run, args, transport_params) + _run_telnyx(run, args, transport_params) case "twilio": - run_twilio(run, args, transport_params) + _run_twilio(run, args, transport_params) case "webrtc": - run_webrtc(run, args, transport_params) + _run_webrtc(run, args, transport_params) def main( @@ -321,7 +343,26 @@ def main( parser: Optional[argparse.ArgumentParser] = None, transport_params: Mapping[str, Callable] = {}, ): - """Main entry point for running Pipecat applications with transport selection.""" + """Run a Pipecat bot with transport selection. + + Args: + run: The bot function to execute. Must accept (transport, args, handle_sigint). + parser: Optional argument parser. If None, creates a default one. + transport_params: Mapping of transport names to parameter factory functions. + Each factory should return transport-specific parameters when called. + + Command-line arguments: + --host: Server host address (default: localhost) + --port: Server port (default: 7860) + -t/--transport: Transport type (daily, livekit, webrtc, twilio, telnyx, plivo) + -x/--proxy: Public proxy hostname for telephony webhooks + --esp32: Enable SDP munging for ESP32 compatibility + -v/--verbose: Increase logging verbosity + + The function handles argument parsing, transport setup, and bot execution. + Different transports may use FastAPI servers (WebRTC, telephony) or direct + execution (Daily, LiveKit). + """ if not parser: parser = argparse.ArgumentParser(description="Pipecat Bot Runner") @@ -359,4 +400,4 @@ def main( logger.remove(0) logger.add(sys.stderr, level="TRACE" if args.verbose else "DEBUG") - run_main(run, args, transport_params) + _run_main(run, args, transport_params) diff --git a/src/pipecat/runner/utils.py b/src/pipecat/runner/utils.py index 3d7776662..bc3e0a9f5 100644 --- a/src/pipecat/runner/utils.py +++ b/src/pipecat/runner/utils.py @@ -4,10 +4,32 @@ # SPDX-License-Identifier: BSD 2-Clause License # -"""Common transport utility functions shared between server.py and run.py.""" +"""Transport utility functions and FastAPI route setup helpers. + +This module provides common functionality for setting up transport-specific +FastAPI routes and handling WebRTC/WebSocket connections. It includes SDP +manipulation utilities for WebRTC compatibility and transport detection helpers. + +Key features: + +- WebRTC route setup with connection management +- WebSocket route setup for telephony providers +- SDP munging for ESP32 and other WebRTC compatibility +- Transport client ID detection across different transport types +- Video capture utilities for Daily transports + +The utilities are designed to be transport-agnostic where possible, with +specific handlers for each transport type's unique requirements. + +Example:: + + from pipecat.runner.utils import setup_webrtc_routes + + app = FastAPI() + setup_webrtc_routes(app, bot_runner_function, host="localhost") +""" import json -import os import re from typing import Any, Callable, Dict