Programmatic Call Transfer to Queue via Genesys Cloud Conversations API

What You Will Build

• This tutorial demonstrates how to execute a programmatic transfer of an active voice conversation to a specific routing queue using the Genesys Cloud Conversations API.
• The solution utilizes the PATCH method on the /api/v2/conversations/voice/{conversationId} endpoint to modify the conversation participant’s routing state.
• The implementation is provided in Python using the purecloudplatformclientv2 SDK, with equivalent raw HTTP curl commands for validation.

Prerequisites

• OAuth Client Type: A Machine-to-Machine (M2M) OAuth application with the agent or admin role assigned.
• Required Scopes:
  • conversations:read (to validate conversation state)
  • conversations:write (to execute the transfer)
  • routing:queue:read (optional, for validating the target queue ID)
• SDK Version: purecloudplatformclientv2 >= 22.0.0 (Python).
• Runtime: Python 3.8+.
• External Dependencies:
  • purecloudplatformclientv2
  • requests (for raw HTTP examples if needed)

pip install purecloudplatformclientv2

Authentication Setup

Genesys Cloud uses OAuth 2.0 for authentication. For programmatic integrations, the Client Credentials flow is standard. You must obtain an access token before making any API calls.

Python SDK Authentication

The Genesys Cloud Python SDK handles token caching and refresh logic automatically when configured correctly. You must provide the Client ID, Client Secret, and the environment (e.g., mypurecloud.com).

from purecloudplatformclientv2 import Configuration
from purecloudplatformclientv2.rest import ApiException

# Initialize the configuration
configuration = Configuration()

# Set the environment (e.g., mypurecloud.com, euw2.pure.cloud, etc.)
configuration.host = "https://api.mypurecloud.com"

# Set OAuth credentials
configuration.client_id = "YOUR_CLIENT_ID"
configuration.client_secret = "YOUR_CLIENT_SECRET"

# Optional: Set the base path if using a specific region
# configuration.host = "https://api.euw2.pure.cloud"

print("Authentication configuration initialized.")

Critical Note: The Configuration object must be passed to the API client instance. Failure to do so results in 401 Unauthorized errors.

Implementation

Step 1: Identify the Active Conversation and Participant

To transfer a call, you must know two unique identifiers:

1. conversationId: The UUID of the voice conversation.
2. participantId: The UUID of the participant you are transferring (usually the customer).

You cannot transfer a conversation without specifying which participant is being moved. If you are transferring the customer, you use the customer’s participant ID. If you are transferring yourself (the agent) to a queue, you use your own participant ID.

Scenario: An agent is on a call with a customer. The agent determines the customer needs specialized billing support. The agent triggers a transfer to the “Billing Support” queue.

First, we retrieve the conversation details to identify the participantId of the customer.

from purecloudplatformclientv2 import ConversationApi
from purecloudplatformclientv2.models import ConversationParticipant

def get_customer_participant_id(conversation_id: str, configuration: Configuration) -> str:
    """
    Retrieves the participant ID of the customer in a voice conversation.
    Assumes the first participant is the customer and the second is the agent.
    """
    api_instance = ConversationApi(configuration)
    
    try:
        # Fetch conversation details
        conversation = api_instance.get_conversation_voice(conversation_id=conversation_id)
        
        # Iterate through participants to find the customer
        # In a standard inbound call, the first participant is typically the customer
        if conversation.participants and len(conversation.participant_ids) > 0:
            # We need to check the participant details to determine role
            # However, for simplicity in this tutorial, we assume the first ID is the customer
            # In production, check participant.type == "customer"
            return conversation.participant_ids[0]
        else:
            raise ValueError("No participants found in conversation.")
            
    except ApiException as e:
        if e.status == 404:
            print(f"Conversation {conversation_id} not found.")
        elif e.status == 401:
            print("Authentication failed. Check Client ID/Secret.")
        else:
            print(f"API Error: {e.status} - {e.reason}")
        raise

# Example Usage
# customer_participant_id = get_customer_participant_id("123e4567-e89b-12d3-a456-426614174000", configuration)

Step 2: Construct the Transfer Payload

The core action is a PATCH request to the conversation endpoint. The body must contain a participants array with the specific participantId and a wrapupCode (optional but recommended for audit trails) and routingData or queueId.

To transfer to a queue, you must set the queueId in the participant’s routingData.

Required Payload Structure:

{
  "participants": [
    {
      "id": "CUSTOMER_PARTICIPANT_UUID",
      "queueId": "TARGET_QUEUE_UUID",
      "wrapupCode": "TRANSFERRED",
      "state": "queued"
    }
  ]
}

