Programmatically Transfer a Call to Another Queue Using the Conversations API

What You Will Build

• You will write a script that identifies an active voice conversation and modifies its routing intent to transfer it to a specific Genesys Cloud queue.
• This tutorial uses the Genesys Cloud CX Conversations API (/api/v2/conversations) and the PureCloudPlatformClientV2 Python SDK.
• The implementation is provided in Python, utilizing the official Genesys Cloud SDK for type safety and automatic retry logic.

Prerequisites

Before executing the code, ensure the following requirements are met:

• OAuth Client: You must have a Public or Confidential OAuth client created in the Genesys Cloud Admin Portal.
• Required Scopes: The OAuth token must include the scope conversation:write. Without this scope, the PATCH request will return a 403 Forbidden error.
• SDK Version: This tutorial uses genesyscloud Python SDK version 13.0.0 or higher.
• Runtime: Python 3.8 or higher.
• Dependencies: Install the SDK via pip:

  pip install genesyscloud
  

Authentication Setup

Genesys Cloud APIs require a valid OAuth 2.0 Bearer token. For programmatic access, the Client Credentials Flow is the standard approach. This flow exchanges your Client ID and Client Secret for an access token.

The Genesys Cloud Python SDK handles token caching and refreshing automatically when initialized correctly. You must provide the region (e.g., us-east-1, eu-west-1) to ensure the token is requested from the correct identity provider.

from genesyscloud.platform.client import PureCloudPlatformClientV2
import os

def get_platform_client():
    """
    Initializes the Genesys Cloud Platform Client.
    Uses environment variables for credentials to avoid hardcoding secrets.
    """
    client_id = os.environ.get("GENESYS_CLIENT_ID")
    client_secret = os.environ.get("GENESYS_CLIENT_SECRET")
    region = os.environ.get("GENESYS_REGION", "us-east-1")

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

    # Initialize the client
    platform_client = PureCloudPlatformClientV2()

    # Configure the client with credentials and region
    # The SDK will automatically handle token acquisition and refresh
    platform_client.set_region(region)
    platform_client.set_client_id(client_id)
    platform_client.set_client_secret(client_secret)

    return platform_client

Implementation

Step 1: Retrieve the Active Conversation

To transfer a call, you must first identify the specific conversation ID. In a production environment, this ID usually arrives via a webhook (Event Streams) or is retrieved by querying active conversations for a specific user or queue.

For this tutorial, we will query the conversations API to find the most recent active voice conversation associated with a specific user. This simulates the “find the call” step.

Endpoint: GET /api/v2/conversations
Scope: conversation:read

from genesyscloud.conversations.api import ConversationApi
from genesyscloud.conversations.model import ConversationQuery

def find_active_voice_conversation(platform_client, user_id: str) -> str:
    """
    Finds the most recent active voice conversation for a given user.
    Returns the conversation ID if found, otherwise raises an exception.
    """
    conversation_api = ConversationApi(platform_client)

    try:
        # Query for active conversations
        # expand=user ensures we get user details to filter by agent
        response = conversation_api.post_conversations_query(
            body=ConversationQuery(
                expand=["user"],
                state="active"
            )
        )

        if not response or not response.conversations:
            raise LookupError("No active conversations found.")

        # Iterate through conversations to find a voice call involving the specific user
        for conv in response.conversations:
            if conv.type != "voice":
                continue

            # Check if the target user is a participant in this conversation
            # Note: In a real scenario, you might filter by queue ID or other attributes
            for participant in conv.participants:
                if participant.id == user_id:
                    print(f"Found active voice conversation: {conv.id}")
                    return conv.id

        raise LookupError(f"No active voice conversation found for user {user_id}.")

    except Exception as e:
        print(f"Error retrieving conversations: {e}")
        raise

Step 2: Prepare the Transfer Payload

The core of the transfer logic lies in the routingData object within the conversation’s attributes. Genesys Cloud uses a flexible attributes model to store routing intent.

To transfer a call to a queue, you must set the routingData attribute with the following structure:

• routingData.queueId: The UUID of the target queue.
• routingData.skillRequirements: (Optional) Specific skills required for the next agent.
• routingData.attributes: (Optional) Additional metadata passed to the next queue.

If you do not specify routingData, the system may re-route based on the existing context or fail if the current context is no longer valid. Explicitly setting the queue ID ensures deterministic routing.

Important: You cannot change the type of the conversation. You can only modify attributes and routing data.

from genesyscloud.conversations.model import ConversationPatchRequest

