Sending Structured Messages (Quick Replies, Cards) via the Genesys Cloud Open Messaging API

What You Will Build

• This tutorial demonstrates how to programmatically send rich, structured messages containing quick reply buttons and informational cards to a user within a Genesys Cloud Messaging conversation.
• The implementation uses the Genesys Cloud Python SDK (genesys-cloud-purecloud-platform-client) to interact with the Messaging API.
• The code is written in Python 3.9+ and requires an OAuth application configured with the appropriate messaging scopes.

Prerequisites

OAuth Configuration

• Client Type: Confidential Client (Client Credentials Grant) or Private Key JWT.
• Required Scopes:
  • messaging:conversation:write: Required to send messages to a conversation.
  • messaging:conversation:read: Required to retrieve conversation details if needed for context.
  • user:read: Often required to verify the identity of the sending agent or bot.

SDK Installation

• Install the official Genesys Cloud Python SDK.

  pip install purecloudplatformclientv2
  

Runtime Requirements

• Python 3.9 or higher.
• A valid Genesys Cloud Organization ID.
• A configured Messaging Channel (e.g., SMS, Web Chat, WhatsApp) where you have an active conversation.

Authentication Setup

The Genesys Cloud API requires OAuth 2.0 authentication. For server-to-server integrations, the Client Credentials flow is standard. The SDK handles token caching and refreshing automatically, but you must configure the client credentials correctly.

import os
from purecloudplatformclientv2 import Configuration, ApiClient, MessagingApi

def get_authenticated_messaging_api() -> MessagingApi:
    """
    Initializes and returns an authenticated MessagingApi instance.
    """
    # Load credentials from environment variables
    client_id = os.getenv("GENESYS_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLIENT_SECRET")
    environment = os.getenv("GENESYS_ENVIRONMENT", "us-east-1") # e.g., us-east-1, eu-west-1

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

    # Configure the SDK
    config = Configuration(
        host=f"https://api.{environment}.mygenesys.com",
        client_id=client_id,
        client_secret=client_secret
    )

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

    # Initialize the Messaging API instance
    messaging_api = MessagingApi(api_client)

    return messaging_api

Note: The Configuration object caches the access token. Subsequent calls to the API will reuse the token until it expires, at which point the SDK automatically refreshes it. Do not create a new Configuration or ApiClient object for every request, as this bypasses the cache and increases latency.

Implementation

Step 1: Constructing the Structured Message Payload

The Genesys Cloud Open Messaging API uses a specific JSON structure defined by the Message model. To send rich content, you must use the content field with a contentType that supports structure, such as application/vnd.genesis.message.card+json or application/vnd.genesis.message.quick-reply+json.

For this tutorial, we will construct a message containing a Card with Quick Reply buttons. This requires building a complex nested object.

from purecloudplatformclientv2 import (
    Message,
    MessageContent,
    Card,
    CardSection,
    QuickReply,
    QuickReplyButton
)

def build_rich_message_payload() -> Message:
    """
    Constructs a Message object containing a Card with Quick Reply buttons.
    """
    # Define the quick reply buttons
    button_1 = QuickReplyButton(
        display_text="Yes",
        value="confirmed_yes"
    )
    button_2 = QuickReplyButton(
        display_text="No",
        value="confirmed_no"
    )

    # Define the card section
    # Note: The title and description are optional but recommended for accessibility
    card_section = CardSection(
        title="Order Confirmation",
        description="Would you like to confirm your order #12345?",
        quick_replies=[QuickReply(button=button_1), QuickReply(button=button_2)]
    )

    # Define the card
    # The type 'generic' is standard for most rich messages
    message_card = Card(
        card_type="generic",
        sections=[card_section]
    )

    # Define the message content
    # The contentType must match the structure of the content object
    message_content = MessageContent(
        content_type="application/vnd.genesis.message.card+json",
        content=message_card
    )

    # Create the final Message object
    # The 'from' field identifies the sender (usually a bot or agent ID)
    # The 'to' field is typically omitted in the send request, as it is derived from the conversation
    message = Message(
        content=message_content,
        from_id="bot-id-here" # Replace with your actual Bot or User ID
    )

    return message

Key Parameters Explained:

• content_type: Must be application/vnd.genesis.message.card+json for cards. If you were sending only quick replies without a card structure, you might use application/vnd.genesis.message.quick-reply+json.
• quick_replies: An array of QuickReply objects. Each QuickReply contains a button object with display_text (what the user sees) and value (what is sent back to the API when clicked).
• from_id: The ID of the entity sending the message. This must be a valid User ID (for agents) or Bot ID. If you use an invalid ID, the API returns a 400 Bad Request.

