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

Final cleanup

Browse Source
This commit is contained in:
Mark Backman
2025-07-25 21:25:01 -04:00
parent df211fb5a3
commit 7253077bb7
6 changed files with 235 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
src/pipecat/runner/__init__.py
View File

@@ -0,0 +1 @@
"""Pipecat runner package for local and cloud bot execution."""

106
src/pipecat/runner/cloud.py
View File

@@ -4,7 +4,63 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Cloud-compatible development server - simplified without subprocesses."""
"""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.
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
# 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
transport = DailyTransport(session_args.room_url, session_args.token, session_args.body)
elif hasattr(session_args, 'webrtc_connection'):
# WebRTC transport setup
transport = SmallWebRTCTransport(..., webrtc_connection=session_args.webrtc_connection)
# Run your bot logic
await run_bot_logic(transport)
Supported transports:
- Daily - Creates rooms and tokens, runs bot as participant
- LiveKit - Creates rooms and tokens, runs bot as agent
- WebRTC - Provides local WebRTC interface with prebuilt UI
- Telephony - Handles webhook and WebSocket connections for Twilio, Telnyx, Plivo
Example::
async def bot(session_args):
# Detect transport type from session_args
if hasattr(session_args, "room_url"):
# Daily or LiveKit
transport = create_daily_transport(session_args)
elif hasattr(session_args, "webrtc_connection"):
# WebRTC
transport = create_webrtc_transport(session_args)
# Your bot implementation
await run_pipeline(transport)
if __name__ == "__main__":
from pipecat.runner.cloud import main
main()
Then run: `python bot.py -t daily` or `python bot.py -t webrtc`
"""
import argparse
import asyncio
@@ -20,14 +76,13 @@ from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import RedirectResponse
from loguru import logger
# Import the common transport utility functions
from .transport_utilities import setup_websocket_routes
from pipecat.runner.utils import setup_websocket_routes
load_dotenv(override=True)
os.environ["LOCAL_RUN"] = "1"
def get_bot_module():
def _get_bot_module():
"""Get the bot module from the calling script."""
import importlib.util
@@ -68,12 +123,12 @@ def get_bot_module():
)
async def run_bot_directly(transport_type: str, **kwargs):
async def _run_bot_directly(transport_type: str, **kwargs):
"""Run a bot directly in the same process - no subprocess needed."""
if transport_type == "webrtc":
if "webrtc_connection" in kwargs:
# Direct WebRTC connection
bot_module = get_bot_module()
bot_module = _get_bot_module()
class WebRTCSessionArgs:
def __init__(self, webrtc_connection):
@@ -88,7 +143,7 @@ async def run_bot_directly(transport_type: str, **kwargs):
elif transport_type in ["twilio", "telnyx", "plivo"]:
if "websocket" in kwargs:
# Direct WebSocket connection
bot_module = get_bot_module()
bot_module = _get_bot_module()
class WebSocketSessionArgs:
def __init__(self, transport_type, websocket, call_info):
@@ -104,7 +159,7 @@ async def run_bot_directly(transport_type: str, **kwargs):
await bot_module.bot(session_args)
def create_server_app(transport_type: str, host: str = "0.0.0.0", proxy: str = None):
def _create_server_app(transport_type: str, host: str = "0.0.0.0", proxy: str = None):
"""Create FastAPI app with transport-specific routes."""
app = FastAPI()
@@ -163,13 +218,13 @@ def create_server_app(transport_type: str, host: str = "0.0.0.0", proxy: str = N
# Run bot directly
background_tasks.add_task(
run_bot_directly, "webrtc", webrtc_connection=pipecat_connection
_run_bot_directly, "webrtc", webrtc_connection=pipecat_connection
)
answer = pipecat_connection.get_answer()
if host and host != "0.0.0.0":
from .transport_utilities import smallwebrtc_sdp_munging
from pipecat.runner.utils import smallwebrtc_sdp_munging
answer["sdp"] = smallwebrtc_sdp_munging(answer["sdp"], host)
@@ -189,7 +244,7 @@ def create_server_app(transport_type: str, host: str = "0.0.0.0", proxy: str = N
app.router.lifespan_context = lifespan
elif transport_type in ["twilio", "telnyx", "plivo"]:
setup_websocket_routes(app, run_bot_directly, transport_type, proxy)
setup_websocket_routes(app, _run_bot_directly, transport_type, proxy)
# Add general routes
@app.get("/")
@@ -201,13 +256,13 @@ def create_server_app(transport_type: str, host: str = "0.0.0.0", proxy: str = N
# Create Daily room and start bot
import aiohttp
from .daily_runner import configure
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()
bot_module = _get_bot_module()
class DailySessionArgs:
def __init__(self, room_url, token):
@@ -226,12 +281,12 @@ def create_server_app(transport_type: str, host: str = "0.0.0.0", proxy: str = N
elif transport_type == "livekit":
# Create LiveKit room and start bot
from .livekit_runner import configure
from pipecat.runner.livekit import configure
url, token, room_name = await configure()
# Start the bot in the background to join the room
bot_module = get_bot_module()
bot_module = _get_bot_module()
class LiveKitSessionArgs:
def __init__(self, url, token, room_name):
@@ -262,13 +317,13 @@ def create_server_app(transport_type: str, host: str = "0.0.0.0", proxy: str = N
if transport_type == "daily":
import aiohttp
from .daily_runner import configure
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()
bot_module = _get_bot_module()
class DailySessionArgs:
def __init__(self, room_url, token):
@@ -294,7 +349,20 @@ def create_server_app(transport_type: str, host: str = "0.0.0.0", proxy: str = N
def main():
"""Main entry point for cloud-compatible server."""
"""Start the cloud-compatible development server.
Parses command-line arguments and starts a FastAPI server configured
for the specified transport type. The server will discover and run
any bot() function found in the current directory.
Command-line arguments:
--host: Server host address (default: 0.0.0.0)
--port: Server port (default: 7860)
-t/--transport: Transport type (daily, livekit, webrtc, twilio, telnyx, plivo)
-x/--proxy: Public proxy hostname for telephony webhooks
The bot file must contain a `bot(session_args)` function as the entry point.
"""
parser = argparse.ArgumentParser(description="Pipecat Cloud-Compatible Development Server")
parser.add_argument("--host", type=str, default="0.0.0.0", help="Host address")
parser.add_argument("--port", type=int, default=7860, help="Port number")
@@ -311,7 +379,7 @@ def main():
args = parser.parse_args()
# Create the app with transport-specific setup
app = create_server_app(args.transport, args.host, args.proxy)
app = _create_server_app(args.transport, args.host, args.proxy)
# Run the server
uvicorn.run(app, host=args.host, port=args.port)

29
src/pipecat/runner/daily.py
View File

@@ -4,7 +4,30 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Daily.co room configuration utilities for Pipecat examples."""
"""Daily room and token configuration utilities.
This module provides helper functions for creating and configuring Daily rooms
and authentication tokens. It handles both command-line argument parsing and
environment variable configuration.
The module supports creating temporary rooms for development or using existing
rooms specified via arguments or environment variables.
Required environment variables:
- DAILY_API_KEY - Daily API key for room/token creation
- DAILY_SAMPLE_ROOM_URL (optional) - Existing room URL to use
- DAILY_SAMPLE_ROOM_TOKEN (optional) - Existing token to use
Example::
import aiohttp
from pipecat.runner.daily import configure
async with aiohttp.ClientSession() as session:
room_url, token = await configure(session)
# Use room_url and token with DailyTransport
"""
import argparse
import os
@@ -16,7 +39,7 @@ from pipecat.transports.services.helpers.daily_rest import DailyRESTHelper
async def configure(aiohttp_session: aiohttp.ClientSession):
"""Configure Daily.co room URL and token from arguments or environment.
"""Configure Daily room URL and token from arguments or environment.
Args:
aiohttp_session: HTTP session for making API requests.
@@ -34,7 +57,7 @@ async def configure(aiohttp_session: aiohttp.ClientSession):
async def configure_with_args(
aiohttp_session: aiohttp.ClientSession, parser: Optional[argparse.ArgumentParser] = None
):
"""Configure Daily.co room with command-line argument parsing.
"""Configure Daily room with command-line argument parsing.
Args:
aiohttp_session: HTTP session for making API requests.

