diff --git a/src/pipecat/runner/__init__.py b/src/pipecat/runner/__init__.py
new file mode 100644
index 000000000..e69de29bb
diff --git a/src/pipecat/runner/daily_runner.py b/src/pipecat/runner/daily_runner.py
new file mode 100644
index 000000000..d8d6acd81
--- /dev/null
+++ b/src/pipecat/runner/daily_runner.py
@@ -0,0 +1,89 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""Daily.co room configuration utilities for Pipecat examples."""
+
+import argparse
+import os
+from typing import Optional
+
+import aiohttp
+
+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.
+
+    Args:
+        aiohttp_session: HTTP session for making API requests.
+
+    Returns:
+        Tuple containing the room URL and authentication token.
+
+    Raises:
+        Exception: If room URL or API key are not provided.
+    """
+    (url, token, _) = await configure_with_args(aiohttp_session)
+    return (url, token)
+
+
+async def configure_with_args(
+    aiohttp_session: aiohttp.ClientSession, parser: Optional[argparse.ArgumentParser] = None
+):
+    """Configure Daily.co room with command-line argument parsing.
+
+    Args:
+        aiohttp_session: HTTP session for making API requests.
+        parser: Optional argument parser. If None, creates a default one.
+
+    Returns:
+        Tuple containing room URL, authentication token, and parsed arguments.
+
+    Raises:
+        Exception: If room URL or API key are not provided via arguments or environment.
+    """
+    if not parser:
+        parser = argparse.ArgumentParser(description="Daily AI SDK Bot Sample")
+    parser.add_argument(
+        "-u", "--url", type=str, required=False, help="URL of the Daily room to join"
+    )
+    parser.add_argument(
+        "-k",
+        "--apikey",
+        type=str,
+        required=False,
+        help="Daily API Key (needed to create an owner token for the room)",
+    )
+
+    args, unknown = parser.parse_known_args()
+
+    url = args.url or os.getenv("DAILY_SAMPLE_ROOM_URL")
+    key = args.apikey or os.getenv("DAILY_API_KEY")
+
+    if not url:
+        raise Exception(
+            "No Daily room specified. use the -u/--url option from the command line, or set DAILY_SAMPLE_ROOM_URL in your environment to specify a Daily room URL."
+        )
+
+    if not key:
+        raise Exception(
+            "No Daily API key specified. use the -k/--apikey option from the command line, or set DAILY_API_KEY in your environment to specify a Daily API key, available from https://dashboard.daily.co/developers."
+        )
+
+    daily_rest_helper = DailyRESTHelper(
+        daily_api_key=key,
+        daily_api_url=os.getenv("DAILY_API_URL", "https://api.daily.co/v1"),
+        aiohttp_session=aiohttp_session,
+    )
+
+    # Create a meeting token for the given room with an expiration 1 hour in
+    # the future.
+    expiry_time: float = 60 * 60
+
+    token = await daily_rest_helper.get_token(url, expiry_time)
+
+    return (url, token, args)
diff --git a/src/pipecat/runner/livekit_runner.py b/src/pipecat/runner/livekit_runner.py
new file mode 100644
index 000000000..344f65d52
--- /dev/null
+++ b/src/pipecat/runner/livekit_runner.py
@@ -0,0 +1,126 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+"""LiveKit room configuration utilities for Pipecat applications."""
+
+import argparse
+import os
+from typing import Optional
+
+from livekit import api
+from loguru import logger
+
+
+def generate_token(room_name: str, participant_name: str, api_key: str, api_secret: str) -> str:
+    """Generate a LiveKit access token for a participant.
+
+    Args:
+        room_name: Name of the LiveKit room.
+        participant_name: Name of the participant.
+        api_key: LiveKit API key.
+        api_secret: LiveKit API secret.
+
+    Returns:
+        JWT token string for room access.
+    """
+    token = api.AccessToken(api_key, api_secret)
+    token.with_identity(participant_name).with_name(participant_name).with_grants(
+        api.VideoGrants(
+            room_join=True,
+            room=room_name,
+        )
+    )
+
+    return token.to_jwt()
+
+
+def generate_token_with_agent(
+    room_name: str, participant_name: str, api_key: str, api_secret: str
+) -> str:
+    """Generate a LiveKit access token for an agent participant.
+
+    Args:
+        room_name: Name of the LiveKit room.
+        participant_name: Name of the participant.
+        api_key: LiveKit API key.
+        api_secret: LiveKit API secret.
+
+    Returns:
+        JWT token string for agent room access.
+    """
+    token = api.AccessToken(api_key, api_secret)
+    token.with_identity(participant_name).with_name(participant_name).with_grants(
+        api.VideoGrants(
+            room_join=True,
+            room=room_name,
+            agent=True,  # This makes LiveKit client know agent has joined
+        )
+    )
+
+    return token.to_jwt()
+
+
+async def configure():
+    """Configure LiveKit room URL and token from arguments or environment.
+
+    Returns:
+        Tuple containing the server URL, authentication token, and room name.
+
+    Raises:
+        Exception: If required LiveKit configuration is not provided.
+    """
+    (url, token, room_name, _) = await configure_with_args()
+    return (url, token, room_name)
+
+
+async def configure_with_args(parser: Optional[argparse.ArgumentParser] = None):
+    """Configure LiveKit room with command-line argument parsing.
+
+    Args:
+        parser: Optional argument parser. If None, creates a default one.
+
+    Returns:
+        Tuple containing server URL, authentication token, room name, and parsed arguments.
+
+    Raises:
+        Exception: If required LiveKit configuration is not provided via arguments or environment.
+    """
+    if not parser:
+        parser = argparse.ArgumentParser(description="LiveKit AI SDK Bot Sample")
+    parser.add_argument(
+        "-r", "--room", type=str, required=False, help="Name of the LiveKit room to join"
+    )
+    parser.add_argument("-u", "--url", type=str, required=False, help="URL of the LiveKit server")
+
+    args, unknown = parser.parse_known_args()
+
+    room_name = args.room or os.getenv("LIVEKIT_ROOM_NAME")
+    url = args.url or os.getenv("LIVEKIT_URL")
+    api_key = os.getenv("LIVEKIT_API_KEY")
+    api_secret = os.getenv("LIVEKIT_API_SECRET")
+
+    if not room_name:
+        raise Exception(
+            "No LiveKit room specified. Use the -r/--room option from the command line, or set LIVEKIT_ROOM_NAME in your environment."
+        )
+
+    if not url:
+        raise Exception(
+            "No LiveKit server URL specified. Use the -u/--url option from the command line, or set LIVEKIT_URL in your environment."
+        )
+
+    if not api_key or not api_secret:
+        raise Exception(
+            "LIVEKIT_API_KEY and LIVEKIT_API_SECRET must be set in environment variables."
+        )
+
+    token = generate_token_with_agent(room_name, "Pipecat Agent", api_key, api_secret)
+
+    # Generate user token for testing/debugging
+    user_token = generate_token(room_name, "User", api_key, api_secret)
+    logger.info(f"User token: {user_token}")
+
+    return (url, token, room_name, args)
diff --git a/src/pipecat/runner/run.py b/src/pipecat/runner/run.py
new file mode 100644
index 000000000..c82211d57
--- /dev/null
+++ b/src/pipecat/runner/run.py
@@ -0,0 +1,649 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# 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
+"""
+
+import argparse
+import asyncio
+import json
+import os
+import re
+import sys
+from contextlib import asynccontextmanager
+from typing import Any, 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 loguru import logger
+
+from pipecat.serializers.plivo import PlivoFrameSerializer
+from pipecat.serializers.telnyx import TelnyxFrameSerializer
+from pipecat.serializers.twilio import TwilioFrameSerializer
+from pipecat.transports.base_transport import BaseTransport, TransportParams
+from pipecat.transports.network.fastapi_websocket import (
+    FastAPIWebsocketParams,
+    FastAPIWebsocketTransport,
+)
+from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
+from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
+from pipecat.transports.services.daily import DailyParams, DailyTransport
+from pipecat.transports.services.livekit import LiveKitParams, LiveKitTransport
+
+# Load environment variables
+load_dotenv(override=True)
+
+
+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.
+    """
+    if isinstance(transport, SmallWebRTCTransport):
+        return client.pc_id
+    elif isinstance(transport, DailyTransport):
+        return client["id"]
+    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).
+    """
+    if isinstance(transport, DailyTransport):
+        await transport.capture_participant_video(
+            client["id"], framerate=framerate, video_source="camera"
+        )
+
+
+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).
+    """
+    if isinstance(transport, DailyTransport):
+        await transport.capture_participant_video(
+            client["id"], framerate=framerate, video_source="screenVideo"
+        )
+
+
+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.
+    """
+    logger.info("Running with DailyTransport...")
+
+    from pipecat.runner.daily_runner import configure
+
+    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.
+    """
+    logger.info("Running with LiveKitTransport...")
+
+    from pipecat.runner.livekit_runner import configure
+
+    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 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.
+    """
+    logger.info("Running with SmallWebRTCTransport...")
+
+    from pipecat_ai_small_webrtc_prebuilt.frontend import SmallWebRTCPrebuiltUI
+
+    app = FastAPI()
+
+    # 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.
+
+        Args:
+            request: WebRTC offer request containing SDP and connection details.
+            background_tasks: FastAPI background tasks for running applications.
+
+        Returns:
+            WebRTC answer with connection details.
+        """
+        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.
+
+                Args:
+                    webrtc_connection: The closed WebRTC connection.
+                """
+                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]()
+            transport = SmallWebRTCTransport(params=params, webrtc_connection=pipecat_connection)
+            background_tasks.add_task(run, transport, args, False)
+
+        answer = pipecat_connection.get_answer()
+
+        if args.esp32 and args.host:
+            answer["sdp"] = smallwebrtc_sdp_munging(answer["sdp"], args.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.
+
+        Args:
+            app: The FastAPI application instance.
+
+        Yields:
+            Control to the FastAPI application runtime.
+        """
+        yield  # Run app
+        coros = [pc.disconnect() for pc in pcs_map.values()]
+        await asyncio.gather(*coros)
+        pcs_map.clear()
+
+    uvicorn.run(app, host=args.host, port=args.port)
+
+
+def run_twilio(
+    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.
+    """
+    logger.info("Running with FastAPIWebsocketTransport (Twilio)...")
+
+    app = FastAPI()
+
+    app.add_middleware(
+        CORSMiddleware,
+        allow_origins=["*"],  # Allow all origins for testing
+        allow_credentials=True,
+        allow_methods=["*"],
+        allow_headers=["*"],
+    )
+
+    @app.post("/")
+    async def start_call():
+        """Handle Twilio webhook and return TwiML response.
+
+        Returns:
+            TwiML XML response directing call to WebSocket stream.
+        """
+        logger.debug("POST TwiML")
+
+        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")
+
+    @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)
+
+    uvicorn.run(app, host=args.host, port=args.port)
+
+
+def run_telnyx(
+    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.
+    """
+    logger.info("Running with FastAPIWebsocketTransport (Telnyx)...")
+
+    app = FastAPI()
+
+    app.add_middleware(
+        CORSMiddleware,
+        allow_origins=["*"],  # Allow all origins for testing
+        allow_credentials=True,
+        allow_methods=["*"],
+        allow_headers=["*"],
+    )
+
+    @app.post("/")
+    async def start_call():
+        """Handle Telnyx webhook and return TeXML response.
+
+        Returns:
+            TeXML XML response directing call to WebSocket stream.
+        """
+        logger.debug("POST TeXML")
+
+        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")
+
+    @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)
+
+    uvicorn.run(app, host=args.host, port=args.port)
+
+
+def run_plivo(
+    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.
+    """
+    logger.info("Running with FastAPIWebsocketTransport (Plivo)...")
+
+    app = FastAPI()
+
+    app.add_middleware(
+        CORSMiddleware,
+        allow_origins=["*"],  # Allow all origins for testing
+        allow_credentials=True,
+        allow_methods=["*"],
+        allow_headers=["*"],
+    )
+
+    @app.post("/")
+    async def start_call():
+        """Handle Plivo webhook and return Plivo XML response.
+
+        Returns:
+            Plivo XML response directing call to WebSocket stream.
+        """
+        logger.debug("POST Plivo XML")
+
+        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")
+
+    @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)
+
+    uvicorn.run(app, host=args.host, port=args.port)
+
+
+def run_main(
+    run: Callable,
+    args: argparse.Namespace,
+    transport_params: Mapping[str, Callable] = {},
+):
+    """Run the application with the specified transport type.
+
+    Args:
+        run: The function to run.
+        args: Parsed command-line arguments.
+        transport_params: Mapping of transport names to parameter factory functions.
+    """
+    if args.transport not in transport_params:
+        logger.error(f"Transport '{args.transport}' not supported by this application.")
+        return
+
+    match args.transport:
+        case "daily":
+            run_daily(run, args, transport_params)
+        case "livekit":
+            run_livekit(run, args, transport_params)
+        case "plivo":
+            run_plivo(run, args, transport_params)
+        case "telnyx":
+            run_telnyx(run, args, transport_params)
+        case "twilio":
+            run_twilio(run, args, transport_params)
+        case "webrtc":
+            run_webrtc(run, args, transport_params)
+
+
+def main(
+    run: Callable,
+    *,
+    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.
+    """
+    if not parser:
+        parser = argparse.ArgumentParser(description="Pipecat Bot Runner")
+    parser.add_argument(
+        "--host", default="localhost", help="Host for HTTP server (default: localhost)"
+    )
+    parser.add_argument(
+        "--port", type=int, default=7860, help="Port for HTTP server (default: 7860)"
+    )
+    parser.add_argument(
+        "--transport",
+        "-t",
+        type=str,
+        choices=["daily", "livekit", "plivo", "telnyx", "twilio", "webrtc"],
+        default="webrtc",
+        help="The transport this application should use",
+    )
+    parser.add_argument(
+        "--proxy", "-x", help="A public proxy host name (no protocol, e.g. proxy.example.com)"
+    )
+    parser.add_argument(
+        "--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":
+        logger.error("For ESP32, you need to specify `--host IP` so we can do SDP munging.")
+        return
+
+    # Log level
+    logger.remove(0)
+    logger.add(sys.stderr, level="TRACE" if args.verbose else "DEBUG")
+
+    # Import the bot file
+    run_main(run, args, transport_params)