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(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(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(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(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(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(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}:analyze", data=payload)
|
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(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")
|
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(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(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)
|
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)
|
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(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(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(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")
|
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(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")
|
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(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(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(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(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)
|