How to Initiate a Cobrowse Session Programmatically via the Genesys Cloud Conversations API

What You Will Build

• This tutorial demonstrates how to programmatically create and initiate a cobrowse session between a supervisor and an agent using the Genesys Cloud Conversations API.
• The solution utilizes the Genesys Cloud conversations and cobrowse API endpoints to establish a secure, real-time sharing context.
• The implementation is provided in Python using the genesyscloud SDK and httpx for raw API calls where SDK coverage is limited.

Prerequisites

Before executing the code, ensure you have the following environment configured:

• OAuth Client Credentials: You require a Genesys Cloud OAuth client with the following scopes:
  • cobrowse:session:write (to create sessions)
  • cobrowse:session:read (to check session status)
  • conversations:write (to associate the session with an existing conversation, if applicable)
  • user:read (to verify user identities)
• SDK Version: genesyscloud Python SDK version 150.0.0 or higher.
• Runtime: Python 3.9+.
• External Dependencies:

  pip install genesyscloud httpx
  

• Active Conversation ID: Cobrowse sessions are typically tied to an existing voice or digital conversation. You must have a valid conversationId from an active interaction.

Authentication Setup

Genesys Cloud uses OAuth 2.0 for authentication. For server-to-server integrations, the Client Credentials flow is standard. The following code sets up the authentication client and handles token acquisition.

import httpx
import os
import time
from typing import Optional, Dict

class GenesysAuth:
    def __init__(self, org_id: str, client_id: str, client_secret: str):
        self.org_id = org_id
        self.client_id = client_id
        self.client_secret = client_secret
        self.base_url = f"https://{org_id}.mypurecloud.com"
        self.token_endpoint = f"{self.base_url}/oauth/token"
        self._access_token: Optional[str] = None
        self._token_expires_at: float = 0

    def get_access_token(self) -> str:
        """
        Retrieves an OAuth access token.
        Caches the token until it expires to avoid unnecessary API calls.
        """
        current_time = time.time()
        if self._access_token and current_time < self._token_expires_at - 30:
            return self._access_token

        payload = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret
        }

        try:
            response = httpx.post(
                self.token_endpoint,
                data=payload,
                timeout=10.0
            )
            response.raise_for_status()
            token_data = response.json()
            
            self._access_token = token_data["access_token"]
            self._token_expires_at = current_time + token_data["expires_in"]
            
            return self._access_token
        except httpx.HTTPStatusError as e:
            if e.response.status_code == 401:
                raise RuntimeError("Invalid Client ID or Secret.")
            raise RuntimeError(f"Authentication failed: {e.response.text}")

# Initialize Auth
AUTH = GenesysAuth(
    org_id=os.environ["GENESYS_ORG_ID"],
    client_id=os.environ["GENESYS_CLIENT_ID"],
    client_secret=os.environ["GENESYS_CLIENT_SECRET"]
)

Implementation

Step 1: Locate an Active Conversation and Participants

Cobrowse cannot exist in a vacuum. It must be attached to a conversation. First, we retrieve an active conversation and identify the userId of the agent and the userId of the supervisor (the user initiating the cobrowse).

import httpx
import json
from typing import List, Dict, Any

def get_active_conversations(auth: GenesysAuth, limit: int = 5) -> List[Dict[str, Any]]:
    """
    Retrieves a list of active conversations to find a target session.
    Endpoint: GET /api/v2/conversations
    Scope: conversations:read
    """
    url = f"{auth.base_url}/api/v2/conversations"
    headers = {
        "Authorization": f"Bearer {auth.get_access_token()}",
        "Content-Type": "application/json"
    }
    
    params = {
        "state": "active",
        "limit": limit
    }

    response = httpx.get(url, headers=headers, params=params, timeout=10.0)
    response.raise_for_status()
    
    data = response.json()
    return data.get("entities", [])