def create_transfer_payload(target_queue_id: str) -> ConversationPatchRequest:
    """
    Constructs the JSON payload required to transfer a call to a new queue.
    """
    # The attributes object holds the routing intent
    # This structure is documented in the Genesys Cloud API Reference for Conversation Attributes
    routing_attributes = {
        "routingData": {
            "queueId": target_queue_id,
            "skillRequirements": [],  # Empty list resets skill requirements, or specify skills here
            "attributes": {
                "transferReason": "Programmatic Transfer via API",
                "originalQueueId": "previous-queue-uuid"  # Example metadata
            }
        }
    }

    # Create the patch request object
    # We use 'attributes' to update the routingData
    # Note: The SDK maps this to the correct JSON structure
    return ConversationPatchRequest(
        attributes=routing_attributes
    )

Step 3: Execute the Transfer via PATCH

Now that we have the conversation ID and the transfer payload, we execute the PATCH request.

Endpoint: PATCH /api/v2/conversations/{conversationId}
Scope: conversation:write

The patch_conversations_conversation method in the SDK sends the partial update to the server. The server validates the routingData and updates the conversation’s state. If successful, the conversation remains “active” but is now queued for the new destination.

from genesyscloud.conversations.api import ConversationApi
import httpx

def transfer_call(platform_client, conversation_id: str, target_queue_id: str) -> bool:
    """
    Executes the transfer of an active voice conversation to a new queue.
    """
    conversation_api = ConversationApi(platform_client)

    # Create the payload
    patch_body = create_transfer_payload(target_queue_id)

    try:
        # Execute the PATCH request
        # The SDK handles serialization of ConversationPatchRequest to JSON
        response = conversation_api.patch_conversations_conversation(
            conversation_id=conversation_id,
            body=patch_body
        )

        # Check the response status
        # A successful PATCH usually returns 200 OK with the updated conversation object
        if response:
            print(f"Successfully transferred conversation {conversation_id} to queue {target_queue_id}.")
            
            # Verify the routing data was updated in the response
            if response.attributes and "routingData" in response.attributes:
                updated_queue = response.attributes["routingData"].get("queueId")
                print(f"Verified: Conversation is now routed to queue {updated_queue}")
                return True
            else:
                print("Warning: Transfer executed, but routingData not present in response attributes.")
                return True
        else:
            print("Error: Empty response from transfer request.")
            return False

    except httpx.HTTPStatusError as e:
        # Handle specific HTTP errors
        if e.response.status_code == 404:
            print(f"Conversation {conversation_id} not found. It may have ended.")
        elif e.response.status_code == 400:
            print(f"Bad Request: Invalid payload or conversation state. Response: {e.response.text}")
        elif e.response.status_code == 403:
            print(f"Forbidden: Check OAuth scopes. You need 'conversation:write'.")
        else:
            print(f"HTTP Error {e.response.status_code}: {e.response.text}")
        return False
    except Exception as e:
        print(f"Unexpected error during transfer: {e}")
        return False

Complete Working Example

Below is the full, copy-pasteable Python script. It combines authentication, conversation lookup, and transfer logic into a single executable module.

Instructions:

1. Set the environment variables: GENESYS_CLIENT_ID, GENESYS_CLIENT_SECRET, GENESYS_REGION.
2. Replace YOUR_USER_ID with an active agent’s UUID.
3. Replace TARGET_QUEUE_ID with the UUID of the destination queue.
4. Run the script.

import os
import sys
import httpx
from genesyscloud.platform.client import PureCloudPlatformClientV2
from genesyscloud.conversations.api import ConversationApi
from genesyscloud.conversations.model import ConversationQuery, ConversationPatchRequest

def get_platform_client() -> PureCloudPlatformClientV2:
    """Initializes the Genesys Cloud Platform Client."""
    client_id = os.environ.get("GENESYS_CLIENT_ID")
    client_secret = os.environ.get("GENESYS_CLIENT_SECRET")
    region = os.environ.get("GENESYS_REGION", "us-east-1")

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

    platform_client = PureCloudPlatformClientV2()
    platform_client.set_region(region)
    platform_client.set_client_id(client_id)
    platform_client.set_client_secret(client_secret)

    return platform_client

def find_active_voice_conversation(platform_client: PureCloudPlatformClientV2, user_id: str) -> str:
    """
    Finds the most recent active voice conversation for a given user.
    """
    conversation_api = ConversationApi(platform_client)

    try:
        response = conversation_api.post_conversations_query(
            body=ConversationQuery(
                expand=["user"],
                state="active"
            )
        )

        if not response or not response.conversations:
            raise LookupError("No active conversations found.")

        for conv in response.conversations:
            if conv.type != "voice":
                continue

            for participant in conv.participants:
                if participant.id == user_id:
                    print(f"Found active voice conversation: {conv.id}")
                    return conv.id

        raise LookupError(f"No active voice conversation found for user {user_id}.")

    except Exception as e:
        print(f"Error retrieving conversations: {e}")
        raise

