From 8e8e42717b14e6ef4c8bf1724003285b546e8d80 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Tue, 9 Sep 2025 21:29:48 -0400 Subject: [PATCH 1/6] Add to and from phone information to the development runner --- CHANGELOG.md | 22 +++++++++++++++ src/pipecat/runner/utils.py | 54 ++++++++++++++++++++++++++++++++++--- 2 files changed, 72 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c882fa7c5..9ec29ac20 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,28 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 Using the universal `LLMContext` and associated `LLMContextAggregatorPair` is a pre-requisite for using `LLMSwitcher` to switch between LLMs at runtime. +- Added `to` and `from` phone number information to the development runner for + Twilio, Telnyx, Plivo, and Exotel. Note: for Twilio and Plivo, `to` and `from` + phone number information must be manually provided. + + - Twilio: TwiML returned must contain: + ``` + + + + + ``` + - Plivo: Add `to` and `from` information as `extraHeaders` in the XML `Stream` + property. + + This phone information acts as an identifer for inbound calls, allowing you + to programmatically inject information to your bot file based on corresponding + user information. + +- Added the ability to retrieve custom data from the development runner for + Twilio, Telnyx, Plivo, and Exotel. This provides the ability to inject custom + data into the call for outbound calling cases. + - Added video streaming support to `LiveKitTransport`. - Added `OpenAIRealtimeLLMService` and `AzureRealtimeLLMService` which provide diff --git a/src/pipecat/runner/utils.py b/src/pipecat/runner/utils.py index 585bd02b2..cfa32c19d 100644 --- a/src/pipecat/runner/utils.py +++ b/src/pipecat/runner/utils.py @@ -99,14 +99,20 @@ async def parse_telephony_websocket(websocket: WebSocket): 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} - - Exotel: {"stream_id": str, "call_id": str, "account_sid": str} + - Twilio: {"stream_id": str, "call_id": str, "from": str, "to": str} + - Telnyx: {"stream_id": str, "call_control_id": str, "outbound_encoding": str, "from": str, "to": str} + - Plivo: {"stream_id": str, "call_id": str, "from": str, "to": str} + - Exotel: {"stream_id": str, "call_id": str, "account_sid": str, "from": str, "to": str} Example usage:: transport_type, call_data = await parse_telephony_websocket(websocket) + + # Access call information (works for all providers when available) + from_number = call_data.get("from", "") + to_number = call_data.get("to", "") + + # Access provider-specific fields if transport_type == "telnyx": outbound_encoding = call_data["outbound_encoding"] """ @@ -151,9 +157,13 @@ async def parse_telephony_websocket(websocket: WebSocket): # Extract provider-specific data if transport_type == "twilio": start_data = call_data_raw.get("start", {}) + custom_params = start_data.get("customParameters", {}) + call_data = { "stream_id": start_data.get("streamSid"), "call_id": start_data.get("callSid"), + "from": custom_params.get("from", ""), + "to": custom_params.get("to", ""), } elif transport_type == "telnyx": @@ -163,13 +173,47 @@ async def parse_telephony_websocket(websocket: WebSocket): "outbound_encoding": call_data_raw.get("start", {}) .get("media_format", {}) .get("encoding"), + "from": call_data_raw.get("start", {}).get("from", ""), + "to": call_data_raw.get("start", {}).get("to", ""), } elif transport_type == "plivo": start_data = call_data_raw.get("start", {}) + custom_params = start_data.get("customParameters", {}) + + # Extract from/to from extra_headers if available + from_number = "" + to_number = "" + + # First try customParameters (for query parameter approach) + if custom_params.get("from"): + from_number = custom_params.get("from", "") + if custom_params.get("to"): + to_number = custom_params.get("to", "") + + # If not found in customParameters, try extra_headers + if not from_number or not to_number: + # Check for extra_headers in both root level and start event + extra_headers = call_data_raw.get("extra_headers", "") or start_data.get( + "extra_headers", "" + ) + if extra_headers: + # Parse format: "{X-PH-from: 14129162450, X-PH-to: 17242775935}" + import re + + from_match = re.search(r"X-PH-from:\s*([^,\s}]+)", extra_headers) + to_match = re.search(r"X-PH-to:\s*([^,\s}]+)", extra_headers) + + if from_match and not from_number: + from_number = from_match.group(1).strip() + if to_match and not to_number: + to_number = to_match.group(1).strip() + call_data = { "stream_id": start_data.get("streamId"), "call_id": start_data.get("callId"), + "from": from_number, + "to": to_number, } elif transport_type == "exotel": @@ -178,6 +222,8 @@ async def parse_telephony_websocket(websocket: WebSocket): "stream_id": start_data.get("stream_sid"), "call_id": start_data.get("call_sid"), "account_sid": start_data.get("account_sid"), + "from": start_data.get("from", ""), + "to": start_data.get("to", ""), } else: From d5956764364cf8fc4933140f7b97cc23249ba113 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 11 Sep 2025 09:54:19 -0400 Subject: [PATCH 2/6] Add custom data handling for Twilio --- src/pipecat/runner/utils.py | 22 ++++++++++------------ 1 file changed, 10 insertions(+), 12 deletions(-) diff --git a/src/pipecat/runner/utils.py b/src/pipecat/runner/utils.py index cfa32c19d..3f3b190cb 100644 --- a/src/pipecat/runner/utils.py +++ b/src/pipecat/runner/utils.py @@ -99,7 +99,11 @@ async def parse_telephony_websocket(websocket: WebSocket): tuple: (transport_type: str, call_data: dict) call_data contains provider-specific fields: - - Twilio: {"stream_id": str, "call_id": str, "from": str, "to": str} + - Twilio: { + "stream_id": str, + "call_id": str, + "body": dict + } - Telnyx: {"stream_id": str, "call_control_id": str, "outbound_encoding": str, "from": str, "to": str} - Plivo: {"stream_id": str, "call_id": str, "from": str, "to": str} - Exotel: {"stream_id": str, "call_id": str, "account_sid": str, "from": str, "to": str} @@ -107,14 +111,8 @@ async def parse_telephony_websocket(websocket: WebSocket): Example usage:: transport_type, call_data = await parse_telephony_websocket(websocket) - - # Access call information (works for all providers when available) - from_number = call_data.get("from", "") - to_number = call_data.get("to", "") - - # Access provider-specific fields - if transport_type == "telnyx": - outbound_encoding = call_data["outbound_encoding"] + if transport_type == "twilio": + user_id = call_data["body"]["user_id"] """ # Read first two messages start_data = websocket.iter_text() @@ -157,13 +155,13 @@ async def parse_telephony_websocket(websocket: WebSocket): # Extract provider-specific data if transport_type == "twilio": start_data = call_data_raw.get("start", {}) - custom_params = start_data.get("customParameters", {}) + body_data = start_data.get("customParameters", {}) call_data = { "stream_id": start_data.get("streamSid"), "call_id": start_data.get("callSid"), - "from": custom_params.get("from", ""), - "to": custom_params.get("to", ""), + # All custom parameters + "body": body_data, } elif transport_type == "telnyx": From 99cfcb1d4e5bef3b0182c0cff99ce5f098f63765 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 11 Sep 2025 14:44:42 -0400 Subject: [PATCH 3/6] Parsed custom data from Plivo extraHeaders --- src/pipecat/runner/utils.py | 109 +++++++++++++++++++++++++----------- 1 file changed, 76 insertions(+), 33 deletions(-) diff --git a/src/pipecat/runner/utils.py b/src/pipecat/runner/utils.py index 3f3b190cb..a21f49441 100644 --- a/src/pipecat/runner/utils.py +++ b/src/pipecat/runner/utils.py @@ -45,6 +45,54 @@ from pipecat.runner.types import ( from pipecat.transports.base_transport import BaseTransport +def _parse_plivo_extra_headers(extra_headers_str: str) -> dict: + """Parse Plivo extra headers string into a dictionary. + + Args: + extra_headers_str: String in format "X-PH-key: value, X-PH-key2: value2" + + Returns: + Dictionary with parsed headers (X-PH- prefix removed) + """ + if not isinstance(extra_headers_str, str) or not extra_headers_str: + return {} + + # Plivo uses "X-PH-key: value" format + pattern = r"X-PH-(\w+):\s*([^,}]+)" + matches = re.findall(pattern, extra_headers_str) + + return {key: value.strip() for key, value in matches} + + +def _extract_plivo_custom_parameters(parsed_headers: dict) -> dict: + """Extract custom parameters from Plivo headers, excluding built-in ones. + + Args: + parsed_headers: Dictionary of all parsed headers + + Returns: + Dictionary containing only custom parameters + """ + # Built-in Plivo headers that should not be treated as custom parameters + builtin_headers = { + "from", + "to", + "ALegRequestUUID", + "ALegUUID", + "BillRate", + "CallStatus", + "Direction", + "Event", + "ParentAuthID", + "RequestUUID", + "RouteType", + "STIRAttestation", + "STIRVerification", + } + + return {key: value for key, value in parsed_headers.items() if key not in builtin_headers} + + def _detect_transport_type_from_message(message_data: dict) -> str: """Attempt to auto-detect transport type from WebSocket message structure.""" logger.trace("=== Auto-Detection Analysis ===") @@ -104,9 +152,27 @@ async def parse_telephony_websocket(websocket: WebSocket): "call_id": str, "body": dict } - - Telnyx: {"stream_id": str, "call_control_id": str, "outbound_encoding": str, "from": str, "to": str} - - Plivo: {"stream_id": str, "call_id": str, "from": str, "to": str} - - Exotel: {"stream_id": str, "call_id": str, "account_sid": str, "from": str, "to": str} + - Telnyx: { + "stream_id": str, + "call_control_id": str, + "outbound_encoding": str, + "from": str, + "to": str, + } + - Plivo: { + "stream_id": str, + "call_id": str, + "from": str, + "to": str, + "custom_parameters": dict, + } + - Exotel: { + "stream_id": str, + "call_id": str, + "account_sid": str, + "from": str, + "to": str, + } Example usage:: @@ -177,41 +243,18 @@ async def parse_telephony_websocket(websocket: WebSocket): elif transport_type == "plivo": start_data = call_data_raw.get("start", {}) - custom_params = start_data.get("customParameters", {}) + extra_headers_str = call_data_raw.get("extra_headers", "") - # Extract from/to from extra_headers if available - from_number = "" - to_number = "" - - # First try customParameters (for query parameter approach) - if custom_params.get("from"): - from_number = custom_params.get("from", "") - if custom_params.get("to"): - to_number = custom_params.get("to", "") - - # If not found in customParameters, try extra_headers - if not from_number or not to_number: - # Check for extra_headers in both root level and start event - extra_headers = call_data_raw.get("extra_headers", "") or start_data.get( - "extra_headers", "" - ) - if extra_headers: - # Parse format: "{X-PH-from: 14129162450, X-PH-to: 17242775935}" - import re - - from_match = re.search(r"X-PH-from:\s*([^,\s}]+)", extra_headers) - to_match = re.search(r"X-PH-to:\s*([^,\s}]+)", extra_headers) - - if from_match and not from_number: - from_number = from_match.group(1).strip() - if to_match and not to_number: - to_number = to_match.group(1).strip() + # Parse Plivo extra headers (format: "X-PH-key: value, X-PH-key2: value2") + parsed_headers = _parse_plivo_extra_headers(extra_headers_str) + custom_parameters = _extract_plivo_custom_parameters(parsed_headers) call_data = { "stream_id": start_data.get("streamId"), "call_id": start_data.get("callId"), - "from": from_number, - "to": to_number, + "from": parsed_headers.get("from"), + "to": parsed_headers.get("to"), + "custom_parameters": custom_parameters, } elif transport_type == "exotel": From e5ed0424e483e0445f9088f784b2b54a84d2b22a Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Fri, 12 Sep 2025 11:08:06 -0400 Subject: [PATCH 4/6] Remove to/from data from Plivo, as it will rely on body information --- src/pipecat/runner/utils.py | 61 ------------------------------------- 1 file changed, 61 deletions(-) diff --git a/src/pipecat/runner/utils.py b/src/pipecat/runner/utils.py index a21f49441..c505fdaa6 100644 --- a/src/pipecat/runner/utils.py +++ b/src/pipecat/runner/utils.py @@ -45,54 +45,6 @@ from pipecat.runner.types import ( from pipecat.transports.base_transport import BaseTransport -def _parse_plivo_extra_headers(extra_headers_str: str) -> dict: - """Parse Plivo extra headers string into a dictionary. - - Args: - extra_headers_str: String in format "X-PH-key: value, X-PH-key2: value2" - - Returns: - Dictionary with parsed headers (X-PH- prefix removed) - """ - if not isinstance(extra_headers_str, str) or not extra_headers_str: - return {} - - # Plivo uses "X-PH-key: value" format - pattern = r"X-PH-(\w+):\s*([^,}]+)" - matches = re.findall(pattern, extra_headers_str) - - return {key: value.strip() for key, value in matches} - - -def _extract_plivo_custom_parameters(parsed_headers: dict) -> dict: - """Extract custom parameters from Plivo headers, excluding built-in ones. - - Args: - parsed_headers: Dictionary of all parsed headers - - Returns: - Dictionary containing only custom parameters - """ - # Built-in Plivo headers that should not be treated as custom parameters - builtin_headers = { - "from", - "to", - "ALegRequestUUID", - "ALegUUID", - "BillRate", - "CallStatus", - "Direction", - "Event", - "ParentAuthID", - "RequestUUID", - "RouteType", - "STIRAttestation", - "STIRVerification", - } - - return {key: value for key, value in parsed_headers.items() if key not in builtin_headers} - - def _detect_transport_type_from_message(message_data: dict) -> str: """Attempt to auto-detect transport type from WebSocket message structure.""" logger.trace("=== Auto-Detection Analysis ===") @@ -162,9 +114,6 @@ async def parse_telephony_websocket(websocket: WebSocket): - Plivo: { "stream_id": str, "call_id": str, - "from": str, - "to": str, - "custom_parameters": dict, } - Exotel: { "stream_id": str, @@ -222,7 +171,6 @@ async def parse_telephony_websocket(websocket: WebSocket): if transport_type == "twilio": start_data = call_data_raw.get("start", {}) body_data = start_data.get("customParameters", {}) - call_data = { "stream_id": start_data.get("streamSid"), "call_id": start_data.get("callSid"), @@ -243,18 +191,9 @@ async def parse_telephony_websocket(websocket: WebSocket): elif transport_type == "plivo": start_data = call_data_raw.get("start", {}) - extra_headers_str = call_data_raw.get("extra_headers", "") - - # Parse Plivo extra headers (format: "X-PH-key: value, X-PH-key2: value2") - parsed_headers = _parse_plivo_extra_headers(extra_headers_str) - custom_parameters = _extract_plivo_custom_parameters(parsed_headers) - call_data = { "stream_id": start_data.get("streamId"), "call_id": start_data.get("callId"), - "from": parsed_headers.get("from"), - "to": parsed_headers.get("to"), - "custom_parameters": custom_parameters, } elif transport_type == "exotel": From 22633a63b0dbbac385b4d9165a276c9a2acfca24 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Fri, 12 Sep 2025 11:15:03 -0400 Subject: [PATCH 5/6] Update changelog --- CHANGELOG.md | 30 +++++++++++------------------- 1 file changed, 11 insertions(+), 19 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9ec29ac20..5a3f882e9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,27 +13,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 Using the universal `LLMContext` and associated `LLMContextAggregatorPair` is a pre-requisite for using `LLMSwitcher` to switch between LLMs at runtime. -- Added `to` and `from` phone number information to the development runner for - Twilio, Telnyx, Plivo, and Exotel. Note: for Twilio and Plivo, `to` and `from` - phone number information must be manually provided. +- Added new fields to the development runner's `parse_telephony_websocket` + method in support of providing dynamic data to a bot. - - Twilio: TwiML returned must contain: - ``` - - - - - ``` - - Plivo: Add `to` and `from` information as `extraHeaders` in the XML `Stream` - property. + - Twilio: Added a new `body` parameter, which parses the websocket message + for `customParameters`. Provide data via the `Parameter` nouns in your + TwiML to use this feature. + - Telnyx & Exotel: Both providers make the `to` and `from` phone numbers + available in the websocket messages. You can now access these numbers as + `call_data["to"]` and `call_data["from"]`. - This phone information acts as an identifer for inbound calls, allowing you - to programmatically inject information to your bot file based on corresponding - user information. - -- Added the ability to retrieve custom data from the development runner for - Twilio, Telnyx, Plivo, and Exotel. This provides the ability to inject custom - data into the call for outbound calling cases. + Note: Each telephony provider offers different features. Refer to the + corresponding example in `pipecat-examples` to see how to pass custom data + to your bot. - Added video streaming support to `LiveKitTransport`. From 2f496ac74fa9fb4b617fcb1cde114b90bf91a42d Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Fri, 12 Sep 2025 11:28:12 -0400 Subject: [PATCH 6/6] Add optional body parameter to WebsocketRunnerArguments --- CHANGELOG.md | 4 ++++ src/pipecat/runner/types.py | 2 ++ 2 files changed, 6 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5a3f882e9..3362a24b1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -27,6 +27,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 corresponding example in `pipecat-examples` to see how to pass custom data to your bot. +- Added `body` to the `WebsocketRunnerArguments` as an optional parameter. + Custom `body` information can be passed from the server into the bot file via + the `bot()` method using this new parameter. + - Added video streaming support to `LiveKitTransport`. - Added `OpenAIRealtimeLLMService` and `AzureRealtimeLLMService` which provide diff --git a/src/pipecat/runner/types.py b/src/pipecat/runner/types.py index 842765d25..89aecd84c 100644 --- a/src/pipecat/runner/types.py +++ b/src/pipecat/runner/types.py @@ -51,9 +51,11 @@ class WebSocketRunnerArguments(RunnerArguments): Parameters: websocket: WebSocket connection for audio streaming + body: Additional request data """ websocket: WebSocket + body: Optional[Any] = field(default_factory=dict) @dataclass