Programmatic Call Transfer via the Genesys Cloud Conversations API

What You Will Build

• This tutorial demonstrates how to programmatically transfer an active voice conversation from one queue to another using the Genesys Cloud Conversations API.
• It utilizes the PATCH /api/v2/conversations/voice/{conversationId} endpoint with the transfer action.
• The implementation is provided in Python using the official genesyscloud SDK and raw requests for clarity on payload structure.

Prerequisites

• OAuth Client Type: Confidential Client (Client Credentials Grant).
• Required Scopes:
  • conversation:transfer:write (To perform the transfer)
  • conversation:view (To read conversation details and status)
  • queue:member:view (Optional, if you need to validate queue membership beforehand)
• SDK Version: genesyscloud >= 1.0.0 (Python) or @genesyscloud/genesyscloud (Node.js).
• Runtime: Python 3.8+.
• Dependencies: pip install genesyscloud requests

Authentication Setup

Before interacting with the Conversations API, you must obtain a valid access token. The Genesys Cloud API uses OAuth 2.0 Client Credentials flow for server-to-server integrations.

The following Python function handles token acquisition and caching. In production, implement a TTL (Time-To-Live) check to refresh the token before it expires (typically 3600 seconds).

import requests
import time
from typing import Optional

class GenesysAuth:
    def __init__(self, client_id: str, client_secret: str, environment: str = "my.genesys.cloud"):
        self.client_id = client_id
        self.client_secret = client_secret
        self.environment = environment
        self.token_url = f"https://{environment}/oauth/token"
        self.access_token: Optional[str] = None
        self.token_expiry: float = 0.0

    def get_access_token(self) -> str:
        """
        Retrieves a new access token if the current one is expired or missing.
        """
        if self.access_token and time.time() < self.token_expiry:
            return self.access_token

        headers = {
            "Content-Type": "application/x-www-form-urlencoded"
        }
        data = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret
        }

        response = requests.post(self.token_url, headers=headers, data=data)
        response.raise_for_status()

        token_data = response.json()
        self.access_token = token_data["access_token"]
        # Set expiry slightly before actual expiry to avoid race conditions
        self.token_expiry = time.time() + token_data["expires_in"] - 60

        return self.access_token

# Initialize with your credentials
# auth = GenesysAuth("your_client_id", "your_client_secret")
# token = auth.get_access_token()

Implementation

Step 1: Identify the Conversation and Target Queue

To transfer a call, you need two distinct identifiers:

1. Conversation ID: The UUID of the active voice conversation.
2. Target Queue ID: The UUID of the queue you want to transfer the call to.

You cannot transfer a call that is not in the queued, waiting, or connected state. Additionally, the system user performing the transfer must have the conversation:transfer:write permission.

If you do not have the Queue ID, you can retrieve it by listing queues or searching by name. Here is how to find a queue by name using the Search API, which is often more reliable than iterating through all queues.

import requests

def find_queue_by_name(auth: GenesysAuth, queue_name: str) -> Optional[str]:
    """
    Searches for a queue by name and returns its ID.
    """
    token = auth.get_access_token()
    # Using the Search API to find queues by name
    url = f"https://{auth.environment}/api/v2/search"
    
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }

    # Query for entity type 'queue' and match name
    payload = {
        "query": {
            "bool": {
                "must": [
                    {
                        "term": {
                            "entityType": "queue"
                        }
                    },
                    {
                        "match": {
                            "name": queue_name
                        }
                    }
                ]
            }
        }
    }

    response = requests.post(url, json=payload, headers=headers)
    
    if response.status_code == 200:
        results = response.json().get("results", [])
        if results:
            # Return the ID of the first match
            return results[0].get("id")
    
    return None

# Example usage:
# queue_id = find_queue_by_name(auth, "Support Tier 2")
# if not queue_id:
#     raise ValueError("Queue not found")

Step 2: Construct the Transfer Payload

The core of the transfer operation is the PATCH request to /api/v2/conversations/voice/{conversationId}.

The body of this request must contain an array of actions. For a queue transfer, the action object requires:

• action: The string literal "transfer".
• toType: The string literal "queue".
• toId: The UUID of the target queue.

Optional but recommended fields:

