Implementing Cost Attribution Models for Multi-Department Cloud Telephony Usage

What This Guide Covers

You are building a cost attribution system that breaks down your organization’s Genesys Cloud telephony bill - which arrives as a single monthly invoice covering all regions, departments, and business units - into per-department, per-queue, and per-campaign cost allocations. When complete, each department head will receive a monthly chargeback report showing exactly how much telephony spend their teams consumed, broken down by inbound toll-free minutes, outbound DID usage, BYOC trunk charges, and per-minute metered costs. This replaces the current model where IT absorbs the entire invoice and has no visibility into which departments drive the most telephony spend.

Prerequisites, Roles & Licensing

• Genesys Cloud: Any CX tier.
• Permissions required:
  • Analytics > Conversation Detail > View
  • Billing > Subscription > View (for usage data)
  • Routing > Queue > View
• Data destination: A data warehouse (Snowflake, BigQuery, or Redshift) or a simple PostgreSQL database for cost aggregation.
• Billing data source: Genesys Cloud Usage API or exported CSV billing reports from the Admin → Billing panel.

The Implementation Deep-Dive

1. The Cost Attribution Architecture

[Genesys Cloud Billing API: Usage Data]
    |
    v
[ETL: Extract per-conversation media usage]
    |
    v
[Enrich: Map conversation → queue → department via org metadata]
    |
    v
[Rate Card Application: minutes × $/min per media type per region]
    |
    v
[Aggregation: Roll up costs by department, queue, campaign]
    |
    v
[Chargeback Report: Monthly cost-per-department breakdown]

2. Extracting Per-Conversation Usage Data

The Analytics API provides conversation-level detail including talk duration, hold duration, and media type - the raw inputs for cost attribution:

import requests
from datetime import datetime, timedelta
from collections import defaultdict

GENESYS_API = "https://api.mypurecloud.com"

def extract_conversation_usage(access_token: str, start_date: str, end_date: str) -> list[dict]:
    """
    Extracts per-conversation usage data for cost attribution.
    Returns: list of {conversation_id, queue, department, media_type, talk_minutes, direction}
    """
    headers = {
        "Authorization": f"Bearer {access_token}",
        "Content-Type": "application/json"
    }
    
    query = {
        "interval": f"{start_date}/{end_date}",
        "order": "asc",
        "orderBy": "conversationStart",
        "paging": {"pageSize": 100, "pageNumber": 1},
        "segmentFilters": [
            {
                "type": "and",
                "predicates": [
                    {"type": "dimension", "dimension": "mediaType", "value": "voice"}
                ]
            }
        ]
    }
    
    all_records = []
    page = 1
    
    while True:
        query["paging"]["pageNumber"] = page
        resp = requests.post(
            f"{GENESYS_API}/api/v2/analytics/conversations/details/query",
            headers=headers, json=query
        )
        data = resp.json()
        conversations = data.get("conversations", [])
        
        if not conversations:
            break
        
        for conv in conversations:
            conv_id = conv["conversationId"]
            
            for participant in conv.get("participants", []):
                if participant.get("purpose") != "agent":
                    continue
                
                for session in participant.get("sessions", []):
                    for segment in session.get("segments", []):
                        if segment.get("segmentType") == "interact":
                            talk_ms = segment.get("segmentEnd", 0) - segment.get("segmentStart", 0)
                            
                            queue_id = next(
                                (m.get("value") for m in session.get("metrics", []) 
                                 if m.get("name") == "oQueueId"), None
                            )
                            
                            all_records.append({
                                "conversation_id": conv_id,
                                "queue_id": queue_id,
                                "direction": session.get("direction", "unknown"),
                                "talk_minutes": round(talk_ms / 60000, 2),
                                "media_type": session.get("mediaType", "voice"),
                                "ani": session.get("ani", ""),
                                "dnis": session.get("dnis", ""),
                                "timestamp": conv.get("conversationStart")
                            })
        
        page += 1
        if page > data.get("totalHits", 0) / 100 + 1:
            break
    
    return all_records

