Sending proactive notifications to a customer who previously had a web messaging session

What You Will Build

• You will build a Python script that identifies a customer based on their previous web messaging session ID and sends a proactive outbound message to their browser.
• This tutorial uses the Genesys Cloud CX Messaging API (/api/v2/messaging/messages) and the Conversations Search API to locate the target user.
• The implementation is written in Python 3.9+ using the requests library and httpx for asynchronous operations where appropriate.

Prerequisites

• OAuth Client: A Genesys Cloud OAuth client with the messaging:outbound:send and analytics:conversation:search scopes.
• SDK Version: This tutorial uses the REST API directly via requests for maximum transparency, but the logic applies to the PureCloudPlatformClientV2 Python SDK (v130.0.0+).
• Runtime: Python 3.9 or higher.
• Dependencies:
  • requests (for HTTP calls)
  • pydantic (for data validation, optional but recommended)
  • uuid (standard library, for generating message IDs)

Authentication Setup

Genesys Cloud uses OAuth 2.0 for authentication. For server-to-server integrations (like this proactive notification service), you will use the Client Credentials grant type.

You must store your Client ID and Client Secret securely. Never hardcode these in your application. Use environment variables or a secrets manager.

Step 1: Obtain an Access Token

The following function handles the token retrieval. It caches the token to avoid unnecessary API calls, as tokens are valid for 3600 seconds.

import os
import time
import requests
from typing import Optional

class GenesysAuth:
    def __init__(self, client_id: str, client_secret: str, env: str = "mypurecloud.com"):
        self.client_id = client_id
        self.client_secret = client_secret
        self.env = env
        self.access_token: Optional[str] = None
        self.token_expiry: float = 0
        self.base_url = f"https://api.{env}"

    def get_token(self) -> str:
        """
        Retrieves an OAuth access token.
        Returns the token string.
        Raises ValueError if authentication fails.
        """
        # Check if we have a valid cached token
        if self.access_token and time.time() < self.token_expiry:
            return self.access_token

        token_url = f"{self.base_url}/oauth/token"
        payload = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret
        }

        headers = {
            "Content-Type": "application/json"
        }

        response = requests.post(token_url, json=payload, headers=headers)

        if response.status_code != 200:
            raise ValueError(f"Authentication failed: {response.status_code} - {response.text}")

        token_data = response.json()
        self.access_token = token_data["access_token"]
        self.token_expiry = time.time() + token_data["expires_in"] - 60  # Buffer of 60 seconds

        return self.access_token

    def get_headers(self) -> dict:
        """Returns headers with the current Bearer token."""
        return {
            "Authorization": f"Bearer {self.get_token()}",
            "Content-Type": "application/json"
        }

Implementation

Step 1: Locate the Customer’s Session

To send a proactive message, you must target a specific user session. Genesys Cloud Web Messaging sessions are identified by a sessionId (often stored as a cookie on the client side) or by linking the conversation to a specific userId if the user was authenticated.

In most proactive scenarios, you do not have the active sessionId because the user is currently offline or idle. However, if you are targeting a user who is currently on your site but idle, or if you have stored their userId from a previous authenticated session, you can search for their recent conversations.

If you are sending a message to a user who is not currently online, Web Messaging cannot deliver it directly to the browser. You must use Outbound Messaging to a phone number or email, or use the Engagement API to push a notification to a mobile app.

Note: True “proactive web messaging” to a browser requires the browser to have an open WebSocket connection. If the page is closed, the connection is dead. This tutorial assumes you are sending an outbound message to a user who has previously opted in, using their contact information linked to their previous web session.

For this tutorial, we will assume you have the userId or email of the customer from your CRM, and you want to send them a message via the Outbound Messaging API, which can trigger a web message if they are online, or fall back to SMS/Email depending on your routing configuration.

However, the most robust way to “proactively” message a web user who is currently idle but connected is to use the Conversations API to find their active conversation and inject a message.

Let us focus on the scenario where you have the Conversation ID or Session ID from a previous interaction, and you want to send a message to that specific channel.

Critical Constraint: You cannot send a message to a closed conversation. You must verify the conversation is still open or initiate a new outbound engagement.

For this tutorial, we will use the Outbound Messaging API to send a message to a user identified by their email, which is linked to their previous web session. This is the standard pattern for “re-engagement.”

Step 2: Construct the Outbound Message Payload

The endpoint /api/v2/messaging/messages accepts a JSON payload. The structure depends on the channel. For web messaging re-engagement, we typically use the email or sms channel if the web socket is closed, or we use the userId if we are pushing to a registered client.

To send a message to a user via their previous web channel, you must use the Engagement API or Outbound Messaging with the web channel type, targeting the userId.

Here is the payload structure for an outbound web message:

{
  "type": "web",
  "to": {
    "userId": "string",
    "email": "string"
  },
  "from": {
    "userId": "string"
  },
  "body": {
    "text": "Hello, we noticed you were looking at our pricing page. Do you have any questions?"
  }
}

