Sending Proactive Notifications to a Customer with a History of Web Messaging

What You Will Build

• You will build a Python script that identifies a specific customer by their external ID or phone number and sends them a proactive web message notification.
• This solution uses the Genesys Cloud CX API v2 endpoints for User Search, Interaction Management, and Webchat Proactive Messaging.
• The tutorial covers Python using the requests library and the official genesyscloud SDK.

Prerequisites

• OAuth Client: A Genesys Cloud OAuth client with the webchat:proactive:send and interaction:read scopes.
• SDK Version: genesys-cloud-purecloud-platform-client v130.0.0 or later.
• Runtime: Python 3.8+.
• Dependencies: pip install requests genesys-cloud-purecloud-platform-client.
• Environment Variables: You must have GENESYS_CLOUD_CLIENT_ID, GENESYS_CLOUD_CLIENT_SECRET, and GENESYS_CLOUD_REGION set in your environment.

Authentication Setup

Genesys Cloud uses OAuth 2.0 Client Credentials grant for server-to-server integrations. You must obtain a valid access token before making any API calls. The token expires after one hour, so production code should implement a refresh mechanism. For this tutorial, we will use a helper function to fetch a fresh token.

import os
import requests
import time
from typing import Optional

def get_access_token() -> str:
    """
    Fetches a new OAuth2 access token using Client Credentials.
    In production, implement caching to avoid fetching a new token on every request.
    """
    client_id = os.getenv("GENESYS_CLOUD_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLOUD_CLIENT_SECRET")
    region = os.getenv("GENESYS_CLOUD_REGION", "us-east-1")
    
    if not client_id or not client_secret:
        raise ValueError("GENESYS_CLOUD_CLIENT_ID and GENESYS_CLOUD_CLIENT_SECRET must be set.")

    # Construct the OAuth endpoint based on the region
    if region == "us-east-1":
        auth_host = "https://api.mypurecloud.com"
    elif region == "eu-west-1":
        auth_host = "https://api.eu.pure.cloud"
    elif region == "ap-southeast-1":
        auth_host = "https://api.ap.pure.cloud"
    else:
        raise ValueError(f"Unsupported region: {region}")

    url = f"{auth_host}/oauth/token"
    headers = {
        "Content-Type": "application/x-www-form-urlencoded"
    }
    data = {
        "grant_type": "client_credentials",
        "client_id": client_id,
        "client_secret": client_secret
    }

    response = requests.post(url, headers=headers, data=data)
    
    if response.status_code != 200:
        raise Exception(f"Failed to get access token: {response.status_code} - {response.text}")

    token_data = response.json()
    return token_data["access_token"]

# Example usage
token = get_access_token()
print(f"Access Token obtained: {token[:20]}...")

Implementation

Step 1: Locate the Customer Interaction Context

To send a proactive message, you need to target a specific device or user. Genesys Cloud Webchat proactive messages are typically sent to a device identified by a device_id or by matching a user’s external_id (often an email or phone number) within a specific webchat widget.

First, we must find the recent interactions associated with the customer. We will search for interactions by the customer’s phone number or email. This step ensures we are targeting a known entity and allows us to retrieve the device_id if it was stored in previous interaction metadata.

Required Scope: interaction:read

import json
from datetime import datetime, timedelta

def find_recent_interaction(token: str, customer_phone: str, region: str) -> Optional[dict]:
    """
    Searches for recent webchat interactions for a given phone number.
    Returns the most recent interaction object or None.
    """
    auth_host = get_auth_host(region)
    url = f"{auth_host}/api/v2/interactions/details/query"
    
    # Define the time window for the search (last 30 days)
    end_time = datetime.utcnow().isoformat() + "Z"
    start_time = (datetime.utcnow() - timedelta(days=30)).isoformat() + "Z"
    
    # Query body for interaction search
    query_body = {
        "groupBy": "interaction",
        "filter": {
            "type": "and",
            "clauses": [
                {
                    "type": "field",
                    "field": "routing.phoneNumber",
                    "op": "eq",
                    "values": [customer_phone]
                },
                {
                    "type": "field",
                    "field": "routing.mediaType",
                    "op": "eq",
                    "values": ["webchat"]
                },
                {
                    "type": "field",
                    "field": "routing.startTime",
                    "op": "gte",
                    "values": [start_time]
                }
            ]
        },
        "sort": [
            {
                "field": "routing.startTime",
                "order": "desc"
            }
        ],
        "size": 1
    }

    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }

    response = requests.post(url, headers=headers, json=query_body)
    
    if response.status_code == 200:
        data = response.json()
        if data.get("total") > 0:
            return data["results"][0]
        else:
            return None
    else:
        raise Exception(f"Failed to query interactions: {response.status_code} - {response.text}")

def get_auth_host(region: str) -> str:
    if region == "us-east-1":
        return "https://api.mypurecloud.com"
    elif region == "eu-west-1":
        return "https://api.eu.pure.cloud"
    elif region == "ap-southeast-1":
        return "https://api.ap.pure.cloud"
    else:
        raise ValueError(f"Unsupported region: {region}")

