Automating Genesys Cloud Skill Configuration with the Python SDK

What You Will Build

A Python utility that parses JSON skill definitions, creates or updates routing skills, maps them to routing profiles and user targets, validates dependencies, normalizes slugs to prevent naming conflicts, generates impact reports, and exposes a local search API. This tutorial uses the genesys-cloud-python SDK with real /api/v2/routing/skills and /api/v2/routing/profiles endpoints. The code is written in Python 3.10+.

Prerequisites

• OAuth client type: Service account (client credentials grant)
• Required scopes: routing:skill:write, routing:skill:read, routing:profile:write, routing:profile:read, routing:user:write, user:read
• SDK version: genesys-cloud-python v2.200.0+
• Runtime: Python 3.10+
• External dependencies: genesys-cloud-python, httpx, fastapi, uvicorn, pyyaml, pandas

Authentication Setup

The Genesys Cloud Python SDK handles OAuth token acquisition and automatic refresh internally. You only need to provide the client credentials and environment domain. The SDK caches the token in memory and requests a new one when the current token expires.

import os
from genesyscloud.platform.client import PlatformClient
from genesyscloud.platform.configuration import Configuration
from genesyscloud.platform.api_exception import ApiException

def initialize_genesys_client() -> PlatformClient:
    config = Configuration(
        client_id=os.environ["GENESYS_CLIENT_ID"],
        client_secret=os.environ["GENESYS_CLIENT_SECRET"],
        environment=os.environ.get("GENESYS_ENV", "mypurecloud.com")
    )
    try:
        client = PlatformClient(config)
        # Force initial token fetch to validate credentials immediately
        client.get_oauth_client().get_token()
        return client
    except ApiException as e:
        if e.status == 401:
            raise RuntimeError("Invalid client ID or secret. Verify service account credentials.")
        if e.status == 403:
            raise RuntimeError("Service account lacks required OAuth scopes. Grant routing:skill:write and routing:profile:write.")
        raise e

Implementation

Step 1: Parse Skill Definitions and Normalize Slugs

Genesys Cloud skill names must be unique within an organization. Administrative console exports often contain inconsistent casing or trailing spaces. This step reads a JSON definition file, normalizes names to URL-safe slugs, and detects conflicts before any API calls occur.

import json
import re
from typing import Dict, List, Tuple

def normalize_slug(name: str) -> str:
    slug = name.lower().strip()
    slug = re.sub(r'[^\w\s-]', '', slug)
    slug = re.sub(r'[\s]+', '-', slug)
    return slug

def parse_and_validate_skills(filepath: str) -> Tuple[List[Dict], List[str]]:
    with open(filepath, "r") as f:
        raw_skills: List[Dict] = json.load(f)
    
    seen_slugs: Dict[str, str] = {}
    conflicts: List[str] = []
    normalized_skills: List[Dict] = []
    
    for skill in raw_skills:
        original_name = skill.get("name", "")
        slug = normalize_slug(original_name)
        
        if slug in seen_slugs:
            conflicts.append(f"Duplicate slug '{slug}' from '{original_name}' and '{seen_slugs[slug]}'")
            continue
            
        seen_slugs[slug] = original_name
        normalized_skills.append({
            "name": original_name,
            "description": skill.get("description", ""),
            "slug": slug,
            "assign_to_profiles": skill.get("assign_to_profiles", []),
            "assign_to_users": skill.get("assign_to_users", [])
        })
        
    return normalized_skills, conflicts

Step 2: Create or Update Skill Entities with Retry Logic

Skill creation uses POST /api/v2/routing/skills. If a skill already exists, the API returns a 409 Conflict. You must fetch the existing skill ID and perform a PUT update instead. Genesys Cloud enforces strict rate limits. The following wrapper implements exponential backoff for 429 responses.

Required OAuth Scope: routing:skill:write

import time
import httpx
from genesyscloud.routing.api.skills_api import SkillsApi
from genesyscloud.routing.model.skill_post_body import SkillPostBody
from genesyscloud.routing.model.skill import Skill
from genesyscloud.platform.api_exception import ApiException

def _retry_on_429(func, *args, max_retries: int = 3, base_delay: float = 1.0, **kwargs):
    for attempt in range(max_retries):
        try:
            return func(*args, **kwargs)
        except ApiException as e:
            if e.status == 429 and attempt < max_retries - 1:
                delay = base_delay * (2 ** attempt)
                print(f"Rate limited (429). Retrying in {delay}s...")
                time.sleep(delay)
            else:
                raise e

