Send Canned Responses Programmatically via Genesys Cloud Conversations API

What You Will Build

• A Python script that injects a canned response into an active web chat conversation using the Genesys Cloud Conversations API.
• This solution uses the /api/v2/conversations/messages endpoint to post a message on behalf of an agent or user.
• The tutorial covers Python with the genesyscloud SDK and raw requests for HTTP-level visibility.

Prerequisites

• OAuth Client Type: Confidential Client (Client Credentials Grant) or JWT (Service Account).
• Required Scopes:
  • conversation:read (to locate the conversation).
  • conversation:message:write (to send the message).
  • user:read (to identify the agent sending the message, if acting as an agent).
• SDK Version: genesyscloud Python SDK >= 100.0.0.
• Runtime: Python 3.8+.
• Dependencies:
  • pip install genesyscloud
  • pip install httpx (for raw HTTP examples if needed, though SDK is preferred).

Authentication Setup

Genesys Cloud APIs require a valid OAuth 2.0 access token. For backend services, the Client Credentials Grant is the standard. The token must contain the scopes listed above.

The genesyscloud SDK handles token refresh automatically if configured correctly. Below is the initialization pattern.

import os
from purecloudplatformclientv2 import Configuration, ApiClient

def get_api_client():
    """
    Initializes and returns a configured PureCloudPlatformClientV2 instance.
    Uses environment variables for security.
    """
    # Environment variables must be set:
    # GENESYS_CLIENT_ID, GENESYS_CLIENT_SECRET, GENESYS_BASE_URL
    
    config = Configuration()
    config.host = os.getenv("GENESYS_BASE_URL", "https://api.mypurecloud.com")
    config.client_id = os.getenv("GENESYS_CLIENT_ID")
    config.client_secret = os.getenv("GENESYS_CLIENT_SECRET")
    
    if not config.client_id or not config.client_secret:
        raise ValueError("GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET must be set.")

    # The SDK uses a default token provider that caches and refreshes tokens
    api_client = ApiClient(configuration=config)
    return api_client

Implementation

Step 1: Locate the Active Conversation

To send a message, you need the conversationId. In a live chat scenario, this ID is often passed from the frontend via WebChat SDK events. However, if you are building a backend service that monitors for new chats, you must query the conversation list.

The endpoint GET /api/v2/conversations supports filtering by type and state.

Endpoint: GET /api/v2/conversations
Method: GET
Scopes: conversation:read

from purecloudplatformclientv2 import ConversationApi
from purecloudplatformclientv2.rest import ApiException

def find_active_chat_conversation(api_client: ApiClient, user_id: str) -> str:
    """
    Finds the most recent active chat conversation for a specific user.
    
    Args:
        api_client: Authenticated ApiClient instance.
        user_id: The ID of the agent or user involved in the chat.
        
    Returns:
        The conversation ID string.
    """
    conversation_api = ConversationApi(api_client)
    
    try:
        # Filter for webchat, active status, and specific user involvement
        # Note: The API returns a paginated list. We fetch the first page.
        response = conversation_api.get_conversations(
            type="webchat",
            status="active",
            expand=["participants"], # Expand participants to check user involvement
            limit=10
        )
        
        for conversation in response.entities:
            # Check if the target user is a participant
            for participant in conversation.participants:
                if participant.user.id == user_id:
                    return conversation.id
                    
        raise Exception("No active webchat conversation found for the user.")
        
    except ApiException as e:
        print(f"Exception when calling ConversationApi->get_conversations: {e}")
        raise

Step 2: Construct the Canned Response Message

The core action is posting a message to the conversation. The API endpoint is POST /api/v2/conversations/{conversationId}/messages.

When sending a canned response, you are essentially creating a Message object. The critical fields are:

1. from: The UserIdentity of the sender (usually the agent).
2. to: The UserIdentity of the recipient (the customer).
3. text: The content of the canned response.

Endpoint: POST /api/v2/conversations/{conversationId}/messages
Method: POST
Scopes: conversation:message:write

Request Body Structure:

{
  "from": {
    "id": "agent-user-id-here",
    "name": "Agent Name",
    "type": "user"
  },
  "to": [
    {
      "id": "customer-user-id-here",
      "name": "Customer Name",
      "type": "user"
    }
  ],
  "text": "Here is the canned response content.",
  "timestamp": "2023-10-27T10:00:00.000Z"
}

