Implementing Dynamic Skill Assignment in Genesys Cloud Routing with Python SDK

What You Will Build

• This script polls the Genesys Cloud Interaction API for new conversations, extracts caller intent attributes, and matches them against a YAML-defined skill matrix.
• The client uses the Genesys Cloud Python SDK to attach routing skills via PUT /api/v2/routing/interactions/{interactionId} and validates interaction existence when the API returns 404.
• The implementation triggers supervisor email notifications via the Communications API when high-priority skills are applied, with full OAuth token management, pagination handling, and exponential backoff for rate limits.
• Language: Python 3.9+

Prerequisites

• OAuth Client Type: Confidential Client (Client Credentials Flow)
• Required OAuth Scopes: interaction:read, routing:interaction:write, communications:write, user:read
• SDK: genesys-cloud>=2.0.0
• Runtime: Python 3.9 or higher
• External Dependencies: requests>=2.31.0, pyyaml>=6.0, httpx>=0.25.0
• Genesys Cloud Environment: A valid organization with routing skills, interactions, and at least one supervisor user with email notification enabled

Authentication Setup

Genesys Cloud uses OAuth 2.0 Client Credentials for server-to-server integrations. The following code implements token retrieval with in-memory caching and automatic refresh when the token expires. The SDK does not manage tokens natively, so you must pass a valid bearer token to the Configuration object.

import time
import requests
from typing import Optional

class TokenManager:
    def __init__(self, client_id: str, client_secret: str, org_host: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.token_url = f"https://{org_host}/oauth/token"
        self._token: Optional[str] = None
        self._expires_at: float = 0.0

    def get_token(self) -> str:
        if time.time() < self._expires_at - 60:
            return self._token

        payload = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret,
            "scope": "interaction:read routing:interaction:write communications:write user:read"
        }

        response = requests.post(self.token_url, data=payload)
        response.raise_for_status()

        data = response.json()
        self._token = data["access_token"]
        self._expires_at = time.time() + data["expires_in"]
        return self._token

The TokenManager class caches the access token and subtracts a 60-second safety margin before the actual expiry. Every SDK client initialization will call get_token() to ensure the bearer token remains valid across long-running polling cycles.

Implementation

Step 1: Initialize SDK and Load Skill Matrix Configuration

The routing logic depends on a YAML configuration that maps intent keywords to skill IDs, priority levels, and supervisor contact information. The SDK requires a Configuration object bound to an ApiClient.

import yaml
import logging
from genesys_cloud.rest import Configuration, ApiClient
from genesys_cloud.apis.interaction import InteractionApi
from genesys_cloud.apis.routing import RoutingApi
from genesys_cloud.apis.user_management import UserManagementApi
from genesys_cloud.models import RoutingInteraction, RoutingSkill

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")

def load_skill_matrix(config_path: str) -> list[dict]:
    with open(config_path, "r") as f:
        config = yaml.safe_load(f)
    return config.get("skill_matrix", [])

def init_apis(org_host: str, token_manager: TokenManager):
    config = Configuration(
        host=f"https://{org_host}",
        access_token=token_manager.get_token()
    )
    api_client = ApiClient(config)
    return (
        InteractionApi(api_client),
        RoutingApi(api_client),
        UserManagementApi(api_client)
    )

The load_skill_matrix function parses the YAML file. The init_apis function constructs the three required SDK clients: InteractionApi for polling conversations, RoutingApi for skill assignment, and UserManagementApi for resolving supervisor IDs. All clients share the same underlying ApiClient instance, which reuses the TCP connection pool.

Step 2: Poll Interaction API and Extract Attributes

The Interaction API supports pagination via pageSize and pageNumber. You must request expand=attributes to retrieve custom caller data. The following function fetches interactions within a specific time window and extracts the intent attribute.

from datetime import datetime, timezone, timedelta

def fetch_active_interactions(interaction_api: InteractionApi, since: datetime, page_size: int = 100) -> list:
    date_from = since.isoformat()
    date_to = datetime.now(timezone.utc).isoformat()
    all_interactions = []
    page_number = 1
    has_more = True

    while has_more:
        try:
            result = interaction_api.get_interactions(
                expand="attributes",
                date_from=date_from,
                date_to=date_to,
                page_size=page_size,
                page_number=page_number
            )
        except Exception as e:
            logging.error("Failed to fetch interactions: %s", e)
            break

        if not result.entities:
            break

        all_interactions.extend(result.entities)
        if page_number >= result.page_count:
            has_more = False
        else:
            page_number += 1

    return all_interactions

