Skip to content

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: typing.Any,
) -> None:
    """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, view=None)

Gets a single conversation by name or ID.

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

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

create_conversation ¶

create_conversation(conversation, conversation_id=None, parent=None)

Creates or ingests a new conversation.

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

def create_conversation(
    self,
    conversation: dict[str, Any],
    conversation_id: str | None = None,
    parent: str | None = None,
) -> dict[str, Any]:
    """Creates or ingests a new conversation."""
    parent = parent or self.parent
    params = (
        {"conversationId": conversation_id} if conversation_id else None
    )
    return self._request(
        "POST", f"{parent}/conversations", data=conversation, params=params
    )

delete_conversation ¶

delete_conversation(name, force=True)

Deletes a conversation by resource name or ID.

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

def delete_conversation(self, name: str, force: bool = True) -> None:
    """Deletes a conversation by resource name or ID."""
    if not name.startswith("projects/"):
        name = f"{self.parent}/conversations/{name}"
    params = {"force": "true" if force else "false"}
    self._request("DELETE", name, params=params)

analyze_conversation ¶

analyze_conversation(name, annotator_selector=None)

Triggers single-conversation analysis (e.g. for scorecard dry runs).

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

def analyze_conversation(
    self,
    name: str,
    annotator_selector: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Triggers single-conversation analysis (e.g. for scorecard
    dry runs)."""
    if not name.startswith("projects/"):
        name = f"{self.parent}/conversations/{name}"

    payload = {}
    if annotator_selector:
        payload["annotatorSelector"] = annotator_selector
    return self._request("POST", f"{name}/analyses", data=payload)

bulk_analyze_conversations ¶

bulk_analyze_conversations(parent=None, filter_str=None, annotator_selector=None, analysis_percentage=100.0)

Triggers batch analysis over matching conversations.

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

def bulk_analyze_conversations(
    self,
    parent: str | None = None,
    filter_str: str | None = None,
    annotator_selector: dict[str, Any] | None = None,
    analysis_percentage: float = 100.0,
) -> dict[str, Any]:
    """Triggers batch analysis over matching conversations."""
    parent = parent or self.parent
    payload = {
        "filter": filter_str or "",
        "analysisPercentage": analysis_percentage,
    }
    if annotator_selector:
        payload["annotatorSelector"] = annotator_selector
    return self._request(
        "POST", f"{parent}/conversations:bulkAnalyze", data=payload
    )

list_issue_models ¶

list_issue_models(parent=None)

Lists issue models (topic models) under the specified parent.

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

def list_issue_models(
    self, parent: str | None = None
) -> list[dict[str, Any]]:
    """Lists issue models (topic models) under the specified parent."""
    parent = parent or self.parent
    return self._list_paginated(f"{parent}/issueModels", "issueModels")

get_issue_model ¶

get_issue_model(name)

Gets details of an issue model by name or ID.

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

def get_issue_model(self, name: str) -> dict[str, Any]:
    """Gets details of an issue model by name or ID."""
    if not name.startswith("projects/"):
        name = f"{self.parent}/issueModels/{name}"
    return self._request("GET", name)

create_issue_model ¶

create_issue_model(display_name, input_data_config=None, parent=None, issue_model_id=None, model_type='TYPE_V2')

Creates a new issue model for topic modelling.

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

def create_issue_model(
    self,
    display_name: str,
    input_data_config: dict[str, Any] | None = None,
    parent: str | None = None,
    issue_model_id: str | None = None,
    model_type: str = "TYPE_V2",
) -> dict[str, Any]:
    """Creates a new issue model for topic modelling."""
    parent = parent or self.parent
    payload = {
        "displayName": display_name,
        "modelType": model_type,
        "inputDataConfig": input_data_config or {"medium": "CHAT"},
    }
    params = {"issueModelId": issue_model_id} if issue_model_id else None
    return self._request(
        "POST", f"{parent}/issueModels", data=payload, params=params
    )

update_issue_model ¶

update_issue_model(name, issue_model, update_mask='*')

Updates an issue model.

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

def update_issue_model(
    self,
    name: str,
    issue_model: dict[str, Any],
    update_mask: str = "*",
) -> dict[str, Any]:
    """Updates an issue model."""
    if not name.startswith("projects/"):
        name = f"{self.parent}/issueModels/{name}"
    params = {"updateMask": update_mask}
    return self._request("PATCH", name, data=issue_model, params=params)