Step 2: Identify the Target Webchat Widget and Device

Proactive messages are sent to a specific Webchat Widget. You must know the widgetId of the webchat interface where you want the message to appear. Additionally, if you want to ensure the message goes to a specific browser instance, you need the device_id.

If you do not have the device_id from the previous interaction, you can send the proactive message to all devices associated with a user’s external_id (if configured in your Genesys Cloud instance to track users by external ID). However, the most robust method for “previous session” targeting is using the device_id stored in the interaction metadata.

For this example, we assume you have the widgetId and you are targeting by external_id (e.g., email) or device_id. If you do not have the device_id, you must rely on the Genesys Cloud User Management or Interaction metadata to map the customer to a device.

Note: The proactive message API does not “search” for devices. It pushes to a known device_id or a user identified by external_id within a widget.

Let us assume we are targeting a user by their external_id (email) in a specific widget.

Required Scope: webchat:proactive:send

Step 3: Construct and Send the Proactive Message

The core API call is POST /api/v2/webchat/proactive/messages. This endpoint accepts a payload containing the widget ID, the target device/user, and the message content.

def send_proactive_message(token: str, widget_id: str, target_external_id: str, message_body: str, region: str) -> dict:
    """
    Sends a proactive webchat message to a user identified by external_id.
    """
    auth_host = get_auth_host(region)
    url = f"{auth_host}/api/v2/webchat/proactive/messages"
    
    # The payload structure for proactive messaging
    payload = {
        "widgetId": widget_id,
        "target": {
            "type": "externalId",  # Can also be "deviceId"
            "value": target_external_id
        },
        "message": {
            "type": "text",
            "text": message_body
        },
        "metadata": {
            "source": "proactive_campaign",
            "campaign_id": "summer_sale_2024"
        }
    }

    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }

    response = requests.post(url, headers=headers, json=payload)
    
    if response.status_code == 202:
        # 202 Accepted means the message is queued for delivery
        return {"status": "queued", "response": response.json()}
    elif response.status_code == 404:
        # Target user/device not found or widget does not exist
        return {"status": "not_found", "error": response.json()}
    elif response.status_code == 429:
        # Rate limited
        retry_after = response.headers.get("Retry-After", 1)
        return {"status": "rate_limited", "retry_after": int(retry_after)}
    else:
        raise Exception(f"Failed to send proactive message: {response.status_code} - {response.text}")

Complete Working Example

This script combines the authentication, interaction lookup (to verify the user exists), and the proactive message sending. It demonstrates a safe workflow: verify the user had a recent interaction, then send the message.

import os
import sys
import time
import requests
from datetime import datetime, timedelta
from typing import Optional

# --- Configuration ---
REGION = os.getenv("GENESYS_CLOUD_REGION", "us-east-1")
CLIENT_ID = os.getenv("GENESYS_CLOUD_CLIENT_ID")
CLIENT_SECRET = os.getenv("GENESYS_CLOUD_CLIENT_SECRET")
WIDGET_ID = os.getenv("GENESYS_CLOUD_WIDGET_ID") # Required: Your Webchat Widget ID

# --- Helper Functions ---

def get_auth_host(region: str) -> str:
    if region == "us-east-1":
        return "https://api.mypurecloud.com"
    elif region == "eu-west-1":
        return "https://api.eu.pure.cloud"
    elif region == "ap-southeast-1":
        return "https://api.ap.pure.cloud"
    else:
        raise ValueError(f"Unsupported region: {region}")

def get_access_token() -> str:
    client_id = CLIENT_ID
    client_secret = CLIENT_SECRET
    region = REGION
    
    if not client_id or not client_secret:
        raise ValueError("Environment variables GENESYS_CLOUD_CLIENT_ID and GENESYS_CLOUD_CLIENT_SECRET must be set.")

    auth_host = get_auth_host(region)
    url = f"{auth_host}/oauth/token"
    headers = {"Content-Type": "application/x-www-form-urlencoded"}
    data = {
        "grant_type": "client_credentials",
        "client_id": client_id,
        "client_secret": client_secret
    }

    response = requests.post(url, headers=headers, data=data)
    
    if response.status_code != 200:
        raise Exception(f"Failed to get access token: {response.status_code} - {response.text}")

    return response.json()["access_token"]

def verify_recent_interaction(token: str, customer_phone: str) -> bool:
    """
    Checks if the customer had a webchat interaction in the last 30 days.
    """
    auth_host = get_auth_host(REGION)
    url = f"{auth_host}/api/v2/interactions/details/query"
    
    end_time = datetime.utcnow().isoformat() + "Z"
    start_time = (datetime.utcnow() - timedelta(days=30)).isoformat() + "Z"
    
    query_body = {
        "groupBy": "interaction",
        "filter": {
            "type": "and",
            "clauses": [
                {"type": "field", "field": "routing.phoneNumber", "op": "eq", "values": [customer_phone]},
                {"type": "field", "field": "routing.mediaType", "op": "eq", "values": ["webchat"]},
                {"type": "field", "field": "routing.startTime", "op": "gte", "values": [start_time]}
            ]
        },
        "size": 1
    }

    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }

    response = requests.post(url, headers=headers, json=query_body)
    
    if response.status_code == 200:
        data = response.json()
        return data.get("total", 0) > 0
    else:
        raise Exception(f"Interaction check failed: {response.status_code}")

