Automating Wrap-Up Code Assignment Based on Post-Call Survey Results Using the Genesys Cloud Interaction API and Python

What You Will Build

• A Python script that processes post-call survey responses and automatically assigns the corresponding wrap-up code to a Genesys Cloud interaction.
• This implementation uses the Genesys Cloud Interaction API (/api/v2/interactions/events) and the official Python SDK.
• The tutorial covers Python 3.9+ with type hints, JWT authentication, and production-grade error handling.

Prerequisites

• OAuth Client Type: Confidential Client (Server-to-Server) with JWT grant type enabled.
• Required Scopes: interaction:write, interaction:view
• SDK Version: genesys-cloud-purecloud-platform-client v130.0.0+
• Runtime: Python 3.9+
• Dependencies: httpx>=0.24.0, pyjwt>=2.8.0, genesys-cloud-purecloud-platform-client>=130.0.0

Authentication Setup

Genesys Cloud server-to-server integrations require a JWT grant flow. The following code generates a signed JWT, exchanges it for an access token, and implements token caching with automatic refresh logic.

import time
import httpx
from datetime import datetime, timezone, timedelta
import jwt
from typing import Optional, Dict, Any

class GenesysAuth:
    def __init__(self, org_id: str, client_id: str, private_key_pem: str, env: str = "mypurecloud.com"):
        self.org_id = org_id
        self.client_id = client_id
        self.private_key = private_key_pem
        self.base_url = f"https://api.{env}"
        self.token_url = f"{self.base_url}/oauth/token"
        self.access_token: Optional[str] = None
        self.token_expires_at: Optional[float] = None

    def _generate_jwt(self) -> str:
        now = datetime.now(timezone.utc)
        payload = {
            "iss": self.client_id,
            "sub": self.client_id,
            "aud": self.token_url,
            "iat": now,
            "exp": now + timedelta(minutes=10),
            "jti": str(time.time())
        }
        return jwt.encode(payload, self.private_key, algorithm="RS256")

    def get_access_token(self) -> str:
        if self.access_token and self.token_expires_at and time.time() < self.token_expires_at - 60:
            return self.access_token

        jwt_token = self._generate_jwt()
        headers = {"Content-Type": "application/x-www-form-urlencoded"}
        data = {
            "grant_type": "urn:ietf:params:oauth:grant-type:jwt-bearer",
            "assertion": jwt_token
        }

        response = httpx.post(self.token_url, headers=headers, data=data)
        response.raise_for_status()
        token_data = response.json()

        self.access_token = token_data["access_token"]
        self.token_expires_at = time.time() + token_data["expires_in"] - 60
        return self.access_token

The get_access_token() method caches the token and refreshes it 60 seconds before expiration. This prevents unnecessary OAuth calls during high-throughput survey processing.

Implementation

Step 1: Map Survey Results to Wrap-Up Code IDs

Wrap-up codes in Genesys Cloud are identified by their internal UUID. You must retrieve these IDs once and cache them. The following function queries the wrap-up code definitions and builds a lookup dictionary.

from genesyscloud.platform_client import PureCloudPlatformClientV2
from genesyscloud.wrapup_code.api import WrapupCodeApi

def load_wrapup_codes(platform_client: PureCloudPlatformClientV2, division_id: str) -> Dict[str, str]:
    """
    Retrieves all wrap-up codes for a division and returns a mapping of code strings to UUIDs.
    Scope required: interaction:view
    """
    wrapup_api = WrapupCodeApi(platform_client)
    code_map = {}

    # Pagination handling for wrap-up codes
    page_size = 25
    page_number = 1
    while True:
        try:
            response = wrapup_api.get_wrapupcodes(
                division_id=division_id,
                page_size=page_size,
                page_number=page_number
            )
            if not response.entities:
                break

            for code in response.entities:
                if code.code:
                    code_map[code.code] = code.id

            if page_number >= response.page_count:
                break
            page_number += 1
        except httpx.HTTPStatusError as e:
            if e.response.status_code == 429:
                retry_after = int(e.response.headers.get("Retry-After", 5))
                time.sleep(retry_after)
                continue
            raise

    return code_map

