Sending Proactive Notifications to a Customer via Genesys Cloud Web Messaging

What You Will Build

• One sentence: This tutorial demonstrates how to programmatically initiate a web messaging session with a customer who previously engaged with your web widget, using their stored contactId.
• One sentence: This uses the Genesys Cloud CX Web Messaging API (/api/v2/webchat/messaging) and the Python SDK.
• One sentence: The programming language covered is Python 3.9+.

Prerequisites

• OAuth Client Type: Public or Confidential Client with the webchat:messaging scope.
• Required Scopes: webchat:messaging, webchat:session:read (if you need to verify session status before sending).
• SDK Version: genesys-cloud-purecloud-platform-client >= 150.0.0.
• Runtime: Python 3.9 or higher.
• Dependencies: genesys-cloud-purecloud-platform-client, pyjwt (optional, for debugging tokens), httpx (for direct API calls if preferred, though SDK is recommended).

Install the SDK via pip:

pip install genesys-cloud-purecloud-platform-client

Authentication Setup

Genesys Cloud uses OAuth 2.0. For server-to-server communication (like sending proactive messages), you should use the Client Credentials Grant. This flow does not require user interaction and returns an access token with the scopes granted to your client application.

Below is a robust authentication helper that caches tokens and handles expiration automatically.

import time
import os
from typing import Optional
from purecloud_platform_client import Configuration, ApiClient

class GenesysAuth:
    def __init__(self, client_id: str, client_secret: str, environment: str = "mypurecloud.com"):
        self.client_id = client_id
        self.client_secret = client_secret
        self.environment = environment
        self.api_client: Optional[ApiClient] = None
        self.token_expiry: float = 0
        self.access_token: str = ""

    def get_api_client(self) -> ApiClient:
        """
        Returns a configured ApiClient. Handles token refresh if expired.
        """
        if self.api_client is None or time.time() > self.token_expiry:
            self._refresh_token()
        return self.api_client

    def _refresh_token(self) -> None:
        """
        Fetches a new OAuth token using Client Credentials Grant.
        """
        config = Configuration()
        # Set the base path based on environment (e.g., us-east-1.mypurecloud.com)
        config.host = f"https://{self.environment}"
        
        # Get token
        token_response = config.get_access_token(
            client_id=self.client_id,
            client_secret=self.client_secret,
            scopes=["webchat:messaging"] # Only request needed scopes
        )
        
        self.access_token = token_response.access_token
        # Set expiry slightly before actual expiry to avoid race conditions
        self.token_expiry = time.time() + token_response.expires_in - 10
        
        self.api_client = ApiClient(configuration=config)

Implementation

Step 1: Identify the Previous Customer Session

Before sending a proactive message, you must have a valid contactId or session_id from a previous interaction. In a typical architecture, this ID is stored in your CRM or database when the customer first opens the web widget.

If you do not have the ID stored, you cannot send a proactive message to a specific anonymous user. Web Messaging in Genesys Cloud is session-based. The contactId is the unique identifier for that specific browser session’s engagement with Genesys.

For this tutorial, we assume you have retrieved the contactId from your database. If you need to find it, you would query the webchat:session:read endpoint, but that requires the session_id which is tied to the browser. Therefore, proactive messaging relies entirely on your application storing the contactId during the initial connect or message event.

Step 2: Construct the Proactive Message Payload

To send a message, you use the POST /api/v2/webchat/messaging endpoint. The critical parameter here is contactId. You also need to specify the type of the message. For proactive notifications, the type is usually agent or bot, depending on whether you want it to appear as a human agent or an automated bot.

The request body requires:

1. contactId: The ID of the customer session.
2. type: The sender type (agent or bot).
3. message: The content of the message.

Important: If the session has been terminated (e.g., the customer closed the browser or the session timed out), sending a message will fail. You must handle this gracefully.

Here is the code to send the message using the SDK.

from purecloud_platform_client import WebMessagingApi
from purecloud_platform_client.models import WebMessageRequestBody