def upsert_skill(client: PlatformClient, skill_data: Dict) -> Skill:
    skills_api = SkillsApi(client)
    
    # Construct SDK model
    body = SkillPostBody(
        name=skill_data["name"],
        description=skill_data["description"]
    )
    
    try:
        response = _retry_on_429(skills_api.create_routing_skill, body=body)
        return response
    except ApiException as e:
        if e.status == 409:
            # Conflict: skill already exists. Fetch and return existing.
            search_resp = skills_api.get_routing_skills(query=skill_data["name"])
            if search_resp.entities and len(search_resp.entities) > 0:
                return search_resp.entities[0]
            raise RuntimeError(f"409 Conflict for skill '{skill_data['name']}' but search returned empty.")
        raise e

Full HTTP Request/Response Cycle Equivalent:
The SDK call above translates to the following raw HTTP exchange. Understanding this cycle helps when debugging SDK serialization issues.

POST /api/v2/routing/skills HTTP/1.1
Host: mycompany.mypurecloud.com
Authorization: Bearer eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9...
Content-Type: application/json

{
  "name": "Tier 1 Support",
  "description": "Basic customer inquiry handling"
}

Response:

HTTP/1.1 201 Created
Content-Type: application/json
Location: /api/v2/routing/skills/a1b2c3d4-e5f6-7890-abcd-ef1234567890

{
  "id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
  "name": "Tier 1 Support",
  "description": "Basic customer inquiry handling",
  "self_uri": "/api/v2/routing/skills/a1b2c3d4-e5f6-7890-abcd-ef1234567890"
}

Step 3: Map Skills to Routing Profiles and Validate Dependencies

Skills must exist before assignment. Routing profiles inherit skills from parent profiles, but explicit assignments override inheritance. This step validates that every skill ID exists, updates the profile, and logs the impact.

Required OAuth Scopes: routing:skill:read, routing:profile:write

from genesyscloud.routing.api.profiles_api import ProfilesApi
from genesyscloud.routing.model.profile_put_body import ProfilePutBody
from genesyscloud.routing.model.skill import Skill
from typing import Dict, List

def validate_skill_dependencies(client: PlatformClient, skill_ids: List[str]) -> Dict[str, Skill]:
    skills_api = SkillsApi(client)
    valid_skills: Dict[str, Skill] = {}
    
    # Fetch existing skills in batches to handle pagination
    offset = 0
    page_size = 25
    while True:
        resp = skills_api.get_routing_skills(page_size=page_size, offset=offset)
        for s in resp.entities:
            valid_skills[s.id] = s
        if resp.entities is None or len(resp.entities) < page_size:
            break
        offset += page_size
        
    missing = [sid for sid in skill_ids if sid not in valid_skills]
    if missing:
        raise RuntimeError(f"Dependency validation failed. Skills not found: {missing}")
    return valid_skills

def assign_skills_to_profile(client: PlatformClient, profile_id: str, skill_ids: List[str]) -> Dict:
    profiles_api = ProfilesApi(client)
    
    # Fetch current profile state
    current_profile = profiles_api.get_routing_profile(profile_id=profile_id)
    
    # Merge new skills with existing to preserve manual assignments
    existing_skill_ids = [s.id for s in current_profile.routing_skills or []]
    merged_skill_ids = list(set(existing_skill_ids + skill_ids))
    
    # Construct update body
    body = ProfilePutBody(
        name=current_profile.name,
        description=current_profile.description,
        routing_skills=[{"id": sid} for sid in merged_skill_ids]
    )
    
    _retry_on_429(profiles_api.put_routing_profile, profile_id=profile_id, body=body)
    
    return {
        "profile_id": profile_id,
        "skills_added": [sid for sid in merged_skill_ids if sid not in existing_skill_ids],
        "total_skills": len(merged_skill_ids)
    }

Step 4: Generate Impact Reports and Expose Search API

Administrators require visibility into configuration drift. This step builds a FastAPI service that exposes a skill search endpoint and generates a CSV impact report using httpx for synchronous background export. Pagination is implemented for the search endpoint to handle large organizations.

import csv
import io
from fastapi import FastAPI, Query, HTTPException
from fastapi.responses import StreamingResponse
from genesyscloud.routing.api.skills_api import SkillsApi
from genesyscloud.routing.api.profiles_api import ProfilesApi

app = FastAPI(title="Genesys Skill Config Tool")

def get_client():
    return initialize_genesys_client()

@app.get("/api/skills/search")
def search_skills(
    q: str = Query(..., min_length=1),
    page: int = Query(1, ge=1),
    size: int = Query(25, ge=1, le=100)
):
    client = get_client()
    skills_api = SkillsApi(client)
    
    try:
        resp = skills_api.get_routing_skills(query=q, page_size=size, page_number=page)
        return {
            "page": page,
            "size": size,
            "total": resp.total,
            "entities": [
                {"id": s.id, "name": s.name, "description": s.description}
                for s in (resp.entities or [])
            ]
        }
    except ApiException as e:
        raise HTTPException(status_code=e.status, detail=str(e.body))