The loop continues until page_number reaches result.page_count. Each interaction entity contains an attributes dictionary. You will read the intent from interaction.attributes.get("intent", "") in the next step.

Step 3: Match Intent, Validate IDs, and Handle 404 Responses

The routing assignment must occur only when the interaction exists and matches a known intent. The API returns 404 if the interaction was deleted, transferred, or already closed. You must validate the ID before attempting the skill update.

def validate_interaction_id(interaction_api: InteractionApi, interaction_id: str) -> bool:
    try:
        interaction_api.get_interaction(interaction_id=interaction_id)
        return True
    except Exception as e:
        status_code = getattr(e, "status", None)
        if status_code == 404:
            logging.warning("Interaction %s not found. Skipping assignment.", interaction_id)
            return False
        raise

This function calls GET /api/v2/interactions/{interactionId}. If the response is 404, the function logs the event and returns False. Any other exception propagates to the caller. You will call this validation before constructing the PUT payload.

Step 4: Apply Routing Skills via PUT and Implement Retry Logic

The PUT /api/v2/routing/interactions/{interactionId} endpoint replaces the entire routing configuration. You must fetch the current routing state, append the new skill, and submit the complete object. The following function implements exponential backoff for 429 rate limit responses.

import time as time_module

def apply_routing_skill(
    routing_api: RoutingApi,
    interaction_id: str,
    skill_id: str,
    priority: str,
    max_retries: int = 5
) -> bool:
    base_delay = 1.0

    for attempt in range(max_retries):
        try:
            current = routing_api.get_routing_interaction(interaction_id=interaction_id)
            new_skill = RoutingSkill(skill_id=skill_id, priority=priority)

            if not current.routing_skills:
                current.routing_skills = []

            existing_skill_ids = {s.skill_id for s in current.routing_skills}
            if skill_id not in existing_skill_ids:
                current.routing_skills.append(new_skill)
                routing_api.update_routing_interaction(
                    interaction_id=interaction_id,
                    routing_interaction=current
                )
                logging.info("Applied skill %s to interaction %s.", skill_id, interaction_id)
                return True
            else:
                logging.info("Skill %s already assigned to interaction %s.", skill_id, interaction_id)
                return True

        except Exception as e:
            status_code = getattr(e, "status", None)
            if status_code == 429:
                delay = base_delay * (2 ** attempt)
                logging.warning("Rate limited (429). Retrying in %.2f seconds.", delay)
                time_module.sleep(delay)
                continue
            elif status_code == 404:
                logging.error("Routing interaction %s not found during update.", interaction_id)
                return False
            else:
                logging.error("Failed to apply skill: %s", e)
                return False

    logging.error("Max retries exceeded for interaction %s.", interaction_id)
    return False

The retry loop doubles the delay on each 429 response. The function fetches the existing RoutingInteraction, merges the new RoutingSkill, and submits via update_routing_interaction. Duplicate skill IDs are skipped to prevent payload conflicts.

Step 5: Trigger Supervisor Notifications for High Priority Matches

When the YAML configuration marks a skill as high priority, the system must notify the designated supervisor. The Communications API accepts email targets via POST /api/v2/communications/messages. You must resolve the supervisor email from the configuration and construct a valid message payload.

import httpx

def notify_supervisor(token_manager: TokenManager, org_host: str, supervisor_email: str, interaction_id: str, skill_id: str) -> None:
    token = token_manager.get_token()
    url = f"https://{org_host}/api/v2/communications/messages"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    payload = {
        "targets": [{"type": "email", "address": supervisor_email}],
        "body": {
            "contentType": "text/plain",
            "content": f"High priority skill {skill_id} applied to interaction {interaction_id}. Immediate review required."
        },
        "subject": f"Genesys Alert: High Priority Skill Assignment ({interaction_id})"
    }

    with httpx.Client() as client:
        try:
            response = client.post(url, headers=headers, json=payload)
            response.raise_for_status()
            logging.info("Supervisor notification sent for interaction %s.", interaction_id)
        except httpx.HTTPStatusError as e:
            logging.error("Notification failed with status %s: %s", e.response.status_code, e.response.text)
        except Exception as e:
            logging.error("Notification request failed: %s", e)

The function uses httpx for explicit HTTP cycle visibility. The payload targets an email address and includes a plain-text body with the interaction ID and skill ID. The request reuses the same OAuth token manager to avoid scope mismatches.

