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

What You Will Build

• This tutorial demonstrates how to execute a blind transfer of an active voice conversation from one routing queue to another using the Genesys Cloud Conversations API.
• The implementation uses the PATCH method on the /api/v2/conversations/voice/{conversationId} endpoint with the PureCloudPlatformClientV2 Python SDK.
• The code is written in Python 3.10+ and handles OAuth2 authentication, conversation state validation, and error response parsing.

Prerequisites

• OAuth Client: A Genesys Cloud OAuth application with the client_credentials grant type.
• Required Scopes:
  • conversations:read (to verify conversation status and participants)
  • conversations:write (to execute the transfer action)
• SDK Version: genesys-cloud-sdk-python version 120.0.0 or later.
• Runtime: Python 3.10 or higher.
• Dependencies:

  pip install genesys-cloud-sdk-python python-dotenv
  

Authentication Setup

Genesys Cloud APIs require OAuth 2.0 Bearer tokens. For server-side integrations, the client_credentials flow is standard. The SDK handles token caching and automatic refresh, but you must initialize the client correctly.

Create a .env file in your project root with the following variables:

GENESYS_CLOUD_REGION=us-east-1
GENESYS_CLOUD_CLIENT_ID=your-client-id
GENESYS_CLOUD_CLIENT_SECRET=your-client-secret

Initialize the API client using the configuration builder. This approach ensures that the correct regional endpoint is used and that the token manager is instantiated with the correct credentials.

import os
from dotenv import load_dotenv
from platformclientv2 import Configuration, ApiClient
from platformclientv2.auth import OAuthClientCredentials

load_dotenv()

def get_api_client() -> ApiClient:
    """
    Initializes and returns a configured Genesys Cloud API client.
    """
    config = Configuration(
        host=os.getenv("GENESYS_CLOUD_REGION"),
        client_id=os.getenv("GENESYS_CLOUD_CLIENT_ID"),
        client_secret=os.getenv("GENESYS_CLOUD_CLIENT_SECRET")
    )
    
    # The SDK automatically handles token acquisition and refresh
    api_client = ApiClient(configuration=config)
    return api_client

Implementation

Step 1: Validate Conversation Status

Before issuing a transfer command, you must verify that the conversation exists and is in a state that allows transfer. You cannot transfer a conversation that is already terminated or parked. The most reliable way to determine the current state is to GET the conversation details.

We will retrieve the conversation by ID and check the state field.

from platformclientv2 import ConversationApi
from platformclientv2.api_exception import ApiException

def get_conversation_details(api_client: ApiClient, conversation_id: str) -> dict:
    """
    Retrieves conversation details to validate state.
    
    Args:
        api_client: The initialized API client.
        conversation_id: The UUID of the conversation.
        
    Returns:
        The conversation object.
        
    Raises:
        ApiException: If the API call fails.
    """
    conversation_api = ConversationApi(api_client)
    
    try:
        # GET /api/v2/conversations/voice/{conversationId}
        response = conversation_api.get_conversation_voice(conversation_id)
        return response
    except ApiException as e:
        if e.status == 404:
            raise ValueError(f"Conversation {conversation_id} not found.")
        elif e.status == 401:
            raise ValueError("Authentication failed. Check OAuth credentials.")
        else:
            raise e

Expected Response Snippet:

{
  "id": "a1b2c3d4-e5f6-7890-g1h2-i3j4k5l6m7n8",
  "externalConvId": null,
  "type": "voice",
  "state": "active",
  "createdTime": "2023-10-27T10:00:00.000Z",
  "modifiedTime": "2023-10-27T10:05:00.000Z",
  "participants": [
    {
      "id": "participant-id-1",
      "role": "agent",
      "state": "connected"
    },
    {
      "id": "participant-id-2",
      "role": "customer",
      "state": "connected"
    }
  ]
}

If response.state is not active or ringing, the transfer should be aborted. A transfer on a terminated call will result in a 409 Conflict.

Step 2: Construct the Transfer Payload

The PATCH operation on the Conversations API uses a specific action model to modify conversation attributes. To transfer a voice call, you must use the transfer action.

The critical fields are:

• action: Must be set to "transfer".
• from: The ID of the participant initiating the transfer (usually the agent).
• to: The ID of the destination queue.
• type: The type of transfer. Use "blind" for a blind transfer (agent drops immediately) or "consultative" for a consultative transfer (agent stays on bridge). This tutorial focuses on blind.

You must obtain the Queue ID first. If you do not know the Queue ID, you can search for it by name.

from platformclientv2 import RoutingApi
from platformclientv2.models import QueueSearchRequest

