Send a Canned Response During a Chat Interaction via the Conversations API

What You Will Build

• One sentence: This tutorial demonstrates how to programmatically inject a pre-approved canned response into an active real-time chat session.
• One sentence: This uses the Genesys Cloud CX Conversations API (POST /api/v2/conversations/chats/{conversationId}/messages) and the Python SDK.
• One sentence: The implementation is covered in Python using the genesyscloud package.

Prerequisites

• OAuth Client Type: Service Account or Authorization Code Grant.
• Required Scopes:
  • conversation:chat:write (to send messages)
  • cannedresponse:read (to fetch the canned response text if not hardcoded)
  • conversation:read (optional, to verify conversation status)
• SDK Version: genesyscloud >= 2.0.0.
• Language/Runtime: Python 3.8+.
• External Dependencies:
  • genesyscloud
  • requests (for manual HTTP examples)

Authentication Setup

Genesys Cloud APIs require a valid Bearer token. For service accounts, you exchange a client ID and client secret for a token using the client_credentials grant type.

import os
import requests
from typing import Optional

def get_access_token(client_id: str, client_secret: str, base_url: str = "https://api.mypurecloud.com") -> str:
    """
    Retrieves an OAuth2 access token for a Genesys Cloud Service Account.
    """
    token_url = f"{base_url}/oauth/token"
    payload = {
        "grant_type": "client_credentials",
        "client_id": client_id,
        "client_secret": client_secret,
        "scope": "conversation:chat:write cannedresponse:read"
    }
    headers = {
        "Content-Type": "application/x-www-form-urlencoded"
    }

    response = requests.post(token_url, data=payload, headers=headers)
    
    if response.status_code == 200:
        return response.json().get("access_token")
    else:
        raise Exception(f"Authentication failed with status {response.status_code}: {response.text}")

# Example usage
# TOKEN = get_access_token(os.getenv("GENESYS_CLIENT_ID"), os.getenv("GENESYS_CLIENT_SECRET"))

Note: In production, cache the token and refresh it before expiration (typically 1 hour). Do not call this endpoint for every message send.

Implementation

Step 1: Identify the Conversation and Canned Response

To send a canned response, you need two identifiers:

1. The conversationId (UUID of the active chat).
2. The cannedResponse object or its text content.

While you can hardcode the text, best practice involves fetching the canned response by ID to ensure version control and localization compliance.

import requests
import json

def get_canned_response(token: str, canned_response_id: str, base_url: str = "https://api.mypurecloud.com") -> dict:
    """
    Fetches a specific canned response by ID.
    """
    url = f"{base_url}/api/v2/cannedresponses/{canned_response_id}"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }

    response = requests.get(url, headers=headers)
    
    if response.status_code == 200:
        return response.json()
    elif response.status_code == 404:
        raise ValueError(f"Canned response ID {canned_response_id} not found.")
    else:
        raise Exception(f"Failed to fetch canned response: {response.status_code} {response.text}")

# Example payload returned:
# {
#   "id": "canned-uuid-123",
#   "name": "Welcome Message",
#   "description": "Standard welcome",
#   "body": "Hello, thank you for contacting support. An agent will be with you shortly.",
#   "type": "text",
#   ...
# }

Step 2: Construct the Message Payload

The Conversations API endpoint for sending chat messages is POST /api/v2/conversations/chats/{conversationId}/messages.

The request body must adhere to the ChatMessage schema. Critical fields include:

• type: Must be "user" for agent-initiated messages or "customer" for simulated customer messages. For canned responses sent by an agent, use "user".
• text: The content of the message. This comes from the body field of the canned response.
• from: The ID of the user (agent) sending the message. This is crucial for audit trails. If omitted, the system may attribute it to the service account or fail, depending on configuration.

Important: If the canned response contains HTML (type html), you must set the type field in the message payload to html and ensure the text is valid HTML. For plain text, use text.

def build_message_payload(canned_response: dict, agent_id: str) -> dict:
    """
    Constructs the JSON payload for sending a chat message.
    """
    # Determine message type based on canned response type
    msg_type = "text"
    if canned_response.get("type") == "html":
        msg_type = "html"
    
    payload = {
        "type": "user",  # Indicates this is an agent/system message
        "from": {
            "id": agent_id,
            "name": "Support Agent"  # Optional, but good for debugging
        },
        "text": canned_response.get("body", ""),
        "contentType": msg_type  # 'text' or 'html'
    }
    
    return payload

Step 3: Send the Message via Conversations API

Now we execute the POST request. This operation is synchronous. The API returns a 201 Created status if successful.

def send_canned_response_message(token: str, conversation_id: str, payload: dict, base_url: str = "https://api.mypurecloud.com") -> dict:
    """
    Sends a message to an active chat conversation.
    """
    url = f"{base_url}/api/v2/conversations/chats/{conversation_id}/messages"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }

    response = requests.post(url, headers=headers, json=payload)
    
    if response.status_code == 201:
        return response.json()
    elif response.status_code == 404:
        raise ValueError(f"Conversation ID {conversation_id} not found or ended.")
    elif response.status_code == 409:
        raise ValueError(f"Conversation conflict. The conversation may be ended or invalid for messaging.")
    else:
        raise Exception(f"Failed to send message: {response.status_code} {response.text}")

