Sending Proactive Notifications to a Customer with History via Genesys Cloud APIs

What You Will Build

• You will build a service that retrieves historical web message transcripts for a specific customer and uses that context to send a proactive outbound web message or email notification.
• This tutorial uses the Genesys Cloud CX Analytics API to retrieve conversation details and the Engage API (or Email API) to dispatch the proactive message.
• The implementation is provided in Python using the purecloud-platform-client-v2 SDK and the requests library for raw API calls where the SDK lacks specific proactive messaging helpers.

Prerequisites

• OAuth Client Type: Confidential Client (Client Credentials Grant) or Public Client (Authorization Code Grant with PKCE). For backend services, Confidential Client is recommended.
• Required Scopes:
  • analytics:conversation:read (to retrieve historical transcript data)
  • message:outbound:create (to send proactive web messages via Engage)
  • email:outbound:send (if falling back to email)
  • user:read (to identify the agent or service account sending the message)
• SDK Version: purecloud-platform-client-v2 >= 164.0.0 (Python)
• Runtime: Python 3.9+
• Dependencies:
  • pip install purecloud-platform-client-v2
  • pip install requests
  • pip install python-dotenv (for secure credential management)

Authentication Setup

Genesys Cloud uses OAuth 2.0 for all API access. For a backend service sending proactive notifications, you will typically use the Client Credentials Grant flow. This flow requires a registered OAuth Client in the Genesys Cloud Admin Console with the appropriate scopes granted.

Step 1: Configure Environment Variables

Create a .env file in your project root. Never hardcode credentials.

# .env
GENESYS_CLOUD_REGION=us-east-1 # e.g., eu-west-1, au-southeast-2
GENESYS_CLOUD_CLIENT_ID=your_client_id_here
GENESYS_CLOUD_CLIENT_SECRET=your_client_secret_here

Step 2: Initialize the SDK Client

The Genesys Cloud Python SDK handles token acquisition and refresh automatically when initialized with a client ID and secret.

import os
from dotenv import load_dotenv
from purecloud_platform_client_v2 import PlatformClient, Configuration

load_dotenv()

def get_platform_client() -> PlatformClient:
    """
    Initializes and returns a configured Genesys Cloud Platform Client.
    """
    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")

    if not client_id or not client_secret:
        raise ValueError("GENESYS_CLOUD_CLIENT_ID and GENESYS_CLOUD_CLIENT_SECRET are required.")

    # Configure the API client
    config = Configuration()
    config.host = f"https://api.{region}.mypurecloud.com"
    config.client_id = client_id
    config.client_secret = client_secret

    # Create the platform client
    platform_client = PlatformClient(config)
    
    # Explicitly set the authentication method for client credentials
    platform_client.set_auth(
        client_id=client_id,
        client_secret=client_secret
    )
    
    return platform_client

Implementation

The workflow consists of three distinct phases:

1. Retrieve History: Query the Analytics API for past web message conversations associated with the customer.
2. Enrich Context: Parse the transcript to extract relevant context (e.g., previous issue, name, sentiment).
3. Send Proactive Message: Construct and dispatch the new message using the Engage API.

Step 1: Retrieve Historical Conversation Details

To send a context-aware proactive notification, you first need to prove the customer exists in your system and retrieve their last interaction. The POST /api/v2/analytics/conversations/details/query endpoint is the primary tool for this.

Required Scope: analytics:conversation:read

from purecloud_platform_client_v2 import AnalyticsApi, ConversationDetailQueryRequest
from purecloud_platform_client_v2.api_exception import ApiException
import json

