How to Initiate a Cobrowse Session Programmatically via the Conversations API

How to Initiate a Cobrowse Session Programmatically via the Conversations API

What You Will Build

  • One sentence: This tutorial demonstrates how to programmatically create a cobrowse session and inject it into an active webchat or voice conversation.
  • One sentence: This uses the Genesys Cloud CX Conversations API and the Cobrowse API endpoints.
  • One sentence: The examples are written in Python using the genesys-cloud-python-sdk and httpx for direct API calls.

Prerequisites

  • OAuth Client Type: Confidential Client (Client Credentials Grant) or Public Client (PKCE/Authorization Code). For backend automation, Confidential Client is standard.
  • Required Scopes:
    • cobrowse:initiate (to start the session)
    • conversation:webchat or conversation:voice (depending on the channel)
    • conversation:read (to verify session status)
  • SDK Version: genesys-cloud-python-sdk >= 1.100.0
  • Runtime: Python 3.9+
  • Dependencies:
    • pip install genesys-cloud-python-sdk
    • pip install httpx
    • pip install python-dotenv

Authentication Setup

Genesys Cloud APIs require a valid Bearer token. For programmatic initiation, you typically use the Client Credentials Grant flow if your application runs on a server. If you are building a client-side widget that initiates cobrowse, you would use PKCE. This tutorial assumes a backend service initiating the session.

Security Note: Never expose your client_secret in client-side code.

import httpx
import os
from dotenv import load_dotenv

load_dotenv()

# Configuration from environment variables
REGION = os.getenv("GENESYS_REGION", "mypurecloud.com")
CLIENT_ID = os.getenv("GENESYS_CLIENT_ID")
CLIENT_SECRET = os.getenv("GENESYS_CLIENT_SECRET")

def get_access_token() -> str:
    """
    Retrieves an OAuth 2.0 access token using Client Credentials Grant.
    """
    if not CLIENT_ID or not CLIENT_SECRET:
        raise ValueError("GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET must be set")

    url = f"https://api.{REGION}/oauth/token"
    headers = {
        "Content-Type": "application/x-www-form-urlencoded"
    }
    data = {
        "grant_type": "client_credentials",
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET,
        "scope": "cobrowse:initiate conversation:webchat conversation:read"
    }

    with httpx.Client() as client:
        response = client.post(url, headers=headers, data=data)
        
        if response.status_code != 200:
            raise RuntimeError(f"Failed to obtain token: {response.text}")
            
        return response.json()["access_token"]

# Cache the token in memory or a database in production. 
# For this tutorial, we fetch it fresh.
TOKEN = get_access_token()

Implementation

Step 1: Identify the Active Conversation

Before initiating cobrowse, you must have an active conversation. Cobrowse is a media type that can be added to an existing Webchat or Voice conversation. You cannot initiate cobrowse into a vacuum; it must be associated with a specific conversationId.

If you do not have an active conversation, you must first create one (e.g., via the Webchat API). For this tutorial, we assume an active Webchat conversation exists. We will retrieve the conversation ID by querying recent conversations.

import httpx

def get_latest_webchat_conversation(token: str) -> str:
    """
    Retrieves the most recent active Webchat conversation ID.
    In production, you would pass the conversation ID directly from your application state.
    """
    url = f"https://api.{REGION}/api/v2/conversations/webchat"
    headers = {
        "Authorization": f"Bearer {token}",
        "Accept": "application/json"
    }
    
    # Query for active conversations
    params = {
        "active": "true",
        "pageSize": "1"
    }

    with httpx.Client() as client:
        response = client.get(url, headers=headers, params=params)
        
        if response.status_code != 200:
            raise RuntimeError(f"Failed to fetch conversations: {response.text}")
            
        data = response.json()
        if not data.get("entities"):
            raise RuntimeError("No active webchat conversations found.")
            
        return data["entities"][0]["id"]

# conversation_id = get_latest_webchat_conversation(TOKEN)
# For testing, replace the line above with a real conversation ID from your Genesys Cloud environment.
conversation_id = "REPLACE_WITH_REAL_CONVERSATION_ID"

Step 2: Initiate the Cobrowse Session

To initiate cobrowse, you must make a POST request to the /api/v2/cobrowse/sessions endpoint. However, simply creating the session is not enough. You must also add the cobrowse media to the conversation.

The Genesys Cloud Conversations API allows you to add media to an existing conversation. The critical step is constructing the correct payload for the POST /api/v2/conversations/{conversationId}/media endpoint.

Required OAuth Scope: cobrowse:initiate

Endpoint: POST /api/v2/cobrowse/sessions

First, we create the cobrowse session resource. This returns a cobrowseSessionId and a joinUrl (for the agent) and potentially a viewerUrl (depending on configuration).

