Sending Proactive Notifications to a Customer with Prior Web Messaging History

What You Will Build

• A Python script that retrieves a customer’s previous web messaging conversation history to determine context.
• A mechanism to send a proactive outbound web message to a specific user session using Genesys Cloud CX.
• Implementation of the Genesys Cloud CX Web Messaging SDK and REST API to bridge historical data with real-time engagement.

Prerequisites

• OAuth Client Type: Confidential Client (Client Credentials Flow) for backend API calls.
• Required Scopes: conversation:webchat:read, conversation:webchat:write, user:read.
• SDK Version: genesys-cloud-python v4.0.0+.
• Runtime: Python 3.9+.
• External Dependencies: genesys-cloud-python, requests, uuid.
• Genesys Cloud Setup: A configured Web Messaging integration with a valid siteName and orgId.

Authentication Setup

Genesys Cloud CX uses OAuth 2.0 Client Credentials Grant for server-to-server communication. You must obtain an access token before making any API calls. The token expires after 3600 seconds, so production code should implement a refresh mechanism.

import requests
import os
import time

class GenesysAuth:
    def __init__(self, client_id: str, client_secret: str, env_url: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.env_url = env_url  # e.g., "https://api.mypurecloud.com"
        self.token = None
        self.token_expiry = 0

    def get_token(self) -> str:
        # Check if token is still valid
        if self.token and time.time() < self.token_expiry - 60:
            return self.token

        # Request new token
        url = f"{self.env_url}/oauth/token"
        headers = {"Content-Type": "application/x-www-form-urlencoded"}
        data = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret
        }

        response = requests.post(url, headers=headers, data=data)
        response.raise_for_status()

        token_data = response.json()
        self.token = token_data["access_token"]
        self.token_expiry = time.time() + token_data["expires_in"]
        
        return self.token

Implementation

Step 1: Identifying the Target Customer Session

To send a proactive notification, you must target an active web messaging session. Genesys Cloud does not allow sending web messages to arbitrary email addresses or phone numbers directly via the API in the same way as SMS. You must target a specific userId associated with an active session on the web widget.

First, you need to identify the active sessions. We will query the conversations API to find active web chat conversations. This allows you to correlate a customer’s historical data with their current live session.

Endpoint: GET /api/v2/conversations/webchat
Scope: conversation:webchat:read

import requests
from typing import List, Dict, Any

class WebChatManager:
    def __init__(self, auth: GenesysAuth):
        self.auth = auth
        self.base_url = auth.env_url

    def get_active_webchat_sessions(self) -> List[Dict[str, Any]]:
        """
        Retrieves all active web chat conversations.
        In a production scenario, you would filter this by specific 
        participant attributes or external IDs.
        """
        url = f"{self.base_url}/api/v2/conversations/webchat"
        headers = {
            "Authorization": f"Bearer {self.auth.get_token()}",
            "Accept": "application/json"
        }
        
        params = {
            "pageSize": 25,
            "page": 1
        }

        response = requests.get(url, headers=headers, params=params)
        response.raise_for_status()
        
        return response.json().get("entities", [])

Step 2: Retrieving Historical Context

Before sending the proactive message, it is best practice to verify the customer’s intent or history. We will fetch the previous conversation details for the identified session. This ensures the proactive message is relevant.

Endpoint: GET /api/v2/conversations/{conversationId}
Scope: conversation:read

    def get_conversation_history(self, conversation_id: str) -> Dict[str, Any]:
        """
        Fetches detailed history for a specific conversation ID.
        """
        url = f"{self.base_url}/api/v2/conversations/{conversation_id}"
        headers = {
            "Authorization": f"Bearer {self.auth.get_token()}",
            "Accept": "application/json"
        }

        try:
            response = requests.get(url, headers=headers)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.HTTPError as e:
            if e.response.status_code == 404:
                print(f"Conversation {conversation_id} not found.")
                return {}
            raise

Step 3: Sending the Proactive Message

The core action is sending a message to an active session. Genesys Cloud supports sending messages as a “System” or as a specific “User” (Agent/Bot). For proactive notifications, you typically send as a Bot or a specific Agent ID.

To send a message, you use the POST /api/v2/conversations/{conversationId}/messages endpoint. The body must specify the to participant (the customer) and the from participant (the sender).