def get_last_web_message_history(platform_client: PlatformClient, email_address: str, max_records: int = 5):
    """
    Retrieves the last N web message conversations for a specific email address.
    """
    analytics_api = AnalyticsApi(platform_client)
    
    # Construct the query body
    # We filter by communication type 'webchat' and the participant's email
    query_body = ConversationDetailQueryRequest(
        interval="2023-01-01T00:00:00Z/2023-12-31T23:59:59Z", # Adjust interval as needed
        group_by=["participant"],
        metrics=["durationSeconds"],
        filter=[
            {
                "type": "equals",
                "field": "conversation.communicationType",
                "value": "webchat"
            },
            {
                "type": "contains",
                "field": "conversation.participants.email",
                "value": email_address
            }
        ],
        size=max_records,
        sort=[{"field": "conversation.startTime", "direction": "desc"}]
    )

    try:
        # Execute the query
        response = analytics_api.post_analytics_conversations_details_query(body=query_body)
        
        if not response.entity:
            print("No historical conversations found.")
            return None

        # The response contains summary data. To get the full transcript, 
        # we need the conversation IDs.
        conversation_ids = [item.id for item in response.entity]
        
        # Fetch full details for the most recent conversation
        if conversation_ids:
            latest_conv_id = conversation_ids[0]
            return fetch_full_transcript(platform_client, latest_conv_id)
        
        return None

    except ApiError as e:
        print(f"Analytics API Error: {e.status} - {e.reason}")
        raise

def fetch_full_transcript(platform_client: PlatformClient, conversation_id: str):
    """
    Fetches the full conversation transcript including messages.
    """
    analytics_api = AnalyticsApi(platform_client)
    
    # Query for the specific conversation details
    query_body = ConversationDetailQueryRequest(
        interval="2023-01-01T00:00:00Z/2023-12-31T23:59:59Z",
        group_by=[], # No grouping to get individual conversation rows
        metrics=[],
        filter=[
            {
                "type": "equals",
                "field": "conversation.id",
                "value": conversation_id
            }
        ],
        size=1
    )

    try:
        response = analytics_api.post_analytics_conversations_details_query(body=query_body)
        if response.entity and len(response.entity) > 0:
            return response.entity[0]
        return None
    except ApiError as e:
        print(f"Failed to fetch transcript: {e.reason}")
        return None

Step 2: Extract Context from Transcript

The Analytics API returns a structured object. The interactions field contains the sequence of messages. We need to parse this to find the customer’s last issue or intent.

from typing import Dict, Optional

def extract_customer_context(conversation_data: Dict) -> Optional[Dict]:
    """
    Parses the conversation data to extract the customer's last message and name.
    """
    if not conversation_data or not conversation_data.interactions:
        return None

    customer_name = "Valued Customer"
    last_customer_message = ""

    # Iterate through interactions to find customer messages
    # Interactions are ordered chronologically
    for interaction in conversation_data.interactions:
        if interaction.direction == "inbound": # Message from customer to agent/system
            if interaction.sender and interaction.sender.name:
                customer_name = interaction.sender.name
            if interaction.text:
                last_customer_message = interaction.text

    return {
        "name": customer_name,
        "last_message": last_customer_message,
        "conversation_id": conversation_data.id
    }

Step 3: Send Proactive Web Message via Engage API

Genesys Cloud’s “Proactive Messaging” feature allows you to send a web message to a user who has previously chatted with you, provided they have the chat widget open or are on a tracked page. This is done via the Engage API.

Required Scope: message:outbound:create

Endpoint: POST /api/v2/engage/messages

Note: The SDK does not always have a direct method for every Engage endpoint. Using the requests library with the SDK’s token is often cleaner for newer Engage features.

import requests
from purecloud_platform_client_v2 import PlatformClient

def send_proactive_web_message(platform_client: PlatformClient, customer_email: str, context: Dict):
    """
    Sends a proactive web message to a customer using the Engage API.
    
    Args:
        platform_client: The initialized Genesys Cloud client.
        customer_email: The email address of the customer.
        context: Dictionary containing 'name' and 'last_message'.
    """
    # Get the current access token from the platform client
    # The SDK caches this token. We extract it for the raw HTTP request.
    auth_client = platform_client.get_auth_client()
    access_token = auth_client.get_access_token()
    
    if not access_token:
        raise Exception("Failed to retrieve access token.")

    base_url = platform_client.get_config().host.replace("api.", "") # Engage API might use different subdomain
    engage_url = f"https://engage.{base_url}/api/v2/engage/messages"
    
    # Construct the message payload
    # The 'to' field requires a specific structure for web messaging
    payload = {
        "to": {
            "type": "email",
            "address": customer_email
        },
        "from": {
            "type": "email",
            "address": "support@yourcompany.com", # Must be a verified sender in Genesys
            "name": "Customer Support Bot"
        },
        "subject": "Follow up on your recent inquiry",
        "body": {
            "contentType": "text/html",
            "content": f"""
            <h2>Hello {context.get('name', 'there')}</h2>
            <p>We noticed you recently asked about: <strong>{context.get('last_message', 'an issue')}</strong>.</p>
            <p>Did we resolve your issue? Click below to chat with an agent if you need more help.</p>
            """
        },
        "messageType": "proactive" # Critical: Marks this as a proactive notification
    }

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

    try:
        response = requests.post(engage_url, json=payload, headers=headers)
        
        if response.status_code == 202: # Accepted
            print(f"Proactive message sent successfully. ID: {response.json().get('id')}")
            return response.json()
        elif response.status_code == 400:
            print(f"Bad Request: {response.text}")
            raise ValueError(f"Invalid payload: {response.text}")
        elif response.status_code == 429:
            print("Rate limited. Please retry later.")
            raise Exception("Rate limited")
        else:
            print(f"Error sending message: {response.status_code} - {response.text}")
            raise Exception(f"Engage API Error: {response.text}")
            
    except requests.exceptions.RequestException as e:
        print(f"Network error: {e}")
        raise