def extract_participants(conversation: Dict[str, Any]) -> Dict[str, str]:
    """
    Extracts agent and supervisor IDs from a conversation entity.
    This is a simplified logic. In production, you would filter by role or specific user IDs.
    """
    participants = conversation.get("participants", [])
    agent_id = None
    supervisor_id = None # In a real scenario, this might be the current user's ID
    
    for p in participants:
        if p.get("role") == "agent":
            agent_id = p.get("userId")
            break
            
    # For this tutorial, we assume the supervisor is a specific user ID known to the system
    # or we use the OAuth client's associated user if it's a user-context flow.
    # Here we will pass the supervisor ID explicitly in the main function.
    
    return {
        "agent_id": agent_id,
        "conversation_id": conversation.get("id")
    }

Step 2: Create the Cobrowse Session

The core action is creating the session. Genesys Cloud provides a dedicated endpoint for cobrowse sessions. You must specify the conversationId to link the cobrowse to the active interaction.

API Endpoint: POST /api/v2/cobrowse/sessions

Required Payload Fields:

• conversationId: The ID of the active conversation.
• participants: An array of user IDs involved in the cobrowse. Typically, this includes the agent and the supervisor.

def create_cobrowse_session(
    auth: GenesysAuth,
    conversation_id: str,
    agent_user_id: str,
    supervisor_user_id: str
) -> Dict[str, Any]:
    """
    Initiates a cobrowse session.
    Endpoint: POST /api/v2/cobrowse/sessions
    Scope: cobrowse:session:write
    
    Args:
        auth: GenesysAuth instance
        conversation_id: ID of the active conversation
        agent_user_id: User ID of the agent to observe
        supervisor_user_id: User ID of the supervisor initiating the session
        
    Returns:
        The created session object containing the session ID and join URLs.
    """
    url = f"{auth.base_url}/api/v2/cobrowse/sessions"
    headers = {
        "Authorization": f"Bearer {auth.get_access_token()}",
        "Content-Type": "application/json"
    }

    # The payload defines who is in the session and which conversation it attaches to
    payload = {
        "conversationId": conversation_id,
        "participants": [
            {
                "userId": agent_user_id,
                "role": "agent" # Explicitly define role if supported by your org config
            },
            {
                "userId": supervisor_user_id,
                "role": "supervisor"
            }
        ],
        "settings": {
            "allowRemoteControl": False, # Set to True if you want the supervisor to control the agent's browser
            "timeoutInMinutes": 15
        }
    }

    try:
        response = httpx.post(url, headers=headers, json=payload, timeout=10.0)
        
        # Handle specific error codes
        if response.status_code == 409:
            # Conflict: A session might already exist for this conversation
            print("Warning: A cobrowse session may already exist for this conversation.")
            return response.json()
        elif response.status_code == 400:
            # Bad Request: Check payload structure or user permissions
            raise RuntimeError(f"Bad Request: {response.json()}")
        elif response.status_code == 403:
            # Forbidden: User lacks permission to cobrowse
            raise RuntimeError("Forbidden: The supervisor user lacks cobrowse permissions.")
            
        response.raise_for_status()
        return response.json()

    except httpx.HTTPStatusError as e:
        print(f"HTTP Error during session creation: {e.response.status_code}")
        print(f"Response Body: {e.response.text}")
        raise

Step 3: Retrieve Session Join Tokens

Once the session is created, the users need tokens to join the actual browser-based cobrowse interface. The API returns a sessionId. You must then fetch the specific join tokens for each participant.

API Endpoint: POST /api/v2/cobrowse/sessions/{sessionId}/participants/{participantId}/tokens

def get_cobrowse_join_token(
    auth: GenesysAuth,
    session_id: str,
    participant_user_id: str
) -> Dict[str, Any]:
    """
    Generates a join token for a specific participant in a cobrowse session.
    Endpoint: POST /api/v2/cobrowse/sessions/{sessionId}/participants/{participantId}/tokens
    Scope: cobrowse:session:read
    
    Args:
        auth: GenesysAuth instance
        session_id: The ID returned from create_cobrowse_session
        participant_user_id: The user ID of the person joining (agent or supervisor)
        
    Returns:
        Token object containing the 'token' and 'expiresAt' fields.
    """
    url = f"{auth.base_url}/api/v2/cobrowse/sessions/{session_id}/participants/{participant_user_id}/tokens"
    headers = {
        "Authorization": f"Bearer {auth.get_access_token()}",
        "Content-Type": "application/json"
    }

    try:
        response = httpx.post(url, headers=headers, timeout=10.0)
        response.raise_for_status()
        return response.json()
    except httpx.HTTPStatusError as e:
        print(f"Failed to generate join token: {e.response.status_code}")
        raise

