How to Disconnect a Participant from a Genesys Cloud Conference Call

How to Disconnect a Participant from a Genesys Cloud Conference Call

What You Will Build

  • A script that identifies a specific participant within an active Genesys Cloud conversation and sends a disconnect command to remove them from the session.
  • This tutorial uses the Genesys Cloud CX Conversations API (v2) to interact with real-time call data.
  • The implementation is demonstrated in Python using the official Genesys Cloud SDK.

Prerequisites

  • OAuth Client Type: A Genesys Cloud Private Web App or Server-to-Server OAuth client.
  • Required Scopes: conversation:read, conversation:write, and interaction:read (if retrieving metadata).
  • SDK Version: genesys-cloud-sdk version 130.0.0 or higher.
  • Language/Runtime: Python 3.9+.
  • External Dependencies: genesys-cloud-sdk, requests (for direct HTTP fallback examples), python-dotenv (for secure credential management).

Install the SDK via pip:

pip install genesys-cloud-sdk python-dotenv

Authentication Setup

Genesys Cloud uses OAuth 2.0. For server-side integrations, the Client Credentials Grant is the standard flow. You must configure your OAuth client in the Genesys Cloud Admin console with the necessary permissions.

The following code initializes the SDK client. This setup handles token acquisition and refreshing automatically if configured correctly.

import os
from dotenv import load_dotenv
from purecloudplatformclientv2.rest import ApiException
from purecloudplatformclientv2 import PlatformClient, ConversationApi

# Load environment variables from .env file
load_dotenv()

def get_platform_client() -> PlatformClient:
    """
    Initializes and returns a configured Genesys Cloud Platform Client.
    """
    # Retrieve credentials from environment variables
    client_id = os.getenv("GENESYS_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLIENT_SECRET")
    
    if not client_id or not client_secret:
        raise EnvironmentError("GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET must be set.")

    # Initialize the platform client
    client = PlatformClient()

    # Configure OAuth
    try:
        # The SDK handles token caching automatically in memory
        client.oauth_client.set_credentials(
            client_id=client_id,
            client_secret=client_secret
        )
        
        # Verify connectivity by fetching the user info (optional but recommended for debugging)
        # This ensures the token is valid before proceeding to conversation logic
        client.oauth_client.get_access_token()
        
    except Exception as e:
        print(f"Authentication failed: {e}")
        raise

    return client

Implementation

Step 1: Locate the Active Conversation

To disconnect a participant, you first need the conversationId. In a production environment, this ID is typically passed via a webhook or retrieved from an existing context. For this tutorial, we will query for active conversations involving a specific user or number.

If you already possess the conversationId, skip to Step 2.

from purecloudplatformclientv2 import ConversationApi, ConversationQueryBody

def find_active_conversation(client: PlatformClient, participant_identifier: str) -> str | None:
    """
    Finds an active conversation containing a specific participant identifier.
    This is a simplified search. In production, use a stored conversationId from a callback.
    
    Args:
        client: The authenticated PlatformClient.
        participant_identifier: Phone number or user ID involved in the call.
        
    Returns:
        The conversationId string, or None if not found.
    """
    api_instance = ConversationApi(client)
    
    # Define the query body
    # Note: Querying by participant details can be resource-intensive. 
    # Prefer storing the conversationId when the call is initiated.
    query_body = ConversationQueryBody(
        conversation_ids=[], # Leave empty to search all
        status="active",     # Only look at active calls
        # We cannot directly filter by participant in the basic query endpoint efficiently.
        # A better approach in production is to maintain a map of active calls.
        # For this example, we will fetch recent active conversations and filter locally.
    )
    
    try:
        # Fetch active conversations
        # Limit to 10 for this example to avoid rate limits
        response = api_instance.post_conversations_search(
            body=query_body,
        )
        
        if not response.entities:
            print("No active conversations found.")
            return None
            
        for conv in response.entities:
            # Check participants manually
            for participant in conv.participants:
                # Check if the identifier matches the address or external ID
                if participant_identifier in participant.address or participant_identifier in str(participant.external_id):
                    print(f"Found conversation ID: {conv.id}")
                    return conv.id
                    
    except ApiException as e:
        print(f"Exception when calling ConversationApi->post_conversations_search: {e}")
        return None

    return None

Expected Response Structure:
The post_conversations_search endpoint returns a ConversationSearchResponse. The critical field is entities, which is a list of Conversation objects. Each conversation contains a participants array.

