Send Canned Responses During Chat Interactions via the Genesys Cloud Conversations API

What You Will Build

• A Python script that joins an active chat conversation and sends a specific canned response to the customer.
• This tutorial uses the Genesys Cloud Platform API v2 and the gen-api-client-python SDK.
• The implementation covers Python 3.9+ with the official PureCloudPlatformClientV2 library.

Prerequisites

• OAuth Client Type: Public or Confidential Client with the scope conversation:chat:write.
• SDK Version: gen-api-client-python >= 150.0.0.
• Language/Runtime: Python 3.9 or later.
• External Dependencies:
  • gen-api-client-python: The official Genesys Cloud SDK.
  • requests: For OAuth token acquisition (if not using the SDK’s built-in auth helper).

Install the SDK via pip:

pip install gen-api-client-python

Authentication Setup

Genesys Cloud uses OAuth 2.0 for authentication. For server-to-server integrations, the Client Credentials Grant flow is standard. You must configure an OAuth client in the Genesys Cloud Admin Console with the conversation:chat:write scope.

The following code demonstrates how to initialize the PureCloudPlatformClientV2 instance. This instance handles token caching and automatic refresh, so you do not need to manually manage token expiration in your application logic.

from purecloud_platform_client import (
    Configuration,
    PureCloudPlatformClientV2,
    ApiClient
)
import os

def get_platform_client() -> PureCloudPlatformClientV2:
    """
    Initializes and returns a configured PureCloudPlatformClientV2 instance.
    """
    # Retrieve credentials from environment variables
    client_id = os.getenv("GENESYS_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLIENT_SECRET")
    environment = os.getenv("GENESYS_ENVIRONMENT", "mypurecloud.com")

    if not client_id or not client_secret:
        raise ValueError("GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET must be set.")

    # Configure the API client
    configuration = Configuration(
        host=f"https://api.{environment}",
        oauth_client_id=client_id,
        oauth_client_secret=client_secret
    )

    # Create the API client
    api_client = ApiClient(configuration)

    # Initialize the platform client
    platform_client = PureCloudPlatformClientV2(api_client)
    
    return platform_client

Required Scope: conversation:chat:write

If you attempt to call chat endpoints without this scope, the API will return a 403 Forbidden error. Ensure your OAuth client in the Admin Console has this scope explicitly added.

Implementation

Step 1: Identify the Conversation and Canned Response

Before sending a message, you need two identifiers:

1. The conversationId of the active chat session.
2. The id of the canned response you wish to send.

You can retrieve the conversation ID from your application state (e.g., from a webhook payload or the Chat SDK). To find the canned response ID, you can search for it by name using the Knowledge API.

def find_canned_response_id(platform_client: PureCloudPlatformClientV2, search_term: str) -> str:
    """
    Searches for a canned response by name and returns its ID.
    
    Args:
        platform_client: The initialized PureCloudPlatformClientV2 instance.
        search_term: The name of the canned response to find.
        
    Returns:
        The ID of the first matching canned response.
        
    Raises:
        ValueError: If no canned response is found.
    """
    from purecloud_platform_client.api_client import ApiClient
    from purecloud_platform_client.rest import ApiException

    knowledge_api = platform_client.knowledge_api
    
    # Search for canned responses
    # Note: The knowledge API returns snippets. We look for type 'CannedResponse'.
    try:
        search_result = knowledge_api.post_knowledge_search(
            body={
                "search": {
                    "text": search_term,
                    "types": ["CannedResponse"]
                },
                "size": 1
            }
        )
        
        if search_result.snippets and len(search_result.snippets) > 0:
            return search_result.snippets[0].id
        
        raise ValueError(f"No canned response found with search term: {search_term}")
        
    except ApiException as e:
        if e.status == 401:
            print("Authentication failed. Check your OAuth client credentials.")
        elif e.status == 403:
            print("Forbidden. Ensure the OAuth client has 'knowledge:knowledge:read' scope.")
        else:
            print(f"Error searching canned response: {e}")
        raise

Required Scope: knowledge:knowledge:read

This step requires a separate scope because reading knowledge base items (including canned responses) is a distinct permission from writing to conversations.

Step 2: Join the Conversation (If Necessary)

To send a message on behalf of a user (such as an agent or a bot), that user must be a participant in the conversation. If your OAuth client represents a system bot, you must join the conversation first.

If you are acting on behalf of an existing agent who is already in the chat, you can skip this step. However, for automated systems, joining is mandatory.