Important Note on timestamp:
If you omit timestamp, Genesys Cloud generates one server-side. If you provide it, ensure it is ISO 8601 format. For real-time canned responses, omitting it is safer to maintain chronological accuracy.

Important Note on to:
In a webchat, the to array usually contains the external user. You can find the external user’s ID from the conversation participants list retrieved in Step 1. If you are unsure of the specific external user ID, you can sometimes omit the to field if the API context implies the other participant, but explicitly defining to is the robust approach for programmatic injection.

Step 3: Sending the Message via SDK

Here is the function that constructs the payload and sends it.

from purecloudplatformclientv2 import PostConversationMessageRequest, UserIdentity

def send_canned_response(
    api_client: ApiClient,
    conversation_id: str,
    agent_id: str,
    customer_id: str,
    canned_text: str
) -> dict:
    """
    Sends a canned response message to a chat conversation.
    
    Args:
        api_client: Authenticated ApiClient instance.
        conversation_id: The ID of the chat conversation.
        agent_id: The ID of the agent sending the message.
        customer_id: The ID of the customer receiving the message.
        canned_text: The string content of the canned response.
        
    Returns:
        The response metadata containing the message ID.
    """
    conversation_api = ConversationApi(api_client)
    
    # Construct the 'from' identity
    sender = UserIdentity(
        id=agent_id,
        name="Support Agent", # Optional: Display name
        type="user"
    )
    
    # Construct the 'to' identity list
    recipient = UserIdentity(
        id=customer_id,
        name="Customer", # Optional: Display name
        type="user"
    )
    
    # Build the request object
    message_request = PostConversationMessageRequest(
        from_=sender,
        to=[recipient],
        text=canned_text
    )
    
    try:
        # Execute the API call
        response = conversation_api.post_conversations_messages(
            conversation_id=conversation_id,
            body=message_request
        )
        
        return {
            "success": True,
            "message_id": response.id if response else None,
            "status_code": 200
        }
        
    except ApiException as e:
        # Handle specific errors
        if e.status == 401:
            print("Authentication failed. Check OAuth token.")
        elif e.status == 403:
            print("Permission denied. Check scopes: conversation:message:write")
        elif e.status == 404:
            print(f"Conversation {conversation_id} not found.")
        else:
            print(f"Unexpected error: {e}")
        raise

Step 4: Handling Pagination and Rate Limits

While sending a single message does not involve pagination, fetching the conversation list (Step 1) does. The get_conversations endpoint returns a divisions and nextPageUri if more data exists.

For high-volume scenarios (e.g., sending canned responses to thousands of chats), you will hit rate limits (429 Too Many Requests). The genesyscloud SDK does not automatically retry 429s in all versions. You should implement exponential backoff.

import time
import random

def send_with_retry(api_client, conversation_id, agent_id, customer_id, text, max_retries=3):
    """
    Wraps send_canned_response with exponential backoff for 429 errors.
    """
    for attempt in range(max_retries):
        try:
            return send_canned_response(api_client, conversation_id, agent_id, customer_id, text)
        except ApiException as e:
            if e.status == 429:
                # Extract Retry-After header if present, otherwise default
                retry_after = e.headers.get('Retry-After', 1)
                # Add jitter to prevent thundering herd
                wait_time = float(retry_after) + random.uniform(0, 1)
                print(f"Rate limited. Retrying in {wait_time:.2f} seconds...")
                time.sleep(wait_time)
            else:
                # Non-retryable error
                raise
    raise Exception("Max retries exceeded for sending canned response.")

Complete Working Example

This script combines authentication, conversation lookup, and message sending. It assumes you have an active webchat and know the Agent ID.

import os
import sys
from purecloudplatformclientv2 import Configuration, ApiClient, ConversationApi, PostConversationMessageRequest, UserIdentity
from purecloudplatformclientv2.rest import ApiException

# --- Configuration ---
# Set these in your environment
CLIENT_ID = os.getenv("GENESYS_CLIENT_ID")
CLIENT_SECRET = os.getenv("GENESYS_CLIENT_SECRET")
BASE_URL = os.getenv("GENESYS_BASE_URL", "https://api.mypurecloud.com")
AGENT_ID = os.getenv("GENESYS_AGENT_ID") # The user ID of the agent sending the canned response