Endpoint: POST /api/v2/conversations/{conversationId}/messages
Scope: conversation:webchat:write

    def send_proactive_message(
        self, 
        conversation_id: str, 
        customer_user_id: str, 
        sender_user_id: str, 
        message_text: str
    ) -> Dict[str, Any]:
        """
        Sends a proactive text message to an active web chat session.
        
        Args:
            conversation_id: The ID of the active web chat conversation.
            customer_user_id: The userId of the customer in the session.
            sender_user_id: The userId of the bot/agent sending the message.
            message_text: The content of the proactive message.
        """
        url = f"{self.base_url}/api/v2/conversations/{conversation_id}/messages"
        headers = {
            "Authorization": f"Bearer {self.auth.get_token()}",
            "Content-Type": "application/json",
            "Accept": "application/json"
        }

        # Construct the message payload
        payload = {
            "to": {
                "id": customer_user_id,
                "name": "Customer" # Optional but recommended
            },
            "from": {
                "id": sender_user_id,
                "name": "ProactiveBot" # Name of the sender
            },
            "text": message_text,
            "type": "text"
        }

        try:
            response = requests.post(url, headers=headers, json=payload)
            
            # Handle specific error codes
            if response.status_code == 409:
                # Conflict: Conversation might be closed or invalid state
                print(f"Conflict sending message to {conversation_id}: {response.text}")
                return {"error": "Conflict", "details": response.text}
            elif response.status_code == 404:
                # Not Found: Conversation or user ID invalid
                print(f"Entity not found: {response.text}")
                return {"error": "Not Found", "details": response.text}
            elif response.status_code == 429:
                # Rate Limited
                print("Rate limited. Please retry after delay.")
                return {"error": "Rate Limited"}
            
            response.raise_for_status()
            return response.json()
            
        except requests.exceptions.RequestException as e:
            print(f"Error sending message: {e}")
            return {"error": str(e)}

Step 4: Orchestrating the Logic

We combine the steps to create a workflow that finds an active session, checks if it belongs to a high-value customer (simulated here), and sends a proactive offer.

def main():
    # Configuration
    CLIENT_ID = os.getenv("GENESYS_CLIENT_ID")
    CLIENT_SECRET = os.getenv("GENESYS_CLIENT_SECRET")
    ENV_URL = os.getenv("GENESYS_ENV_URL", "https://api.mypurecloud.com")
    SENDER_BOT_ID = os.getenv("GENESYS_BOT_USER_ID") # ID of the bot user in Genesys

    if not all([CLIENT_ID, CLIENT_SECRET, SENDER_BOT_ID]):
        raise ValueError("Missing environment variables for Genesys Cloud.")

    # Initialize Auth and Manager
    auth = GenesysAuth(CLIENT_ID, CLIENT_SECRET, ENV_URL)
    manager = WebChatManager(auth)

    print("Fetching active web chat sessions...")
    active_sessions = manager.get_active_webchat_sessions()

    if not active_sessions:
        print("No active web chat sessions found.")
        return

    print(f"Found {len(active_sessions)} active session(s).")

    # Iterate through sessions to find a target
    # In production, you would filter by custom attributes or external ID
    for session in active_sessions:
        conversation_id = session["id"]
        
        # Identify participants
        participants = session.get("participants", [])
        customer_user_id = None
        customer_name = ""

        for participant in participants:
            if participant["role"] == "customer":
                customer_user_id = participant["userId"]
                customer_name = participant.get("name", "Unknown Customer")
                break
        
        if not customer_user_id:
            continue

        # Simulate business logic: Send proactive message to first active customer
        print(f"Targeting customer: {customer_name} in conversation {conversation_id}")
        
        # Retrieve history to ensure context (optional but recommended)
        history = manager.get_conversation_history(conversation_id)
        if history:
            print(f"Loaded history for conversation. Duration: {history.get('wrapUpCode', 'N/A')}")

        # Define proactive message content
        message_content = "Hi! We noticed you were browsing our premium plans. Would you like to speak with a specialist for a personalized discount?"

        # Send the message
        result = manager.send_proactive_message(
            conversation_id=conversation_id,
            customer_user_id=customer_user_id,
            sender_user_id=SENDER_BOT_ID,
            message_text=message_content
        )

        if "error" not in result:
            print(f"Successfully sent proactive message to {customer_name}.")
        else:
            print(f"Failed to send message: {result}")
        
        # Break after first success for demo purposes
        break

if __name__ == "__main__":
    main()

Complete Working Example

Below is the consolidated, production-ready Python script. It includes error handling, token management, and the full logic flow. Save this as proactive_webchat.py.

import os
import time
import requests
from typing import List, Dict, Any, Optional