Key Parameters:

• id: The UUID of the participant being transferred.
• queueId: The UUID of the target routing queue.
• state: Must be set to queued to initiate the transfer process.
• wrapupCode: Optional. Setting this immediately closes the current interaction for the agent and starts the new queued interaction. If omitted, the agent remains connected until the transfer completes or fails.

Step 3: Execute the Transfer via SDK

Using the ConversationApi class, we call patch_conversation_voice.

from purecloudplatformclientv2 import ConversationApi
from purecloudplatformclientv2.models import ConversationPatch, ConversationParticipant
from purecloudplatformclientv2.rest import ApiException

def transfer_call_to_queue(
    configuration: Configuration,
    conversation_id: str,
    participant_id: str,
    queue_id: str,
    wrapup_code: str = "TRANSFERRED"
) -> bool:
    """
    Transfers a specific participant in a voice conversation to a target queue.
    
    Args:
        configuration: The authenticated Configuration object.
        conversation_id: UUID of the voice conversation.
        participant_id: UUID of the participant to transfer (usually the customer).
        queue_id: UUID of the target routing queue.
        wrapup_code: Optional wrap-up code to apply to the agent's leg.
    
    Returns:
        True if successful, False otherwise.
    """
    api_instance = ConversationApi(configuration)
    
    # Construct the participant update object
    participant = ConversationParticipant(
        id=participant_id,
        queue_id=queue_id,
        state="queued",
        wrapup_code=wrapup_code
    )
    
    # Construct the patch request body
    patch_body = ConversationPatch(
        participants=[participant]
    )
    
    try:
        # Execute the PATCH request
        # Note: The SDK method is patch_conversation_voice
        api_instance.patch_conversation_voice(
            conversation_id=conversation_id,
            body=patch_body
        )
        
        print(f"Successfully transferred participant {participant_id} to queue {queue_id}.")
        return True
        
    except ApiException as e:
        # Handle specific error codes
        if e.status == 400:
            print("Bad Request: Check participant_id and queue_id validity.")
        elif e.status == 404:
            print("Not Found: Conversation or Participant does not exist.")
        elif e.status == 409:
            print("Conflict: Conversation state prevents transfer (e.g., already queued).")
        elif e.status == 429:
            print("Rate Limited: Too many requests. Implement retry logic.")
        else:
            print(f"Unexpected API Error: {e.status} - {e.body}")
        
        return False

Step 4: Handling Edge Cases and Validation

Before executing the transfer, it is critical to validate that the conversation is in a transferable state. A conversation cannot be transferred if:

1. The participant is already in the queued state.
2. The conversation is ended or ended_by_system.
3. The target queue does not exist or is disabled.

Validation Logic:

def validate_transfer_state(conversation_id: str, configuration: Configuration) -> bool:
    """
    Checks if the conversation is in a state that allows transfer.
    """
    api_instance = ConversationApi(configuration)
    try:
        conversation = api_instance.get_conversation_voice(conversation_id=conversation_id)
        
        # Check conversation state
        if conversation.state in ["ended", "ended_by_system", "ended_by_user"]:
            print("Error: Conversation has already ended.")
            return False
            
        # Check if any participant is already queued
        if conversation.participants:
            for p in conversation.participants:
                if p.state == "queued":
                    print("Error: Participant is already in queue.")
                    return False
                    
        return True
        
    except ApiException as e:
        print(f"Validation Error: {e.status}")
        return False

Complete Working Example

The following script combines authentication, validation, and the transfer action into a single executable module.

import os
from purecloudplatformclientv2 import Configuration, ConversationApi
from purecloudplatformclientv2.models import ConversationPatch, ConversationParticipant
from purecloudplatformclientv2.rest import ApiException

