client

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.

class pipecat.transports.whatsapp.client.WhatsAppClient(whatsapp_token: str, phone_number_id: str, session: ClientSession, ice_servers: list[RTCIceServer] | None = None, whatsapp_secret: str | None = None)[source]

Bases: object

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.

Parameters:

  • _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

__init__(whatsapp_token: str, phone_number_id: str, session: ClientSession, ice_servers: list[RTCIceServer] | None = None, whatsapp_secret: str | None = None) → None[source]

Initialize the WhatsApp client.

Parameters:

  • 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

  • whatsapp_secret – WhatsApp APP secret for validating that the webhook request came from WhatsApp.

update_ice_servers(ice_servers: list[RTCIceServer] | None = None)[source]

Update the list of ICE servers used for WebRTC connections.

update_whatsapp_secret(whatsapp_secret: str | None = None)[source]

Update the WhatsApp APP secret for validating that the webhook request came from WhatsApp.

update_whatsapp_token(whatsapp_token: str)[source]

Update the WhatsApp API access token.

update_whatsapp_phone_number_id(phone_number_id: str)[source]

Update the WhatsApp phone number ID for authentication.

async terminate_all_calls() → None[source]

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.

async handle_verify_webhook_request(params: dict[str, str], expected_verification_token: str) → int[source]

Handle a verify webhook request from WhatsApp.

Parameters:

  • params – Dictionary containing webhook parameters from query string

  • expected_verification_token – The expected verification token to validate against

Returns:

The challenge value if verification succeeds

Return type:

int

Raises:

ValueError – If verification fails due to missing parameters or invalid token

async handle_webhook_request(request: WhatsAppWebhookRequest, connection_callback: Callable[[SmallWebRTCConnection], Awaitable[None]] | None = None, raw_body: bytes | None = None, sha256_signature: str | None = None) → bool[source]

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.

Parameters:

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

  • raw_body – Optional bytes containing the raw request body.

  • sha256_signature – Optional X-Hub-Signature-256 header value from the request.

Returns:

True if the webhook request was handled successfully, False otherwise

Return type:

bool

Raises:

  • ValueError – If the webhook request contains no supported events

  • Exception – If connection establishment or API calls fail