Common
Common is the authentication base class that all other CXAS SCRAPI classes inherit from. It handles GCP credential management and shared utilities, so you don't need to worry about authentication plumbing in subclasses.
You typically won't instantiate Common directly — instead, you'll use classes like Apps, Agents, or Sessions that inherit from it.
Reference
Common(creds_path=None, creds_dict=None, creds=None, scope=None, app_name=None, user_agent_extension=None)
Core Class for managing Auth and shared functions in CX Agent Studio.
Source code in src/cxas_scrapi/core/common.py
| def __init__(
self,
creds_path: str = None,
creds_dict: Dict[str, str] = None,
creds: Any = None,
scope: List[str] = None,
app_name: str = None, # Optional: used to determine client_options
user_agent_extension: str = None,
):
self.scopes = GLOBAL_SCOPES
if scope:
self.scopes += scope
oauth_token = os.environ.get("CXAS_OAUTH_TOKEN")
if creds:
self.creds = creds
if hasattr(self.creds, "refresh"):
try:
self.creds.refresh(Request())
except Exception:
pass
self.token = getattr(self.creds, "token", None)
elif creds_path:
self.creds = service_account.Credentials.from_service_account_file(
creds_path, scopes=self.scopes
)
self.creds.refresh(Request())
self.token = self.creds.token
elif creds_dict:
self.creds = service_account.Credentials.from_service_account_info(
creds_dict, scopes=self.scopes
)
self.creds.refresh(Request())
self.token = self.creds.token
elif oauth_token:
self.creds = Credentials(token=oauth_token)
self.token = oauth_token
else:
self.creds, _ = default(scopes=self.scopes)
self.creds.refresh(Request())
self.token = self.creds.token
self.app_name = app_name
# Calculate standard client options if app_name/resource provided
self.client_options = None
if app_name:
self.client_options = self._get_client_options(app_name)
self.project_id = self._get_project_id(app_name)
self.location = self._get_location(app_name)
else:
self.project_id = None
self.location = None
try:
sdk_version = importlib.metadata.version("cxas-scrapi")
except importlib.metadata.PackageNotFoundError:
sdk_version = "unknown"
self.user_agent = f"cxas-scrapi/{sdk_version}"
if user_agent_extension:
self.user_agent += f":{user_agent_extension}"
self.client_info = ClientInfo(user_agent=self.user_agent)
|
staticmethod
sanitize_expectation_id(text)
Creates a safe ID from the text (MD5 hash).
Source code in src/cxas_scrapi/core/common.py
| @staticmethod
def sanitize_expectation_id(text: str) -> str:
"""Creates a safe ID from the text (MD5 hash)."""
return hashlib.md5(text.encode("utf-8")).hexdigest()
|
staticmethod
get_agent_text_from_outputs(outputs, separator='\n')
Extracts and concatenates text responses from a list of output objects.
Parameters:
| Name | Type | Description | Default |
outputs | List[Any] | A list of output objects (dict or protobuf) from a Session flow execution. | required |
separator | str | String used to join multiple text responses. | '\n' |
Returns:
| Type | Description |
str | A string containing the concatenated text from all outputs. |
Source code in src/cxas_scrapi/core/common.py
| @staticmethod
def get_agent_text_from_outputs(
outputs: List[Any], separator: str = "\n"
) -> str:
"""Extracts and concatenates text responses from a list of output
objects.
Args:
outputs: A list of output objects (dict or protobuf) from a
Session flow execution.
separator: String used to join multiple text responses.
Returns:
A string containing the concatenated text from all outputs.
"""
agent_texts = []
for output in outputs:
if hasattr(output, "text") and getattr(output, "text", ""):
agent_texts.append(output.text)
elif (
isinstance(output, dict) and "text" in output and output["text"]
):
agent_texts.append(output["text"])
return separator.join(agent_texts)
|
get_grpc_transport(client_class)
Creates a customer gRPC transport for CXAS SCRAPI calls.
Source code in src/cxas_scrapi/core/common.py
| def get_grpc_transport(self, client_class: type):
"""Creates a customer gRPC transport for CXAS SCRAPI calls."""
transport_class = client_class.get_transport_class("grpc")
host = "ces.googleapis.com"
client_opts = getattr(self, "client_options", None)
if client_opts and "api_endpoint" in client_opts:
host = self.client_options["api_endpoint"]
channel = transport_class.create_channel(
host=host,
credentials=self.creds,
options=[("grpc.primary_user_agent", self.user_agent)]
)
return transport_class(channel=channel)
|
recurse_proto_repeated_composite(repeated_object)
Recursively converts RepeatedComposite objects to lists.
Source code in src/cxas_scrapi/core/common.py
| def recurse_proto_repeated_composite(self, repeated_object):
"""Recursively converts RepeatedComposite objects to lists."""
repeated_list = []
for item in repeated_object:
if isinstance(item, repeated.RepeatedComposite):
processed_item = self.recurse_proto_repeated_composite(item)
repeated_list.append(processed_item)
elif isinstance(item, maps.MapComposite):
processed_item = self.recurse_proto_marshal_to_dict(item)
repeated_list.append(processed_item)
else:
repeated_list.append(item)
return repeated_list
|
recurse_proto_marshal_to_dict(marshal_object)
Recursively converts MapComposite objects to dicts.
Source code in src/cxas_scrapi/core/common.py
| def recurse_proto_marshal_to_dict(self, marshal_object):
"""Recursively converts MapComposite objects to dicts."""
new_dict = {}
for k, v in marshal_object.items():
processed_v = v
if isinstance(v, maps.MapComposite):
processed_v = self.recurse_proto_marshal_to_dict(v)
elif isinstance(v, repeated.RepeatedComposite):
processed_v = self.recurse_proto_repeated_composite(v)
new_dict[k] = processed_v
return new_dict
|