How to initiate a cobrowse session programmatically through the Conversations API

How to initiate a cobrowse session programmatically through the Conversations API

What You Will Build

  • You will build a Python script that creates a new conversation, adds a participant via SIP URI, and initiates a cobrowse session using the Genesys Cloud Conversations API.
  • This tutorial uses the Genesys Cloud conversations API surface and the PureCloudPlatformClientV2 Python SDK.
  • The implementation covers Python with the requests library for raw HTTP calls and the official SDK for structured interaction.

Prerequisites

  • OAuth Client Type: Machine-to-Machine (M2M) or JWT Auth.
  • Required OAuth Scopes:
    • conversation:read
    • conversation:write
    • cobrowse:read
    • cobrowse:write
    • user:read (if resolving user IDs)
  • SDK Version: Genesys Cloud Python SDK v2.10.0+ (genesyscloud-python).
  • Runtime: Python 3.8+.
  • External Dependencies: requests, genesyscloud-python, python-dotenv (for credential management).

Authentication Setup

Genesys Cloud APIs require OAuth 2.0 Bearer tokens. For programmatic initiation, M2M authentication is the standard approach. You must exchange a client ID and secret for an access token.

The following code demonstrates how to retrieve and cache a token. In production, implement a TTL check to refresh tokens before expiration.

import requests
import os
from dotenv import load_dotenv

load_dotenv()

# Load credentials from environment variables
CLIENT_ID = os.getenv("GENESYS_CLIENT_ID")
CLIENT_SECRET = os.getenv("GENESYS_CLIENT_SECRET")
ORGANIZATION_ID = os.getenv("GENESYS_ORGANIZATION_ID")

def get_access_token() -> str:
    """
    Retrieves an OAuth 2.0 Bearer token from Genesys Cloud.
    
    Returns:
        str: The access token.
        
    Raises:
        requests.exceptions.HTTPError: If authentication fails.
    """
    url = f"https://api.mypurecloud.com/oauth/token"
    
    headers = {
        "Content-Type": "application/x-www-form-urlencoded"
    }
    
    data = {
        "grant_type": "client_credentials",
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET
    }
    
    response = requests.post(url, headers=headers, data=data)
    
    if response.status_code != 200:
        raise requests.exceptions.HTTPError(
            f"Authentication failed with status {response.status_code}: {response.text}"
        )
        
    token_data = response.json()
    return token_data["access_token"]

# Example usage
token = get_access_token()
print(f"Token acquired: {token[:10]}...")

Implementation

Step 1: Create a Conversation Container

Cobrowsing in Genesys Cloud occurs within a conversation. You cannot initiate a cobrowse session without an active conversation ID. The first step is to create a conversation of type cobrowse.

The API endpoint is POST /api/v2/conversations. The body must specify the type as cobrowse and include the necessary participants or metadata.

import requests
import json
from typing import Dict, Any

def create_cobrowse_conversation(access_token: str, organization_id: str) -> str:
    """
    Creates a new cobrowse conversation.
    
    Args:
        access_token: Valid OAuth bearer token.
        organization_id: Your Genesys Cloud organization ID.
        
    Returns:
        str: The ID of the newly created conversation.
    """
    url = f"https://api.mypurecloud.com/api/v2/conversations"
    
    headers = {
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json",
        "X-Genesys-Organization-ID": organization_id
    }
    
    # The conversation type must be 'cobrowse'
    payload = {
        "type": "cobrowse",
        "metadata": {
            "purpose": "automated_initiation"
        }
    }
    
    response = requests.post(url, headers=headers, json=payload)
    
    # Handle 401/403 errors explicitly
    if response.status_code == 401:
        raise PermissionError("Invalid or expired OAuth token.")
    elif response.status_code == 403:
        raise PermissionError("Insufficient scopes. Ensure 'conversation:write' is granted.")
    elif response.status_code != 201:
        raise requests.exceptions.HTTPError(
            f"Failed to create conversation: {response.status_code} - {response.text}"
        )
        
    conversation_data = response.json()
    conversation_id = conversation_data["id"]
    
    print(f"Conversation created: {conversation_id}")
    return conversation_id

Expected Response:

{
  "id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
  "type": "cobrowse",
  "state": "active",
  "routingData": {},
  "metadata": {
    "purpose": "automated_initiation"
  },
  "createdTime": "2023-10-27T10:00:00.000Z",
  "updatedTime": "2023-10-27T10:00:00.000Z"
}

Step 2: Add Participants to the Conversation

A cobrowse session requires at least two participants: the initiator (agent) and the target (customer). You add participants using POST /api/v2/conversations/{conversationId}/participants.