def send_proactive_message(
    api_client: ApiClient,
    contact_id: str,
    message_text: str,
    sender_type: str = "bot"
) -> dict:
    """
    Sends a proactive message to a specific contact ID.
    
    Args:
        api_client: The authenticated ApiClient.
        contact_id: The contact ID from the previous session.
        message_text: The text content of the message.
        sender_type: 'agent' or 'bot'. Default is 'bot'.
        
    Returns:
        dict: The response from the API.
    """
    api_instance = WebMessagingApi(api_client)
    
    # Construct the request body
    body = WebMessageRequestBody(
        contact_id=contact_id,
        type=sender_type,
        message=message_text
    )
    
    try:
        # Execute the API call
        # The API returns a 200 OK with the message details if successful
        response = api_instance.post_webchat_messaging(body=body)
        
        print(f"Message sent successfully. Message ID: {response.id}")
        return response
        # Note: In the Python SDK, the response object might be a WebMessage object
        
    except Exception as e:
        # Handle specific HTTP errors
        if hasattr(e, 'status') and e.status == 400:
            print(f"Bad Request: The contact ID might be invalid or the session is expired. Details: {e.body}")
        elif hasattr(e, 'status') and e.status == 404:
            print(f"Not Found: The contact ID does not exist in Genesys Cloud.")
        elif hasattr(e, 'status') and e.status == 429:
            print(f"Rate Limited: Too many requests. Wait before retrying.")
        else:
            print(f"Error sending message: {e}")
        raise e

Step 3: Handling Session Expiration and Retries

Web messaging sessions in Genesys Cloud have a timeout (default is often 15-30 minutes of inactivity). If you attempt to send a proactive message to an expired session, the API will return a 400 Bad Request or 404 Not Found.

You should implement a retry mechanism with exponential backoff for transient errors (429, 5xx), but fail fast for 400/404 errors as they indicate a permanent state (invalid/expired contact).

import time
from purecloud_platform_client import ApiClient

def send_with_retry(
    auth: GenesysAuth,
    contact_id: str,
    message_text: str,
    max_retries: int = 3
) -> bool:
    """
    Sends a message with retry logic for rate limits and server errors.
    Fails immediately on 400/404.
    """
    api_client = auth.get_api_client()
    
    for attempt in range(max_retries):
        try:
            send_proactive_message(api_client, contact_id, message_text)
            return True
            
        except Exception as e:
            # Check if it is a retryable error
            is_retryable = False
            if hasattr(e, 'status'):
                status = e.status
                if status == 429:
                    is_retryable = True
                    wait_time = 2 ** attempt # Exponential backoff
                    print(f"Rate limited. Retrying in {wait_time} seconds...")
                elif 500 <= status < 600:
                    is_retryable = True
                    wait_time = 2 ** attempt
                    print(f"Server error {status}. Retrying in {wait_time} seconds...")
            
            if is_retryable:
                time.sleep(wait_time)
            else:
                # Non-retryable error (e.g., 400, 404)
                print(f"Non-retryable error on attempt {attempt + 1}. Aborting.")
                return False
                
    print(f"Failed to send message after {max_retries} retries.")
    return False

Complete Working Example

Below is a complete, runnable Python script. Replace the placeholder values with your Genesys Cloud credentials and a valid contactId.

import os
import time
from purecloud_platform_client import Configuration, ApiClient, WebMessagingApi
from purecloud_platform_client.models import WebMessageRequestBody

class GenesysAuth:
    def __init__(self, client_id: str, client_secret: str, environment: str = "mypurecloud.com"):
        self.client_id = client_id
        self.client_secret = client_secret
        self.environment = environment
        self.api_client = None
        self.token_expiry = 0

    def get_api_client(self) -> ApiClient:
        if self.api_client is None or time.time() > self.token_expiry:
            self._refresh_token()
        return self.api_client

    def _refresh_token(self) -> None:
        config = Configuration()
        config.host = f"https://{self.environment}"
        
        token_response = config.get_access_token(
            client_id=self.client_id,
            client_secret=self.client_secret,
            scopes=["webchat:messaging"]
        )
        
        self.token_expiry = time.time() + token_response.expires_in - 10
        self.api_client = ApiClient(configuration=config)

