diff --git a/src/pipecat/runner/cloud.py b/src/pipecat/runner/cloud.py index 01dd983dc..aedae4669 100644 --- a/src/pipecat/runner/cloud.py +++ b/src/pipecat/runner/cloud.py @@ -7,10 +7,11 @@ """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. +structured for Pipecat Cloud deployment. The runner enables you to run Pipecat +bots locally or deployed without requiring any code changes. It supports +multiple transport types and handles room/token management automatically. -It requires the pipecatcloud package for proper session argument types. +It requires the `pipecatcloud` package for proper session argument types. Install with:: @@ -72,7 +73,13 @@ Example:: from pipecat.runner.cloud import main main() -Then run: `python bot.py -t daily` or `python bot.py -t webrtc` +To run locally: + +- Daily: `python bot.py -t daily` +- WebRTC: `python bot.py -t webrtc` +- Telephony: `python bot.py -t twilio -x your_username.ngork.io` +- Telephony: `python bot.py -t telnyx -x your_username.ngork.io` +- Telephony: `python bot.py -t plivo -x your_username.ngork """ import argparse @@ -90,7 +97,6 @@ from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import HTMLResponse, RedirectResponse from loguru import logger -# Require pipecatcloud for cloud-compatible bots try: from pipecatcloud.agent import DailySessionArguments, WebSocketSessionArguments except ImportError: @@ -180,169 +186,171 @@ def _create_server_app(transport_type: str, host: str = "localhost", proxy: str allow_headers=["*"], ) - # Add transport-specific routes + # Set up transport-specific routes if transport_type == "webrtc": - # WebRTC setup - 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) - - bot_module = _get_bot_module() - - session_args = SmallWebRTCSessionArguments( - webrtc_connection=pipecat_connection, - session_id=None, - ) - background_tasks.add_task(bot_module.bot, session_args) - - answer = pipecat_connection.get_answer() - 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 - + _setup_webrtc_routes(app) + elif transport_type == "daily": + _setup_daily_routes(app) elif transport_type in ["twilio", "telnyx", "plivo"]: - # Direct telephony WebSocket handling (no utils dependency) + _setup_telephony_routes(app, transport_type, proxy) + else: + logger.warning(f"Unknown transport type: {transport_type}") - @app.post("/") - async def start_call(): - """Handle telephony webhook and return XML response.""" - logger.debug(f"POST {transport_type.upper()} XML") + return app - if transport_type == "twilio": - xml_content = f""" + +def _setup_webrtc_routes(app: FastAPI): + """Set up WebRTC-specific routes.""" + 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 + + # Store connections by pc_id + pcs_map: Dict[str, SmallWebRTCConnection] = {} + + # Mount the frontend + 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) + + bot_module = _get_bot_module() + session_args = SmallWebRTCSessionArguments( + webrtc_connection=pipecat_connection, + session_id=None, + ) + background_tasks.add_task(bot_module.bot, session_args) + + answer = pipecat_connection.get_answer() + pcs_map[answer["pc_id"]] = pipecat_connection + return answer + + @asynccontextmanager + async def lifespan(app: FastAPI): + """Manage FastAPI application lifecycle and cleanup connections.""" + yield + coros = [pc.disconnect() for pc in pcs_map.values()] + await asyncio.gather(*coros) + pcs_map.clear() + + app.router.lifespan_context = lifespan + + +def _setup_daily_routes(app: FastAPI): + """Set up Daily-specific routes.""" + + @app.get("/") + async def start_agent(): + """Launch a Daily bot and redirect to room.""" + print("Starting bot with Daily transport") + + import aiohttp + + 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() + session_args = DailySessionArguments( + room_url=room_url, token=token, body={}, session_id=None + ) + asyncio.create_task(bot_module.bot(session_args)) + return RedirectResponse(room_url) + + @app.post("/connect") + async def rtvi_connect(): + """Launch a Daily bot and return connection info for RTVI clients.""" + print("Starting bot with Daily transport") + + import aiohttp + + 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() + session_args = DailySessionArguments( + room_url=room_url, token=token, body={}, session_id=None + ) + asyncio.create_task(bot_module.bot(session_args)) + return {"room_url": room_url, "token": token} + + +def _setup_telephony_routes(app: FastAPI, transport_type: str, proxy: str): + """Set up telephony-specific routes.""" + # XML response templates + XML_TEMPLATES = { + "twilio": f""" -""" - elif transport_type == "telnyx": - xml_content = f""" +""", + "telnyx": f""" -""" - elif transport_type == "plivo": - xml_content = f""" +""", + "plivo": f""" wss://{proxy}/ws -""" - else: - xml_content = "" +""", + } - return HTMLResponse(content=xml_content, media_type="application/xml") + @app.post("/") + async def start_call(): + """Handle telephony webhook and return XML response.""" + logger.debug(f"POST {transport_type.upper()} XML") + xml_content = XML_TEMPLATES.get(transport_type, "") + 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") - await _run_telephony_bot(websocket) + @app.websocket("/ws") + async def websocket_endpoint(websocket: WebSocket): + """Handle WebSocket connections for telephony.""" + await websocket.accept() + logger.debug("WebSocket connection accepted") + await _run_telephony_bot(websocket) - # 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": - # Create Daily room and start bot - import aiohttp - - 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() - session_args = DailySessionArguments( - room_url=room_url, token=token, body={}, session_id=None - ) - asyncio.create_task(bot_module.bot(session_args)) - return RedirectResponse(room_url) - - elif transport_type == "webrtc": - return RedirectResponse("/client/") - else: - 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": - import aiohttp - - 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() - session_args = DailySessionArguments( - room_url=room_url, token=token, body={}, session_id=None - ) - asyncio.create_task(bot_module.bot(session_args)) - return {"room_url": room_url, "token": token} - - else: - return { - "error": f"RTVI connect not supported for {transport_type} transport. Use Daily." - } - - return app + """Simple status endpoint for telephony transports.""" + return {"status": f"Bot started with {transport_type}"} def main(): diff --git a/src/pipecat/runner/local.py b/src/pipecat/runner/local.py index 402150927..c10a6363b 100644 --- a/src/pipecat/runner/local.py +++ b/src/pipecat/runner/local.py @@ -57,13 +57,10 @@ from loguru import logger load_dotenv(override=True) -def setup_websocket_routes( +def _setup_websocket_routes( app: FastAPI, transport_runner: Callable, transport_type: str, proxy: str = None ): - """Set up WebSocket routes for telephony providers. - - This is used by the local runner (pipecat.runner.local) only. - """ + """Set up WebSocket routes for telephony providers.""" app.add_middleware( CORSMiddleware, allow_origins=["*"], @@ -279,7 +276,7 @@ def _run_twilio( await run(transport, MockArgs(), False) - setup_websocket_routes(app, twilio_runner, "twilio", args.proxy) + _setup_websocket_routes(app, twilio_runner, "twilio", args.proxy) uvicorn.run(app, host=args.host, port=args.port) @@ -321,7 +318,7 @@ def _run_telnyx( await run(transport, MockArgs(), False) - setup_websocket_routes(app, telnyx_runner, "telnyx", args.proxy) + _setup_websocket_routes(app, telnyx_runner, "telnyx", args.proxy) uvicorn.run(app, host=args.host, port=args.port) @@ -361,7 +358,7 @@ def _run_plivo( await run(transport, MockArgs(), False) - setup_websocket_routes(app, plivo_runner, "plivo", args.proxy) + _setup_websocket_routes(app, plivo_runner, "plivo", args.proxy) uvicorn.run(app, host=args.host, port=args.port) diff --git a/src/pipecat/runner/utils.py b/src/pipecat/runner/utils.py index 0617b1267..80963d156 100644 --- a/src/pipecat/runner/utils.py +++ b/src/pipecat/runner/utils.py @@ -42,7 +42,7 @@ from pipecat.transports.base_transport import BaseTransport def detect_transport_type_from_message(message_data: dict) -> str: """Attempt to auto-detect transport type from WebSocket message structure.""" - logger.debug("=== Auto-Detection Analysis ===") + logger.trace("=== Auto-Detection Analysis ===") # Twilio detection if ( @@ -51,7 +51,7 @@ def detect_transport_type_from_message(message_data: dict) -> str: and "streamSid" in message_data.get("start", {}) and "callSid" in message_data.get("start", {}) ): - logger.debug("Auto-detected: TWILIO") + logger.trace("Auto-detected: TWILIO") return "twilio" # Telnyx detection @@ -60,7 +60,7 @@ def detect_transport_type_from_message(message_data: dict) -> str: and "start" in message_data and "call_control_id" in message_data.get("start", {}) ): - logger.debug("Auto-detected: TELNYX") + logger.trace("Auto-detected: TELNYX") return "telnyx" # Plivo detection @@ -69,10 +69,10 @@ def detect_transport_type_from_message(message_data: dict) -> str: and "streamId" in message_data.get("start", {}) and "callId" in message_data.get("start", {}) ): - logger.debug("Auto-detected: PLIVO") + logger.trace("Auto-detected: PLIVO") return "plivo" - logger.debug("Auto-detection failed - unknown format") + logger.trace("Auto-detection failed - unknown format") return "unknown" @@ -82,15 +82,13 @@ async def parse_telephony_websocket(websocket: WebSocket): Returns: tuple: (transport_type: str, stream_id: str, call_id: str) """ - logger.info("=== Parsing Telephony WebSocket ===") - # Read first two messages start_data = websocket.iter_text() try: # First message first_message_raw = await start_data.__anext__() - logger.debug(f"First message: {first_message_raw}") + logger.trace(f"First message: {first_message_raw}") try: first_message = json.loads(first_message_raw) except json.JSONDecodeError: @@ -98,7 +96,7 @@ async def parse_telephony_websocket(websocket: WebSocket): # Second message second_message_raw = await start_data.__anext__() - logger.debug(f"Second message: {second_message_raw}") + logger.trace(f"Second message: {second_message_raw}") try: second_message = json.loads(second_message_raw) except json.JSONDecodeError: @@ -112,11 +110,11 @@ async def parse_telephony_websocket(websocket: WebSocket): if detected_type_first != "unknown": transport_type = detected_type_first call_data = first_message - logger.info(f"Detected transport: {transport_type} (from first message)") + logger.debug(f"Detected transport: {transport_type} (from first message)") elif detected_type_second != "unknown": transport_type = detected_type_second call_data = second_message - logger.info(f"Detected transport: {transport_type} (from second message)") + logger.debug(f"Detected transport: {transport_type} (from second message)") else: transport_type = "unknown" call_data = second_message @@ -141,7 +139,7 @@ async def parse_telephony_websocket(websocket: WebSocket): stream_id = None call_id = None - logger.info(f"Parsed - Type: {transport_type}, StreamId: {stream_id}, CallId: {call_id}") + logger.debug(f"Parsed - Type: {transport_type}, StreamId: {stream_id}, CallId: {call_id}") return transport_type, stream_id, call_id except Exception as e: