Programmatically Close Web Messaging Sessions from the Backend
What You Will Build
- A backend service that terminates active Web Messaging conversations by transitioning them to the “closed” state.
- This solution uses the Genesys Cloud CX Platform API (
/api/v2/conversations/webchat). - The implementation is provided in Python using the
requestslibrary for direct HTTP interaction.
Prerequisites
- OAuth Client Type: Service Account (Client Credentials Flow).
- Required Scopes:
conversation:writeis mandatory to update conversation states.conversation:readis helpful for debugging. - API Version: Genesys Cloud CX REST API v2.
- Language/Runtime: Python 3.8+.
- External Dependencies:
requests: For HTTP communication.python-dotenv: For managing environment variables securely.
Authentication Setup
Genesys Cloud CX uses OAuth 2.0 for API authentication. For backend services, the Client Credentials flow is the standard pattern. This flow exchanges a client ID and secret for an access token valid for one hour.
The following Python class handles token acquisition and caching. In production, you should implement a thread-safe cache to avoid requesting a new token on every API call.
import os
import time
import requests
from typing import Optional
class GenesysAuth:
def __init__(self):
self.client_id = os.getenv("GENESYS_CLIENT_ID")
self.client_secret = os.getenv("GENESYS_CLIENT_SECRET")
self.environment = os.getenv("GENESYS_ENVIRONMENT", "mypurecloud.com")
self.token_url = f"https://login.{self.environment}/oauth/token"
self.access_token: Optional[str] = None
self.token_expiry: float = 0
def get_access_token(self) -> str:
"""
Returns a valid OAuth access token.
Handles caching to prevent unnecessary requests.
"""
# Check if current token is still valid (subtract 60s buffer)
if self.access_token and time.time() < (self.token_expiry - 60):
return self.access_token
# Request new 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(self.token_url, headers=headers, data=data)
response.raise_for_status()
token_data = response.json()
self.access_token = token_data["access_token"]
# Cache expiry time
self.token_expiry = time.time() + token_data["expires_in"]
return self.access_token
except requests.exceptions.HTTPError as e:
if response.status_code == 401:
raise Exception("Invalid Client ID or Secret.") from e
raise Exception(f"Authentication failed: {response.text}") from e
def get_auth_header(self) -> dict:
"""
Returns the Authorization header dict for API calls.
"""
token = self.get_access_token()
return {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json"
}
Implementation
Step 1: Identify the Active Conversation
To close a session, you must possess the conversationId. This ID is typically generated when the Web Messaging widget initiates a conversation or passed to your backend via a webhook.
If you do not have the ID, you can query active conversations. However, for closing a specific session, we assume the ID is available. The Genesys API distinguishes between different media types. Web Messaging uses the webchat media type.
Endpoint: GET /api/v2/conversations/webchat/{conversationId}
This step is optional if you already possess the ID, but it is critical for validation. Attempting to close a non-existent or already closed conversation will result in a 404 or 409 error.
import requests
class GenesysMessenger:
def __init__(self, auth: GenesysAuth):
self.auth = auth
self.base_url = f"https://api.{auth.environment}"
def validate_conversation(self, conversation_id: str) -> dict:
"""
Fetches conversation details to ensure it exists and is active.
"""
url = f"{self.base_url}/api/v2/conversations/webchat/{conversation_id}"
headers = self.auth.get_auth_header()
try:
response = requests.get(url, headers=headers)
response.raise_for_status()
return response.json()
except requests.exceptions.HTTPError as e:
if response.status_code == 404:
raise ValueError(f"Conversation {conversation_id} not found.") from e
raise Exception(f"Failed to fetch conversation: {response.text}") from e
Step 2: Transition the Conversation to Closed
The core logic involves sending a PATCH request to the Web Messaging conversation endpoint. The Genesys API uses a specific state machine for conversations. To close a session, you must transition the state field from active to closed.
Endpoint: PATCH /api/v2/conversations/webchat/{conversationId}
OAuth Scope: conversation:write
Request Body:
{
"state": "closed"
}
It is important to note that you cannot close a conversation that is in the queued state. It must first be routed to an agent or bot. If the conversation is active, you may close it directly. If it is ringing (waiting for agent answer), you must first answer it or let it ring out, depending on your business logic. For this tutorial, we assume the conversation is active.
import json
import logging
logger = logging.getLogger(__name__)
class GenesysMessenger:
# ... (previous code) ...
def close_conversation(self, conversation_id: str, reason: str = "Automated Closure") -> dict:
"""
Transitions the Web Messaging conversation to the 'closed' state.
Args:
conversation_id: The unique ID of the conversation.
reason: Optional reason for closing, stored in analytics.
Returns:
The updated conversation object.
"""
url = f"{self.base_url}/api/v2/conversations/webchat/{conversation_id}"
headers = self.auth.get_auth_header()
payload = {
"state": "closed"
}
# Optional: Add a transcript note or custom attribute if needed
# payload["wrapUpCode"] = "resolved"
try:
response = requests.patch(url, headers=headers, json=payload)
# Handle specific status codes
if response.status_code == 200:
logger.info(f"Successfully closed conversation {conversation_id}")
return response.json()
elif response.status_code == 409:
# Conflict: Conversation is not in a state that allows closing
raise Exception(f"Cannot close conversation {conversation_id}. Current state may not allow closure.") from response
elif response.status_code == 404:
raise Exception(f"Conversation {conversation_id} not found.") from response
else:
response.raise_for_status()
except requests.exceptions.RequestException as e:
logger.error(f"Error closing conversation {conversation_id}: {e}")
raise
Step 3: Handling Edge Cases and Retries
Web Messaging APIs can occasionally return 429 Too Many Requests if you are closing many sessions rapidly. Implementing exponential backoff is best practice.
Additionally, the API is eventually consistent. If you close a session and immediately query it, it might still appear as active for a few hundred milliseconds. If your business logic requires immediate confirmation, you should poll the GET endpoint briefly.
import time
import random
class GenesysMessenger:
# ... (previous code) ...
def close_conversation_with_retry(self, conversation_id: str, max_retries: int = 3) -> dict:
"""
Attempts to close the conversation with exponential backoff on 429 errors.
"""
for attempt in range(max_retries):
try:
return self.close_conversation(conversation_id)
except requests.exceptions.HTTPError as e:
if e.response.status_code == 429:
# Exponential backoff: 1s, 2s, 4s... plus jitter
wait_time = (2 ** attempt) + random.uniform(0, 1)
logger.warning(f"Rate limited. Retrying in {wait_time:.2f} seconds...")
time.sleep(wait_time)
continue
else:
raise
except Exception as e:
# Non-retryable errors should fail immediately
raise
raise Exception(f"Failed to close conversation {conversation_id} after {max_retries} attempts.")
Complete Working Example
This script demonstrates the full lifecycle: authentication, validation, and closure.
import os
import sys
import logging
import requests
from typing import Optional
# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# --- Authentication Class ---
class GenesysAuth:
def __init__(self):
self.client_id = os.getenv("GENESYS_CLIENT_ID")
self.client_secret = os.getenv("GENESYS_CLIENT_SECRET")
self.environment = os.getenv("GENESYS_ENVIRONMENT", "mypurecloud.com")
self.token_url = f"https://login.{self.environment}/oauth/token"
self.access_token: Optional[str] = None
self.token_expiry: float = 0
def get_access_token(self) -> str:
import time
if self.access_token and time.time() < (self.token_expiry - 60):
return self.access_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(self.token_url, headers=headers, data=data)
response.raise_for_status()
token_data = response.json()
self.access_token = token_data["access_token"]
self.token_expiry = time.time() + token_data["expires_in"]
return self.access_token
except requests.exceptions.HTTPError as e:
raise Exception(f"Authentication failed: {response.text}") from e
def get_auth_header(self) -> dict:
token = self.get_access_token()
return {
"Authorization": f"Bearer {token}",
"Content-Type": "application/json"
}
# --- Messenger Logic Class ---
class GenesysMessenger:
def __init__(self, auth: GenesysAuth):
self.auth = auth
self.base_url = f"https://api.{auth.environment}"
def close_conversation(self, conversation_id: str) -> dict:
url = f"{self.base_url}/api/v2/conversations/webchat/{conversation_id}"
headers = self.auth.get_auth_header()
payload = {
"state": "closed"
}
try:
response = requests.patch(url, headers=headers, json=payload)
response.raise_for_status()
logger.info(f"Conversation {conversation_id} closed successfully.")
return response.json()
except requests.exceptions.HTTPError as e:
if e.response.status_code == 409:
logger.error(f"Conflict: Conversation {conversation_id} cannot be closed. Check current state.")
elif e.response.status_code == 404:
logger.error(f"Not Found: Conversation {conversation_id} does not exist.")
else:
logger.error(f"HTTP Error: {e.response.status_code} - {e.response.text}")
raise
# --- Main Execution ---
def main():
# 1. Initialize Authentication
try:
auth = GenesysAuth()
# Trigger a token fetch to validate credentials early
auth.get_access_token()
logger.info("Authentication successful.")
except Exception as e:
logger.error(f"Failed to initialize authentication: {e}")
sys.exit(1)
# 2. Initialize Messenger
messenger = GenesysMessenger(auth)
# 3. Define Conversation ID
# In a real scenario, this comes from a webhook or database
conversation_id = os.getenv("CONVERSATION_ID")
if not conversation_id:
logger.error("CONVERSATION_ID environment variable is not set.")
sys.exit(1)
# 4. Close the Session
try:
result = messenger.close_conversation(conversation_id)
logger.info("Final State:", result.get("state"))
except Exception as e:
logger.error(f"Operation failed: {e}")
sys.exit(1)
if __name__ == "__main__":
main()
Common Errors & Debugging
Error: 409 Conflict
What causes it: The conversation is not in a state that allows closure. Common states that block closure include queued (waiting for agent) or ringing (alerting agent).
How to fix it: Check the current state using the GET /api/v2/conversations/webchat/{id} endpoint. If the state is queued, you must first route the conversation to an agent or abandon it. If ringing, you must answer it.
Code Fix:
# Check state before closing
current_state = validate_conversation(conversation_id)["state"]
if current_state == "queued":
# Logic to abandon or route
pass
elif current_state == "active":
close_conversation(conversation_id)
Error: 403 Forbidden
What causes it: The OAuth token does not have the conversation:write scope.
How to fix it: Go to the Genesys Cloud Admin Portal, navigate to Admin > Security > Integrations, find your service account, and ensure conversation:write is checked in the API scopes.
Code Fix: Regenerate the token after updating scopes.
Error: 404 Not Found
What causes it: The conversationId is invalid or the conversation has already been deleted/archived.
How to fix it: Verify the ID format. Web Messaging IDs are typically UUIDs. Ensure the ID was copied from the correct environment (Production vs. Sandbox).