Mimic Pipecat Cloud websocket handling in the cloud runner, add a websocket message parser called parse_telephony_websocket in utils.py, update examples to use the new functionality

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
+        transport = FastAPIWebsocketTransport(
-        params = FastAPIWebsocketParams(
+            websocket=session_args.websocket,
-            audio_in_enabled=True,
+            params=FastAPIWebsocketParams(
-            audio_out_enabled=True,
+                audio_in_enabled=True,
-            vad_analyzer=SileroVADAnalyzer(),
+                audio_out_enabled=True,
-            add_wav_header=False,
+                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)
 
 

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

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

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):
+    async def bot(session_args):
-        # Detect transport type from 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(
+    # Just pass the WebSocket - let the bot handle parsing
-        websocket=websocket,
+    session_args = WebSocketSessionArguments(websocket=websocket, session_id=None)
-        session_id=None,
-    )
-    # Add transport-specific attributes
-    session_args.call_info = call_info
-    session_args.transport_type = transport_type
 
     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
+        # Direct telephony WebSocket handling (no utils dependency)
-        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"]
-                )
 
-        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"""<?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")
+            await _run_telephony_bot(websocket)
 
     # Add general routes
     @app.get("/")

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 import BackgroundTasks, FastAPI, WebSocket
-from fastapi.responses import RedirectResponse
+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"""<?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)
+
+
 def _run_webrtc(
     run: Callable, args: argparse.Namespace, transport_params: Mapping[str, Callable] = {}
 ):

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 RedirectResponse
-from fastapi.responses import HTMLResponse, 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"""<?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)