diff --git a/examples/runner-examples/cloud-multi-bot.py b/examples/runner-examples/cloud-multi-bot.py index 2a21aead6..ef9c8de89 100644 --- a/examples/runner-examples/cloud-multi-bot.py +++ b/examples/runner-examples/cloud-multi-bot.py @@ -4,6 +4,17 @@ # SPDX-License-Identifier: BSD 2-Clause License # +"""Pipecat Cloud-compatible bot example. + +Transports are: + +- Daily +- SmallWebRTC +- Twilio +- Telnyx +- Plivo +""" + import os from dotenv import load_dotenv @@ -25,11 +36,14 @@ try: except ImportError: raise ImportError( "pipecatcloud package is required for cloud-compatible bots. " - "Install with: pip install pipecat-ai[[pipecatcloud]]" + "Install with: pip install pipecat-ai[pipecatcloud]" ) load_dotenv(override=True) +# Check if we're running locally +IS_LOCAL_RUN = os.environ.get("LOCAL_RUN", "0") == "1" + async def run_bot(transport): """Main bot logic that works with any transport.""" @@ -101,12 +115,18 @@ async def bot( if isinstance(session_args, DailySessionArguments): from pipecat.transports.services.daily import DailyParams, DailyTransport + if not IS_LOCAL_RUN: + from pipecat.audio.filters.krisp_filter import KrispFilter + transport = DailyTransport( session_args.room_url, session_args.token, "Pipecat Bot", params=DailyParams( audio_in_enabled=True, + audio_in_filter=None + if IS_LOCAL_RUN + else KrispFilter(), # Only use Krisp in production audio_out_enabled=True, vad_analyzer=SileroVADAnalyzer(), ), @@ -126,56 +146,62 @@ async def bot( ) elif isinstance(session_args, WebSocketSessionArguments): + # Use the utility to parse WebSocket data + from pipecat.runner.utils import parse_telephony_websocket + + transport_type, stream_id, call_id = await parse_telephony_websocket(session_args.websocket) + logger.info(f"Auto-detected transport: {transport_type}") + + # Create transport based on detected type + if transport_type == "twilio": + from pipecat.serializers.twilio import TwilioFrameSerializer + + serializer = TwilioFrameSerializer( + stream_sid=stream_id, + call_sid=call_id, + account_sid=os.getenv("TWILIO_ACCOUNT_SID", ""), + auth_token=os.getenv("TWILIO_AUTH_TOKEN", ""), + ) + + elif transport_type == "telnyx": + from pipecat.serializers.telnyx import TelnyxFrameSerializer + + serializer = TelnyxFrameSerializer( + stream_id=stream_id, + call_control_id=call_id, + outbound_encoding="PCMU", # Set manually + inbound_encoding="PCMU", + ) + + elif transport_type == "plivo": + from pipecat.serializers.plivo import PlivoFrameSerializer + + serializer = PlivoFrameSerializer( + stream_id=stream_id, + call_id=call_id, + ) + + else: + # Generic fallback + serializer = None + + # Create the transport from pipecat.transports.network.fastapi_websocket import ( FastAPIWebsocketParams, FastAPIWebsocketTransport, ) - # Create base parameters for telephony - params = FastAPIWebsocketParams( - audio_in_enabled=True, - audio_out_enabled=True, - vad_analyzer=SileroVADAnalyzer(), - add_wav_header=False, + transport = FastAPIWebsocketTransport( + websocket=session_args.websocket, + params=FastAPIWebsocketParams( + audio_in_enabled=True, + audio_out_enabled=True, + add_wav_header=False, + vad_analyzer=SileroVADAnalyzer(), + serializer=serializer, + ), ) - # Create appropriate serializer based on transport type - transport_type = getattr(session_args, "transport_type", "unknown") - call_info = getattr(session_args, "call_info", {}) - - if transport_type == "twilio": - from pipecat.serializers.twilio import TwilioFrameSerializer - - params.serializer = TwilioFrameSerializer( - stream_sid=call_info["stream_sid"], - call_sid=call_info["call_sid"], - account_sid=os.getenv("TWILIO_ACCOUNT_SID", ""), - auth_token=os.getenv("TWILIO_AUTH_TOKEN", ""), - ) - elif transport_type == "telnyx": - from pipecat.serializers.telnyx import TelnyxFrameSerializer - - params.serializer = TelnyxFrameSerializer( - stream_id=call_info["stream_id"], - call_control_id=call_info["call_control_id"], - outbound_encoding=call_info["outbound_encoding"], - inbound_encoding="PCMU", - ) - elif transport_type == "plivo": - from pipecat.serializers.plivo import PlivoFrameSerializer - - params.serializer = PlivoFrameSerializer( - stream_id=call_info["stream_id"], - call_id=call_info["call_id"], - ) - else: - raise ValueError(f"Unsupported WebSocket transport type: {transport_type}") - - transport = FastAPIWebsocketTransport(websocket=session_args.websocket, params=params) - - else: - raise ValueError(f"Unsupported session arguments type: {type(session_args)}") - await run_bot(transport) diff --git a/examples/runner-examples/cloud-simple-bot.py b/examples/runner-examples/cloud-simple-bot.py index 345920ae0..f021cdeaf 100644 --- a/examples/runner-examples/cloud-simple-bot.py +++ b/examples/runner-examples/cloud-simple-bot.py @@ -4,6 +4,10 @@ # SPDX-License-Identifier: BSD 2-Clause License # +"""Pipecat Cloud-compatible bot example. + +Transports are Daily or SmallWebRTC.""" + import os from dataclasses import dataclass from typing import Any, Optional diff --git a/examples/runner-examples/requirements.txt b/examples/runner-examples/requirements.txt index 182ac8324..29809da53 100644 --- a/examples/runner-examples/requirements.txt +++ b/examples/runner-examples/requirements.txt @@ -1,4 +1,4 @@ pipecatcloud -pipecat-ai[openai,daily,deepgram,cartesia,silero] +pipecat-ai[openai,daily,deepgram,cartesia,silero,webrtc,websocket] python-dotenv pipecat-ai-small-webrtc-prebuilt \ No newline at end of file diff --git a/src/pipecat/runner/cloud.py b/src/pipecat/runner/cloud.py index 265be77bb..01dd983dc 100644 --- a/src/pipecat/runner/cloud.py +++ b/src/pipecat/runner/cloud.py @@ -23,8 +23,8 @@ are established. Bot function signature:: async def bot(session_args: DailySessionArguments | SmallWebRTCSessionArguments | WebSocketSessionArguments): + # Type-safe access to session arguments - # Create transport based on session_args type if isinstance(session_args, DailySessionArguments): # Daily transport setup - guaranteed to have room_url, token from pipecat.transports.services.daily import DailyTransport, DailyParams @@ -58,13 +58,11 @@ Supported transports: Example:: - async def bot(session_args: DailySessionArguments | SmallWebRTCSessionArguments): - # Detect transport type from session_args + async def bot(session_args): + # Type-safe transport detection if isinstance(session_args, DailySessionArguments): - # Daily transport = create_daily_transport(session_args) elif isinstance(session_args, SmallWebRTCSessionArguments): - # WebRTC transport = create_webrtc_transport(session_args) # Your bot implementation @@ -89,11 +87,9 @@ import uvicorn from dotenv import load_dotenv from fastapi import BackgroundTasks, FastAPI, WebSocket from fastapi.middleware.cors import CORSMiddleware -from fastapi.responses import RedirectResponse +from fastapi.responses import HTMLResponse, RedirectResponse from loguru import logger -from pipecat.runner.utils import setup_websocket_routes - # Require pipecatcloud for cloud-compatible bots try: from pipecatcloud.agent import DailySessionArguments, WebSocketSessionArguments @@ -162,17 +158,12 @@ def _get_bot_module(): ) -async def _run_telephony_bot(transport_type: str, websocket: WebSocket, call_info): +async def _run_telephony_bot(websocket: WebSocket): """Run a bot for telephony transports.""" bot_module = _get_bot_module() - session_args = WebSocketSessionArguments( - websocket=websocket, - session_id=None, - ) - # Add transport-specific attributes - session_args.call_info = call_info - session_args.transport_type = transport_type + # Just pass the WebSocket - let the bot handle parsing + session_args = WebSocketSessionArguments(websocket=websocket, session_id=None) await bot_module.bot(session_args) @@ -257,14 +248,45 @@ def _create_server_app(transport_type: str, host: str = "localhost", proxy: str app.router.lifespan_context = lifespan elif transport_type in ["twilio", "telnyx", "plivo"]: - # Create a wrapper function for telephony - async def telephony_runner(transport_type_inner: str, **kwargs): - if "websocket" in kwargs and "call_info" in kwargs: - await _run_telephony_bot( - transport_type_inner, kwargs["websocket"], kwargs["call_info"] - ) + # Direct telephony WebSocket handling (no utils dependency) - setup_websocket_routes(app, telephony_runner, transport_type, proxy) + @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""" + + + + + +""" + elif transport_type == "telnyx": + xml_content = f""" + + + + + +""" + elif transport_type == "plivo": + xml_content = f""" + + wss://{proxy}/ws +""" + else: + xml_content = "" + + 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") + await _run_telephony_bot(websocket) # Add general routes @app.get("/") diff --git a/src/pipecat/runner/local.py b/src/pipecat/runner/local.py index a143ee2a6..402150927 100644 --- a/src/pipecat/runner/local.py +++ b/src/pipecat/runner/local.py @@ -40,6 +40,7 @@ Then run: `python bot.py -t webrtc` import argparse import asyncio +import json import os import sys from contextlib import asynccontextmanager @@ -48,15 +49,112 @@ from typing import Callable, Dict, Mapping, Optional import aiohttp import uvicorn from dotenv import load_dotenv -from fastapi import BackgroundTasks, FastAPI -from fastapi.responses import RedirectResponse +from fastapi import BackgroundTasks, FastAPI, WebSocket +from fastapi.middleware.cors import CORSMiddleware +from fastapi.responses import HTMLResponse, RedirectResponse from loguru import logger -from pipecat.runner.utils import setup_websocket_routes - load_dotenv(override=True) +def setup_websocket_routes( + app: FastAPI, transport_runner: Callable, transport_type: str, proxy: str = None +): + """Set up WebSocket routes for telephony providers. + + This is used by the local runner (pipecat.runner.local) only. + """ + 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""" + + + + + +""" + elif transport_type == "telnyx": + xml_content = f""" + + + + + +""" + elif transport_type == "plivo": + xml_content = f""" + + wss://{proxy}/ws +""" + else: + xml_content = "" + + 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) + + def _run_webrtc( run: Callable, args: argparse.Namespace, transport_params: Mapping[str, Callable] = {} ): diff --git a/src/pipecat/runner/utils.py b/src/pipecat/runner/utils.py index bc3e0a9f5..0617b1267 100644 --- a/src/pipecat/runner/utils.py +++ b/src/pipecat/runner/utils.py @@ -34,15 +34,130 @@ 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 fastapi.responses import RedirectResponse from loguru import logger from pipecat.transports.base_transport import BaseTransport +def detect_transport_type_from_message(message_data: dict) -> str: + """Attempt to auto-detect transport type from WebSocket message structure.""" + logger.debug("=== Auto-Detection Analysis ===") + + # Twilio detection + if ( + message_data.get("event") == "start" + and "start" in message_data + and "streamSid" in message_data.get("start", {}) + and "callSid" in message_data.get("start", {}) + ): + logger.debug("Auto-detected: TWILIO") + return "twilio" + + # Telnyx detection + if ( + "stream_id" in message_data + and "start" in message_data + and "call_control_id" in message_data.get("start", {}) + ): + logger.debug("Auto-detected: TELNYX") + return "telnyx" + + # Plivo detection + if ( + "start" in message_data + and "streamId" in message_data.get("start", {}) + and "callId" in message_data.get("start", {}) + ): + logger.debug("Auto-detected: PLIVO") + return "plivo" + + logger.debug("Auto-detection failed - unknown format") + return "unknown" + + +async def parse_telephony_websocket(websocket: WebSocket): + """Parse telephony WebSocket messages and return transport type and basic call data. + + Returns: + tuple: (transport_type: str, stream_id: str, call_id: str) + """ + logger.info("=== Parsing Telephony WebSocket ===") + + # Read first two messages + start_data = websocket.iter_text() + + try: + # First message + first_message_raw = await start_data.__anext__() + logger.debug(f"First message: {first_message_raw}") + try: + first_message = json.loads(first_message_raw) + except json.JSONDecodeError: + first_message = {} + + # Second message + second_message_raw = await start_data.__anext__() + logger.debug(f"Second message: {second_message_raw}") + try: + second_message = json.loads(second_message_raw) + except json.JSONDecodeError: + second_message = {} + + # Try auto-detection on both messages + detected_type_first = detect_transport_type_from_message(first_message) + detected_type_second = detect_transport_type_from_message(second_message) + + # Use the successful detection + if detected_type_first != "unknown": + transport_type = detected_type_first + call_data = first_message + logger.info(f"Detected transport: {transport_type} (from first message)") + elif detected_type_second != "unknown": + transport_type = detected_type_second + call_data = second_message + logger.info(f"Detected transport: {transport_type} (from second message)") + else: + transport_type = "unknown" + call_data = second_message + logger.warning("Could not auto-detect transport type") + + # Extract just the essential fields + if transport_type == "twilio": + start_data = call_data.get("start", {}) + stream_id = start_data.get("streamSid") + call_id = start_data.get("callSid") + + elif transport_type == "telnyx": + stream_id = call_data.get("stream_id") + call_id = call_data.get("start", {}).get("call_control_id") + + elif transport_type == "plivo": + start_data = call_data.get("start", {}) + stream_id = start_data.get("streamId") + call_id = start_data.get("callId") + + else: + stream_id = None + call_id = None + + logger.info(f"Parsed - Type: {transport_type}, StreamId: {stream_id}, CallId: {call_id}") + return transport_type, stream_id, call_id + + except Exception as e: + logger.error(f"Error parsing telephony WebSocket: {e}") + raise + + def get_install_command(transport: str) -> str: - """Get the pip install command for a specific transport.""" + """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]", @@ -241,98 +356,3 @@ def setup_webrtc_routes(app: FastAPI, transport_runner: Callable, host: str = No 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""" - - - - - -""" - elif transport_type == "telnyx": - xml_content = f""" - - - - - -""" - elif transport_type == "plivo": - xml_content = f""" - - wss://{proxy}/ws -""" - else: - xml_content = "" - - 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)