Sharing Genesys Cloud Agent Assist Real-Time Notes via WebSocket API with Python

What You Will Build

A production-ready Python module that streams Agent Assist notes to collaborators in real time, validates payloads against gateway constraints, resolves version conflicts, syncs with external case systems, and maintains audit trails. The implementation uses the Genesys Cloud Python SDK for authentication and REST validation, combined with websockets and httpx for low-latency streaming and external synchronization. This tutorial covers Python 3.9+.

Prerequisites

• OAuth 2.0 Client Credentials flow with scopes: agentassist:note:write, agentassist:note:read, realtime:subscribe
• Genesys Cloud Python SDK version 3.0.0 or later
• Python 3.9+ runtime
• External dependencies: websockets>=12.0, httpx>=0.25.0, pydantic>=2.5.0, genesys-cloud-sdk>=3.0.0
• A deployed Agent Assist configuration with note sharing enabled in your Genesys Cloud environment

Authentication Setup

The Genesys Cloud Python SDK handles OAuth token acquisition and automatic refresh. You must initialize the platform client and extract the active access token for WebSocket handshake authorization.

import os
from genesyscloud.platform_client_v2 import PureCloudPlatformClientV2
from genesyscloud.auth.client_credentials_auth import ClientCredentialsAuth
from typing import Optional

def get_platform_client() -> PureCloudPlatformClientV2:
    client = PureCloudPlatformClientV2()
    auth = ClientCredentialsAuth(
        os.environ["GENESYS_CLIENT_ID"],
        os.environ["GENESYS_CLIENT_SECRET"],
        os.environ["GENESYS_REGION"]
    )
    client.set_auth(auth)
    return client

def get_active_token(client: PureCloudPlatformClientV2) -> str:
    """Extracts the current bearer token from the SDK auth provider."""
    auth_provider = client.get_auth()
    token_data = auth_provider.get_token()
    if not token_data or "access_token" not in token_data:
        raise RuntimeError("Failed to retrieve OAuth access token")
    return token_data["access_token"]

The SDK caches tokens in memory and refreshes them automatically before expiration. When initializing the WebSocket connection, you must pass the token as a query parameter or in the Authorization header depending on your environment configuration. The standard Genesys Cloud Realtime API expects the token as a query parameter: ?access_token={token}.

Implementation

Step 1: WebSocket Connection and Realtime Subscription

Genesys Cloud exposes real-time streaming through the /api/v2/realtime/{domain} WebSocket endpoint. You must authenticate the handshake, subscribe to the note sharing channel, and maintain connection health with ping/pong handling.

import asyncio
import websockets
import json
import logging
from typing import Dict, Any

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

class NoteWebSocketClient:
    def __init__(self, region: str, access_token: str):
        self.ws_url = f"wss://api.{region}.mypurecloud.com/api/v2/realtime/agent-assist/notes?access_token={access_token}"
        self.websocket: Optional[websockets.WebSocketClientProtocol] = None
        self._shutdown_event = asyncio.Event()

    async def connect(self) -> None:
        logger.info("Establishing WebSocket connection to Genesys Cloud Realtime API")
        async with websockets.connect(self.ws_url, ping_interval=20, ping_timeout=10) as ws:
            self.websocket = ws
            # Subscribe to note sharing events
            subscription_payload = {
                "type": "subscribe",
                "domain": "agent-assist/notes",
                "filter": {"eventTypes": ["note.shared", "note.updated", "note.conflict"]}
            }
            await ws.send(json.dumps(subscription_payload))
            logger.info("Subscription request sent. Awaiting handshake confirmation")
            
            # Process incoming messages until shutdown
            await self._listen_loop(ws)

    async def _listen_loop(self, ws: websockets.WebSocketClientProtocol) -> None:
        try:
            async for message in ws:
                data = json.loads(message)
                await self._handle_inbound_message(data)
        except websockets.ConnectionClosed as e:
            logger.warning("WebSocket connection closed: %s", e)
        except Exception as e:
            logger.error("Unexpected error in WebSocket loop: %s", e)
        finally:
            self._shutdown_event.set()

    async def _handle_inbound_message(self, data: Dict[str, Any]) -> None:
        if data.get("type") == "handshake":
            logger.info("Realtime gateway handshake confirmed")
        elif data.get("type") == "note.shared":
            logger.info("Received shared note event: %s", data.get("noteId"))
        elif data.get("type") == "error":
            logger.error("Gateway error: %s", data.get("message"))

The filter object restricts inbound traffic to note-specific events. The Realtime API enforces a maximum subscription payload size of 8 KB. Always validate filter syntax before transmission.

Step 2: Payload Construction and Schema Validation

Before transmitting a note share, you must validate the payload against assist gateway constraints. The gateway rejects payloads exceeding 16 KB, disallows unescaped HTML, and requires explicit collaborator references. You will use Pydantic for schema validation and implement content sanitization.

import re
from pydantic import BaseModel, field_validator, ValidationError
from typing import List, Dict, Any

MAX_NOTE_SIZE_BYTES = 16384
ALLOWED_HTML_TAGS = {"b", "i", "u", "br", "p", "ul", "li"}

class CollaboratorReference(BaseModel):
    userId: str
    permission: str  # "read" or "write"

class NoteSharePayload(BaseModel):
    noteId: str
    content: str
    version: int
    collaborators: List[CollaboratorReference]
    metadata: Dict[str, Any] = {}

    @field_validator("content")
    @classmethod
    def sanitize_content(cls, v: str) -> str:
        # Remove script tags and event handlers
        cleaned = re.sub(r"<script[^>]*>.*?</script>", "", v, flags=re.IGNORECASE | re.DOTALL)
        cleaned = re.sub(r"on\w+=","", cleaned, flags=re.IGNORECASE)
        return cleaned

    @field_validator("content")
    @classmethod
    def enforce_size_limit(cls, v: str) -> str:
        if len(v.encode("utf-8")) > MAX_NOTE_SIZE_BYTES:
            raise ValueError(f"Note content exceeds maximum size limit of {MAX_NOTE_SIZE_BYTES} bytes")
        return v

    @field_validator("collaborators")
    @classmethod
    def validate_permissions(cls, v: List[CollaboratorReference]) -> List[CollaboratorReference]:
        for collab in v:
            if collab.permission not in ("read", "write"):
                raise ValueError(f"Invalid collaborator permission: {collab.permission}")
        return v

def validate_share_payload(payload_dict: Dict[str, Any]) -> NoteSharePayload:
    try:
        return NoteSharePayload(**payload_dict)
    except ValidationError as e:
        logger.error("Payload validation failed: %s", e)
        raise

This validation pipeline prevents sharing failures caused by oversized payloads, malicious content injection, or invalid permission tokens. The gateway returns a 400 Bad Request if these constraints are violated.

Step 3: Atomic SEND Operations and Conflict Resolution

The Realtime API uses atomic SEND operations for note distribution. You must frame the message correctly, track optimistic updates locally, and handle version conflicts when multiple agents modify the same note concurrently.

import time
from httpx import AsyncClient
from typing import Optional

class NoteShareManager:
    def __init__(self, region: str, access_token: str, base_url: str):
        self.region = region
        self.access_token = access_token
        self.base_url = base_url
        self.http_client = AsyncClient(
            headers={"Authorization": f"Bearer {access_token}", "Content-Type": "application/json"},
            timeout=10.0
        )
        self.version_matrix: Dict[str, int] = {}  # Optimistic update tracking

    async def send_note_share(self, ws: websockets.WebSocketClientProtocol, payload: NoteSharePayload) -> Dict[str, Any]:
        send_frame = {
            "type": "send",
            "operation": "share_note",
            "noteId": payload.noteId,
            "content": payload.content,
            "version": payload.version,
            "collaborators": [c.dict() for c in payload.collaborators],
            "timestamp": time.time()
        }
        
        # Optimistic version increment
        self.version_matrix[payload.noteId] = payload.version + 1
        
        await ws.send(json.dumps(send_frame))
        logger.info("Atomic SEND operation queued for note %s at version %d", payload.noteId, payload.version)
        return send_frame

    async def resolve_version_conflict(self, note_id: str, server_version: int) -> Optional[NoteSharePayload]:
        """Fetches latest state from REST API to reconcile conflicts."""
        try:
            response = await self.http_client.get(f"{self.base_url}/api/v2/agent-assist/notes/{note_id}")
            if response.status_code == 404:
                logger.warning("Note %s not found during conflict resolution", note_id)
                return None
            response.raise_for_status()
            
            note_data = response.json()
            self.version_matrix[note_id] = note_data["version"]
            
            # Reconstruct payload with server truth
            reconciled = NoteSharePayload(
                noteId=note_id,
                content=note_data.get("content", ""),
                version=note_data["version"],
                collaborators=[]  # Reset or re-fetch collaborators as needed
            )
            logger.info("Conflict resolved for note %s. Synced to server version %d", note_id, note_data["version"])
            return reconciled
        except httpx.HTTPStatusError as e:
            logger.error("Conflict resolution failed: HTTP %d", e.response.status_code)
            return None

The version_matrix dictionary tracks client-side optimistic increments. When the gateway returns a conflict directive (typically via a note.conflict event containing the authoritative version), you must fetch the latest state via the REST endpoint /api/v2/agent-assist/notes/{noteId}, update the local matrix, and retry the SEND operation.

Step 4: Webhook Synchronization and Audit Logging

External case systems require event synchronization. You will implement a webhook callback dispatcher, track share latency, and generate structured audit logs for governance compliance.

import uuid
from datetime import datetime, timezone
from typing import List

class ShareAuditLogger:
    def __init__(self, webhook_url: str):
        self.webhook_url = webhook_url
        self.http_client = AsyncClient(timeout=5.0)
        self.latency_samples: List[float] = []
        self.sync_accuracy_count = 0
        self.total_sync_attempts = 0

    async def dispatch_and_log(self, note_id: str, payload_version: int, send_timestamp: float, success: bool) -> None:
        self.total_sync_attempts += 1
        latency = time.time() - send_timestamp
        self.latency_samples.append(latency)
        
        if success:
            self.sync_accuracy_count += 1
        
        audit_record = {
            "auditId": str(uuid.uuid4()),
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "noteId": note_id,
            "clientVersion": payload_version,
            "shareLatencyMs": round(latency * 1000, 2),
            "syncSuccess": success,
            "governanceTag": "agent-assist-note-share"
        }
        
        # Log locally
        logger.info("AUDIT: %s", json.dumps(audit_record))
        
        # Sync with external case system
        try:
            await self.http_client.post(self.webhook_url, json=audit_record)
        except Exception as e:
            logger.warning("Webhook sync failed for audit record %s: %s", audit_record["auditId"], e)

    def get_metrics(self) -> Dict[str, float]:
        if not self.latency_samples:
            return {"avgLatencyMs": 0.0, "syncAccuracyRate": 0.0}
        avg_latency = sum(self.latency_samples) / len(self.latency_samples)
        accuracy = self.sync_accuracy_count / self.total_sync_attempts if self.total_sync_attempts > 0 else 0.0
        return {"avgLatencyMs": round(avg_latency * 1000, 2), "syncAccuracyRate": round(accuracy, 4)}

The audit logger captures latency from SEND initiation to gateway acknowledgment. It posts structured JSON to your external webhook endpoint for case system alignment. The sync_accuracy_rate metric helps identify network degradation or gateway throttling during peak assist scaling.

Complete Working Example

import asyncio
import os
import json
import logging
from typing import Dict, Any, Optional
import websockets
import httpx
from pydantic import BaseModel, field_validator, ValidationError
from genesyscloud.platform_client_v2 import PureCloudPlatformClientV2
from genesyscloud.auth.client_credentials_auth import ClientCredentialsAuth

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

# --- Models & Validation ---
MAX_NOTE_SIZE_BYTES = 16384

class CollaboratorReference(BaseModel):
    userId: str
    permission: str

class NoteSharePayload(BaseModel):
    noteId: str
    content: str
    version: int
    collaborators: list[CollaboratorReference]

    @field_validator("content")
    @classmethod
    def sanitize_and_limit(cls, v: str) -> str:
        import re
        cleaned = re.sub(r"<script[^>]*>.*?</script>", "", v, flags=re.IGNORECASE | re.DOTALL)
        if len(cleaned.encode("utf-8")) > MAX_NOTE_SIZE_BYTES:
            raise ValueError(f"Content exceeds {MAX_NOTE_SIZE_BYTES} bytes")
        return cleaned

# --- Core Sharer Class ---
class AgentAssistNoteSharer:
    def __init__(self, region: str, client_id: str, client_secret: str, webhook_url: str):
        self.region = region
        self.webhook_url = webhook_url
        self.base_url = f"https://api.{region}.mypurecloud.com"
        self.audit_logger = ShareAuditLogger(webhook_url)
        self.version_matrix: Dict[str, int] = {}
        
        # Auth setup
        self.client = PureCloudPlatformClientV2()
        self.auth = ClientCredentialsAuth(client_id, client_secret, region)
        self.client.set_auth(self.auth)
        
        # HTTP client for REST validation & conflict resolution
        self.http = httpx.AsyncClient(
            headers={"Content-Type": "application/json"},
            timeout=10.0
        )

    def _get_token(self) -> str:
        token_data = self.auth.get_token()
        return token_data["access_token"]

    async def resolve_conflict(self, note_id: str, server_version: int) -> Optional[NoteSharePayload]:
        token = self._get_token()
        self.http.headers["Authorization"] = f"Bearer {token}"
        try:
            resp = await self.http.get(f"{self.base_url}/api/v2/agent-assist/notes/{note_id}")
            resp.raise_for_status()
            data = resp.json()
            self.version_matrix[note_id] = data["version"]
            return NoteSharePayload(
                noteId=note_id,
                content=data.get("content", ""),
                version=data["version"],
                collaborators=[]
            )
        except httpx.HTTPError as e:
            logger.error("Conflict resolution failed: %s", e)
            return None

    async def share_note(self, payload: NoteSharePayload) -> Dict[str, Any]:
        import time
        ws_url = f"wss://api.{self.region}.mypurecloud.com/api/v2/realtime/agent-assist/notes?access_token={self._get_token()}"
        
        try:
            async with websockets.connect(ws_url, ping_interval=20) as ws:
                # Subscribe
                await ws.send(json.dumps({"type": "subscribe", "domain": "agent-assist/notes", "filter": {"eventTypes": ["note.shared", "note.conflict"]}}))
                
                send_ts = time.time()
                send_frame = {
                    "type": "send",
                    "operation": "share_note",
                    "noteId": payload.noteId,
                    "content": payload.content,
                    "version": payload.version,
                    "collaborators": [c.dict() for c in payload.collaborators],
                    "timestamp": send_ts
                }
                self.version_matrix[payload.noteId] = payload.version + 1
                
                await ws.send(json.dumps(send_frame))
                logger.info("SEND queued for %s v%d", payload.noteId, payload.version)
                
                # Await confirmation or conflict
                async for msg in ws:
                    data = json.loads(msg)
                    if data.get("type") == "handshake":
                        continue
                    elif data.get("type") == "note.shared":
                        await self.audit_logger.dispatch_and_log(payload.noteId, payload.version, send_ts, success=True)
                        return {"status": "success", "message": "Note shared successfully"}
                    elif data.get("type") == "note.conflict":
                        server_ver = data.get("serverVersion")
                        logger.warning("Version conflict detected. Server version: %s", server_ver)
                        await self.audit_logger.dispatch_and_log(payload.noteId, payload.version, send_ts, success=False)
                        reconciled = await self.resolve_conflict(payload.noteId, server_ver)
                        if reconciled:
                            return await self.share_note(reconciled)
                        return {"status": "failed", "reason": "conflict_unresolvable"}
                    elif data.get("type") == "error":
                        await self.audit_logger.dispatch_and_log(payload.noteId, payload.version, send_ts, success=False)
                        return {"status": "failed", "reason": data.get("message")}
        except Exception as e:
            logger.error("WebSocket share operation failed: %s", e)
            return {"status": "failed", "reason": str(e)}

# --- Audit & Metrics Helper ---
class ShareAuditLogger:
    def __init__(self, webhook_url: str):
        self.webhook_url = webhook_url
        self.http = httpx.AsyncClient(timeout=5.0)
        self.latency_samples = []
        self.success_count = 0
        self.total_attempts = 0

    async def dispatch_and_log(self, note_id: str, version: int, send_ts: float, success: bool):
        self.total_attempts += 1
        latency = time.time() - send_ts
        self.latency_samples.append(latency)
        if success:
            self.success_count += 1
        
        record = {
            "auditId": str(uuid.uuid4()),
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "noteId": note_id,
            "version": version,
            "latencyMs": round(latency * 1000, 2),
            "success": success
        }
        logger.info("AUDIT: %s", json.dumps(record))
        try:
            await self.http.post(self.webhook_url, json=record)
        except Exception as e:
            logger.warning("Webhook dispatch failed: %s", e)

    def metrics(self):
        avg = sum(self.latency_samples) / len(self.latency_samples) * 1000 if self.latency_samples else 0
        acc = self.success_count / self.total_attempts if self.total_attempts > 0 else 0
        return {"avgLatencyMs": round(avg, 2), "accuracyRate": round(acc, 4)}

# --- Execution ---
async def main():
    sharer = AgentAssistNoteSharer(
        region="us-east-1",
        client_id=os.environ["GENESYS_CLIENT_ID"],
        client_secret=os.environ["GENESYS_CLIENT_SECRET"],
        webhook_url="https://your-external-cms.com/api/genesys-sync"
    )
    
    try:
        payload = NoteSharePayload(
            noteId="a1b2c3d4-e5f6-7890-abcd-ef1234567890",
            content="Customer requires escalation to tier 2. Verify warranty status before proceeding.",
            version=1,
            collaborators=[{"userId": "user-123", "permission": "write"}, {"userId": "user-456", "permission": "read"}]
        )
        
        result = await sharer.share_note(payload)
        print(json.dumps(result, indent=2))
        print(json.dumps(sharer.audit_logger.metrics(), indent=2))
    except ValidationError as ve:
        logger.error("Payload validation failed: %s", ve)
    except Exception as e:
        logger.error("Execution failed: %s", e)

if __name__ == "__main__":
    import uuid
    from datetime import datetime, timezone
    import time
    asyncio.run(main())

Common Errors and Debugging

Error: 401 Unauthorized

• Cause: OAuth token expired or missing required scopes (agentassist:note:write, realtime:subscribe).
• Fix: Ensure the SDK auth provider refreshes tokens before WebSocket handshake. Call auth.get_token() immediately before constructing the WebSocket URL.
• Code: Add a token validation step that checks auth.get_token().get("expires_in") and forces auth.refresh_token() if expiration is within 30 seconds.

Error: 429 Too Many Requests

• Cause: Exceeding Realtime API rate limits during high-volume assist scaling.
• Fix: Implement exponential backoff with jitter before retrying SEND operations. The SDK does not handle WebSocket rate limits automatically.
• Code: Wrap ws.send() in a retry loop with await asyncio.sleep(min(2 ** attempt + random.uniform(0, 0.5), 10)).

Error: 400 Bad Request (Schema Violation)

• Cause: Payload exceeds 16 KB, contains blocked HTML, or uses invalid collaborator permissions.
• Fix: Run all payloads through the NoteSharePayload Pydantic model before transmission. The field_validator catches size and sanitization violations early.
• Code: The sanitize_and_limit validator enforces the 16384-byte ceiling and strips script tags. Ensure external data sources are pre-cleaned.

Error: WebSocket Connection Closed (Code 1006)

• Cause: Network interruption or gateway idle timeout.
• Fix: Enable ping_interval=20 and ping_timeout=10 in websockets.connect(). Implement an automatic reconnect loop that re-authenticates and re-subscribes.
• Code: Wrap the connect() logic in a while not shutdown: loop with asyncio.sleep(5) on disconnection.

Official References

• Genesys Cloud Realtime API Documentation
• Agent Assist Notes REST API
• Genesys Cloud Python SDK Reference
• OAuth 2.0 Scopes Reference