• reasonCode: A reason code object if your organization requires transfer reasons.
• wrapUpCode: If you are transferring a connected agent, you might need to specify how the current leg should close.

Below is the structure of the JSON payload. Note that the toId must be a valid Queue ID.

{
  "actions": [
    {
      "action": "transfer",
      "toType": "queue",
      "toId": "00000000-0000-0000-0000-000000000000"
    }
  ]
}

Step 3: Execute the Transfer

The following Python function performs the actual transfer. It includes error handling for common status codes such as 404 (Conversation not found), 403 (Insufficient permissions), and 409 (Conversation state does not allow transfer).

import requests
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def transfer_call_to_queue(
    auth: GenesysAuth, 
    conversation_id: str, 
    target_queue_id: str,
    environment: str = "my.genesys.cloud"
) -> dict:
    """
    Transfers an active voice conversation to a specified queue.
    
    Args:
        auth: GenesysAuth instance with valid token.
        conversation_id: UUID of the active conversation.
        target_queue_id: UUID of the destination queue.
        environment: The Genesys Cloud environment URL.
        
    Returns:
        The JSON response from the API.
        
    Raises:
        requests.exceptions.HTTPError: If the API returns a non-2xx status.
    """
    token = auth.get_access_token()
    url = f"https://{environment}/api/v2/conversations/voice/{conversation_id}"
    
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
        "Accept": "application/json"
    }

    payload = {
        "actions": [
            {
                "action": "transfer",
                "toType": "queue",
                "toId": target_queue_id
            }
        ]
    }

    try:
        response = requests.patch(url, json=payload, headers=headers)
        
        # Check for success. Note: PATCH often returns 200 or 204.
        if response.status_code == 204:
            logger.info(f"Successfully transferred conversation {conversation_id} to queue {target_queue_id}.")
            return {}
        
        response.raise_for_status()
        return response.json()

    except requests.exceptions.HTTPError as http_err:
        logger.error(f"HTTP error occurred: {http_err}")
        logger.error(f"Response body: {response.text}")
        
        # Specific error handling
        if response.status_code == 404:
            raise ValueError(f"Conversation {conversation_id} not found or does not exist.")
        elif response.status_code == 403:
            raise PermissionError("Insufficient permissions. Ensure the client has 'conversation:transfer:write'.")
        elif response.status_code == 409:
            raise RuntimeError("Conflict: The conversation is in a state that does not allow transfer (e.g., closed, ringing).")
        else:
            raise http_err
    except requests.exceptions.RequestException as req_err:
        logger.error(f"Request failed: {req_err}")
        raise req_err

Complete Working Example

This script combines authentication, queue lookup, and the transfer action into a single runnable module. It assumes you have the conversation_id from an external source (e.g., a webhook or database).

#!/usr/bin/env python3
"""
Genesys Cloud Call Transfer Example
Transfers an active voice conversation to a specified queue.
"""

import sys
import requests
import time
import logging
from typing import Optional

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

class GenesysAuth:
    def __init__(self, client_id: str, client_secret: str, environment: str = "my.genesys.cloud"):
        self.client_id = client_id
        self.client_secret = client_secret
        self.environment = environment
        self.token_url = f"https://{environment}/oauth/token"
        self.access_token: Optional[str] = None
        self.token_expiry: float = 0.0

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

        headers = {"Content-Type": "application/x-www-form-urlencoded"}
        data = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret
        }

        try:
            response = requests.post(self.token_url, headers=headers, data=data)
            response.raise_for_status()
            token_data = response.json()
            self.access_token = token_data["access_token"]
            self.token_expiry = time.time() + token_data["expires_in"] - 60
            return self.access_token
        except requests.exceptions.RequestException as e:
            logger.error(f"Failed to obtain access token: {e}")
            raise

def find_queue_by_name(auth: GenesysAuth, queue_name: str) -> Optional[str]:
    token = auth.get_access_token()
    url = f"https://{auth.environment}/api/v2/search"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    payload = {
        "query": {
            "bool": {
                "must": [
                    {"term": {"entityType": "queue"}},
                    {"match": {"name": queue_name}}
                ]
            }
        }
    }

    try:
        response = requests.post(url, json=payload, headers=headers)
        if response.status_code == 200:
            results = response.json().get("results", [])
            if results:
                return results[0].get("id")
    except Exception as e:
        logger.error(f"Error finding queue: {e}")
    return None