def transfer_call(platform_client: PureCloudPlatformClientV2, conversation_id: str, target_queue_id: str) -> bool:
    """
    Executes the transfer of an active voice conversation to a new queue.
    """
    conversation_api = ConversationApi(platform_client)

    # Construct the routing attributes
    routing_attributes = {
        "routingData": {
            "queueId": target_queue_id,
            "skillRequirements": [],
            "attributes": {
                "transferReason": "API Transfer"
            }
        }
    }

    patch_body = ConversationPatchRequest(
        attributes=routing_attributes
    )

    try:
        print(f"Initiating transfer of conversation {conversation_id} to queue {target_queue_id}...")
        
        response = conversation_api.patch_conversations_conversation(
            conversation_id=conversation_id,
            body=patch_body
        )

        if response:
            print("Transfer request accepted.")
            if response.attributes and "routingData" in response.attributes:
                updated_queue = response.attributes["routingData"].get("queueId")
                print(f"Verification: Conversation is now routed to queue {updated_queue}")
            return True
        else:
            print("Error: Empty response from transfer request.")
            return False

    except httpx.HTTPStatusError as e:
        print(f"HTTP Error {e.response.status_code}: {e.response.text}")
        return False
    except Exception as e:
        print(f"Unexpected error during transfer: {e}")
        return False

def main():
    # Configuration
    # Replace these with actual UUIDs from your Genesys Cloud instance
    TARGET_USER_ID = os.environ.get("TARGET_USER_ID", "YOUR_USER_ID_UUID")
    TARGET_QUEUE_ID = os.environ.get("TARGET_QUEUE_ID", "YOUR_QUEUE_ID_UUID")

    if TARGET_USER_ID == "YOUR_USER_ID_UUID" or TARGET_QUEUE_ID == "YOUR_QUEUE_ID_UUID":
        print("Please set TARGET_USER_ID and TARGET_QUEUE_ID environment variables or edit the script.")
        sys.exit(1)

    try:
        # 1. Authenticate
        print("Initializing Genesys Cloud Client...")
        platform_client = get_platform_client()

        # 2. Find the conversation
        print("Searching for active voice conversation...")
        conversation_id = find_active_voice_conversation(platform_client, TARGET_USER_ID)

        # 3. Transfer the call
        print("Transferring call...")
        success = transfer_call(platform_client, conversation_id, TARGET_QUEUE_ID)

        if success:
            print("Process completed successfully.")
        else:
            print("Process failed.")
            sys.exit(1)

    except ValueError as ve:
        print(f"Configuration Error: {ve}")
        sys.exit(1)
    except LookupError as le:
        print(f"Lookup Error: {le}")
        sys.exit(1)
    except Exception as e:
        print(f"Fatal Error: {e}")
        sys.exit(1)

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 403 Forbidden

Cause: The OAuth token used does not have the conversation:write scope.
Fix:

1. Go to the Genesys Cloud Admin Portal.
2. Navigate to Developers > API Access > OAuth Clients.
3. Select your client.
4. In the Scopes tab, ensure conversation:write is checked.
5. Regenerate your token or restart your application to pick up the new scope.

Error: 400 Bad Request - “Invalid routingData”

Cause: The queueId provided does not exist, is not a valid UUID, or the conversation type is incompatible (e.g., trying to route a chat to a voice queue without proper configuration).
Fix:

1. Verify the TARGET_QUEUE_ID is a valid UUID of an existing queue in your Genesys Cloud instance.
2. Ensure the conversation type (voice) matches the queue’s supported media types.
3. Check that the routingData structure matches the API specification exactly. Missing queueId inside routingData will cause failure.

Error: 404 Not Found

Cause: The conversation_id is invalid or the conversation has already ended.
Fix:

1. Ensure the conversation_id passed to the PATCH endpoint is correct.
2. Check the conversation state. If the call was hung up before the PATCH request completed, the conversation is no longer active and cannot be modified.
3. Implement idempotency checks or retry logic with a short delay if this occurs in high-throughput environments.

Error: 429 Too Many Requests

Cause: You have exceeded the rate limit for the Conversations API.
Fix:

1. The Genesys Cloud Python SDK includes built-in retry logic for 429 errors. Ensure you are not bypassing the SDK with raw requests calls.
2. If using the SDK, check the retry_config settings.
3. Implement exponential backoff in your application logic if you are making bulk updates.

Official References

• Genesys Cloud Conversations API Documentation
• Conversation Patch Request Model
• OAuth Scopes Reference
• Genesys Cloud Python SDK Documentation