def send_proactive_message(
    api_client: ApiClient,
    contact_id: str,
    message_text: str,
    sender_type: str = "bot"
):
    api_instance = WebMessagingApi(api_client)
    
    body = WebMessageRequestBody(
        contact_id=contact_id,
        type=sender_type,
        message=message_text
    )
    
    try:
        response = api_instance.post_webchat_messaging(body=body)
        print(f"Success: Message sent with ID {response.id}")
        return response
    except Exception as e:
        if hasattr(e, 'status'):
            if e.status == 400:
                raise ValueError(f"Invalid Contact ID or Session Expired: {e.body}")
            elif e.status == 404:
                raise ValueError(f"Contact ID not found: {e.body}")
            elif e.status == 429:
                raise ConnectionError("Rate limited. Please retry later.")
        raise e

def main():
    # Configuration
    CLIENT_ID = os.getenv("GENESYS_CLIENT_ID", "YOUR_CLIENT_ID")
    CLIENT_SECRET = os.getenv("GENESYS_CLIENT_SECRET", "YOUR_CLIENT_SECRET")
    ENVIRONMENT = os.getenv("GENESYS_ENVIRONMENT", "us-east-1.mypurecloud.com")
    CONTACT_ID = os.getenv("GENESYS_CONTACT_ID", "YOUR_PREVIOUS_CONTACT_ID")
    MESSAGE = "Hello! We noticed you were looking at our summer collection. Do you have any questions?"

    # Initialize Auth
    auth = GenesysAuth(CLIENT_ID, CLIENT_SECRET, ENVIRONMENT)
    
    # Get API Client
    api_client = auth.get_api_client()
    
    # Send Message
    try:
        send_proactive_message(
            api_client=api_client,
            contact_id=CONTACT_ID,
            message_text=MESSAGE,
            sender_type="bot"
        )
    except ValueError as ve:
        print(f"Validation Error: {ve}")
    except ConnectionError as ce:
        print(f"Connection Error: {ce}")
    except Exception as e:
        print(f"Unexpected Error: {e}")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 400 Bad Request - “Contact ID is invalid or session has expired”

• What causes it: The contactId provided does not correspond to an active web messaging session. Sessions expire after a period of inactivity (configurable in Genesys Cloud, default is often 15 minutes). Alternatively, the contactId format is incorrect.
• How to fix it:
  1. Verify the contactId was captured from a recent session.
  2. Check the session timeout settings in your Genesys Cloud Web Messaging configuration.
  3. If the session is expired, you cannot “wake it up.” You must treat this as a new session. You can prompt the user to start a new chat via your web widget instead.
• Code showing the fix:

  try:
      send_proactive_message(...)
  except ValueError as e:
      if "expired" in str(e).lower():
          # Logic to notify user to start new chat
          print("Session expired. Please start a new chat.")
      else:
          print(f"Error: {e}")
  

Error: 403 Forbidden - “Insufficient Permissions”

• What causes it: The OAuth token used does not have the webchat:messaging scope.
• How to fix it: Ensure your OAuth client in Genesys Cloud has the webchat:messaging scope assigned. When requesting the token, ensure you specify scopes=["webchat:messaging"].
• Code showing the fix:

  # In _refresh_token
  token_response = config.get_access_token(
      client_id=self.client_id,
      client_secret=self.client_secret,
      scopes=["webchat:messaging"] # Ensure this scope is present
  )
  

Error: 429 Too Many Requests

• What causes it: You are exceeding the API rate limits for the Web Messaging endpoint.
• How to fix it: Implement exponential backoff and retry logic. Do not retry immediately.
• Code showing the fix: See the send_with_retry function in Step 3.

Error: 404 Not Found

• What causes it: The contactId does not exist in Genesys Cloud.
• How to fix it: Verify the contactId was stored correctly from the initial widget connection. If you are using a custom integration, ensure the contactId is passed correctly in the connect or message events.

Official References

• Genesys Cloud Web Messaging API Documentation
• Genesys Cloud Python SDK Reference
• Web Messaging Scopes