From 0c3e526c1932013d19fc6b1daeb83437f3d6cfbd Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Wed, 30 Jul 2025 18:53:21 -0400 Subject: [PATCH] Remove dependency on pipecatcloud, define new *RunnerArguments types --- .../runner-examples/01-all-transport-bot.py | 22 +++--- .../01-all-transport-factory-bot.py | 16 ++--- .../runner-examples/02-two-transport-bot.py | 19 ++---- .../03-single-transport-bot.py | 4 +- pyproject.toml | 3 +- src/pipecat/runner/run.py | 67 +++++-------------- src/pipecat/runner/types.py | 60 +++++++++++++++++ src/pipecat/runner/utils.py | 26 +++---- 8 files changed, 111 insertions(+), 106 deletions(-) create mode 100644 src/pipecat/runner/types.py diff --git a/examples/runner-examples/01-all-transport-bot.py b/examples/runner-examples/01-all-transport-bot.py index 00f2aad38..6d884dc76 100644 --- a/examples/runner-examples/01-all-transport-bot.py +++ b/examples/runner-examples/01-all-transport-bot.py @@ -26,19 +26,15 @@ 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.run import SmallWebRTCSessionArguments +from pipecat.runner.types import ( + DailyRunnerArguments, + SmallWebRTCRunnerArguments, + WebSocketRunnerArguments, +) 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) # Check if we're running locally @@ -108,11 +104,11 @@ async def run_bot(transport): async def bot( - session_args: DailySessionArguments | SmallWebRTCSessionArguments | WebSocketSessionArguments, + session_args: DailyRunnerArguments | SmallWebRTCRunnerArguments | WebSocketRunnerArguments, ): """Main bot entry point compatible with Pipecat Cloud.""" - if isinstance(session_args, DailySessionArguments): + if isinstance(session_args, DailyRunnerArguments): from pipecat.transports.services.daily import DailyParams, DailyTransport if not IS_LOCAL_RUN: @@ -132,7 +128,7 @@ async def bot( ), ) - elif isinstance(session_args, SmallWebRTCSessionArguments): + elif isinstance(session_args, SmallWebRTCRunnerArguments): from pipecat.transports.base_transport import TransportParams from pipecat.transports.network.small_webrtc import SmallWebRTCTransport @@ -145,7 +141,7 @@ async def bot( webrtc_connection=session_args.webrtc_connection, ) - elif isinstance(session_args, WebSocketSessionArguments): + elif isinstance(session_args, WebSocketRunnerArguments): # Use the utility to parse WebSocket data from pipecat.runner.utils import parse_telephony_websocket diff --git a/examples/runner-examples/01-all-transport-factory-bot.py b/examples/runner-examples/01-all-transport-factory-bot.py index 4f1f20d36..859d92683 100644 --- a/examples/runner-examples/01-all-transport-factory-bot.py +++ b/examples/runner-examples/01-all-transport-factory-bot.py @@ -26,6 +26,12 @@ 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.types import ( + DailyRunnerArguments, + SmallWebRTCRunnerArguments, + WebSocketRunnerArguments, +) +from pipecat.runner.utils import create_transport from pipecat.services.cartesia.tts import CartesiaTTSService from pipecat.services.deepgram.stt import DeepgramSTTService from pipecat.services.openai.llm import OpenAILLMService @@ -33,14 +39,6 @@ from pipecat.transports.base_transport import TransportParams from pipecat.transports.network.fastapi_websocket import FastAPIWebsocketParams from pipecat.transports.services.daily import DailyParams -try: - from pipecatcloud.agent import DailySessionArguments, WebSocketSessionArguments - - from pipecat.runner.run import SmallWebRTCSessionArguments - from pipecat.runner.utils import create_transport -except ImportError: - raise ImportError("Install with: pip install pipecat-ai[runner]") - load_dotenv(override=True) # Define transport configurations using factory functions @@ -139,7 +137,7 @@ async def run_bot(transport): async def bot( - session_args: DailySessionArguments | SmallWebRTCSessionArguments | WebSocketSessionArguments, + session_args: DailyRunnerArguments | SmallWebRTCRunnerArguments | WebSocketRunnerArguments, ): """Main bot entry point compatible with Pipecat Cloud.""" transport = await create_transport(session_args, transport_params) diff --git a/examples/runner-examples/02-two-transport-bot.py b/examples/runner-examples/02-two-transport-bot.py index d7757ad0c..7d3e4d4ef 100644 --- a/examples/runner-examples/02-two-transport-bot.py +++ b/examples/runner-examples/02-two-transport-bot.py @@ -19,19 +19,14 @@ 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.run import SmallWebRTCSessionArguments +from pipecat.runner.types import ( + DailyRunnerArguments, + SmallWebRTCRunnerArguments, +) 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) @@ -101,10 +96,10 @@ async def run_bot(transport): await runner.run(task) -async def bot(session_args: DailySessionArguments | SmallWebRTCSessionArguments): +async def bot(session_args: DailyRunnerArguments | SmallWebRTCRunnerArguments): """Main bot entry point compatible with Pipecat Cloud.""" - if isinstance(session_args, DailySessionArguments): + if isinstance(session_args, DailyRunnerArguments): from pipecat.transports.services.daily import DailyParams, DailyTransport if not IS_LOCAL_RUN: @@ -124,7 +119,7 @@ async def bot(session_args: DailySessionArguments | SmallWebRTCSessionArguments) ), ) - elif isinstance(session_args, SmallWebRTCSessionArguments): + elif isinstance(session_args, SmallWebRTCRunnerArguments): from pipecat.transports.base_transport import TransportParams from pipecat.transports.network.small_webrtc import SmallWebRTCTransport diff --git a/examples/runner-examples/03-single-transport-bot.py b/examples/runner-examples/03-single-transport-bot.py index cae886f32..84dfadba8 100644 --- a/examples/runner-examples/03-single-transport-bot.py +++ b/examples/runner-examples/03-single-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.run import SmallWebRTCSessionArguments +from pipecat.runner.types import SmallWebRTCRunnerArguments from pipecat.services.cartesia.tts import CartesiaTTSService from pipecat.services.deepgram.stt import DeepgramSTTService from pipecat.services.openai.llm import OpenAILLMService @@ -91,7 +91,7 @@ async def run_bot(transport): await runner.run(task) -async def bot(session_args: SmallWebRTCSessionArguments): +async def bot(session_args: SmallWebRTCRunnerArguments): """Main bot entry point compatible with Pipecat Cloud.""" transport = SmallWebRTCTransport( diff --git a/pyproject.toml b/pyproject.toml index 8478fafdf..5f0bc5ef0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -81,12 +81,11 @@ 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" ] riva = [ "nvidia-riva-client~=2.21.1" ] -runner = [ "python-dotenv>=1.0.0,<2.0.0", "uvicorn>=0.32.0,<1.0.0", "fastapi>=0.115.6,<0.117.0", "pipecatcloud>=0.2.0", "pipecat-ai-small-webrtc-prebuilt>=1.0.0"] +runner = [ "python-dotenv>=1.0.0,<2.0.0", "uvicorn>=0.32.0,<1.0.0", "fastapi>=0.115.6,<0.117.0", "pipecat-ai-small-webrtc-prebuilt>=1.0.0"] sambanova = [] sentry = [ "sentry-sdk~=2.23.1" ] local-smart-turn = [ "coremltools>=8.0", "transformers", "torch~=2.5.0", "torchaudio~=2.5.0" ] diff --git a/src/pipecat/runner/run.py b/src/pipecat/runner/run.py index fd3224c4b..64e7808bc 100644 --- a/src/pipecat/runner/run.py +++ b/src/pipecat/runner/run.py @@ -21,7 +21,7 @@ are established. Single transport example:: - async def bot(session_args: DailySessionArguments): + async def bot(session_args: DailyRunnerArguments): transport = DailyTransport( session_args.room_url, session_args.token, @@ -39,11 +39,11 @@ Multiple transport example:: async def bot(session_args): # Type-safe transport detection - if isinstance(session_args, DailySessionArguments): + if isinstance(session_args, DailyRunnerArguments): transport = setup_daily_transport(session_args) # Your application code - elif isinstance(session_args, SmallWebRTCSessionArguments): + elif isinstance(session_args, SmallWebRTCRunnerArguments): transport = setup_webrtc_transport(session_args) # Your application code - elif isinstance(session_args, WebSocketSessionArguments): + elif isinstance(session_args, WebSocketRunnerArguments): transport = setup_telephony_transport(session_args) # Your application code # Your bot implementation @@ -69,11 +69,16 @@ import asyncio import os import sys from contextlib import asynccontextmanager -from dataclasses import dataclass -from typing import Any, Dict, Optional +from typing import Dict from loguru import logger +from pipecat.runner.types import ( + DailyRunnerArguments, + SmallWebRTCRunnerArguments, + WebSocketRunnerArguments, +) + try: import uvicorn from dotenv import load_dotenv @@ -87,32 +92,6 @@ except ImportError as e: "Runner dependencies required. Install with: pip install pipecat-ai[runner]" ) from e -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[runner]" - ) - - -# Define WebRTC type locally until it's added to pipecatcloud -@dataclass -class SmallWebRTCSessionArguments: - """Small WebRTC session arguments. - - .. deprecated:: 0.0.77 - This will be replaced by pipecatcloud.agent.SmallWebRTCSessionArguments - once WebRTC support is added to the `pipecatcloud` package. - - Parameters: - websocket: The WebSocket connection for the session. - session_id: Optional session ID for tracking. - """ - - webrtc_connection: Any - session_id: Optional[str] = None - load_dotenv(override=True) os.environ["LOCAL_RUN"] = "1" @@ -165,7 +144,7 @@ async def _run_telephony_bot(websocket: WebSocket): bot_module = _get_bot_module() # Just pass the WebSocket - let the bot handle parsing - session_args = WebSocketSessionArguments(websocket=websocket, session_id=None) + session_args = WebSocketRunnerArguments(websocket=websocket) await bot_module.bot(session_args) @@ -204,7 +183,7 @@ def _setup_webrtc_routes(app: FastAPI, esp32_mode: bool = False, host: str = "lo from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection except ImportError as e: - logger.error(f"WebRTC transport dependencies not installed.") + logger.error(f"WebRTC transport dependencies not installed: {e}") return # Store connections by pc_id @@ -242,10 +221,7 @@ def _setup_webrtc_routes(app: FastAPI, esp32_mode: bool = False, host: str = "lo pcs_map.pop(webrtc_connection.pc_id, None) bot_module = _get_bot_module() - session_args = SmallWebRTCSessionArguments( - webrtc_connection=pipecat_connection, - session_id=None, - ) + session_args = SmallWebRTCRunnerArguments(webrtc_connection=pipecat_connection) background_tasks.add_task(bot_module.bot, session_args) answer = pipecat_connection.get_answer() @@ -287,9 +263,7 @@ def _setup_daily_routes(app: FastAPI): # Start the bot in the background bot_module = _get_bot_module() - session_args = DailySessionArguments( - room_url=room_url, token=token, body={}, session_id=None - ) + session_args = DailyRunnerArguments(room_url=room_url, token=token, body={}) asyncio.create_task(bot_module.bot(session_args)) return RedirectResponse(room_url) @@ -307,9 +281,7 @@ def _setup_daily_routes(app: FastAPI): # Start the bot in the background bot_module = _get_bot_module() - session_args = DailySessionArguments( - room_url=room_url, token=token, body={}, session_id=None - ) + session_args = DailyRunnerArguments(room_url=room_url, token=token, body={}) asyncio.create_task(bot_module.bot(session_args)) return {"room_url": room_url, "token": token} @@ -373,12 +345,7 @@ async def _run_daily_direct(): async with aiohttp.ClientSession() as session: room_url, token = await configure(session) - # Create session args like the cloud runner would - from pipecatcloud.agent import DailySessionArguments - - session_args = DailySessionArguments( - room_url=room_url, token=token, body={}, session_id=None - ) + session_args = DailyRunnerArguments(room_url=room_url, token=token, body={}) # Get the bot module and run it directly bot_module = _get_bot_module() diff --git a/src/pipecat/runner/types.py b/src/pipecat/runner/types.py new file mode 100644 index 000000000..d093b09dd --- /dev/null +++ b/src/pipecat/runner/types.py @@ -0,0 +1,60 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Runner session argument types for the development runner. + +These types are used by the development runner to pass transport-specific +information to bot functions. +""" + +from dataclasses import dataclass +from typing import Any + +from fastapi import WebSocket + + +@dataclass +class RunnerArguments: + """Base class for runner session arguments.""" + + pass + + +@dataclass +class DailyRunnerArguments(RunnerArguments): + """Daily transport session arguments for the runner. + + Parameters: + room_url: Daily room URL to join + token: Authentication token for the room + body: Additional request data + """ + + room_url: str + token: str + body: Any + + +@dataclass +class WebSocketRunnerArguments(RunnerArguments): + """WebSocket transport session arguments for the runner. + + Parameters: + websocket: WebSocket connection for audio streaming + """ + + websocket: WebSocket + + +@dataclass +class SmallWebRTCRunnerArguments(RunnerArguments): + """Small WebRTC transport session arguments for the runner. + + Parameters: + webrtc_connection: Pre-configured WebRTC peer connection + """ + + webrtc_connection: Any diff --git a/src/pipecat/runner/utils.py b/src/pipecat/runner/utils.py index 8797c9ae7..cd0781f78 100644 --- a/src/pipecat/runner/utils.py +++ b/src/pipecat/runner/utils.py @@ -37,6 +37,11 @@ from typing import Any, Callable, Dict, Optional from fastapi import WebSocket from loguru import logger +from pipecat.runner.types import ( + DailyRunnerArguments, + SmallWebRTCRunnerArguments, + WebSocketRunnerArguments, +) from pipecat.transports.base_transport import BaseTransport @@ -439,23 +444,8 @@ async def create_transport( transport = await create_transport(session_args, transport_params) """ - # Import session types - try: - from pipecatcloud.agent import DailySessionArguments, WebSocketSessionArguments - except ImportError: - raise ImportError( - "pipecatcloud package required. Install with: pip install pipecat-ai[runner]" - ) - - try: - from pipecat.runner.run import SmallWebRTCSessionArguments - except ImportError: - raise ImportError( - "SmallWebRTCSessionArguments not found. Make sure you've installed pipecat-ai[runner]." - ) - # Create transport based on session args type - if isinstance(session_args, DailySessionArguments): + if isinstance(session_args, DailyRunnerArguments): params = _get_transport_params("daily", transport_params) from pipecat.transports.services.daily import DailyTransport @@ -467,7 +457,7 @@ async def create_transport( params=params, ) - elif isinstance(session_args, SmallWebRTCSessionArguments): + elif isinstance(session_args, SmallWebRTCRunnerArguments): params = _get_transport_params("webrtc", transport_params) from pipecat.transports.network.small_webrtc import SmallWebRTCTransport @@ -477,7 +467,7 @@ async def create_transport( webrtc_connection=session_args.webrtc_connection, ) - elif isinstance(session_args, WebSocketSessionArguments): + elif isinstance(session_args, WebSocketRunnerArguments): # Parse once to determine the provider and get data transport_type, call_data = await parse_telephony_websocket(session_args.websocket) params = _get_transport_params(transport_type, transport_params)