diff --git a/examples/runner-examples/cloud-multi-bot.py b/examples/runner-examples/01-all-transport-bot.py similarity index 98% rename from examples/runner-examples/cloud-multi-bot.py rename to examples/runner-examples/01-all-transport-bot.py index ef9c8de89..cd60377e1 100644 --- a/examples/runner-examples/cloud-multi-bot.py +++ b/examples/runner-examples/01-all-transport-bot.py @@ -26,7 +26,7 @@ from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext from pipecat.processors.frameworks.rtvi import RTVIConfig, RTVIObserver, RTVIProcessor -from pipecat.runner.cloud import SmallWebRTCSessionArguments +from pipecat.runner.run import SmallWebRTCSessionArguments from pipecat.services.cartesia.tts import CartesiaTTSService from pipecat.services.deepgram.stt import DeepgramSTTService from pipecat.services.openai.llm import OpenAILLMService @@ -206,6 +206,6 @@ async def bot( if __name__ == "__main__": - from pipecat.runner.cloud import main + from pipecat.runner.run import main main() diff --git a/examples/runner-examples/cloud-simple-bot.py b/examples/runner-examples/02-two-transport-bot.py similarity index 97% rename from examples/runner-examples/cloud-simple-bot.py rename to examples/runner-examples/02-two-transport-bot.py index 58f62b6e3..d7757ad0c 100644 --- a/examples/runner-examples/cloud-simple-bot.py +++ b/examples/runner-examples/02-two-transport-bot.py @@ -19,7 +19,7 @@ from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext from pipecat.processors.frameworks.rtvi import RTVIConfig, RTVIObserver, RTVIProcessor -from pipecat.runner.cloud import SmallWebRTCSessionArguments +from pipecat.runner.run import SmallWebRTCSessionArguments from pipecat.services.cartesia.tts import CartesiaTTSService from pipecat.services.deepgram.stt import DeepgramSTTService from pipecat.services.openai.llm import OpenAILLMService @@ -141,6 +141,6 @@ async def bot(session_args: DailySessionArguments | SmallWebRTCSessionArguments) if __name__ == "__main__": - from pipecat.runner.cloud import main + from pipecat.runner.run import main main() diff --git a/examples/runner-examples/cloud-simple-bot2.py b/examples/runner-examples/03-two-transport-boy.py similarity index 98% rename from examples/runner-examples/cloud-simple-bot2.py rename to examples/runner-examples/03-two-transport-boy.py index 89f2cd319..7665f43fb 100644 --- a/examples/runner-examples/cloud-simple-bot2.py +++ b/examples/runner-examples/03-two-transport-boy.py @@ -19,7 +19,7 @@ from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext from pipecat.processors.frameworks.rtvi import RTVIConfig, RTVIObserver, RTVIProcessor -from pipecat.runner.cloud import SmallWebRTCSessionArguments +from pipecat.runner.run import SmallWebRTCSessionArguments from pipecat.services.cartesia.tts import CartesiaTTSService from pipecat.services.deepgram.stt import DeepgramSTTService from pipecat.services.openai.llm import OpenAILLMService @@ -160,6 +160,6 @@ async def bot( if __name__ == "__main__": - from pipecat.runner.cloud import main + from pipecat.runner.run import main main() diff --git a/examples/runner-examples/Dockerfile b/examples/runner-examples/Dockerfile index 70a179871..01ca0b133 100644 --- a/examples/runner-examples/Dockerfile +++ b/examples/runner-examples/Dockerfile @@ -4,4 +4,4 @@ COPY ./requirements.txt requirements.txt RUN pip install --no-cache-dir --upgrade -r requirements.txt -COPY ./cloud-simple-bot.py bot.py +COPY ./02-two-transport-bot.py bot.py diff --git a/examples/runner-examples/local-multi-bot.py b/examples/runner-examples/local-multi-bot.py deleted file mode 100644 index b542c6c15..000000000 --- a/examples/runner-examples/local-multi-bot.py +++ /dev/null @@ -1,122 +0,0 @@ -# -# Copyright (c) 2024–2025, Daily -# -# SPDX-License-Identifier: BSD 2-Clause License -# - -import argparse -import os - -from dotenv import load_dotenv -from loguru import logger - -from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.pipeline.pipeline import Pipeline -from pipecat.pipeline.runner import PipelineRunner -from pipecat.pipeline.task import PipelineParams, PipelineTask -from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext -from pipecat.processors.frameworks.rtvi import RTVIConfig, RTVIObserver, RTVIProcessor -from pipecat.services.cartesia.tts import CartesiaTTSService -from pipecat.services.deepgram.stt import DeepgramSTTService -from pipecat.services.openai.llm import OpenAILLMService -from pipecat.transports.base_transport import BaseTransport, TransportParams - -load_dotenv(override=True) - - -def create_transport_params(transport_name): - """Create transport parameters based on transport name.""" - base_config = { - "audio_in_enabled": True, - "audio_out_enabled": True, - "vad_analyzer": SileroVADAnalyzer(), - } - - if transport_name == "daily": - from pipecat.transports.services.daily import DailyParams - - return DailyParams(**base_config) - elif transport_name == "livekit": - from pipecat.transports.services.livekit import LiveKitParams - - return LiveKitParams(**base_config) - elif transport_name in ["plivo", "telnyx", "twilio"]: - from pipecat.transports.network.fastapi_websocket import FastAPIWebsocketParams - - return FastAPIWebsocketParams(**base_config) - else: # webrtc - return TransportParams(**base_config) - - -async def run_bot(transport: BaseTransport, _: argparse.Namespace, handle_sigint: bool): - logger.info(f"Starting bot") - - stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) - - tts = CartesiaTTSService( - api_key=os.getenv("CARTESIA_API_KEY"), - voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady - ) - - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) - - messages = [ - { - "role": "system", - "content": "You are a friendly AI assistant. Respond naturally and keep your answers conversational.", - }, - ] - - context = OpenAILLMContext(messages) - context_aggregator = llm.create_context_aggregator(context) - - rtvi = RTVIProcessor(config=RTVIConfig(config=[])) - - pipeline = Pipeline( - [ - transport.input(), # Transport user input - rtvi, - stt, - context_aggregator.user(), # User responses - llm, # LLM - tts, # TTS - transport.output(), # Transport bot output - context_aggregator.assistant(), # Assistant spoken responses - ] - ) - - task = PipelineTask( - pipeline, - params=PipelineParams( - enable_metrics=True, - enable_usage_metrics=True, - ), - observers=[RTVIObserver(rtvi)], - ) - - @transport.event_handler("on_client_connected") - async def on_client_connected(transport, client): - logger.info(f"Client connected") - # Kick off the conversation. - messages.append({"role": "system", "content": "Say hello and briefly introduce yourself."}) - await task.queue_frames([context_aggregator.user().get_context_frame()]) - - @transport.event_handler("on_client_disconnected") - async def on_client_disconnected(transport, client): - logger.info(f"Client disconnected") - await task.cancel() - - runner = PipelineRunner(handle_sigint=handle_sigint) - - await runner.run(task) - - -if __name__ == "__main__": - from pipecat.runner.local import main - - transport_params = { - transport: lambda t=transport: create_transport_params(t) - for transport in ["daily", "livekit", "plivo", "telnyx", "twilio", "webrtc"] - } - - main(run_bot, transport_params=transport_params) diff --git a/examples/runner-examples/local-simple-bot.py b/examples/runner-examples/local-simple-bot.py deleted file mode 100644 index c31e498ac..000000000 --- a/examples/runner-examples/local-simple-bot.py +++ /dev/null @@ -1,101 +0,0 @@ -# -# Copyright (c) 2024–2025, Daily -# -# SPDX-License-Identifier: BSD 2-Clause License -# - -import argparse -import os - -from dotenv import load_dotenv -from loguru import logger - -from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.pipeline.pipeline import Pipeline -from pipecat.pipeline.runner import PipelineRunner -from pipecat.pipeline.task import PipelineParams, PipelineTask -from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext -from pipecat.processors.frameworks.rtvi import RTVIConfig, RTVIObserver, RTVIProcessor -from pipecat.services.cartesia.tts import CartesiaTTSService -from pipecat.services.deepgram.stt import DeepgramSTTService -from pipecat.services.openai.llm import OpenAILLMService -from pipecat.transports.base_transport import BaseTransport, TransportParams - -load_dotenv(override=True) - - -async def run_bot(transport: BaseTransport, _: argparse.Namespace, handle_sigint: bool): - logger.info(f"Starting bot") - - stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) - - tts = CartesiaTTSService( - api_key=os.getenv("CARTESIA_API_KEY"), - voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady - ) - - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) - - messages = [ - { - "role": "system", - "content": "You are a friendly AI assistant. Respond naturally and keep your answers conversational.", - }, - ] - - context = OpenAILLMContext(messages) - context_aggregator = llm.create_context_aggregator(context) - - rtvi = RTVIProcessor(config=RTVIConfig(config=[])) - - pipeline = Pipeline( - [ - transport.input(), # Transport user input - rtvi, # RTVI processor - stt, - context_aggregator.user(), # User responses - llm, # LLM - tts, # TTS - transport.output(), # Transport bot output - context_aggregator.assistant(), # Assistant spoken responses - ] - ) - - task = PipelineTask( - pipeline, - params=PipelineParams( - enable_metrics=True, - enable_usage_metrics=True, - ), - observers=[RTVIObserver(rtvi)], - ) - - @transport.event_handler("on_client_connected") - async def on_client_connected(transport, client): - logger.info(f"Client connected") - # Kick off the conversation. - messages.append({"role": "system", "content": "Say hello and briefly introduce yourself."}) - await task.queue_frames([context_aggregator.user().get_context_frame()]) - - @transport.event_handler("on_client_disconnected") - async def on_client_disconnected(transport, client): - logger.info(f"Client disconnected") - await task.cancel() - - runner = PipelineRunner(handle_sigint=handle_sigint) - - await runner.run(task) - - -if __name__ == "__main__": - from pipecat.runner.local import main - - transport_params = { - "webrtc": lambda: TransportParams( - audio_in_enabled=True, - audio_out_enabled=True, - vad_analyzer=SileroVADAnalyzer(), - ), - } - - main(run_bot, transport_params=transport_params) diff --git a/src/pipecat/runner/local.py b/src/pipecat/runner/local.py deleted file mode 100644 index fea8d4a24..000000000 --- a/src/pipecat/runner/local.py +++ /dev/null @@ -1,521 +0,0 @@ -# -# Copyright (c) 2024–2025, Daily -# -# SPDX-License-Identifier: BSD 2-Clause License -# - -"""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. - -Install dependencies with:: - - pip install pipecat-ai[runner] - -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 json -import os -import sys -from contextlib import asynccontextmanager -from typing import Callable, Dict, Mapping, Optional - -from loguru import logger - -try: - 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 -except ImportError as e: - logger.error(f"Runner dependencies not available: {e}") - logger.error("To use Pipecat runners, install with: pip install pipecat-ai[runner]") - raise ImportError( - "Runner dependencies required. Install with: pip install pipecat-ai[runner]" - ) from e - -load_dotenv(override=True) - - -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) - - -def _run_webrtc( - run: Callable, args: argparse.Namespace, transport_params: Mapping[str, Callable] = {} -): - """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.") - logger.debug(f"Import error: {e}") - return - - logger.info("Running with SmallWebRTCTransport...") - - app = FastAPI() - - # Store connections by pc_id (like the working version) - 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 function with SmallWebRTC transport arguments (like working version) - params = transport_params[args.transport]() - transport = SmallWebRTCTransport(params=params, webrtc_connection=pipecat_connection) - - 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 pipecat.runner.utils import smallwebrtc_sdp_munging - - answer["sdp"] = smallwebrtc_sdp_munging(answer["sdp"], args.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 - uvicorn.run(app, host=args.host, port=args.port) - - -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 FastAPIWebsocketTransport - except ImportError as e: - logger.error(f"Twilio transport dependencies not installed.") - logger.debug(f"Import error: {e}") - return - - logger.info("Running with FastAPIWebsocketTransport (Twilio)...") - - app = FastAPI() - - # 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"] - - 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", ""), - ) - - transport = FastAPIWebsocketTransport(websocket=kwargs["websocket"], params=params) - - class MockArgs: - def __init__(self): - self.transport = "twilio" - - 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 using Telnyx transport with FastAPI WebSocket server.""" - try: - from pipecat.serializers.telnyx import TelnyxFrameSerializer - 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}") - return - - logger.info("Running with FastAPIWebsocketTransport (Telnyx)...") - - app = FastAPI() - - # 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"] - - 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", - ) - - transport = FastAPIWebsocketTransport(websocket=kwargs["websocket"], params=params) - - class MockArgs: - def __init__(self): - self.transport = "telnyx" - - 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 using Plivo transport with FastAPI WebSocket server.""" - try: - from pipecat.serializers.plivo import PlivoFrameSerializer - 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}") - return - - logger.info("Running with FastAPIWebsocketTransport (Plivo)...") - - app = FastAPI() - - # 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"] - - params = transport_params["plivo"]() - params.add_wav_header = False - params.serializer = PlivoFrameSerializer( - stream_id=call_info["stream_id"], - call_id=call_info["call_id"], - ) - - transport = FastAPIWebsocketTransport(websocket=kwargs["websocket"], params=params) - - class MockArgs: - def __init__(self): - self.transport = "plivo" - - 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_daily( - run: Callable, args: argparse.Namespace, transport_params: Mapping[str, Callable] = {} -): - """Run using Daily transport.""" - try: - from pipecat.runner.daily 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 - - 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 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.") - return - - match args.transport: - case "daily": - _run_daily(run, args, transport_params) - case "livekit": - _run_livekit(run, args, transport_params) - case "plivo": - _run_plivo(run, args, transport_params) - case "telnyx": - _run_telnyx(run, args, transport_params) - case "twilio": - _run_twilio(run, args, transport_params) - case "webrtc": - _run_webrtc(run, args, transport_params) - - -def main( - run: Callable, - *, - parser: Optional[argparse.ArgumentParser] = None, - transport_params: Mapping[str, Callable] = {}, -): - """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: - - Args: - --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") - - parser.add_argument( - "--host", default="localhost", help="Host for HTTP server (default: localhost)" - ) - parser.add_argument( - "--port", type=int, default=7860, help="Port for HTTP server (default: 7860)" - ) - parser.add_argument( - "--transport", - "-t", - type=str, - choices=list(transport_params.keys()) - if transport_params - else ["daily", "livekit", "plivo", "telnyx", "twilio", "webrtc"], - default="webrtc", - help="The transport this application should use", - ) - parser.add_argument( - "--proxy", "-x", help="A public proxy host name (no protocol, e.g. proxy.example.com)" - ) - parser.add_argument( - "--esp32", action="store_true", default=False, help="Perform SDP munging for the ESP32" - ) - parser.add_argument( - "--verbose", "-v", action="count", default=0, help="Increase logging verbosity" - ) - - args = parser.parse_args() - - if args.esp32 and args.host == "localhost": - logger.error("For ESP32, you need to specify `--host IP` so we can do SDP munging.") - return - - # Log level - logger.remove(0) - logger.add(sys.stderr, level="TRACE" if args.verbose else "DEBUG") - - # Startup messages for browser-accessible transports - if args.transport == "webrtc": - print() - print(f"🚀 WebRTC server starting at http://{args.host}:{args.port}/client") - print(f" Open this URL in your browser to connect!") - print() - - _run_main(run, args, transport_params) diff --git a/src/pipecat/runner/cloud.py b/src/pipecat/runner/run.py similarity index 99% rename from src/pipecat/runner/cloud.py rename to src/pipecat/runner/run.py index bc86e3238..a87a64e42 100644 --- a/src/pipecat/runner/cloud.py +++ b/src/pipecat/runner/run.py @@ -33,7 +33,7 @@ Single transport example:: await run_pipeline(transport) if __name__ == "__main__": - from pipecat.runner.cloud import main + from pipecat.runner.run import main main() Multiple transport example:: diff --git a/src/pipecat/runner/utils.py b/src/pipecat/runner/utils.py index d67c7ffb8..7a4a59775 100644 --- a/src/pipecat/runner/utils.py +++ b/src/pipecat/runner/utils.py @@ -149,26 +149,6 @@ async def parse_telephony_websocket(websocket: WebSocket): raise -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.