def transfer_call_to_queue(auth: GenesysAuth, conversation_id: str, target_queue_id: str) -> bool:
    token = auth.get_access_token()
    url = f"https://{auth.environment}/api/v2/conversations/voice/{conversation_id}"
    
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
        "Accept": "application/json"
    }

    payload = {
        "actions": [
            {
                "action": "transfer",
                "toType": "queue",
                "toId": target_queue_id
            }
        ]
    }

    try:
        response = requests.patch(url, json=payload, headers=headers)
        
        if response.status_code == 204:
            logger.info("Transfer successful (204 No Content).")
            return True
        elif response.status_code == 200:
            logger.info(f"Transfer successful. Response: {response.json()}")
            return True
            
        response.raise_for_status()
        return False

    except requests.exceptions.HTTPError as e:
        logger.error(f"HTTP Error: {e}")
        logger.error(f"Response: {response.text}")
        return False
    except Exception as e:
        logger.error(f"Unexpected error: {e}")
        return False

def main():
    # Configuration
    CLIENT_ID = "your_client_id_here"
    CLIENT_SECRET = "your_client_secret_here"
    ENVIRONMENT = "my.genesys.cloud" # e.g., "usw2.my.genesys.cloud"
    
    CONVERSATION_ID = "a1b2c3d4-e5f6-7890-abcd-ef1234567890" # Replace with active conversation ID
    TARGET_QUEUE_NAME = "Support Tier 2" # Replace with actual queue name

    if CLIENT_ID == "your_client_id_here":
        logger.error("Please update CLIENT_ID and CLIENT_SECRET in the script.")
        sys.exit(1)

    # Initialize Auth
    auth = GenesysAuth(CLIENT_ID, CLIENT_SECRET, ENVIRONMENT)
    
    # Step 1: Get Token
    try:
        auth.get_access_token()
    except Exception as e:
        logger.error("Authentication failed.")
        sys.exit(1)

    # Step 2: Find Queue ID
    queue_id = find_queue_by_name(auth, TARGET_QUEUE_NAME)
    if not queue_id:
        logger.error(f"Queue '{TARGET_QUEUE_NAME}' not found.")
        sys.exit(1)
    
    logger.info(f"Found Queue ID: {queue_id}")

    # Step 3: Transfer Call
    success = transfer_call_to_queue(auth, CONVERSATION_ID, queue_id)
    
    if success:
        logger.info("Process completed successfully.")
    else:
        logger.error("Transfer failed.")
        sys.exit(1)

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 409 Conflict

• Cause: The conversation is in a state that does not permit transfer. Common invalid states include closed, ended, or ringing (if the call has not been answered yet). You can only transfer calls that are queued, waiting, or connected.
• Fix: Verify the current state of the conversation using GET /api/v2/conversations/voice/{conversationId}. If the state is connected, ensure the agent is available to be dropped. If the state is queued, the transfer will move the caller to the new queue immediately.

Error: 403 Forbidden

• Cause: The OAuth token used does not have the conversation:transfer:write scope. Alternatively, the System User associated with the client credentials lacks the necessary permissions in the Genesys Cloud Admin console.
• Fix:
  1. Check the client credentials in Genesys Cloud Admin > Security > Client Credentials. Ensure conversation:transfer:write is checked.
  2. Verify the System User has the “Transfer conversations” permission.

Error: 404 Not Found

• Cause: The conversationId provided does not exist, or it belongs to a different organization (unlikely if using correct credentials). It may also happen if the conversation has aged out of the active cache (conversations are typically retained for 24 hours in the API).
• Fix: Use the Analytics API to look up historical conversations if the ID is older than 24 hours, or ensure you are capturing the ID during the active call lifecycle.

Error: 422 Unprocessable Entity

• Cause: The payload structure is incorrect. For example, using toType: "user" when you intended a queue, or providing an invalid UUID format for toId.
• Fix: Ensure the actions array contains exactly one object with action: "transfer", toType: "queue", and a valid toId.

Official References

• Conversations API Documentation
• Transfer Action Specification
• OAuth 2.0 Client Credentials Flow