3. Department Mapping via Queue Metadata

Map each queue to a department using queue metadata or a maintained mapping table:

# Queue → Department mapping (maintained in config or fetched from Genesys)
QUEUE_DEPARTMENT_MAP = {}  # Populated at runtime

def build_queue_department_map(access_token: str) -> dict:
    """
    Builds a queue_id → department mapping using queue descriptions or custom attributes.
    Convention: Queue description format is 'DepartmentName | Queue Purpose'
    """
    headers = {"Authorization": f"Bearer {access_token}"}
    
    resp = requests.get(
        f"{GENESYS_API}/api/v2/routing/queues",
        headers=headers,
        params={"pageSize": 500}
    )
    
    mapping = {}
    for queue in resp.json().get("entities", []):
        queue_id = queue["id"]
        description = queue.get("description", "")
        
        # Parse department from description (e.g., "Sales | Inbound Leads")
        if "|" in description:
            department = description.split("|")[0].strip()
        else:
            department = "Unassigned"
        
        mapping[queue_id] = {
            "department": department,
            "queue_name": queue.get("name", "Unknown"),
        }
    
    return mapping

def enrich_with_department(records: list[dict], queue_map: dict) -> list[dict]:
    """Adds department information to each conversation record."""
    for record in records:
        q_info = queue_map.get(record.get("queue_id"), {})
        record["department"] = q_info.get("department", "Unassigned")
        record["queue_name"] = q_info.get("queue_name", "Unknown")
    return records

4. Rate Card Application

Apply your carrier’s rate card to convert minutes into dollars:

RATE_CARD = {
    # Per-minute rates by direction and number type
    "inbound": {
        "toll_free": 0.032,    # $/min for 800-number inbound
        "local_did": 0.008,    # $/min for local DID inbound
        "default": 0.015       # $/min fallback
    },
    "outbound": {
        "domestic": 0.012,     # $/min for domestic outbound
        "international": 0.085, # $/min for international outbound
        "default": 0.020
    },
    "platform_fee_per_minute": 0.004  # Genesys Cloud metered voice fee
}

def classify_number_type(number: str) -> str:
    """Classifies a phone number for rate card lookup."""
    if not number:
        return "default"
    if number.startswith("+1800") or number.startswith("+1888") or number.startswith("+1877"):
        return "toll_free"
    if number.startswith("+1"):
        return "domestic"
    if number.startswith("+"):
        return "international"
    return "local_did"

def calculate_costs(records: list[dict]) -> list[dict]:
    """Applies rate card to each conversation record."""
    for record in records:
        direction = record.get("direction", "inbound")
        minutes = record.get("talk_minutes", 0)
        
        if direction == "inbound":
            number_type = classify_number_type(record.get("dnis", ""))
            rate = RATE_CARD["inbound"].get(number_type, RATE_CARD["inbound"]["default"])
        else:
            number_type = classify_number_type(record.get("ani", ""))
            rate = RATE_CARD["outbound"].get(number_type, RATE_CARD["outbound"]["default"])
        
        carrier_cost = round(minutes * rate, 4)
        platform_cost = round(minutes * RATE_CARD["platform_fee_per_minute"], 4)
        
        record["carrier_cost"] = carrier_cost
        record["platform_cost"] = platform_cost
        record["total_cost"] = round(carrier_cost + platform_cost, 4)
        record["rate_applied"] = rate
        record["number_type"] = number_type
    
    return records

5. Generating the Chargeback Report

