How to Mute/Unmute an Agent Microphone via the Genesys Cloud API

What You Will Build

• You will write code to programmatically mute and unmute an agent’s microphone during an active voice conversation.
• This tutorial utilizes the Genesys Cloud v2 Conversations API (/api/v2/conversations/voice/participants/{participantId}).
• The primary implementation language is Python, with supplementary examples in JavaScript/TypeScript.

Prerequisites

• OAuth Client Type: You require a Client Credentials grant flow client. Ensure the client has the necessary permissions to modify conversation participants.
• Required Scopes: The OAuth token must include the conversation:participant:write scope. Without this, the API will return a 403 Forbidden error.
• SDK Version: Genesys Cloud Python SDK (genesyscloud) version 1.0.0 or later. For JavaScript, use the @genesyscloud/platform-client-sdk package.
• Runtime Requirements: Python 3.8+ or Node.js 16+.
• External Dependencies:
  • Python: pip install genesyscloud requests
  • JavaScript: npm install @genesyscloud/platform-client-sdk axios

Authentication Setup

Before interacting with the Conversations API, you must obtain a valid access token. The Genesys Cloud API uses OAuth 2.0. For server-side integrations that control agent states or conversation parameters, the Client Credentials flow is standard.

Python Authentication Helper

Create a helper function to fetch the token. In a production environment, cache this token and refresh it before expiration. Do not fetch a new token for every API call.

import requests
import time
from typing import Optional

class GenesysAuth:
    def __init__(self, client_id: str, client_secret: str, region: str = "mypurecloud.com"):
        self.client_id = client_id
        self.client_secret = client_secret
        self.region = region
        self.token_url = f"https://api.{region}/oauth/token"
        self.access_token: Optional[str] = None
        self.token_expiry: float = 0

    def get_access_token(self) -> str:
        # Return cached token if valid
        if self.access_token and time.time() < self.token_expiry:
            return self.access_token

        # Fetch new token
        response = requests.post(
            self.token_url,
            data={
                "grant_type": "client_credentials",
                "client_id": self.client_id,
                "client_secret": self.client_secret
            }
        )
        
        if response.status_code != 200:
            raise Exception(f"Failed to obtain token: {response.status_code} - {response.text}")

        data = response.json()
        self.access_token = data["access_token"]
        # Cache for 90% of the lifetime to avoid edge-case expiration
        self.token_expiry = time.time() + (data["expires_in"] * 0.9)
        
        return self.access_token

Implementation

Step 1: Identify the Active Conversation and Participant

You cannot mute a microphone without knowing the specific conversationId and participantId. The agent must be in an active voice conversation.

1. Find the Conversation: Query the user’s current interactions.
2. Find the Participant: Locate the participant object corresponding to the agent.

Finding the Active Voice Conversation (Python)

Use the GET /api/v2/conversations/voice endpoint or the user-specific interaction endpoint. The most reliable method for a specific agent is querying their current interactions.

import requests
from typing import Dict, Any

class GenesysConversationManager:
    def __init__(self, auth: GenesysAuth):
        self.auth = auth
        self.base_url = f"https://api.{auth.region}"

    def get_active_voice_conversation(self, user_id: str) -> Dict[str, Any]:
        """
        Retrieves the first active voice conversation for a given user.
        """
        token = self.auth.get_access_token()
        headers = {
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json"
        }

        # Fetch all current conversations for the user
        # Note: In high-volume scenarios, paginate if necessary. 
        # For a single agent check, the default limit is usually sufficient.
        url = f"{self.base_url}/api/v2/conversations/voice"
        params = {
            "pageSize": 100,
            "filter": f"participants.userId eq '{user_id}'"
        }

        response = requests.get(url, headers=headers, params=params)
        
        if response.status_code != 200:
            raise Exception(f"Failed to fetch conversations: {response.status_code} - {response.text}")

        conversations = response.json().get("entities", [])
        
        # Filter for active conversations (state: 'connected' or 'ringing')
        active_convo = None
        for convo in conversations:
            if convo.get("state") in ["connected", "ringing", "holding"]:
                active_convo = convo
                break
        
        if not active_convo:
            raise Exception(f"No active voice conversation found for user {user_id}")
            
        return active_convo

Step 2: Locate the Participant ID

Once you have the conversation object, you must iterate through its participants array to find the one matching the agent’s userId.

    def get_agent_participant_id(self, conversation: Dict[str, Any], user_id: str) -> str:
        """
        Finds the participantId for the specific user within the conversation.
        """
        participants = conversation.get("participants", [])
        
        for participant in participants:
            if participant.get("userId") == user_id:
                return participant["id"]
        
        raise Exception(f"User {user_id} is not a participant in conversation {conversation['id']}")

Step 3: Execute the Mute/Unmute Command

The core action uses the PATCH /api/v2/conversations/voice/participants/{participantId} endpoint.

Critical Parameter: The muted field.

• true: Mutes the agent’s microphone (no audio sent from agent to others).
• false: Unmutes the agent’s microphone.

