Initiating Outbound Calls on Behalf of an Agent via REST API

Initiating Outbound Calls on Behalf of an Agent via REST API

What You Will Build

  • A Python script that programmatically initiates a voice conversation from a specified user (agent) to an external phone number.
  • This tutorial uses the Genesys Cloud CX API endpoint POST /api/v2/conversations/calls directly via the requests library to demonstrate precise control over call initiation parameters.
  • The implementation is written in Python 3.8+, utilizing the requests library for HTTP communication and pyjwt is not required as OAuth handling is done via the standard token endpoint.

Prerequisites

  • OAuth Client Type: Service Account (Client Credentials Grant) or Resource Owner Password Credentials (ROPC) if acting as a specific user. For this tutorial, we assume a Service Account with sufficient permissions to create conversations on behalf of others.
  • Required Scopes:
    • conversation:write (Required to create the conversation)
    • user:read (Required to resolve user IDs if you only have user names/emails)
    • routing:user:read (Optional, if you need to verify agent availability status before calling)
  • API Version: Genesys Cloud CX API v2.
  • Language/Runtime: Python 3.8 or higher.
  • External Dependencies:
    • requests: For HTTP calls.
    • python-dotenv: For secure environment variable management (optional but recommended).

Install dependencies:

pip install requests python-dotenv

Authentication Setup

Genesys Cloud uses OAuth 2.0 for authentication. Before making any API calls, you must obtain an access token. For server-to-server integrations, the Client Credentials Grant is the standard approach.

The following function handles the token acquisition and includes basic caching logic to avoid unnecessary token requests.

import requests
import time
import os
from typing import Optional

# Load environment variables
from dotenv import load_dotenv
load_dotenv()

GENESYS_CLOUD_REGION = os.getenv("GENESYS_CLOUD_REGION", "mypurecloud.com")
CLIENT_ID = os.getenv("CLIENT_ID")
CLIENT_SECRET = os.getenv("CLIENT_SECRET")

class GenesysAuth:
    def __init__(self, client_id: str, client_secret: str, region: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.region = region
        self.token_url = f"https://api.{region}/oauth/token"
        self.access_token: Optional[str] = None
        self.token_expiry: float = 0

    def get_token(self) -> str:
        """
        Retrieves an OAuth access token.
        Returns a cached token if it is still valid.
        """
        if self.access_token and time.time() < self.token_expiry:
            return self.access_token

        payload = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret
        }

        headers = {
            "Content-Type": "application/x-www-form-urlencoded"
        }

        try:
            response = requests.post(self.token_url, data=payload, headers=headers)
            response.raise_for_status()
            data = response.json()
            
            self.access_token = data["access_token"]
            # Set expiry slightly before actual expiry to allow for refresh buffer
            self.token_expiry = time.time() + (data["expires_in"] - 60)
            
            return self.access_token
        except requests.exceptions.HTTPError as e:
            raise Exception(f"OAuth Authentication Failed: {response.status_code} - {response.text}") from e
        except requests.exceptions.RequestException as e:
            raise Exception(f"Network error during authentication: {e}") from e

# Initialize Auth
auth_client = GenesysAuth(CLIENT_ID, CLIENT_SECRET, GENESYS_CLOUD_REGION)

Implementation

Step 1: Identify the Agent and Target Number

Before initiating the call, you must know the ID of the agent who will place the call and the destination number. If you only have the agent’s name or email, you must query the User API first.

OAuth Scope: user:read

def get_user_by_email(auth: GenesysAuth, email: str, region: str) -> Optional[str]:
    """
    Retrieves the Genesys Cloud User ID for a given email address.
    """
    base_url = f"https://api.{region}"
    endpoint = f"{base_url}/api/v2/users"
    
    headers = {
        "Authorization": f"Bearer {auth.get_token()}",
        "Content-Type": "application/json"
    }
    
    # Query parameters for searching
    params = {
        "email": email,
        "pageSize": 1,
        "pageNumber": 1
    }

    try:
        response = requests.get(endpoint, headers=headers, params=params)
        response.raise_for_status()
        data = response.json()
        
        if data["entities"] and len(data["entities"]) > 0:
            return data["entities"][0]["id"]
        else:
            return None
    except requests.exceptions.HTTPError as e:
        print(f"HTTP Error fetching user: {response.status_code} - {response.text}")
        return None
    except requests.exceptions.RequestException as e:
        print(f"Request Exception: {e}")
        return None