def find_queue_id(api_client: ApiClient, queue_name: str) -> str:
    """
    Searches for a queue by name and returns its ID.
    
    Args:
        api_client: The initialized API client.
        queue_name: The display name of the target queue.
        
    Returns:
        The Queue ID string.
    """
    routing_api = RoutingApi(api_client)
    
    # POST /api/v2/routing/queues/search
    search_request = QueueSearchRequest(
        query=queue_name,
        size=1
    )
    
    try:
        response = routing_api.post_routing_queues_search(search_request)
        if response.entities and len(response.entities) > 0:
            return response.entities[0].id
        else:
            raise ValueError(f"Queue '{queue_name}' not found.")
    except ApiException as e:
        raise ValueError(f"Failed to find queue: {e.body}")

Once you have the Queue ID and the Agent Participant ID, construct the transfer request body.

Required OAuth Scope: conversations:write

Step 3: Execute the Transfer

Now we perform the PATCH call. The SDK provides a dedicated method for this, but it is important to understand that this is an asynchronous operation in the backend. The API returns 200 OK immediately if the request is accepted. The actual transfer happens shortly after.

from platformclientv2.models import ConversationPatch, ConversationPatchAction

def transfer_call(api_client: ApiClient, conversation_id: str, agent_participant_id: str, target_queue_id: str) -> bool:
    """
    Executes a blind transfer of the conversation to a target queue.
    
    Args:
        api_client: The initialized API client.
        conversation_id: The UUID of the conversation.
        agent_participant_id: The participant ID of the agent initiating the transfer.
        target_queue_id: The ID of the destination queue.
        
    Returns:
        True if the transfer request was accepted.
        
    Raises:
        ApiException: If the API call fails.
    """
    conversation_api = ConversationApi(api_client)
    
    # Construct the action object
    # Note: The SDK uses a specific model for the action
    transfer_action = ConversationPatchAction(
        action="transfer",
        from_id=agent_participant_id,
        to_id=target_queue_id,
        type="blind"
    )
    
    # Construct the patch request body
    patch_body = ConversationPatch(
        actions=[transfer_action]
    )
    
    try:
        # PATCH /api/v2/conversations/voice/{conversationId}
        # The response body is typically empty or minimal for PATCH operations
        conversation_api.patch_conversation_voice(
            conversation_id,
            body=patch_body
        )
        return True
    except ApiException as e:
        if e.status == 409:
            raise ValueError("Transfer failed: Conversation is in an invalid state (e.g., already terminated or transferring).")
        elif e.status == 400:
            raise ValueError(f"Bad Request: {e.body}. Check participant IDs and queue ID.")
        else:
            raise e

HTTP Request Details:

• Method: PATCH
• Path: /api/v2/conversations/voice/{conversationId}
• Headers:
  • Authorization: Bearer <token>
  • Content-Type: application/json
• Body:

  {
    "actions": [
      {
        "action": "transfer",
        "from": "agent-participant-uuid",
        "to": "target-queue-uuid",
        "type": "blind"
      }
    ]
  }
  

Response:

• 200 OK: The transfer request has been accepted. The conversation state will change to transfer or ringing shortly.

Complete Working Example

This script combines all steps into a single executable module. It assumes you have a .env file configured.

import os
import sys
from dotenv import load_dotenv
from platformclientv2 import Configuration, ApiClient, ConversationApi, RoutingApi
from platformclientv2.auth import OAuthClientCredentials
from platformclientv2.api_exception import ApiException
from platformclientv2.models import ConversationPatch, ConversationPatchAction, QueueSearchRequest

load_dotenv()

def initialize_api_client() -> ApiClient:
    """Initializes the Genesys Cloud API client."""
    try:
        config = Configuration(
            host=os.getenv("GENESYS_CLOUD_REGION"),
            client_id=os.getenv("GENESYS_CLOUD_CLIENT_ID"),
            client_secret=os.getenv("GENESYS_CLOUD_CLIENT_SECRET")
        )
        return ApiClient(configuration=config)
    except Exception as e:
        print(f"Failed to initialize API client: {e}")
        sys.exit(1)

def validate_conversation(api_client: ApiClient, conversation_id: str) -> str:
    """Validates the conversation state and returns the agent participant ID."""
    conversation_api = ConversationApi(api_client)
    try:
        conv = conversation_api.get_conversation_voice(conversation_id)
        if conv.state not in ["active", "ringing"]:
            raise ValueError(f"Cannot transfer conversation in state '{conv.state}'.")
        
        # Identify the agent participant
        agent_id = None
        for participant in conv.participants:
            if participant.role == "agent":
                agent_id = participant.id
                break
        
        if not agent_id:
            raise ValueError("No agent participant found in this conversation.")
            
        return agent_id
    except ApiException as e:
        raise ValueError(f"Error retrieving conversation: {e.body}")