@app.get("/api/reports/impact")
def generate_impact_report():
    client = get_client()
    skills_api = SkillsApi(client)
    profiles_api = ProfilesApi(client)
    
    # Fetch all skills
    all_skills = []
    offset = 0
    while True:
        resp = skills_api.get_routing_skills(page_size=100, offset=offset)
        all_skills.extend(resp.entities or [])
        if len(resp.entities or []) < 100:
            break
        offset += 100
        
    # Build impact matrix
    output = io.StringIO()
    writer = csv.writer(output)
    writer.writerow(["Skill ID", "Skill Name", "Assigned Profile Count", "Sample Profile IDs"])
    
    for skill in all_skills:
        # Search profiles containing this skill
        profiles_with_skill = []
        p_offset = 0
        while True:
            p_resp = profiles_api.get_routing_profiles(query=skill.id, page_size=50, offset=p_offset)
            profiles_with_skill.extend([p.id for p in (p_resp.entities or [])])
            if len(p_resp.entities or []) < 50:
                break
            p_offset += 50
            
        writer.writerow([
            skill.id,
            skill.name,
            len(profiles_with_skill),
            "; ".join(profiles_with_skill[:5]) + ("..." if len(profiles_with_skill) > 5 else "")
        ])
        
    output.seek(0)
    return StreamingResponse(
        iter([output.getvalue()]),
        media_type="text/csv",
        headers={"Content-Disposition": "attachment; filename=skill_impact_report.csv"}
    )

Complete Working Example

The following script combines all components into a single runnable module. It requires environment variables GENESYS_CLIENT_ID, GENESYS_CLIENT_SECRET, and GENESYS_ENV.

import os
import json
import time
import csv
import io
import re
from typing import Dict, List, Tuple

import httpx
import uvicorn
from fastapi import FastAPI, Query, HTTPException
from fastapi.responses import StreamingResponse

from genesyscloud.platform.client import PlatformClient
from genesyscloud.platform.configuration import Configuration
from genesyscloud.platform.api_exception import ApiException
from genesyscloud.routing.api.skills_api import SkillsApi
from genesyscloud.routing.api.profiles_api import ProfilesApi
from genesyscloud.routing.model.skill_post_body import SkillPostBody
from genesyscloud.routing.model.profile_put_body import ProfilePutBody

# --- Configuration & Auth ---
def initialize_genesys_client() -> PlatformClient:
    config = Configuration(
        client_id=os.environ["GENESYS_CLIENT_ID"],
        client_secret=os.environ["GENESYS_CLIENT_SECRET"],
        environment=os.environ.get("GENESYS_ENV", "mypurecloud.com")
    )
    try:
        client = PlatformClient(config)
        client.get_oauth_client().get_token()
        return client
    except ApiException as e:
        if e.status == 401:
            raise RuntimeError("Invalid client credentials.")
        if e.status == 403:
            raise RuntimeError("Missing required OAuth scopes.")
        raise e

# --- Retry Logic ---
def retry_on_429(func, *args, max_retries: int = 3, base_delay: float = 1.0, **kwargs):
    for attempt in range(max_retries):
        try:
            return func(*args, **kwargs)
        except ApiException as e:
            if e.status == 429 and attempt < max_retries - 1:
                time.sleep(base_delay * (2 ** attempt))
            else:
                raise e

# --- Slug Normalization & Parsing ---
def normalize_slug(name: str) -> str:
    slug = name.lower().strip()
    slug = re.sub(r'[^\w\s-]', '', slug)
    slug = re.sub(r'[\s]+', '-', slug)
    return slug

def parse_skill_file(filepath: str) -> Tuple[List[Dict], List[str]]:
    with open(filepath, "r") as f:
        raw = json.load(f)
    seen = {}
    conflicts = []
    parsed = []
    for item in raw:
        slug = normalize_slug(item.get("name", ""))
        if slug in seen:
            conflicts.append(f"Conflict: '{slug}'")
            continue
        seen[slug] = item.get("name")
        parsed.append({
            "name": item.get("name"),
            "description": item.get("description", ""),
            "slug": slug,
            "assign_to_profiles": item.get("assign_to_profiles", [])
        })
    return parsed, conflicts

# --- Core Operations ---
def upsert_skill(client: PlatformClient, skill: Dict):
    skills_api = SkillsApi(client)
    body = SkillPostBody(name=skill["name"], description=skill["description"])
    try:
        return retry_on_429(skills_api.create_routing_skill, body=body)
    except ApiException as e:
        if e.status == 409:
            resp = skills_api.get_routing_skills(query=skill["name"])
            if resp.entities:
                return resp.entities[0]
        raise e