Complete Working Example

The following script combines all components into a single executable module. Save it as dynamic_skill_assigner.py and run it with python dynamic_skill_assigner.py.

import time
import yaml
import logging
import httpx
from datetime import datetime, timezone, timedelta
from typing import Optional

from genesys_cloud.rest import Configuration, ApiClient
from genesys_cloud.apis.interaction import InteractionApi
from genesys_cloud.apis.routing import RoutingApi
from genesys_cloud.models import RoutingSkill

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")

class TokenManager:
    def __init__(self, client_id: str, client_secret: str, org_host: str):
        self.client_id = client_id
        self.client_secret = client_secret
        self.token_url = f"https://{org_host}/oauth/token"
        self._token: Optional[str] = None
        self._expires_at: float = 0.0

    def get_token(self) -> str:
        if time.time() < self._expires_at - 60:
            return self._token
        payload = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret,
            "scope": "interaction:read routing:interaction:write communications:write user:read"
        }
        import requests
        response = requests.post(self.token_url, data=payload)
        response.raise_for_status()
        data = response.json()
        self._token = data["access_token"]
        self._expires_at = time.time() + data["expires_in"]
        return self._token

def load_skill_matrix(config_path: str) -> list[dict]:
    with open(config_path, "r") as f:
        config = yaml.safe_load(f)
    return config.get("skill_matrix", [])

def fetch_active_interactions(interaction_api: InteractionApi, since: datetime, page_size: int = 100) -> list:
    date_from = since.isoformat()
    date_to = datetime.now(timezone.utc).isoformat()
    all_interactions = []
    page_number = 1
    has_more = True
    while has_more:
        try:
            result = interaction_api.get_interactions(
                expand="attributes",
                date_from=date_from,
                date_to=date_to,
                page_size=page_size,
                page_number=page_number
            )
        except Exception as e:
            logging.error("Failed to fetch interactions: %s", e)
            break
        if not result.entities:
            break
        all_interactions.extend(result.entities)
        if page_number >= result.page_count:
            has_more = False
        else:
            page_number += 1
    return all_interactions

def validate_interaction_id(interaction_api: InteractionApi, interaction_id: str) -> bool:
    try:
        interaction_api.get_interaction(interaction_id=interaction_id)
        return True
    except Exception as e:
        status_code = getattr(e, "status", None)
        if status_code == 404:
            logging.warning("Interaction %s not found. Skipping assignment.", interaction_id)
            return False
        raise

def apply_routing_skill(routing_api: RoutingApi, interaction_id: str, skill_id: str, priority: str, max_retries: int = 5) -> bool:
    base_delay = 1.0
    for attempt in range(max_retries):
        try:
            current = routing_api.get_routing_interaction(interaction_id=interaction_id)
            new_skill = RoutingSkill(skill_id=skill_id, priority=priority)
            if not current.routing_skills:
                current.routing_skills = []
            existing_skill_ids = {s.skill_id for s in current.routing_skills}
            if skill_id not in existing_skill_ids:
                current.routing_skills.append(new_skill)
                routing_api.update_routing_interaction(
                    interaction_id=interaction_id,
                    routing_interaction=current
                )
                logging.info("Applied skill %s to interaction %s.", skill_id, interaction_id)
                return True
            else:
                logging.info("Skill %s already assigned to interaction %s.", skill_id, interaction_id)
                return True
        except Exception as e:
            status_code = getattr(e, "status", None)
            if status_code == 429:
                delay = base_delay * (2 ** attempt)
                logging.warning("Rate limited (429). Retrying in %.2f seconds.", delay)
                time.sleep(delay)
                continue
            elif status_code == 404:
                logging.error("Routing interaction %s not found during update.", interaction_id)
                return False
            else:
                logging.error("Failed to apply skill: %s", e)
                return False
    logging.error("Max retries exceeded for interaction %s.", interaction_id)
    return False

def notify_supervisor(token_manager: TokenManager, org_host: str, supervisor_email: str, interaction_id: str, skill_id: str) -> None:
    token = token_manager.get_token()
    url = f"https://{org_host}/api/v2/communications/messages"
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    payload = {
        "targets": [{"type": "email", "address": supervisor_email}],
        "body": {
            "contentType": "text/plain",
            "content": f"High priority skill {skill_id} applied to interaction {interaction_id}. Immediate review required."
        },
        "subject": f"Genesys Alert: High Priority Skill Assignment ({interaction_id})"
    }
    with httpx.Client() as client:
        try:
            response = client.post(url, headers=headers, json=payload)
            response.raise_for_status()
            logging.info("Supervisor notification sent for interaction %s.", interaction_id)
        except httpx.HTTPStatusError as e:
            logging.error("Notification failed with status %s: %s", e.response.status_code, e.response.text)
        except Exception as e:
            logging.error("Notification request failed: %s", e)

