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

Added support for WhatsApp User-initiated Calls.

Browse Source
This commit is contained in:
Filipi Fuchter
2025-09-02 18:05:00 -03:00
parent 49c1f0bd08
commit 8fbd9b5af7
4 changed files with 719 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
CHANGELOG.md
View File

@@ -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.

5
src/pipecat/transports/whatsapp/__init__.py Normal file
View File

@@ -0,0 +1,5 @@
#
# Copyright (c) 2024–2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#

345
src/pipecat/transports/whatsapp/api.py Normal file
View File

@@ -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()

364
src/pipecat/transports/whatsapp/client.py Normal file
View File

@@ -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