def create_cobrowse_session(token: str) -> dict:
    """
    Creates a new cobrowse session resource in Genesys Cloud.
    Returns the session details including the session ID.
    """
    url = f"https://api.{REGION}/api/v2/cobrowse/sessions"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
        "Accept": "application/json"
    }
    
    # The body is often empty for simple initiation, but you can specify 
    # initial settings if supported by your org's cobrowse configuration.
    body = {} 

    with httpx.Client() as client:
        response = client.post(url, headers=headers, json=body)
        
        if response.status_code != 201:
            raise RuntimeError(f"Failed to create cobrowse session: {response.text}")
            
        return response.json()

cobrowse_data = create_cobrowse_session(TOKEN)
cobrowse_session_id = cobrowse_data["id"]
print(f"Cobrowse Session Created: {cobrowse_session_id}")

Step 3: Add Cobrowse Media to the Conversation

Now that we have a cobrowseSessionId, we must attach it to the active conversation. This is done via the Conversations API.

Endpoint: POST /api/v2/conversations/{conversationId}/media

Required OAuth Scope: conversation:webchat (or conversation:voice)

The payload must specify the type as cobrowse and include the cobrowseSessionId.

def add_cobrowse_to_conversation(token: str, conversation_id: str, cobrowse_session_id: str) -> dict:
    """
    Adds the cobrowse media to an existing conversation.
    This action invites the customer to join the cobrowse session.
    """
    url = f"https://api.{REGION}/api/v2/conversations/{conversation_id}/media"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
        "Accept": "application/json"
    }
    
    body = {
        "type": "cobrowse",
        "cobrowseSessionId": cobrowse_session_id
    }

    with httpx.Client() as client:
        response = client.post(url, headers=headers, json=body)
        
        if response.status_code not in [200, 201]:
            raise RuntimeError(f"Failed to add cobrowse to conversation: {response.text}")
            
        return response.json()

media_result = add_cobrowse_to_conversation(TOKEN, conversation_id, cobrowse_session_id)
print(f"Cobrowse added to conversation. Media ID: {media_result.get('id', 'N/A')}")

Step 4: Verify Session Status

After adding the media, you should verify that the cobrowse session is active and linked correctly. You can query the conversation details to see the media list.

Endpoint: GET /api/v2/conversations/{conversationId}

def verify_cobrowse_status(token: str, conversation_id: str) -> None:
    """
    Fetches conversation details to verify cobrowse media is present.
    """
    url = f"https://api.{REGION}/api/v2/conversations/{conversation_id}"
    headers = {
        "Authorization": f"Bearer {token}",
        "Accept": "application/json"
    }

    with httpx.Client() as client:
        response = client.get(url, headers=headers)
        
        if response.status_code != 200:
            raise RuntimeError(f"Failed to fetch conversation: {response.text}")
            
        conversation = response.json()
        media_list = conversation.get("media", [])
        
        cobrowse_media = None
        for media in media_list:
            if media.get("type") == "cobrowse":
                cobrowse_media = media
                break
                
        if cobrowse_media:
            print("Success: Cobrowse media is active in the conversation.")
            print(f"  Session ID: {cobrowse_media.get('cobrowseSessionId')}")
            print(f"  Status: {cobrowse_media.get('state')}")
        else:
            print("Warning: Cobrowse media not found in conversation.")

verify_cobrowse_status(TOKEN, conversation_id)

Complete Working Example

Below is the full, copy-pasteable script. Ensure you have set the environment variables GENESYS_CLIENT_ID, GENESYS_CLIENT_SECRET, and GENESYS_REGION before running.

import httpx
import os
import sys
from dotenv import load_dotenv

load_dotenv()

# --- Configuration ---
REGION = os.getenv("GENESYS_REGION", "mypurecloud.com")
CLIENT_ID = os.getenv("GENESYS_CLIENT_ID")
CLIENT_SECRET = os.getenv("GENESYS_CLIENT_SECRET")
CONVERSATION_ID = os.getenv("GENESYS_CONVERSATION_ID")  # Must be an active Webchat conversation

if not CLIENT_ID or not CLIENT_SECRET:
    print("Error: GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET must be set.")
    sys.exit(1)

if not CONVERSATION_ID:
    print("Error: GENESYS_CONVERSATION_ID must be set to an active Webchat conversation ID.")
    sys.exit(1)

# --- Helper Functions ---

def get_access_token() -> str:
    url = f"https://api.{REGION}/oauth/token"
    headers = {"Content-Type": "application/x-www-form-urlencoded"}
    data = {
        "grant_type": "client_credentials",
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET,
        "scope": "cobrowse:initiate conversation:webchat conversation:read"
    }
    
    with httpx.Client() as client:
        response = client.post(url, headers=headers, data=data)
        if response.status_code != 200:
            raise RuntimeError(f"Auth Failed: {response.text}")
        return response.json()["access_token"]