# Example Usage:
# agent_id = get_user_by_email(auth_client, "agent.name@company.com", GENESYS_CLOUD_REGION)
# if not agent_id:
#     raise ValueError("Agent not found")

For the remainder of this tutorial, we assume agent_id is available. The target number (to_number) should be in E.164 format (e.g., +14155552671).

Step 2: Construct the Conversation Call Payload

The core of this operation is the POST /api/v2/conversations/calls endpoint. This endpoint creates a new voice conversation. To initiate an outbound call on behalf of an agent, you must structure the participants array correctly.

Key parameters:

  • from: The caller ID. This can be a user ID (if the user has a phone number assigned) or a direct phone number string. When using a user ID, Genesys Cloud resolves the user’s configured outbound caller ID.
  • to: The destination number.
  • type: Must be user if referencing a user ID, or phone if referencing a number directly.
  • role: Typically agent for the initiator.

OAuth Scope: conversation:write

def prepare_call_payload(agent_id: str, to_number: str) -> dict:
    """
    Constructs the JSON payload for the outbound call.
    """
    return {
        "participants": [
            {
                "from": {
                    "id": agent_id,
                    "type": "user"
                },
                "to": {
                    "id": to_number,
                    "type": "phone"
                },
                "role": "agent"
            }
        ]
    }

Step 3: Initiate the Call

This step sends the HTTP POST request to Genesys Cloud. You must handle potential errors such as:

  • 400 Bad Request: Invalid phone number format or invalid user ID.
  • 401 Unauthorized: Expired or invalid token.
  • 403 Forbidden: The service account lacks conversation:write permissions or the user ID is not a valid agent.
  • 429 Too Many Requests: Rate limiting.
def initiate_outbound_call(auth: GenesysAuth, agent_id: str, to_number: str, region: str) -> dict:
    """
    Initiates an outbound call using the Genesys Cloud API.
    """
    base_url = f"https://api.{region}"
    endpoint = f"{base_url}/api/v2/conversations/calls"
    
    headers = {
        "Authorization": f"Bearer {auth.get_token()}",
        "Content-Type": "application/json"
    }
    
    payload = prepare_call_payload(agent_id, to_number)

    try:
        response = requests.post(endpoint, headers=headers, json=payload)
        
        # Check for success
        if response.status_code == 201:
            result = response.json()
            print(f"Call initiated successfully. Conversation ID: {result['id']}")
            return result
        else:
            # Handle specific error codes
            error_body = response.json() if response.text else {}
            raise Exception(f"API Error {response.status_code}: {error_body}")
            
    except requests.exceptions.HTTPError as e:
        print(f"HTTP Error: {e}")
        raise
    except requests.exceptions.RequestException as e:
        print(f"Network Error: {e}")
        raise
    except Exception as e:
        print(f"Unexpected Error: {e}")
        raise

Complete Working Example

Below is the full, copy-pasteable script. It combines authentication, user resolution, and call initiation into a single workflow.

import requests
import time
import os
import sys
from typing import Optional

from dotenv import load_dotenv

# Load environment variables from .env file
load_dotenv()

# Configuration
GENESYS_CLOUD_REGION = os.getenv("GENESYS_CLOUD_REGION", "mypurecloud.com")
CLIENT_ID = os.getenv("CLIENT_ID")
CLIENT_SECRET = os.getenv("CLIENT_SECRET")

if not CLIENT_ID or not CLIENT_SECRET:
    raise EnvironmentError("CLIENT_ID and CLIENT_SECRET must be set in environment variables.")