def main():
    # 1. Configuration and Authentication
    configuration = Configuration()
    configuration.host = "https://api.mypurecloud.com"
    configuration.client_id = os.getenv("GENESYS_CLIENT_ID")
    configuration.client_secret = os.getenv("GENESYS_CLIENT_SECRET")
    
    if not configuration.client_id or not configuration.client_secret:
        raise EnvironmentError("GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET must be set.")

    # 2. Input Parameters
    # These should be passed from your application logic (e.g., CTI integration)
    CONVERSATION_ID = "123e4567-e89b-12d3-a456-426614174000" # Replace with actual ID
    CUSTOMER_PARTICIPANT_ID = "7c9e6679-7425-40de-944b-e07fc1f90ae7" # Replace with actual ID
    TARGET_QUEUE_ID = "b5a4c3d2-1f2e-3d4c-5b6a-7e8f9d0c1b2a" # Replace with actual Queue ID
    
    # 3. Validation
    print(f"Validating conversation {CONVERSATION_ID}...")
    if not validate_transfer_state(CONVERSATION_ID, configuration):
        print("Transfer aborted due to validation failure.")
        return

    # 4. Execution
    print(f"Initiating transfer to queue {TARGET_QUEUE_ID}...")
    success = transfer_call_to_queue(
        configuration=configuration,
        conversation_id=CONVERSATION_ID,
        participant_id=CUSTOMER_PARTICIPANT_ID,
        queue_id=TARGET_QUEUE_ID,
        wrapup_code="TRANSFERRED"
    )
    
    if success:
        print("Transfer initiated successfully. Monitor queue position via WebRTC or API polling.")
    else:
        print("Transfer failed. Check logs for details.")

def validate_transfer_state(conversation_id: str, configuration: Configuration) -> bool:
    api_instance = ConversationApi(configuration)
    try:
        conversation = api_instance.get_conversation_voice(conversation_id=conversation_id)
        if conversation.state in ["ended", "ended_by_system", "ended_by_user"]:
            print("Error: Conversation has already ended.")
            return False
        if conversation.participants:
            for p in conversation.participants:
                if p.state == "queued":
                    print("Error: Participant is already in queue.")
                    return False
        return True
    except ApiException as e:
        print(f"Validation Error: {e.status}")
        return False

def transfer_call_to_queue(
    configuration: Configuration,
    conversation_id: str,
    participant_id: str,
    queue_id: str,
    wrapup_code: str = "TRANSFERRED"
) -> bool:
    api_instance = ConversationApi(configuration)
    
    participant = ConversationParticipant(
        id=participant_id,
        queue_id=queue_id,
        state="queued",
        wrapup_code=wrapup_code
    )
    
    patch_body = ConversationPatch(
        participants=[participant]
    )
    
    try:
        api_instance.patch_conversation_voice(
            conversation_id=conversation_id,
            body=patch_body
        )
        return True
    except ApiException as e:
        if e.status == 429:
            print("Rate Limited. Implement exponential backoff.")
        else:
            print(f"API Error: {e.status} - {e.body}")
        return False

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 400 Bad Request

Cause: The most common cause is an invalid queueId or participantId. Another cause is attempting to set state="queued" on a participant who is already in that state.

Fix:

1. Verify the queueId exists using GET /api/v2/routing/queues/{queueId}.
2. Verify the participantId belongs to the specified conversationId.
3. Ensure the participant is not already queued.

# Debugging Code: Verify Queue Existence
from purecloudplatformclientv2 import RoutingApi
routing_api = RoutingApi(configuration)
try:
    queue = routing_api.get_routing_queue(queue_id=TARGET_QUEUE_ID)
    print(f"Queue found: {queue.name}")
except ApiException:
    print("Queue ID is invalid.")

Error: 404 Not Found

Cause: The conversationId does not exist, or the conversation has ended and been purged from active memory (though history remains, you cannot PATCH an ended conversation).

Fix: Check the conversation state using GET /api/v2/conversations/voice/{conversationId}. If the state is ended, the transfer is impossible.

Error: 409 Conflict

Cause: The conversation state is inconsistent with the requested action. For example, trying to transfer a participant who is currently connected but the system detects a race condition where they have already been updated by another process (such as the agent clicking transfer in the UI).

Fix: Implement optimistic locking. Retrieve the conversation, check the version field, and include it in the PATCH header if supported, or simply retry the operation after a short delay.

import time

def retry_transfer(max_retries=3, delay=1):
    for attempt in range(max_retries):
        success = transfer_call_to_queue(...)
        if success:
            return True
        if attempt < max_retries - 1:
            time.sleep(delay * (attempt + 1)) # Exponential backoff
    return False

Error: 429 Too Many Requests

Cause: You have exceeded the rate limit for the Conversations API. Genesys Cloud enforces rate limits per OAuth client and per endpoint.

Fix: Implement exponential backoff with jitter. The response headers will include Retry-After.

response = api_instance.patch_conversation_voice_with_http_info(...)
if response.status == 429:
    retry_after = int(response.headers.get('Retry-After', 5))
    time.sleep(retry_after)
    # Retry logic here

Official References

• Genesys Cloud Conversations API Documentation
• PATCH /api/v2/conversations/voice/{conversationId} Reference
• Python SDK Reference
• Routing Queues API