24
src/pipecat/runner/livekit.py
View File

@@ -4,7 +4,29 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""LiveKit room configuration utilities for Pipecat applications."""
"""LiveKit room and token configuration utilities.
This module provides helper functions for creating and configuring LiveKit
rooms and authentication tokens. It handles JWT token generation with
appropriate grants for both regular participants and AI agents.
The module supports creating tokens for development and testing, with
automatic agent detection for proper room permissions.
Required environment variables:
- LIVEKIT_API_KEY - LiveKit API key
- LIVEKIT_API_SECRET - LiveKit API secret
- LIVEKIT_URL - LiveKit server URL
- LIVEKIT_ROOM_NAME - Room name to join
Example::
from pipecat.runner.livekit import configure
url, token, room_name = await configure()
# Use with LiveKitTransport
"""
import argparse
import os

107
src/pipecat/runner/local.py
View File

@@ -4,29 +4,60 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Direct execution runner for local-only examples."""
"""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.
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 os
import sys
from contextlib import asynccontextmanager
from typing import Callable, Dict, Mapping, Optional
import aiohttp
import uvicorn
from dotenv import load_dotenv
from fastapi import BackgroundTasks, FastAPI
from fastapi.concurrency import asynccontextmanager
from fastapi.responses import RedirectResponse
from loguru import logger
# Import the common transport utility functions
from .transport_utilities import setup_websocket_routes
from pipecat.runner.utils import setup_websocket_routes
load_dotenv(override=True)
def run_webrtc(
def _run_webrtc(
run: Callable, args: argparse.Namespace, transport_params: Mapping[str, Callable] = {}
):
"""Run using WebRTC transport with FastAPI server."""
@@ -91,7 +122,7 @@ def run_webrtc(
answer = pipecat_connection.get_answer()
if args.esp32 and args.host:
from .transport_utilities import smallwebrtc_sdp_munging
from pipecat.runner.utils import smallwebrtc_sdp_munging
answer["sdp"] = smallwebrtc_sdp_munging(answer["sdp"], args.host)
@@ -112,16 +143,13 @@ def run_webrtc(
uvicorn.run(app, host=args.host, port=args.port)
def run_twilio(
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 (
FastAPIWebsocketParams,
FastAPIWebsocketTransport,
)
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}")
@@ -157,16 +185,13 @@ def run_twilio(
uvicorn.run(app, host=args.host, port=args.port)
def run_telnyx(
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 (
FastAPIWebsocketParams,
FastAPIWebsocketTransport,
)
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}")
@@ -202,16 +227,13 @@ def run_telnyx(
uvicorn.run(app, host=args.host, port=args.port)
def run_plivo(
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 (
FastAPIWebsocketParams,
FastAPIWebsocketTransport,
)
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}")
@@ -245,10 +267,10 @@ def run_plivo(
uvicorn.run(app, host=args.host, port=args.port)
def run_daily(
def _run_daily(
run: Callable, args: argparse.Namespace, transport_params: Mapping[str, Callable] = {}
):
"""Run using Daily.co transport."""
"""Run using Daily transport."""
try:
from pipecat.runner.daily import configure
from pipecat.transports.services.daily import DailyParams, DailyTransport
@@ -269,7 +291,7 @@ def run_daily(
asyncio.run(run_daily_impl())
def run_livekit(
def _run_livekit(
run: Callable, args: argparse.Namespace, transport_params: Mapping[str, Callable] = {}
):
"""Run using LiveKit transport."""
@@ -292,7 +314,7 @@ def run_livekit(
asyncio.run(run_livekit_impl())
def run_main(
def _run_main(
run: Callable, args: argparse.Namespace, transport_params: Mapping[str, Callable] = {}
):
"""Run the application with the specified transport type."""
@@ -302,17 +324,17 @@ def run_main(
match args.transport:
case "daily":
run_daily(run, args, transport_params)
_run_daily(run, args, transport_params)
case "livekit":
run_livekit(run, args, transport_params)
_run_livekit(run, args, transport_params)
case "plivo":
run_plivo(run, args, transport_params)
_run_plivo(run, args, transport_params)
case "telnyx":
run_telnyx(run, args, transport_params)
_run_telnyx(run, args, transport_params)
case "twilio":
run_twilio(run, args, transport_params)
_run_twilio(run, args, transport_params)
case "webrtc":
run_webrtc(run, args, transport_params)
_run_webrtc(run, args, transport_params)
def main(
@@ -321,7 +343,26 @@ def main(
parser: Optional[argparse.ArgumentParser] = None,
transport_params: Mapping[str, Callable] = {},
):
"""Main entry point for running Pipecat applications with transport selection."""
"""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:
--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")
@@ -359,4 +400,4 @@ def main(
logger.remove(0)
logger.add(sys.stderr, level="TRACE" if args.verbose else "DEBUG")
run_main(run, args, transport_params)
_run_main(run, args, transport_params)

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

@@ -4,10 +4,32 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Common transport utility functions shared between server.py and run.py."""
"""Transport utility functions and FastAPI route setup helpers.
This module provides common functionality for setting up transport-specific
FastAPI routes and handling WebRTC/WebSocket connections. It includes SDP
manipulation utilities for WebRTC compatibility and transport detection helpers.
Key features:
- WebRTC route setup with connection management
- WebSocket route setup for telephony providers
- SDP munging for ESP32 and other WebRTC compatibility
- Transport client ID detection across different transport types
- Video capture utilities for Daily transports
The utilities are designed to be transport-agnostic where possible, with
specific handlers for each transport type's unique requirements.
Example::
from pipecat.runner.utils import setup_webrtc_routes
app = FastAPI()
setup_webrtc_routes(app, bot_runner_function, host="localhost")
"""
import json
import os
import re
from typing import Any, Callable, Dict