Programmatically Close a Genesys Cloud Web Messaging Session

What You Will Build

• A backend service that identifies an active Web Messaging conversation and programmatically terminates the session.
• This tutorial uses the Genesys Cloud PureCloudPlatformClientV2 Python SDK and the underlying REST API for granular control.
• The programming language covered is Python 3.9+.

Prerequisites

• OAuth Client Type: A Client Credentials grant type is recommended for backend services. Alternatively, a Private Key grant type works for user-centric operations.
• Required Scopes:
  • webmessaging:conversation:view (to find the conversation)
  • webmessaging:conversation:update (to update the conversation state)
  • webmessaging:session:update (to close the specific client session)
• SDK Version: genesys-cloud-sdk-purecloud >= 11.0.0
• Runtime: Python 3.9 or later.
• Dependencies:

  pip install genesys-cloud-sdk-purecloud
  

Authentication Setup

Genesys Cloud uses OAuth 2.0 for all API access. For backend services, the Client Credentials flow is the standard pattern because it does not require a user context.

Below is the initialization code using the Python SDK. This handles the token acquisition and caching automatically.

import os
from purecloud_platform_client_v2 import PlatformClient

def init_platform_client() -> PlatformClient:
    """
    Initializes and returns a configured PlatformClient instance.
    """
    platform_client = PlatformClient()
    
    # Configure authentication using Client Credentials
    # Ensure these environment variables are set in your deployment
    auth = platform_client.auth
    auth.client_id = os.getenv("GENESYS_CLIENT_ID")
    auth.client_secret = os.getenv("GENESYS_CLIENT_SECRET")
    auth.environment = os.getenv("GENESYS_ENVIRONMENT", "mypurecloud.com")
    
    # Optional: Set a custom timeout if your network is slow
    # platform_client.set_request_timeout(10)
    
    return platform_client

Note on Token Refresh: The SDK manages token expiration internally. If a request fails with a 401 due to an expired token, the SDK automatically attempts to refresh the token and retries the original request. You do not need to implement manual refresh logic.

Implementation

Step 1: Identify the Active Conversation

To close a session, you must first locate the specific conversation ID. In Genesys Cloud, a “Conversation” is the server-side container for the interaction. A single conversation can contain multiple “Sessions” (e.g., if a customer reloads the page, a new session might be created within the same conversation, or if multiple agents are involved).

We will query the Conversation API to find the active Web Messaging conversation associated with a specific user identifier (such as an external ID or a custom attribute).

Endpoint: GET /api/v2/conversations
Scope: webmessaging:conversation:view

from purecloud_platform_client_v2 import ConversationApi
from purecloud_platform_client_v2.models import ConversationQueryRequest

def find_active_webmessaging_conversation(platform_client: PlatformClient, user_external_id: str) -> str | None:
    """
    Searches for an active Web Messaging conversation linked to a user external ID.
    
    Args:
        platform_client: The initialized PlatformClient.
        user_external_id: The external ID assigned to the customer in the messaging client.
        
    Returns:
        The conversation ID string, or None if no active conversation is found.
    """
    api_instance = ConversationApi(platform_client)
    
    # Construct the query body
    # We look for conversations with:
    # 1. Type 'webmessaging'
    # 2. State 'active' (or 'pending', depending on your workflow)
    # 3. A participant with the specific external ID
    query_body = ConversationQueryRequest(
        type="webmessaging",
        state="active",
        # Filter by participant external ID
        participants=[{
            "externalId": user_external_id
        }]
    )
    
    try:
        response = api_instance.post_conversations_query(body=query_body)
        
        # Check if any conversations were returned
        if response.conversations and len(response.conversations) > 0:
            # Return the first active conversation found
            return response.conversations[0].id
        else:
            print(f"No active web messaging conversation found for external ID: {user_external_id}")
            return None
            
    except Exception as e:
        print(f"Error querying conversations: {e}")
        return None

Step 2: Retrieve Session Details

Once you have the Conversation ID, you need to identify the specific Session ID to close. A conversation object contains a list of participants and sessions. For Web Messaging, the client session is what renders the UI. Closing the session triggers the client to disconnect.

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

from purecloud_platform_client_v2 import ConversationApi
from purecloud_platform_client_v2.models import WebMessagingConversation

def get_webmessaging_sessions(platform_client: PlatformClient, conversation_id: str) -> list:
    """
    Retrieves the list of sessions within a Web Messaging conversation.
    
    Args:
        platform_client: The initialized PlatformClient.
        conversation_id: The ID of the conversation.
        
    Returns:
        A list of session objects from the conversation.
    """
    api_instance = ConversationApi(platform_client)
    
    try:
        # Fetch the detailed Web Messaging conversation
        response: WebMessagingConversation = api_instance.get_conversations_webmessaging_conversation(
            conversation_id=conversation_id
        )
        
        # The sessions are nested within the conversation object
        # Note: In some SDK versions, sessions might be accessed via response.sessions
        if response.sessions:
            return response.sessions
        else:
            print("No sessions found in this conversation.")
            return []
            
    except Exception as e:
        print(f"Error retrieving conversation details: {e}")
        return []

Step 3: Close the Session Programmatically

To close the session, you must send a PATCH request to the Session endpoint. This updates the session state to closed. This action signals the Genesys Cloud Web Messaging client to terminate the connection and display the “session closed” UI state.

Endpoint: PATCH /api/v2/conversations/webmessaging/{conversationId}/sessions/{sessionId}
Scope: webmessaging:session:update

from purecloud_platform_client_v2 import ConversationApi
from purecloud_platform_client_v2.models import WebMessagingSessionUpdateRequest

def close_webmessaging_session(platform_client: PlatformClient, conversation_id: str, session_id: str) -> bool:
    """
    Closes a specific Web Messaging session.
    
    Args:
        platform_client: The initialized PlatformClient.
        conversation_id: The ID of the conversation.
        session_id: The ID of the session to close.
        
    Returns:
        True if the session was successfully closed, False otherwise.
    """
    api_instance = ConversationApi(platform_client)
    
    # Define the update request
    # Setting the state to 'closed' terminates the client connection
    update_body = WebMessagingSessionUpdateRequest(
        state="closed"
    )
    
    try:
        # Execute the PATCH request
        # The API does not return a body for this operation, but returns a 204 No Content on success
        api_instance.patch_conversations_webmessaging_conversation_session(
            conversation_id=conversation_id,
            session_id=session_id,
            body=update_body
        )
        print(f"Successfully closed session {session_id} in conversation {conversation_id}.")
        return True
        
    except Exception as e:
        # Handle specific HTTP errors
        if hasattr(e, 'status_code'):
            if e.status_code == 404:
                print(f"Session {session_id} not found in conversation {conversation_id}.")
            elif e.status_code == 409:
                print(f"Conflict: Session {session_id} may already be closed or in an invalid state.")
            else:
                print(f"HTTP Error {e.status_code}: {e.body}")
        else:
            print(f"Unexpected error closing session: {e}")
        return False

Complete Working Example

The following script combines all steps into a single executable module. It assumes you have set the environment variables GENESYS_CLIENT_ID, GENESYS_CLIENT_SECRET, and optionally GENESYS_ENVIRONMENT.

import os
import sys
from purecloud_platform_client_v2 import PlatformClient, ConversationApi
from purecloud_platform_client_v2.models import ConversationQueryRequest, WebMessagingSessionUpdateRequest, WebMessagingConversation

# --- Configuration ---
GENESYS_CLIENT_ID = os.getenv("GENESYS_CLIENT_ID")
GENESYS_CLIENT_SECRET = os.getenv("GENESYS_CLIENT_SECRET")
GENESYS_ENVIRONMENT = os.getenv("GENESYS_ENVIRONMENT", "mypurecloud.com")

def init_platform_client() -> PlatformClient:
    platform_client = PlatformClient()
    platform_client.auth.client_id = GENESYS_CLIENT_ID
    platform_client.auth.client_secret = GENESYS_CLIENT_SECRET
    platform_client.auth.environment = GENESYS_ENVIRONMENT
    return platform_client

def find_conversation_by_external_id(platform_client: PlatformClient, external_id: str) -> str | None:
    api_instance = ConversationApi(platform_client)
    query_body = ConversationQueryRequest(
        type="webmessaging",
        state="active",
        participants=[{"externalId": external_id}]
    )
    try:
        response = api_instance.post_conversations_query(body=query_body)
        if response.conversations:
            return response.conversations[0].id
    except Exception as e:
        print(f"Query failed: {e}")
    return None

def close_session_for_conversation(platform_client: PlatformClient, conversation_id: str, external_id: str) -> bool:
    api_instance = ConversationApi(platform_client)
    
    # 1. Get Conversation Details to find Session ID
    try:
        conv_response: WebMessagingConversation = api_instance.get_conversations_webmessaging_conversation(conversation_id)
    except Exception as e:
        print(f"Failed to get conversation details: {e}")
        return False

    if not conv_response.sessions:
        print("No sessions found in conversation.")
        return False

    # 2. Find the session matching the external ID
    # Note: In Web Messaging, the session is often tied to the participant with the external ID
    target_session_id = None
    for session in conv_response.sessions:
        # Check if this session belongs to the participant with our external ID
        # This logic depends on how your client sets the external ID. 
        # Usually, you match the session ID to the participant's session ID.
        # For simplicity, we assume the first active session belongs to the user if we filtered by external ID earlier.
        # A more robust check involves iterating participants and matching session IDs.
        if session.state != "closed":
            target_session_id = session.id
            break
            
    if not target_session_id:
        print("No active session found to close.")
        return False

    # 3. Close the Session
    update_body = WebMessagingSessionUpdateRequest(state="closed")
    try:
        api_instance.patch_conversations_webmessaging_conversation_session(
            conversation_id=conversation_id,
            session_id=target_session_id,
            body=update_body
        )
        return True
    except Exception as e:
        print(f"Failed to close session: {e}")
        return False

def main():
    if not GENESYS_CLIENT_ID or not GENESYS_CLIENT_SECRET:
        print("Error: GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET must be set.")
        sys.exit(1)

    # Example External ID provided via command line argument
    if len(sys.argv) < 2:
        print("Usage: python close_webmessaging_session.py <user_external_id>")
        sys.exit(1)
        
    user_external_id = sys.argv[1]
    
    print(f"Attempting to close Web Messaging session for external ID: {user_external_id}")
    
    platform_client = init_platform_client()
    
    # Step 1: Find Conversation
    conversation_id = find_conversation_by_external_id(platform_client, user_external_id)
    
    if not conversation_id:
        print("No active conversation found. Nothing to close.")
        return

    print(f"Found active conversation: {conversation_id}")
    
    # Step 2 & 3: Close Session
    success = close_session_for_conversation(platform_client, conversation_id, user_external_id)
    
    if success:
        print("Session closed successfully.")
    else:
        print("Failed to close session.")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 403 Forbidden

Cause: The OAuth token does not have the required scopes.
Fix: Ensure your OAuth client in the Genesys Cloud Admin portal has the webmessaging:conversation:update and webmessaging:session:update scopes enabled. If you are using Client Credentials, verify that the application has the necessary permissions assigned to it.

Error: 404 Not Found

Cause: The conversation ID or session ID does not exist, or the conversation has already ended/archived.
Fix:

1. Verify the conversation ID returned from the search step is correct.
2. Check if the conversation state is still active. If the conversation has naturally timed out or was manually ended by an agent, the session may no longer be addressable.
3. Ensure the externalId used in the query exactly matches the one set in the Web Messaging client JavaScript initialization.

Error: 409 Conflict

Cause: The session is already in a closed state.
Fix: Add a check in your code to verify session.state before attempting the PATCH request. If the state is already closed, skip the API call to avoid unnecessary errors.

Error: 429 Too Many Requests

Cause: You have exceeded the API rate limits for your organization.
Fix: Implement exponential backoff in your retry logic. The Python SDK does not automatically retry 429s by default. You can wrap the API calls in a retry decorator.

import time
from functools import wraps

def retry_on_429(max_retries=3, base_delay=1):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if hasattr(e, 'status_code') and e.status_code == 429:
                        delay = base_delay * (2 ** attempt)
                        print(f"Rate limited. Retrying in {delay} seconds...")
                        time.sleep(delay)
                    else:
                        raise
            raise Exception("Max retries exceeded for 429 error.")
        return wrapper
    return decorator

# Usage:
# @retry_on_429
# def close_webmessaging_session(...): ...

Official References

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