# Example Response Body:
# {
#   "id": "message-uuid-456",
#   "type": "user",
#   "from": { ... },
#   "to": [],
#   "text": "Hello, thank you for contacting support...",
#   "contentType": "text",
#   "timestamp": "2023-10-27T10:00:00.000Z",
#   "conversationId": "conv-uuid-789"
# }

Complete Working Example

This script combines authentication, retrieval, and sending into a single executable module. It includes basic error handling and logging.

import os
import sys
import requests
import logging
from typing import Optional, Dict, Any

# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

GENESYS_BASE_URL = "https://api.mypurecloud.com"
CLIENT_ID = os.getenv("GENESYS_CLIENT_ID")
CLIENT_SECRET = os.getenv("GENESYS_CLIENT_SECRET")

def get_access_token(client_id: str, client_secret: str) -> str:
    token_url = f"{GENESYS_BASE_URL}/oauth/token"
    payload = {
        "grant_type": "client_credentials",
        "client_id": client_id,
        "client_secret": client_secret,
        "scope": "conversation:chat:write cannedresponse:read"
    }
    headers = {"Content-Type": "application/x-www-form-urlencoded"}
    
    try:
        response = requests.post(token_url, data=payload, headers=headers, timeout=10)
        response.raise_for_status()
        return response.json().get("access_token")
    except requests.exceptions.RequestException as e:
        logger.error(f"Authentication failed: {e}")
        raise

def get_canned_response(token: str, canned_response_id: str) -> Dict[str, Any]:
    url = f"{GENESYS_BASE_URL}/api/v2/cannedresponses/{canned_response_id}"
    headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
    
    try:
        response = requests.get(url, headers=headers, timeout=10)
        response.raise_for_status()
        return response.json()
    except requests.exceptions.HTTPError as e:
        if response.status_code == 404:
            logger.error(f"Canned response {canned_response_id} not found.")
        else:
            logger.error(f"Error fetching canned response: {e}")
        raise

def send_chat_message(token: str, conversation_id: str, agent_id: str, canned_response: Dict[str, Any]) -> Dict[str, Any]:
    msg_type = "html" if canned_response.get("type") == "html" else "text"
    
    payload = {
        "type": "user",
        "from": {
            "id": agent_id
        },
        "text": canned_response.get("body", ""),
        "contentType": msg_type
    }
    
    url = f"{GENESYS_BASE_URL}/api/v2/conversations/chats/{conversation_id}/messages"
    headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
    
    try:
        response = requests.post(url, headers=headers, json=payload, timeout=10)
        response.raise_for_status()
        logger.info(f"Message sent successfully. ID: {response.json().get('id')}")
        return response.json()
    except requests.exceptions.HTTPError as e:
        logger.error(f"Failed to send message to conversation {conversation_id}: {e}")
        raise

def main():
    if not CLIENT_ID or not CLIENT_SECRET:
        logger.error("Missing GENESYS_CLIENT_ID or GENESYS_CLIENT_SECRET environment variables.")
        sys.exit(1)

    # Configuration
    CONVERSATION_ID = "YOUR_CONVERSATION_UUID"
    CANNED_RESPONSE_ID = "YOUR_CANNED_RESPONSE_UUID"
    AGENT_ID = "YOUR_AGENT_UUID"  # The user ID of the agent sending the message

    try:
        # 1. Authenticate
        logger.info("Authenticating...")
        token = get_access_token(CLIENT_ID, CLIENT_SECRET)

        # 2. Fetch Canned Response
        logger.info(f"Fetching canned response: {CANNED_RESPONSE_ID}")
        canned_response = get_canned_response(token, CANNED_RESPONSE_ID)
        logger.info(f"Retrieved canned response: {canned_response.get('name')}")

        # 3. Send Message
        logger.info(f"Sending message to conversation: {CONVERSATION_ID}")
        result = send_chat_message(token, CONVERSATION_ID, AGENT_ID, canned_response)
        logger.info(f"Success. Message ID: {result['id']}")

    except Exception as e:
        logger.error(f"Execution failed: {e}")
        sys.exit(1)

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 403 Forbidden

What causes it: The OAuth token lacks the conversation:chat:write scope, or the service account does not have the necessary permissions in the Genesys Cloud admin console.
How to fix it:

1. Verify the scope parameter in the get_access_token function includes conversation:chat:write.
2. In the Genesys Cloud Admin, navigate to Manage > Users > Service Accounts. Ensure the service account has a role that permits “Send Chat Messages” or similar permissions.

Error: 404 Not Found

What causes it: The conversationId is invalid, does not exist, or the conversation has ended and been archived. The Conversations API only allows messaging on active conversations.
How to fix it:

1. Verify the conversationId is correct.
2. Check the conversation status via GET /api/v2/conversations/chats/{conversationId}. If the status is ended, you cannot send new messages via this endpoint.

Error: 409 Conflict

What causes it: The conversation is in a state that prevents messaging (e.g., bridging, or already closed by the system).
How to fix it: Ensure the conversation is in an active state. Do not attempt to send messages after the customer has disconnected if the conversation is automatically closing.

Error: Message Not Appearing in UI

What causes it: The from field in the payload is missing or contains an invalid userId.
How to fix it: Always include the from object with a valid Genesys Cloud User ID. If you omit this, the system may reject the message or attribute it incorrectly, causing it to fail visibility checks in the agent desktop.

Official References

• Send Chat Message API Documentation
• Canned Responses API Documentation
• Python SDK Reference