From 8fbd9b5af7a573b4e51d6b7ce74709a0a4041610 Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Tue, 2 Sep 2025 18:05:00 -0300 Subject: [PATCH] Added support for WhatsApp User-initiated Calls. --- CHANGELOG.md | 5 + src/pipecat/transports/whatsapp/__init__.py | 5 + src/pipecat/transports/whatsapp/api.py | 345 +++++++++++++++++++ src/pipecat/transports/whatsapp/client.py | 364 ++++++++++++++++++++ 4 files changed, 719 insertions(+) create mode 100644 src/pipecat/transports/whatsapp/__init__.py create mode 100644 src/pipecat/transports/whatsapp/api.py create mode 100644 src/pipecat/transports/whatsapp/client.py diff --git a/CHANGELOG.md b/CHANGELOG.md index c8648fbb2..af3969560 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added support for WhatsApp User-initiated Calls. + - Added new audio filter `AICFilter`, speech enhancement for improving VAD/STT performance, no ONNX dependency. See https://ai-coustics.com/sdk/ @@ -85,6 +87,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Fixed `SmallWebRTCTransport` to not use `mid` to decide if the transceiver should + be `sendrecv` or not. + - Fixed an issue where Deepgram swallowed `asyncio.CancelledError` during disconnect, preventing tasks from being cancelled. diff --git a/src/pipecat/transports/whatsapp/__init__.py b/src/pipecat/transports/whatsapp/__init__.py new file mode 100644 index 000000000..d23112945 --- /dev/null +++ b/src/pipecat/transports/whatsapp/__init__.py @@ -0,0 +1,5 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# diff --git a/src/pipecat/transports/whatsapp/api.py b/src/pipecat/transports/whatsapp/api.py new file mode 100644 index 000000000..2fa47950d --- /dev/null +++ b/src/pipecat/transports/whatsapp/api.py @@ -0,0 +1,345 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""WhatsApp API. + +API to communicate with WhatsApp Cloud API. +""" + +from typing import Any, Dict, List, Optional, Union + +import aiohttp +from loguru import logger +from pydantic import BaseModel, Field + + +# ---------------------------- +# Pydantic Models for WhatsApp +# ---------------------------- +class WhatsAppSession(BaseModel): + """WebRTC session information for WhatsApp calls. + + Attributes: + sdp: Session Description Protocol (SDP) data for WebRTC connection + sdp_type: Type of SDP (e.g., "offer", "answer") + """ + + sdp: str + sdp_type: str + + +class WhatsAppError(BaseModel): + """Error information from WhatsApp API responses. + + Attributes: + code: Error code number + message: Human-readable error message + href: URL for more information about the error + error_data: Additional error-specific data + """ + + code: int + message: str + href: str + error_data: Dict[str, Any] + + +class WhatsAppConnectCall(BaseModel): + """Incoming call connection event data. + + Represents a user-initiated call that requires handling. This is sent + when a WhatsApp user initiates a call to your business number. + + Attributes: + id: Unique call identifier + from_: Phone number of the caller (WhatsApp ID format) + to: Your business phone number that received the call + event: Always "connect" for incoming calls + timestamp: ISO 8601 timestamp when the call was initiated + direction: Optional call direction ("inbound" for user-initiated calls) + session: WebRTC session data containing SDP offer from the caller + """ + + id: str + from_: str = Field(..., alias="from") + to: str + event: str # "connect" + timestamp: str + direction: Optional[str] + session: WhatsAppSession + + +class WhatsAppTerminateCall(BaseModel): + """Call termination event data. + + Represents the end of a call session, whether completed successfully, + failed, or was rejected by either party. + + Attributes: + id: Unique call identifier (matches the connect event) + from_: Phone number of the caller + to: Your business phone number + event: Always "terminate" for call end events + timestamp: ISO 8601 timestamp when the call ended + direction: Optional call direction + biz_opaque_callback_data: Optional business-specific callback data + status: Call completion status ("FAILED", "COMPLETED", "REJECTED") + start_time: ISO 8601 timestamp when call actually started (after acceptance) + end_time: ISO 8601 timestamp when call ended + duration: Call duration in seconds (only for completed calls) + """ + + id: str + from_: str = Field(..., alias="from") + to: str + event: str # "terminate" + timestamp: str + direction: Optional[str] + biz_opaque_callback_data: Optional[str] = None + status: Optional[str] = None # "FAILED" or "COMPLETED" or "REJECTED" + start_time: Optional[str] = None + end_time: Optional[str] = None + duration: Optional[int] = None + + +class WhatsAppProfile(BaseModel): + """User profile information. + + Attributes: + name: Display name of the WhatsApp user + """ + + name: str + + +class WhatsAppContact(BaseModel): + """Contact information for a WhatsApp user. + + Attributes: + profile: User's profile information + wa_id: WhatsApp ID (phone number in international format without +) + """ + + profile: WhatsAppProfile + wa_id: str + + +class WhatsAppMetadata(BaseModel): + """Business phone number metadata. + + Attributes: + display_phone_number: Formatted phone number for display + phone_number_id: WhatsApp Business API phone number ID + """ + + display_phone_number: str + phone_number_id: str + + +class WhatsAppConnectCallValue(BaseModel): + """Webhook payload for incoming call events. + + Attributes: + messaging_product: Always "whatsapp" + metadata: Business phone number information + contacts: List of contact information for involved parties + calls: List of call connection events + """ + + messaging_product: str + metadata: WhatsAppMetadata + contacts: List[WhatsAppContact] + calls: List[WhatsAppConnectCall] + + +class WhatsAppTerminateCallValue(BaseModel): + """Webhook payload for call termination events. + + Attributes: + messaging_product: Always "whatsapp" + metadata: Business phone number information + calls: List of call termination events + errors: Optional list of errors that occurred during the call + """ + + messaging_product: str + metadata: WhatsAppMetadata + calls: List[WhatsAppTerminateCall] + errors: Optional[List[WhatsAppError]] = None + + +class WhatsAppChange(BaseModel): + """Webhook change event wrapper. + + Attributes: + value: The actual event data (connect or terminate) + field: Always "calls" for calling webhooks + """ + + value: Union[WhatsAppConnectCallValue, WhatsAppTerminateCallValue] + field: str + + +class WhatsAppEntry(BaseModel): + """Webhook entry containing one or more changes. + + Attributes: + id: WhatsApp Business Account ID + changes: List of change events in this webhook delivery + """ + + id: str + changes: List[WhatsAppChange] + + +class WhatsAppWebhookRequest(BaseModel): + """Complete webhook request from WhatsApp. + + This is the top-level structure for all webhook deliveries from + the WhatsApp Cloud API for calling events. + + Attributes: + object: Always "whatsapp_business_account" + entry: List of webhook entries (usually contains one entry) + """ + + object: str + entry: List[WhatsAppEntry] + + +class WhatsAppApi: + """WhatsApp Cloud API client for handling calls. + + This class provides methods to interact with the WhatsApp Cloud API + for managing voice calls, including answering, rejecting, and terminating calls. + + Attributes: + BASE_URL: Base URL for WhatsApp Graph API v23.0 + phone_number_id: Your WhatsApp Business phone number ID + session: aiohttp client session for making HTTP requests + whatsapp_url: Complete URL for the calls endpoint + whatsapp_token: Bearer token for API authentication + """ + + BASE_URL = f"https://graph.facebook.com/v23.0/" + + def __init__( + self, whatsapp_token: str, phone_number_id: str, session: aiohttp.ClientSession + ) -> None: + """Initialize the WhatsApp API client. + + Args: + whatsapp_token: WhatsApp access token for authentication + phone_number_id: Your business phone number ID from WhatsApp Business API + session: aiohttp ClientSession for making HTTP requests + """ + self._phone_number_id = phone_number_id + self._session = session + self._whatsapp_url = f"{self.BASE_URL}{phone_number_id}/calls" + self._whatsapp_token = whatsapp_token + + async def answer_call_to_whatsapp(self, call_id: str, action: str, sdp: str, from_: str): + """Answer an incoming WhatsApp call. + + This method handles the call answering process, supporting both "pre_accept" + and "accept" actions as required by the WhatsApp calling workflow. + + Args: + call_id: Unique identifier for the call (from connect webhook) + action: Action to perform ("pre_accept" or "accept") + sdp: Session Description Protocol answer for WebRTC connection + from_: Caller's phone number (WhatsApp ID format) + + Returns: + Dict containing the API response with success status and any error details + + Note: + Calls must be pre-accepted before being accepted. The typical flow is: + 1. Receive connect webhook + 2. Call with action="pre_accept" + 3. Call with action="accept" + """ + logger.debug(f"Answering call {call_id} to WhatsApp, action:{action}") + async with self._session.post( + self._whatsapp_url, + headers={ + "Authorization": f"Bearer {self._whatsapp_token}", + "Content-Type": "application/json", + }, + json={ + "messaging_product": "whatsapp", + "to": from_, + "action": action, + "call_id": call_id, + "session": {"sdp": sdp, "sdp_type": "answer"}, + }, + ) as response: + return await response.json() + + async def reject_call_to_whatsapp(self, call_id: str): + """Reject an incoming WhatsApp call. + + This method rejects a call that was received via connect webhook. + The caller will receive a rejection notification and a terminate + webhook will be sent with status "REJECTED". + + Args: + call_id: Unique identifier for the call (from connect webhook) + + Returns: + Dict containing the API response with success status and any error details + + Note: + This should be called instead of answer_call_to_whatsapp when you want + to decline the incoming call. The caller will see the call as rejected. + """ + logger.debug(f"Rejecting call {call_id}") + async with self._session.post( + self._whatsapp_url, + headers={ + "Authorization": f"Bearer {self._whatsapp_token}", + "Content-Type": "application/json", + }, + json={ + "messaging_product": "whatsapp", + "action": "reject", + "call_id": call_id, + }, + ) as response: + return await response.json() + + async def terminate_call_to_whatsapp(self, call_id: str): + """Terminate an active WhatsApp call. + + This method ends an ongoing call that has been previously accepted. + Both parties will be disconnected and a terminate webhook will be + sent with status "COMPLETED". + + Args: + call_id: Unique identifier for the active call + + Returns: + Dict containing the API response with success status and any error details + + Note: + This should only be called for calls that have been accepted and are + currently active. For incoming calls that haven't been accepted yet, + use reject_call_to_whatsapp instead. + """ + logger.debug(f"Terminating call {call_id}") + async with self._session.post( + self._whatsapp_url, + headers={ + "Authorization": f"Bearer {self._whatsapp_token}", + "Content-Type": "application/json", + }, + json={ + "messaging_product": "whatsapp", + "action": "terminate", + "call_id": call_id, + }, + ) as response: + return await response.json() diff --git a/src/pipecat/transports/whatsapp/client.py b/src/pipecat/transports/whatsapp/client.py new file mode 100644 index 000000000..25035c651 --- /dev/null +++ b/src/pipecat/transports/whatsapp/client.py @@ -0,0 +1,364 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""WhatsApp API Client. + +This module provides a client for communicating with the WhatsApp Cloud API, +handling webhook requests, managing WebRTC connections, and processing +WhatsApp call events. +""" + +import asyncio +from typing import Awaitable, Callable, Dict, List, Optional, Union + +import aiohttp +from loguru import logger + +from pipecat.transports.network.webrtc_connection import IceServer, SmallWebRTCConnection +from pipecat.transports.whatsapp.api import ( + WhatsAppApi, + WhatsAppConnectCall, + WhatsAppConnectCallValue, + WhatsAppTerminateCall, + WhatsAppTerminateCallValue, + WhatsAppWebhookRequest, +) + + +class WhatsAppClient: + """WhatsApp Cloud API client for handling calls and webhook requests. + + This client manages WhatsApp call connections using WebRTC, processes webhook + events from WhatsApp, and maintains ongoing call state. It supports both + incoming call handling and call termination through the WhatsApp Cloud API. + + Attributes: + _whatsapp_api: WhatsApp API instance for making API calls + _ongoing_calls_map: Dictionary mapping call IDs to WebRTC connections + _ice_servers: List of ICE servers for WebRTC connections + """ + + def __init__( + self, + whatsapp_token: str, + phone_number_id: str, + session: aiohttp.ClientSession, + ice_servers: Optional[List[IceServer]] = None, + ) -> None: + """Initialize the WhatsApp client. + + Args: + whatsapp_token: WhatsApp API access token + phone_number_id: WhatsApp phone number ID for the business account + session: aiohttp session for making HTTP requests + ice_servers: List of ICE servers for WebRTC connections. If None, + defaults to Google's public STUN server + """ + self._whatsapp_api = WhatsAppApi( + whatsapp_token=whatsapp_token, phone_number_id=phone_number_id, session=session + ) + self._ongoing_calls_map: Dict[str, SmallWebRTCConnection] = {} + + # Set default ICE servers if none provided + if ice_servers is None: + self._ice_servers = [IceServer(urls="stun:stun.l.google.com:19302")] + else: + self._ice_servers = ice_servers + + async def terminate_all_calls(self) -> None: + """Terminate all ongoing WhatsApp calls. + + This method will: + 1. Send termination requests to WhatsApp API for each ongoing call + 2. Disconnect all WebRTC connections + 3. Clear the ongoing calls map + + All terminations are executed concurrently for efficiency. + """ + logger.debug("Will terminate all ongoing WhatsApp calls") + + if not self._ongoing_calls_map: + logger.debug("No ongoing calls to terminate") + return + + logger.debug(f"Terminating {len(self._ongoing_calls_map)} ongoing calls") + + # Terminate each call via WhatsApp API + termination_tasks = [] + for call_id, pipecat_connection in self._ongoing_calls_map.items(): + logger.debug(f"Terminating call {call_id}") + # Call WhatsApp API to terminate the call + if self._whatsapp_api: + termination_tasks.append(self._whatsapp_api.terminate_call_to_whatsapp(call_id)) + # Disconnect the pipecat connection + termination_tasks.append(pipecat_connection.disconnect()) + + # Execute all terminations concurrently + await asyncio.gather(*termination_tasks, return_exceptions=True) + + # Clear the ongoing calls map + self._ongoing_calls_map.clear() + logger.debug("All calls terminated successfully") + + async def handle_verify_webhook_request( + self, params: Dict[str, str], expected_verification_token: str + ) -> int: + """Handle a verify webhook request from WhatsApp. + + Args: + params: Dictionary containing webhook parameters from query string + expected_verification_token: The expected verification token to validate against + + Returns: + int: The challenge value if verification succeeds + + Raises: + ValueError: If verification fails due to missing parameters or invalid token + """ + mode = params.get("hub.mode") + challenge = params.get("hub.challenge") + verify_token = params.get("hub.verify_token") + + if not mode or not challenge or not verify_token: + raise ValueError("Missing required webhook verification parameters") + + if mode != "subscribe": + raise ValueError(f"Invalid hub mode: expected 'subscribe', got '{mode}'") + + if verify_token != expected_verification_token: + raise ValueError("Webhook verification token mismatch") + + return int(challenge) + + async def handle_webhook_request( + self, + request: WhatsAppWebhookRequest, + connection_callback: Optional[Callable[[SmallWebRTCConnection], Awaitable[None]]] = None, + ) -> bool: + """Handle a webhook request from WhatsApp. + + This method processes incoming webhook requests and handles both + connect and terminate events. For connect events, it establishes + a WebRTC connection and optionally invokes a callback with the + new connection. + + Args: + request: The webhook request from WhatsApp containing call events + connection_callback: Optional callback function to invoke when a new + WebRTC connection is established. The callback + receives the SmallWebRTCConnection instance. + + Returns: + bool: True if the webhook request was handled successfully, False otherwise + + Raises: + ValueError: If the webhook request contains no supported events + Exception: If connection establishment or API calls fail + """ + try: + for entry in request.entry: + for change in entry.changes: + # Handle connect events + if isinstance(change.value, WhatsAppConnectCallValue): + for call in change.value.calls: + if call.event == "connect": + logger.debug(f"Processing connect event for call {call.id}") + try: + connection = await self._handle_connect_event(call) + + # Invoke callback if provided + if connection_callback and connection: + try: + await connection_callback(connection) + logger.debug( + f"Connection callback executed successfully for call {call.id}" + ) + except Exception as callback_error: + logger.error( + f"Connection callback failed for call {call.id}: {callback_error}" + ) + # Continue execution despite callback failure + + return True + except Exception as connect_error: + logger.error( + f"Failed to handle connect event for call {call.id}: {connect_error}" + ) + raise + + # Handle terminate events + elif isinstance(change.value, WhatsAppTerminateCallValue): + for call in change.value.calls: + if call.event == "terminate": + logger.debug(f"Processing terminate event for call {call.id}") + try: + return await self._handle_terminate_event(call) + except Exception as terminate_error: + logger.error( + f"Failed to handle terminate event for call {call.id}: {terminate_error}" + ) + raise + + # No supported events found + error_msg = "No supported event found in webhook request" + logger.warning(f"{error_msg}: {request}") + raise ValueError(error_msg) + + except Exception as e: + logger.error(f"Error processing webhook request: {e}") + logger.debug(f"Webhook request details: {request}") + raise + + def _filter_sdp_for_whatsapp(self, sdp: str) -> str: + """Filter SDP to be compatible with WhatsApp requirements. + + WhatsApp only supports SHA-256 fingerprints, so this method removes + other fingerprint types from the SDP. + + Args: + sdp: The original SDP string + + Returns: + Filtered SDP string compatible with WhatsApp + """ + lines = sdp.splitlines() + filtered = [] + for line in lines: + if line.startswith("a=fingerprint:") and not line.startswith("a=fingerprint:sha-256"): + continue # drop sha-384 / sha-512 + filtered.append(line) + return "\r\n".join(filtered) + "\r\n" + + async def _handle_connect_event(self, call: WhatsAppConnectCall) -> SmallWebRTCConnection: + """Handle a CONNECT event by establishing WebRTC connection and accepting the call. + + This method: + 1. Creates a new WebRTC connection using configured ICE servers + 2. Initializes the connection with the provided SDP + 3. Generates an SDP answer and filters it for WhatsApp compatibility + 4. Pre-accepts the call with WhatsApp API + 5. Accepts the call with WhatsApp API + 6. Stores the connection for later management + + Args: + call: WhatsApp connect call event + + Returns: + The established SmallWebRTCConnection instance + + Raises: + Exception: If pre-accept or accept API calls fail + """ + logger.debug(f"Incoming call from {call.from_}, call_id: {call.id}") + + pipecat_connection = None + try: + # Create and initialize WebRTC connection + pipecat_connection = SmallWebRTCConnection(self._ice_servers) + await pipecat_connection.initialize(sdp=call.session.sdp, type=call.session.sdp_type) + sdp_answer = pipecat_connection.get_answer().get("sdp") + sdp_answer = self._filter_sdp_for_whatsapp(sdp_answer) + + logger.debug(f"SDP answer generated for call {call.id}") + + # Pre-accept the call + try: + pre_accept_resp = await self._whatsapp_api.answer_call_to_whatsapp( + call.id, "pre_accept", sdp_answer, call.from_ + ) + if not pre_accept_resp.get("success", False): + logger.error(f"Failed to pre-accept call {call.id}: {pre_accept_resp}") + raise Exception(f"Failed to pre-accept call: {pre_accept_resp}") + + logger.debug(f"Pre-accept successful for call {call.id}") + except Exception as e: + logger.error(f"Pre-accept API call failed for call {call.id}: {e}") + raise Exception(f"Failed to pre-accept call: {e}") + + # Accept the call + try: + accept_resp = await self._whatsapp_api.answer_call_to_whatsapp( + call.id, "accept", sdp_answer, call.from_ + ) + if not accept_resp.get("success", False): + logger.error(f"Failed to accept call {call.id}: {accept_resp}") + raise Exception(f"Failed to accept call: {accept_resp}") + + logger.debug(f"Accept successful for call {call.id}") + except Exception as e: + logger.error(f"Accept API call failed for call {call.id}: {e}") + raise Exception(f"Failed to accept call: {e}") + + # Store the connection for management + self._ongoing_calls_map[call.id] = pipecat_connection + + # Set up disconnect handler + @pipecat_connection.event_handler("closed") + async def handle_disconnected(webrtc_connection: SmallWebRTCConnection): + logger.debug( + f"Peer connection closed: {webrtc_connection.pc_id} for call {call.id}" + ) + # Clean up from ongoing calls map + self._ongoing_calls_map.pop(call.id, None) + + logger.debug(f"WebRTC connection established successfully for call {call.id}") + return pipecat_connection + + except Exception as e: + # Clean up connection on failure + if pipecat_connection: + try: + await pipecat_connection.disconnect() + except Exception as cleanup_error: + logger.error( + f"Failed to cleanup connection for call {call.id}: {cleanup_error}" + ) + + logger.error(f"Failed to handle connect event for call {call.id}: {e}") + raise + + async def _handle_terminate_event(self, call: WhatsAppTerminateCall) -> bool: + """Handle a TERMINATE event by cleaning up resources and logging call completion. + + This method: + 1. Logs call termination details including duration if available + 2. Disconnects the associated WebRTC connection + 3. Removes the call from the ongoing calls map + + Args: + call: WhatsApp terminate call event + + Returns: + bool: True if the call was terminated successfully, False otherwise + """ + logger.debug(f"Call terminated from {call.from_}, call_id: {call.id}") + logger.debug(f"Call status: {call.status}") + if call.duration: + logger.debug(f"Call duration: {call.duration} seconds") + + try: + if call.id in self._ongoing_calls_map: + pipecat_connection = self._ongoing_calls_map[call.id] + logger.debug(f"Disconnecting WebRTC connection for call {call.id}") + + try: + await pipecat_connection.disconnect() + logger.debug(f"WebRTC connection disconnected successfully for call {call.id}") + except Exception as disconnect_error: + logger.error( + f"Failed to disconnect WebRTC connection for call {call.id}: {disconnect_error}" + ) + + # Remove from ongoing calls map + self._ongoing_calls_map.pop(call.id, None) + else: + logger.warning(f"Call {call.id} not found in ongoing calls map") + + return True + + except Exception as e: + logger.error(f"Error handling terminate event for call {call.id}: {e}") + return False