diff --git a/src/pipecat/runner/run.py b/src/pipecat/runner/run.py index 81da18ad8..80e706e2e 100644 --- a/src/pipecat/runner/run.py +++ b/src/pipecat/runner/run.py @@ -4,305 +4,39 @@ # SPDX-License-Identifier: BSD 2-Clause License # -"""Pipecat runner with support for multiple transport types. - -This module provides a unified interface for running Pipecat applications -across different transport types including Daily.co, WebRTC, and Twilio. It -handles setup, configuration, and lifecycle management for each transport type. - -Example usage: - DailyTransport:: - - python bot.py --transport daily - - LiveKitTransport:: - - python bot.py --transport livekit - - Plivo:: - - python bot.py --transport plivo --proxy username.ngrok.io - # Note: Concurrently, run an ngrok tunnel to your local server: - # ngrok http 7860 - - SmallWebRTCTransport:: - - python bot.py --transport webrtc - - Telnyx:: - - python bot.py --transport telnyx --proxy username.ngrok.io - # Note: Concurrently, run an ngrok tunnel to your local server: - # ngrok http 7860 - - Twilio:: - - python bot.py --transport twilio --proxy username.ngrok.io - # Note: Concurrently, run an ngrok tunnel to your local server: - # ngrok http 7860 -""" +"""Direct execution runner for local-only examples.""" import argparse import asyncio -import json import os -import re import sys -from contextlib import asynccontextmanager -from typing import Any, Callable, Dict, Mapping, Optional +from typing import Callable, Dict, Mapping, Optional import aiohttp import uvicorn from dotenv import load_dotenv -from fastapi import BackgroundTasks, FastAPI, WebSocket -from fastapi.middleware.cors import CORSMiddleware -from fastapi.responses import HTMLResponse, RedirectResponse +from fastapi import BackgroundTasks, FastAPI +from fastapi.concurrency import asynccontextmanager +from fastapi.responses import RedirectResponse from loguru import logger -from pipecat.transports.base_transport import BaseTransport, TransportParams +# Import the common transport utility functions +from .transport_utilities import setup_websocket_routes -# Load environment variables load_dotenv(override=True) -def get_install_command(transport: str) -> str: - """Get the pip install command for a specific transport. - - Args: - transport: The transport name. - - Returns: - The pip install command string. - """ - install_map = { - "daily": "pip install pipecat-ai[daily]", - "livekit": "pip install pipecat-ai[livekit]", - "webrtc": "pip install pipecat-ai[webrtc]", - "twilio": "pip install pipecat-ai[websocket]", - "telnyx": "pip install pipecat-ai[websocket]", - "plivo": "pip install pipecat-ai[websocket]", - } - return install_map.get(transport, f"pip install pipecat-ai[{transport}]") - - -def get_transport_client_id(transport: BaseTransport, client: Any) -> str: - """Get client identifier from transport-specific client object. - - Args: - transport: The transport instance. - client: Transport-specific client object. - - Returns: - Client identifier string, empty if transport not supported. - """ - # Import conditionally to avoid dependency issues - try: - from pipecat.transports.network.small_webrtc import SmallWebRTCTransport - - if isinstance(transport, SmallWebRTCTransport): - return client.pc_id - except ImportError: - pass - - try: - from pipecat.transports.services.daily import DailyTransport - - if isinstance(transport, DailyTransport): - return client["id"] - except ImportError: - pass - - logger.warning(f"Unable to get client id from unsupported transport {type(transport)}") - return "" - - -async def maybe_capture_participant_camera( - transport: BaseTransport, client: Any, framerate: int = 0 -): - """Capture participant camera video if transport supports it. - - Args: - transport: The transport instance. - client: Transport-specific client object. - framerate: Video capture framerate. Defaults to 0 (auto). - """ - try: - from pipecat.transports.services.daily import DailyTransport - - if isinstance(transport, DailyTransport): - await transport.capture_participant_video( - client["id"], framerate=framerate, video_source="camera" - ) - except ImportError: - pass - - -async def maybe_capture_participant_screen( - transport: BaseTransport, client: Any, framerate: int = 0 -): - """Capture participant screen video if transport supports it. - - Args: - transport: The transport instance. - client: Transport-specific client object. - framerate: Video capture framerate. Defaults to 0 (auto). - """ - try: - from pipecat.transports.services.daily import DailyTransport - - if isinstance(transport, DailyTransport): - await transport.capture_participant_video( - client["id"], framerate=framerate, video_source="screenVideo" - ) - except ImportError: - pass - - -def smallwebrtc_sdp_cleanup_ice_candidates(text: str, pattern: str) -> str: - """Clean up ICE candidates in SDP text for SmallWebRTC. - - Args: - text: SDP text to clean up. - pattern: Pattern to match for candidate filtering. - - Returns: - Cleaned SDP text with filtered ICE candidates. - """ - result = [] - lines = text.splitlines() - for line in lines: - if re.search("a=candidate", line): - if re.search(pattern, line) and not re.search("raddr", line): - result.append(line) - else: - result.append(line) - return "\r\n".join(result) - - -def smallwebrtc_sdp_cleanup_fingerprints(text: str) -> str: - """Remove unsupported fingerprint algorithms from SDP text. - - Args: - text: SDP text to clean up. - - Returns: - SDP text with sha-384 and sha-512 fingerprints removed. - """ - result = [] - lines = text.splitlines() - for line in lines: - if not re.search("sha-384", line) and not re.search("sha-512", line): - result.append(line) - return "\r\n".join(result) - - -def smallwebrtc_sdp_munging(sdp: str, host: str) -> str: - """Apply SDP modifications for SmallWebRTC compatibility. - - Args: - sdp: Original SDP string. - host: Host address for ICE candidate filtering. - - Returns: - Modified SDP string with fingerprint and ICE candidate cleanup. - """ - sdp = smallwebrtc_sdp_cleanup_fingerprints(sdp) - sdp = smallwebrtc_sdp_cleanup_ice_candidates(sdp, host) - return sdp - - -def run_daily( - run: Callable, - args: argparse.Namespace, - transport_params: Mapping[str, Callable] = {}, -): - """Run using Daily.co transport. - - Args: - run: The function to run. - args: Parsed command-line arguments. - transport_params: Mapping of transport names to parameter factory functions. - """ - try: - from pipecat.runner.daily_runner import configure - from pipecat.transports.services.daily import DailyParams, DailyTransport - except ImportError as e: - logger.error( - f"Daily transport dependencies not installed. Install with: {get_install_command('daily')}" - ) - logger.debug(f"Import error: {e}") - return - - logger.info("Running with DailyTransport...") - - async def run_daily_impl(): - async with aiohttp.ClientSession() as session: - (room_url, token) = await configure(session) - - # Run function with DailyTransport transport arguments. - params: DailyParams = transport_params[args.transport]() - transport = DailyTransport(room_url, token, "Pipecat", params=params) - await run(transport, args, True) - - asyncio.run(run_daily_impl()) - - -def run_livekit( - run: Callable, - args: argparse.Namespace, - transport_params: Mapping[str, Callable] = {}, -): - """Run using LiveKit transport. - - Args: - run: The function to run. - args: Parsed command-line arguments. - transport_params: Mapping of transport names to parameter factory functions. - """ - try: - from pipecat.runner.livekit_runner import configure - from pipecat.transports.services.livekit import LiveKitParams, LiveKitTransport - except ImportError as e: - logger.error( - f"LiveKit transport dependencies not installed. Install with: {get_install_command('livekit')}" - ) - logger.debug(f"Import error: {e}") - return - - logger.info("Running with LiveKitTransport...") - - async def run_livekit_impl(): - (url, token, room_name) = await configure() - - # Run function with LiveKit transport arguments. - params: LiveKitParams = transport_params[args.transport]() - transport = LiveKitTransport(url=url, token=token, room_name=room_name, params=params) - await run(transport, args, True) - - asyncio.run(run_livekit_impl()) - - def run_webrtc( - run: Callable, - args: argparse.Namespace, - transport_params: Mapping[str, Callable] = {}, + run: Callable, args: argparse.Namespace, transport_params: Mapping[str, Callable] = {} ): - """Run using WebRTC transport with FastAPI server. - - Args: - run: The function to run. - args: Parsed command-line arguments. - transport_params: Mapping of transport names to parameter factory functions. - """ + """Run using WebRTC transport with FastAPI server.""" try: from pipecat_ai_small_webrtc_prebuilt.frontend import SmallWebRTCPrebuiltUI from pipecat.transports.network.small_webrtc import SmallWebRTCTransport from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection except ImportError as e: - logger.error( - f"WebRTC transport dependencies not installed. Install with: {get_install_command('webrtc')}" - ) + logger.error(f"WebRTC transport dependencies not installed.") logger.debug(f"Import error: {e}") return @@ -310,7 +44,7 @@ def run_webrtc( app = FastAPI() - # Store connections by pc_id + # Store connections by pc_id (like the working version) pcs_map: Dict[str, SmallWebRTCConnection] = {} # Mount the frontend at / @@ -323,15 +57,7 @@ def run_webrtc( @app.post("/api/offer") async def offer(request: dict, background_tasks: BackgroundTasks): - """Handle WebRTC offer requests and manage peer connections. - - Args: - request: WebRTC offer request containing SDP and connection details. - background_tasks: FastAPI background tasks for running applications. - - Returns: - WebRTC answer with connection details. - """ + """Handle WebRTC offer requests and manage peer connections.""" pc_id = request.get("pc_id") if pc_id and pc_id in pcs_map: @@ -348,22 +74,25 @@ def run_webrtc( @pipecat_connection.event_handler("closed") async def handle_disconnected(webrtc_connection: SmallWebRTCConnection): - """Handle WebRTC connection closure and cleanup. - - Args: - webrtc_connection: The closed WebRTC connection. - """ + """Handle WebRTC connection closure and cleanup.""" logger.info(f"Discarding peer connection for pc_id: {webrtc_connection.pc_id}") pcs_map.pop(webrtc_connection.pc_id, None) - # Run function with SmallWebRTC transport arguments. - params: TransportParams = transport_params[args.transport]() + # Run function with SmallWebRTC transport arguments (like working version) + params = transport_params[args.transport]() transport = SmallWebRTCTransport(params=params, webrtc_connection=pipecat_connection) - background_tasks.add_task(run, transport, args, False) + + class MockArgs: + def __init__(self): + self.transport = "webrtc" + + background_tasks.add_task(run, transport, MockArgs(), False) answer = pipecat_connection.get_answer() if args.esp32 and args.host: + from .transport_utilities import smallwebrtc_sdp_munging + answer["sdp"] = smallwebrtc_sdp_munging(answer["sdp"], args.host) # Updating the peer connection inside the map @@ -373,34 +102,20 @@ def run_webrtc( @asynccontextmanager async def lifespan(app: FastAPI): - """Manage FastAPI application lifecycle and cleanup connections. - - Args: - app: The FastAPI application instance. - - Yields: - Control to the FastAPI application runtime. - """ + """Manage FastAPI application lifecycle and cleanup connections.""" yield # Run app coros = [pc.disconnect() for pc in pcs_map.values()] await asyncio.gather(*coros) pcs_map.clear() + app.router.lifespan_context = lifespan uvicorn.run(app, host=args.host, port=args.port) def run_twilio( - run: Callable, - args: argparse.Namespace, - transport_params: Mapping[str, Callable] = {}, + run: Callable, args: argparse.Namespace, transport_params: Mapping[str, Callable] = {} ): - """Run using Twilio transport with FastAPI WebSocket server. - - Args: - run: The function to run. - args: Parsed command-line arguments. - transport_params: Mapping of transport names to parameter factory functions. - """ + """Run using Twilio transport with FastAPI WebSocket server.""" try: from pipecat.serializers.twilio import TwilioFrameSerializer from pipecat.transports.network.fastapi_websocket import ( @@ -408,9 +123,7 @@ def run_twilio( FastAPIWebsocketTransport, ) except ImportError as e: - logger.error( - f"Twilio transport dependencies not installed. Install with: {get_install_command('twilio')}" - ) + logger.error(f"Twilio transport dependencies not installed.") logger.debug(f"Import error: {e}") return @@ -418,79 +131,36 @@ def run_twilio( app = FastAPI() - app.add_middleware( - CORSMiddleware, - allow_origins=["*"], # Allow all origins for testing - allow_credentials=True, - allow_methods=["*"], - allow_headers=["*"], - ) + # Twilio WebSocket handler + async def twilio_runner(transport_type: str, **kwargs): + if "websocket" in kwargs and "call_info" in kwargs: + call_info = kwargs["call_info"] - @app.post("/") - async def start_call(): - """Handle Twilio webhook and return TwiML response. + params = transport_params["twilio"]() + params.add_wav_header = False + params.serializer = TwilioFrameSerializer( + stream_sid=call_info["stream_sid"], + call_sid=call_info["call_sid"], + account_sid=os.getenv("TWILIO_ACCOUNT_SID", ""), + auth_token=os.getenv("TWILIO_AUTH_TOKEN", ""), + ) - Returns: - TwiML XML response directing call to WebSocket stream. - """ - logger.debug("POST TwiML") + transport = FastAPIWebsocketTransport(websocket=kwargs["websocket"], params=params) - xml_content = f""" - - - - - - - """ - return HTMLResponse(content=xml_content, media_type="application/xml") + class MockArgs: + def __init__(self): + self.transport = "twilio" - @app.websocket("/ws") - async def websocket_endpoint(websocket: WebSocket): - """Handle Twilio WebSocket connections for voice streaming. - - Args: - websocket: The WebSocket connection from Twilio. - """ - await websocket.accept() - - logger.debug("WebSocket connection accepted") - - # Reading Twilio data. - start_data = websocket.iter_text() - await start_data.__anext__() - call_data = json.loads(await start_data.__anext__()) - print(call_data, flush=True) - stream_sid = call_data["start"]["streamSid"] - call_sid = call_data["start"]["callSid"] - - # Create websocket transport and update params. - params: FastAPIWebsocketParams = transport_params[args.transport]() - params.add_wav_header = False - params.serializer = TwilioFrameSerializer( - stream_sid=stream_sid, - call_sid=call_sid, - account_sid=os.getenv("TWILIO_ACCOUNT_SID", ""), - auth_token=os.getenv("TWILIO_AUTH_TOKEN", ""), - ) - transport = FastAPIWebsocketTransport(websocket=websocket, params=params) - await run(transport, args, False) + await run(transport, MockArgs(), False) + setup_websocket_routes(app, twilio_runner, "twilio", args.proxy) uvicorn.run(app, host=args.host, port=args.port) def run_telnyx( - run: Callable, - args: argparse.Namespace, - transport_params: Mapping[str, Callable] = {}, + run: Callable, args: argparse.Namespace, transport_params: Mapping[str, Callable] = {} ): - """Run using Telnyx transport with FastAPI WebSocket server. - - Args: - run: The function to run. - args: Parsed command-line arguments. - transport_params: Mapping of transport names to parameter factory functions. - """ + """Run using Telnyx transport with FastAPI WebSocket server.""" try: from pipecat.serializers.telnyx import TelnyxFrameSerializer from pipecat.transports.network.fastapi_websocket import ( @@ -498,9 +168,7 @@ def run_telnyx( FastAPIWebsocketTransport, ) except ImportError as e: - logger.error( - f"Telnyx transport dependencies not installed. Install with: {get_install_command('telnyx')}" - ) + logger.error(f"Telnyx transport dependencies not installed.") logger.debug(f"Import error: {e}") return @@ -508,80 +176,36 @@ def run_telnyx( app = FastAPI() - app.add_middleware( - CORSMiddleware, - allow_origins=["*"], # Allow all origins for testing - allow_credentials=True, - allow_methods=["*"], - allow_headers=["*"], - ) + # Telnyx WebSocket handler + async def telnyx_runner(transport_type: str, **kwargs): + if "websocket" in kwargs and "call_info" in kwargs: + call_info = kwargs["call_info"] - @app.post("/") - async def start_call(): - """Handle Telnyx webhook and return TeXML response. + params = transport_params["telnyx"]() + params.add_wav_header = False + params.serializer = TelnyxFrameSerializer( + stream_id=call_info["stream_id"], + call_control_id=call_info["call_control_id"], + outbound_encoding=call_info["outbound_encoding"], + inbound_encoding="PCMU", + ) - Returns: - TeXML XML response directing call to WebSocket stream. - """ - logger.debug("POST TeXML") + transport = FastAPIWebsocketTransport(websocket=kwargs["websocket"], params=params) - xml_content = f""" - - - - - - - """ - return HTMLResponse(content=xml_content, media_type="application/xml") + class MockArgs: + def __init__(self): + self.transport = "telnyx" - @app.websocket("/ws") - async def websocket_endpoint(websocket: WebSocket): - """Handle Telnyx WebSocket connections for voice streaming. - - Args: - websocket: The WebSocket connection from Telnyx. - """ - await websocket.accept() - - logger.debug("WebSocket connection accepted") - - # Reading Telnyx data. - start_data = websocket.iter_text() - await start_data.__anext__() - call_data = json.loads(await start_data.__anext__()) - print(call_data, flush=True) - stream_id = call_data["stream_id"] - call_control_id = call_data["start"]["call_control_id"] - outbound_encoding = call_data["start"]["media_format"]["encoding"] - - # Create websocket transport and update params. - params: FastAPIWebsocketParams = transport_params[args.transport]() - params.add_wav_header = False - params.serializer = TelnyxFrameSerializer( - stream_id=stream_id, - call_control_id=call_control_id, - outbound_encoding=outbound_encoding, - inbound_encoding="PCMU", # You might want to make this configurable - ) - transport = FastAPIWebsocketTransport(websocket=websocket, params=params) - await run(transport, args, False) + await run(transport, MockArgs(), False) + setup_websocket_routes(app, telnyx_runner, "telnyx", args.proxy) uvicorn.run(app, host=args.host, port=args.port) def run_plivo( - run: Callable, - args: argparse.Namespace, - transport_params: Mapping[str, Callable] = {}, + run: Callable, args: argparse.Namespace, transport_params: Mapping[str, Callable] = {} ): - """Run using Plivo transport with FastAPI WebSocket server. - - Args: - run: The function to run. - args: Parsed command-line arguments. - transport_params: Mapping of transport names to parameter factory functions. - """ + """Run using Plivo transport with FastAPI WebSocket server.""" try: from pipecat.serializers.plivo import PlivoFrameSerializer from pipecat.transports.network.fastapi_websocket import ( @@ -589,9 +213,7 @@ def run_plivo( FastAPIWebsocketTransport, ) except ImportError as e: - logger.error( - f"Plivo transport dependencies not installed. Install with: {get_install_command('plivo')}" - ) + logger.error(f"Plivo transport dependencies not installed.") logger.debug(f"Import error: {e}") return @@ -599,89 +221,83 @@ def run_plivo( app = FastAPI() - app.add_middleware( - CORSMiddleware, - allow_origins=["*"], # Allow all origins for testing - allow_credentials=True, - allow_methods=["*"], - allow_headers=["*"], - ) + # Plivo WebSocket handler + async def plivo_runner(transport_type: str, **kwargs): + if "websocket" in kwargs and "call_info" in kwargs: + call_info = kwargs["call_info"] - @app.post("/") - async def start_call(): - """Handle Plivo webhook and return Plivo XML response. + params = transport_params["plivo"]() + params.add_wav_header = False + params.serializer = PlivoFrameSerializer( + stream_id=call_info["stream_id"], + call_id=call_info["call_id"], + ) - Returns: - Plivo XML response directing call to WebSocket stream. - """ - logger.debug("POST Plivo XML") + transport = FastAPIWebsocketTransport(websocket=kwargs["websocket"], params=params) - xml_content = f""" - - wss://{args.proxy}/ws - - """ - return HTMLResponse(content=xml_content, media_type="application/xml") + class MockArgs: + def __init__(self): + self.transport = "plivo" - @app.websocket("/ws") - async def websocket_endpoint(websocket: WebSocket): - """Handle Plivo WebSocket connections for voice streaming. - - Args: - websocket: The WebSocket connection from Plivo. - """ - await websocket.accept() - - logger.debug("WebSocket connection accepted") - - # Reading Plivo data. - start_data = websocket.iter_text() - start_message = json.loads(await start_data.__anext__()) - - logger.debug(f"Received start message: {start_message}") - - # Extract stream_id and call_id from the start event - start_info = start_message.get("start", {}) - stream_id = start_info.get("streamId") - call_id = start_info.get("callId") - - if not stream_id: - logger.error("No streamId found in start message") - await websocket.close() - return - - logger.info(f"WebSocket connection accepted for stream: {stream_id}, call: {call_id}") - - # Create websocket transport and update params. - params: FastAPIWebsocketParams = transport_params[args.transport]() - params.add_wav_header = False - params.serializer = PlivoFrameSerializer( - stream_id=stream_id, - call_id=call_id, - ) - transport = FastAPIWebsocketTransport(websocket=websocket, params=params) - await run(transport, args, False) + await run(transport, MockArgs(), False) + setup_websocket_routes(app, plivo_runner, "plivo", args.proxy) uvicorn.run(app, host=args.host, port=args.port) -def run_main( - run: Callable, - args: argparse.Namespace, - transport_params: Mapping[str, Callable] = {}, +def run_daily( + run: Callable, args: argparse.Namespace, transport_params: Mapping[str, Callable] = {} ): - """Run the application with the specified transport type. + """Run using Daily.co transport.""" + try: + from pipecat.runner.daily_runner import configure + from pipecat.transports.services.daily import DailyParams, DailyTransport + except ImportError as e: + logger.error(f"Daily transport dependencies not installed.") + logger.debug(f"Import error: {e}") + return - Args: - run: The function to run. - args: Parsed command-line arguments. - transport_params: Mapping of transport names to parameter factory functions. - """ + logger.info("Running with DailyTransport...") + + async def run_daily_impl(): + async with aiohttp.ClientSession() as session: + (room_url, token) = await configure(session) + params: DailyParams = transport_params[args.transport]() + transport = DailyTransport(room_url, token, "Pipecat", params=params) + await run(transport, args, True) + + asyncio.run(run_daily_impl()) + + +def run_livekit( + run: Callable, args: argparse.Namespace, transport_params: Mapping[str, Callable] = {} +): + """Run using LiveKit transport.""" + try: + from pipecat.runner.livekit_runner import configure + from pipecat.transports.services.livekit import LiveKitParams, LiveKitTransport + except ImportError as e: + logger.error(f"LiveKit transport dependencies not installed.") + logger.debug(f"Import error: {e}") + return + + logger.info("Running with LiveKitTransport...") + + async def run_livekit_impl(): + (url, token, room_name) = await configure() + params: LiveKitParams = transport_params[args.transport]() + transport = LiveKitTransport(url=url, token=token, room_name=room_name, params=params) + await run(transport, args, True) + + asyncio.run(run_livekit_impl()) + + +def run_main( + run: Callable, args: argparse.Namespace, transport_params: Mapping[str, Callable] = {} +): + """Run the application with the specified transport type.""" if args.transport not in transport_params: logger.error(f"Transport '{args.transport}' not supported by this application.") - logger.info( - f"To add {args.transport} support, install with: {get_install_command(args.transport)}" - ) return match args.transport: @@ -705,15 +321,10 @@ def main( parser: Optional[argparse.ArgumentParser] = None, transport_params: Mapping[str, Callable] = {}, ): - """Main entry point for running Pipecat applications with transport selection. - - Args: - run: The function to run. - parser: Optional argument parser. If None, creates a default one. - transport_params: Mapping of transport names to parameter factory functions. - """ + """Main entry point for running Pipecat applications with transport selection.""" if not parser: parser = argparse.ArgumentParser(description="Pipecat Bot Runner") + parser.add_argument( "--host", default="localhost", help="Host for HTTP server (default: localhost)" ) @@ -724,7 +335,9 @@ def main( "--transport", "-t", type=str, - choices=["daily", "livekit", "plivo", "telnyx", "twilio", "webrtc"], + choices=list(transport_params.keys()) + if transport_params + else ["daily", "livekit", "plivo", "telnyx", "twilio", "webrtc"], default="webrtc", help="The transport this application should use", ) @@ -735,6 +348,7 @@ def main( "--esp32", action="store_true", default=False, help="Perform SDP munging for the ESP32" ) parser.add_argument("--verbose", "-v", action="count", default=0) + args = parser.parse_args() if args.esp32 and args.host == "localhost": @@ -745,5 +359,4 @@ def main( logger.remove(0) logger.add(sys.stderr, level="TRACE" if args.verbose else "DEBUG") - # Import the bot file run_main(run, args, transport_params) diff --git a/src/pipecat/runner/runner.py b/src/pipecat/runner/runner.py new file mode 100644 index 000000000..fdfc74a57 --- /dev/null +++ b/src/pipecat/runner/runner.py @@ -0,0 +1,143 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import argparse +import asyncio +import importlib.util +import os +import sys + + +def find_and_import_bot(): + """Find and import the bot function from the current working directory.""" + cwd = os.getcwd() + + # Add current working directory to Python path if not already there + if cwd not in sys.path: + sys.path.insert(0, cwd) + + # Try to find bot.py in current directory + bot_file = os.path.join(cwd, "bot.py") + if os.path.exists(bot_file): + spec = importlib.util.spec_from_file_location("bot", bot_file) + bot_module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(bot_module) + return bot_module + + # Try to import bot module directly + try: + import bot + + return bot + except ImportError: + pass + + # Look for any .py file in current directory that has a bot function + for filename in os.listdir(cwd): + if filename.endswith(".py") and filename != "runner.py": + try: + module_name = filename[:-3] # Remove .py extension + spec = importlib.util.spec_from_file_location( + module_name, os.path.join(cwd, filename) + ) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + + if hasattr(module, "bot"): + return module + except Exception: + continue + + raise ImportError( + "Could not find 'bot' function. Make sure your bot file has a 'bot' function." + ) + + +def main(): + """Parse args and launch the bot with specified transport.""" + parser = argparse.ArgumentParser(description="Pipecat Bot Runner") + parser.add_argument("-u", "--url", type=str, required=False, help="Daily room URL") + parser.add_argument("-t", "--token", type=str, required=False, help="Daily room token") + parser.add_argument( + "--transport", + type=str, + choices=["daily", "livekit", "webrtc"], + default="daily", + help="Transport type", + ) + parser.add_argument("--room", type=str, required=False, help="LiveKit room name") + + args, unknown = parser.parse_known_args() + + # Find and import the bot function + try: + bot_module = find_and_import_bot() + bot_function = bot_module.bot + except ImportError as e: + print(f"Error: {e}") + sys.exit(1) + + if args.transport == "daily": + url = args.url or os.getenv("DAILY_SAMPLE_ROOM_URL") + token = args.token or os.getenv("DAILY_SAMPLE_ROOM_TOKEN") + + if not url or not token: + raise Exception("Daily room URL and token are required.") + + # Create Daily session arguments + try: + from pipecatcloud.agent import DailySessionArguments + + session_args = DailySessionArguments( + room_url=url, + token=token, + body={}, + session_id=None, + ) + except ImportError: + # Fallback for local development + class LocalDailySessionArgs: + def __init__(self, room_url, token, body=None): + self.room_url = room_url + self.token = token + self.body = body or {} + + session_args = LocalDailySessionArgs(url, token) + + elif args.transport == "livekit": + url = args.url or os.getenv("LIVEKIT_URL") + token = args.token or os.getenv("LIVEKIT_TOKEN") + room_name = args.room or os.getenv("LIVEKIT_ROOM_NAME") + + if not url or not token or not room_name: + raise Exception("LiveKit URL, token, and room name are required.") + + class LiveKitSessionArgs: + def __init__(self, url, token, room_name): + self.url = url + self.token = token + self.room_name = room_name + self.body = {} + + session_args = LiveKitSessionArgs(url, token, room_name) + + elif args.transport == "webrtc": + # For WebRTC subprocess mode (not typically used) + class WebRTCSessionArgs: + def __init__(self): + self.transport_type = "webrtc" + self.body = {} + + session_args = WebRTCSessionArgs() + + else: + raise Exception(f"Unsupported transport: {args.transport}") + + asyncio.run(bot_function(session_args)) + + +if __name__ == "__main__": + main() diff --git a/src/pipecat/runner/server.py b/src/pipecat/runner/server.py new file mode 100644 index 000000000..de2736100 --- /dev/null +++ b/src/pipecat/runner/server.py @@ -0,0 +1,303 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Cloud-compatible development server that uses subprocess to run bots.""" + +import argparse +import asyncio +import os +import subprocess +import sys +from typing import Dict + +import uvicorn +from dotenv import load_dotenv +from fastapi import BackgroundTasks, FastAPI +from fastapi.concurrency import asynccontextmanager +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_webrtc_routes, setup_websocket_routes + +load_dotenv(override=True) +os.environ["LOCAL_RUN"] = "1" + +# Track bot processes +bot_procs = {} + + +def cleanup(): + """Cleanup function to terminate all bot processes.""" + for entry in bot_procs.values(): + proc = entry[0] + proc.terminate() + proc.wait() + + +def get_bot_module(): + """Get the bot module from the calling script.""" + # Get the main module (the file that was executed) + main_module = sys.modules["__main__"] + + # Check if it has a bot function + if hasattr(main_module, "bot"): + return main_module + + # Try to import 'bot' module from current directory + try: + import bot + + return bot + except ImportError: + raise ImportError( + "Could not find 'bot' function. Make sure your script has a 'bot' function or there's a 'bot.py' file in the current directory." + ) + + +async def run_subprocess_bot(transport_type: str, **kwargs): + """Run a bot via subprocess - used by transport handlers.""" + if transport_type == "daily": + import aiohttp + + from .daily_runner import configure + + async with aiohttp.ClientSession() as session: + room_url, token = await configure(session) + proc = subprocess.Popen( + [ + f"LOCAL_RUN=1 python3 -m pipecat.runner.runner -u {room_url} -t {token} --transport daily" + ], + shell=True, + bufsize=1, + cwd=os.getcwd(), + ) # Run from current working directory + bot_procs[proc.pid] = (proc, room_url) + return room_url, token + + elif transport_type == "livekit": + from .livekit_runner import configure + + url, token, room_name = await configure() + proc = subprocess.Popen( + [ + f"LOCAL_RUN=1 python3 -m pipecat.runner.runner --transport livekit --url {url} --token {token} --room {room_name}" + ], + shell=True, + bufsize=1, + cwd=os.getcwd(), + ) # Run from current working directory + bot_procs[proc.pid] = (proc, url) + return url, token, room_name + + elif transport_type == "webrtc": + if "webrtc_connection" in kwargs: + # Direct connection mode - run bot directly + bot_module = get_bot_module() + + class WebRTCSessionArgs: + def __init__(self, webrtc_connection): + self.transport_type = "webrtc" + self.webrtc_connection = webrtc_connection + self.body = {} + self.handle_sigint = False + + session_args = WebRTCSessionArgs(kwargs["webrtc_connection"]) + await bot_module.bot(session_args) + else: + # Subprocess mode (rarely used for WebRTC) + proc = subprocess.Popen( + [f"LOCAL_RUN=1 python3 -m pipecat.runner.runner --transport webrtc"], + shell=True, + bufsize=1, + cwd=os.getcwd(), + ) # Run from current working directory + bot_procs[proc.pid] = (proc, "webrtc") + + elif transport_type in ["twilio", "telnyx", "plivo"]: + if "websocket" in kwargs: + # Direct WebSocket mode - run bot directly + bot_module = get_bot_module() + + class WebSocketSessionArgs: + def __init__(self, transport_type, websocket, call_info): + self.transport_type = transport_type + self.websocket = websocket + self.call_info = call_info + self.body = {} + self.handle_sigint = False + + session_args = WebSocketSessionArgs( + transport_type, kwargs["websocket"], kwargs["call_info"] + ) + await bot_module.bot(session_args) + else: + # Subprocess mode (rarely used for telephony) + proc = subprocess.Popen( + [f"LOCAL_RUN=1 python3 -m pipecat.runner.runner --transport {transport_type}"], + shell=True, + bufsize=1, + cwd=os.getcwd(), + ) # Run from current working directory + bot_procs[proc.pid] = (proc, transport_type) + + +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() + + app.add_middleware( + CORSMiddleware, + allow_origins=["*"], + allow_credentials=True, + allow_methods=["*"], + allow_headers=["*"], + ) + + # Add transport-specific routes + if transport_type == "webrtc": + # Direct WebRTC setup (like the working run.py version) + try: + from pipecat_ai_small_webrtc_prebuilt.frontend import SmallWebRTCPrebuiltUI + + from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection + except ImportError as e: + logger.error(f"WebRTC transport dependencies not installed.") + return app + + # Store connections by pc_id + pcs_map: Dict[str, SmallWebRTCConnection] = {} + + # Mount the frontend at / + app.mount("/client", SmallWebRTCPrebuiltUI) + + @app.get("/", include_in_schema=False) + async def root_redirect(): + """Redirect root requests to client interface.""" + return RedirectResponse(url="/client/") + + @app.post("/api/offer") + async def offer(request: dict, background_tasks: BackgroundTasks): + """Handle WebRTC offer requests and manage peer connections.""" + pc_id = request.get("pc_id") + + if pc_id and pc_id in pcs_map: + pipecat_connection = pcs_map[pc_id] + logger.info(f"Reusing existing connection for pc_id: {pc_id}") + await pipecat_connection.renegotiate( + sdp=request["sdp"], + type=request["type"], + restart_pc=request.get("restart_pc", False), + ) + else: + pipecat_connection = SmallWebRTCConnection() + await pipecat_connection.initialize(sdp=request["sdp"], type=request["type"]) + + @pipecat_connection.event_handler("closed") + async def handle_disconnected(webrtc_connection: SmallWebRTCConnection): + """Handle WebRTC connection closure and cleanup.""" + logger.info(f"Discarding peer connection for pc_id: {webrtc_connection.pc_id}") + pcs_map.pop(webrtc_connection.pc_id, None) + + # Run bot directly instead of through run_subprocess_bot + background_tasks.add_task( + run_subprocess_bot, "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 + + answer["sdp"] = smallwebrtc_sdp_munging(answer["sdp"], host) + + # Updating the peer connection inside the map + pcs_map[answer["pc_id"]] = pipecat_connection + + return answer + + @asynccontextmanager + async def lifespan(app: FastAPI): + """Manage FastAPI application lifecycle and cleanup connections.""" + yield # Run app + coros = [pc.disconnect() for pc in pcs_map.values()] + await asyncio.gather(*coros) + pcs_map.clear() + + app.router.lifespan_context = lifespan + + elif transport_type in ["twilio", "telnyx", "plivo"]: + setup_websocket_routes(app, run_subprocess_bot, transport_type, proxy) + + # Add general routes + @app.get("/") + async def start_agent(): + """Launch a bot and redirect appropriately.""" + print(f"Starting bot with {transport_type} transport") + + if transport_type == "daily": + result = await run_subprocess_bot("daily") + room_url, token = result + return RedirectResponse(room_url) + elif transport_type == "livekit": + result = await run_subprocess_bot("livekit") + url, token, room_name = result + return RedirectResponse(url) + elif transport_type == "webrtc": + await run_subprocess_bot("webrtc") + return RedirectResponse("/client/") + else: + await run_subprocess_bot(transport_type) + return {"status": f"Bot started with {transport_type}"} + + @app.post("/connect") + async def rtvi_connect(): + """Launch a bot and return connection info for RTVI clients.""" + print(f"Starting bot with {transport_type} transport") + + if transport_type == "daily": + result = await run_subprocess_bot("daily") + room_url, token = result + return {"transport": "daily", "room_url": room_url, "token": token} + elif transport_type == "webrtc": + await run_subprocess_bot("webrtc") + return {"transport": "webrtc", "client_url": "/client/"} + else: + # RTVI only supports Daily and WebRTC + return { + "error": f"RTVI connect not supported for {transport_type} transport. Use Daily or WebRTC." + } + + return app + + +def main(): + """Main entry point for cloud-compatible server.""" + 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") + parser.add_argument( + "-t", + "--transport", + type=str, + choices=["daily", "livekit", "webrtc", "twilio", "telnyx", "plivo"], + default="webrtc", + help="Transport type", + ) + parser.add_argument("--proxy", "-x", help="Public proxy host name") + + args = parser.parse_args() + + # Create the app with transport-specific setup + app = create_server_app(args.transport, args.host, args.proxy) + + # Run the server + uvicorn.run(app, host=args.host, port=args.port) + + +if __name__ == "__main__": + main() diff --git a/src/pipecat/runner/transport_utilities.py b/src/pipecat/runner/transport_utilities.py new file mode 100644 index 000000000..3d7776662 --- /dev/null +++ b/src/pipecat/runner/transport_utilities.py @@ -0,0 +1,316 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Common transport utility functions shared between server.py and run.py.""" + +import json +import os +import re +from typing import Any, Callable, Dict + +from fastapi import BackgroundTasks, FastAPI, WebSocket +from fastapi.middleware.cors import CORSMiddleware +from fastapi.responses import HTMLResponse, RedirectResponse +from loguru import logger + +from pipecat.transports.base_transport import BaseTransport + + +def get_install_command(transport: str) -> str: + """Get the pip install command for a specific transport.""" + install_map = { + "daily": "pip install pipecat-ai[daily]", + "livekit": "pip install pipecat-ai[livekit]", + "webrtc": "pip install pipecat-ai[webrtc]", + "twilio": "pip install pipecat-ai[websocket]", + "telnyx": "pip install pipecat-ai[websocket]", + "plivo": "pip install pipecat-ai[websocket]", + } + return install_map.get(transport, f"pip install pipecat-ai[{transport}]") + + +def get_transport_client_id(transport: BaseTransport, client: Any) -> str: + """Get client identifier from transport-specific client object. + + Args: + transport: The transport instance. + client: Transport-specific client object. + + Returns: + Client identifier string, empty if transport not supported. + """ + # Import conditionally to avoid dependency issues + try: + from pipecat.transports.network.small_webrtc import SmallWebRTCTransport + + if isinstance(transport, SmallWebRTCTransport): + return client.pc_id + except ImportError: + pass + + try: + from pipecat.transports.services.daily import DailyTransport + + if isinstance(transport, DailyTransport): + return client["id"] + except ImportError: + pass + + logger.warning(f"Unable to get client id from unsupported transport {type(transport)}") + return "" + + +async def maybe_capture_participant_camera( + transport: BaseTransport, client: Any, framerate: int = 0 +): + """Capture participant camera video if transport supports it. + + Args: + transport: The transport instance. + client: Transport-specific client object. + framerate: Video capture framerate. Defaults to 0 (auto). + """ + try: + from pipecat.transports.services.daily import DailyTransport + + if isinstance(transport, DailyTransport): + await transport.capture_participant_video( + client["id"], framerate=framerate, video_source="camera" + ) + except ImportError: + pass + + +async def maybe_capture_participant_screen( + transport: BaseTransport, client: Any, framerate: int = 0 +): + """Capture participant screen video if transport supports it. + + Args: + transport: The transport instance. + client: Transport-specific client object. + framerate: Video capture framerate. Defaults to 0 (auto). + """ + try: + from pipecat.transports.services.daily import DailyTransport + + if isinstance(transport, DailyTransport): + await transport.capture_participant_video( + client["id"], framerate=framerate, video_source="screenVideo" + ) + except ImportError: + pass + + +def smallwebrtc_sdp_cleanup_ice_candidates(text: str, pattern: str) -> str: + """Clean up ICE candidates in SDP text for SmallWebRTC. + + Args: + text: SDP text to clean up. + pattern: Pattern to match for candidate filtering. + + Returns: + Cleaned SDP text with filtered ICE candidates. + """ + result = [] + lines = text.splitlines() + for line in lines: + if re.search("a=candidate", line): + if re.search(pattern, line) and not re.search("raddr", line): + result.append(line) + else: + result.append(line) + return "\r\n".join(result) + + +def smallwebrtc_sdp_cleanup_fingerprints(text: str) -> str: + """Remove unsupported fingerprint algorithms from SDP text. + + Args: + text: SDP text to clean up. + + Returns: + SDP text with sha-384 and sha-512 fingerprints removed. + """ + result = [] + lines = text.splitlines() + for line in lines: + if not re.search("sha-384", line) and not re.search("sha-512", line): + result.append(line) + return "\r\n".join(result) + + +def smallwebrtc_sdp_munging(sdp: str, host: str) -> str: + """Apply SDP modifications for SmallWebRTC compatibility. + + Args: + sdp: Original SDP string. + host: Host address for ICE candidate filtering. + + Returns: + Modified SDP string with fingerprint and ICE candidate cleanup. + """ + sdp = smallwebrtc_sdp_cleanup_fingerprints(sdp) + sdp = smallwebrtc_sdp_cleanup_ice_candidates(sdp, host) + return sdp + + +def setup_webrtc_routes(app: FastAPI, transport_runner: Callable, host: str = None): + """Set up WebRTC routes for an app.""" + try: + from pipecat_ai_small_webrtc_prebuilt.frontend import SmallWebRTCPrebuiltUI + + from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection + except ImportError as e: + logger.error( + f"WebRTC transport dependencies not installed. Install with: {get_install_command('webrtc')}" + ) + logger.debug(f"Import error: {e}") + return + + # Store connections by pc_id + pcs_map: Dict[str, SmallWebRTCConnection] = {} + + # Mount the frontend at / + app.mount("/client", SmallWebRTCPrebuiltUI) + + @app.get("/", include_in_schema=False) + async def root_redirect(): + """Redirect root requests to client interface.""" + return RedirectResponse(url="/client/") + + @app.post("/api/offer") + async def offer(request: dict, background_tasks: BackgroundTasks): + """Handle WebRTC offer requests and manage peer connections.""" + pc_id = request.get("pc_id") + + if pc_id and pc_id in pcs_map: + pipecat_connection = pcs_map[pc_id] + logger.info(f"Reusing existing connection for pc_id: {pc_id}") + await pipecat_connection.renegotiate( + sdp=request["sdp"], + type=request["type"], + restart_pc=request.get("restart_pc", False), + ) + else: + pipecat_connection = SmallWebRTCConnection() + await pipecat_connection.initialize(sdp=request["sdp"], type=request["type"]) + + @pipecat_connection.event_handler("closed") + async def handle_disconnected(webrtc_connection: SmallWebRTCConnection): + """Handle WebRTC connection closure and cleanup.""" + logger.info(f"Discarding peer connection for pc_id: {webrtc_connection.pc_id}") + pcs_map.pop(webrtc_connection.pc_id, None) + + # Run transport with the connection + background_tasks.add_task( + transport_runner, "webrtc", webrtc_connection=pipecat_connection + ) + + answer = pipecat_connection.get_answer() + + if host: + answer["sdp"] = smallwebrtc_sdp_munging(answer["sdp"], host) + + # Updating the peer connection inside the map + pcs_map[answer["pc_id"]] = pipecat_connection + + return answer + + +def setup_websocket_routes( + app: FastAPI, transport_runner: Callable, transport_type: str, proxy: str = None +): + """Set up WebSocket routes for telephony providers.""" + app.add_middleware( + CORSMiddleware, + allow_origins=["*"], + allow_credentials=True, + allow_methods=["*"], + allow_headers=["*"], + ) + + @app.post("/") + async def start_call(): + """Handle telephony webhook and return XML response.""" + logger.debug(f"POST {transport_type.upper()} XML") + + if transport_type == "twilio": + xml_content = f""" + + + + + +""" + elif transport_type == "telnyx": + xml_content = f""" + + + + + +""" + elif transport_type == "plivo": + xml_content = f""" + + wss://{proxy}/ws +""" + else: + xml_content = "" + + return HTMLResponse(content=xml_content, media_type="application/xml") + + @app.websocket("/ws") + async def websocket_endpoint(websocket: WebSocket): + """Handle WebSocket connections for telephony.""" + await websocket.accept() + logger.debug("WebSocket connection accepted") + + # Parse transport-specific data + start_data = websocket.iter_text() + + if transport_type == "twilio": + await start_data.__anext__() + call_data = json.loads(await start_data.__anext__()) + print(call_data, flush=True) + stream_sid = call_data["start"]["streamSid"] + call_sid = call_data["start"]["callSid"] + call_info = {"stream_sid": stream_sid, "call_sid": call_sid} + + elif transport_type == "telnyx": + await start_data.__anext__() + call_data = json.loads(await start_data.__anext__()) + print(call_data, flush=True) + stream_id = call_data["stream_id"] + call_control_id = call_data["start"]["call_control_id"] + outbound_encoding = call_data["start"]["media_format"]["encoding"] + call_info = { + "stream_id": stream_id, + "call_control_id": call_control_id, + "outbound_encoding": outbound_encoding, + } + + elif transport_type == "plivo": + start_message = json.loads(await start_data.__anext__()) + logger.debug(f"Received start message: {start_message}") + + start_info = start_message.get("start", {}) + stream_id = start_info.get("streamId") + call_id = start_info.get("callId") + + if not stream_id: + logger.error("No streamId found in start message") + await websocket.close() + return + + logger.info(f"WebSocket connection accepted for stream: {stream_id}, call: {call_id}") + call_info = {"stream_id": stream_id, "call_id": call_id} + else: + call_info = {} + + # Run transport with the websocket connection + await transport_runner(transport_type, websocket=websocket, call_info=call_info)