This function handles pagination and implements retry logic for 429 Too Many Requests responses. The Retry-After header dictates the sleep duration. If the header is missing, the script defaults to 5 seconds.

Step 2: Process Survey Payload and Determine Wrap-Up

Post-call surveys typically arrive via webhook or queue consumer. The payload contains an interactionId and response scores. The mapping logic translates survey outcomes into the corresponding wrap-up code string.

from typing import Optional

def determine_wrapup_code(survey_payload: Dict[str, Any], code_map: Dict[str, str]) -> Optional[str]:
    """
    Analyzes survey scores and returns the corresponding wrap-up code UUID.
    Returns None if no wrap-up should be assigned.
    """
    csat_score = survey_payload.get("csat_score", 0)
    nps_score = survey_payload.get("nps_score", 0)
    issue_type = survey_payload.get("issue_type", "").upper()

    # Business logic mapping
    if nps_score <= 6:
        target_code = "WU_PROMPT_ESCALATION"
    elif nps_score >= 9 and csat_score >= 4:
        target_code = "WU_PROMPT_SUCCESS"
    elif issue_type == "TECHNICAL":
        target_code = "WU_PROMPT_TECH_ISSUE"
    else:
        target_code = "WU_PROMPT_GENERAL"

    return code_map.get(target_code)

The function returns the UUID string. If the business logic does not match a known code, the function returns None, preventing invalid API calls.

Step 3: Assign Wrap-Up Code via Interaction API

The Interaction API accepts wrap-up events through POST /api/v2/interactions/events. The SDK method post_interactions_events constructs the request. You must include the interactionId, eventType, and wrapUpCode object.

from genesyscloud.interaction.api import InteractionApi
from genesyscloud.model.post_interactions_events_request import PostInteractionsEventsRequest
from genesyscloud.model.wrapup_code import WrapupCode

def assign_wrapup_code(
    platform_client: PureCloudPlatformClientV2,
    interaction_id: str,
    wrapup_code_id: str,
    wrapup_time: Optional[str] = None
) -> Dict[str, Any]:
    """
    Assigns a wrap-up code to a completed interaction.
    Scope required: interaction:write
    """
    interaction_api = InteractionApi(platform_client)

    if not wrapup_time:
        wrapup_time = datetime.now(timezone.utc).isoformat()

    wrapup_code_obj = WrapupCode(id=wrapup_code_id, code=None)

    event_request = PostInteractionsEventsRequest(
        event_type="wrapup",
        interaction_id=interaction_id,
        wrap_up_code=wrapup_code_obj,
        wrap_up_time=wrapup_time
    )

    try:
        response = interaction_api.post_interactions_events(event_request)
        return response.to_dict()
    except httpx.HTTPStatusError as e:
        status = e.response.status_code
        if status == 429:
            retry_after = int(e.response.headers.get("Retry-After", 5))
            time.sleep(retry_after)
            # Retry once after rate limit
            return assign_wrapup_code(platform_client, interaction_id, wrapup_code_id, wrapup_time)
        elif status == 404:
            raise ValueError(f"Interaction ID {interaction_id} not found")
        elif status == 400:
            raise ValueError(f"Invalid wrap-up code or interaction state: {e.response.text}")
        else:
            raise

The PostInteractionsEventsRequest object serializes to the exact JSON payload expected by the API. The wrapup event type transitions the interaction to a wrapped state and attaches the code. The retry logic handles transient rate limits without blocking the queue consumer.

Complete Working Example

The following script combines authentication, lookup, mapping, and assignment into a single executable module. Replace the placeholder credentials and division ID before execution.

import time
import httpx
from datetime import datetime, timezone, timedelta
import jwt
from typing import Optional, Dict, Any

from genesyscloud.platform_client import PureCloudPlatformClientV2
from genesyscloud.wrapup_code.api import WrapupCodeApi
from genesyscloud.interaction.api import InteractionApi
from genesyscloud.model.post_interactions_events_request import PostInteractionsEventsRequest
from genesyscloud.model.wrapup_code import WrapupCode

