How to Transfer a Call to Another Queue Programmatically Using the Conversations API

What You Will Build

• A Python script that programmatically transfers an active voice conversation to a specific target queue using the Genesys Cloud Conversations API.
• This implementation utilizes the PATCH method on the /api/v2/communications/conversations/{conversationId} endpoint to update the conversation’s routing configuration.
• The code is written in Python 3.9+ using the genesyscloud SDK and the httpx library for raw HTTP fallback examples.

Prerequisites

• OAuth Client Type: A Private Key authentication client (Service Account) or a Client Credentials client with sufficient permissions.
• Required Scopes:
  • conversation:view (to retrieve the current conversation state)
  • conversation:update (to modify the conversation routing)
  • routing:queue:view (to validate the target queue ID)
• SDK Version: genesyscloud Python SDK version 140.0.0 or higher.
• Runtime Requirements: Python 3.9+.
• External Dependencies:
  • genesyscloud
  • httpx (for raw API examples and robust error handling)

Authentication Setup

The Genesys Cloud API requires OAuth 2.0 authentication. For backend integrations, the Private Key flow is the standard. The SDK handles the token acquisition and refresh automatically when initialized correctly.

Below is the setup for the SDK client. Ensure you have your private_key loaded from a secure environment variable or file.

import os
from purecloudplatformclientv2 import ApiClient, Configuration

def get_pure_cloud_api_client():
    """
    Initializes and returns a configured PureCloudPlatformClientV2 instance.
    Uses Private Key authentication.
    """
    # Load private key from environment variable for security
    private_key = os.getenv("GENESYS_PRIVATE_KEY")
    client_id = os.getenv("GENESYS_CLIENT_ID")
    client_secret = os.getenv("GENESYS_CLIENT_SECRET")
    region = os.getenv("GENESYS_REGION", "us-east-1")  # e.g., us-east-1, eu-west-1

    if not private_key or not client_id:
        raise ValueError("GENESYS_PRIVATE_KEY and GENESYS_CLIENT_ID environment variables are required.")

    # Initialize the API client configuration
    config = Configuration()
    config.region = region
    
    # Set up private key authentication
    config.private_key = private_key
    config.client_id = client_id
    
    # Optional: Set client secret if using a specific type of private key auth
    # Note: PureCloudPlatformClientV2 primarily uses client_id + private_key
    
    api_client = ApiClient(configuration=config)
    return api_client

Implementation

Transferring a call to a queue via the API is not a dedicated “transfer” endpoint. Instead, it is a state update. You must send a PATCH request to the conversation resource, replacing the existing routing object with a new one that specifies the target queueId.

Step 1: Retrieve the Target Queue ID

Before updating the conversation, you must know the ID of the target queue. You cannot transfer a call to a queue name; you must use the UUID.

API Endpoint: GET /api/v2/routing/queues

Required Scope: routing:queue:view

from purecloudplatformclientv2 import RoutingApi
from purecloudplatformclientv2.rest import ApiException

def find_queue_id(api_client, queue_name):
    """
    Searches for a queue by name and returns its ID.
    Raises an exception if the queue is not found.
    """
    routing_api = RoutingApi(api_client)
    
    try:
        # Fetch queues with a basic filter
        # Note: For large organizations, use the query API instead of listing all
        response = routing_api.get_routing_queues(page_size=100)
        
        for queue in response.entities:
            if queue.name.lower() == queue_name.lower():
                return queue.id
        
        raise ValueError(f"Queue '{queue_name}' not found.")
        
    except ApiException as e:
        print(f"Exception when calling RoutingApi->get_routing_queues: {e}")
        raise

Step 2: Construct the Patch Payload

The core of the transfer logic lies in the JSON payload sent to the PATCH request. The payload must contain a routing object. Within routing, you specify the queueId.

Crucially, you must also define the priority and optionally the skillRequirements. If you do not specify these, the system may default to priority 50. It is best practice to explicitly set the priority to ensure the call enters the queue with the intended urgency.

API Endpoint: PATCH /api/v2/communications/conversations/{conversationId}

Required Scope: conversation:update

Payload Structure:

{
  "routing": {
    "queueId": "target-queue-uuid-here",
    "priority": 50,
    "skillRequirements": []
  }
}

Note on skillRequirements: If the target queue requires specific skills for routing and you do not provide them, the agent selection might fail or default. For a simple queue transfer, an empty array is often sufficient if the queue does not enforce skill-based routing strictly, but checking the queue configuration is recommended.

