diff --git a/src/pipecat/runner/utils.py b/src/pipecat/runner/utils.py index 7a4a59775..8cc703b32 100644 --- a/src/pipecat/runner/utils.py +++ b/src/pipecat/runner/utils.py @@ -30,8 +30,9 @@ Example:: """ import json +import os import re -from typing import Any +from typing import Any, Callable, Dict, Optional from fastapi import WebSocket from loguru import logger @@ -273,3 +274,199 @@ def smallwebrtc_sdp_munging(sdp: str, host: str) -> str: sdp = _smallwebrtc_sdp_cleanup_fingerprints(sdp) sdp = _smallwebrtc_sdp_cleanup_ice_candidates(sdp, host) return sdp + + +def _get_transport_params(transport_key: str, transport_params: Dict[str, Callable]) -> Any: + """Get transport parameters from factory function. + + Args: + transport_key: The transport key to look up + transport_params: Dict mapping transport names to parameter factory functions + + Returns: + Transport parameters from the factory function + + Raises: + ValueError: If transport key is missing from transport_params + """ + if transport_key not in transport_params: + raise ValueError( + f"Missing transport params for '{transport_key}'. " + f"Please add '{transport_key}' key to your transport_params dict." + ) + + params = transport_params[transport_key]() + logger.debug(f"Using transport params for {transport_key}") + return params + + +async def _create_telephony_transport( + websocket: WebSocket, + params: Optional[Any] = None, + transport_type: str = None, + stream_id: str = None, + call_id: str = None, +) -> BaseTransport: + """Create a telephony transport with pre-parsed WebSocket data. + + Args: + websocket: FastAPI WebSocket connection from telephony provider + params: FastAPIWebsocketParams (required) + transport_type: Pre-detected provider type ("twilio", "telnyx", "plivo") + stream_id: Pre-parsed stream ID + call_id: Pre-parsed call ID + + Returns: + Configured FastAPIWebsocketTransport ready for telephony use. + """ + from pipecat.transports.network.fastapi_websocket import FastAPIWebsocketTransport + + if params is None: + raise ValueError( + "FastAPIWebsocketParams must be provided. " + "The serializer and add_wav_header will be set automatically." + ) + + # Always set add_wav_header to False for telephony + params.add_wav_header = False + + logger.info(f"Using pre-detected telephony provider: {transport_type}") + + if transport_type == "twilio": + from pipecat.serializers.twilio import TwilioFrameSerializer + + params.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 + + # Note: This would need additional parsing for telnyx-specific fields + raise NotImplementedError("Telnyx auto-detection not fully implemented yet") + elif transport_type == "plivo": + from pipecat.serializers.plivo import PlivoFrameSerializer + + params.serializer = PlivoFrameSerializer( + stream_id=stream_id, + call_id=call_id, + auth_id=os.getenv("PLIVO_AUTH_ID", ""), + auth_token=os.getenv("PLIVO_AUTH_TOKEN", ""), + ) + else: + raise ValueError( + f"Unsupported telephony provider: {transport_type}. " + f"Supported providers: twilio, telnyx, plivo" + ) + + return FastAPIWebsocketTransport(websocket=websocket, params=params) + + +async def create_transport( + session_args: Any, transport_params: Dict[str, Callable] +) -> BaseTransport: + """Create a transport from session arguments using factory functions. + + This function uses the clean transport_params factory pattern where users + define a dictionary mapping transport names to parameter factory functions. + + Args: + session_args: Session arguments from the runner. + transport_params: Dict mapping transport names to parameter factory functions. + Keys should be: "daily", "webrtc", "twilio", "telnyx", "plivo" + Values should be functions that return transport parameters when called. + + Returns: + Configured transport instance. + + Raises: + ValueError: If transport key is missing from transport_params or session args type is unsupported. + ImportError: If required dependencies are not installed. + + Example:: + + transport_params = { + "daily": lambda: DailyParams( + audio_in_enabled=True, + audio_out_enabled=True, + vad_analyzer=SileroVADAnalyzer(), + ), + "webrtc": lambda: TransportParams( + audio_in_enabled=True, + audio_out_enabled=True, + vad_analyzer=SileroVADAnalyzer(), + ), + "twilio": lambda: FastAPIWebsocketParams( + audio_in_enabled=True, + audio_out_enabled=True, + vad_analyzer=SileroVADAnalyzer(), + # add_wav_header and serializer will be set automatically + ), + "telnyx": lambda: FastAPIWebsocketParams( + audio_in_enabled=True, + audio_out_enabled=True, + vad_analyzer=SileroVADAnalyzer(), + # add_wav_header and serializer will be set automatically + ), + "plivo": lambda: FastAPIWebsocketParams( + audio_in_enabled=True, + audio_out_enabled=True, + vad_analyzer=SileroVADAnalyzer(), + # add_wav_header and serializer will be set automatically + ), + } + + transport = await create_transport(session_args, transport_params) + """ + # Import session types + try: + from pipecatcloud.agent import DailySessionArguments, WebSocketSessionArguments + except ImportError: + raise ImportError( + "pipecatcloud package required. Install with: pip install pipecat-ai[runner]" + ) + + try: + from pipecat.runner.run import SmallWebRTCSessionArguments + except ImportError: + raise ImportError( + "SmallWebRTCSessionArguments not found. Make sure you've installed pipecat-ai[runner]." + ) + + # Create transport based on session args type + if isinstance(session_args, DailySessionArguments): + params = _get_transport_params("daily", transport_params) + + from pipecat.transports.services.daily import DailyTransport + + return DailyTransport( + session_args.room_url, + session_args.token, + "Pipecat Bot", + params=params, + ) + + elif isinstance(session_args, SmallWebRTCSessionArguments): + params = _get_transport_params("webrtc", transport_params) + + from pipecat.transports.network.small_webrtc import SmallWebRTCTransport + + return SmallWebRTCTransport( + params=params, + webrtc_connection=session_args.webrtc_connection, + ) + + elif isinstance(session_args, WebSocketSessionArguments): + # Parse once to determine the provider and get data + transport_type, stream_id, call_id = await parse_telephony_websocket(session_args.websocket) + params = _get_transport_params(transport_type, transport_params) + + # Create telephony transport with pre-parsed data + return await _create_telephony_transport( + session_args.websocket, params, transport_type, stream_id, call_id + ) + + else: + raise ValueError(f"Unsupported session arguments type: {type(session_args)}")