Step 3: Send the Message via API

We will create a function that sends this payload.

import json
from typing import Dict, Any

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

    def send_proactive_web_message(self, user_id: str, email: str, from_user_id: str, message_text: str) -> Dict[str, Any]:
        """
        Sends a proactive outbound web message to a user.
        
        Args:
            user_id: The Genesys Cloud userId of the recipient.
            email: The email address of the recipient (for fallback).
            from_user_id: The Genesys Cloud userId of the sender (agent or bot).
            message_text: The content of the message.
            
        Returns:
            The API response dictionary.
        """
        url = f"{self.base_url}/api/v2/messaging/messages"
        
        payload = {
            "type": "web",
            "to": {
                "userId": user_id,
                "email": email
            },
            "from": {
                "userId": from_user_id
            },
            "body": {
                "text": message_text
            }
        }

        headers = self.auth.get_headers()
        
        try:
            response = requests.post(url, json=payload, headers=headers)
            
            # Handle specific errors
            if response.status_code == 401:
                raise Exception("Authentication failed. Token may be expired.")
            elif response.status_code == 403:
                raise Exception("Forbidden. Check OAuth scopes: messaging:outbound:send")
            elif response.status_code == 429:
                raise Exception("Rate limit exceeded. Implement retry logic.")
            elif response.status_code >= 400:
                raise Exception(f"API Error: {response.status_code} - {response.text}")
            
            return response.json()
            
        except requests.exceptions.RequestException as e:
            raise Exception(f"Network error: {str(e)}")

Step 4: Handling Rate Limits and Retries

Genesys Cloud APIs enforce rate limits. A 429 status code indicates you have exceeded the limit. You must implement exponential backoff.

import time
import random

def send_with_retry(messaging: GenesysMessaging, user_id: str, email: str, from_user_id: str, message_text: str, max_retries: int = 3) -> Dict[str, Any]:
    """
    Sends a message with exponential backoff retry logic for 429 errors.
    """
    attempt = 0
    while attempt < max_retries:
        try:
            return messaging.send_proactive_web_message(user_id, email, from_user_id, message_text)
        except Exception as e:
            if "429" in str(e):
                wait_time = (2 ** attempt) + random.uniform(0, 1)
                print(f"Rate limited. Retrying in {wait_time:.2f} seconds...")
                time.sleep(wait_time)
                attempt += 1
            else:
                raise e
    raise Exception("Max retries exceeded for 429 error.")

Complete Working Example

This script ties together authentication, message construction, and sending with error handling.

import os
import sys
import json
import time
import requests
from typing import Dict, Any, Optional

# --- Configuration ---
CLIENT_ID = os.environ.get("GENESYS_CLIENT_ID")
CLIENT_SECRET = os.environ.get("GENESYS_CLIENT_SECRET")
ENVIRONMENT = os.environ.get("GENESYS_ENV", "mypurecloud.com")

# --- Authentication Class ---
class GenesysAuth:
    def __init__(self, client_id: str, client_secret: str, env: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.env = env
        self.access_token: Optional[str] = None
        self.token_expiry: float = 0
        self.base_url = f"https://api.{env}"

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

        token_url = f"{self.base_url}/oauth/token"
        payload = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret
        }

        headers = {"Content-Type": "application/json"}
        response = requests.post(token_url, json=payload, headers=headers)

        if response.status_code != 200:
            raise ValueError(f"Auth failed: {response.status_code} - {response.text}")

        data = response.json()
        self.access_token = data["access_token"]
        self.token_expiry = time.time() + data["expires_in"] - 60
        return self.access_token

    def get_headers(self) -> dict:
        return {
            "Authorization": f"Bearer {self.get_token()}",
            "Content-Type": "application/json"
        }

# --- Messaging Service Class ---
class ProactiveMessagingService:
    def __init__(self, auth: GenesysAuth):
        self.auth = auth
        self.base_url = auth.base_url

    def send_message(self, user_id: str, email: str, from_user_id: str, text: str) -> Dict[str, Any]:
        url = f"{self.base_url}/api/v2/messaging/messages"
        
        payload = {
            "type": "web",
            "to": {
                "userId": user_id,
                "email": email
            },
            "from": {
                "userId": from_user_id
            },
            "body": {
                "text": text
            }
        }

        response = requests.post(url, json=payload, headers=self.auth.get_headers())
        
        if response.status_code == 200:
            return {"success": True, "data": response.json()}
        elif response.status_code == 429:
            raise Exception("Rate Limit 429")
        else:
            raise Exception(f"Failed: {response.status_code} - {response.text}")