def main():
    ORG_HOST = "mycompany.genesiscloud.com"
    CLIENT_ID = "your_client_id"
    CLIENT_SECRET = "your_client_secret"
    CONFIG_PATH = "skill_matrix.yaml"
    POLL_INTERVAL_SECONDS = 30

    token_manager = TokenManager(CLIENT_ID, CLIENT_SECRET, ORG_HOST)
    config = Configuration(host=f"https://{ORG_HOST}", access_token=token_manager.get_token())
    api_client = ApiClient(config)
    interaction_api = InteractionApi(api_client)
    routing_api = RoutingApi(api_client)
    skill_matrix = load_skill_matrix(CONFIG_PATH)

    last_processed = datetime.now(timezone.utc) - timedelta(minutes=5)

    logging.info("Starting dynamic skill assignment poller.")
    while True:
        interactions = fetch_active_interactions(interaction_api, last_processed)
        for interaction in interactions:
            intent = interaction.attributes.get("intent", "") if interaction.attributes else ""
            matched_rule = next((r for r in skill_matrix if r.get("intent_keyword") in intent.lower()), None)
            if not matched_rule:
                continue

            interaction_id = interaction.id
            if not validate_interaction_id(interaction_api, interaction_id):
                continue

            success = apply_routing_skill(
                routing_api,
                interaction_id,
                matched_rule["skill_id"],
                matched_rule["priority"]
            )
            if success and matched_rule["priority"] == "high":
                notify_supervisor(token_manager, ORG_HOST, matched_rule["supervisor_email"], interaction_id, matched_rule["skill_id"])

        last_processed = datetime.now(timezone.utc)
        time.sleep(POLL_INTERVAL_SECONDS)

if __name__ == "__main__":
    main()

The main function establishes the polling loop. It fetches interactions, matches intent against the YAML rules, validates the ID, applies the skill, and triggers notifications when the priority matches “high”. The loop sleeps for 30 seconds between cycles to respect API rate limits.

Common Errors & Debugging

Error: 401 Unauthorized

• Cause: The OAuth token expired or the client credentials are incorrect.
• Fix: Verify CLIENT_ID and CLIENT_SECRET in the configuration. Ensure the token manager refreshes the token before each API call. Check the OAuth client scope list in the Genesys Cloud admin console.
• Code: The TokenManager class already implements automatic refresh with a 60-second safety margin. If 401 persists, add logging.debug("Token refresh triggered") inside get_token() to confirm execution.

Error: 403 Forbidden

• Cause: The OAuth client lacks the required scope, or the organization enforces role-based access control that blocks programmatic writes.
• Fix: Grant the client credentials the routing:interaction:write and communications:write scopes. Assign the OAuth client to a system role with “Routing Administration” and “Messaging” permissions.
• Code: The scope string in TokenManager explicitly requests all required permissions. Verify the exact scope names match your tenant configuration.

Error: 404 Not Found on Routing Update

• Cause: The interaction was deleted, merged, or routed to a different queue before the PUT request executed.
• Fix: The validate_interaction_id function catches this condition early. If the error occurs during update_routing_interaction, the retry loop catches the 404 status code and returns False without crashing the poller.
• Code: The apply_routing_skill function checks status_code == 404 and logs the event. You can add a dead-letter queue or database record for auditing skipped interactions.

Error: 429 Too Many Requests

• Cause: The polling interval is too aggressive, or multiple instances of the script are running concurrently.
• Fix: Increase POLL_INTERVAL_SECONDS to 60 or higher. The retry loop implements exponential backoff with a base delay of 1 second. The delay doubles on each subsequent 429 response.
• Code: The apply_routing_skill function sleeps for base_delay * (2 ** attempt) seconds. Monitor the Retry-After header in the 429 response body if Genesys Cloud returns a specific wait time.

Official References

• Genesys Cloud Interactions API Documentation
• Genesys Cloud Routing Interactions API Documentation
• Genesys Cloud Communications Messages API Documentation
• Genesys Cloud Python SDK Reference
• Genesys Cloud OAuth 2.0 Client Credentials Guide