Step 4: Fallback to Email (Optional but Recommended)

If the customer is not currently active on the web widget, the proactive web message may fail or not be delivered immediately. A robust system often sends an email as a fallback.

def send_fallback_email(platform_client: PlatformClient, customer_email: str, context: Dict):
    """
    Sends a standard email using the Email API.
    """
    email_api = platform_client.email_api # Assuming EmailApi is imported from SDK
    
    # Construct the email body object
    email_body = {
        "from": {
            "address": "support@yourcompany.com",
            "name": "Support Team"
        },
        "to": [
            {
                "address": customer_email,
                "name": context.get('name', 'Customer')
            }
        ],
        "subject": "Checking in on your recent support request",
        "body": {
            "contentType": "text/html",
            "content": f"<p>Hi {context.get('name', 'there')},</p><p>Following up on: {context.get('last_message')}</p>"
        }
    }
    
    try:
        response = email_api.post_email_outbound_send(body=email_body)
        print(f"Email sent. ID: {response.id}")
        return response
    except Exception as e:
        print(f"Email failed: {e}")
        return None

Complete Working Example

Below is the consolidated script. Save this as proactive_notification.py.

import os
import sys
from dotenv import load_dotenv
from purecloud_platform_client_v2 import PlatformClient, Configuration, AnalyticsApi, ConversationDetailQueryRequest, ApiError
import requests
from typing import Dict, Optional

# Load environment variables
load_dotenv()

def get_platform_client() -> PlatformClient:
    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")

    if not client_id or not client_secret:
        raise ValueError("Missing credentials in .env file.")

    config = Configuration()
    config.host = f"https://api.{region}.mypurecloud.com"
    config.client_id = client_id
    config.client_secret = client_secret

    platform_client = PlatformClient(config)
    platform_client.set_auth(client_id=client_id, client_secret=client_secret)
    return platform_client

def get_last_web_message_history(platform_client: PlatformClient, email_address: str) -> Optional[Dict]:
    analytics_api = AnalyticsApi(platform_client)
    
    # Define a wide interval to catch recent history
    query_body = ConversationDetailQueryRequest(
        interval="2023-01-01T00:00:00Z/2024-12-31T23:59:59Z",
        group_by=["participant"],
        metrics=["durationSeconds"],
        filter=[
            {"type": "equals", "field": "conversation.communicationType", "value": "webchat"},
            {"type": "contains", "field": "conversation.participants.email", "value": email_address}
        ],
        size=1,
        sort=[{"field": "conversation.startTime", "direction": "desc"}]
    )

    try:
        response = analytics_api.post_analytics_conversations_details_query(body=query_body)
        if not response.entity or len(response.entity) == 0:
            return None
        
        latest_conv_id = response.entity[0].id
        
        # Fetch full transcript
        transcript_query = ConversationDetailQueryRequest(
            interval="2023-01-01T00:00:00Z/2024-12-31T23:59:59Z",
            group_by=[],
            metrics=[],
            filter=[{"type": "equals", "field": "conversation.id", "value": latest_conv_id}],
            size=1
        )
        
        transcript_response = analytics_api.post_analytics_conversations_details_query(body=transcript_query)
        if transcript_response.entity and len(transcript_response.entity) > 0:
            return transcript_response.entity[0]
        return None
    except ApiError as e:
        print(f"Analytics Error: {e.reason}")
        return None