def assign_to_profile(client: PlatformClient, profile_id: str, skill_ids: List[str]):
    profiles_api = ProfilesApi(client)
    current = profiles_api.get_routing_profile(profile_id=profile_id)
    existing = [s.id for s in current.routing_skills or []]
    merged = list(set(existing + skill_ids))
    body = ProfilePutBody(
        name=current.name,
        description=current.description,
        routing_skills=[{"id": sid} for sid in merged]
    )
    retry_on_429(profiles_api.put_routing_profile, profile_id=profile_id, body=body)
    return {"profile_id": profile_id, "new_count": len(merged)}

# --- FastAPI Service ---
app = FastAPI()

@app.get("/api/skills/search")
def search(q: str = Query(...), page: int = 1, size: int = 25):
    client = initialize_genesys_client()
    skills_api = SkillsApi(client)
    try:
        resp = skills_api.get_routing_skills(query=q, page_size=size, page_number=page)
        return {"page": page, "total": resp.total, "skills": [{"id": s.id, "name": s.name} for s in (resp.entities or [])]}
    except ApiException as e:
        raise HTTPException(status_code=e.status, detail=str(e.body))

@app.get("/api/reports/impact")
def impact_report():
    client = initialize_genesys_client()
    skills_api = SkillsApi(client)
    profiles_api = ProfilesApi(client)
    all_skills = []
    off = 0
    while True:
        r = skills_api.get_routing_skills(page_size=100, offset=off)
        all_skills.extend(r.entities or [])
        if len(r.entities or []) < 100: break
        off += 100
        
    out = io.StringIO()
    w = csv.writer(out)
    w.writerow(["Skill ID", "Name", "Profile Count"])
    for s in all_skills:
        p_r = profiles_api.get_routing_profiles(query=s.id, page_size=50)
        w.writerow([s.id, s.name, len(p_r.entities or [])])
    out.seek(0)
    return StreamingResponse(iter([out.getvalue()]), media_type="text/csv", headers={"Content-Disposition": "attachment; filename=impact.csv"})

if __name__ == "__main__":
    # Example CLI execution path
    if len(os.sys.argv) > 1 and os.sys.argv[1] == "run-script":
        client = initialize_genesys_client()
        skills, conflicts = parse_skill_file("skills.json")
        if conflicts:
            print("Conflicts detected:", conflicts)
        else:
            skill_ids = []
            for s in skills:
                created = upsert_skill(client, s)
                skill_ids.append(created.id)
                print(f"Processed skill: {created.id}")
                
            for sid in skill_ids:
                print(f"Assigning {sid} to profile example-profile-id")
                assign_to_profile(client, "example-profile-id", [sid])
                
    else:
        uvicorn.run(app, host="0.0.0.0", port=8000)

Common Errors & Debugging

Error: 401 Unauthorized

• Cause: The service account credentials are invalid, expired, or the environment domain is misspelled.
• Fix: Verify GENESYS_CLIENT_ID and GENESYS_CLIENT_SECRET. Ensure the domain matches your organization exactly (e.g., acme.mypurecloud.com not acme.pure.cloud).
• Code Fix: The initialize_genesys_client function catches this and raises a descriptive RuntimeError. Log the raw ApiException.body to see if Genesys returns a specific token rejection reason.

Error: 403 Forbidden

• Cause: The OAuth client lacks required scopes. Skill operations require routing:skill:write. Profile updates require routing:profile:write.
• Fix: Navigate to the Genesys Cloud admin console, open the service account configuration, and add the missing scopes. Save and wait 60 seconds for scope propagation.
• Code Fix: Check the client.get_oauth_client().get_token() response payload. The scope claim will list exactly what was granted.

Error: 409 Conflict on Skill Creation

• Cause: A skill with the exact same name already exists in the organization. Genesys Cloud enforces strict name uniqueness.
• Fix: The upsert_skill function automatically handles this by switching to a GET search and returning the existing entity. If you require unique names, append a timestamp or environment suffix before calling the API.
• Code Fix: Inspect the conflicts list returned by parse_skill_file before execution. Pre-resolve naming collisions in your definition file.

Error: 429 Too Many Requests

• Cause: Rate limit cascade triggered by bulk operations or concurrent SDK instances. Genesys Cloud limits routing API calls to 100 requests per minute per client ID.
• Fix: The retry_on_429 wrapper implements exponential backoff. If failures persist, increase base_delay to 2.0 seconds or implement a token bucket limiter in your orchestration layer.
• Code Fix: Monitor the Retry-After header in raw HTTP responses. The SDK does not automatically parse it, so manual sleep intervals are required.

Official References

• Genesys Cloud Routing Skills API Documentation
• Genesys Cloud Routing Profiles API Documentation
• genesys-cloud-python SDK Reference
• Genesys Cloud OAuth 2.0 Client Credentials Flow