Implementing the Guest API to Send and Receive Messages Without the Messenger Widget

What You Will Build

• A headless chat client that authenticates as an anonymous guest and exchanges messages with a Genesys Cloud agent.
• Implementation uses the Genesys Cloud Conversational Messaging Guest API (/api/v2/conversations/messaging/guests) and WebSocket streaming.
• The tutorial covers Python using the websockets library and requests for HTTP operations.

Prerequisites

• OAuth Client: A confidential client application registered in Genesys Cloud with the webchat or api grant type.
• Required Scopes: webchat:guest:read, webchat:guest:write, webchat:guest:send.
• SDK Version: Genesys Cloud Python SDK genesyscloud v13.0.0+ (though this tutorial uses direct HTTP/WebSocket for lower-level control and clarity).
• Runtime: Python 3.9+.
• Dependencies: pip install requests websockets python-dotenv.
• Environment: A Genesys Cloud organization with Messaging enabled and a Queue configured to accept web chats.

Authentication Setup

The Guest API requires two distinct authentication steps. First, you must authenticate your application to Genesys Cloud to obtain a bearer token. Second, you must register a “Guest” identity with the Messaging service. This guest identity is temporary and tied to the specific session.

Step 1: Application OAuth Token

You need a valid OAuth 2.0 bearer token to call the Guest registration endpoint. Use the client_credentials flow.

import requests
import os
from urllib.parse import quote