Required Headers:

• Content-Type: application/json
• Authorization: Bearer <token>

Request Body:

{
  "muted": true
}

Python Implementation: Mute/Unmute

    def toggle_mute(self, conversation_id: str, participant_id: str, muted: bool) -> Dict[str, Any]:
        """
        Mutes or unmutes a participant in a voice conversation.
        
        Args:
            conversation_id: The ID of the voice conversation.
            participant_id: The ID of the participant to mute.
            muted: True to mute, False to unmute.
        """
        token = self.auth.get_access_token()
        headers = {
            "Authorization": f"Bearer {token}",
            "Content-Type": "application/json"
        }

        # Construct the URL for the specific participant in the specific conversation
        # Note: The API path is /conversations/voice/participants/{participantId}
        # The conversation context is implicitly handled by the participant belonging to that conversation.
        url = f"{self.base_url}/api/v2/conversations/voice/participants/{participant_id}"
        
        payload = {
            "muted": muted
        }

        # Use PATCH method as per API specification
        response = requests.patch(url, headers=headers, json=payload)

        if response.status_code == 200:
            return response.json()
        elif response.status_code == 404:
            raise Exception(f"Participant {participant_id} not found in conversation.")
        elif response.status_code == 403:
            raise Exception("Insufficient permissions. Ensure 'conversation:participant:write' scope is present.")
        else:
            raise Exception(f"Failed to toggle mute: {response.status_code} - {response.text}")

JavaScript/TypeScript Implementation

For frontend or Node.js environments, use the official SDK or Axios. The logic remains identical.

import { PlatformClient } from '@genesyscloud/platform-client-sdk';

interface GenesysConfig {
  clientId: string;
  clientSecret: string;
  region: string;
}

export class GenesysCallController {
  private platformClient: PlatformClient;

  constructor(config: GenesysConfig) {
    this.platformClient = new PlatformClient(config.region);
    // Initialize OAuth (assuming client credentials)
    this.platformClient.login(config.clientId, config.clientSecret);
  }

  async toggleMute(conversationId: string, participantId: string, muted: boolean): Promise<void> {
    try {
      // The SDK method for updating a participant
      // Note: In the JS SDK, the method is typically part of the ConversationsVoiceApi
      const conversationsVoiceApi = this.platformClient.ConversationsVoiceApi;
      
      const body = {
        muted: muted
      };

      // Execute the update
      await conversationsVoiceApi.postConversationsVoiceParticipantsUpdate({
        conversationId: conversationId,
        participantId: participantId,
        body: body
      });

      console.log(`Successfully set muted=${muted} for participant ${participantId}`);
    } catch (error) {
      if (error instanceof Error) {
        console.error("Error toggling mute:", error.message);
      }
      throw error;
    }
  }

  async findActiveConversationForUser(userId: string): Promise<string | null> {
    const conversationsVoiceApi = this.platformClient.ConversationsVoiceApi;
    
    // Fetch conversations
    const response = await conversationsVoiceApi.getConversationsVoice({
      pageSize: 100,
      filter: `participants.userId eq '${userId}'`
    });

    const entities = response.entities || [];
    const activeConvo = entities.find(c => ['connected', 'ringing', 'holding'].includes(c.state));
    
    return activeConvo ? activeConvo.id : null;
  }
}

Complete Working Example

Below is a complete, runnable Python script. It assumes you have stored your credentials in environment variables.

import os
import sys
import time
import requests
from typing import Optional, Dict, Any

# --- Configuration ---
CLIENT_ID = os.getenv("GENESYS_CLIENT_ID")
CLIENT_SECRET = os.getenv("GENESYS_CLIENT_SECRET")
REGION = os.getenv("GENESYS_REGION", "mypurecloud.com")
AGENT_USER_ID = os.getenv("AGENT_USER_ID") # The ID of the agent to control

if not all([CLIENT_ID, CLIENT_SECRET, AGENT_USER_ID]):
    print("Error: Missing environment variables. Set GENESYS_CLIENT_ID, GENESYS_CLIENT_SECRET, and AGENT_USER_ID.")
    sys.exit(1)