delete_issue_model ¶

delete_issue_model(name)

Deletes an issue model.

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

def delete_issue_model(self, name: str) -> None:
    """Deletes an issue model."""
    if not name.startswith("projects/"):
        name = f"{self.parent}/issueModels/{name}"
    self._request("DELETE", name)

deploy_issue_model ¶

deploy_issue_model(name)

Deploys an issue model so it actively tags incoming conversations.

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

def deploy_issue_model(self, name: str) -> dict[str, Any]:
    """Deploys an issue model so it actively tags incoming conversations."""
    if not name.startswith("projects/"):
        name = f"{self.parent}/issueModels/{name}"
    return self._request("POST", f"{name}:deploy", data={})

undeploy_issue_model ¶

undeploy_issue_model(name)

Undeploys an active issue model.

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

def undeploy_issue_model(self, name: str) -> dict[str, Any]:
    """Undeploys an active issue model."""
    if not name.startswith("projects/"):
        name = f"{self.parent}/issueModels/{name}"
    return self._request("POST", f"{name}:undeploy", data={})

calculate_issue_model_stats ¶

calculate_issue_model_stats(name)

Calculates and returns issue model statistics.

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

def calculate_issue_model_stats(self, name: str) -> dict[str, Any]:
    """Calculates and returns issue model statistics."""
    if not name.startswith("projects/"):
        name = f"{self.parent}/issueModels/{name}"
    return self._request("GET", f"{name}:calculateIssueModelStats")

list_issues ¶

list_issues(parent_issue_model)

Lists issues (topics) belonging to an issue model.

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

def list_issues(self, parent_issue_model: str) -> list[dict[str, Any]]:
    """Lists issues (topics) belonging to an issue model."""
    if not parent_issue_model.startswith("projects/"):
        parent_issue_model = (
            f"{self.parent}/issueModels/{parent_issue_model}"
        )
    return self._list_paginated(f"{parent_issue_model}/issues", "issues")

get_issue ¶

get_issue(name)

Gets a single issue (topic) by name.

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

def get_issue(self, name: str) -> dict[str, Any]:
    """Gets a single issue (topic) by name."""
    return self._request("GET", name)

list_analysis_rules ¶

list_analysis_rules(parent=None)

Lists analysis rules in the specified parent.

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

def list_analysis_rules(
    self, parent: str | None = None
) -> list[dict[str, Any]]:
    """Lists analysis rules in the specified parent."""
    parent = parent or self.parent
    return self._list_paginated(f"{parent}/analysisRules", "analysisRules")

get_analysis_rule ¶

get_analysis_rule(name)

Gets an analysis rule by name or ID.

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

def get_analysis_rule(self, name: str) -> dict[str, Any]:
    """Gets an analysis rule by name or ID."""
    if not name.startswith("projects/"):
        name = f"{self.parent}/analysisRules/{name}"
    return self._request("GET", name)

create_analysis_rule ¶

create_analysis_rule(display_name, conversation_filter, annotator_selector, active=True, parent=None, analysis_rule_id=None)

Creates a new analysis rule for automated evaluations.

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

def create_analysis_rule(
    self,
    display_name: str,
    conversation_filter: str,
    annotator_selector: dict[str, Any],
    active: bool = True,
    parent: str | None = None,
    analysis_rule_id: str | None = None,
) -> dict[str, Any]:
    """Creates a new analysis rule for automated evaluations."""
    parent = parent or self.parent
    payload = {
        "displayName": display_name,
        "conversationFilter": conversation_filter,
        "annotatorSelector": annotator_selector,
        "active": active,
    }
    params = (
        {"analysisRuleId": analysis_rule_id} if analysis_rule_id else None
    )
    return self._request(
        "POST", f"{parent}/analysisRules", data=payload, params=params
    )

update_analysis_rule ¶

update_analysis_rule(name, analysis_rule, update_mask='*')

Updates an analysis rule.

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

def update_analysis_rule(
    self,
    name: str,
    analysis_rule: dict[str, Any],
    update_mask: str = "*",
) -> dict[str, Any]:
    """Updates an analysis rule."""
    if not name.startswith("projects/"):
        name = f"{self.parent}/analysisRules/{name}"
    params = {"updateMask": update_mask}
    return self._request("PATCH", name, data=analysis_rule, params=params)

activate_analysis_rule ¶