def generate_chargeback_report(records: list[dict]) -> dict:
    """
    Aggregates costs by department and generates a chargeback summary.
    """
    dept_costs = defaultdict(lambda: {
        "total_cost": 0, "carrier_cost": 0, "platform_cost": 0,
        "total_minutes": 0, "call_count": 0,
        "inbound_minutes": 0, "outbound_minutes": 0
    })
    
    for r in records:
        dept = r.get("department", "Unassigned")
        dept_costs[dept]["total_cost"] += r["total_cost"]
        dept_costs[dept]["carrier_cost"] += r["carrier_cost"]
        dept_costs[dept]["platform_cost"] += r["platform_cost"]
        dept_costs[dept]["total_minutes"] += r["talk_minutes"]
        dept_costs[dept]["call_count"] += 1
        
        if r.get("direction") == "inbound":
            dept_costs[dept]["inbound_minutes"] += r["talk_minutes"]
        else:
            dept_costs[dept]["outbound_minutes"] += r["talk_minutes"]
    
    # Sort by total cost descending
    sorted_depts = sorted(dept_costs.items(), key=lambda x: x[1]["total_cost"], reverse=True)
    
    grand_total = sum(d["total_cost"] for d in dept_costs.values())
    
    report = {
        "period": "2026-04",
        "grand_total": round(grand_total, 2),
        "departments": []
    }
    
    for dept_name, costs in sorted_depts:
        report["departments"].append({
            "department": dept_name,
            "total_cost": round(costs["total_cost"], 2),
            "share_pct": round(costs["total_cost"] / grand_total * 100, 1) if grand_total > 0 else 0,
            "call_count": costs["call_count"],
            "total_minutes": round(costs["total_minutes"], 1),
            "avg_cost_per_call": round(costs["total_cost"] / costs["call_count"], 3) if costs["call_count"] > 0 else 0
        })
    
    return report

Example output:

{
  "period": "2026-04",
  "grand_total": 14523.87,
  "departments": [
    {"department": "Sales", "total_cost": 6241.30, "share_pct": 43.0, "call_count": 18420, "total_minutes": 52100.5},
    {"department": "Technical Support", "total_cost": 4891.22, "share_pct": 33.7, "call_count": 12050, "total_minutes": 41230.0},
    {"department": "Billing", "total_cost": 2103.45, "share_pct": 14.5, "call_count": 8200, "total_minutes": 16400.0},
    {"department": "Retention", "total_cost": 1287.90, "share_pct": 8.9, "call_count": 3100, "total_minutes": 9800.0}
  ]
}

Validation, Edge Cases & Troubleshooting

Edge Case 1: Transferred Calls Attribute Cost to the Wrong Department

A call arrives in the Sales queue (2 minutes) then transfers to Technical Support (15 minutes). The entire 17-minute cost is attributed to Sales because the first queue was used for classification.
Solution: Split cost attribution at the segment level. Each queue segment’s duration is costed independently. The 2-minute Sales segment uses the Sales rate, and the 15-minute Tech Support segment uses the Tech Support rate. This requires parsing segment-level data rather than conversation-level aggregates.

Edge Case 2: BYOC Trunk Costs Don’t Appear in Genesys Usage API

Your organization uses BYOC Premise with a third-party carrier. The Genesys Cloud Usage API shows $0 for telephony because Genesys isn’t the carrier - the carrier bills you directly.
Solution: Import your carrier’s CDR (Call Detail Records) into the pipeline as a secondary data source. Join carrier CDRs to Genesys conversations using ANI + DNIS + timestamp matching (±5 second tolerance). Apply the carrier’s rate card to carrier CDRs and Genesys’s platform fee separately.

Edge Case 3: Shared Queues Serve Multiple Departments

Your “General Inquiry” queue handles calls for Sales, Support, and Billing. The queue-to-department mapping doesn’t work because one queue maps to three departments.
Solution: Use wrap-up codes as the department attribution signal instead of queue. Require agents to select a department-indicating wrap-up code at the end of each interaction. The cost pipeline reads the wrap-up code rather than the queue for shared-queue scenarios.

Official References

• Genesys Cloud Usage API
• Genesys Cloud Analytics Conversations API
• Genesys Cloud Billing Overview
• Genesys Cloud Queue Administration