Deployments
Deployments helps you manage the deployment configurations of a CXAS app. A deployment in CX Agent Studio represents an environment configuration (such as a telephony integration or a channel-specific setup) that determines how your app is published and reachable.
Use this class when you want to list deployments, inspect their configuration, or script deployment updates as part of a release pipeline.
Quick Example
from cxas_scrapi import Deployments
app_name = "projects/my-project/locations/us/apps/my-app-id"
deployments = Deployments(app_name=app_name)
# List all deployments
all_deployments = deployments.list_deployments()
for d in all_deployments:
print(d.display_name, d.name)
# Create a Sessions client pinned to a specific deployment
from cxas_scrapi import Sessions
sessions = Sessions(
app_name=app_name,
deployment_id="my-deployment-id",
)
response = sessions.run(session_id="test-1", text="Hello")
Reference
Deployments(app_name, creds_path=None, creds_dict=None, creds=None, scope=None, **kwargs)
Bases: Apps
Core Class for managing Deployment Resources.
Initializes the Deployments client.
Source code in src/cxas_scrapi/core/deployments.py
| def __init__(
self,
app_name: str,
creds_path: str | None = None,
creds_dict: dict[str, str] | None = None,
creds: Any = None,
scope: list[str] | None = None,
**kwargs,
):
"""Initializes the Deployments client."""
project_id = Common._get_project_id(app_name)
location = Common._get_location(app_name)
if not project_id or not location:
raise ValueError(
f"Invalid app_name format: {app_name}. "
"Expected format: "
"projects/<project>/locations/<location>/apps/<app>"
)
super().__init__(
project_id=project_id,
location=location,
creds_path=creds_path,
creds_dict=creds_dict,
creds=creds,
scope=scope,
**kwargs,
)
self.resource_type = "deployments"
self.app_name = app_name
|
Lists deployments within a specific app.
Source code in src/cxas_scrapi/core/deployments.py
| def list_deployments(self) -> list[types.Deployment]:
"""Lists deployments within a specific app."""
request = types.ListDeploymentsRequest(parent=self.app_name)
response = self.client.list_deployments(request=request)
return list(response)
|
get_deployments_map(reverse=False)
Creates a map of Deployment full names to display names.
Parameters:
| Name | Type | Description | Default |
reverse | bool | If True, map display_name -> name. | False |
Source code in src/cxas_scrapi/core/deployments.py
| def get_deployments_map(self, reverse: bool = False) -> dict[str, str]:
"""Creates a map of Deployment full names to display names.
Args:
reverse: If True, map display_name -> name.
"""
deployments = self.list_deployments()
deployments_dict: dict[str, str] = {}
for deployment in deployments:
display_name = deployment.display_name
name = deployment.name
if display_name and name:
if reverse:
deployments_dict[display_name] = name
else:
deployments_dict[name] = display_name
return deployments_dict
|
get_deployment(deployment_id)
Gets a specific deployment.
Source code in src/cxas_scrapi/core/deployments.py
| def get_deployment(self, deployment_id: str) -> types.Deployment:
"""Gets a specific deployment."""
request = types.GetDeploymentRequest(
name=f"{self.app_name}/deployments/{deployment_id}"
)
return self.client.get_deployment(request=request)
|
create_deployment(deployment_id, display_name, app_version, channel_type=API, modality=None, theme=None, web_widget_title=None, disable_dtmf=False, disable_barge_in_control=False, traffic_split=None)
Creates a new deployment with specified configuration.
Note: modality, theme, and web_widget_title are only applicable when channel_type is ChannelType.WEB_UI.
Source code in src/cxas_scrapi/core/deployments.py
| def create_deployment(
self,
deployment_id: str,
display_name: str,
app_version: str,
channel_type: ChannelType | str = ChannelType.API,
modality: Modality | str | None = None,
theme: Theme | str | None = None,
web_widget_title: str | None = None,
disable_dtmf: bool = False,
disable_barge_in_control: bool = False,
traffic_split: dict[str, int] | None = None,
) -> types.Deployment:
"""Creates a new deployment with specified configuration.
Note: `modality`, `theme`, and `web_widget_title` are only applicable
when `channel_type` is `ChannelType.WEB_UI`.
"""
if app_version and not app_version.startswith("projects/"):
app_version = f"{self.app_name}/versions/{app_version}"
deployment = types.Deployment(
display_name=display_name, app_version=app_version
)
# Convert string to enum if needed
if isinstance(channel_type, str):
channel_type = self.ChannelType[channel_type.upper()]
channel_profile = types.ChannelProfile()
channel_profile.channel_type = getattr(
types.common.ChannelProfile.ChannelType, channel_type.value
)
channel_profile.disable_dtmf = disable_dtmf
channel_profile.disable_barge_in_control = disable_barge_in_control
if channel_type == self.ChannelType.WEB_UI:
wwc_kwargs = {
"modality": modality or self.Modality.CHAT_AND_VOICE,
"theme": theme or self.Theme.LIGHT,
}
if web_widget_title:
wwc_kwargs["web_widget_title"] = web_widget_title
wwc = self._build_web_widget_config(wwc_kwargs)
if wwc:
channel_profile.web_widget_config = wwc
deployment.channel_profile = channel_profile
if traffic_split:
if len(traffic_split) < 2:
raise ValueError(
"Traffic split requires at least two versions."
)
versions_client = Versions(app_name=self.app_name, creds=self.creds)
existing_versions = versions_client.list_versions()
existing_version_names = [v.name for v in existing_versions]
experiment_config = types.ExperimentConfig()
version_release = types.ExperimentConfig.VersionRelease()
version_release.state = types.ExperimentConfig.State.RUNNING
for version, split in traffic_split.items():
v_name = version
if not v_name.startswith("projects/"):
v_name = f"{self.app_name}/versions/{version}"
if v_name not in existing_version_names:
raise ValueError(
f"Version {v_name} does not exist. Valid versions: "
f"{[v.split('/')[-1] for v in existing_version_names]}"
)
allocation = (
types.ExperimentConfig.VersionRelease.TrafficAllocation()
)
allocation.app_version = v_name
allocation.traffic_percentage = split
version_release.traffic_allocations.append(allocation)
experiment_config.version_release = version_release
deployment.experiment_config = experiment_config
request = types.CreateDeploymentRequest(
parent=self.app_name,
deployment_id=deployment_id,
deployment=deployment,
)
return self.client.create_deployment(request=request)
|
update_deployment(deployment_id, **kwargs)
Updates specific fields of an existing Deployment.
Source code in src/cxas_scrapi/core/deployments.py
| def update_deployment(
self, deployment_id: str, **kwargs
) -> types.Deployment:
"""Updates specific fields of an existing Deployment."""
deployment = types.Deployment(
name=f"{self.app_name}/deployments/{deployment_id}"
)
mask_paths = []
channel_profile_fields = [
"channel_type",
"modality",
"theme",
"web_widget_title",
"disable_dtmf",
"disable_barge_in_control",
]
has_channel_profile_update = any(
k in kwargs for k in channel_profile_fields
)
if has_channel_profile_update:
channel_profile = types.ChannelProfile()
if "channel_type" in kwargs:
channel_type = kwargs.pop("channel_type")
if isinstance(channel_type, str):
channel_type = self.ChannelType[channel_type.upper()]
channel_profile.channel_type = getattr(
types.common.ChannelProfile.ChannelType, channel_type.value
)
mask_paths.append("channel_profile.channel_type")
if "disable_dtmf" in kwargs:
channel_profile.disable_dtmf = kwargs.pop("disable_dtmf")
mask_paths.append("channel_profile.disable_dtmf")
if "disable_barge_in_control" in kwargs:
channel_profile.disable_barge_in_control = kwargs.pop(
"disable_barge_in_control"
)
mask_paths.append("channel_profile.disable_barge_in_control")
wwc = self._build_web_widget_config(kwargs, mask_paths)
if wwc:
channel_profile.web_widget_config = wwc
deployment.channel_profile = channel_profile
if "traffic_split" in kwargs:
traffic_split = kwargs.pop("traffic_split")
if len(traffic_split) < 2:
raise ValueError(
"Traffic split requires at least two versions."
)
versions_client = Versions(app_name=self.app_name, creds=self.creds)
existing_versions = versions_client.list_versions()
existing_version_names = [v.name for v in existing_versions]
experiment_config = types.ExperimentConfig()
version_release = types.ExperimentConfig.VersionRelease()
version_release.state = types.ExperimentConfig.State.RUNNING
for version, split in traffic_split.items():
v_name = version
if not v_name.startswith("projects/"):
v_name = f"{self.app_name}/versions/{version}"
if v_name not in existing_version_names:
raise ValueError(
f"Version {v_name} does not exist. Valid versions: "
f"{[v.split('/')[-1] for v in existing_version_names]}"
)
allocation = (
types.ExperimentConfig.VersionRelease.TrafficAllocation()
)
allocation.app_version = v_name
allocation.traffic_percentage = split
version_release.traffic_allocations.append(allocation)
experiment_config.version_release = version_release
deployment.experiment_config = experiment_config
mask_paths.append("experiment_config")
elif "app_version" in kwargs:
# If promoting a new version without a traffic split,
# clear any existing experiment
deployment.experiment_config = types.ExperimentConfig()
mask_paths.append("experiment_config")
# Handle remaining kwargs as top-level fields
for key, value in kwargs.items():
val_to_set = value
is_app_ver = key == "app_version"
if is_app_ver and value and not value.startswith("projects/"):
val_to_set = f"{self.app_name}/versions/{value}"
setattr(deployment, key, val_to_set)
mask_paths.append(key)
request = types.UpdateDeploymentRequest(
deployment=deployment,
update_mask=field_mask_pb2.FieldMask(paths=mask_paths),
)
return self.client.update_deployment(request=request)
|
delete_deployment(deployment_id)
Deletes a specific deployment.
Source code in src/cxas_scrapi/core/deployments.py
| def delete_deployment(self, deployment_id: str) -> None:
"""Deletes a specific deployment."""
request = types.DeleteDeploymentRequest(
name=f"{self.app_name}/deployments/{deployment_id}"
)
self.client.delete_deployment(request=request)
|