How to Programmatically Close a Web Messaging Session from the Backend
What You Will Build
- This code demonstrates how to terminate an active Web Messaging conversation from a backend service when the customer is no longer present.
- This tutorial uses the Genesys Cloud CX API v2 and the Python SDK (
genesyscloud). - The implementation is provided in Python 3.9+, utilizing the
requestslibrary for HTTP interaction and the official SDK for structured operations.
Prerequisites
- OAuth Client Type: Service Account or Confidential Client.
- Required Scopes:
conversations:close(Required to close the conversation)conversations:read(Optional, required if you need to fetch conversation details before closing)
- SDK Version:
genesyscloud>= 140.0.0 (Python SDK). - Runtime: Python 3.9 or higher.
- Dependencies:
genesyscloudrequests
pip install genesyscloud requests
Authentication Setup
Genesys Cloud uses OAuth 2.0. For backend services, the Client Credentials flow is the standard. The following code initializes the authentication client and retrieves an access token. In production, you should cache this token and refresh it before expiration, but for this tutorial, we will fetch a fresh token each time to ensure the code is self-contained.
import os
from genesyscloud.auth.api_client import ApiClient
from genesyscloud.auth.client_credentials_client import ClientCredentialsClient
def get_access_token():
"""
Retrieves an OAuth access token using Client Credentials flow.
"""
# Environment variables should hold these secrets
client_id = os.getenv("GENESYS_CLIENT_ID")
client_secret = os.getenv("GENESYS_CLIENT_SECRET")
if not client_id or not client_secret:
raise ValueError("GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET must be set in environment.")
# Initialize the API client with the Genesys Cloud environment
# Default is 'mypurecloud.com', change if using a different environment (e.g., 'au-pure.cloud')
api_client = ApiClient(base_url="https://api.mypurecloud.com")
# Create the credentials client
credentials_client = ClientCredentialsClient(
api_client=api_client,
client_id=client_id,
client_secret=client_secret
)
# Fetch the token
token = credentials_client.get_access_token()
return token.access_token
# Example usage
try:
token = get_access_token()
print(f"Token acquired successfully. Expires in: {token.expires_in} seconds")
except Exception as e:
print(f"Authentication failed: {e}")
exit(1)
Implementation
Step 1: Identify the Target Conversation
Before closing a session, you must identify the specific conversationId. In a Web Messaging context, this ID is generated when the widget initiates contact. If you do not have the ID stored in your external database, you must query the Analytics or Conversations API to find it based on a unique identifier (such as the user’s email or a custom attribute).
For this tutorial, we assume the conversationId is known or passed into the function. If you need to search for it, you would use GET /api/v2/analytics/conversations/details/query.
Step 2: Construct the Close Payload
To close a conversation, you must send a POST request to /api/v2/conversations/{conversationId}/close. The body of this request must contain a wrapUpCode. Even if you do not use wrap-up codes in your queue configuration, Genesys Cloud requires a valid ID for the close operation.
The most common default wrap-up code ID is 99999999-9999-9999-9999-999999999999 (often labeled as “No Wrap-up Code” or “Default”). However, it is safer to retrieve the actual default wrap-up code ID from your organization to avoid 400 Bad Request errors.
Here is how to retrieve the default wrap-up code ID using the SDK:
from genesyscloud.wrap_up_codes.api import Wrap_up_codesApi
from genesyscloud.wrap_up_codes.model import GetWrapupcodesbody
def get_default_wrapup_code_id(api_client):
"""
Retrieves the ID of the default wrap-up code.
"""
wrapup_codes_api = Wrap_up_codesApi(api_client)
# Fetch all wrap-up codes
# Note: In a high-volume environment, cache this result.
response = wrapup_codes_api.post_wrap_up_codes(
body=GetWrapupcodesbody(
page_size=25,
expand=["default"]
)
)
# Find the code marked as default
default_code = None
if response.entities:
for code in response.entities:
if code.default is True:
default_code = code.id
break
if not default_code:
# Fallback to a known default ID if none is explicitly marked, though rare
# This is a common fallback ID in many Genesys orgs
return "99999999-9999-9999-9999-999999999999"
return default_code
Step 3: Execute the Close Operation
With the conversationId and the wrapUpCodeId, you can now close the conversation. The conversation object in Genesys Cloud has a lifecycle. A Web Messaging conversation is typically in the active or queued state. Closing it moves it to the closed state.
It is critical to handle the 409 Conflict error. This occurs if the conversation is already closed or if another agent/system has already closed it.
from genesyscloud.conversations.api import ConversationsApi
from genesyscloud.conversations.model import PostConversationClosebody
import requests
def close_web_messaging_session(conversation_id: str, api_client: ApiClient, wrapup_code_id: str):
"""
Closes a specific Web Messaging conversation.
Args:
conversation_id: The UUID of the conversation to close.
api_client: The authenticated ApiClient instance.
wrapup_code_id: The UUID of the wrap-up code to apply.
"""
conversations_api = ConversationsApi(api_client)
# Construct the close body
close_body = PostConversationClosebody(
wrap_up_code_id=wrapup_code_id
)
try:
# Execute the close
# The API returns a 204 No Content on success
conversations_api.post_conversation_close(
conversation_id=conversation_id,
body=close_body
)
print(f"Successfully closed conversation: {conversation_id}")
return True
except requests.exceptions.HTTPError as err:
status_code = err.response.status_code
if status_code == 409:
# Conflict: Already closed or locked
print(f"Conversation {conversation_id} is already closed or locked.")
return False
elif status_code == 404:
# Not Found: Conversation ID does not exist
print(f"Conversation {conversation_id} not found.")
return False
elif status_code == 401 or status_code == 403:
# Auth Error: Token invalid or missing scope
print(f"Authentication/Authorization error: {err.response.text}")
raise
else:
# Other errors
print(f"Unexpected error closing conversation: {err.response.text}")
raise
except Exception as e:
print(f"An unexpected error occurred: {e}")
raise
Complete Working Example
This script combines authentication, wrap-up code retrieval, and the close operation into a single executable flow. It assumes you have a conversation_id ready.
import os
import sys
import requests
from genesyscloud.auth.api_client import ApiClient
from genesyscloud.auth.client_credentials_client import ClientCredentialsClient
from genesyscloud.wrap_up_codes.api import Wrap_up_codesApi
from genesyscloud.wrap_up_codes.model import GetWrapupcodesbody
from genesyscloud.conversations.api import ConversationsApi
from genesyscloud.conversations.model import PostConversationClosebody
def get_access_token():
"""Retrieves an OAuth access token using Client Credentials flow."""
client_id = os.getenv("GENESYS_CLIENT_ID")
client_secret = os.getenv("GENESYS_CLIENT_SECRET")
if not client_id or not client_secret:
raise ValueError("GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET must be set.")
api_client = ApiClient(base_url="https://api.mypurecloud.com")
credentials_client = ClientCredentialsClient(
api_client=api_client,
client_id=client_id,
client_secret=client_secret
)
token = credentials_client.get_access_token()
return token.access_token, api_client
def get_default_wrapup_code_id(api_client: ApiClient):
"""Retrieves the ID of the default wrap-up code."""
wrapup_codes_api = Wrap_up_codesApi(api_client)
try:
response = wrapup_codes_api.post_wrap_up_codes(
body=GetWrapupcodesbody(page_size=25, expand=["default"])
)
if response.entities:
for code in response.entities:
if code.default is True:
return code.id
# Fallback to standard default ID
return "99999999-9999-9999-9999-999999999999"
except Exception as e:
print(f"Error fetching wrap-up codes: {e}")
# Return fallback on error to allow closure if possible
return "99999999-9999-9999-9999-999999999999"
def close_conversation(conversation_id: str, api_client: ApiClient, wrapup_code_id: str):
"""Closes the specified conversation."""
conversations_api = ConversationsApi(api_client)
close_body = PostConversationClosebody(wrap_up_code_id=wrapup_code_id)
try:
conversations_api.post_conversation_close(
conversation_id=conversation_id,
body=close_body
)
print(f"SUCCESS: Conversation {conversation_id} has been closed.")
return True
except requests.exceptions.HTTPError as err:
status_code = err.response.status_code
if status_code == 409:
print(f"INFO: Conversation {conversation_id} is already closed or locked.")
return True # Treat as success for idempotency
elif status_code == 404:
print(f"ERROR: Conversation {conversation_id} not found.")
return False
elif status_code in [401, 403]:
print(f"AUTH ERROR: {err.response.text}")
return False
else:
print(f"HTTP ERROR {status_code}: {err.response.text}")
return False
except Exception as e:
print(f"UNEXPECTED ERROR: {e}")
return False
def main():
# 1. Configuration
# Replace with a real conversation ID from your Genesys Cloud instance
TARGET_CONVERSATION_ID = os.getenv("TARGET_CONVERSATION_ID", "00000000-0000-0000-0000-000000000000")
if TARGET_CONVERSATION_ID == "00000000-0000-0000-0000-000000000000":
print("Please set TARGET_CONVERSATION_ID environment variable.")
sys.exit(1)
try:
# 2. Authenticate
print("Authenticating...")
token, api_client = get_access_token()
# 3. Get Wrap-up Code
print("Fetching default wrap-up code...")
wrapup_id = get_default_wrapup_code_id(api_client)
print(f"Using wrap-up code ID: {wrapup_id}")
# 4. Close Conversation
print(f"Closing conversation: {TARGET_CONVERSATION_ID}")
success = close_conversation(TARGET_CONVERSATION_ID, api_client, wrapup_id)
if not success:
sys.exit(1)
except Exception as e:
print(f"FATAL: {e}")
sys.exit(1)
if __name__ == "__main__":
main()
Common Errors & Debugging
Error: 401 Unauthorized
- Cause: The OAuth token has expired, is invalid, or the client credentials are incorrect.
- Fix: Verify
GENESYS_CLIENT_IDandGENESYS_CLIENT_SECRET. Ensure the token is refreshed before use in long-running processes. - Debug Code:
if err.response.status_code == 401: print("Token expired or invalid. Re-authenticating...") # Trigger re-authentication logic
Error: 403 Forbidden
- Cause: The OAuth client lacks the
conversations:closescope. - Fix: Go to Genesys Cloud Admin > Security > OAuth Clients > [Your Client] > Scopes. Add
conversations:close. - Debug Code:
if err.response.status_code == 403: print("Missing scope: conversations:close. Check OAuth Client configuration.")
Error: 409 Conflict
- Cause: The conversation is already in a
closedstate, or it is currently being handled by an agent who has locked it. - Fix: This is often an idempotency issue. If your goal is simply to ensure the conversation is closed, treat this as a success. If you need to know the current state, query the conversation details first.
- Debug Code:
if err.response.status_code == 409: # Log and continue, do not fail the process print("Conversation already closed or locked. Ignoring.")
Error: 400 Bad Request
- Cause: The
wrapUpCodeIdprovided is invalid or does not exist in the organization. - Fix: Ensure you are using a valid wrap-up code ID. Use the
get_default_wrapup_code_idfunction provided in Step 2 to dynamically fetch a valid ID. - Debug Code:
if err.response.status_code == 400: print(f"Invalid request body: {err.response.text}") # Check if wrapUpCodeId is a valid UUID and exists in the org