From 49435cd1154aa6c26d881b220336d1771a37ffb5 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Sat, 26 Jul 2025 21:56:06 -0400 Subject: [PATCH] Add typing support for session_args, fix debug logging in cloud.py --- examples/quickstart/cloud-multi-bot.py | 60 ++++++----- examples/quickstart/cloud-simple-bot.py | 31 ++++-- examples/quickstart/local-multi-bot.py | 5 + pyproject.toml | 1 + src/pipecat/runner/cloud.py | 135 ++++++++++++++---------- src/pipecat/runner/local.py | 4 +- 6 files changed, 148 insertions(+), 88 deletions(-) diff --git a/examples/quickstart/cloud-multi-bot.py b/examples/quickstart/cloud-multi-bot.py index 9d72eb32d..314eca81f 100644 --- a/examples/quickstart/cloud-multi-bot.py +++ b/examples/quickstart/cloud-multi-bot.py @@ -12,16 +12,25 @@ 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 PipelineTask +from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.runner.cloud import SmallWebRTCSessionArguments from pipecat.services.cartesia.tts import CartesiaTTSService from pipecat.services.deepgram.stt import DeepgramSTTService from pipecat.services.openai.llm import OpenAILLMService +try: + from pipecatcloud.agent import DailySessionArguments, WebSocketSessionArguments +except ImportError: + raise ImportError( + "pipecatcloud package is required for cloud-compatible bots. " + "Install with: pip install pipecat-ai[[pipecatcloud]]" + ) + load_dotenv(override=True) -async def run_bot_logic(transport, handle_sigint: bool = True): +async def run_bot(transport): """Main bot logic that works with any transport.""" logger.info(f"Starting bot") @@ -56,7 +65,13 @@ async def run_bot_logic(transport, handle_sigint: bool = True): ] ) - task = PipelineTask(pipeline) + task = PipelineTask( + pipeline, + params=PipelineParams( + enable_metrics=True, + enable_usage_metrics=True, + ), + ) @transport.event_handler("on_client_connected") async def on_client_connected(transport, client): @@ -69,18 +84,16 @@ async def run_bot_logic(transport, handle_sigint: bool = True): logger.info("Client disconnected") await task.cancel() - runner = PipelineRunner(handle_sigint=handle_sigint) + runner = PipelineRunner(handle_sigint=False) await runner.run(task) -async def bot(session_args): +async def bot( + session_args: DailySessionArguments | SmallWebRTCSessionArguments | WebSocketSessionArguments, +): """Main bot entry point compatible with Pipecat Cloud.""" - # Get handle_sigint from session_args, default to True for Daily - handle_sigint = getattr(session_args, "handle_sigint", True) - - if hasattr(session_args, "room_url") and hasattr(session_args, "token"): - # Daily session arguments (cloud or local) + if isinstance(session_args, DailySessionArguments): from pipecat.transports.services.daily import DailyParams, DailyTransport transport = DailyTransport( @@ -94,8 +107,7 @@ async def bot(session_args): ), ) - elif hasattr(session_args, "webrtc_connection"): - # WebRTC session arguments (local only, created by server.py) + elif isinstance(session_args, SmallWebRTCSessionArguments): from pipecat.transports.base_transport import TransportParams from pipecat.transports.network.small_webrtc import SmallWebRTCTransport @@ -108,14 +120,13 @@ async def bot(session_args): webrtc_connection=session_args.webrtc_connection, ) - elif hasattr(session_args, "websocket"): - # WebSocket session arguments (for telephony providers) + elif isinstance(session_args, WebSocketSessionArguments): from pipecat.transports.network.fastapi_websocket import ( FastAPIWebsocketParams, FastAPIWebsocketTransport, ) - # Create appropriate serializer based on transport type + # Create base parameters for telephony params = FastAPIWebsocketParams( audio_in_enabled=True, audio_out_enabled=True, @@ -123,43 +134,44 @@ async def bot(session_args): add_wav_header=False, ) - if session_args.transport_type == "twilio": + # Create appropriate serializer based on transport type + transport_type = getattr(session_args, "transport_type", "unknown") + call_info = getattr(session_args, "call_info", {}) + + if transport_type == "twilio": from pipecat.serializers.twilio import TwilioFrameSerializer - call_info = session_args.call_info 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", ""), ) - elif session_args.transport_type == "telnyx": + elif transport_type == "telnyx": from pipecat.serializers.telnyx import TelnyxFrameSerializer - call_info = session_args.call_info 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", ) - elif session_args.transport_type == "plivo": + elif transport_type == "plivo": from pipecat.serializers.plivo import PlivoFrameSerializer - call_info = session_args.call_info params.serializer = PlivoFrameSerializer( stream_id=call_info["stream_id"], call_id=call_info["call_id"], ) else: - raise ValueError(f"Unsupported WebSocket transport type: {session_args.transport_type}") + raise ValueError(f"Unsupported WebSocket transport type: {transport_type}") transport = FastAPIWebsocketTransport(websocket=session_args.websocket, params=params) else: - raise ValueError(f"Unknown session arguments: {session_args}") + raise ValueError(f"Unsupported session arguments type: {type(session_args)}") - await run_bot_logic(transport, handle_sigint) + await run_bot(transport) if __name__ == "__main__": diff --git a/examples/quickstart/cloud-simple-bot.py b/examples/quickstart/cloud-simple-bot.py index 3d408409e..b14eef563 100644 --- a/examples/quickstart/cloud-simple-bot.py +++ b/examples/quickstart/cloud-simple-bot.py @@ -12,16 +12,25 @@ 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 PipelineTask +from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.runner.cloud import SmallWebRTCSessionArguments from pipecat.services.cartesia.tts import CartesiaTTSService from pipecat.services.deepgram.stt import DeepgramSTTService from pipecat.services.openai.llm import OpenAILLMService +try: + from pipecatcloud.agent import DailySessionArguments +except ImportError: + raise ImportError( + "pipecatcloud package is required for cloud-compatible bots. " + "Install with: pip install pipecat-ai[[pipecatcloud]]" + ) + load_dotenv(override=True) -async def run_bot_logic(transport): +async def run_bot(transport): """Main bot logic that works with any transport.""" logger.info(f"Starting bot") @@ -56,7 +65,13 @@ async def run_bot_logic(transport): ] ) - task = PipelineTask(pipeline) + task = PipelineTask( + pipeline, + params=PipelineParams( + enable_metrics=True, + enable_usage_metrics=True, + ), + ) @transport.event_handler("on_client_connected") async def on_client_connected(transport, client): @@ -73,11 +88,10 @@ async def run_bot_logic(transport): await runner.run(task) -async def bot(session_args): +async def bot(session_args: DailySessionArguments | SmallWebRTCSessionArguments): """Main bot entry point compatible with Pipecat Cloud.""" - if hasattr(session_args, "room_url"): - # Daily session arguments (cloud or local) + if isinstance(session_args, DailySessionArguments): from pipecat.transports.services.daily import DailyParams, DailyTransport transport = DailyTransport( @@ -91,8 +105,7 @@ async def bot(session_args): ), ) - elif hasattr(session_args, "webrtc_connection"): - # WebRTC session arguments (local only, created by server.py) + elif isinstance(session_args, SmallWebRTCSessionArguments): from pipecat.transports.base_transport import TransportParams from pipecat.transports.network.small_webrtc import SmallWebRTCTransport @@ -105,7 +118,7 @@ async def bot(session_args): webrtc_connection=session_args.webrtc_connection, ) - await run_bot_logic(transport) + await run_bot(transport) if __name__ == "__main__": diff --git a/examples/quickstart/local-multi-bot.py b/examples/quickstart/local-multi-bot.py index 081a31406..b542c6c15 100644 --- a/examples/quickstart/local-multi-bot.py +++ b/examples/quickstart/local-multi-bot.py @@ -15,6 +15,7 @@ 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 @@ -69,9 +70,12 @@ async def run_bot(transport: BaseTransport, _: argparse.Namespace, handle_sigint 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 @@ -87,6 +91,7 @@ async def run_bot(transport: BaseTransport, _: argparse.Namespace, handle_sigint enable_metrics=True, enable_usage_metrics=True, ), + observers=[RTVIObserver(rtvi)], ) @transport.event_handler("on_client_connected") diff --git a/pyproject.toml b/pyproject.toml index 0c5be3a27..7cf38f7c7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -79,6 +79,7 @@ openai = [ "websockets>=13.1,<15.0" ] openpipe = [ "openpipe~=4.50.0" ] openrouter = [] perplexity = [] +pipecatcloud = [ "pipecatcloud>=0.2.0" ] playht = [ "pyht>=0.1.6", "websockets>=13.1,<15.0" ] qwen = [] rime = [ "websockets>=13.1,<15.0" ] diff --git a/src/pipecat/runner/cloud.py b/src/pipecat/runner/cloud.py index f9d440775..265be77bb 100644 --- a/src/pipecat/runner/cloud.py +++ b/src/pipecat/runner/cloud.py @@ -10,29 +10,42 @@ 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. +It requires the pipecatcloud package for proper session argument types. + +Install with:: + + pip install pipecat-ai[pipecatcloud] + All bots must implement a `bot(session_args)` async function as the entry point. The server automatically discovers and executes this function when connections are established. Bot function signature:: - async def bot(session_args): - # session_args contains transport-specific connection information + async def bot(session_args: DailySessionArguments | SmallWebRTCSessionArguments | WebSocketSessionArguments): - # For Daily: session_args.room_url, session_args.token, session_args.body - # For WebRTC: session_args.webrtc_connection - # For Telephony: session_args.websocket - - # Create transport based on session_args attributes - if hasattr(session_args, 'room_url'): - # Daily transport setup + # Create transport based on session_args type + if isinstance(session_args, DailySessionArguments): + # Daily transport setup - guaranteed to have room_url, token from pipecat.transports.services.daily import DailyTransport, DailyParams - transport = DailyTransport(session_args.room_url, session_args.token, "Bot", DailyParams(...)) - elif hasattr(session_args, 'webrtc_connection'): - # WebRTC transport setup + transport = DailyTransport( + session_args.room_url, + session_args.token, + "Bot", + DailyParams(...) + ) + elif isinstance(session_args, SmallWebRTCSessionArguments): + # WebRTC transport setup - guaranteed to have webrtc_connection from pipecat.transports.network.small_webrtc import SmallWebRTCTransport from pipecat.transports.base_transport import TransportParams - transport = SmallWebRTCTransport(TransportParams(...), webrtc_connection=session_args.webrtc_connection) + transport = SmallWebRTCTransport( + TransportParams(...), + webrtc_connection=session_args.webrtc_connection + ) + elif isinstance(session_args, WebSocketSessionArguments): + # Telephony setup - guaranteed to have websocket + # Access call_info and transport_type via additional attributes + pass # Run your bot logic await run_bot_logic(transport) @@ -45,12 +58,12 @@ Supported transports: Example:: - async def bot(session_args): + async def bot(session_args: DailySessionArguments | SmallWebRTCSessionArguments): # Detect transport type from session_args - if hasattr(session_args, "room_url"): + if isinstance(session_args, DailySessionArguments): # Daily transport = create_daily_transport(session_args) - elif hasattr(session_args, "webrtc_connection"): + elif isinstance(session_args, SmallWebRTCSessionArguments): # WebRTC transport = create_webrtc_transport(session_args) @@ -69,17 +82,41 @@ import asyncio import os import sys from contextlib import asynccontextmanager -from typing import Dict +from dataclasses import dataclass +from typing import Any, Dict, Optional import uvicorn from dotenv import load_dotenv -from fastapi import BackgroundTasks, FastAPI +from fastapi import BackgroundTasks, FastAPI, WebSocket from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import RedirectResponse from loguru import logger from pipecat.runner.utils import setup_websocket_routes +# Require pipecatcloud for cloud-compatible bots +try: + from pipecatcloud.agent import DailySessionArguments, WebSocketSessionArguments +except ImportError: + raise ImportError( + "pipecatcloud package is required for cloud-compatible bots. " + "Install with: pip install pipecat-ai[pipecatcloud]" + ) + + +# Define WebRTC type locally until it's added to pipecatcloud +@dataclass +class SmallWebRTCSessionArguments: + """Small WebRTC session arguments for local development. + + This will be replaced by pipecatcloud.agent.SmallWebRTCSessionArguments + when WebRTC support is added to Pipecat Cloud. + """ + + webrtc_connection: Any + session_id: Optional[str] = None + + load_dotenv(override=True) os.environ["LOCAL_RUN"] = "1" @@ -125,19 +162,18 @@ def _get_bot_module(): ) -async def _run_telephony_bot(transport_type: str, websocket, call_info): +async def _run_telephony_bot(transport_type: str, websocket: WebSocket, call_info): """Run a bot for telephony transports.""" 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 = WebSocketSessionArguments( + websocket=websocket, + session_id=None, + ) + # Add transport-specific attributes + session_args.call_info = call_info + session_args.transport_type = transport_type - session_args = WebSocketSessionArgs(transport_type, websocket, call_info) await bot_module.bot(session_args) @@ -198,17 +234,12 @@ def _create_server_app(transport_type: str, host: str = "localhost", proxy: str logger.info(f"Discarding peer connection for pc_id: {webrtc_connection.pc_id}") pcs_map.pop(webrtc_connection.pc_id, None) - # 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(pipecat_connection) + session_args = SmallWebRTCSessionArguments( + webrtc_connection=pipecat_connection, + session_id=None, + ) background_tasks.add_task(bot_module.bot, session_args) answer = pipecat_connection.get_answer() @@ -252,15 +283,9 @@ def _create_server_app(transport_type: str, host: str = "localhost", proxy: str # Start the bot in the background to join the room bot_module = _get_bot_module() - - class DailySessionArgs: - def __init__(self, room_url, token): - self.room_url = room_url - self.token = token - self.body = {} - self.handle_sigint = False - - session_args = DailySessionArgs(room_url, token) + 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) @@ -284,15 +309,9 @@ def _create_server_app(transport_type: str, host: str = "localhost", proxy: str # Start the bot in the background bot_module = _get_bot_module() - - class DailySessionArgs: - def __init__(self, room_url, token): - self.room_url = room_url - self.token = token - self.body = {} - self.handle_sigint = False - - session_args = DailySessionArgs(room_url, token) + 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} @@ -316,6 +335,7 @@ def main(): --port: Server port (default: 7860) -t/--transport: Transport type (daily, webrtc, twilio, telnyx, plivo) -x/--proxy: Public proxy hostname for telephony webhooks + -v/--verbose: Increase logging verbosity The bot file must contain a `bot(session_args)` function as the entry point. """ @@ -331,9 +351,16 @@ def main(): help="Transport type", ) parser.add_argument("--proxy", "-x", help="Public proxy host name") + parser.add_argument( + "--verbose", "-v", action="count", default=0, help="Increase logging verbosity" + ) args = parser.parse_args() + # Log level + logger.remove() + logger.add(sys.stderr, level="TRACE" if args.verbose else "DEBUG") + # Print startup message if args.transport == "webrtc": print() diff --git a/src/pipecat/runner/local.py b/src/pipecat/runner/local.py index 6e167fb2f..a143ee2a6 100644 --- a/src/pipecat/runner/local.py +++ b/src/pipecat/runner/local.py @@ -388,7 +388,9 @@ def main( 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) + parser.add_argument( + "--verbose", "-v", action="count", default=0, help="Increase logging verbosity" + ) args = parser.parse_args()