def initialize_client() -> ApiClient:
    config = Configuration()
    config.host = BASE_URL
    config.client_id = CLIENT_ID
    config.client_secret = CLIENT_SECRET
    return ApiClient(configuration=config)

def get_active_chat_for_agent(api_client: ApiClient, agent_id: str) -> tuple:
    """
    Returns (conversation_id, customer_id) for the first active chat involving the agent.
    """
    conversation_api = ConversationApi(api_client)
    
    try:
        response = conversation_api.get_conversations(
            type="webchat",
            status="active",
            expand=["participants"],
            limit=5
        )
        
        for conv in response.entities:
            for participant in conv.participants:
                if participant.user.id == agent_id:
                    # Found the agent, now find the other participant (customer)
                    for other_participant in conv.participants:
                        if other_participant.user.id != agent_id:
                            return conv.id, other_participant.user.id
        
        return None, None
        
    except ApiException as e:
        print(f"Error fetching conversations: {e}")
        raise

def send_canned(api_client: ApiClient, conv_id: str, customer_id: str, text: str):
    conversation_api = ConversationApi(api_client)
    
    sender = UserIdentity(id=AGENT_ID, name="System Agent", type="user")
    recipient = UserIdentity(id=customer_id, name="Customer", type="user")
    
    body = PostConversationMessageRequest(
        from_=sender,
        to=[recipient],
        text=text
    )
    
    try:
        resp = conversation_api.post_conversations_messages(conv_id, body=body)
        print(f"Message sent successfully. Message ID: {resp.id}")
    except ApiException as e:
        print(f"Failed to send message: {e}")
        if e.status == 429:
            print("Hit rate limit. Implement retry logic in production.")

def main():
    if not CLIENT_ID or not CLIENT_SECRET or not AGENT_ID:
        print("Error: Missing environment variables GENESYS_CLIENT_ID, GENESYS_CLIENT_SECRET, or GENESYS_AGENT_ID")
        sys.exit(1)
        
    api_client = initialize_client()
    
    print("Searching for active chat conversation...")
    conv_id, customer_id = get_active_chat_for_agent(api_client, AGENT_ID)
    
    if not conv_id:
        print("No active chat found for the agent.")
        return
        
    print(f"Found conversation: {conv_id}")
    print(f"Sending canned response to customer: {customer_id}")
    
    canned_text = "Thank you for contacting support. Your ticket has been updated."
    send_canned(api_client, conv_id, customer_id, canned_text)

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 403 Forbidden

Cause: The OAuth token lacks the conversation:message:write scope.
Fix: Ensure your OAuth client in the Genesys Cloud Admin Console has the “Conversations” permission set to “Read and Write” or explicitly add the conversation:message:write scope to the client configuration.
Code Check:

# Verify scopes in token response if using raw HTTP
# "scope": "conversation:read conversation:message:write ..."

Error: 404 Not Found

Cause: The conversation_id is invalid or the conversation has ended.
Fix: Verify the conversation status is active. If the chat has ended, you cannot send messages to it via the Conversations API. You must use the Archives API for historical data, but you cannot inject new messages into closed conversations.
Debugging:

# Check conversation status before sending
conv = conversation_api.get_conversation(conversation_id)
if conv.status != "active":
    raise Exception("Conversation is not active.")

Error: 400 Bad Request - “Invalid Participant”

Cause: The customer_id provided in the to array is not a participant in the conversation.
Fix: Ensure you retrieve the customer_id directly from the conversation’s participants list. Do not guess the user ID.
Code Fix:

# Always derive recipient ID from the conversation object
for p in response.entities[0].participants:
    if p.user.id != AGENT_ID:
        customer_id = p.user.id

Error: 429 Too Many Requests

Cause: You are sending messages too rapidly. Genesys Cloud enforces rate limits per client and per conversation.
Fix: Implement exponential backoff. Do not fire-and-forget in a tight loop without delays.
Code Fix: Use the send_with_retry function provided in Step 4.

Official References

• Genesys Cloud Conversations API Documentation
• POST /api/v2/conversations/{conversationId}/messages
• Python SDK Reference