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
conversationsAPI surface and thePureCloudPlatformClientV2Python SDK. - The implementation covers Python with the
requestslibrary 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:readconversation:writecobrowse:readcobrowse:writeuser: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_idandclient_secretare 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_IDcorresponds 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-Afterheaders in 429 responses. Respect these headers to avoid further throttling.