def send_proactive_notification(token: str, widget_id: str, target_external_id: str, message_text: str) -> dict:
    """
    Sends the proactive message.
    """
    auth_host = get_auth_host(REGION)
    url = f"{auth_host}/api/v2/webchat/proactive/messages"
    
    payload = {
        "widgetId": widget_id,
        "target": {
            "type": "externalId",
            "value": target_external_id
        },
        "message": {
            "type": "text",
            "text": message_text
        }
    }

    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }

    # Implement basic retry logic for 429s
    for attempt in range(3):
        response = requests.post(url, headers=headers, json=payload)
        
        if response.status_code == 202:
            return {"success": True, "message": "Message queued", "data": response.json()}
        elif response.status_code == 429:
            retry_after = int(response.headers.get("Retry-After", 2 ** attempt))
            print(f"Rate limited. Retrying in {retry_after} seconds...")
            time.sleep(retry_after)
        else:
            raise Exception(f"Failed to send message: {response.status_code} - {response.text}")
            
    raise Exception("Max retries exceeded for proactive message.")

# --- Main Execution ---

def main():
    # 1. Setup
    if not WIDGET_ID:
        print("Error: GENESYS_CLOUD_WIDGET_ID environment variable not set.")
        sys.exit(1)

    customer_phone = "+15550199822" # Example customer phone
    customer_external_id = "customer@example.com" # Example external ID (email)
    message_content = "Hello! We noticed you had a recent session with us. Is there anything else we can help you with today?"

    try:
        # 2. Authenticate
        print("Authenticating...")
        token = get_access_token()
        
        # 3. Verify Context (Optional but recommended)
        print(f"Checking recent interactions for {customer_phone}...")
        has_history = verify_recent_interaction(token, customer_phone)
        
        if not has_history:
            print("No recent webchat interactions found. Skipping proactive message to avoid spam.")
            return

        # 4. Send Proactive Message
        print(f"Sending proactive message to {customer_external_id}...")
        result = send_proactive_notification(token, WIDGET_ID, customer_external_id, message_content)
        
        print("Success:", result)

    except Exception as e:
        print(f"Error: {e}")
        sys.exit(1)

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 403 Forbidden

Cause: The OAuth token does not have the webchat:proactive:send scope. This is the most common error when starting with proactive messaging.

Fix:

1. Go to the Genesys Cloud Admin console.
2. Navigate to Setup > Integrations > OAuth 2.0 Clients.
3. Select your client.
4. Edit the Scopes.
5. Ensure webchat:proactive:send is checked.
6. Save and regenerate the token.

Code Check:
Verify your token request includes the correct scopes if you are using a custom token generation flow. In the Client Credentials flow, the scopes are defined in the client configuration in the admin console, not in the POST body.

Error: 404 Not Found

Cause:

1. The widgetId provided does not exist or is not active.
2. The target (external ID or device ID) does not exist in the Genesys Cloud system for that widget.

Fix:

1. Verify the widgetId by listing webchat widgets: GET /api/v2/webchat/widgets.
2. Ensure the target_external_id matches exactly how it was stored in the user’s profile during their previous interaction. Genesys Cloud is case-sensitive for external IDs.

Error: 429 Too Many Requests

Cause: You are exceeding the rate limit for the proactive messaging endpoint. The default rate limit is typically 100 requests per minute per client, but this can vary by license tier.

Fix:
Implement exponential backoff. The code example above includes a basic retry loop with Retry-After header parsing.

# Improved retry logic snippet
import time

def send_with_retry(url, headers, payload, max_retries=3):
    for attempt in range(max_retries):
        response = requests.post(url, headers=headers, json=payload)
        if response.status_code == 429:
            wait_time = int(response.headers.get("Retry-After", 2 ** attempt))
            print(f"Rate limited. Waiting {wait_time}s...")
            time.sleep(wait_time)
        else:
            return response
    raise Exception("Max retries exceeded")

Error: Message Not Receiving

Cause: The user’s browser does not have the webchat widget open, or the proactive messaging feature is not enabled in the Webchat Widget configuration.

Fix:

1. In the Genesys Cloud Admin console, go to Channels > Webchat > Widgets.
2. Select your widget.
3. Go to the Proactive tab.
4. Ensure “Enable Proactive Messaging” is checked.
5. Ensure the user has the widget open in their browser. Proactive messages are only delivered to active sessions or registered devices if push notifications are configured. Note: Webchat proactive messages typically require the user to be on the page with the widget loaded. If you need to reach users offline, consider using SMS or Email channels instead, or configure Webchat to integrate with a mobile push notification service.

Official References

• Genesys Cloud Webchat Proactive Messaging API
• Genesys Cloud Interaction Search API
• OAuth 2.0 Client Credentials Flow
• Genesys Cloud Python SDK Documentation