Step 2: Sending the Message to a Conversation

Once the payload is constructed, you send it to the MessagingApi. The endpoint is POST /api/v2/messaging/conversations/{conversationId}/messages.

from purecloudplatformclientv2.rest import ApiException

def send_rich_message(messaging_api: MessagingApi, conversation_id: str, message: Message) -> dict:
    """
    Sends a structured message to a specific conversation.
    
    Args:
        messaging_api: The authenticated MessagingApi instance.
        conversation_id: The ID of the target conversation.
        message: The Message object to send.
    
    Returns:
        The response body from the API.
    """
    try:
        # The SDK method maps to POST /api/v2/messaging/conversations/{conversationId}/messages
        response = messaging_api.post_messaging_conversations_message(
            conversation_id=conversation_id,
            body=message
        )
        
        # Log success
        print(f"Message sent successfully to conversation {conversation_id}")
        return response.body
        
    except ApiException as e:
        # Handle specific API errors
        if e.status == 401:
            print("Authentication failed. Check your OAuth token and scopes.")
        elif e.status == 403:
            print("Forbidden. Ensure the client has 'messaging:conversation:write' scope.")
        elif e.status == 404:
            print(f"Conversation {conversation_id} not found.")
        elif e.status == 400:
            print(f"Bad Request. Check the message structure. Error: {e.body}")
        elif e.status == 429:
            print("Rate limited. Implement exponential backoff.")
        else:
            print(f"Unexpected API Error: {e.status} - {e.body}")
        
        raise e

Error Handling Strategy:

• 400 Bad Request: Often caused by malformed JSON or invalid content_type. Validate the Message object before sending.
• 403 Forbidden: Usually indicates missing scopes. Verify that messaging:conversation:write is granted to your OAuth client.
• 429 Too Many Requests: Genesys Cloud enforces rate limits. If you hit this, implement a retry mechanism with exponential backoff. The SDK does not handle retries automatically for all operations, so explicit handling is recommended for production systems.

Step 3: Handling Asynchronous Delivery and Status

The post_messaging_conversations_message call is asynchronous from the perspective of message delivery. The API returns 202 Accepted immediately, indicating that the message has been queued for delivery. It does not guarantee that the user has received it.

To track delivery status, you must poll the GET /api/v2/messaging/conversations/{conversationId}/messages endpoint or use Webhooks.

import time

def wait_for_message_delivery(messaging_api: MessagingApi, conversation_id: str, message_id: str, timeout_seconds: int = 30) -> bool:
    """
    Polls the conversation to check if the message has been delivered.
    
    Args:
        messaging_api: The authenticated MessagingApi instance.
        conversation_id: The ID of the target conversation.
        message_id: The ID of the sent message.
        timeout_seconds: Maximum time to wait for delivery.
    
    Returns:
        True if delivered, False if timeout.
    """
    start_time = time.time()
    
    while time.time() - start_time < timeout_seconds:
        try:
            # Retrieve messages from the conversation
            # We filter by the specific message ID if possible, or iterate through recent messages
            messages_response = messaging_api.get_messaging_conversations_messages(
                conversation_id=conversation_id,
                limit=10 # Retrieve last 10 messages
            )
            
            # Check if our message is in the response and has a delivery status
            for msg in messages_response.body.messages:
                if msg.id == message_id:
                    # The 'status' field indicates delivery state
                    # Common statuses: 'sent', 'delivered', 'read', 'failed'
                    if msg.status in ['delivered', 'read']:
                        return True
                    elif msg.status == 'failed':
                        print(f"Message delivery failed: {msg.delivery_status}")
                        return False
            
            # Wait before polling again
            time.sleep(2)
            
        except ApiException as e:
            print(f"Error polling for message status: {e.body}")
            return False
            
    return False

Note: In production, polling is inefficient. Use Genesys Cloud Webhooks to subscribe to conversation:message:received or conversation:message:updated events. This pushes delivery status updates to your server in real-time.

Complete Working Example

This script combines authentication, payload construction, sending, and basic status checking. Replace the placeholder values with your actual Genesys Cloud credentials and IDs.

import os
import time
from purecloudplatformclientv2 import (
    Configuration, ApiClient, MessagingApi,
    Message, MessageContent, Card, CardSection,
    QuickReply, QuickReplyButton
)
from purecloudplatformclientv2.rest import ApiException

