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

Checkpoint: local working, server: daily,webrtc working

Browse Source
This commit is contained in:
Mark Backman
2025-07-25 16:47:42 -04:00
parent 2cdc42adf6
commit b7e415a659
4 changed files with 903 additions and 528 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -4,305 +4,39 @@
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Pipecat runner with support for multiple transport types.
This module provides a unified interface for running Pipecat applications
across different transport types including Daily.co, WebRTC, and Twilio. It
handles setup, configuration, and lifecycle management for each transport type.
Example usage:
DailyTransport::
python bot.py --transport daily
LiveKitTransport::
python bot.py --transport livekit
Plivo::
python bot.py --transport plivo --proxy username.ngrok.io
# Note: Concurrently, run an ngrok tunnel to your local server:
# ngrok http 7860
SmallWebRTCTransport::
python bot.py --transport webrtc
Telnyx::
python bot.py --transport telnyx --proxy username.ngrok.io
# Note: Concurrently, run an ngrok tunnel to your local server:
# ngrok http 7860
Twilio::
python bot.py --transport twilio --proxy username.ngrok.io
# Note: Concurrently, run an ngrok tunnel to your local server:
# ngrok http 7860
"""
"""Direct execution runner for local-only examples."""
import argparse
import asyncio
import json
import os
import re
import sys
from contextlib import asynccontextmanager
from typing import Any, Callable, Dict, Mapping, Optional
from typing import Callable, Dict, Mapping, Optional
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
from fastapi import BackgroundTasks, FastAPI
from fastapi.concurrency import asynccontextmanager
from fastapi.responses import RedirectResponse
from loguru import logger
from pipecat.transports.base_transport import BaseTransport, TransportParams
# Import the common transport utility functions
from .transport_utilities import setup_websocket_routes
# Load environment variables
load_dotenv(override=True)
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.
Args:
transport: The transport instance.
client: Transport-specific client object.
Returns:
Client identifier string, empty if transport not supported.
"""
# Import conditionally to avoid dependency issues
try:
from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
if isinstance(transport, SmallWebRTCTransport):
return client.pc_id
except ImportError:
pass
try:
from pipecat.transports.services.daily import DailyTransport
if isinstance(transport, DailyTransport):
return client["id"]
except ImportError:
pass
logger.warning(f"Unable to get client id from unsupported transport {type(transport)}")
return ""
async def maybe_capture_participant_camera(
transport: BaseTransport, client: Any, framerate: int = 0
):
"""Capture participant camera video if transport supports it.
Args:
transport: The transport instance.
client: Transport-specific client object.
framerate: Video capture framerate. Defaults to 0 (auto).
"""
try:
from pipecat.transports.services.daily import DailyTransport
if isinstance(transport, DailyTransport):
await transport.capture_participant_video(
client["id"], framerate=framerate, video_source="camera"
)
except ImportError:
pass
async def maybe_capture_participant_screen(
transport: BaseTransport, client: Any, framerate: int = 0
):
"""Capture participant screen video if transport supports it.
Args:
transport: The transport instance.
client: Transport-specific client object.
framerate: Video capture framerate. Defaults to 0 (auto).
"""
try:
from pipecat.transports.services.daily import DailyTransport
if isinstance(transport, DailyTransport):
await transport.capture_participant_video(
client["id"], framerate=framerate, video_source="screenVideo"
)
except ImportError:
pass
def smallwebrtc_sdp_cleanup_ice_candidates(text: str, pattern: str) -> str:
"""Clean up ICE candidates in SDP text for SmallWebRTC.
Args:
text: SDP text to clean up.
pattern: Pattern to match for candidate filtering.
Returns:
Cleaned SDP text with filtered ICE candidates.
"""
result = []
lines = text.splitlines()
for line in lines:
if re.search("a=candidate", line):
if re.search(pattern, line) and not re.search("raddr", line):
result.append(line)
else:
result.append(line)
return "\r\n".join(result)
def smallwebrtc_sdp_cleanup_fingerprints(text: str) -> str:
"""Remove unsupported fingerprint algorithms from SDP text.
Args:
text: SDP text to clean up.
Returns:
SDP text with sha-384 and sha-512 fingerprints removed.
"""
result = []
lines = text.splitlines()
for line in lines:
if not re.search("sha-384", line) and not re.search("sha-512", line):
result.append(line)
return "\r\n".join(result)
def smallwebrtc_sdp_munging(sdp: str, host: str) -> str:
"""Apply SDP modifications for SmallWebRTC compatibility.
Args:
sdp: Original SDP string.
host: Host address for ICE candidate filtering.
Returns:
Modified SDP string with fingerprint and ICE candidate cleanup.
"""
sdp = smallwebrtc_sdp_cleanup_fingerprints(sdp)
sdp = smallwebrtc_sdp_cleanup_ice_candidates(sdp, host)
return sdp
def run_daily(
run: Callable,
args: argparse.Namespace,
transport_params: Mapping[str, Callable] = {},
):
"""Run using Daily.co transport.
Args:
run: The function to run.
args: Parsed command-line arguments.
transport_params: Mapping of transport names to parameter factory functions.
"""
try:
from pipecat.runner.daily_runner import configure
from pipecat.transports.services.daily import DailyParams, DailyTransport
except ImportError as e:
logger.error(
f"Daily transport dependencies not installed. Install with: {get_install_command('daily')}"
)
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)
# Run function with DailyTransport transport arguments.
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.
Args:
run: The function to run.
args: Parsed command-line arguments.
transport_params: Mapping of transport names to parameter factory functions.
"""
try:
from pipecat.runner.livekit_runner import configure
from pipecat.transports.services.livekit import LiveKitParams, LiveKitTransport
except ImportError as e:
logger.error(
f"LiveKit transport dependencies not installed. Install with: {get_install_command('livekit')}"
)
logger.debug(f"Import error: {e}")
return
logger.info("Running with LiveKitTransport...")
async def run_livekit_impl():
(url, token, room_name) = await configure()
# Run function with LiveKit transport arguments.
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_webrtc(
run: Callable,
args: argparse.Namespace,
transport_params: Mapping[str, Callable] = {},
run: Callable, args: argparse.Namespace, transport_params: Mapping[str, Callable] = {}
):
"""Run using WebRTC transport with FastAPI server.
Args:
run: The function to run.
args: Parsed command-line arguments.
transport_params: Mapping of transport names to parameter factory functions.
"""
"""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. Install with: {get_install_command('webrtc')}"
)
logger.error(f"WebRTC transport dependencies not installed.")
logger.debug(f"Import error: {e}")
return
@@ -310,7 +44,7 @@ def run_webrtc(
app = FastAPI()
# Store connections by pc_id
# Store connections by pc_id (like the working version)
pcs_map: Dict[str, SmallWebRTCConnection] = {}
# Mount the frontend at /
@@ -323,15 +57,7 @@ def run_webrtc(
@app.post("/api/offer")
async def offer(request: dict, background_tasks: BackgroundTasks):
"""Handle WebRTC offer requests and manage peer connections.
Args:
request: WebRTC offer request containing SDP and connection details.
background_tasks: FastAPI background tasks for running applications.
Returns:
WebRTC answer with connection details.
"""
"""Handle WebRTC offer requests and manage peer connections."""
pc_id = request.get("pc_id")
if pc_id and pc_id in pcs_map:
@@ -348,22 +74,25 @@ def run_webrtc(
@pipecat_connection.event_handler("closed")
async def handle_disconnected(webrtc_connection: SmallWebRTCConnection):
"""Handle WebRTC connection closure and cleanup.
Args:
webrtc_connection: The closed WebRTC connection.
"""
"""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.
params: TransportParams = transport_params[args.transport]()
# Run function with SmallWebRTC transport arguments (like working version)
params = transport_params[args.transport]()
transport = SmallWebRTCTransport(params=params, webrtc_connection=pipecat_connection)
background_tasks.add_task(run, transport, args, False)
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 .transport_utilities import smallwebrtc_sdp_munging
answer["sdp"] = smallwebrtc_sdp_munging(answer["sdp"], args.host)
# Updating the peer connection inside the map
@@ -373,34 +102,20 @@ def run_webrtc(
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Manage FastAPI application lifecycle and cleanup connections.
Args:
app: The FastAPI application instance.
Yields:
Control to the FastAPI application runtime.
"""
"""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: Callable, args: argparse.Namespace, transport_params: Mapping[str, Callable] = {}
):
"""Run using Twilio transport with FastAPI WebSocket server.
Args:
run: The function to run.
args: Parsed command-line arguments.
transport_params: Mapping of transport names to parameter factory functions.
"""
"""Run using Twilio transport with FastAPI WebSocket server."""
try:
from pipecat.serializers.twilio import TwilioFrameSerializer
from pipecat.transports.network.fastapi_websocket import (
@@ -408,9 +123,7 @@ def run_twilio(
FastAPIWebsocketTransport,
)
except ImportError as e:
logger.error(
f"Twilio transport dependencies not installed. Install with: {get_install_command('twilio')}"
)
logger.error(f"Twilio transport dependencies not installed.")
logger.debug(f"Import error: {e}")
return
@@ -418,79 +131,36 @@ def run_twilio(
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # Allow all origins for testing
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 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"]
@app.post("/")
async def start_call():
"""Handle Twilio webhook and return TwiML response.
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", ""),
)
Returns:
TwiML XML response directing call to WebSocket stream.
"""
logger.debug("POST TwiML")
transport = FastAPIWebsocketTransport(websocket=kwargs["websocket"], params=params)
xml_content = f"""<?xml version="1.0" encoding="UTF-8"?>
<Response>
<Connect>
<Stream url="wss://{args.proxy}/ws"></Stream>
</Connect>
<Pause length="40"/>
</Response>
"""
return HTMLResponse(content=xml_content, media_type="application/xml")
class MockArgs:
def __init__(self):
self.transport = "twilio"
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
"""Handle Twilio WebSocket connections for voice streaming.
Args:
websocket: The WebSocket connection from Twilio.
"""
await websocket.accept()
logger.debug("WebSocket connection accepted")
# Reading Twilio data.
start_data = websocket.iter_text()
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"]
# Create websocket transport and update params.
params: FastAPIWebsocketParams = transport_params[args.transport]()
params.add_wav_header = False
params.serializer = TwilioFrameSerializer(
stream_sid=stream_sid,
call_sid=call_sid,
account_sid=os.getenv("TWILIO_ACCOUNT_SID", ""),
auth_token=os.getenv("TWILIO_AUTH_TOKEN", ""),
)
transport = FastAPIWebsocketTransport(websocket=websocket, params=params)
await run(transport, args, False)
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: Callable, args: argparse.Namespace, transport_params: Mapping[str, Callable] = {}
):
"""Run using Telnyx transport with FastAPI WebSocket server.
Args:
run: The function to run.
args: Parsed command-line arguments.
transport_params: Mapping of transport names to parameter factory functions.
"""
"""Run using Telnyx transport with FastAPI WebSocket server."""
try:
from pipecat.serializers.telnyx import TelnyxFrameSerializer
from pipecat.transports.network.fastapi_websocket import (
@@ -498,9 +168,7 @@ def run_telnyx(
FastAPIWebsocketTransport,
)
except ImportError as e:
logger.error(
f"Telnyx transport dependencies not installed. Install with: {get_install_command('telnyx')}"
)
logger.error(f"Telnyx transport dependencies not installed.")
logger.debug(f"Import error: {e}")
return
@@ -508,80 +176,36 @@ def run_telnyx(
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # Allow all origins for testing
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 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"]
@app.post("/")
async def start_call():
"""Handle Telnyx webhook and return TeXML response.
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",
)
Returns:
TeXML XML response directing call to WebSocket stream.
"""
logger.debug("POST TeXML")
transport = FastAPIWebsocketTransport(websocket=kwargs["websocket"], params=params)
xml_content = f"""<?xml version="1.0" encoding="UTF-8"?>
<Response>
<Connect>
<Stream url="wss://{args.proxy}/ws" bidirectionalMode="rtp"></Stream>
</Connect>
<Pause length="40"/>
</Response>
"""
return HTMLResponse(content=xml_content, media_type="application/xml")
class MockArgs:
def __init__(self):
self.transport = "telnyx"
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
"""Handle Telnyx WebSocket connections for voice streaming.
Args:
websocket: The WebSocket connection from Telnyx.
"""
await websocket.accept()
logger.debug("WebSocket connection accepted")
# Reading Telnyx data.
start_data = websocket.iter_text()
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"]
# Create websocket transport and update params.
params: FastAPIWebsocketParams = transport_params[args.transport]()
params.add_wav_header = False
params.serializer = TelnyxFrameSerializer(
stream_id=stream_id,
call_control_id=call_control_id,
outbound_encoding=outbound_encoding,
inbound_encoding="PCMU", # You might want to make this configurable
)
transport = FastAPIWebsocketTransport(websocket=websocket, params=params)
await run(transport, args, False)
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: Callable, args: argparse.Namespace, transport_params: Mapping[str, Callable] = {}
):
"""Run using Plivo transport with FastAPI WebSocket server.
Args:
run: The function to run.
args: Parsed command-line arguments.
transport_params: Mapping of transport names to parameter factory functions.
"""
"""Run using Plivo transport with FastAPI WebSocket server."""
try:
from pipecat.serializers.plivo import PlivoFrameSerializer
from pipecat.transports.network.fastapi_websocket import (
@@ -589,9 +213,7 @@ def run_plivo(
FastAPIWebsocketTransport,
)
except ImportError as e:
logger.error(
f"Plivo transport dependencies not installed. Install with: {get_install_command('plivo')}"
)
logger.error(f"Plivo transport dependencies not installed.")
logger.debug(f"Import error: {e}")
return
@@ -599,89 +221,83 @@ def run_plivo(
app = FastAPI()
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # Allow all origins for testing
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 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"]
@app.post("/")
async def start_call():
"""Handle Plivo webhook and return Plivo XML response.
params = transport_params["plivo"]()
params.add_wav_header = False
params.serializer = PlivoFrameSerializer(
stream_id=call_info["stream_id"],
call_id=call_info["call_id"],
)
Returns:
Plivo XML response directing call to WebSocket stream.
"""
logger.debug("POST Plivo XML")
transport = FastAPIWebsocketTransport(websocket=kwargs["websocket"], params=params)
xml_content = f"""<?xml version="1.0" encoding="UTF-8"?>
<Response>
<Stream bidirectional="true" keepCallAlive="true" contentType="audio/x-mulaw;rate=8000">wss://{args.proxy}/ws</Stream>
</Response>
"""
return HTMLResponse(content=xml_content, media_type="application/xml")
class MockArgs:
def __init__(self):
self.transport = "plivo"
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
"""Handle Plivo WebSocket connections for voice streaming.
Args:
websocket: The WebSocket connection from Plivo.
"""
await websocket.accept()
logger.debug("WebSocket connection accepted")
# Reading Plivo data.
start_data = websocket.iter_text()
start_message = json.loads(await start_data.__anext__())
logger.debug(f"Received start message: {start_message}")
# Extract stream_id and call_id from the start event
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}")
# Create websocket transport and update params.
params: FastAPIWebsocketParams = transport_params[args.transport]()
params.add_wav_header = False
params.serializer = PlivoFrameSerializer(
stream_id=stream_id,
call_id=call_id,
)
transport = FastAPIWebsocketTransport(websocket=websocket, params=params)
await run(transport, args, False)
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_main(
run: Callable,
args: argparse.Namespace,
transport_params: Mapping[str, Callable] = {},
def run_daily(
run: Callable, args: argparse.Namespace, transport_params: Mapping[str, Callable] = {}
):
"""Run the application with the specified transport type.
"""Run using Daily.co transport."""
try:
from pipecat.runner.daily_runner 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
Args:
run: The function to run.
args: Parsed command-line arguments.
transport_params: Mapping of transport names to parameter factory functions.
"""
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_runner 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.")
logger.info(
f"To add {args.transport} support, install with: {get_install_command(args.transport)}"
)
return
match args.transport:
@@ -705,15 +321,10 @@ def main(
parser: Optional[argparse.ArgumentParser] = None,
transport_params: Mapping[str, Callable] = {},
):
"""Main entry point for running Pipecat applications with transport selection.
Args:
run: The function to run.
parser: Optional argument parser. If None, creates a default one.
transport_params: Mapping of transport names to parameter factory functions.
"""
"""Main entry point for running Pipecat applications with transport selection."""
if not parser:
parser = argparse.ArgumentParser(description="Pipecat Bot Runner")
parser.add_argument(
"--host", default="localhost", help="Host for HTTP server (default: localhost)"
)
@@ -724,7 +335,9 @@ def main(
"--transport",
"-t",
type=str,
choices=["daily", "livekit", "plivo", "telnyx", "twilio", "webrtc"],
choices=list(transport_params.keys())
if transport_params
else ["daily", "livekit", "plivo", "telnyx", "twilio", "webrtc"],
default="webrtc",
help="The transport this application should use",
)
@@ -735,6 +348,7 @@ def main(
"--esp32", action="store_true", default=False, help="Perform SDP munging for the ESP32"
)
parser.add_argument("--verbose", "-v", action="count", default=0)
args = parser.parse_args()
if args.esp32 and args.host == "localhost":
@@ -745,5 +359,4 @@ def main(
logger.remove(0)
logger.add(sys.stderr, level="TRACE" if args.verbose else "DEBUG")
# Import the bot file
run_main(run, args, transport_params)

143
src/pipecat/runner/runner.py Normal file
View File

@@ -0,0 +1,143 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import argparse
import asyncio
import importlib.util
import os
import sys
def find_and_import_bot():
"""Find and import the bot function from the current working directory."""
cwd = os.getcwd()
# Add current working directory to Python path if not already there
if cwd not in sys.path:
sys.path.insert(0, cwd)
# Try to find bot.py in current directory
bot_file = os.path.join(cwd, "bot.py")
if os.path.exists(bot_file):
spec = importlib.util.spec_from_file_location("bot", bot_file)
bot_module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(bot_module)
return bot_module
# Try to import bot module directly
try:
import bot
return bot
except ImportError:
pass
# Look for any .py file in current directory that has a bot function
for filename in os.listdir(cwd):
if filename.endswith(".py") and filename != "runner.py":
try:
module_name = filename[:-3] # Remove .py extension
spec = importlib.util.spec_from_file_location(
module_name, os.path.join(cwd, filename)
)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
if hasattr(module, "bot"):
return module
except Exception:
continue
raise ImportError(
"Could not find 'bot' function. Make sure your bot file has a 'bot' function."
)
def main():
"""Parse args and launch the bot with specified transport."""
parser = argparse.ArgumentParser(description="Pipecat Bot Runner")
parser.add_argument("-u", "--url", type=str, required=False, help="Daily room URL")
parser.add_argument("-t", "--token", type=str, required=False, help="Daily room token")
parser.add_argument(
"--transport",
type=str,
choices=["daily", "livekit", "webrtc"],
default="daily",
help="Transport type",
)
parser.add_argument("--room", type=str, required=False, help="LiveKit room name")
args, unknown = parser.parse_known_args()
# Find and import the bot function
try:
bot_module = find_and_import_bot()
bot_function = bot_module.bot
except ImportError as e:
print(f"Error: {e}")
sys.exit(1)
if args.transport == "daily":
url = args.url or os.getenv("DAILY_SAMPLE_ROOM_URL")
token = args.token or os.getenv("DAILY_SAMPLE_ROOM_TOKEN")
if not url or not token:
raise Exception("Daily room URL and token are required.")
# Create Daily session arguments
try:
from pipecatcloud.agent import DailySessionArguments
session_args = DailySessionArguments(
room_url=url,
token=token,
body={},
session_id=None,
)
except ImportError:
# Fallback for local development
class LocalDailySessionArgs:
def __init__(self, room_url, token, body=None):
self.room_url = room_url
self.token = token
self.body = body or {}
session_args = LocalDailySessionArgs(url, token)
elif args.transport == "livekit":
url = args.url or os.getenv("LIVEKIT_URL")
token = args.token or os.getenv("LIVEKIT_TOKEN")
room_name = args.room or os.getenv("LIVEKIT_ROOM_NAME")
if not url or not token or not room_name:
raise Exception("LiveKit URL, token, and room name are required.")
class LiveKitSessionArgs:
def __init__(self, url, token, room_name):
self.url = url
self.token = token
self.room_name = room_name
self.body = {}
session_args = LiveKitSessionArgs(url, token, room_name)
elif args.transport == "webrtc":
# For WebRTC subprocess mode (not typically used)
class WebRTCSessionArgs:
def __init__(self):
self.transport_type = "webrtc"
self.body = {}
session_args = WebRTCSessionArgs()
else:
raise Exception(f"Unsupported transport: {args.transport}")
asyncio.run(bot_function(session_args))
if __name__ == "__main__":
main()

303
src/pipecat/runner/server.py Normal file
View File

@@ -0,0 +1,303 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Cloud-compatible development server that uses subprocess to run bots."""
import argparse
import asyncio
import os
import subprocess
import sys
from typing import Dict
import uvicorn
from dotenv import load_dotenv
from fastapi import BackgroundTasks, FastAPI
from fastapi.concurrency import asynccontextmanager
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_webrtc_routes, setup_websocket_routes
load_dotenv(override=True)
os.environ["LOCAL_RUN"] = "1"
# Track bot processes
bot_procs = {}
def cleanup():
"""Cleanup function to terminate all bot processes."""
for entry in bot_procs.values():
proc = entry[0]
proc.terminate()
proc.wait()
def get_bot_module():
"""Get the bot module from the calling script."""
# Get the main module (the file that was executed)
main_module = sys.modules["__main__"]
# Check if it has a bot function
if hasattr(main_module, "bot"):
return main_module
# Try to import 'bot' module from current directory
try:
import bot
return bot
except ImportError:
raise ImportError(
"Could not find 'bot' function. Make sure your script has a 'bot' function or there's a 'bot.py' file in the current directory."
)
async def run_subprocess_bot(transport_type: str, **kwargs):
"""Run a bot via subprocess - used by transport handlers."""
if transport_type == "daily":
import aiohttp
from .daily_runner import configure
async with aiohttp.ClientSession() as session:
room_url, token = await configure(session)
proc = subprocess.Popen(
[
f"LOCAL_RUN=1 python3 -m pipecat.runner.runner -u {room_url} -t {token} --transport daily"
],
shell=True,
bufsize=1,
cwd=os.getcwd(),
) # Run from current working directory
bot_procs[proc.pid] = (proc, room_url)
return room_url, token
elif transport_type == "livekit":
from .livekit_runner import configure
url, token, room_name = await configure()
proc = subprocess.Popen(
[
f"LOCAL_RUN=1 python3 -m pipecat.runner.runner --transport livekit --url {url} --token {token} --room {room_name}"
],
shell=True,
bufsize=1,
cwd=os.getcwd(),
) # Run from current working directory
bot_procs[proc.pid] = (proc, url)
return url, token, room_name
elif transport_type == "webrtc":
if "webrtc_connection" in kwargs:
# Direct connection mode - 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(kwargs["webrtc_connection"])
await bot_module.bot(session_args)
else:
# Subprocess mode (rarely used for WebRTC)
proc = subprocess.Popen(
[f"LOCAL_RUN=1 python3 -m pipecat.runner.runner --transport webrtc"],
shell=True,
bufsize=1,
cwd=os.getcwd(),
) # Run from current working directory
bot_procs[proc.pid] = (proc, "webrtc")
elif transport_type in ["twilio", "telnyx", "plivo"]:
if "websocket" in kwargs:
# Direct WebSocket mode - run bot directly
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 = WebSocketSessionArgs(
transport_type, kwargs["websocket"], kwargs["call_info"]
)
await bot_module.bot(session_args)
else:
# Subprocess mode (rarely used for telephony)
proc = subprocess.Popen(
[f"LOCAL_RUN=1 python3 -m pipecat.runner.runner --transport {transport_type}"],
shell=True,
bufsize=1,
cwd=os.getcwd(),
) # Run from current working directory
bot_procs[proc.pid] = (proc, transport_type)
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()
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Add transport-specific routes
if transport_type == "webrtc":
# Direct WebRTC setup (like the working run.py version)
try:
from pipecat_ai_small_webrtc_prebuilt.frontend import SmallWebRTCPrebuiltUI
from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
except ImportError as e:
logger.error(f"WebRTC transport dependencies not installed.")
return app
# Store connections by pc_id
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 bot directly instead of through run_subprocess_bot
background_tasks.add_task(
run_subprocess_bot, "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
answer["sdp"] = smallwebrtc_sdp_munging(answer["sdp"], 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
elif transport_type in ["twilio", "telnyx", "plivo"]:
setup_websocket_routes(app, run_subprocess_bot, transport_type, proxy)
# Add general routes
@app.get("/")
async def start_agent():
"""Launch a bot and redirect appropriately."""
print(f"Starting bot with {transport_type} transport")
if transport_type == "daily":
result = await run_subprocess_bot("daily")
room_url, token = result
return RedirectResponse(room_url)
elif transport_type == "livekit":
result = await run_subprocess_bot("livekit")
url, token, room_name = result
return RedirectResponse(url)
elif transport_type == "webrtc":
await run_subprocess_bot("webrtc")
return RedirectResponse("/client/")
else:
await run_subprocess_bot(transport_type)
return {"status": f"Bot started with {transport_type}"}
@app.post("/connect")
async def rtvi_connect():
"""Launch a bot and return connection info for RTVI clients."""
print(f"Starting bot with {transport_type} transport")
if transport_type == "daily":
result = await run_subprocess_bot("daily")
room_url, token = result
return {"transport": "daily", "room_url": room_url, "token": token}
elif transport_type == "webrtc":
await run_subprocess_bot("webrtc")
return {"transport": "webrtc", "client_url": "/client/"}
else:
# RTVI only supports Daily and WebRTC
return {
"error": f"RTVI connect not supported for {transport_type} transport. Use Daily or WebRTC."
}
return app
def main():
"""Main entry point for cloud-compatible server."""
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")
parser.add_argument(
"-t",
"--transport",
type=str,
choices=["daily", "livekit", "webrtc", "twilio", "telnyx", "plivo"],
default="webrtc",
help="Transport type",
)
parser.add_argument("--proxy", "-x", help="Public proxy host name")
args = parser.parse_args()
# Create the app with transport-specific setup
app = create_server_app(args.transport, args.host, args.proxy)
# Run the server
uvicorn.run(app, host=args.host, port=args.port)
if __name__ == "__main__":
main()

316
src/pipecat/runner/transport_utilities.py Normal file
View File

@@ -0,0 +1,316 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
"""Common transport utility functions shared between server.py and run.py."""
import json
import os
import re
from typing import Any, Callable, Dict
from fastapi import BackgroundTasks, FastAPI, WebSocket
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import HTMLResponse, RedirectResponse
from loguru import logger
from pipecat.transports.base_transport import BaseTransport
def get_install_command(transport: str) -> str:
"""Get the pip install command for a specific transport."""
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.
Args:
transport: The transport instance.
client: Transport-specific client object.
Returns:
Client identifier string, empty if transport not supported.
"""
# Import conditionally to avoid dependency issues
try:
from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
if isinstance(transport, SmallWebRTCTransport):
return client.pc_id
except ImportError:
pass
try:
from pipecat.transports.services.daily import DailyTransport
if isinstance(transport, DailyTransport):
return client["id"]
except ImportError:
pass
logger.warning(f"Unable to get client id from unsupported transport {type(transport)}")
return ""
async def maybe_capture_participant_camera(
transport: BaseTransport, client: Any, framerate: int = 0
):
"""Capture participant camera video if transport supports it.
Args:
transport: The transport instance.
client: Transport-specific client object.
framerate: Video capture framerate. Defaults to 0 (auto).
"""
try:
from pipecat.transports.services.daily import DailyTransport
if isinstance(transport, DailyTransport):
await transport.capture_participant_video(
client["id"], framerate=framerate, video_source="camera"
)
except ImportError:
pass
async def maybe_capture_participant_screen(
transport: BaseTransport, client: Any, framerate: int = 0
):
"""Capture participant screen video if transport supports it.
Args:
transport: The transport instance.
client: Transport-specific client object.
framerate: Video capture framerate. Defaults to 0 (auto).
"""
try:
from pipecat.transports.services.daily import DailyTransport
if isinstance(transport, DailyTransport):
await transport.capture_participant_video(
client["id"], framerate=framerate, video_source="screenVideo"
)
except ImportError:
pass
def smallwebrtc_sdp_cleanup_ice_candidates(text: str, pattern: str) -> str:
"""Clean up ICE candidates in SDP text for SmallWebRTC.
Args:
text: SDP text to clean up.
pattern: Pattern to match for candidate filtering.
Returns:
Cleaned SDP text with filtered ICE candidates.
"""
result = []
lines = text.splitlines()
for line in lines:
if re.search("a=candidate", line):
if re.search(pattern, line) and not re.search("raddr", line):
result.append(line)
else:
result.append(line)
return "\r\n".join(result)
def smallwebrtc_sdp_cleanup_fingerprints(text: str) -> str:
"""Remove unsupported fingerprint algorithms from SDP text.
Args:
text: SDP text to clean up.
Returns:
SDP text with sha-384 and sha-512 fingerprints removed.
"""
result = []
lines = text.splitlines()
for line in lines:
if not re.search("sha-384", line) and not re.search("sha-512", line):
result.append(line)
return "\r\n".join(result)
def smallwebrtc_sdp_munging(sdp: str, host: str) -> str:
"""Apply SDP modifications for SmallWebRTC compatibility.
Args:
sdp: Original SDP string.
host: Host address for ICE candidate filtering.
Returns:
Modified SDP string with fingerprint and ICE candidate cleanup.
"""
sdp = smallwebrtc_sdp_cleanup_fingerprints(sdp)
sdp = smallwebrtc_sdp_cleanup_ice_candidates(sdp, host)
return sdp
def setup_webrtc_routes(app: FastAPI, transport_runner: Callable, host: str = None):
"""Set up WebRTC routes for an app."""
try:
from pipecat_ai_small_webrtc_prebuilt.frontend import SmallWebRTCPrebuiltUI
from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
except ImportError as e:
logger.error(
f"WebRTC transport dependencies not installed. Install with: {get_install_command('webrtc')}"
)
logger.debug(f"Import error: {e}")
return
# Store connections by pc_id
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 transport with the connection
background_tasks.add_task(
transport_runner, "webrtc", webrtc_connection=pipecat_connection
)
answer = pipecat_connection.get_answer()
if host:
answer["sdp"] = smallwebrtc_sdp_munging(answer["sdp"], host)
# Updating the peer connection inside the map
pcs_map[answer["pc_id"]] = pipecat_connection
return answer
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"""<?xml version="1.0" encoding="UTF-8"?>
<Response>
<Connect>
<Stream url="wss://{proxy}/ws"></Stream>
</Connect>
<Pause length="40"/>
</Response>"""
elif transport_type == "telnyx":
xml_content = f"""<?xml version="1.0" encoding="UTF-8"?>
<Response>
<Connect>
<Stream url="wss://{proxy}/ws" bidirectionalMode="rtp"></Stream>
</Connect>
<Pause length="40"/>
</Response>"""
elif transport_type == "plivo":
xml_content = f"""<?xml version="1.0" encoding="UTF-8"?>
<Response>
<Stream bidirectional="true" keepCallAlive="true" contentType="audio/x-mulaw;rate=8000">wss://{proxy}/ws</Stream>
</Response>"""
else:
xml_content = "<Response></Response>"
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)