How to Inject Canned Responses into Active Chats via the Genesys Cloud Conversations API

How-To-Articles
1

How to Inject Canned Responses into Active Chats via the Genesys Cloud Conversations API

What You Will Build

  • A Python script that authenticates against Genesys Cloud, identifies an active chat session, and injects a pre-defined text message into the conversation stream on behalf of an agent.
  • This tutorial uses the Genesys Cloud Conversations API (/api/v2/conversations) and the purecloudplatformclientv2 Python SDK.
  • The implementation covers the required POST method, payload construction, and handling of rate limits and invalid session errors.

Prerequisites

  • OAuth Client Type: Private Key or Client Credentials.
  • Required Scopes: chat:conversation:write (to send messages) and chat:conversation:read (to verify session status if needed).
  • SDK Version: purecloudplatformclientv2 v142.0.0 or later.
  • Language/Runtime: Python 3.8+
  • External Dependencies:
    • purecloudplatformclientv2: The official Genesys Cloud Python SDK.
    • requests: For fallback HTTP calls if SDK methods are insufficient (though this tutorial uses the SDK primarily).

Authentication Setup

Genesys Cloud uses OAuth 2.0 for all API access. For server-side integrations (like this one), the Private Key flow is the standard. You must generate a private key in the Genesys Cloud Admin Console under Organization > API Access.

The SDK handles the token acquisition and refresh automatically. You only need to initialize the Configuration object with your client ID, private key, and environment URL.

import purecloudplatformclientv2
from purecloudplatformclientv2.rest import ApiException

def get_platform_client(client_id: str, private_key_path: str, environment: str = "mypurecloud.com") -> purecloudplatformclientv2.ApiClient:
    """
    Initializes the Genesys Cloud Platform Client with Private Key authentication.
    
    Args:
        client_id: The OAuth Client ID from Genesys Cloud.
        private_key_path: Path to the PEM file containing the private key.
        environment: The Genesys Cloud environment domain (e.g., usw2.pure.cloud).
    
    Returns:
        A configured ApiClient instance.
    """
    # Load the private key from a file
    with open(private_key_path, 'r') as key_file:
        private_key = key_file.read()

    # Configure the OAuth2 client
    config = purecloudplatformclientv2.Configuration(
        host=f"https://{environment}",
        client_id=client_id,
        private_key=private_key
    )

    # Create the API client
    api_client = purecloudplatformclientv2.ApiClient(configuration=config)
    
    return api_client

Note: Ensure the private key file has restricted permissions (e.g., chmod 600) on your server. The SDK will automatically request an access token on the first API call and refresh it when it expires.

Implementation

Sending a canned response in Genesys Cloud is technically equivalent to sending any other message in a conversation. There is no specific “SendCannedResponse” endpoint. Instead, you use the generic POST /api/v2/conversations/{conversationId}/messages endpoint. The “canned” nature of the response is determined by the content of the message body, which you can pull from your own database or a static configuration file.

Step 1: Identify the Active Conversation

Before sending a message, you must have the conversationId. In a real-world scenario, this ID is passed to your backend via a webhook (e.g., Chat.Message.Send event) or retrieved via the Routing API.

For this tutorial, we assume you have already captured the conversationId. If you need to find it dynamically, you would query /api/v2/conversations/chats with filters for the agent ID and status.

Required Scope: chat:conversation:read

def get_active_chat_conversation_id(api_client: purecloudplatformclientv2.ApiClient, agent_id: str) -> str | None:
    """
    Retrieves the ID of an active chat conversation for a specific agent.
    
    Args:
        api_client: The initialized Genesys Cloud API client.
        agent_id: The UUID of the agent.
    
    Returns:
        The conversation ID string, or None if no active chats are found.
    """
    conversations_api = purecloudplatformclientv2.ConversationsApi(api_client)
    
    try:
        # Query for active chat conversations assigned to the agent
        # Note: This is a simplified query. In production, use specific filters for efficiency.
        body = purecloudplatformclientv2.ConversationQueryRequest(
            conversation_types=["chat"],
            query_filter=purecloudplatformclientv2.ChatQueryFilter(
                participant_ids=[agent_id]
            ),
            status="active"
        )
        
        response = conversations_api.post_conversations_chat_search(body=body)
        
        if response.conversations and len(response.conversations) > 0:
            return response.conversations[0].id
        else:
            print("No active chat conversations found for this agent.")
            return None
            
    except ApiException as e:
        print(f"Exception when calling ConversationsApi->post_conversations_chat_search: {e}")
        return None