# --- Main Execution ---
def main():
    if not CLIENT_ID or not CLIENT_SECRET:
        print("Error: Set GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET environment variables.")
        sys.exit(1)

    auth = GenesysAuth(CLIENT_ID, CLIENT_SECRET, ENVIRONMENT)
    service = ProactiveMessagingService(auth)

    # Example Usage
    target_user_id = "12345678-1234-1234-1234-123456789012"  # Replace with actual userId
    target_email = "customer@example.com"
    sender_user_id = "87654321-4321-4321-4321-210987654321"  # Replace with actual agent/bot userId
    message_content = "Hi there! We noticed you left an item in your cart. Would you like help completing your purchase?"

    try:
        result = service.send_message(target_user_id, target_email, sender_user_id, message_content)
        print("Message sent successfully:")
        print(json.dumps(result, indent=2))
    except Exception as e:
        print(f"Error sending message: {e}")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 401 Unauthorized

• Cause: The OAuth token is expired, invalid, or the client credentials are incorrect.
• Fix: Ensure GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET are correct. Verify the token caching logic in GenesysAuth is not returning a stale token.

Error: 403 Forbidden

• Cause: The OAuth client lacks the required scope messaging:outbound:send.
• Fix: Go to Genesys Cloud Admin > Security > OAuth. Edit your client application. Add messaging:outbound:send to the scopes. If you are using analytics:conversation:search, add that scope as well.

Error: 422 Unprocessable Entity

• Cause: The payload structure is invalid. For example, missing userId in the to object or invalid type.
• Fix: Validate the JSON payload against the Genesys Cloud API specification. Ensure userId is a valid UUID string.

Error: 429 Too Many Requests

• Cause: You have exceeded the rate limit for the messaging API.
• Fix: Implement exponential backoff as shown in the send_with_retry function. Reduce the frequency of proactive messages.

Official References

• Genesys Cloud Messaging API Documentation
• Genesys Cloud OAuth 2.0 Guide
• Python SDK Reference

Check your payload structure. The 409 Conflict means the server sees an active session for that webchatSessionId and rejects the duplicate initiation request. You must reuse the existing conversationId instead of creating a new one.

If I recall correctly, the 409 Conflict is not a bug but a strict state enforcement by the Web Messaging service. You are likely hitting the endpoint that initiates a new session rather than injecting a message into an existing conversation. The suggestion above is correct, but you need to be precise about which API call to swap.

Here is the correct Laravel/Guzzle implementation to resume an existing session and send a message:

// Use the existing conversationId, not webchatSessionId for the message body
$resumeUrl = "https://api.mypurecloud.com/api/v2/conversations/webchat";
$messagePayload = [
 'from' => [
 'id' => $existingContactId, 
 'type' => 'contact'
 ],
 'text' => 'Proactive notification content...',
 'type' => 'message'
];

try {
 $response = $client->post($resumeUrl, [
 'json' => $messagePayload,
 'headers' => [
 'Authorization' => 'Bearer ' . $accessToken,
 'Content-Type' => 'application/json',
 'X-Genesys-Client-Id' => 'your_client_id',
 'X-Genesys-Client-Version' => '2.0' // Ensure version match
 ]
 ]);
} catch (\GuzzleHttp\Exception\ClientException $e) {
 // Log specific 409 details for debugging session state
 error_log($e->getResponse()->getBody()->getContents());
}

The key is ensuring your webchatSessionId matches the active session in Genesys Cloud’s memory. If that ID has expired or been invalidated by a timeout, you cannot resume. In my Vue dashboard projects, I handle this by checking the WebSocket connection state before attempting the POST. If the client-side session is stale, I force a re-initialization instead of trying to push into a dead conversation. Also, verify you have the webchat:write scope on your OAuth token, as missing scopes often throw confusing 400/409 errors in Guzzle.

It depends, but generally… the 409 is expected behavior when the client attempts to instantiate a new session object while a live webchatSessionId persists in the Genesys Cloud state store. I have traced similar failures in Node.js handlers where the token lacked messaging:conversation:write, causing the API to reject the duplicate initiation attempt as a conflict rather than a permission error. To resolve this, you must stop calling the session creation endpoint and instead route your Laravel Guzzle request to /api/v2/conversations/messaging/{conversationId}/messages. Ensure your OAuth token includes the messaging:conversation:write scope; without it, the server will mask the scope mismatch with a 409. I recommend adding a pre-flight check to validate the session status via GET /api/v2/conversations/messaging/{conversationId} before attempting the POST. This prevents redundant writes and aligns with least-privilege scope design principles. If the session is closed, you can initiate a new one, but for active sessions, strictly reuse the existing conversationId.

My usual workaround is to routing the conflict to an EventBridge rule instead of handling it in the API layer. The 409 means the session is active. Use aws-sdk/client-eventbridge to trigger a Step Function that calls /api/v2/conversations/messaging/{conversationId}/messages. This keeps your Lambda stateless.

const rule = new Rule(this, 'ProactiveMsgRule', {
 eventPattern: { source: ['com.genesys.cloud'] }
});