Choosing the Right OAuth Grant for Server-Side Analytics: Client Credentials vs Authorization Code
What You Will Build
- A Python script that authenticates to Genesys Cloud CX and retrieves high-volume conversation analytics data without requiring user interaction.
- This tutorial uses the Genesys Cloud CX REST API (
/api/v2/analytics/conversations/details/query) and the official Python SDK (genesys-cloud-sdk). - The programming language covered is Python 3.9+.
Prerequisites
- OAuth Client Type: A Genesys Cloud application with the Client Credentials grant type enabled. This is non-negotiable for server-side reporting applications that run without a human user present.
- Required Scopes:
analytics:conversation:readandconversation:read. If you need user-specific metadata, adduser:read. - SDK Version:
genesys-cloud-sdkv10.0.0 or higher. - Runtime: Python 3.9+ with
pipinstalled. - Dependencies:
pip install genesys-cloud-sdk requests python-dotenv
Authentication Setup
The choice between Client Credentials and Authorization Code flows defines the lifecycle of your token and the identity of the API caller. For a server-side reporting app, you must use Client Credentials.
Why Client Credentials?
The Authorization Code flow requires a human user to log in via a browser, approve permissions, and return a code to your backend. This flow expires quickly (typically 1 hour) and requires a refresh token mechanism tied to a specific user session. If your reporting script runs on a cron job at 3 AM, there is no user to authorize the flow. The Authorization Code flow will fail.
The Client Credentials flow uses the client_id and client_secret to request an access token directly from the Genesys Cloud authorization server. The token is valid for one hour, but since your code handles the refresh logic automatically, the user experience is seamless: the script runs, authenticates, fetches data, and exits.
Obtaining Credentials
- Log in to the Genesys Cloud Admin Portal.
- Navigate to Developers > Apps.
- Create a new Application.
- In the Auth Type dropdown, select Client Credentials.
- Copy the Client ID and Client Secret. Store these in a
.envfile.
# .env
GENESYS_CLIENT_ID=your_client_id_here
GENESYS_CLIENT_SECRET=your_client_secret_here
GENESYS_REGION=us-east-1
Implementation
Step 1: Initializing the SDK with Client Credentials
The Genesys Cloud Python SDK simplifies the OAuth handshake. You do not need to manually construct the POST request to /oauth/token. The SDK handles the token acquisition, caching, and automatic refresh when the token nears expiration.
We will use the PlatformClient class to handle authentication.
import os
import sys
from dotenv import load_dotenv
from purecloudplatformclientv2 import (
PlatformClient,
AnalyticsApi,
ConversationDetailQueryRequest,
ConversationDetailQueryResponse
)
# Load environment variables
load_dotenv()
def initialize_platform_client() -> PlatformClient:
"""
Initializes the Genesys Cloud Platform Client using Client Credentials.
This handles token acquisition and refresh automatically.
"""
client_id = os.getenv("GENESYS_CLIENT_ID")
client_secret = os.getenv("GENESYS_CLIENT_SECRET")
region = os.getenv("GENESYS_REGION", "us-east-1")
if not client_id or not client_secret:
raise ValueError("GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET must be set in .env")
# Determine the base URL based on region
if region == "us-east-1":
host = "api.mypurecloud.com"
elif region == "eu-west-1":
host = "api.eu.purecloud.com"
else:
host = f"api.{region}.mypurecloud.com"
platform_client = PlatformClient(host=host)
# Authenticate using Client Credentials
# The SDK manages the token lifecycle here
platform_client.set_credentials(
client_id=client_id,
client_secret=client_secret,
grant_type="client_credentials"
)
return platform_client
if __name__ == "__main__":
try:
client = initialize_platform_client()
print("Authentication successful. Token acquired.")
except Exception as e:
print(f"Authentication failed: {e}")
sys.exit(1)
Expected Response:
If successful, the console prints “Authentication successful. Token acquired.” The SDK stores the access token in memory. If the token expires during a long-running script, the SDK intercepts subsequent API calls, refreshes the token in the background, and retries the failed call.
Error Handling:
- 401 Unauthorized: Check your
client_idandclient_secret. Ensure the application is Active in the Admin Portal. - 403 Forbidden: Ensure the application has the required OAuth scopes assigned in the Admin Portal.
Step 2: Constructing the Analytics Query
Reporting in Genesys Cloud is not a simple “get all records” operation. It uses a query-based model. You must define a time window, the metrics you want, and the grouping dimensions.
For this tutorial, we will retrieve inbound call details for the last 24 hours, grouped by queue.
from datetime import datetime, timedelta, timezone
from purecloudplatformclientv2 import (
AnalyticsApi,
ConversationDetailQueryRequest,
Interval,
QueryFilter,
QueryGroupBy
)
def build_analytics_query(platform_client: PlatformClient) -> ConversationDetailQueryRequest:
"""
Constructs a query for conversation details over the last 24 hours.
"""
analytics_api = AnalyticsApi(platform_client)
# Define the time interval (Last 24 hours)
now = datetime.now(timezone.utc)
start_time = now - timedelta(hours=24)
# Format as ISO 8601 strings
start_time_str = start_time.isoformat()
end_time_str = now.isoformat()
# Define the query body
query_body = {
"interval": Interval(
start=start_time_str,
end=end_time_str
),
"groupBy": [
QueryGroupBy(value="queue")
],
"filters": [
QueryFilter(
attribute="direction",
operator="eq",
value="inbound"
)
],
"metrics": [
"interactions",
"handled",
"abandoned",
"avgHandleTime"
]
}
return query_body
Key Parameters Explained:
interval: Must be ISO 8601 format. The Genesys Cloud API requires UTC timestamps. Local timezones will cause calculation errors.groupBy: This determines the rows in your result set. Grouping byqueuereturns one row per queue. If you omit this, you get a single aggregated row for the entire organization.filters: Reduces the dataset before aggregation. Filtering bydirection == inboundexcludes outbound calls from your report.metrics: These are the columns in your result set. You can only request metrics that are available for the selectedgroupBydimensions.
Step 3: Executing the Query and Handling Pagination
The /api/v2/analytics/conversations/details/query endpoint returns a maximum of 1,000 rows per page. If your organization has more than 1,000 queues (unlikely) or if you group by a high-cardinality dimension like agent, you must handle pagination.
The Genesys Cloud Python SDK provides a convenient get_analytics_conversations_details_query method that returns a paginated iterator.
def fetch_analytics_data(platform_client: PlatformClient, query_body: dict) -> list:
"""
Fetches analytics data with automatic pagination handling.
"""
analytics_api = AnalyticsApi(platform_client)
all_data = []
try:
# The SDK handles pagination automatically when using the iterator
# We pass the query body directly to the method
response = analytics_api.post_analytics_conversations_details_query(
body=query_body,
limit=1000 # Max allowed per page
)
# The response object contains the data and a 'nextPage' token if more data exists
if response.data:
all_data.extend(response.data)
# Handle manual pagination if necessary
# Note: The Python SDK's post method does not auto-paginate in all versions.
# We must check for 'nextPage' token manually for robustness.
page_token = response.next_page
while page_token:
# Fetch next page using the token
response = analytics_api.post_analytics_conversations_details_query(
body=query_body,
limit=1000,
page_token=page_token
)
if response.data:
all_data.extend(response.data)
else:
break
page_token = response.next_page
except Exception as e:
print(f"Error fetching analytics data: {e}")
raise e
return all_data
Why Manual Pagination?
While some SDKs auto-paginate, the Genesys Cloud Analytics API is stateful. The page_token is tied to the specific query parameters. Changing any parameter (even whitespace) invalidates the token. By explicitly passing the page_token from the previous response, we ensure data consistency.
Step 4: Processing and Serializing Results
The raw response from the Analytics API is a complex nested object. For reporting purposes, we typically want a flat list of dictionaries that can be exported to CSV or JSON.
import json
def process_results(data: list) -> list:
"""
Flattens the analytics response into a list of dictionaries.
"""
processed_data = []
for item in data:
# Extract the group-by values
queue_name = item.group_by.get("queue", {}).get("name", "Unknown")
# Extract metrics
interactions = item.metrics.get("interactions", 0)
handled = item.metrics.get("handled", 0)
abandoned = item.metrics.get("abandoned", 0)
avg_handle_time = item.metrics.get("avgHandleTime", 0)
processed_item = {
"queue_name": queue_name,
"total_interactions": interactions,
"handled_calls": handled,
"abandoned_calls": abandoned,
"avg_handle_time_seconds": avg_handle_time
}
processed_data.append(processed_item)
return processed_data
def export_to_json(data: list, filename: str = "report.json"):
"""
Exports the processed data to a JSON file.
"""
with open(filename, 'w') as f:
json.dump(data, f, indent=2)
print(f"Data exported to {filename}")
Complete Working Example
This is the full, copy-pasteable script. Save it as generate_report.py.
import os
import sys
import json
from datetime import datetime, timedelta, timezone
from dotenv import load_dotenv
from purecloudplatformclientv2 import (
PlatformClient,
AnalyticsApi,
Interval,
QueryFilter,
QueryGroupBy
)
def initialize_platform_client() -> PlatformClient:
load_dotenv()
client_id = os.getenv("GENESYS_CLIENT_ID")
client_secret = os.getenv("GENESYS_CLIENT_SECRET")
region = os.getenv("GENESYS_REGION", "us-east-1")
if not client_id or not client_secret:
raise ValueError("GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET must be set in .env")
host_map = {
"us-east-1": "api.mypurecloud.com",
"eu-west-1": "api.eu.purecloud.com"
}
host = host_map.get(region, f"api.{region}.mypurecloud.com")
platform_client = PlatformClient(host=host)
platform_client.set_credentials(
client_id=client_id,
client_secret=client_secret,
grant_type="client_credentials"
)
return platform_client
def build_analytics_query() -> dict:
now = datetime.now(timezone.utc)
start_time = now - timedelta(hours=24)
return {
"interval": Interval(
start=start_time.isoformat(),
end=now.isoformat()
),
"groupBy": [
QueryGroupBy(value="queue")
],
"filters": [
QueryFilter(
attribute="direction",
operator="eq",
value="inbound"
)
],
"metrics": [
"interactions",
"handled",
"abandoned",
"avgHandleTime"
]
}
def fetch_and_process_data(platform_client: PlatformClient) -> list:
analytics_api = AnalyticsApi(platform_client)
query_body = build_analytics_query()
all_data = []
try:
# First page
response = analytics_api.post_analytics_conversations_details_query(
body=query_body,
limit=1000
)
if response.data:
all_data.extend(response.data)
page_token = response.next_page
# Subsequent pages
while page_token:
response = analytics_api.post_analytics_conversations_details_query(
body=query_body,
limit=1000,
page_token=page_token
)
if response.data:
all_data.extend(response.data)
else:
break
page_token = response.next_page
except Exception as e:
print(f"Error fetching data: {e}")
sys.exit(1)
# Process results
processed_data = []
for item in all_data:
queue_name = item.group_by.get("queue", {}).get("name", "Unknown")
interactions = item.metrics.get("interactions", 0)
handled = item.metrics.get("handled", 0)
abandoned = item.metrics.get("abandoned", 0)
avg_handle_time = item.metrics.get("avgHandleTime", 0)
processed_data.append({
"queue_name": queue_name,
"total_interactions": interactions,
"handled_calls": handled,
"abandoned_calls": abandoned,
"avg_handle_time_seconds": avg_handle_time
})
return processed_data
if __name__ == "__main__":
try:
print("Initializing client...")
client = initialize_platform_client()
print("Fetching analytics data...")
data = fetch_and_process_data(client)
print(f"Retrieved {len(data)} rows.")
# Export
with open("analytics_report.json", "w") as f:
json.dump(data, f, indent=2)
print("Report saved to analytics_report.json")
except Exception as e:
print(f"Fatal error: {e}")
sys.exit(1)
Common Errors & Debugging
Error: 401 Unauthorized
- Cause: The
client_idorclient_secretis incorrect, or the application is not active. - Fix: Verify the credentials in your
.envfile. Check the Admin Portal to ensure the application status is Active. - Code Check: Ensure you are using
grant_type="client_credentials"inset_credentials.
Error: 403 Forbidden
- Cause: The application lacks the required OAuth scope.
- Fix: In the Admin Portal, go to Developers > Apps > [Your App] > Scopes. Add
analytics:conversation:read. - Code Check: Ensure your
metricslist only includes metrics available for the selectedgroupBy. RequestingavgHandleTimewithout grouping byagentorqueuemay return null or cause errors in some contexts.
Error: 429 Too Many Requests
- Cause: You have exceeded the Genesys Cloud API rate limit. Analytics queries are heavy and may consume significant quota.
- Fix: Implement exponential backoff. The Genesys Cloud SDK does not automatically retry 429 errors for analytics queries. You must wrap the API call in a retry loop.
- Code Fix:
import time def fetch_with_retry(api_call, max_retries=3): for attempt in range(max_retries): try: return api_call() except Exception as e: if "429" in str(e): wait_time = 2 ** attempt print(f"Rate limited. Retrying in {wait_time} seconds...") time.sleep(wait_time) else: raise e raise Exception("Max retries exceeded")
Error: Empty Data
- Cause: The time interval is in the future, or no conversations match the filter.
- Fix: Ensure
start_timeis in the past. Ensurefiltersare not too restrictive. Test with a broader filter first (e.g., removedirectionfilter).