class GenesysAuth:
    def __init__(self, org_id: str, client_id: str, private_key_pem: str, env: str = "mypurecloud.com"):
        self.org_id = org_id
        self.client_id = client_id
        self.private_key = private_key_pem
        self.base_url = f"https://api.{env}"
        self.token_url = f"{self.base_url}/oauth/token"
        self.access_token: Optional[str] = None
        self.token_expires_at: Optional[float] = None

    def _generate_jwt(self) -> str:
        now = datetime.now(timezone.utc)
        payload = {
            "iss": self.client_id,
            "sub": self.client_id,
            "aud": self.token_url,
            "iat": now,
            "exp": now + timedelta(minutes=10),
            "jti": str(time.time())
        }
        return jwt.encode(payload, self.private_key, algorithm="RS256")

    def get_access_token(self) -> str:
        if self.access_token and self.token_expires_at and time.time() < self.token_expires_at - 60:
            return self.access_token

        jwt_token = self._generate_jwt()
        headers = {"Content-Type": "application/x-www-form-urlencoded"}
        data = {
            "grant_type": "urn:ietf:params:oauth:grant-type:jwt-bearer",
            "assertion": jwt_token
        }

        response = httpx.post(self.token_url, headers=headers, data=data)
        response.raise_for_status()
        token_data = response.json()

        self.access_token = token_data["access_token"]
        self.token_expires_at = time.time() + token_data["expires_in"] - 60
        return self.access_token

def load_wrapup_codes(platform_client: PureCloudPlatformClientV2, division_id: str) -> Dict[str, str]:
    wrapup_api = WrapupCodeApi(platform_client)
    code_map = {}
    page_size = 25
    page_number = 1
    while True:
        try:
            response = wrapup_api.get_wrapupcodes(
                division_id=division_id,
                page_size=page_size,
                page_number=page_number
            )
            if not response.entities:
                break
            for code in response.entities:
                if code.code:
                    code_map[code.code] = code.id
            if page_number >= response.page_count:
                break
            page_number += 1
        except httpx.HTTPStatusError as e:
            if e.response.status_code == 429:
                retry_after = int(e.response.headers.get("Retry-After", 5))
                time.sleep(retry_after)
                continue
            raise
    return code_map

def determine_wrapup_code(survey_payload: Dict[str, Any], code_map: Dict[str, str]) -> Optional[str]:
    csat_score = survey_payload.get("csat_score", 0)
    nps_score = survey_payload.get("nps_score", 0)
    issue_type = survey_payload.get("issue_type", "").upper()

    if nps_score <= 6:
        target_code = "WU_PROMPT_ESCALATION"
    elif nps_score >= 9 and csat_score >= 4:
        target_code = "WU_PROMPT_SUCCESS"
    elif issue_type == "TECHNICAL":
        target_code = "WU_PROMPT_TECH_ISSUE"
    else:
        target_code = "WU_PROMPT_GENERAL"

    return code_map.get(target_code)

def assign_wrapup_code(
    platform_client: PureCloudPlatformClientV2,
    interaction_id: str,
    wrapup_code_id: str,
    wrapup_time: Optional[str] = None
) -> Dict[str, Any]:
    interaction_api = InteractionApi(platform_client)
    if not wrapup_time:
        wrapup_time = datetime.now(timezone.utc).isoformat()

    wrapup_code_obj = WrapupCode(id=wrapup_code_id, code=None)
    event_request = PostInteractionsEventsRequest(
        event_type="wrapup",
        interaction_id=interaction_id,
        wrap_up_code=wrapup_code_obj,
        wrap_up_time=wrapup_time
    )

    try:
        response = interaction_api.post_interactions_events(event_request)
        return response.to_dict()
    except httpx.HTTPStatusError as e:
        status = e.response.status_code
        if status == 429:
            retry_after = int(e.response.headers.get("Retry-After", 5))
            time.sleep(retry_after)
            return assign_wrapup_code(platform_client, interaction_id, wrapup_code_id, wrapup_time)
        elif status == 404:
            raise ValueError(f"Interaction ID {interaction_id} not found")
        elif status == 400:
            raise ValueError(f"Invalid wrap-up code or interaction state: {e.response.text}")
        else:
            raise