Step 3: Execute the Transfer

Now, we combine the retrieval and the update. We will use the CommunicationsApi from the SDK. The SDK method patch_communications_conversation handles the HTTP PATCH verb automatically.

from purecloudplatformclientv2 import CommunicationsApi
from purecloudplatformclientv2.models import ConversationPatchRequest
from purecloudplatformclientv2.models import RoutingData
from purecloudplatformclientv2.rest import ApiException

def transfer_call_to_queue(api_client, conversation_id, target_queue_id, priority=50):
    """
    Transfers an active conversation to a specified queue.
    
    Args:
        api_client: The initialized PureCloudPlatformClientV2 instance.
        conversation_id (str): The UUID of the active conversation.
        target_queue_id (str): The UUID of the destination queue.
        priority (int): The priority level for the call in the new queue (0-100, lower is higher priority).
        
    Returns:
        bool: True if successful, False otherwise.
    """
    communications_api = CommunicationsApi(api_client)
    
    # 1. Construct the Routing Data object
    # This is the core object that tells Genesys where to send the call
    routing_data = RoutingData(
        queue_id=target_queue_id,
        priority=priority,
        skill_requirements=[]  # Adjust if skill-based routing is required
    )
    
    # 2. Construct the Patch Request
    # The SDK requires a specific model for the PATCH body
    patch_request = ConversationPatchRequest(
        routing=routing_data
    )
    
    try:
        # 3. Execute the PATCH request
        # The SDK maps this to PATCH /api/v2/communications/conversations/{conversationId}
        response = communications_api.patch_communications_conversation(
            conversation_id=conversation_id,
            body=patch_request
        )
        
        print(f"Conversation {conversation_id} successfully transferred to queue {target_queue_id}.")
        return True
        
    except ApiException as e:
        # Handle specific HTTP errors
        if e.status == 401:
            print("Authentication failed. Check your Private Key and Client ID.")
        elif e.status == 403:
            print("Forbidden. Ensure the OAuth token has 'conversation:update' scope.")
        elif e.status == 404:
            print(f"Conversation {conversation_id} not found. It may have ended.")
        elif e.status == 409:
            print("Conflict. The conversation may not be in a valid state for transfer (e.g., already in a queue or ended).")
        elif e.status == 429:
            print("Rate limit exceeded. Implement exponential backoff.")
        else:
            print(f"Unexpected API Error: {e.status} - {e.reason}")
        
        return False

Step 4: Handling Edge Cases and Validation

Before sending the PATCH request, you should verify the conversation is actually in a state that allows transfer. Transferring a call that is already in a queue, is completed, or is in a transfer state already will result in a 409 Conflict or 404 Not Found.

You can check the conversation state using GET /api/v2/communications/conversations/{conversationId}.

from purecloudplatformclientv2 import CommunicationsApi
from purecloudplatformclientv2.rest import ApiException

def can_transfer_conversation(api_client, conversation_id):
    """
    Checks if the conversation is in a 'queued' or 'connected' state that allows transfer.
    Returns True if transferable, False otherwise.
    """
    communications_api = CommunicationsApi(api_client)
    
    try:
        response = communications_api.get_communications_conversation(conversation_id)
        
        # Check the state of the conversation
        # Valid states for transfer usually include 'queued', 'connected', 'held'
        # Invalid states include 'completed', 'busy', 'no-answer', 'declined'
        
        state = response.state
        
        if state in ['completed', 'busy', 'no-answer', 'declined', 'ringing']:
            print(f"Conversation state '{state}' does not allow transfer.")
            return False
            
        return True
        
    except ApiException as e:
        if e.status == 404:
            print("Conversation not found.")
            return False
        raise

Complete Working Example

This script combines all steps into a single executable module. It authenticates, finds a queue by name, validates the conversation state, and performs the transfer.

import os
import sys
from purecloudplatformclientv2 import ApiClient, Configuration, CommunicationsApi, RoutingApi
from purecloudplatformclientv2.models import ConversationPatchRequest, RoutingData
from purecloudplatformclientv2.rest import ApiException

def setup_api_client():
    """Initializes the Genesys Cloud API client."""
    private_key = os.getenv("GENESYS_PRIVATE_KEY")
    client_id = os.getenv("GENESYS_CLIENT_ID")
    region = os.getenv("GENESYS_REGION", "us-east-1")

    if not private_key or not client_id:
        raise EnvironmentError("Missing GENESYS_PRIVATE_KEY or GENESYS_CLIENT_ID environment variables.")

    config = Configuration()
    config.region = region
    config.private_key = private_key
    config.client_id = client_id
    
    return ApiClient(configuration=config)