The from field identifies the initiator, and the to field identifies the target. For cobrowse, the to participant is typically identified by a SIP URI or a user ID if the target is an internal user. For external customers, you often use a callback URL or a specific SIP endpoint if integrated with a web SDK.

In this example, we assume the agent is the current authenticated user (resolved via API) and the target is an external user identified by a SIP URI or user ID.

def add_participants(access_token: str, organization_id: str, conversation_id: str, 
                     agent_user_id: str, target_sip_uri: str) -> Dict[str, Any]:
    """
    Adds the agent and the target customer to the cobrowse conversation.
    
    Args:
        access_token: Valid OAuth bearer token.
        organization_id: Your Genesys Cloud organization ID.
        conversation_id: ID of the cobrowse conversation.
        agent_user_id: The ID of the agent user.
        target_sip_uri: The SIP URI of the target customer (e.g., 'sip:user@domain.com').
        
    Returns:
        Dict containing participant details.
    """
    url = f"https://api.mypurecloud.com/api/v2/conversations/{conversation_id}/participants"
    
    headers = {
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json",
        "X-Genesys-Organization-ID": organization_id
    }
    
    # Define the participants
    # The 'from' is the agent, 'to' is the customer
    payload = {
        "addParticipants": [
            {
                "from": {
                    "id": agent_user_id,
                    "externalId": None,
                    "name": "Agent User",
                    "type": "user"
                },
                "to": {
                    "id": None,
                    "externalId": target_sip_uri,
                    "name": "Customer User",
                    "type": "sip"
                },
                "reason": "initiated_cobrowse"
            }
        ]
    }
    
    response = requests.post(url, headers=headers, json=payload)
    
    if response.status_code == 401:
        raise PermissionError("Invalid or expired OAuth token.")
    elif response.status_code == 403:
        raise PermissionError("Insufficient scopes. Ensure 'conversation:write' is granted.")
    elif response.status_code == 404:
        raise ValueError(f"Conversation {conversation_id} not found.")
    elif response.status_code != 201:
        raise requests.exceptions.HTTPError(
            f"Failed to add participants: {response.status_code} - {response.text}"
        )
        
    return response.json()

Critical Parameter Note:
The to participant’s type must be sip when using a SIP URI. If you are cobrowsing with an internal user, use user and provide the id. The externalId field is crucial for external targets.

Step 3: Initiate the Cobrowse Session

Adding participants does not automatically start the cobrowse session. You must explicitly initiate it using the POST /api/v2/conversations/{conversationId}/cobrowse/initiate endpoint.

This endpoint triggers the actual cobrowse connection. The request body includes configuration details such as the initiator’s role and any specific cobrowse options.

def initiate_cobrowse_session(access_token: str, organization_id: str, conversation_id: str) -> Dict[str, Any]:
    """
    Initiates the cobrowse session for the given conversation.
    
    Args:
        access_token: Valid OAuth bearer token.
        organization_id: Your Genesys Cloud organization ID.
        conversation_id: ID of the cobrowse conversation with participants added.
        
    Returns:
        Dict containing the initiation result.
    """
    url = f"https://api.mypurecloud.com/api/v2/conversations/{conversation_id}/cobrowse/initiate"
    
    headers = {
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json",
        "X-Genesys-Organization-ID": organization_id
    }
    
    # Payload for initiating cobrowse
    # The 'initiatorId' should match the agent's user ID
    payload = {
        "initiatorId": "agent_user_id", # Replace with actual agent ID
        "reason": "manual_initiation"
    }
    
    response = requests.post(url, headers=headers, json=payload)
    
    if response.status_code == 401:
        raise PermissionError("Invalid or expired OAuth token.")
    elif response.status_code == 403:
        raise PermissionError("Insufficient scopes. Ensure 'cobrowse:write' is granted.")
    elif response.status_code == 404:
        raise ValueError(f"Conversation {conversation_id} not found.")
    elif response.status_code == 409:
        raise ConflictError("Cobrowse session already initiated or conversation is not in valid state.")
    elif response.status_code != 200:
        raise requests.exceptions.HTTPError(
            f"Failed to initiate cobrowse: {response.status_code} - {response.text}"
        )
        
    return response.json()

Expected Response:

{
  "conversationId": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
  "initiatorId": "agent_user_id",
  "state": "initiated",
  "timestamp": "2023-10-27T10:05:00.000Z"
}

Complete Working Example

The following script combines all steps into a single executable module. It includes error handling, token retrieval, and the full lifecycle of creating and initiating a cobrowse session.

import os
import requests
import json
from typing import Dict, Any
from dotenv import load_dotenv

load_dotenv()