def process_survey(survey_payload: Dict[str, Any], code_map: Dict[str, str], platform_client: PureCloudPlatformClientV2) -> None:
    interaction_id = survey_payload.get("interactionId")
    if not interaction_id:
        raise ValueError("Survey payload missing interactionId")

    wrapup_code_id = determine_wrapup_code(survey_payload, code_map)
    if not wrapup_code_id:
        print("No matching wrap-up code found for survey results.")
        return

    result = assign_wrapup_code(platform_client, interaction_id, wrapup_code_id)
    print(f"Wrap-up assigned successfully. Event ID: {result.get('eventId')}")

if __name__ == "__main__":
    # Configuration
    ORG_ID = "your-org-id"
    CLIENT_ID = "your-client-id"
    PRIVATE_KEY = """-----BEGIN PRIVATE KEY-----
    YOUR_PEM_PRIVATE_KEY_HERE
    -----END PRIVATE KEY-----"""
    ENVIRONMENT = "mypurecloud.com"
    DIVISION_ID = "your-division-id"

    # Sample survey payload
    SAMPLE_SURVEY = {
        "interactionId": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
        "csat_score": 5,
        "nps_score": 10,
        "issue_type": "billing"
    }

    # Initialize
    auth = GenesysAuth(ORG_ID, CLIENT_ID, PRIVATE_KEY, ENVIRONMENT)
    client = PureCloudPlatformClientV2.create(auth.get_access_token, ORG_ID, ENVIRONMENT)

    # Load codes
    code_map = load_wrapup_codes(client, DIVISION_ID)

    # Process
    process_survey(SAMPLE_SURVEY, code_map, client)

Common Errors & Debugging

Error: 401 Unauthorized

• Cause: The JWT signature is invalid, the client ID does not match the registered OAuth client, or the access token has expired.
• Fix: Verify the private key matches the public key uploaded to the Genesys Cloud OAuth client configuration. Ensure the aud claim in the JWT exactly matches the token endpoint URL. Check that the system clock is synchronized with NTP servers.
• Code Fix: The GenesysAuth class automatically refreshes tokens before expiration. If the error persists, log the JWT payload before signing to validate claims.

Error: 403 Forbidden

• Cause: The OAuth client lacks the interaction:write scope, or the client is restricted to a specific division that does not contain the target interaction.
• Fix: Navigate to the OAuth client settings and add interaction:write to the scope list. Verify the division ID passed to load_wrapup_codes matches the interaction division.
• Code Fix: Inspect the token response. The scope field must contain interaction:write. If missing, update the client configuration and regenerate the token.

Error: 429 Too Many Requests

• Cause: The script exceeds the Genesys Cloud rate limit for the Interaction API (typically 100 requests per second per client).
• Fix: Implement exponential backoff and respect the Retry-After header. The assign_wrapup_code function includes a single retry loop. For production queues, implement a token bucket or sliding window rate limiter.
• Code Fix: The existing retry logic sleeps for the duration specified in Retry-After. Increase the sleep duration or add jitter if cascading failures occur.

Error: 400 Bad Request

• Cause: The interaction is not in a completed state, the wrap-up code UUID does not exist, or the wrapUpTime is invalid.
• Fix: Verify the interaction status via GET /api/v2/interactions/{interactionId}. Ensure the wrap-up code was successfully loaded from the correct division. Format timestamps in ISO 8601 with timezone designators.
• Code Fix: Add a validation step before calling post_interactions_events to check interaction state. Log the exact wrapup_code_id being sent to confirm it matches the division lookup.

Official References

• Interaction Events API Documentation
• Python SDK Reference
• OAuth 2.0 JWT Grant Flow
• Wrap-Up Code Management
• Rate Limiting and Retry Guidelines