Complete Working Example

The following script combines all steps into a single executable module. It finds an active conversation, starts a cobrowse session, and prints the join URLs for both the agent and the supervisor.

Note: You must replace SUPERVISOR_USER_ID with a valid User ID from your Genesys Cloud instance who has cobrowse permissions enabled.

import os
import sys
import httpx
import time
import json
from typing import Optional, Dict, Any, List

# --- Authentication Class (from Step 1) ---
class GenesysAuth:
    def __init__(self, org_id: str, client_id: str, client_secret: str):
        self.org_id = org_id
        self.client_id = client_id
        self.client_secret = client_secret
        self.base_url = f"https://{org_id}.mypurecloud.com"
        self.token_endpoint = f"{self.base_url}/oauth/token"
        self._access_token: Optional[str] = None
        self._token_expires_at: float = 0

    def get_access_token(self) -> str:
        current_time = time.time()
        if self._access_token and current_time < self._token_expires_at - 30:
            return self._access_token

        payload = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret
        }

        try:
            response = httpx.post(self.token_endpoint, data=payload, timeout=10.0)
            response.raise_for_status()
            token_data = response.json()
            self._access_token = token_data["access_token"]
            self._token_expires_at = current_time + token_data["expires_in"]
            return self._access_token
        except httpx.HTTPStatusError as e:
            raise RuntimeError(f"Auth Error: {e.response.text}")

# --- Helper Functions ---

def get_active_conversation(auth: GenesysAuth) -> Optional[Dict[str, Any]]:
    """Fetches the first active conversation."""
    url = f"{auth.base_url}/api/v2/conversations"
    headers = {"Authorization": f"Bearer {auth.get_access_token()}", "Content-Type": "application/json"}
    params = {"state": "active", "limit": 1}
    
    try:
        resp = httpx.get(url, headers=headers, params=params, timeout=10.0)
        resp.raise_for_status()
        entities = resp.json().get("entities", [])
        if not entities:
            return None
        return entities[0]
    except Exception as e:
        print(f"Error fetching conversations: {e}")
        return None

def create_session(auth: GenesysAuth, conversation_id: str, agent_id: str, supervisor_id: str) -> Dict[str, Any]:
    """Creates the cobrowse session."""
    url = f"{auth.base_url}/api/v2/cobrowse/sessions"
    headers = {"Authorization": f"Bearer {auth.get_access_token()}", "Content-Type": "application/json"}
    
    payload = {
        "conversationId": conversation_id,
        "participants": [
            {"userId": agent_id, "role": "agent"},
            {"userId": supervisor_id, "role": "supervisor"}
        ],
        "settings": {
            "allowRemoteControl": False,
            "timeoutInMinutes": 10
        }
    }

    try:
        resp = httpx.post(url, headers=headers, json=payload, timeout=10.0)
        resp.raise_for_status()
        return resp.json()
    except httpx.HTTPStatusError as e:
        if e.response.status_code == 409:
            print("Session already exists for this conversation.")
            # In a real app, you would GET the existing session here
            return None
        raise

def get_join_token(auth: GenesysAuth, session_id: str, user_id: str) -> str:
    """Gets the join token for a user."""
    url = f"{auth.base_url}/api/v2/cobrowse/sessions/{session_id}/participants/{user_id}/tokens"
    headers = {"Authorization": f"Bearer {auth.get_access_token()}", "Content-Type": "application/json"}
    
    resp = httpx.post(url, headers=headers, timeout=10.0)
    resp.raise_for_status()
    return resp.json().get("token")

# --- Main Execution Flow ---