def get_queue_id_by_name(api_client, queue_name):
    """Retrieves the Queue ID given a Queue Name."""
    routing_api = RoutingApi(api_client)
    try:
        # Fetching first page of queues
        response = routing_api.get_routing_queues(page_size=100)
        for queue in response.entities:
            if queue.name.lower() == queue_name.lower():
                return queue.id
        return None
    except ApiException as e:
        print(f"Error fetching queues: {e}")
        return None

def transfer_call(api_client, conversation_id, target_queue_id, priority=50):
    """Executes the transfer via PATCH."""
    communications_api = CommunicationsApi(api_client)
    
    # Prepare the routing object
    routing_data = RoutingData(
        queue_id=target_queue_id,
        priority=priority,
        skill_requirements=[]
    )
    
    patch_body = ConversationPatchRequest(routing=routing_data)
    
    try:
        # Execute PATCH
        communications_api.patch_communications_conversation(
            conversation_id=conversation_id,
            body=patch_body
        )
        print(f"SUCCESS: Call {conversation_id} transferred to Queue {target_queue_id}.")
        return True
    except ApiException as e:
        print(f"FAILED: Transfer failed with status {e.status}. Reason: {e.reason}")
        print(f"Details: {e.body}")
        return False

def main():
    # 1. Configuration
    # In production, load these from a secure vault or .env file
    # os.environ['GENESYS_PRIVATE_KEY'] = "-----BEGIN PRIVATE KEY-----..."
    # os.environ['GENESYS_CLIENT_ID'] = "your-client-id"
    
    # Dummy values for demonstration structure
    conversation_id = os.getenv("CONVERSATION_ID", "dummy-conversation-uuid")
    target_queue_name = os.getenv("TARGET_QUEUE_NAME", "Support Tier 2")
    
    if conversation_id == "dummy-conversation-uuid":
        print("Please set CONVERSATION_ID environment variable.")
        sys.exit(1)

    try:
        # 2. Initialize Client
        api_client = setup_api_client()
        
        # 3. Find Target Queue
        print(f"Searching for queue: {target_queue_name}")
        queue_id = get_queue_id_by_name(api_client, target_queue_name)
        
        if not queue_id:
            print(f"Error: Queue '{target_queue_name}' not found.")
            sys.exit(1)
            
        print(f"Found Queue ID: {queue_id}")
        
        # 4. Transfer Call
        print(f"Initiating transfer for conversation: {conversation_id}")
        success = transfer_call(api_client, conversation_id, queue_id, priority=10)
        
        if not success:
            sys.exit(1)
            
    except Exception as e:
        print(f"Unexpected error: {e}")
        sys.exit(1)

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 409 Conflict

What causes it: The conversation is not in a state that allows modification. This typically happens if the call has already ended (completed), is currently ringing (ringing), or is already in a queue (queued) and you are trying to queue it again without proper parameters.
How to fix it:

1. Check the conversation state using GET /api/v2/communications/conversations/{conversationId}.
2. Ensure the state is connected or held.
3. If the call is already in a queue, ensure you are providing a different queueId or adjusting the priority significantly to trigger a re-queue.

Error: 403 Forbidden

What causes it: The OAuth token associated with the API client does not have the conversation:update scope.
How to fix it:

1. Go to the Genesys Cloud Admin Console.
2. Navigate to Administration > Security > Applications.
3. Select your application.
4. Ensure the Conversations permission set includes Update and View.
5. Regenerate the access token.

Error: 400 Bad Request - “Invalid Queue ID”

What causes it: The queueId provided in the routing object does not exist or is not accessible to the user/application.
How to fix it:

1. Verify the queueId is a valid UUID.
2. Ensure the application has routing:queue:view permissions.
3. Confirm the queue exists in the same region/organization as the conversation.

Error: 429 Too Many Requests

What causes it: You have exceeded the API rate limits for the Conversations API.
How to fix it:

1. Implement exponential backoff in your code.
2. Check the Retry-After header in the response.
3. Consider batching requests if you are transferring multiple calls, though call transfers are usually event-driven and low volume.

Official References

• Conversations API Documentation
• PATCH /communications/conversations/{conversationId} Reference
• Genesys Cloud Python SDK
• OAuth Scopes Reference