def create_cobrowse_session(token: str) -> str:
    """Creates a cobrowse session and returns the session ID."""
    url = f"https://api.{REGION}/api/v2/cobrowse/sessions"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
        "Accept": "application/json"
    }
    
    with httpx.Client() as client:
        response = client.post(url, headers=headers, json={})
        if response.status_code != 201:
            raise RuntimeError(f"Create Session Failed: {response.text}")
        return response.json()["id"]

def add_cobrowse_to_conversation(token: str, conv_id: str, session_id: str) -> None:
    """Attaches the cobrowse session to the conversation."""
    url = f"https://api.{REGION}/api/v2/conversations/{conv_id}/media"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
        "Accept": "application/json"
    }
    body = {
        "type": "cobrowse",
        "cobrowseSessionId": session_id
    }
    
    with httpx.Client() as client:
        response = client.post(url, headers=headers, json=body)
        if response.status_code not in [200, 201]:
            raise RuntimeError(f"Add Media Failed: {response.text}")
        print(f"Cobrowse added to conversation {conv_id}")

def verify_session(token: str, conv_id: str) -> None:
    """Verifies the cobrowse media is present."""
    url = f"https://api.{REGION}/api/v2/conversations/{conv_id}"
    headers = {
        "Authorization": f"Bearer {token}",
        "Accept": "application/json"
    }
    
    with httpx.Client() as client:
        response = client.get(url, headers=headers)
        if response.status_code != 200:
            raise RuntimeError(f"Verify Failed: {response.text}")
        
        conv_data = response.json()
        for media in conv_data.get("media", []):
            if media.get("type") == "cobrowse":
                print(f"Verified: Cobrowse active. Session ID: {media.get('cobrowseSessionId')}")
                return
        
        print("Error: Cobrowse media not found in conversation.")

# --- Main Execution ---

def main():
    try:
        print("1. Obtaining Access Token...")
        token = get_access_token()
        
        print("2. Creating Cobrowse Session...")
        session_id = create_cobrowse_session(token)
        print(f"   Session ID: {session_id}")
        
        print("3. Adding Cobrowse to Conversation...")
        add_cobrowse_to_conversation(token, CONVERSATION_ID, session_id)
        
        print("4. Verifying Session...")
        verify_session(token, CONVERSATION_ID)
        
    except Exception as e:
        print(f"Fatal Error: {e}")
        sys.exit(1)

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 403 Forbidden

  • What causes it: The OAuth token used in the request does not have the required scopes.
  • How to fix it: Ensure your scope parameter in the token request includes cobrowse:initiate and the appropriate conversation scope (e.g., conversation:webchat).
  • Code Check:
    # Ensure scope is correct
    data = {
        "grant_type": "client_credentials",
        "client_id": CLIENT_ID,
        "client_secret": CLIENT_SECRET,
        "scope": "cobrowse:initiate conversation:webchat conversation:read" # <--- Critical
    }
    

Error: 404 Not Found on Conversation

  • What causes it: The conversationId provided does not exist, is not owned by the user associated with the OAuth token, or the conversation has ended.
  • How to fix it: Verify the conversationId is active. Use the GET /api/v2/conversations/{conversationId} endpoint to check if the conversation exists and is in an active state.
  • Debugging:
    # Check conversation status before adding media
    resp = client.get(f"https://api.{REGION}/api/v2/conversations/{conv_id}", headers=headers)
    if resp.json().get("state") != "active":
        raise ValueError("Conversation is not active.")
    

Error: 400 Bad Request on Media Addition

  • What causes it: The conversation type does not support cobrowse, or the cobrowse session ID is invalid/expired.
  • How to fix it:
    1. Confirm the conversation is a Webchat or Voice conversation that has cobrowse enabled in your Genesys Cloud organization settings.
    2. Ensure the cobrowseSessionId was created successfully in Step 2.
    3. Check that the type in the payload is exactly "cobrowse" (lowercase).

Error: 429 Too Many Requests

  • What causes it: You have exceeded the rate limit for the Cobrowse or Conversations API.
  • How to fix it: Implement exponential backoff. Genesys Cloud returns a Retry-After header in 429 responses.
  • Code Example with Retry:
    import time
    
    def post_with_retry(client, url, headers, json_body, max_retries=3):
        for attempt in range(max_retries):
            response = client.post(url, headers=headers, json=json_body)
            if response.status_code == 429:
                retry_after = int(response.headers.get("Retry-After", 2 ** attempt))
                print(f"Rate limited. Retrying in {retry_after} seconds...")
                time.sleep(retry_after)
                continue
            return response
        raise RuntimeError("Max retries exceeded")
    

Official References