class GenesysAuth:
    def __init__(self, client_id: str, client_secret: str, base_url: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.base_url = base_url.rstrip('/')
        self.token_endpoint = f"{self.base_url}/oauth/token"

    def get_token(self) -> str:
        """
        Retrieves a short-lived OAuth token for the application.
        """
        auth_header = f"{quote(self.client_id, safe='')}:{quote(self.client_secret, safe='')}"
        headers = {
            "Authorization": f"Basic {auth_header}",
            "Content-Type": "application/x-www-form-urlencoded"
        }
        data = {
            "grant_type": "client_credentials",
            "scope": "webchat:guest:read webchat:guest:write webchat:guest:send"
        }
        
        response = requests.post(self.token_endpoint, headers=headers, data=data)
        response.raise_for_status()
        
        token_data = response.json()
        return token_data['access_token']

# Usage
AUTH = GenesysAuth(
    client_id=os.getenv("GENESYS_CLIENT_ID"),
    client_secret=os.getenv("GENESYS_CLIENT_SECRET"),
    base_url=os.getenv("GENESYS_BASE_URL") # e.g., https://api.mypurecloud.com
)
OAUTH_TOKEN = AUTH.get_token()

Step 2: Registering the Guest

With the bearer token, you register a guest. This returns a guestId and a webChatToken. The webChatToken is critical; it is used to authenticate the WebSocket connection.

Endpoint: POST /api/v2/conversations/messaging/guests

Required Header: Authorization: Bearer <token>

def register_guest(oauth_token: str, base_url: str, queue_id: str) -> dict:
    """
    Registers a new guest session.
    
    Args:
        oauth_token: The bearer token from Step 1.
        base_url: The Genesys API base URL.
        queue_id: The ID of the queue the guest wants to chat with.
        
    Returns:
        Dictionary containing guestId, webChatToken, and other session metadata.
    """
    endpoint = f"{base_url}/api/v2/conversations/messaging/guests"
    
    headers = {
        "Authorization": f"Bearer {oauth_token}",
        "Content-Type": "application/json"
    }
    
    payload = {
        "name": "Anonymous Guest",
        "email": "guest@example.com",
        "locale": "en-US",
        "queueId": queue_id,
        "routingData": {
            "priority": 1,
            "skillRequirements": []
        }
    }
    
    response = requests.post(endpoint, headers=headers, json=payload)
    
    if response.status_code == 400:
        print(f"Bad Request: {response.json()}")
        raise Exception("Guest registration failed. Check queue ID and scopes.")
    elif response.status_code == 401:
        print(f"Unauthorized: Token invalid or expired.")
        raise Exception("OAuth token is invalid.")
        
    response.raise_for_status()
    return response.json()

# Usage
# Assume QUEUE_ID is fetched via API or hardcoded for this demo
QUEUE_ID = os.getenv("GENESYS_QUEUE_ID") 
GUEST_DATA = register_guest(OAUTH_TOKEN, AUTH.base_url, QUEUE_ID)
GUEST_ID = GUEST_DATA['id']
WEB_CHAT_TOKEN = GUEST_DATA['webChatToken']

Implementation

Step 1: Establishing the WebSocket Connection

Genesys Cloud uses WebSockets for real-time messaging. The endpoint is not the standard API base URL but a specific messaging host. You must upgrade the HTTP connection to a WebSocket using the webChatToken as the query parameter.

WebSocket URL: wss://messaging.mypurecloud.com/api/v2/conversations/messaging/guests/{guestId}/websocket

Authentication: The webChatToken is passed as a query parameter ?webChatToken={token}.

import websockets
import json
import asyncio

async def connect_websocket(base_url: str, guest_id: str, web_chat_token: str):
    """
    Connects to the Genesys Messaging WebSocket.
    """
    # Determine the messaging host. 
    # Note: The host is typically derived from the base URL domain.
    # For standard Genesys Cloud, it is messaging.{domain}
    domain = base_url.split('//')[1].split('/')[0]
    ws_host = f"wss://messaging.{domain}/api/v2/conversations/messaging/guests/{guest_id}/websocket"
    
    # Append the webChatToken for authentication
    ws_uri = f"{ws_host}?webChatToken={web_chat_token}"
    
    print(f"Connecting to WebSocket: {ws_uri}")
    
    try:
        async with websockets.connect(ws_uri) as websocket:
            print("WebSocket connected successfully.")
            return websocket
    except Exception as e:
        print(f"Failed to connect to WebSocket: {e}")
        raise

Step 2: Sending Messages

To send a message, you do not use the HTTP API. You send a JSON payload over the established WebSocket connection. The payload must conform to the ConversationMessage structure.

Key Fields:

• type: Must be "message".
• conversationId: The ID of the conversation created during guest registration (available in GUEST_DATA['conversationId']).
• text: The actual message content.
• senderId: Must match the guestId.

async def send_message(websocket, conversation_id: str, guest_id: str, text: str) -> dict:
    """
    Sends a text message to the agent via WebSocket.
    
    Args:
        websocket: The active WebSocket connection.
        conversation_id: The ID of the conversation from guest registration.
        guest_id: The ID of the guest.
        text: The message string.
        
    Returns:
        The server's acknowledgment response.
    """
    message_payload = {
        "type": "message",
        "conversationId": conversation_id,
        "senderId": guest_id,
        "text": text,
        "messageType": "text"
    }
    
    # Serialize to JSON
    json_payload = json.dumps(message_payload)
    
    # Send over WebSocket
    await websocket.send(json_payload)
    print(f"Sent: {text}")
    
    # Wait for acknowledgment
    # Note: Genesys may send multiple messages (ack, agent typing, etc.)
    # We look for the specific ack for our message.
    response = await asyncio.wait_for(websocket.recv(), timeout=10)
    resp_data = json.loads(response)
    
    if resp_data.get('type') == 'ack':
        print(f"Message acknowledged. Message ID: {resp_data.get('messageId')}")
        return resp_data
    else:
        print(f"Unexpected response type: {resp_data.get('type')}")
        return resp_data

Step 3: Receiving Messages and Handling Events

The WebSocket stream is bidirectional. You must listen for incoming messages. Common message types include:

• message: Text from the agent.
• ack: Acknowledgment of your sent message.
• typing: Agent is typing.
• conversationMetadata: Updates to the conversation state (e.g., agent joined, closed).

async def listen_for_messages(websocket, duration_seconds: int = 60):
    """
    Listens for incoming messages from the agent or system.
    
    Args:
        websocket: The active WebSocket connection.
        duration_seconds: How long to listen before timing out.
    """
    print(f"Listening for messages for {duration_seconds} seconds...")
    
    try:
        while True:
            # Use wait_for to allow for a timeout
            message = await asyncio.wait_for(websocket.recv(), timeout=duration_seconds)
            data = json.loads(message)
            
            msg_type = data.get('type')
            
            if msg_type == 'message':
                sender = data.get('senderId')
                text = data.get('text', '')
                print(f"\n[Agent] {text}")
                
            elif msg_type == 'typing':
                sender = data.get('senderId')
                print(f"\n[Agent is typing...]")
                
            elif msg_type == 'ack':
                # Already handled in send_message, but good to log if received unexpectedly
                pass
                
            elif msg_type == 'conversationMetadata':
                event = data.get('event')
                print(f"\n[Conversation Event] {event}")
                if event == 'closed':
                    print("Conversation has been closed by the agent.")
                    break
                    
    except asyncio.TimeoutError:
        print("\nListening timed out.")
    except websockets.exceptions.ConnectionClosed:
        print("\nWebSocket connection closed.")

Complete Working Example

This script combines authentication, guest registration, WebSocket connection, and message exchange. It sends a greeting, waits for an agent response, and then sends a follow-up.

import os
import asyncio
import json
import requests
from urllib.parse import quote
import websockets
from dotenv import load_dotenv

# Load environment variables
load_dotenv()

class GenesysHeadlessChat:
    def __init__(self):
        self.client_id = os.getenv("GENESYS_CLIENT_ID")
        self.client_secret = os.getenv("GENESYS_CLIENT_SECRET")
        self.base_url = os.getenv("GENESYS_BASE_URL") # e.g., https://api.mypurecloud.com
        self.queue_id = os.getenv("GENESYS_QUEUE_ID")
        self.oauth_token = None
        self.guest_data = None
        self.websocket = None

    async def run(self):
        try:
            # 1. Authenticate Application
            print("Step 1: Authenticating Application...")
            self.oauth_token = self._get_oauth_token()
            
            # 2. Register Guest
            print("Step 2: Registering Guest...")
            self.guest_data = self._register_guest()
            
            guest_id = self.guest_data['id']
            conversation_id = self.guest_data['conversationId']
            web_chat_token = self.guest_data['webChatToken']
            
            print(f"Guest ID: {guest_id}")
            print(f"Conversation ID: {conversation_id}")
            
            # 3. Connect WebSocket
            print("Step 3: Connecting to WebSocket...")
            self.websocket = await self._connect_websocket(web_chat_token)
            
            # 4. Send Initial Message
            print("Step 4: Sending Initial Message...")
            await self._send_message(conversation_id, guest_id, "Hello, I need assistance with my order.")
            
            # 5. Listen for Response
            print("Step 5: Listening for Agent Response...")
            await self._listen_for_messages(timeout_seconds=30)
            
            # 6. Send Follow-up (if conversation still open)
            # Note: In a real app, you would check if the conversation is still active
            if self.websocket.open:
                print("Step 6: Sending Follow-up...")
                await self._send_message(conversation_id, guest_id, "Thank you for the quick response.")
                await self._listen_for_messages(timeout_seconds=10)
                
        except Exception as e:
            print(f"Error: {e}")
        finally:
            if self.websocket:
                await self.websocket.close()

    def _get_oauth_token(self) -> str:
        auth_header = f"{quote(self.client_id, safe='')}:{quote(self.client_secret, safe='')}"
        headers = {"Authorization": f"Basic {auth_header}", "Content-Type": "application/x-www-form-urlencoded"}
        data = {"grant_type": "client_credentials", "scope": "webchat:guest:read webchat:guest:write webchat:guest:send"}
        
        response = requests.post(f"{self.base_url}/oauth/token", headers=headers, data=data)
        response.raise_for_status()
        return response.json()['access_token']

    def _register_guest(self) -> dict:
        headers = {"Authorization": f"Bearer {self.oauth_token}", "Content-Type": "application/json"}
        payload = {
            "name": "Test Guest",
            "email": "test@example.com",
            "locale": "en-US",
            "queueId": self.queue_id,
            "routingData": {"priority": 1, "skillRequirements": []}
        }
        
        response = requests.post(f"{self.base_url}/api/v2/conversations/messaging/guests", headers=headers, json=payload)
        response.raise_for_status()
        return response.json()

    async def _connect_websocket(self, web_chat_token: str):
        domain = self.base_url.split('//')[1].split('/')[0]
        guest_id = self.guest_data['id']
        ws_uri = f"wss://messaging.{domain}/api/v2/conversations/messaging/guests/{guest_id}/websocket?webChatToken={web_chat_token}"
        
        try:
            ws = await websockets.connect(ws_uri)
            print("WebSocket Connected.")
            return ws
        except Exception as e:
            raise Exception(f"WebSocket connection failed: {e}")

    async def _send_message(self, conversation_id: str, guest_id: str, text: str):
        payload = {
            "type": "message",
            "conversationId": conversation_id,
            "senderId": guest_id,
            "text": text,
            "messageType": "text"
        }
        await self.websocket.send(json.dumps(payload))
        
        # Wait for ack
        ack = await asyncio.wait_for(self.websocket.recv(), timeout=10)
        ack_data = json.loads(ack)
        if ack_data.get('type') == 'ack':
            print(f"Message Sent & Acked: {text}")
        else:
            print(f"Unexpected Ack: {ack_data}")

    async def _listen_for_messages(self, timeout_seconds: int = 30):
        try:
            while True:
                msg = await asyncio.wait_for(self.websocket.recv(), timeout=timeout_seconds)
                data = json.loads(msg)
                
                if data.get('type') == 'message':
                    sender = data.get('senderId')
                    text = data.get('text', '')
                    print(f"[Agent]: {text}")
                elif data.get('type') == 'typing':
                    print("[Agent is typing...]")
                elif data.get('type') == 'conversationMetadata':
                    event = data.get('event')
                    print(f"[Event]: {event}")
                    if event == 'closed':
                        break
        except asyncio.TimeoutError:
            print("No new messages received within timeout.")
        except websockets.exceptions.ConnectionClosed:
            print("Connection closed by server.")

if __name__ == "__main__":
    chat_client = GenesysHeadlessChat()
    asyncio.run(chat_client.run())

Common Errors & Debugging

Error: 401 Unauthorized on Guest Registration

Cause: The OAuth token is expired, invalid, or lacks the required webchat:guest:write scope.
Fix: Ensure your client credentials are correct. Verify the token using the /oauth/introspect endpoint. Check that the scope webchat:guest:write is included in the token request.

Error: WebSocket Connection Refused or 403

Cause: The webChatToken is invalid, expired, or passed incorrectly.
Fix: Ensure the webChatToken is appended as a query parameter ?webChatToken=... to the WebSocket URL. Do not pass it in the headers. The token is single-use and short-lived; register the guest immediately before connecting.

Error: Message Not Received by Agent

Cause: The conversationId in the message payload does not match the one returned during guest registration.
Fix: Verify that you are using the conversationId from the POST /api/v2/conversations/messaging/guests response. Do not generate a new UUID.

Error: websockets.exceptions.InvalidStatusCode

Cause: The WebSocket handshake failed, often due to an incorrect host or missing authentication.
Fix: Double-check the WebSocket host. It must be messaging.{your-domain}, not api.{your-domain}. Ensure the webChatToken is valid.

Official References

• Genesys Cloud Messaging Guest API Documentation
• Genesys Cloud Conversational Messaging Overview
• Python WebSockets Library Documentation