Step 2: Identify the Target Participant

Once you have the conversationId, you must retrieve the full conversation details to identify the participantId of the person you wish to disconnect. You cannot disconnect by phone number alone; you must use the internal participantId.

from purecloudplatformclientv2 import ConversationApi

def get_participant_id_by_address(
    client: PlatformClient, 
    conversation_id: str, 
    target_address: str
) -> str | None:
    """
    Retrieves the participantId for a specific address within a conversation.
    
    Args:
        client: The authenticated PlatformClient.
        conversation_id: The ID of the active conversation.
        target_address: The phone number or user ID to disconnect.
        
    Returns:
        The participantId string, or None if not found.
    """
    api_instance = ConversationApi(client)
    
    try:
        # Fetch full conversation details
        conversation = api_instance.get_conversations_conversation(
            conversation_id=conversation_id
        )
        
        if not conversation.participants:
            print("Conversation has no participants.")
            return None
            
        for participant in conversation.participants:
            # Compare the target address with the participant's address
            # Ensure both are strings for comparison
            if participant.address and str(participant.address) == str(target_address):
                return participant.id
                
    except ApiException as e:
        print(f"Exception when getting conversation details: {e}")
        
    return None

Critical Note on Scopes:
This step requires the conversation:read scope. The get_conversations_conversation endpoint returns the full state of the call, including all participants, their statuses, and addresses.

Step 3: Disconnect the Participant

This is the core action. Genesys Cloud provides a specific endpoint to disconnect a single participant. This is distinct from ending the entire conversation.

Endpoint: DELETE /api/v2/conversations/{conversationId}/participants/{participantId}

Required Scope: conversation:write

from purecloudplatformclientv2 import ConversationApi

def disconnect_participant(
    client: PlatformClient, 
    conversation_id: str, 
    participant_id: str
) -> bool:
    """
    Disconnects a specific participant from a conversation.
    
    Args:
        client: The authenticated PlatformClient.
        conversation_id: The ID of the active conversation.
        participant_id: The ID of the participant to remove.
        
    Returns:
        True if successful, False otherwise.
    """
    api_instance = ConversationApi(client)
    
    try:
        # The SDK method for deleting a participant
        # This sends a DELETE request to /api/v2/conversations/{conversationId}/participants/{participantId}
        api_instance.delete_conversations_conversation_participant(
            conversation_id=conversation_id,
            participant_id=participant_id
        )
        
        print(f"Successfully disconnected participant {participant_id} from conversation {conversation_id}.")
        return True
        
    except ApiException as e:
        # Handle specific error codes
        if e.status == 404:
            print("Conversation or Participant not found. The call may have already ended.")
        elif e.status == 409:
            print("Conflict: The participant may already be disconnected or the conversation is in an invalid state.")
        elif e.status == 429:
            print("Rate limited. Wait before retrying.")
        else:
            print(f"Failed to disconnect participant: {e}")
            
        return False

Edge Case: The Last Participant
If you disconnect the last remaining participant in a conversation, Genesys Cloud automatically terminates the entire conversation. Ensure your logic accounts for this if you are managing multi-party conferences.

Complete Working Example

Below is the complete, runnable Python script. It assumes you have a .env file with GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET.

import os
import sys
from dotenv import load_dotenv
from purecloudplatformclientv2.rest import ApiException
from purecloudplatformclientv2 import PlatformClient, ConversationApi, ConversationQueryBody

# Load environment variables
load_dotenv()

def get_platform_client() -> PlatformClient:
    """Initializes and returns a configured Genesys Cloud Platform Client."""
    client_id = os.getenv("GENESYS_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLIENT_SECRET")
    
    if not client_id or not client_secret:
        raise EnvironmentError("GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET must be set.")

    client = PlatformClient()

    try:
        client.oauth_client.set_credentials(
            client_id=client_id,
            client_secret=client_secret
        )
        # Trigger token fetch
        client.oauth_client.get_access_token()
    except Exception as e:
        print(f"Authentication failed: {e}")
        raise

    return client