# Configuration
CLIENT_ID = os.getenv("GENESYS_CLIENT_ID")
CLIENT_SECRET = os.getenv("GENESYS_CLIENT_SECRET")
ORGANIZATION_ID = os.getenv("GENESYS_ORGANIZATION_ID")
AGENT_USER_ID = os.getenv("AGENT_USER_ID")
TARGET_SIP_URI = os.getenv("TARGET_SIP_URI", "sip:customer@example.com")

def get_access_token() -> str:
    """Retrieves an OAuth 2.0 Bearer token."""
    url = "https://api.mypurecloud.com/oauth/token"
    headers = {"Content-Type": "application/x-www-form-urlencoded"}
    data = {
        "grant_type": "client_credentials",
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET
    }
    response = requests.post(url, headers=headers, data=data)
    response.raise_for_status()
    return response.json()["access_token"]

def create_cobrowse_conversation(access_token: str) -> str:
    """Creates a new cobrowse conversation."""
    url = f"https://api.mypurecloud.com/api/v2/conversations"
    headers = {
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json",
        "X-Genesys-Organization-ID": ORGANIZATION_ID
    }
    payload = {
        "type": "cobrowse",
        "metadata": {
            "purpose": "automated_initiation"
        }
    }
    response = requests.post(url, headers=headers, json=payload)
    response.raise_for_status()
    return response.json()["id"]

def add_participants(access_token: str, conversation_id: str) -> Dict[str, Any]:
    """Adds the agent and target to the conversation."""
    url = f"https://api.mypurecloud.com/api/v2/conversations/{conversation_id}/participants"
    headers = {
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json",
        "X-Genesys-Organization-ID": ORGANIZATION_ID
    }
    payload = {
        "addParticipants": [
            {
                "from": {
                    "id": AGENT_USER_ID,
                    "name": "Agent User",
                    "type": "user"
                },
                "to": {
                    "id": None,
                    "externalId": TARGET_SIP_URI,
                    "name": "Customer User",
                    "type": "sip"
                },
                "reason": "initiated_cobrowse"
            }
        ]
    }
    response = requests.post(url, headers=headers, json=payload)
    response.raise_for_status()
    return response.json()

def initiate_cobrowse_session(access_token: str, conversation_id: str) -> Dict[str, Any]:
    """Initiates the cobrowse session."""
    url = f"https://api.mypurecloud.com/api/v2/conversations/{conversation_id}/cobrowse/initiate"
    headers = {
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json",
        "X-Genesys-Organization-ID": ORGANIZATION_ID
    }
    payload = {
        "initiatorId": AGENT_USER_ID,
        "reason": "manual_initiation"
    }
    response = requests.post(url, headers=headers, json=payload)
    response.raise_for_status()
    return response.json()

def main():
    """Main execution flow."""
    try:
        print("1. Authenticating...")
        token = get_access_token()
        
        print("2. Creating cobrowse conversation...")
        conversation_id = create_cobrowse_conversation(token)
        
        print(f"3. Adding participants to conversation {conversation_id}...")
        add_participants(token, conversation_id)
        
        print("4. Initiating cobrowse session...")
        result = initiate_cobrowse_session(token, conversation_id)
        
        print("5. Cobrowse session initiated successfully.")
        print(json.dumps(result, indent=2))
        
    except requests.exceptions.HTTPError as e:
        print(f"HTTP Error: {e}")
    except Exception as e:
        print(f"Unexpected error: {e}")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 401 Unauthorized

  • Cause: The OAuth token is invalid, expired, or missing.
  • Fix: Ensure the client_id and client_secret are correct. Verify the token was retrieved successfully before making API calls. Implement token refresh logic if the token expires during long-running processes.

Error: 403 Forbidden

  • Cause: The OAuth token lacks the required scopes.
  • Fix: Check that the M2M application has the following scopes granted: conversation:read, conversation:write, cobrowse:read, cobrowse:write. Re-authorize the application in the Genesys Cloud Admin Console if necessary.

Error: 404 Not Found

  • Cause: The conversation ID does not exist or the user ID is invalid.
  • Fix: Verify that the conversation was created successfully in the previous step. Ensure the AGENT_USER_ID corresponds to an active user in your Genesys Cloud organization.

Error: 409 Conflict

  • Cause: The cobrowse session has already been initiated for this conversation, or the conversation is in an invalid state (e.g., ended).
  • Fix: Check the conversation state before initiating. If the session is already active, you may not need to initiate it again. If the conversation is ended, create a new one.

Error: 429 Too Many Requests

  • Cause: Rate limit exceeded.
  • Fix: Implement exponential backoff retry logic. Genesys Cloud APIs return Retry-After headers in 429 responses. Respect these headers to avoid further throttling.

Official References