activate_analysis_rule(name, active=True)

Convenience method to activate or deactivate an analysis rule.

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

def activate_analysis_rule(
    self, name: str, active: bool = True
) -> dict[str, Any]:
    """Convenience method to activate or deactivate an analysis rule."""
    if not name.startswith("projects/"):
        name = f"{self.parent}/analysisRules/{name}"
    params = {"updateMask": "active"}
    return self._request(
        "PATCH", name, data={"active": active}, params=params
    )

delete_analysis_rule ¶

delete_analysis_rule(name)

Deletes an analysis rule.

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

def delete_analysis_rule(self, name: str) -> None:
    """Deletes an analysis rule."""
    if not name.startswith("projects/"):
        name = f"{self.parent}/analysisRules/{name}"
    self._request("DELETE", name)

list_autolabeling_rules ¶

list_autolabeling_rules(parent=None)

Lists autolabeling rules in the specified parent.

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

def list_autolabeling_rules(
    self, parent: str | None = None
) -> list[dict[str, Any]]:
    """Lists autolabeling rules in the specified parent."""
    parent = parent or self.parent
    return self._list_paginated(
        f"{parent}/autoLabelingRules", "autoLabelingRules"
    )

get_autolabeling_rule ¶

get_autolabeling_rule(name)

Gets an autolabeling rule by name or ID.

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

def get_autolabeling_rule(self, name: str) -> dict[str, Any]:
    """Gets an autolabeling rule by name or ID."""
    if not name.startswith("projects/"):
        name = f"{self.parent}/autoLabelingRules/{name}"
    return self._request("GET", name)

create_autolabeling_rule ¶

create_autolabeling_rule(auto_labeling_rule, auto_labeling_rule_id=None, parent=None)

Creates a new autolabeling rule.

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

def create_autolabeling_rule(
    self,
    auto_labeling_rule: dict[str, Any],
    auto_labeling_rule_id: str | None = None,
    parent: str | None = None,
) -> dict[str, Any]:
    """Creates a new autolabeling rule."""
    parent = parent or self.parent
    params = (
        {"autoLabelingRuleId": auto_labeling_rule_id}
        if auto_labeling_rule_id
        else None
    )
    return self._request(
        "POST",
        f"{parent}/autoLabelingRules",
        data=auto_labeling_rule,
        params=params,
    )

update_autolabeling_rule ¶

update_autolabeling_rule(name, auto_labeling_rule, update_mask='*')

Updates an autolabeling rule.

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

def update_autolabeling_rule(
    self,
    name: str,
    auto_labeling_rule: dict[str, Any],
    update_mask: str | list[str] = "*",
) -> dict[str, Any]:
    """Updates an autolabeling rule."""
    if not name.startswith("projects/"):
        name = f"{self.parent}/autoLabelingRules/{name}"
    mask_str = (
        ",".join(update_mask)
        if isinstance(update_mask, (list, tuple))
        else update_mask
    )
    params = {"updateMask": mask_str}
    return self._request(
        "PATCH", name, data=auto_labeling_rule, params=params
    )

delete_autolabeling_rule ¶

delete_autolabeling_rule(name)

Deletes an autolabeling rule.

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

def delete_autolabeling_rule(self, name: str) -> None:
    """Deletes an autolabeling rule."""
    if not name.startswith("projects/"):
        name = f"{self.parent}/autoLabelingRules/{name}"
    self._request("DELETE", name)

list_dashboards ¶

list_dashboards(parent=None, filter_str=None, page_size=100, max_pages=5)

Lists dashboards in the specified parent project/location.

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

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

    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("dashboards", []))
        page_token = res.get("nextPageToken")
        pages += 1
        if not page_token:
            break
    return results

get_dashboard ¶

get_dashboard(name)

Gets a dashboard by name or ID.

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

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

create_dashboard ¶

create_dashboard(dashboard, dashboard_id=None, parent=None)

Creates a new dashboard.

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

def create_dashboard(
    self,
    dashboard: dict[str, Any],
    dashboard_id: str | None = None,
    parent: str | None = None,
) -> dict[str, Any]:
    """Creates a new dashboard."""
    parent = parent or self.parent
    params = {"dashboardId": dashboard_id} if dashboard_id else None
    return self._request(
        "POST",
        f"{parent}/dashboards",
        data=dashboard,
        params=params,
    )

update_dashboard ¶