def extract_context(conversation_data: Dict) -> Dict:
    customer_name = "Valued Customer"
    last_msg = "unknown topic"
    
    if conversation_data.interactions:
        for interaction in conversation_data.interactions:
            if interaction.direction == "inbound":
                if interaction.sender and interaction.sender.name:
                    customer_name = interaction.sender.name
                if interaction.text:
                    last_msg = interaction.text[:100] # Truncate for safety
    return {"name": customer_name, "last_message": last_msg}

def send_proactive_message(platform_client: PlatformClient, email: str, context: Dict):
    auth_client = platform_client.get_auth_client()
    token = auth_client.get_access_token()
    
    if not token:
        raise Exception("No auth token")

    host = platform_client.get_config().host
    region = host.split(".")[1] # Extract region from api.us-east-1.mypurecloud.com
    engage_url = f"https://engage.{region}.mypurecloud.com/api/v2/engage/messages"
    
    payload = {
        "to": {"type": "email", "address": email},
        "from": {"type": "email", "address": "noreply@yourdomain.com", "name": "Support Bot"},
        "subject": "Quick Check-In",
        "body": {
            "contentType": "text/html",
            "content": f"<p>Hi {context['name']},</p><p>We saw you asked about <b>{context['last_message']}</b>. Need more help?</p>"
        },
        "messageType": "proactive"
    }

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

    try:
        res = requests.post(engage_url, json=payload, headers=headers)
        res.raise_for_status()
        print(f"Success: Message sent. ID: {res.json().get('id')}")
    except requests.exceptions.HTTPError as e:
        print(f"HTTP Error: {e.response.status_code} - {e.response.text}")
    except Exception as e:
        print(f"Request Error: {e}")

def main():
    target_email = os.getenv("TARGET_EMAIL", "customer@example.com")
    
    print(f"Initiating proactive notification flow for {target_email}...")
    
    client = get_platform_client()
    
    # 1. Get History
    hist = get_last_web_message_history(client, target_email)
    
    if not hist:
        print("No historical web messages found. Aborting.")
        return

    # 2. Extract Context
    context = extract_context(hist)
    print(f"Context extracted: Name={context['name']}, Topic={context['last_message']}")

    # 3. Send Message
    send_proactive_message(client, target_email, context)

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 401 Unauthorized

Cause: The OAuth token is expired or invalid.
Fix: Ensure the PlatformClient is initialized correctly. The SDK auto-refreshes tokens, but if you are using the raw requests library, you must fetch the token immediately before the call using auth_client.get_access_token(). Do not cache the token string manually across long-running processes.

Error: 403 Forbidden

Cause: The OAuth Client lacks the required scopes.
Fix:

1. Log in to Genesys Cloud Admin.
2. Navigate to Admin > Security > OAuth Clients.
3. Select your client.
4. Ensure analytics:conversation:read and message:outbound:create are checked under Scopes.
5. Save changes. Note: Scope changes may take up to 15 minutes to propagate.

Error: 400 Bad Request - “Invalid recipient type”

Cause: The to field in the Engage API payload is malformed.
Fix: For proactive web messages triggered by email history, the to field must specify "type": "email". If you are trying to target a specific session ID, the type would be "webchat", but that requires the active session ID, which is not available from historical analytics queries.

Error: 429 Too Many Requests

Cause: You have exceeded the API rate limit.
Fix: Implement exponential backoff. The Genesys Cloud API returns Retry-After headers.

import time

def retry_with_backoff(func, *args, max_retries=3, **kwargs):
    for attempt in range(max_retries):
        try:
            return func(*args, **kwargs)
        except requests.exceptions.HTTPError as e:
            if e.response.status_code == 429:
                retry_after = int(e.response.headers.get('Retry-After', 2 ** attempt))
                print(f"Rate limited. Waiting {retry_after} seconds...")
                time.sleep(retry_after)
            else:
                raise
    raise Exception("Max retries exceeded")

Official References

• Genesys Cloud Analytics API Documentation
• Engage API Reference
• Python SDK Documentation
• OAuth Scopes Reference