class GenesysAuth:
    """Handles OAuth2 Client Credentials Flow for Genesys Cloud."""
    def __init__(self, client_id: str, client_secret: str, env_url: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.env_url = env_url
        self.token = None
        self.token_expiry = 0

    def get_token(self) -> str:
        if self.token and time.time() < self.token_expiry - 60:
            return self.token

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

        try:
            response = requests.post(url, headers=headers, data=data, timeout=10)
            response.raise_for_status()
            token_data = response.json()
            self.token = token_data["access_token"]
            self.token_expiry = time.time() + token_data["expires_in"]
            return self.token
        except requests.exceptions.RequestException as e:
            raise Exception(f"Failed to obtain OAuth token: {e}")

class GenesysWebChatService:
    """Service class for interacting with Genesys Cloud Web Chat APIs."""
    def __init__(self, auth: GenesysAuth):
        self.auth = auth
        self.base_url = auth.env_url

    def _get_headers(self) -> Dict[str, str]:
        return {
            "Authorization": f"Bearer {self.auth.get_token()}",
            "Accept": "application/json",
            "Content-Type": "application/json"
        }

    def get_active_conversations(self, page: int = 1, page_size: int = 25) -> List[Dict[str, Any]]:
        """
        Retrieves active web chat conversations.
        Scope: conversation:webchat:read
        """
        url = f"{self.base_url}/api/v2/conversations/webchat"
        params = {"page": page, "pageSize": page_size}

        try:
            response = requests.get(url, headers=self._get_headers(), params=params, timeout=10)
            response.raise_for_status()
            return response.json().get("entities", [])
        except requests.exceptions.HTTPError as e:
            if e.response.status_code == 401:
                print("Authentication failed. Check Client ID/Secret.")
            elif e.response.status_code == 403:
                print("Forbidden. Check OAuth scopes (conversation:webchat:read).")
            raise

    def send_proactive_message(
        self, 
        conversation_id: str, 
        customer_user_id: str, 
        sender_user_id: str, 
        text: str
    ) -> Dict[str, Any]:
        """
        Sends a proactive message to an active web chat session.
        Scope: conversation:webchat:write
        """
        url = f"{self.base_url}/api/v2/conversations/{conversation_id}/messages"
        
        payload = {
            "to": {"id": customer_user_id},
            "from": {"id": sender_user_id},
            "text": text,
            "type": "text"
        }

        try:
            response = requests.post(url, headers=self._get_headers(), json=payload, timeout=10)
            
            if response.status_code == 200 or response.status_code == 201:
                return {"success": True, "data": response.json()}
            elif response.status_code == 409:
                return {"success": False, "error": "Conflict", "message": "Conversation may be closed or invalid state."}
            elif response.status_code == 404:
                return {"success": False, "error": "Not Found", "message": "Conversation or User ID not found."}
            elif response.status_code == 429:
                return {"success": False, "error": "Rate Limited", "message": "Too many requests. Wait and retry."}
            else:
                response.raise_for_status()
                
        except requests.exceptions.RequestException as e:
            return {"success": False, "error": "Request Exception", "message": str(e)}

def main():
    # Load Configuration from Environment Variables
    client_id = os.getenv("GENESYS_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLIENT_SECRET")
    env_url = os.getenv("GENESYS_ENV_URL", "https://api.mypurecloud.com")
    bot_user_id = os.getenv("GENESYS_BOT_USER_ID")

    if not all([client_id, client_secret, bot_user_id]):
        raise EnvironmentError("Missing required environment variables: GENESYS_CLIENT_ID, GENESYS_CLIENT_SECRET, GENESYS_BOT_USER_ID")

    # Initialize Services
    auth = GenesysAuth(client_id, client_secret, env_url)
    webchat_service = GenesysWebChatService(auth)

    try:
        # Step 1: Get Active Sessions
        print("Retrieving active web chat sessions...")
        conversations = webchat_service.get_active_conversations()

        if not conversations:
            print("No active web chat sessions found.")
            return

        # Step 2: Process Each Session
        for conv in conversations:
            conv_id = conv["id"]
            participants = conv.get("participants", [])
            
            customer_id = None
            for p in participants:
                if p.get("role") == "customer":
                    customer_id = p.get("userId")
                    break
            
            if not customer_id:
                continue

            # Step 3: Send Proactive Message
            # In production, apply business logic here to decide IF to message
            message_text = "Hello! We see you are online. How can we assist you with your recent inquiry?"
            
            print(f"Sending proactive message to conversation {conv_id}...")
            result = webchat_service.send_proactive_message(
                conversation_id=conv_id,
                customer_user_id=customer_id,
                sender_user_id=bot_user_id,
                text=message_text
            )

            if result["success"]:
                print(f"Message sent successfully to {customer_id}.")
            else:
                print(f"Failed to send message: {result['error']} - {result.get('message')}")

            # Limit to one message for this demo
            break

    except Exception as e:
        print(f"Fatal error: {e}")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 401 Unauthorized

• Cause: The OAuth token is expired, invalid, or the Client ID/Secret is incorrect.
• Fix: Ensure the GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET environment variables are set correctly. Verify that the OAuth client in Genesys Cloud is enabled and has the correct grant type (Client Credentials).

Error: 403 Forbidden

• Cause: The OAuth client lacks the required scopes.
• Fix: Go to Genesys Cloud Admin > Security > OAuth Clients. Edit your client and ensure conversation:webchat:read and conversation:webchat:write are added to the scopes. Re-authenticate after adding scopes.

Error: 409 Conflict

• Cause: The conversation is no longer active, or the state is inconsistent.
• Fix: Verify that the conversation ID returned from the GET /api/v2/conversations/webchat endpoint is still active. Web chat sessions can close quickly if the user navigates away. Implement a retry mechanism with a small delay if this occurs frequently.

Error: 404 Not Found

• Cause: The conversationId or userId does not exist.
• Fix: Ensure you are using the userId from the participants array of the active conversation, not the conversationId itself. The to field in the message payload must match the exact userId of the customer participant.

Official References

• Genesys Cloud Web Chat API Documentation
• Genesys Cloud Messages API Documentation
• OAuth 2.0 Client Credentials Flow