wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Remove dependency on pipecatcloud, define new *RunnerArguments types

Browse Source
This commit is contained in:
Mark Backman
2025-07-30 18:53:21 -04:00
parent 58fc952192
commit 0c3e526c19
8 changed files with 111 additions and 106 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
examples/runner-examples/01-all-transport-bot.py
View File

@@ -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

16
examples/runner-examples/01-all-transport-factory-bot.py
View File

@@ -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)

19
examples/runner-examples/02-two-transport-bot.py
View File

@@ -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

4
examples/runner-examples/03-single-transport-bot.py
View File

@@ -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(

3
pyproject.toml
View File

@@ -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" ]

67
src/pipecat/runner/run.py
View File

@@ -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()

60
src/pipecat/runner/types.py Normal file
View File

@@ -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

26
src/pipecat/runner/utils.py
View File

@@ -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)