# --- Authentication Class ---
class GenesysAuth:
    def __init__(self, client_id: str, client_secret: str, region: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.region = region
        self.token_url = f"https://api.{region}/oauth/token"
        self.access_token: Optional[str] = None
        self.token_expiry: float = 0

    def get_access_token(self) -> str:
        if self.access_token and time.time() < self.token_expiry:
            return self.access_token

        response = requests.post(
            self.token_url,
            data={
                "grant_type": "client_credentials",
                "client_id": self.client_id,
                "client_secret": self.client_secret
            }
        )
        
        if response.status_code != 200:
            raise Exception(f"Auth failed: {response.status_code} - {response.text}")

        data = response.json()
        self.access_token = data["access_token"]
        self.token_expiry = time.time() + (data["expires_in"] * 0.9)
        return self.access_token

# --- Conversation Manager Class ---
class GenesysConversationManager:
    def __init__(self, auth: GenesysAuth):
        self.auth = auth
        self.base_url = f"https://api.{auth.region}"

    def get_headers(self) -> Dict[str, str]:
        return {
            "Authorization": f"Bearer {self.auth.get_access_token()}",
            "Content-Type": "application/json"
        }

    def get_active_voice_conversation(self, user_id: str) -> Dict[str, Any]:
        url = f"{self.base_url}/api/v2/conversations/voice"
        params = {
            "pageSize": 100,
            "filter": f"participants.userId eq '{user_id}'"
        }

        response = requests.get(url, headers=self.get_headers(), params=params)
        
        if response.status_code != 200:
            raise Exception(f"Fetch conversations failed: {response.status_code} - {response.text}")

        conversations = response.json().get("entities", [])
        
        for convo in conversations:
            if convo.get("state") in ["connected", "ringing", "holding"]:
                return convo
        
        raise Exception(f"No active voice conversation found for user {user_id}")

    def get_agent_participant_id(self, conversation: Dict[str, Any], user_id: str) -> str:
        participants = conversation.get("participants", [])
        for participant in participants:
            if participant.get("userId") == user_id:
                return participant["id"]
        raise Exception(f"User {user_id} not found in conversation {conversation['id']}")

    def toggle_mute(self, participant_id: str, muted: bool) -> Dict[str, Any]:
        url = f"{self.base_url}/api/v2/conversations/voice/participants/{participant_id}"
        payload = {"muted": muted}

        response = requests.patch(url, headers=self.get_headers(), json=payload)

        if response.status_code == 200:
            return response.json()
        elif response.status_code == 403:
            raise Exception("403 Forbidden: Check OAuth scopes. Need 'conversation:participant:write'.")
        else:
            raise Exception(f"Toggle mute failed: {response.status_code} - {response.text}")

# --- Main Execution Logic ---
def main():
    auth = GenesysAuth(CLIENT_ID, CLIENT_SECRET, REGION)
    manager = GenesysConversationManager(auth)

    print(f"Checking active conversations for Agent ID: {AGENT_USER_ID}")

    try:
        # 1. Get Active Conversation
        conversation = manager.get_active_voice_conversation(AGENT_USER_ID)
        convo_id = conversation["id"]
        print(f"Found active conversation: {convo_id}")

        # 2. Get Participant ID
        participant_id = manager.get_agent_participant_id(conversation, AGENT_USER_ID)
        print(f"Participant ID: {participant_id}")

        # 3. Mute the Agent
        print("Muting agent microphone...")
        mute_result = manager.toggle_mute(participant_id, muted=True)
        print(f"Mute successful. State: {mute_result.get('muted')}")

        # Wait 5 seconds to demonstrate effect
        print("Waiting 5 seconds...")
        time.sleep(5)

        # 4. Unmute the Agent
        print("Unmuting agent microphone...")
        unmute_result = manager.toggle_mute(participant_id, muted=False)
        print(f"Unmute successful. State: {unmute_result.get('muted')}")

    except Exception as e:
        print(f"Error: {e}")
        sys.exit(1)

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 403 Forbidden

Cause: The OAuth token does not have the required scope.
Fix: Verify that your OAuth Client Credentials configuration in the Genesys Cloud Admin Portal includes the conversation:participant:write scope.
Debugging:

1. Go to Admin > Security > OAuth Clients.
2. Select your client.
3. Check the Scopes tab.
4. Ensure conversation:participant:write is checked.
5. Regenerate the token.

Error: 404 Not Found

Cause: The participantId or conversationId is invalid, or the conversation has ended.
Fix:

1. Ensure the conversation is still active. If the agent hangs up, the participant object may be removed or archived.
2. Verify that the userId passed to get_active_voice_conversation matches the AGENT_USER_ID exactly.
3. Check if the agent is actually in a voice conversation. If they are idle, no conversation entity will be returned.

Error: 409 Conflict

Cause: Attempting to update a participant in a conversation that has already terminated or is in a state that does not allow modification.
Fix:

1. Check the state of the conversation. Only connected, ringing, and holding states typically allow participant updates.
2. If the conversation is ended, abandoned, or transferred, the API may reject the PATCH request.

Error: 429 Too Many Requests

Cause: Rate limiting. Genesys Cloud APIs enforce rate limits per client ID.
Fix: Implement exponential backoff. Do not retry immediately.
Code Snippet for Retry:

import time
import random

def request_with_retry(func, *args, max_retries=3, **kwargs):
    for attempt in range(max_retries):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            if "429" in str(e) and attempt < max_retries - 1:
                wait_time = (2 ** attempt) + random.uniform(0, 1)
                print(f"Rate limited. Retrying in {wait_time:.2f} seconds...")
                time.sleep(wait_time)
            else:
                raise e

Official References

• Genesys Cloud API: Update Voice Participant
• Genesys Cloud OAuth Scopes Reference
• Genesys Cloud Python SDK Documentation
• Genesys Cloud JavaScript SDK Documentation