Step 2: Construct the Message Payload

The core of this operation is the ConversationMessageCreateRequest. This object defines the direction of the message, the content, and optionally the user sending it.

Key Parameters:

  • from_id: The ID of the participant sending the message. For a canned response, this is typically the Agent’s ID. If you omit this, Genesys Cloud may default to the system or fail depending on the context.
  • to_id: The ID of the recipient. For a chat, this is usually the Customer’s ID.
  • text: The actual canned text.
  • type: Must be user for standard chat messages.
  • direction: outbound indicates the agent is sending to the customer.

Required Scope: chat:conversation:write

from purecloudplatformclientv2 import ConversationMessageCreateRequest

def create_message_payload(agent_id: str, customer_id: str, canned_text: str) -> ConversationMessageCreateRequest:
    """
    Constructs the payload for sending a message in a chat conversation.
    
    Args:
        agent_id: The UUID of the agent sending the canned response.
        customer_id: The UUID of the customer receiving the message.
        canned_text: The string content of the canned response.
    
    Returns:
        A ConversationMessageCreateRequest object.
    """
    return ConversationMessageCreateRequest(
        from_id=agent_id,
        to_id=customer_id,
        text=canned_text,
        type="user",
        direction="outbound"
    )

Step 3: Send the Message via the API

Now we execute the POST request. We will wrap this in a retry logic block to handle transient 429 (Too Many Requests) errors, which are common in high-volume chat environments.

import time
import random

def send_chat_message(api_client: purecloudplatformclientv2.ApiClient, conversation_id: str, message_payload: ConversationMessageCreateRequest, max_retries: int = 3) -> bool:
    """
    Sends a message to a specific conversation with retry logic for rate limits.
    
    Args:
        api_client: The initialized Genesys Cloud API client.
        conversation_id: The UUID of the chat conversation.
        message_payload: The ConversationMessageCreateRequest object.
        max_retries: Number of times to retry on 429 errors.
    
    Returns:
        True if successful, False otherwise.
    """
    conversations_api = purecloudplatformclientv2.ConversationsApi(api_client)
    
    for attempt in range(max_retries):
        try:
            # Execute the API call
            # Endpoint: POST /api/v2/conversations/{conversationId}/messages
            response = conversations_api.post_conversations_message(
                conversation_id=conversation_id,
                body=message_payload
            )
            
            # If we reach here, the message was sent successfully
            print(f"Message sent successfully. Message ID: {response.id}")
            return True
            
        except ApiException as e:
            if e.status == 429:
                # Rate Limited: Implement exponential backoff
                wait_time = (2 ** attempt) + random.uniform(0, 1)
                print(f"Rate limited (429). Retrying in {wait_time:.2f} seconds...")
                time.sleep(wait_time)
                continue
            elif e.status == 404:
                print(f"Conversation not found: {conversation_id}")
                return False
            elif e.status == 403:
                print(f"Forbidden: Ensure your OAuth token has 'chat:conversation:write' scope.")
                return False
            else:
                print(f"Unexpected error: {e}")
                return False
                
    print("Max retries reached. Failed to send message.")
    return False

Complete Working Example

Below is the complete, runnable Python script. It combines authentication, conversation lookup, and message sending.

Prerequisites:

  1. Install the SDK: pip install purecloudplatformclientv2
  2. Create a file named send_canned_response.py.
  3. Place your private key in a file named private_key.pem in the same directory.
import os
import sys
import purecloudplatformclientv2
from purecloudplatformclientv2.rest import ApiException
from purecloudplatformclientv2 import ConversationMessageCreateRequest
import time
import random

# Configuration Constants
CLIENT_ID = "YOUR_CLIENT_ID"
PRIVATE_KEY_PATH = "private_key.pem"
ENVIRONMENT = "mypurecloud.com" # Change to your environment (e.g., usw2.pure.cloud)
AGENT_ID = "YOUR_AGENT_UUID"    # The UUID of the agent sending the message
CANNED_TEXT = "Thank you for contacting us. We have received your request and will respond shortly."

def get_platform_client(client_id: str, private_key_path: str, environment: str) -> purecloudplatformclientv2.ApiClient:
    with open(private_key_path, 'r') as key_file:
        private_key = key_file.read()

    config = purecloudplatformclientv2.Configuration(
        host=f"https://{environment}",
        client_id=client_id,
        private_key=private_key
    )
    return purecloudplatformclientv2.ApiClient(configuration=config)