update_dashboard(name, dashboard, update_mask='*')

Updates an existing dashboard.

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

def update_dashboard(
    self,
    name: str,
    dashboard: dict[str, Any],
    update_mask: str | list[str] = "*",
) -> dict[str, Any]:
    """Updates an existing dashboard."""
    if not name.startswith("projects/"):
        name = f"{self.parent}/dashboards/{name}"
    mask_str = (
        ",".join(update_mask)
        if isinstance(update_mask, (list, tuple))
        else update_mask
    )
    params = {"updateMask": mask_str}
    return self._request("PATCH", name, data=dashboard, params=params)

delete_dashboard ¶

delete_dashboard(name)

Deletes a dashboard by name or ID.

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

def delete_dashboard(self, name: str) -> None:
    """Deletes a dashboard by name or ID."""
    if not name.startswith("projects/"):
        name = f"{self.parent}/dashboards/{name}"
    self._request("DELETE", name)

list_charts ¶

list_charts(parent, page_size=100, max_pages=5)

Lists charts within a dashboard.

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

def list_charts(
    self,
    parent: str,
    page_size: int = 100,
    max_pages: int = 5,
) -> list[dict[str, Any]]:
    """Lists charts within a dashboard."""
    if not parent.startswith("projects/"):
        parent = f"{self.parent}/dashboards/{parent}"
    path = f"{parent}/charts"
    params: dict[str, Any] = {"pageSize": page_size}

    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("charts", []))
        page_token = res.get("nextPageToken")
        pages += 1
        if not page_token:
            break
    return results

get_chart ¶

get_chart(name, dashboard_id=None)

Gets a chart by resource name, or chart ID within a dashboard.

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

def get_chart(
    self, name: str, dashboard_id: str | None = None
) -> dict[str, Any]:
    """Gets a chart by resource name, or chart ID within a dashboard."""
    if not name.startswith("projects/"):
        if not dashboard_id:
            raise ValueError(
                "dashboard_id is required when name is a chart ID."
            )
        dashboard_prefix = (
            dashboard_id
            if dashboard_id.startswith("projects/")
            else f"{self.parent}/dashboards/{dashboard_id}"
        )
        name = f"{dashboard_prefix}/charts/{name}"
    return self._request("GET", name)

create_chart ¶

create_chart(parent, chart, chart_id=None)

Creates a new chart under a dashboard parent.

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

def create_chart(
    self,
    parent: str,
    chart: dict[str, Any],
    chart_id: str | None = None,
) -> dict[str, Any]:
    """Creates a new chart under a dashboard parent."""
    if not parent.startswith("projects/"):
        parent = f"{self.parent}/dashboards/{parent}"
    params = {"chartId": chart_id} if chart_id else None
    return self._request(
        "POST",
        f"{parent}/charts",
        data=chart,
        params=params,
    )

update_chart ¶

update_chart(name, chart, update_mask='*', dashboard_id=None)

Updates a chart.

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

def update_chart(
    self,
    name: str,
    chart: dict[str, Any],
    update_mask: str | list[str] = "*",
    dashboard_id: str | None = None,
) -> dict[str, Any]:
    """Updates a chart."""
    if not name.startswith("projects/"):
        if not dashboard_id:
            raise ValueError(
                "dashboard_id is required when name is a chart ID."
            )
        dashboard_prefix = (
            dashboard_id
            if dashboard_id.startswith("projects/")
            else f"{self.parent}/dashboards/{dashboard_id}"
        )
        name = f"{dashboard_prefix}/charts/{name}"
    mask_str = (
        ",".join(update_mask)
        if isinstance(update_mask, (list, tuple))
        else update_mask
    )
    params = {"updateMask": mask_str}
    return self._request("PATCH", name, data=chart, params=params)

delete_chart ¶

delete_chart(name, dashboard_id=None)

Deletes a chart.

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

def delete_chart(self, name: str, dashboard_id: str | None = None) -> None:
    """Deletes a chart."""
    if not name.startswith("projects/"):
        if not dashboard_id:
            raise ValueError(
                "dashboard_id is required when name is a chart ID."
            )
        dashboard_prefix = (
            dashboard_id
            if dashboard_id.startswith("projects/")
            else f"{self.parent}/dashboards/{dashboard_id}"
        )
        name = f"{dashboard_prefix}/charts/{name}"
    self._request("DELETE", name)