def main():
    # 1. Authenticate
    try:
        messaging_api = get_authenticated_messaging_api()
    except Exception as e:
        print(f"Failed to authenticate: {e}")
        return

    # 2. Define Target Conversation
    # Replace with a valid conversation ID from your Genesys Cloud instance
    conversation_id = os.getenv("GENESYS_CONVERSATION_ID")
    if not conversation_id:
        print("GENESYS_CONVERSATION_ID environment variable not set.")
        return

    # 3. Build Rich Message
    try:
        message = build_rich_message_payload()
    except Exception as e:
        print(f"Failed to build message payload: {e}")
        return

    # 4. Send Message
    try:
        response = send_rich_message(messaging_api, conversation_id, message)
        print(f"Message sent. Response: {response}")
        
        # The response body typically contains the message ID
        message_id = response.get('id')
        if message_id:
            print(f"Message ID: {message_id}")
            
            # 5. Wait for Delivery (Optional)
            # In production, use Webhooks instead of polling
            is_delivered = wait_for_message_delivery(messaging_api, conversation_id, message_id)
            if is_delivered:
                print("Message delivered successfully.")
            else:
                print("Message delivery status unknown or timed out.")
                
    except ApiException as e:
        print(f"API Error: {e.body}")

def get_authenticated_messaging_api() -> MessagingApi:
    client_id = os.getenv("GENESYS_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLIENT_SECRET")
    environment = os.getenv("GENESYS_ENVIRONMENT", "us-east-1")

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

    config = Configuration(
        host=f"https://api.{environment}.mygenesys.com",
        client_id=client_id,
        client_secret=client_secret
    )
    api_client = ApiClient(configuration=config)
    return MessagingApi(api_client)

def build_rich_message_payload() -> Message:
    button_1 = QuickReplyButton(display_text="Yes", value="confirmed_yes")
    button_2 = QuickReplyButton(display_text="No", value="confirmed_no")

    card_section = CardSection(
        title="Order Confirmation",
        description="Would you like to confirm your order #12345?",
        quick_replies=[QuickReply(button=button_1), QuickReply(button=button_2)]
    )

    message_card = Card(card_type="generic", sections=[card_section])

    message_content = MessageContent(
        content_type="application/vnd.genesis.message.card+json",
        content=message_card
    )

    # Use a valid Bot ID or User ID
    sender_id = os.getenv("GENESYS_SENDER_ID", "bot-id-placeholder")
    
    return Message(content=message_content, from_id=sender_id)

def send_rich_message(messaging_api: MessagingApi, conversation_id: str, message: Message) -> dict:
    try:
        response = messaging_api.post_messaging_conversations_message(
            conversation_id=conversation_id,
            body=message
        )
        return response.body
    except ApiException as e:
        print(f"Send Error: {e.status} - {e.body}")
        raise e

def wait_for_message_delivery(messaging_api: MessagingApi, conversation_id: str, message_id: str, timeout_seconds: int = 30) -> bool:
    start_time = time.time()
    while time.time() - start_time < timeout_seconds:
        try:
            messages_response = messaging_api.get_messaging_conversations_messages(
                conversation_id=conversation_id,
                limit=10
            )
            for msg in messages_response.body.messages:
                if msg.id == message_id:
                    if msg.status in ['delivered', 'read']:
                        return True
                    elif msg.status == 'failed':
                        return False
            time.sleep(2)
        except ApiException as e:
            print(f"Polling Error: {e.body}")
            return False
    return False

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 400 Bad Request - Invalid Content Type

• Cause: The content_type in the MessageContent object does not match the structure of the content object. For example, sending a Card object with content_type="text/plain".
• Fix: Ensure content_type is application/vnd.genesis.message.card+json when sending a Card object. Validate the JSON structure against the OpenAPI specification.

Error: 403 Forbidden - Insufficient Scope

• Cause: The OAuth client lacks the messaging:conversation:write scope.
• Fix: Navigate to the Genesys Cloud Admin Console > Integrations > OAuth Applications. Edit your client and add the messaging:conversation:write scope. Re-authorize the application if necessary.

Error: 429 Too Many Requests

• Cause: Exceeding the rate limit for the Messaging API. The limit varies by organization but is typically around 100 requests per second for write operations.
• Fix: Implement exponential backoff. If you receive a 429, wait for a duration equal to 2^retry_count seconds before retrying. Increase the retry_count with each subsequent failure.

Error: Message Not Appearing in Client

• Cause: The message was sent to the correct conversation ID, but the channel (e.g., SMS, WhatsApp) does not support rich cards.
• Fix: Verify the channel capabilities. SMS and some legacy channels do not support Card objects. They only support text/plain. WhatsApp and Web Chat support rich cards. If sending to SMS, use a simple text message instead.

Official References

• Genesys Cloud Messaging API Documentation
• Open Messaging API Specification
• Python SDK Reference
• OAuth Scopes Reference