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/callsdirectly via therequestslibrary to demonstrate precise control over call initiation parameters. - The implementation is written in Python 3.8+, utilizing the
requestslibrary for HTTP communication andpyjwtis 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 beuserif referencing a user ID, orphoneif referencing a number directly.role: Typicallyagentfor 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:writepermissions 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:
- Verify the OAuth Client in the Admin Console has
conversation:writescope. - Ensure the Service Account has a role that includes “Create conversation” permissions.
- Check if the
agent_idprovided 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.