Generate Post-Call Summaries in Genesys Cloud Agent Assist with Python FastAPI

What You Will Build

• When running, this service receives a Genesys interaction identifier, retrieves the full conversation transcript, extracts a structured summary and action items using an LLM, and writes those artifacts to a CRM case record.
• This implementation uses the Genesys Cloud Interaction Analytics API, CRM Integration API, and the OpenAI API.
• The code uses Python 3.10+ with FastAPI and the httpx library for all HTTP operations.

Prerequisites

• OAuth client type and required scopes: Machine-to-Machine (Client Credentials) grant. Required scopes: interaction:read, analytics:conversation:view, integration:read, integration:write.
• SDK version or API version: Genesys Cloud REST API v2.
• Language/runtime requirements: Python 3.10 or higher.
• External dependencies: fastapi, uvicorn, httpx, pydantic, openai, tenacity.

Authentication Setup

Genesys Cloud uses OAuth 2.0 client credentials flow for service-to-service authentication. The token expires after twenty minutes and must be refreshed before expiration or upon receiving a 401 response. The following code establishes a secure token cache with automatic refresh logic.

import os
import httpx
from typing import Optional
from datetime import datetime, timedelta, timezone

class GenesysAuthManager:
    def __init__(self, client_id: str, client_secret: str, environment: str = "mypurecloud.com"):
        self.client_id = client_id
        self.client_secret = client_secret
        self.environment = environment
        self.token: Optional[str] = None
        self.expires_at: Optional[datetime] = None
        self.http_client = httpx.Client(timeout=30.0)

    def _fetch_token(self) -> str:
        url = f"https://{self.environment}/oauth/token"
        payload = {
            "grant_type": "client_credentials",
            "client_id": self.client_id,
            "client_secret": self.client_secret,
            "scope": "interaction:read analytics:conversation:view integration:read integration:write"
        }
        response = self.http_client.post(url, data=payload)
        response.raise_for_status()
        data = response.json()
        self.token = data["access_token"]
        self.expires_at = datetime.now(timezone.utc) + timedelta(seconds=data["expires_in"])
        return self.token

    def get_access_token(self) -> str:
        if self.token is None or self.expires_at is None or datetime.now(timezone.utc) >= self.expires_at:
            return self._fetch_token()
        return self.token

    def get_headers(self) -> dict:
        return {
            "Authorization": f"Bearer {self.get_access_token()}",
            "Content-Type": "application/json"
        }

Implementation

Step 1: Fetch Interaction Transcript via Analytics API

The /api/v2/analytics/conversations/details/query endpoint returns conversation metadata and transcript segments. It supports pagination through the nextPageId field. The following function queries interactions, handles pagination, extracts the transcript array, and implements retry logic for 429 rate limits.

OAuth scope required: analytics:conversation:view

import httpx
import json
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10),
    retry=retry_if_exception_type(httpx.HTTPStatusError)
)
def fetch_transcript(auth: GenesysAuthManager, interaction_id: str) -> list[dict]:
    base_url = f"https://{auth.environment}/api/v2/analytics/conversations/details/query"
    query_payload = {
        "view": "conversation",
        "dateFrom": "2020-01-01T00:00:00Z",
        "dateTo": "2030-01-01T00:00:00Z",
        "pageSize": 25,
        "filter": [
            {"type": "equals", "dimension": "id", "value": interaction_id}
        ]
    }

    transcript_segments = []
    current_url = base_url

    while current_url:
        response = httpx.post(current_url, json=query_payload, headers=auth.get_headers(), timeout=30.0)
        
        if response.status_code == 401:
            auth._fetch_token()
            response = httpx.post(current_url, json=query_payload, headers=auth.get_headers(), timeout=30.0)
        
        if response.status_code == 429:
            retry_after = int(response.headers.get("Retry-After", 5))
            raise httpx.HTTPStatusError(f"Rate limited. Retry after {retry_after}s", request=response.request, response=response)
        
        response.raise_for_status()
        data = response.json()

        if "entities" in data and len(data["entities"]) > 0:
            conversation = data["entities"][0]
            transcript_segments.extend(conversation.get("transcript", []))

        current_url = data.get("nextPageId")
        if current_url:
            current_url = f"https://{auth.environment}/{current_url}"

    if not transcript_segments:
        raise ValueError(f"No transcript found for interaction {interaction_id}")
    
    return transcript_segments

Step 2: Prompt LLM and Extract Action Items

The transcript contains an array of message objects with author, text, and timestamp. You must flatten this into a structured prompt. The following function formats the transcript, sends it to the LLM with a strict JSON schema requirement, and parses the response into a Pydantic model.

from pydantic import BaseModel
from openai import OpenAI
import json

class PostCallSummary(BaseModel):
    executive_summary: str
    action_items: list[str]
    customer_sentiment: str
    confidence_score: float

def generate_llm_summary(transcript: list[dict], openai_client: OpenAI) -> PostCallSummary:
    formatted_transcript = "\n".join([
        f"{seg.get('author', 'Unknown')}: {seg.get('text', '')}" 
        for seg in transcript if seg.get("text")
    ])

    system_prompt = """You are a specialized call center analyst. Extract a post-call summary, action items, and sentiment. 
    Return strictly valid JSON matching the provided schema. Do not include markdown formatting or explanatory text."""

    try:
        response = openai_client.chat.completions.create(
            model="gpt-4o-mini",
            response_format={"type": "json_object"},
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": formatted_transcript}
            ],
            temperature=0.1,
            max_tokens=500
        )
        
        raw_json = response.choices[0].message.content
        if not raw_json:
            raise ValueError("LLM returned empty response")
            
        parsed_data = json.loads(raw_json)
        return PostCallSummary(**parsed_data)
    
    except json.JSONDecodeError as e:
        raise RuntimeError(f"LLM returned malformed JSON: {e}")
    except Exception as e:
        raise RuntimeError(f"LLM generation failed: {e}")

Step 3: Update CRM Case Record and Link Artifacts

Genesys CRM integration uses the /api/v2/integration/instances/{integrationId}/entities/{entityId}/records/{recordId} endpoint to upsert record fields. The entityId corresponds to the CRM object name (e.g., Case). You must map the LLM output to the exact field names configured in your Genesys CRM integration.

OAuth scope required: integration:write

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10),
    retry=retry_if_exception_type(httpx.HTTPStatusError)
)
def update_crm_case(auth: GenesysAuthManager, integration_id: str, case_id: str, summary: PostCallSummary) -> dict:
    url = f"https://{auth.environment}/api/v2/integration/instances/{integration_id}/entities/Case/records/{case_id}"
    
    payload = {
        "fields": {
            "Post_Call_Summary": summary.executive_summary,
            "Action_Items": "; ".join(summary.action_items),
            "Call_Sentiment": summary.customer_sentiment,
            "Summary_Confidence": str(summary.confidence_score)
        }
    }

    response = httpx.put(url, json=payload, headers=auth.get_headers(), timeout=30.0)
    
    if response.status_code == 401:
        auth._fetch_token()
        response = httpx.put(url, json=payload, headers=auth.get_headers(), timeout=30.0)
    
    if response.status_code == 429:
        raise httpx.HTTPStatusError("CRM API rate limited", request=response.request, response=response)
    
    response.raise_for_status()
    return response.json()

Complete Working Example

The following FastAPI application combines all components into a production-ready service. It exposes a single endpoint that accepts an interaction ID and case ID, processes the transcript, and returns the updated CRM record.

import os
import uvicorn
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from openai import OpenAI
import httpx

app = FastAPI(title="Genesys Post-Call Summary Service")

# Initialize dependencies
GENESYS_CLIENT_ID = os.getenv("GENESYS_CLIENT_ID")
GENESYS_CLIENT_SECRET = os.getenv("GENESYS_CLIENT_SECRET")
GENESYS_ENV = os.getenv("GENESYS_ENV", "mypurecloud.com")
INTEGRATION_ID = os.getenv("GENESYS_INTEGRATION_ID")
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")

auth_manager = GenesysAuthManager(GENESYS_CLIENT_ID, GENESYS_CLIENT_SECRET, GENESYS_ENV)
openai_client = OpenAI(api_key=OPENAI_API_KEY)

class SummaryRequest(BaseModel):
    interaction_id: str
    case_id: str

@app.post("/generate-summary")
async def generate_summary_endpoint(req: SummaryRequest) -> dict:
    try:
        # Step 1: Fetch transcript with pagination and retry logic
        transcript = fetch_transcript(auth_manager, req.interaction_id)
        
        # Step 2: Generate structured summary
        summary_data = generate_llm_summary(transcript, openai_client)
        
        # Step 3: Update CRM record
        if not INTEGRATION_ID:
            raise HTTPException(status_code=500, detail="GENESYS_INTEGRATION_ID not configured")
            
        crm_response = update_crm_case(auth_manager, INTEGRATION_ID, req.case_id, summary_data)
        
        return {
            "status": "success",
            "summary": summary_data.model_dump(),
            "crm_update": crm_response
        }
        
    except httpx.HTTPStatusError as e:
        status_code = e.response.status_code
        detail = e.response.text
        raise HTTPException(status_code=status_code, detail=f"Genesys API error: {detail}")
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Processing failed: {str(e)}")

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

Common Errors & Debugging

Error: 401 Unauthorized

• What causes it: The access token has expired, the client credentials are invalid, or the OAuth client lacks the required scopes.
• How to fix it: Verify the client ID and secret match a registered application in Genesys Admin. Ensure the application has the interaction:read, analytics:conversation:view, and integration:write scopes assigned. The GenesysAuthManager automatically refreshes tokens when expiration approaches. If the error persists, check the OAuth client configuration in Genesys Admin under Applications.
• Code showing the fix: The get_headers() method checks expires_at and calls _fetch_token() proactively. The API functions also catch 401 status codes and trigger a manual refresh before retrying the request.

Error: 429 Too Many Requests

• What causes it: Genesys Cloud enforces strict rate limits on analytics and CRM endpoints. Bulk processing or rapid polling triggers this limit.
• How to fix it: Implement exponential backoff. The tenacity decorator in the code handles this automatically. For high-volume workloads, queue requests and process them asynchronously instead of synchronous blocking calls.
• Code showing the fix: The @retry decorator on fetch_transcript and update_crm_case catches httpx.HTTPStatusError, waits between attempts, and raises the error after three failures. The 429 handler explicitly reads the Retry-After header when available.

Error: 400 Bad Request (CRM Field Mismatch)

• What causes it: The field names in the payload do not match the exact field names configured in the Genesys CRM integration schema.
• How to fix it: Navigate to Genesys Admin > Integrations > CRM > Fields. Copy the exact field names (case-sensitive) and update the payload dictionary in update_crm_case. Test with a single field first to isolate mapping errors.
• Code showing the fix: Validate the CRM schema by calling GET /api/v2/integration/instances/{id}/entities/Case to retrieve the field definitions. Match the name property exactly in your payload.

Error: JSONDecodeError from LLM

• What causes it: The LLM model returned natural language text instead of strict JSON, or the response was truncated.
• How to fix it: Enforce response_format={"type": "json_object"} in the OpenAI call. Add a temperature of 0.1 to reduce randomness. Implement a fallback parser that strips markdown code blocks before passing the string to json.loads().
• Code showing the fix: The generate_llm_summary function wraps json.loads() in a try-except block and raises a descriptive RuntimeError. Add a pre-processing step: raw_json = raw_json.strip().removeprefix("```json").removesuffix("```") before parsing.

Official References

• Genesys Cloud Conversation Analytics API
• Genesys Cloud CRM Integration API
• Genesys Cloud OAuth 2.0 Documentation
• OpenAI Python SDK Reference
• FastAPI Documentation