def get_active_chat_conversation_id(api_client: purecloudplatformclientv2.ApiClient, agent_id: str) -> str | None:
    conversations_api = purecloudplatformclientv2.ConversationsApi(api_client)
    
    try:
        body = purecloudplatformclientv2.ConversationQueryRequest(
            conversation_types=["chat"],
            query_filter=purecloudplatformclientv2.ChatQueryFilter(
                participant_ids=[agent_id]
            ),
            status="active"
        )
        
        response = conversations_api.post_conversations_chat_search(body=body)
        
        if response.conversations and len(response.conversations) > 0:
            return response.conversations[0].id
        else:
            return None
            
    except ApiException as e:
        print(f"Exception when searching for conversations: {e}")
        return None

def send_chat_message(api_client: purecloudplatformclientv2.ApiClient, conversation_id: str, message_payload: ConversationMessageCreateRequest, max_retries: int = 3) -> bool:
    conversations_api = purecloudplatformclientv2.ConversationsApi(api_client)
    
    for attempt in range(max_retries):
        try:
            response = conversations_api.post_conversations_message(
                conversation_id=conversation_id,
                body=message_payload
            )
            print(f"Success: Message ID {response.id} sent.")
            return True
            
        except ApiException as e:
            if e.status == 429:
                wait_time = (2 ** attempt) + random.uniform(0, 1)
                print(f"Rate limited. Waiting {wait_time:.2f}s...")
                time.sleep(wait_time)
                continue
            else:
                print(f"Error {e.status}: {e.body}")
                return False
    return False

def main():
    # 1. Initialize API Client
    print("Initializing API Client...")
    api_client = get_platform_client(CLIENT_ID, PRIVATE_KEY_PATH, ENVIRONMENT)
    
    # 2. Find an Active Chat
    print(f"Searching for active chats for agent {AGENT_ID}...")
    conversation_id = get_active_chat_conversation_id(api_client, AGENT_ID)
    
    if not conversation_id:
        print("No active chats found. Please start a chat session and try again.")
        sys.exit(1)
        
    print(f"Found active chat: {conversation_id}")
    
    # 3. Retrieve Customer ID (Optional but recommended for proper 'to_id')
    # In a real scenario, you would parse the conversation details to get the customer ID.
    # For this example, we will assume the customer ID is known or retrieved from the conversation object.
    # NOTE: You must replace CUSTOMER_ID below with the actual UUID of the customer participant.
    CUSTOMER_ID = "REPLACE_WITH_CUSTOMER_UUID" 
    
    if CUSTOMER_ID == "REPLACE_WITH_CUSTOMER_UUID":
        print("Error: You must provide a valid CUSTOMER_ID.")
        sys.exit(1)

    # 4. Construct and Send Message
    message_payload = ConversationMessageCreateRequest(
        from_id=AGENT_ID,
        to_id=CUSTOMER_ID,
        text=CANNED_TEXT,
        type="user",
        direction="outbound"
    )
    
    print(f"Sending canned response: '{CANNED_TEXT}'")
    success = send_chat_message(api_client, conversation_id, message_payload)
    
    if success:
        print("Operation completed successfully.")
    else:
        print("Operation failed.")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 403 Forbidden

Cause: The OAuth token does not have the required scope.
Fix: Ensure your OAuth Client in Genesys Cloud has chat:conversation:write assigned. If you are using a Private Key, verify that the key is associated with a user who has the “Chat Admin” or “Chat Manager” role, or that the OAuth Client itself has the scope granted directly.

Error: 404 Not Found

Cause: The conversationId or participant_id (from_id/to_id) is invalid.
Fix:

  1. Verify the conversationId matches the format xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx.
  2. Ensure the from_id (Agent) is currently a participant in the conversation. You cannot send a message on behalf of an agent who is not in the chat.
  3. Ensure the to_id (Customer) is a valid participant. If you are unsure of the customer ID, you can retrieve the conversation details via GET /api/v2/conversations/{conversationId} and inspect the participants array.

Error: 429 Too Many Requests

Cause: You are sending messages faster than Genesys Cloud allows for that specific client or tenant.
Fix: Implement exponential backoff. The code above includes a basic retry mechanism. For high-volume bots, consider queuing messages and using a rate-limiter library like ratelimit in Python.

Error: Message does not appear in Agent Desktop

Cause: The message was sent but filtered by the Agent Desktop settings or the conversation is in a state that does not display new messages (e.g., “Wrap-up” or “Closed”).
Fix: Check the conversation status. Messages sent to a “Closed” conversation will be stored but may not trigger real-time notifications. Ensure the conversation is active or resolved (depending on your org’s settings) before sending.

Official References