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

Parse telephony data into call_data object, add support for Telnyx, update example

Browse Source
This commit is contained in:
Mark Backman
2025-07-29 21:17:42 -04:00
parent dba182b396
commit f133cf97c8
2 changed files with 58 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
examples/runner-examples/01-all-transport-bot.py
View File

@@ -149,7 +149,7 @@ async def bot(
# Use the utility to parse WebSocket data # Use the utility to parse WebSocket data
from pipecat.runner.utils import parse_telephony_websocket from pipecat.runner.utils import parse_telephony_websocket
transport_type, stream_id, call_id = await parse_telephony_websocket(session_args.websocket) transport_type, call_data = await parse_telephony_websocket(session_args.websocket)
logger.info(f"Auto-detected transport: {transport_type}") logger.info(f"Auto-detected transport: {transport_type}")
# Create transport based on detected type # Create transport based on detected type
@@ -157,8 +157,8 @@ async def bot(
from pipecat.serializers.twilio import TwilioFrameSerializer from pipecat.serializers.twilio import TwilioFrameSerializer
serializer = TwilioFrameSerializer( serializer = TwilioFrameSerializer(
stream_sid=stream_id, stream_sid=call_data["stream_id"],
call_sid=call_id, call_sid=call_data["call_id"],
account_sid=os.getenv("TWILIO_ACCOUNT_SID", ""), account_sid=os.getenv("TWILIO_ACCOUNT_SID", ""),
auth_token=os.getenv("TWILIO_AUTH_TOKEN", ""), auth_token=os.getenv("TWILIO_AUTH_TOKEN", ""),
) )
@@ -167,18 +167,18 @@ async def bot(
from pipecat.serializers.telnyx import TelnyxFrameSerializer from pipecat.serializers.telnyx import TelnyxFrameSerializer
serializer = TelnyxFrameSerializer( serializer = TelnyxFrameSerializer(
stream_id=stream_id, stream_id=call_data["stream_id"],
call_control_id=call_id, call_control_id=call_data["call_control_id"],
outbound_encoding="PCMU", # Set manually outbound_encoding=call_data["outbound_encoding"],
inbound_encoding="PCMU", inbound_encoding="PCMU", # Set manually
) )
elif transport_type == "plivo": elif transport_type == "plivo":
from pipecat.serializers.plivo import PlivoFrameSerializer from pipecat.serializers.plivo import PlivoFrameSerializer
serializer = PlivoFrameSerializer( serializer = PlivoFrameSerializer(
stream_id=stream_id, stream_id=call_data["stream_id"],
call_id=call_id, call_id=call_data["call_id"],
) )
else: else:

80
src/pipecat/runner/utils.py
View File

@@ -26,7 +26,7 @@ Example::
from pipecat.runner.utils import parse_telephony_websocket from pipecat.runner.utils import parse_telephony_websocket
async def telephony_websocket_handler(websocket: WebSocket): async def telephony_websocket_handler(websocket: WebSocket):
transport_type, stream_id, call_id = await parse_telephony_websocket(websocket) transport_type, call_data = await parse_telephony_websocket(websocket)
""" """
import json import json
@@ -77,14 +77,21 @@ def _detect_transport_type_from_message(message_data: dict) -> str:
async def parse_telephony_websocket(websocket: WebSocket): async def parse_telephony_websocket(websocket: WebSocket):
"""Parse telephony WebSocket messages and return transport type and basic call data. """Parse telephony WebSocket messages and return transport type and call data.
Returns: Returns:
tuple: (transport_type: str, stream_id: str, call_id: str) tuple: (transport_type: str, call_data: dict)
call_data contains provider-specific fields:
- Twilio: {"stream_id": str, "call_id": str}
- Telnyx: {"stream_id": str, "call_control_id": str, "outbound_encoding": str}
- Plivo: {"stream_id": str, "call_id": str}
Example usage:: Example usage::
transport_type, stream_id, call_id = await parse_telephony_websocket(websocket) transport_type, call_data = await parse_telephony_websocket(websocket)
if transport_type == "telnyx":
outbound_encoding = call_data["outbound_encoding"]
""" """
# Read first two messages # Read first two messages
start_data = websocket.iter_text() start_data = websocket.iter_text()
@@ -113,37 +120,46 @@ async def parse_telephony_websocket(websocket: WebSocket):
# Use the successful detection # Use the successful detection
if detected_type_first != "unknown": if detected_type_first != "unknown":
transport_type = detected_type_first transport_type = detected_type_first
call_data = first_message call_data_raw = first_message
logger.debug(f"Detected transport: {transport_type} (from first message)") logger.debug(f"Detected transport: {transport_type} (from first message)")
elif detected_type_second != "unknown": elif detected_type_second != "unknown":
transport_type = detected_type_second transport_type = detected_type_second
call_data = second_message call_data_raw = second_message
logger.debug(f"Detected transport: {transport_type} (from second message)") logger.debug(f"Detected transport: {transport_type} (from second message)")
else: else:
transport_type = "unknown" transport_type = "unknown"
call_data = second_message call_data_raw = second_message
logger.warning("Could not auto-detect transport type") logger.warning("Could not auto-detect transport type")
# Extract provider-specific data
if transport_type == "twilio": if transport_type == "twilio":
start_data = call_data.get("start", {}) start_data = call_data_raw.get("start", {})
stream_id = start_data.get("streamSid") call_data = {
call_id = start_data.get("callSid") "stream_id": start_data.get("streamSid"),
"call_id": start_data.get("callSid"),
}
elif transport_type == "telnyx": elif transport_type == "telnyx":
stream_id = call_data.get("stream_id") call_data = {
call_id = call_data.get("start", {}).get("call_control_id") "stream_id": call_data_raw.get("stream_id"),
"call_control_id": call_data_raw.get("start", {}).get("call_control_id"),
"outbound_encoding": call_data_raw.get("start", {})
.get("media_format", {})
.get("encoding"),
}
elif transport_type == "plivo": elif transport_type == "plivo":
start_data = call_data.get("start", {}) start_data = call_data_raw.get("start", {})
stream_id = start_data.get("streamId") call_data = {
call_id = start_data.get("callId") "stream_id": start_data.get("streamId"),
"call_id": start_data.get("callId"),
}
else: else:
stream_id = None call_data = {}
call_id = None
logger.debug(f"Parsed - Type: {transport_type}, StreamId: {stream_id}, CallId: {call_id}") logger.debug(f"Parsed - Type: {transport_type}, Data: {call_data}")
return transport_type, stream_id, call_id return transport_type, call_data
except Exception as e: except Exception as e:
logger.error(f"Error parsing telephony WebSocket: {e}") logger.error(f"Error parsing telephony WebSocket: {e}")
@@ -304,8 +320,7 @@ async def _create_telephony_transport(
websocket: WebSocket, websocket: WebSocket,
params: Optional[Any] = None, params: Optional[Any] = None,
transport_type: str = None, transport_type: str = None,
stream_id: str = None, call_data: dict = None,
call_id: str = None,
) -> BaseTransport: ) -> BaseTransport:
"""Create a telephony transport with pre-parsed WebSocket data. """Create a telephony transport with pre-parsed WebSocket data.
@@ -313,8 +328,7 @@ async def _create_telephony_transport(
websocket: FastAPI WebSocket connection from telephony provider websocket: FastAPI WebSocket connection from telephony provider
params: FastAPIWebsocketParams (required) params: FastAPIWebsocketParams (required)
transport_type: Pre-detected provider type ("twilio", "telnyx", "plivo") transport_type: Pre-detected provider type ("twilio", "telnyx", "plivo")
stream_id: Pre-parsed stream ID call_data: Pre-parsed call data dict with provider-specific fields
call_id: Pre-parsed call ID
Returns: Returns:
Configured FastAPIWebsocketTransport ready for telephony use. Configured FastAPIWebsocketTransport ready for telephony use.
@@ -336,22 +350,26 @@ async def _create_telephony_transport(
from pipecat.serializers.twilio import TwilioFrameSerializer from pipecat.serializers.twilio import TwilioFrameSerializer
params.serializer = TwilioFrameSerializer( params.serializer = TwilioFrameSerializer(
stream_sid=stream_id, stream_sid=call_data["stream_id"],
call_sid=call_id, call_sid=call_data["call_id"],
account_sid=os.getenv("TWILIO_ACCOUNT_SID", ""), account_sid=os.getenv("TWILIO_ACCOUNT_SID", ""),
auth_token=os.getenv("TWILIO_AUTH_TOKEN", ""), auth_token=os.getenv("TWILIO_AUTH_TOKEN", ""),
) )
elif transport_type == "telnyx": elif transport_type == "telnyx":
from pipecat.serializers.telnyx import TelnyxFrameSerializer from pipecat.serializers.telnyx import TelnyxFrameSerializer
# Note: This would need additional parsing for telnyx-specific fields params.serializer = TelnyxFrameSerializer(
raise NotImplementedError("Telnyx auto-detection not fully implemented yet") stream_id=call_data["stream_id"],
call_control_id=call_data["call_control_id"],
outbound_encoding=call_data["outbound_encoding"],
inbound_encoding="PCMU", # Standard default
)
elif transport_type == "plivo": elif transport_type == "plivo":
from pipecat.serializers.plivo import PlivoFrameSerializer from pipecat.serializers.plivo import PlivoFrameSerializer
params.serializer = PlivoFrameSerializer( params.serializer = PlivoFrameSerializer(
stream_id=stream_id, stream_id=call_data["stream_id"],
call_id=call_id, call_id=call_data["call_id"],
auth_id=os.getenv("PLIVO_AUTH_ID", ""), auth_id=os.getenv("PLIVO_AUTH_ID", ""),
auth_token=os.getenv("PLIVO_AUTH_TOKEN", ""), auth_token=os.getenv("PLIVO_AUTH_TOKEN", ""),
) )
@@ -460,12 +478,12 @@ async def create_transport(
elif isinstance(session_args, WebSocketSessionArguments): elif isinstance(session_args, WebSocketSessionArguments):
# Parse once to determine the provider and get data # Parse once to determine the provider and get data
transport_type, stream_id, call_id = await parse_telephony_websocket(session_args.websocket) transport_type, call_data = await parse_telephony_websocket(session_args.websocket)
params = _get_transport_params(transport_type, transport_params) params = _get_transport_params(transport_type, transport_params)
# Create telephony transport with pre-parsed data # Create telephony transport with pre-parsed data
return await _create_telephony_transport( return await _create_telephony_transport(
session_args.websocket, params, transport_type, stream_id, call_id session_args.websocket, params, transport_type, call_data
) )
else: else: