Insights¶

Insights connects CXAS Scrapi to the CCAI Insights API — Google Cloud's contact center analytics platform. Through this class you can access conversation analytics, manage scorecards, and pull quality AI data that helps you understand how well your agent is performing at scale.

The class uses a REST-based client (rather than a gRPC SDK) because the Insights API has a different endpoint pattern from the CES API used by other CXAS classes. The authentication flow is the same — your credentials from Common are reused automatically.

Quick Example¶

from cxas_scrapi import Insights

insights = Insights(
    project_id="my-gcp-project",
    location="us-central1",
    creds_path="/path/to/service_account.json",
)

# List all conversations (conversations indexed in Insights)
conversations = insights.list_conversations()
for conv in conversations:
    print(conv.get("name"), conv.get("duration"))

# List all scorecards
scorecards = insights.list_scorecards()
for sc in scorecards:
    print(sc.get("displayName"))

Reference¶

Insights ¶

Insights(project_id, location='us-central1', api_version='v1', creds_path=None, creds_dict=None, creds=None, scope=None, **kwargs)

Bases: Common

Core Class for managing CCAI Insights Resources and base operations.

Initializes the Insights API base client.

Source code in src/cxas_scrapi/core/insights.py

def __init__(
    self,
    project_id: str,
    location: str = "us-central1",
    api_version: str = "v1",
    creds_path: str | None = None,
    creds_dict: dict[str, str] | None = None,
    creds: Any = None,
    scope: list[str] | None = None,
    **kwargs,
):
    """Initializes the Insights API base client."""
    super().__init__(
        creds_path=creds_path,
        creds_dict=creds_dict,
        creds=creds,
        scope=scope,
        **kwargs,
    )
    self.project_id = project_id
    self.location = location
    self.parent = f"projects/{project_id}/locations/{location}"

    base_endpoint = "contactcenterinsights.googleapis.com"
    if location != "global":
        self._base_url = f"https://{location}-{base_endpoint}/{api_version}"
    else:
        self._base_url = f"https://{base_endpoint}/{api_version}"

    self.session = requests.Session()
    retries = Retry(
        total=5,
        backoff_factor=1,
        status_forcelist=[
            http.HTTPStatus.TOO_MANY_REQUESTS,
            http.HTTPStatus.INTERNAL_SERVER_ERROR,
            http.HTTPStatus.BAD_GATEWAY,
            http.HTTPStatus.SERVICE_UNAVAILABLE,
            http.HTTPStatus.GATEWAY_TIMEOUT,
        ],
    )
    self.session.mount("https://", HTTPAdapter(max_retries=retries))

list_conversations ¶

list_conversations(filter_str=None, view=None, page_size=100, max_pages=5)

Lists conversations in the configured parent location.

Source code in src/cxas_scrapi/core/insights.py

def list_conversations(
    self,
    filter_str: str | None = None,
    view: str | None = None,
    page_size: int = 100,
    max_pages: int = 5,
) -> list[dict[str, Any]]:
    """Lists conversations in the configured parent location."""
    path = f"{self.parent}/conversations"
    params = {"pageSize": page_size}
    if filter_str:
        params["filter"] = filter_str
    if view:
        params["view"] = view

    results = []
    page_token = None
    pages = 0
    while pages < max_pages:
        if page_token:
            params["pageToken"] = page_token
        res = self._request("GET", path, params=params)
        results.extend(res.get("conversations", []))
        page_token = res.get("nextPageToken")
        pages += 1
        if not page_token:
            break
    return results

get_conversation ¶

get_conversation(name)

Gets a single conversation by name or ID.

Source code in src/cxas_scrapi/core/insights.py

def get_conversation(self, name: str) -> dict[str, Any]:
    """Gets a single conversation by name or ID."""
    if not name.startswith("projects/"):
        name = f"{self.parent}/conversations/{name}"
    return self._request("GET", name)