def find_active_conversation_by_number(client: PlatformClient, phone_number: str) -> str | None:
    """
    Simplified search for an active conversation by phone number.
    WARNING: This is inefficient for large systems. Use stored conversationIds in production.
    """
    api_instance = ConversationApi(client)
    query_body = ConversationQueryBody(
        conversation_ids=[],
        status="active"
    )
    
    try:
        response = api_instance.post_conversations_search(body=query_body)
        
        if not response.entities:
            return None
            
        for conv in response.entities:
            for participant in conv.participants:
                if participant.address and str(participant.address) == phone_number:
                    return conv.id
    except ApiException as e:
        print(f"Search failed: {e}")
    return None

def get_participant_id(client: PlatformClient, conversation_id: str, phone_number: str) -> str | None:
    """Finds the participantId associated with a phone number in a specific conversation."""
    api_instance = ConversationApi(client)
    try:
        conversation = api_instance.get_conversations_conversation(conversation_id=conversation_id)
        for participant in conversation.participants:
            if participant.address and str(participant.address) == phone_number:
                return participant.id
    except ApiException as e:
        print(f"Get participant details failed: {e}")
    return None

def disconnect_participant(client: PlatformClient, conversation_id: str, participant_id: str) -> bool:
    """Sends the disconnect command for a specific participant."""
    api_instance = ConversationApi(client)
    try:
        api_instance.delete_conversations_conversation_participant(
            conversation_id=conversation_id,
            participant_id=participant_id
        )
        return True
    except ApiException as e:
        print(f"Disconnect failed ({e.status}): {e.reason}")
        return False

def main():
    # Configuration: Replace with the actual phone number you wish to disconnect
    TARGET_PHONE_NUMBER = os.getenv("TARGET_PHONE_NUMBER", "+15551234567")
    
    if not TARGET_PHONE_NUMBER:
        print("Please set TARGET_PHONE_NUMBER in your .env file.")
        sys.exit(1)

    print(f"Starting process to disconnect {TARGET_PHONE_NUMBER}...")
    
    try:
        # 1. Authenticate
        client = get_platform_client()
        print("Authenticated successfully.")
        
        # 2. Find the conversation
        print("Searching for active conversation...")
        conversation_id = find_active_conversation_by_number(client, TARGET_PHONE_NUMBER)
        
        if not conversation_id:
            print(f"No active conversation found for {TARGET_PHONE_NUMBER}.")
            sys.exit(1)
            
        print(f"Found conversation ID: {conversation_id}")
        
        # 3. Identify the participant
        print("Identifying participant ID...")
        participant_id = get_participant_id(client, conversation_id, TARGET_PHONE_NUMBER)
        
        if not participant_id:
            print(f"Participant {TARGET_PHONE_NUMBER} not found in conversation {conversation_id}.")
            sys.exit(1)
            
        print(f"Participant ID: {participant_id}")
        
        # 4. Disconnect
        print("Initiating disconnect...")
        success = disconnect_participant(client, conversation_id, participant_id)
        
        if success:
            print("Operation completed successfully.")
        else:
            print("Operation failed.")
            
    except Exception as e:
        print(f"Critical error: {e}")
        sys.exit(1)

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 401 Unauthorized

Cause: The OAuth token is expired, invalid, or the client credentials are incorrect.
Fix: Ensure your GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET are correct. The SDK attempts to refresh tokens automatically, but if the refresh token is invalid, you must re-authenticate. Check that the OAuth client in Genesys Cloud is enabled and has not been revoked.

Error: 403 Forbidden

Cause: The OAuth client lacks the required scopes.
Fix: Verify that the client has conversation:write permission. The delete_conversations_conversation_participant endpoint strictly requires write access. Additionally, ensure the user associated with the OAuth client has the “Agent” or “Admin” role with permissions to manage conversations.

Error: 404 Not Found

Cause: The conversationId or participantId is invalid or the call has already ended.
Fix: Conversation IDs are ephemeral. If a call ends, the conversation moves to closed status. Ensure you are querying for active conversations. Double-check that the participantId matches exactly what was returned in the get_conversations_conversation response.

Error: 409 Conflict

Cause: The participant is already disconnected, or the conversation is in a state that does not allow disconnection (e.g., already terminated).
Fix: Check the conversation status before attempting the disconnect. If the status is closed, no action is needed. If the participant status is disconnected, the operation is redundant.

Error: 429 Too Many Requests

Cause: You have exceeded the Genesys Cloud API rate limits.
Fix: Implement exponential backoff. For high-volume disconnection scenarios, consider batching operations or using Webhooks to react to state changes rather than polling and issuing commands rapidly.

Official References