def get_target_queue_id(api_client: ApiClient, queue_name: str) -> str:
    """Finds the queue ID by name."""
    routing_api = RoutingApi(api_client)
    try:
        search_req = QueueSearchRequest(query=queue_name, size=1)
        result = routing_api.post_routing_queues_search(search_req)
        if not result.entities:
            raise ValueError(f"Queue '{queue_name}' not found.")
        return result.entities[0].id
    except ApiException as e:
        raise ValueError(f"Error searching queue: {e.body}")

def execute_transfer(api_client: ApiClient, conversation_id: str, agent_id: str, queue_id: str):
    """Executes the blind transfer."""
    conversation_api = ConversationApi(api_client)
    
    # Build the transfer action
    action = ConversationPatchAction(
        action="transfer",
        from_id=agent_id,
        to_id=queue_id,
        type="blind"
    )
    
    # Build the patch body
    body = ConversationPatch(actions=[action])
    
    try:
        # Execute PATCH
        conversation_api.patch_conversation_voice(conversation_id, body=body)
        print(f"Transfer request accepted for conversation {conversation_id}.")
    except ApiException as e:
        print(f"Transfer failed with status {e.status}: {e.body}")
        raise

def main():
    # Configuration
    CONVERSATION_ID = os.getenv("TEST_CONVERSATION_ID")
    TARGET_QUEUE_NAME = os.getenv("TARGET_QUEUE_NAME", "Support_Tier2")
    
    if not CONVERSATION_ID:
        print("Error: TEST_CONVERSATION_ID not set in .env")
        sys.exit(1)

    print(f"Starting transfer for conversation: {CONVERSATION_ID}")
    
    # Step 1: Initialize Client
    api_client = initialize_api_client()
    
    try:
        # Step 2: Validate Conversation & Get Agent ID
        print("Validating conversation state...")
        agent_id = validate_conversation(api_client, CONVERSATION_ID)
        print(f"Found agent participant: {agent_id}")
        
        # Step 3: Find Target Queue
        print(f"Finding queue: {TARGET_QUEUE_NAME}")
        target_queue_id = get_target_queue_id(api_client, TARGET_QUEUE_NAME)
        print(f"Found target queue ID: {target_queue_id}")
        
        # Step 4: Execute Transfer
        print("Executing blind transfer...")
        execute_transfer(api_client, CONVERSATION_ID, agent_id, target_queue_id)
        print("Transfer initiated successfully.")
        
    except ValueError as ve:
        print(f"Validation Error: {ve}")
    except Exception as e:
        print(f"Unexpected Error: {e}")

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 409 Conflict

• Cause: The conversation is not in a transferable state. This typically happens if the conversation has already been terminated, is currently in the middle of another transfer, or is parked.
• Fix: Check the state field of the conversation via GET /api/v2/conversations/voice/{conversationId}. Ensure the state is active or ringing. If the state is transfer, wait for the previous transfer to complete or fail before retrying.

Error: 400 Bad Request

• Cause: Invalid participant ID or invalid queue ID. The from ID must belong to a participant currently in the conversation. The to ID must be a valid Queue, User, or External Contact ID.
• Fix: Verify that agent_participant_id matches one of the IDs in the participants array of the conversation. Verify that target_queue_id exists in the Routing Queues API.

Error: 403 Forbidden

• Cause: The OAuth token lacks the conversations:write scope, or the application does not have permission to modify conversations.
• Fix: Update the OAuth application scopes in the Genesys Cloud Admin Console to include conversations:write. Ensure the token is refreshed if it was generated with insufficient scopes.

Error: 429 Too Many Requests

• Cause: Rate limiting. The Conversations API has strict rate limits. Executing transfers in a tight loop without delay will trigger this.
• Fix: Implement exponential backoff. Check the Retry-After header in the response.

import time

def safe_transfer_with_retry(api_client, conversation_id, agent_id, queue_id, max_retries=3):
    for attempt in range(max_retries):
        try:
            execute_transfer(api_client, conversation_id, agent_id, queue_id)
            return
        except ApiException as e:
            if e.status == 429:
                wait_time = int(e.headers.get("Retry-After", 2 ** attempt))
                print(f"Rate limited. Waiting {wait_time} seconds...")
                time.sleep(wait_time)
            else:
                raise e
    raise Exception("Max retries exceeded for transfer.")

Official References

• Conversations API Documentation
• PATCH Conversation Endpoint
• Python SDK Reference
• OAuth 2.0 Client Credentials Flow