def main():
    # 1. Configuration
    ORG_ID = os.environ.get("GENESYS_ORG_ID")
    CLIENT_ID = os.environ.get("GENESYS_CLIENT_ID")
    CLIENT_SECRET = os.environ.get("GENESYS_CLIENT_SECRET")
    SUPERVISOR_USER_ID = os.environ.get("GENESYS_SUPERVISOR_USER_ID") # Must be set

    if not all([ORG_ID, CLIENT_ID, CLIENT_SECRET, SUPERVISOR_USER_ID]):
        print("Error: Missing environment variables. Please set GENESYS_ORG_ID, CLIENT_ID, CLIENT_SECRET, and GENESYS_SUPERVISOR_USER_ID.")
        sys.exit(1)

    auth = GenesysAuth(ORG_ID, CLIENT_ID, CLIENT_SECRET)

    # 2. Find an Active Conversation
    print("Searching for active conversations...")
    conversation = get_active_conversation(auth)
    
    if not conversation:
        print("No active conversations found. Please initiate a call or chat first.")
        sys.exit(1)

    conv_id = conversation["id"]
    print(f"Found active conversation: {conv_id}")

    # Extract Agent ID from participants
    agent_id = None
    for p in conversation.get("participants", []):
        if p.get("role") == "agent":
            agent_id = p.get("userId")
            break
    
    if not agent_id:
        print("No agent found in this conversation. Cobrowse requires an agent participant.")
        sys.exit(1)

    print(f"Target Agent ID: {agent_id}")
    print(f"Supervisor ID: {SUPERVISOR_USER_ID}")

    # 3. Create Cobrowse Session
    print("Initiating cobrowse session...")
    session = create_session(auth, conv_id, agent_id, SUPERVISOR_USER_ID)
    
    if not session:
        sys.exit(1)

    session_id = session["id"]
    print(f"Cobrowse Session Created: {session_id}")

    # 4. Generate Join Tokens
    print("Generating join tokens...")
    
    try:
        agent_token = get_join_token(auth, session_id, agent_id)
        supervisor_token = get_join_token(auth, session_id, SUPERVISOR_USER_ID)

        # 5. Output Join URLs
        # The base URL for joining depends on your region. 
        # Standard format: https://<org_id>.mypurecloud.com/cobrowse/join?token=<token>
        
        join_base = f"https://{ORG_ID}.mypurecloud.com/cobrowse/join"
        
        agent_url = f"{join_base}?token={agent_token}"
        supervisor_url = f"{join_base}?token={supervisor_token}"

        print("\n--- Cobrowse Session Ready ---")
        print(f"Agent Join URL:     {agent_url}")
        print(f"Supervisor Join URL: {supervisor_url}")
        print("--------------------------------------------------")

    except Exception as e:
        print(f"Error generating tokens: {e}")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 403 Forbidden

• What causes it: The OAuth client lacks the cobrowse:session:write scope, or the user ID provided for the supervisor does not have the “Cobrowse” capability enabled in their profile settings within Genesys Cloud.
• How to fix it:
  1. Check the OAuth Client scopes in the Genesys Cloud Admin console. Ensure cobrowse:session:write is selected.
  2. Navigate to Admin > Users > [Supervisor Name] > Capabilities. Ensure “Cobrowse” is checked under the relevant interaction type (Voice, Digital, etc.).

Error: 409 Conflict

• What causes it: A cobrowse session is already active for the specified conversationId. Genesys Cloud does not allow multiple concurrent cobrowse sessions on the same conversation.
• How to fix it:
  1. Implement a check before creation. Use GET /api/v2/cobrowse/sessions?conversationId={id} to see if a session exists.
  2. If a session exists, retrieve its ID and generate new join tokens for the participants instead of creating a new session.

Error: 404 Not Found

• What causes it: The conversationId is invalid, inactive, or does not belong to the organization. Alternatively, the userId provided does not exist.
• How to fix it:
  1. Verify the conversation is in active state using the Conversations API.
  2. Ensure the User IDs are valid by testing a simple GET /api/v2/users/{userId} call.

Error: Token Expiration

• What causes it: Cobrowse join tokens have a short lifespan (typically 15 minutes). If the user does not join within this window, the token becomes invalid.
• How to fix it:
  1. Regenerate the token using the POST /api/v2/cobrowse/sessions/{sessionId}/participants/{participantId}/tokens endpoint.
  2. Implement an automated retry mechanism in your frontend application to refresh the token silently before it expires.

Official References

• Create Cobrowse Session API Documentation
• Get Cobrowse Join Token API Documentation
• Genesys Cloud Cobrowse Overview