class GenesysClient:
    def __init__(self, client_id: str, client_secret: str, region: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.region = region
        self.token_url = f"https://api.{region}/oauth/token"
        self.access_token: Optional[str] = None
        self.token_expiry: float = 0

    def _get_token(self) -> str:
        """Retrieves or refreshes the OAuth access token."""
        if self.access_token and time.time() < self.token_expiry:
            return self.access_token

        payload = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret
        }

        headers = {
            "Content-Type": "application/x-www-form-urlencoded"
        }

        try:
            response = requests.post(self.token_url, data=payload, headers=headers)
            response.raise_for_status()
            data = response.json()
            
            self.access_token = data["access_token"]
            self.token_expiry = time.time() + (data["expires_in"] - 60)
            return self.access_token
        except requests.exceptions.HTTPError as e:
            raise Exception(f"OAuth Authentication Failed: {response.status_code} - {response.text}") from e
        except requests.exceptions.RequestException as e:
            raise Exception(f"Network error during authentication: {e}") from e

    def get_user_id_by_email(self, email: str) -> Optional[str]:
        """Resolves a user ID from an email address."""
        endpoint = f"https://api.{self.region}/api/v2/users"
        headers = {
            "Authorization": f"Bearer {self._get_token()}",
            "Content-Type": "application/json"
        }
        params = {
            "email": email,
            "pageSize": 1,
            "pageNumber": 1
        }

        try:
            response = requests.get(endpoint, headers=headers, params=params)
            response.raise_for_status()
            data = response.json()
            
            if data["entities"] and len(data["entities"]) > 0:
                return data["entities"][0]["id"]
            return None
        except requests.exceptions.RequestException as e:
            print(f"Error fetching user: {e}")
            return None

    def make_outbound_call(self, agent_id: str, to_number: str) -> dict:
        """
        Initiates an outbound call on behalf of the specified agent.
        
        Args:
            agent_id (str): The Genesys Cloud User ID of the agent.
            to_number (str): The destination phone number in E.164 format.
            
        Returns:
            dict: The API response containing the conversation ID.
        """
        endpoint = f"https://api.{self.region}/api/v2/conversations/calls"
        headers = {
            "Authorization": f"Bearer {self._get_token()}",
            "Content-Type": "application/json"
        }
        
        payload = {
            "participants": [
                {
                    "from": {
                        "id": agent_id,
                        "type": "user"
                    },
                    "to": {
                        "id": to_number,
                        "type": "phone"
                    },
                    "role": "agent"
                }
            ]
        }

        try:
            response = requests.post(endpoint, headers=headers, json=payload)
            
            if response.status_code == 201:
                result = response.json()
                print(f"Success: Call initiated. Conversation ID: {result['id']}")
                return result
            else:
                error_details = response.json() if response.text else "No details provided"
                raise Exception(f"API Error {response.status_code}: {error_details}")
                
        except requests.exceptions.RequestException as e:
            raise Exception(f"Network error during call initiation: {e}") from e

def main():
    # 1. Initialize Client
    client = GenesysClient(CLIENT_ID, CLIENT_SECRET, GENESYS_CLOUD_REGION)

    # 2. Define Agent and Target
    agent_email = os.getenv("AGENT_EMAIL", "agent@example.com")
    target_number = os.getenv("TARGET_NUMBER", "+14155551234")

    print(f"Looking up user ID for: {agent_email}")
    agent_id = client.get_user_id_by_email(agent_email)

    if not agent_id:
        print("Error: Could not find user with the provided email.")
        sys.exit(1)

    print(f"Found Agent ID: {agent_id}")
    print(f"Initiating call to: {target_number}")

    try:
        # 3. Initiate Call
        result = client.make_outbound_call(agent_id, target_number)
        
        # 4. Optional: Log Conversation ID for tracking
        conversation_id = result["id"]
        print(f"Monitor this conversation at: https://admin.{GENESYS_CLOUD_REGION}/conversations/voice/{conversation_id}")
        
    except Exception as e:
        print(f"Failed to initiate call: {e}")
        sys.exit(1)

if __name__ == "__main__":
    main()

Common Errors & Debugging

Error: 400 Bad Request - Invalid Phone Number

Cause: The to number is not in E.164 format or contains invalid characters. Genesys Cloud is strict about phone number formatting.
Fix: Ensure the number starts with + followed by the country code and number (e.g., +16505551234). Remove any spaces, dashes, or parentheses.

Error: 403 Forbidden - Permission Denied

Cause: The OAuth client used lacks the conversation:write scope, or the Service Account does not have the necessary role permissions to create conversations on behalf of users.
Fix:

  1. Verify the OAuth Client in the Admin Console has conversation:write scope.
  2. Ensure the Service Account has a role that includes “Create conversation” permissions.
  3. Check if the agent_id provided is actually a valid user and is not disabled.

Error: 429 Too Many Requests

Cause: You have exceeded the rate limit for the POST /api/v2/conversations/calls endpoint.
Fix: Implement exponential backoff. The response headers will include Retry-After.

import time

def retry_with_backoff(func, *args, max_retries=3, **kwargs):
    for attempt in range(max_retries):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            if "429" in str(e) and attempt < max_retries - 1:
                wait_time = 2 ** attempt
                print(f"Rate limited. Retrying in {wait_time} seconds...")
                time.sleep(wait_time)
            else:
                raise e

Error: 401 Unauthorized

Cause: The access token is expired or invalid.
Fix: Ensure your GenesysAuth class correctly refreshes the token before each API call. The provided implementation checks token_expiry automatically.

Official References