def join_chat_conversation(platform_client: PureCloudPlatformClientV2, conversation_id: str, user_id: str) -> None:
    """
    Joins a user to an existing chat conversation.
    
    Args:
        platform_client: The initialized PureCloudPlatformClientV2 instance.
        conversation_id: The ID of the chat conversation.
        user_id: The ID of the user joining the conversation.
    """
    from purecloud_platform_client.api_client import ApiClient
    from purecloud_platform_client.rest import ApiException
    from purecloud_platform_client.models import ConversationParticipant

    conversations_api = platform_client.conversations_api
    
    try:
        # Construct the participant body
        participant = ConversationParticipant(
            user_id=user_id,
            type="user",
            state="active"
        )
        
        # Join the conversation
        conversations_api.post_conversations_conversation_participants(
            conversation_id=conversation_id,
            body=participant
        )
        print(f"User {user_id} successfully joined conversation {conversation_id}.")
        
    except ApiException as e:
        if e.status == 409:
            print(f"User {user_id} is already a participant in conversation {conversation_id}.")
        elif e.status == 404:
            print(f"Conversation {conversation_id} not found.")
        else:
            print(f"Error joining conversation: {e}")
        raise

Required Scope: conversation:chat:write

The 409 Conflict response is common if the user is already joined. Your application should handle this gracefully, as it does not prevent sending messages.

Step 3: Send the Canned Response

Now that the user is joined, you can send the canned response. The Genesys Cloud Conversations API allows you to send a message by specifying the text content or by referencing a canned response directly. However, the most robust way to send a canned response is to retrieve its content and send it as a standard text message, or use the sendCannedResponse endpoint if available in your specific API version.

As of the current API version, the recommended approach is to use the post_conversations_conversation_messages endpoint with a body that references the canned response ID. This ensures that any updates to the canned response text are reflected in real-time.

def send_canned_response_message(
    platform_client: PureCloudPlatformClientV2, 
    conversation_id: str, 
    canned_response_id: str, 
    user_id: str
) -> None:
    """
    Sends a canned response to a chat conversation.
    
    Args:
        platform_client: The initialized PureCloudPlatformClientV2 instance.
        conversation_id: The ID of the chat conversation.
        canned_response_id: The ID of the canned response to send.
        user_id: The ID of the user sending the message.
    """
    from purecloud_platform_client.api_client import ApiClient
    from purecloud_platform_client.rest import ApiException
    from purecloud_platform_client.models import ChatMessageSendRequest

    conversations_api = platform_client.conversations_api
    
    try:
        # Construct the message request
        # Note: The ChatMessageSendRequest allows referencing a canned response
        message_request = ChatMessageSendRequest(
            canned_response_id=canned_response_id,
            user_id=user_id,
            type="cannedResponse"
        )
        
        # Send the message
        conversations_api.post_conversations_conversation_messages(
            conversation_id=conversation_id,
            body=message_request
        )
        
        print(f"Canned response {canned_response_id} sent to conversation {conversation_id} by user {user_id}.")
        
    except ApiException as e:
        if e.status == 401:
            print("Authentication failed. Check your OAuth client credentials.")
        elif e.status == 403:
            print("Forbidden. Ensure the OAuth client has 'conversation:chat:write' scope.")
        elif e.status == 404:
            print("Conversation or canned response not found.")
        elif e.status == 429:
            print("Rate limit exceeded. Implement retry logic.")
        else:
            print(f"Error sending message: {e}")
        raise

Required Scope: conversation:chat:write

This endpoint handles the rendering of the canned response text, including any variables defined within the canned response. If you need to pass variables, you must construct the message body with the variables field populated accordingly.

Step 4: Handle Rate Limits and Retries

Genesys Cloud APIs enforce rate limits. A 429 Too Many Requests response indicates that you have exceeded the allowed number of requests per second. You must implement exponential backoff to handle this.

import time
import random

def send_with_retry(
    func, 
    *args, 
    max_retries: int = 3, 
    base_delay: float = 1.0, 
    **kwargs
) -> None:
    """
    Executes an API function with retry logic for 429 errors.
    
    Args:
        func: The API function to execute.
        max_retries: The maximum number of retry attempts.
        base_delay: The initial delay in seconds before the first retry.
    """
    from purecloud_platform_client.rest import ApiException
    
    for attempt in range(max_retries):
        try:
            return func(*args, **kwargs)
        except ApiException as e:
            if e.status == 429:
                # Calculate delay with exponential backoff and jitter
                delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
                print(f"Rate limit exceeded. Retrying in {delay:.2f} seconds...")
                time.sleep(delay)
            else:
                raise

# Usage example
send_with_retry(
    send_canned_response_message,
    platform_client,
    conversation_id="12345678-1234-1234-1234-123456789012",
    canned_response_id="abcdef-1234-5678-90ab-cdef12345678",
    user_id="user-12345678-1234-1234-1234-123456789012"
)

This utility function wraps any API call, automatically retrying on 429 errors. The jitter prevents thundering herd scenarios if multiple clients retry simultaneously.

Complete Working Example

The following script combines all steps into a single executable module. It authenticates, finds a canned response, joins the conversation, and sends the message.

import os
import time
import random
from purecloud_platform_client import (
    Configuration,
    PureCloudPlatformClientV2,
    ApiClient
)
from purecloud_platform_client.rest import ApiException
from purecloud_platform_client.models import ChatMessageSendRequest, ConversationParticipant

def get_platform_client() -> PureCloudPlatformClientV2:
    client_id = os.getenv("GENESYS_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLIENT_SECRET")
    environment = os.getenv("GENESYS_ENVIRONMENT", "mypurecloud.com")

    if not client_id or not client_secret:
        raise ValueError("GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET must be set.")

    configuration = Configuration(
        host=f"https://api.{environment}",
        oauth_client_id=client_id,
        oauth_client_secret=client_secret
    )
    api_client = ApiClient(configuration)
    return PureCloudPlatformClientV2(api_client)

def find_canned_response_id(platform_client: PureCloudPlatformClientV2, search_term: str) -> str:
    knowledge_api = platform_client.knowledge_api
    try:
        search_result = knowledge_api.post_knowledge_search(
            body={
                "search": {
                    "text": search_term,
                    "types": ["CannedResponse"]
                },
                "size": 1
            }
        )
        if search_result.snippets and len(search_result.snippets) > 0:
            return search_result.snippets[0].id
        raise ValueError(f"No canned response found with search term: {search_term}")
    except ApiException as e:
        print(f"Error searching canned response: {e}")
        raise

def join_chat_conversation(platform_client: PureCloudPlatformClientV2, conversation_id: str, user_id: str) -> None:
    conversations_api = platform_client.conversations_api
    try:
        participant = ConversationParticipant(
            user_id=user_id,
            type="user",
            state="active"
        )
        conversations_api.post_conversations_conversation_participants(
            conversation_id=conversation_id,
            body=participant
        )
        print(f"User {user_id} joined conversation {conversation_id}.")
    except ApiException as e:
        if e.status == 409:
            print(f"User {user_id} already in conversation {conversation_id}.")
        else:
            raise

def send_canned_response_message(
    platform_client: PureCloudPlatformClientV2, 
    conversation_id: str, 
    canned_response_id: str, 
    user_id: str
) -> None:
    conversations_api = platform_client.conversations_api
    try:
        message_request = ChatMessageSendRequest(
            canned_response_id=canned_response_id,
            user_id=user_id,
            type="cannedResponse"
        )
        conversations_api.post_conversations_conversation_messages(
            conversation_id=conversation_id,
            body=message_request
        )
        print(f"Canned response sent to conversation {conversation_id}.")
    except ApiException as e:
        raise

def send_with_retry(func, *args, max_retries=3, base_delay=1.0, **kwargs):
    for attempt in range(max_retries):
        try:
            return func(*args, **kwargs)
        except ApiException as e:
            if e.status == 429:
                delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
                print(f"Rate limit exceeded. Retrying in {delay:.2f} seconds...")
                time.sleep(delay)
            else:
                raise

def main():
    # Configuration
    conversation_id = os.getenv("GENESYS_CONVERSATION_ID")
    user_id = os.getenv("GENESYS_USER_ID")
    canned_response_name = os.getenv("GENESYS_CANNED_RESPONSE_NAME", "Hello World")

    if not conversation_id or not user_id:
        raise ValueError("GENESYS_CONVERSATION_ID and GENESYS_USER_ID must be set.")

    # Initialize
    platform_client = get_platform_client()

    try:
        # Step 1: Find Canned Response
        canned_response_id = find_canned_response_id(platform_client, canned_response_name)
        
        # Step 2: Join Conversation
        join_chat_conversation(platform_client, conversation_id, user_id)
        
        # Step 3: Send Message with Retry
        send_with_retry(
            send_canned_response_message,
            platform_client,
            conversation_id,
            canned_response_id,
            user_id
        )
        
    except Exception as e:
        print(f"Failed to send canned response: {e}")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 403 Forbidden

• Cause: The OAuth client lacks the required scope.
• Fix: Ensure the OAuth client has conversation:chat:write and knowledge:knowledge:read scopes.
• Debug: Check the OAuth client configuration in the Genesys Cloud Admin Console under Admin > Security > OAuth Clients.

Error: 404 Not Found

• Cause: The conversation ID or canned response ID is invalid.
• Fix: Verify the IDs exist. Use the Genesys Cloud API Explorer to test the GET /api/v2/conversations/{conversationId} endpoint.
• Debug: Print the IDs before making the API call to ensure they are not null or empty.

Error: 409 Conflict

• Cause: The user is already a participant in the conversation.
• Fix: This is not an error. Ignore the exception or handle it gracefully.
• Debug: Check the response body for details. The API will confirm the user is already joined.

Error: 429 Too Many Requests

• Cause: Rate limit exceeded.
• Fix: Implement exponential backoff with jitter.
• Debug: Check the Retry-After header in the response for the recommended wait time.

Official References

• Conversations API Documentation
• Knowledge API Documentation
• PureCloud Platform Client Python SDK
• OAuth Scopes Reference