diff --git a/tests/unit/vertexai/genai/replays/test_ae_sandbox_snapshots_create.py b/tests/unit/vertexai/genai/replays/test_ae_sandbox_snapshots_create.py new file mode 100644 index 0000000000..f2f1a213ed --- /dev/null +++ b/tests/unit/vertexai/genai/replays/test_ae_sandbox_snapshots_create.py @@ -0,0 +1,40 @@ +"""Tests for Sandbox Snapshot create operation.""" + +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +# pylint: disable=protected-access,bad-continuation,missing-function-docstring,bad-indentation + +from tests.unit.vertexai.genai.replays import pytest_helper +from vertexai._genai import types + + +def test_create_sandbox_snapshot(client): + snapshot = client.agent_engines.sandboxes.snapshots.create( + source_sandbox_environment_name="projects/802583348448/locations/us-central1/reasoningEngines/6130241318758121472/sandboxEnvironments/525190525100228608", + config={ + "display_name": "test_snapshot", + "ttl": "3600s", + "owner": "test_owner", + }, + ) + + assert isinstance(snapshot, types.AgentEngineSandboxSnapshotOperation) + + +pytestmark = pytest_helper.setup( + file=__file__, + globals_for_file=globals(), + test_method="agent_engines.sandboxes.snapshots.create", +) diff --git a/tests/unit/vertexai/genai/replays/test_ae_sandbox_snapshots_delete.py b/tests/unit/vertexai/genai/replays/test_ae_sandbox_snapshots_delete.py new file mode 100644 index 0000000000..d5407ac89f --- /dev/null +++ b/tests/unit/vertexai/genai/replays/test_ae_sandbox_snapshots_delete.py @@ -0,0 +1,35 @@ +"""Tests for Sandbox Snapshot delete operation.""" + +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +# pylint: disable=protected-access,bad-continuation,missing-function-docstring,bad-indentation + +from tests.unit.vertexai.genai.replays import pytest_helper +from vertexai._genai import types + + +def test_delete_sandbox_snapshot(client): + result = client.agent_engines.sandboxes.snapshots.delete( + name="projects/802583348448/locations/us-central1/reasoningEngines/6130241318758121472/sandboxEnvironmentSnapshots/421086565159141376", + ) + + assert isinstance(result, types.DeleteSandboxEnvironmentSnapshotOperation) + + +pytestmark = pytest_helper.setup( + file=__file__, + globals_for_file=globals(), + test_method="agent_engines.sandboxes.snapshots.delete", +) diff --git a/tests/unit/vertexai/genai/replays/test_ae_sandbox_snapshots_get.py b/tests/unit/vertexai/genai/replays/test_ae_sandbox_snapshots_get.py new file mode 100644 index 0000000000..9340fcbaab --- /dev/null +++ b/tests/unit/vertexai/genai/replays/test_ae_sandbox_snapshots_get.py @@ -0,0 +1,37 @@ +"""Tests for Sandbox Snapshot get operation.""" + +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +# pylint: disable=protected-access,bad-continuation,missing-function-docstring,bad-indentation + +from tests.unit.vertexai.genai.replays import pytest_helper +from vertexai._genai import types + + +def test_get_sandbox_snapshot(client): + snapshot_name = "projects/802583348448/locations/us-central1/reasoningEngines/6130241318758121472/sandboxEnvironmentSnapshots/421086565159141376" + fetched_snapshot = client.agent_engines.sandboxes.snapshots.get( + name=snapshot_name, + ) + + assert isinstance(fetched_snapshot, types.SandboxEnvironmentSnapshot) + assert fetched_snapshot.name == snapshot_name + + +pytestmark = pytest_helper.setup( + file=__file__, + globals_for_file=globals(), + test_method="agent_engines.sandboxes.snapshots.get", +) diff --git a/tests/unit/vertexai/genai/replays/test_ae_sandbox_snapshots_list.py b/tests/unit/vertexai/genai/replays/test_ae_sandbox_snapshots_list.py new file mode 100644 index 0000000000..a71a17b8cf --- /dev/null +++ b/tests/unit/vertexai/genai/replays/test_ae_sandbox_snapshots_list.py @@ -0,0 +1,41 @@ +"""Tests for Sandbox Snapshot list operation.""" + +# Copyright 2026 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# +# pylint: disable=protected-access,bad-continuation,missing-function-docstring,bad-indentation + +from tests.unit.vertexai.genai.replays import pytest_helper +from vertexai._genai import types + + +def test_list_sandbox_snapshots(client): + snapshots_list_operation = client.agent_engines.sandboxes.snapshots.list( + name="projects/802583348448/locations/us-central1/reasoningEngines/6130241318758121472", + ) + + assert isinstance( + snapshots_list_operation.sandbox_environment_snapshots[0], + types.SandboxEnvironmentSnapshot, + ) + assert isinstance( + snapshots_list_operation, types.ListSandboxEnvironmentSnapshotsResponse + ) + + +pytestmark = pytest_helper.setup( + file=__file__, + globals_for_file=globals(), + test_method="agent_engines.sandboxes.snapshots.list", +) diff --git a/vertexai/_genai/sandbox_snapshots.py b/vertexai/_genai/sandbox_snapshots.py new file mode 100644 index 0000000000..81595e7ae1 --- /dev/null +++ b/vertexai/_genai/sandbox_snapshots.py @@ -0,0 +1,899 @@ +# Copyright 2025 Google LLC +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +# Code generated by the Google Gen AI SDK generator DO NOT EDIT. + +import json +import logging +from typing import Any, Optional, Union +from urllib.parse import urlencode + +from google.genai import _api_module +from google.genai import _common +from google.genai._common import get_value_by_path as getv +from google.genai._common import set_value_by_path as setv + +from . import types + +logger = logging.getLogger("vertexai_genai.sandboxsnapshots") + + +def _CreateAgentEngineSandboxSnapshotConfig_to_vertex( + from_object: Union[dict[str, Any], object], + parent_object: Optional[dict[str, Any]] = None, +) -> dict[str, Any]: + to_object: dict[str, Any] = {} + + if getv(from_object, ["display_name"]) is not None: + setv(parent_object, ["displayName"], getv(from_object, ["display_name"])) + + if getv(from_object, ["owner"]) is not None: + setv(parent_object, ["owner"], getv(from_object, ["owner"])) + + if getv(from_object, ["ttl"]) is not None: + setv(parent_object, ["ttl"], getv(from_object, ["ttl"])) + + return to_object + + +def _CreateSandboxEnvironmentSnapshotRequestParameters_to_vertex( + from_object: Union[dict[str, Any], object], + parent_object: Optional[dict[str, Any]] = None, +) -> dict[str, Any]: + to_object: dict[str, Any] = {} + if getv(from_object, ["source_sandbox_environment_name"]) is not None: + setv( + to_object, + ["_url", "name"], + getv(from_object, ["source_sandbox_environment_name"]), + ) + + if getv(from_object, ["config"]) is not None: + setv( + to_object, + ["config"], + _CreateAgentEngineSandboxSnapshotConfig_to_vertex( + getv(from_object, ["config"]), to_object + ), + ) + + return to_object + + +def _DeleteSandboxEnvironmentSnapshotRequestParameters_to_vertex( + from_object: Union[dict[str, Any], object], + parent_object: Optional[dict[str, Any]] = None, +) -> dict[str, Any]: + to_object: dict[str, Any] = {} + if getv(from_object, ["name"]) is not None: + setv(to_object, ["_url", "name"], getv(from_object, ["name"])) + + return to_object + + +def _GetAgentEngineSandboxSnapshotOperationParameters_to_vertex( + from_object: Union[dict[str, Any], object], + parent_object: Optional[dict[str, Any]] = None, +) -> dict[str, Any]: + to_object: dict[str, Any] = {} + if getv(from_object, ["operation_name"]) is not None: + setv( + to_object, ["_url", "operationName"], getv(from_object, ["operation_name"]) + ) + + return to_object + + +def _GetSandboxEnvironmentSnapshotRequestParameters_to_vertex( + from_object: Union[dict[str, Any], object], + parent_object: Optional[dict[str, Any]] = None, +) -> dict[str, Any]: + to_object: dict[str, Any] = {} + if getv(from_object, ["name"]) is not None: + setv(to_object, ["_url", "name"], getv(from_object, ["name"])) + + return to_object + + +def _ListSandboxEnvironmentSnapshotsConfig_to_vertex( + from_object: Union[dict[str, Any], object], + parent_object: Optional[dict[str, Any]] = None, +) -> dict[str, Any]: + to_object: dict[str, Any] = {} + + if getv(from_object, ["page_size"]) is not None: + setv(parent_object, ["_query", "pageSize"], getv(from_object, ["page_size"])) + + if getv(from_object, ["page_token"]) is not None: + setv(parent_object, ["_query", "pageToken"], getv(from_object, ["page_token"])) + + if getv(from_object, ["filter"]) is not None: + setv(parent_object, ["_query", "filter"], getv(from_object, ["filter"])) + + return to_object + + +def _ListSandboxEnvironmentSnapshotsRequestParameters_to_vertex( + from_object: Union[dict[str, Any], object], + parent_object: Optional[dict[str, Any]] = None, +) -> dict[str, Any]: + to_object: dict[str, Any] = {} + if getv(from_object, ["name"]) is not None: + setv(to_object, ["_url", "name"], getv(from_object, ["name"])) + + if getv(from_object, ["config"]) is not None: + _ListSandboxEnvironmentSnapshotsConfig_to_vertex( + getv(from_object, ["config"]), to_object + ) + + return to_object + + +class SandboxSnapshots(_api_module.BaseModule): + """Sandbox environment snapshot commands.""" + + def create( + self, + *, + source_sandbox_environment_name: str, + config: Optional[types.CreateAgentEngineSandboxSnapshotConfigOrDict] = None, + ) -> types.AgentEngineSandboxSnapshotOperation: + """ + Snapshots a sandbox environment. + + """ + + parameter_model = types._CreateSandboxEnvironmentSnapshotRequestParameters( + source_sandbox_environment_name=source_sandbox_environment_name, + config=config, + ) + + request_url_dict: Optional[dict[str, str]] + if not self._api_client.vertexai: + raise ValueError( + "This method is only supported in the Gemini Enterprise Agent Platform (previously known as Vertex AI) client." + ) + else: + request_dict = _CreateSandboxEnvironmentSnapshotRequestParameters_to_vertex( + parameter_model + ) + request_url_dict = request_dict.get("_url") + if request_url_dict: + path = "{name}:snapshot".format_map(request_url_dict) + else: + path = "{name}:snapshot" + + query_params = request_dict.get("_query") + if query_params: + path = f"{path}?{urlencode(query_params)}" + # TODO: remove the hack that pops config. + request_dict.pop("config", None) + + http_options: Optional[types.HttpOptions] = None + if ( + parameter_model.config is not None + and parameter_model.config.http_options is not None + ): + http_options = parameter_model.config.http_options + + request_dict = _common.convert_to_dict(request_dict) + request_dict = _common.encode_unserializable_types(request_dict) + + response = self._api_client.request("post", path, request_dict, http_options) + + response_dict = {} if not response.body else json.loads(response.body) + + return_value = types.AgentEngineSandboxSnapshotOperation._from_response( + response=response_dict, + kwargs=( + { + "config": { + "response_schema": getattr( + parameter_model.config, "response_schema", None + ), + "response_json_schema": getattr( + parameter_model.config, "response_json_schema", None + ), + "include_all_fields": getattr( + parameter_model.config, "include_all_fields", None + ), + } + } + if getattr(parameter_model, "config", None) + else {} + ), + ) + + self._api_client._verify_response(return_value) + return return_value + + def delete( + self, + *, + name: str, + config: Optional[types.DeleteSandboxEnvironmentSnapshotConfigOrDict] = None, + ) -> types.DeleteSandboxEnvironmentSnapshotOperation: + """ + Deletes a sandbox environment snapshot. + + """ + + parameter_model = types._DeleteSandboxEnvironmentSnapshotRequestParameters( + name=name, + config=config, + ) + + request_url_dict: Optional[dict[str, str]] + if not self._api_client.vertexai: + raise ValueError( + "This method is only supported in the Gemini Enterprise Agent Platform (previously known as Vertex AI) client." + ) + else: + request_dict = _DeleteSandboxEnvironmentSnapshotRequestParameters_to_vertex( + parameter_model + ) + request_url_dict = request_dict.get("_url") + if request_url_dict: + path = "{name}".format_map(request_url_dict) + else: + path = "{name}" + + query_params = request_dict.get("_query") + if query_params: + path = f"{path}?{urlencode(query_params)}" + # TODO: remove the hack that pops config. + request_dict.pop("config", None) + + http_options: Optional[types.HttpOptions] = None + if ( + parameter_model.config is not None + and parameter_model.config.http_options is not None + ): + http_options = parameter_model.config.http_options + + request_dict = _common.convert_to_dict(request_dict) + request_dict = _common.encode_unserializable_types(request_dict) + + response = self._api_client.request("delete", path, request_dict, http_options) + + response_dict = {} if not response.body else json.loads(response.body) + + return_value = types.DeleteSandboxEnvironmentSnapshotOperation._from_response( + response=response_dict, + kwargs=( + { + "config": { + "response_schema": getattr( + parameter_model.config, "response_schema", None + ), + "response_json_schema": getattr( + parameter_model.config, "response_json_schema", None + ), + "include_all_fields": getattr( + parameter_model.config, "include_all_fields", None + ), + } + } + if getattr(parameter_model, "config", None) + else {} + ), + ) + + self._api_client._verify_response(return_value) + return return_value + + def get( + self, + *, + name: str, + config: Optional[types.GetSandboxEnvironmentSnapshotConfigOrDict] = None, + ) -> types.SandboxEnvironmentSnapshot: + """ + Gets a sandbox environment snapshot. + + """ + + parameter_model = types._GetSandboxEnvironmentSnapshotRequestParameters( + name=name, + config=config, + ) + + request_url_dict: Optional[dict[str, str]] + if not self._api_client.vertexai: + raise ValueError( + "This method is only supported in the Gemini Enterprise Agent Platform (previously known as Vertex AI) client." + ) + else: + request_dict = _GetSandboxEnvironmentSnapshotRequestParameters_to_vertex( + parameter_model + ) + request_url_dict = request_dict.get("_url") + if request_url_dict: + path = "{name}".format_map(request_url_dict) + else: + path = "{name}" + + query_params = request_dict.get("_query") + if query_params: + path = f"{path}?{urlencode(query_params)}" + # TODO: remove the hack that pops config. + request_dict.pop("config", None) + + http_options: Optional[types.HttpOptions] = None + if ( + parameter_model.config is not None + and parameter_model.config.http_options is not None + ): + http_options = parameter_model.config.http_options + + request_dict = _common.convert_to_dict(request_dict) + request_dict = _common.encode_unserializable_types(request_dict) + + response = self._api_client.request("get", path, request_dict, http_options) + + response_dict = {} if not response.body else json.loads(response.body) + + return_value = types.SandboxEnvironmentSnapshot._from_response( + response=response_dict, + kwargs=( + { + "config": { + "response_schema": getattr( + parameter_model.config, "response_schema", None + ), + "response_json_schema": getattr( + parameter_model.config, "response_json_schema", None + ), + "include_all_fields": getattr( + parameter_model.config, "include_all_fields", None + ), + } + } + if getattr(parameter_model, "config", None) + else {} + ), + ) + + self._api_client._verify_response(return_value) + return return_value + + def list( + self, + *, + name: str, + config: Optional[types.ListSandboxEnvironmentSnapshotsConfigOrDict] = None, + ) -> types.ListSandboxEnvironmentSnapshotsResponse: + """ + Lists sandbox environment snapshots. + + """ + + parameter_model = types._ListSandboxEnvironmentSnapshotsRequestParameters( + name=name, + config=config, + ) + + request_url_dict: Optional[dict[str, str]] + if not self._api_client.vertexai: + raise ValueError( + "This method is only supported in the Gemini Enterprise Agent Platform (previously known as Vertex AI) client." + ) + else: + request_dict = _ListSandboxEnvironmentSnapshotsRequestParameters_to_vertex( + parameter_model + ) + request_url_dict = request_dict.get("_url") + if request_url_dict: + path = "{name}/sandboxEnvironmentSnapshots".format_map(request_url_dict) + else: + path = "{name}/sandboxEnvironmentSnapshots" + + query_params = request_dict.get("_query") + if query_params: + path = f"{path}?{urlencode(query_params)}" + # TODO: remove the hack that pops config. + request_dict.pop("config", None) + + http_options: Optional[types.HttpOptions] = None + if ( + parameter_model.config is not None + and parameter_model.config.http_options is not None + ): + http_options = parameter_model.config.http_options + + request_dict = _common.convert_to_dict(request_dict) + request_dict = _common.encode_unserializable_types(request_dict) + + response = self._api_client.request("get", path, request_dict, http_options) + + response_dict = {} if not response.body else json.loads(response.body) + + return_value = types.ListSandboxEnvironmentSnapshotsResponse._from_response( + response=response_dict, + kwargs=( + { + "config": { + "response_schema": getattr( + parameter_model.config, "response_schema", None + ), + "response_json_schema": getattr( + parameter_model.config, "response_json_schema", None + ), + "include_all_fields": getattr( + parameter_model.config, "include_all_fields", None + ), + } + } + if getattr(parameter_model, "config", None) + else {} + ), + ) + + self._api_client._verify_response(return_value) + return return_value + + def _get_sandbox_snapshot_operation( + self, + *, + operation_name: str, + config: Optional[types.GetAgentEngineOperationConfigOrDict] = None, + ) -> types.AgentEngineSandboxSnapshotOperation: + parameter_model = types._GetAgentEngineSandboxSnapshotOperationParameters( + operation_name=operation_name, + config=config, + ) + + request_url_dict: Optional[dict[str, str]] + if not self._api_client.vertexai: + raise ValueError( + "This method is only supported in the Gemini Enterprise Agent Platform (previously known as Vertex AI) client." + ) + else: + request_dict = _GetAgentEngineSandboxSnapshotOperationParameters_to_vertex( + parameter_model + ) + request_url_dict = request_dict.get("_url") + if request_url_dict: + path = "{operationName}".format_map(request_url_dict) + else: + path = "{operationName}" + + query_params = request_dict.get("_query") + if query_params: + path = f"{path}?{urlencode(query_params)}" + # TODO: remove the hack that pops config. + request_dict.pop("config", None) + + http_options: Optional[types.HttpOptions] = None + if ( + parameter_model.config is not None + and parameter_model.config.http_options is not None + ): + http_options = parameter_model.config.http_options + + request_dict = _common.convert_to_dict(request_dict) + request_dict = _common.encode_unserializable_types(request_dict) + + response = self._api_client.request("get", path, request_dict, http_options) + + response_dict = {} if not response.body else json.loads(response.body) + + return_value = types.AgentEngineSandboxSnapshotOperation._from_response( + response=response_dict, + kwargs=( + { + "config": { + "response_schema": getattr( + parameter_model.config, "response_schema", None + ), + "response_json_schema": getattr( + parameter_model.config, "response_json_schema", None + ), + "include_all_fields": getattr( + parameter_model.config, "include_all_fields", None + ), + } + } + if getattr(parameter_model, "config", None) + else {} + ), + ) + + self._api_client._verify_response(return_value) + return return_value + + +class AsyncSandboxSnapshots(_api_module.BaseModule): + """Sandbox environment snapshot commands.""" + + async def create( + self, + *, + source_sandbox_environment_name: str, + config: Optional[types.CreateAgentEngineSandboxSnapshotConfigOrDict] = None, + ) -> types.AgentEngineSandboxSnapshotOperation: + """ + Snapshots a sandbox environment. + + """ + + parameter_model = types._CreateSandboxEnvironmentSnapshotRequestParameters( + source_sandbox_environment_name=source_sandbox_environment_name, + config=config, + ) + + request_url_dict: Optional[dict[str, str]] + if not self._api_client.vertexai: + raise ValueError( + "This method is only supported in the Gemini Enterprise Agent Platform (previously known as Vertex AI) client." + ) + else: + request_dict = _CreateSandboxEnvironmentSnapshotRequestParameters_to_vertex( + parameter_model + ) + request_url_dict = request_dict.get("_url") + if request_url_dict: + path = "{name}:snapshot".format_map(request_url_dict) + else: + path = "{name}:snapshot" + + query_params = request_dict.get("_query") + if query_params: + path = f"{path}?{urlencode(query_params)}" + # TODO: remove the hack that pops config. + request_dict.pop("config", None) + + http_options: Optional[types.HttpOptions] = None + if ( + parameter_model.config is not None + and parameter_model.config.http_options is not None + ): + http_options = parameter_model.config.http_options + + request_dict = _common.convert_to_dict(request_dict) + request_dict = _common.encode_unserializable_types(request_dict) + + response = await self._api_client.async_request( + "post", path, request_dict, http_options + ) + + response_dict = {} if not response.body else json.loads(response.body) + + return_value = types.AgentEngineSandboxSnapshotOperation._from_response( + response=response_dict, + kwargs=( + { + "config": { + "response_schema": getattr( + parameter_model.config, "response_schema", None + ), + "response_json_schema": getattr( + parameter_model.config, "response_json_schema", None + ), + "include_all_fields": getattr( + parameter_model.config, "include_all_fields", None + ), + } + } + if getattr(parameter_model, "config", None) + else {} + ), + ) + + self._api_client._verify_response(return_value) + return return_value + + async def delete( + self, + *, + name: str, + config: Optional[types.DeleteSandboxEnvironmentSnapshotConfigOrDict] = None, + ) -> types.DeleteSandboxEnvironmentSnapshotOperation: + """ + Deletes a sandbox environment snapshot. + + """ + + parameter_model = types._DeleteSandboxEnvironmentSnapshotRequestParameters( + name=name, + config=config, + ) + + request_url_dict: Optional[dict[str, str]] + if not self._api_client.vertexai: + raise ValueError( + "This method is only supported in the Gemini Enterprise Agent Platform (previously known as Vertex AI) client." + ) + else: + request_dict = _DeleteSandboxEnvironmentSnapshotRequestParameters_to_vertex( + parameter_model + ) + request_url_dict = request_dict.get("_url") + if request_url_dict: + path = "{name}".format_map(request_url_dict) + else: + path = "{name}" + + query_params = request_dict.get("_query") + if query_params: + path = f"{path}?{urlencode(query_params)}" + # TODO: remove the hack that pops config. + request_dict.pop("config", None) + + http_options: Optional[types.HttpOptions] = None + if ( + parameter_model.config is not None + and parameter_model.config.http_options is not None + ): + http_options = parameter_model.config.http_options + + request_dict = _common.convert_to_dict(request_dict) + request_dict = _common.encode_unserializable_types(request_dict) + + response = await self._api_client.async_request( + "delete", path, request_dict, http_options + ) + + response_dict = {} if not response.body else json.loads(response.body) + + return_value = types.DeleteSandboxEnvironmentSnapshotOperation._from_response( + response=response_dict, + kwargs=( + { + "config": { + "response_schema": getattr( + parameter_model.config, "response_schema", None + ), + "response_json_schema": getattr( + parameter_model.config, "response_json_schema", None + ), + "include_all_fields": getattr( + parameter_model.config, "include_all_fields", None + ), + } + } + if getattr(parameter_model, "config", None) + else {} + ), + ) + + self._api_client._verify_response(return_value) + return return_value + + async def get( + self, + *, + name: str, + config: Optional[types.GetSandboxEnvironmentSnapshotConfigOrDict] = None, + ) -> types.SandboxEnvironmentSnapshot: + """ + Gets a sandbox environment snapshot. + + """ + + parameter_model = types._GetSandboxEnvironmentSnapshotRequestParameters( + name=name, + config=config, + ) + + request_url_dict: Optional[dict[str, str]] + if not self._api_client.vertexai: + raise ValueError( + "This method is only supported in the Gemini Enterprise Agent Platform (previously known as Vertex AI) client." + ) + else: + request_dict = _GetSandboxEnvironmentSnapshotRequestParameters_to_vertex( + parameter_model + ) + request_url_dict = request_dict.get("_url") + if request_url_dict: + path = "{name}".format_map(request_url_dict) + else: + path = "{name}" + + query_params = request_dict.get("_query") + if query_params: + path = f"{path}?{urlencode(query_params)}" + # TODO: remove the hack that pops config. + request_dict.pop("config", None) + + http_options: Optional[types.HttpOptions] = None + if ( + parameter_model.config is not None + and parameter_model.config.http_options is not None + ): + http_options = parameter_model.config.http_options + + request_dict = _common.convert_to_dict(request_dict) + request_dict = _common.encode_unserializable_types(request_dict) + + response = await self._api_client.async_request( + "get", path, request_dict, http_options + ) + + response_dict = {} if not response.body else json.loads(response.body) + + return_value = types.SandboxEnvironmentSnapshot._from_response( + response=response_dict, + kwargs=( + { + "config": { + "response_schema": getattr( + parameter_model.config, "response_schema", None + ), + "response_json_schema": getattr( + parameter_model.config, "response_json_schema", None + ), + "include_all_fields": getattr( + parameter_model.config, "include_all_fields", None + ), + } + } + if getattr(parameter_model, "config", None) + else {} + ), + ) + + self._api_client._verify_response(return_value) + return return_value + + async def list( + self, + *, + name: str, + config: Optional[types.ListSandboxEnvironmentSnapshotsConfigOrDict] = None, + ) -> types.ListSandboxEnvironmentSnapshotsResponse: + """ + Lists sandbox environment snapshots. + + """ + + parameter_model = types._ListSandboxEnvironmentSnapshotsRequestParameters( + name=name, + config=config, + ) + + request_url_dict: Optional[dict[str, str]] + if not self._api_client.vertexai: + raise ValueError( + "This method is only supported in the Gemini Enterprise Agent Platform (previously known as Vertex AI) client." + ) + else: + request_dict = _ListSandboxEnvironmentSnapshotsRequestParameters_to_vertex( + parameter_model + ) + request_url_dict = request_dict.get("_url") + if request_url_dict: + path = "{name}/sandboxEnvironmentSnapshots".format_map(request_url_dict) + else: + path = "{name}/sandboxEnvironmentSnapshots" + + query_params = request_dict.get("_query") + if query_params: + path = f"{path}?{urlencode(query_params)}" + # TODO: remove the hack that pops config. + request_dict.pop("config", None) + + http_options: Optional[types.HttpOptions] = None + if ( + parameter_model.config is not None + and parameter_model.config.http_options is not None + ): + http_options = parameter_model.config.http_options + + request_dict = _common.convert_to_dict(request_dict) + request_dict = _common.encode_unserializable_types(request_dict) + + response = await self._api_client.async_request( + "get", path, request_dict, http_options + ) + + response_dict = {} if not response.body else json.loads(response.body) + + return_value = types.ListSandboxEnvironmentSnapshotsResponse._from_response( + response=response_dict, + kwargs=( + { + "config": { + "response_schema": getattr( + parameter_model.config, "response_schema", None + ), + "response_json_schema": getattr( + parameter_model.config, "response_json_schema", None + ), + "include_all_fields": getattr( + parameter_model.config, "include_all_fields", None + ), + } + } + if getattr(parameter_model, "config", None) + else {} + ), + ) + + self._api_client._verify_response(return_value) + return return_value + + async def _get_sandbox_snapshot_operation( + self, + *, + operation_name: str, + config: Optional[types.GetAgentEngineOperationConfigOrDict] = None, + ) -> types.AgentEngineSandboxSnapshotOperation: + parameter_model = types._GetAgentEngineSandboxSnapshotOperationParameters( + operation_name=operation_name, + config=config, + ) + + request_url_dict: Optional[dict[str, str]] + if not self._api_client.vertexai: + raise ValueError( + "This method is only supported in the Gemini Enterprise Agent Platform (previously known as Vertex AI) client." + ) + else: + request_dict = _GetAgentEngineSandboxSnapshotOperationParameters_to_vertex( + parameter_model + ) + request_url_dict = request_dict.get("_url") + if request_url_dict: + path = "{operationName}".format_map(request_url_dict) + else: + path = "{operationName}" + + query_params = request_dict.get("_query") + if query_params: + path = f"{path}?{urlencode(query_params)}" + # TODO: remove the hack that pops config. + request_dict.pop("config", None) + + http_options: Optional[types.HttpOptions] = None + if ( + parameter_model.config is not None + and parameter_model.config.http_options is not None + ): + http_options = parameter_model.config.http_options + + request_dict = _common.convert_to_dict(request_dict) + request_dict = _common.encode_unserializable_types(request_dict) + + response = await self._api_client.async_request( + "get", path, request_dict, http_options + ) + + response_dict = {} if not response.body else json.loads(response.body) + + return_value = types.AgentEngineSandboxSnapshotOperation._from_response( + response=response_dict, + kwargs=( + { + "config": { + "response_schema": getattr( + parameter_model.config, "response_schema", None + ), + "response_json_schema": getattr( + parameter_model.config, "response_json_schema", None + ), + "include_all_fields": getattr( + parameter_model.config, "include_all_fields", None + ), + } + } + if getattr(parameter_model, "config", None) + else {} + ), + ) + + self._api_client._verify_response(return_value) + return return_value diff --git a/vertexai/_genai/sandboxes.py b/vertexai/_genai/sandboxes.py index fbd7cc0a15..9edc2cd9be 100644 --- a/vertexai/_genai/sandboxes.py +++ b/vertexai/_genai/sandboxes.py @@ -630,6 +630,7 @@ def _get_sandbox_operation( return return_value _templates = None + _snapshots = None @property def templates(self) -> Any: @@ -640,12 +641,27 @@ def templates(self) -> Any: ) except ImportError as e: raise ImportError( - "The 'agent_engines.sandboxes.sandbox_templates' module requires " + "The 'agent_engines.sandboxes.templates' module requires " "additional packages. Please install them using pip install " "google-cloud-aiplatform[agent_engines]" ) from e return self._templates.SandboxTemplates(self._api_client) + @property + def snapshots(self) -> Any: + if self._snapshots is None: + try: + self._snapshots = __import__("importlib").import_module( + ".sandbox_snapshots", __package__ + ) + except ImportError as e: + raise ImportError( + "The 'agent_engines.sandboxes.snapshots' module requires " + "additional packages. Please install them using pip install " + "google-cloud-aiplatform[sandbox_snapshots]" + ) from e + return self._snapshots.SandboxSnapshots(self._api_client) + def create( self, *, diff --git a/vertexai/_genai/types/__init__.py b/vertexai/_genai/types/__init__.py index 1ce1cfeb62..6d08a89068 100644 --- a/vertexai/_genai/types/__init__.py +++ b/vertexai/_genai/types/__init__.py @@ -40,6 +40,7 @@ from .common import _CreateEvaluationRunParameters from .common import _CreateEvaluationSetParameters from .common import _CreateMultimodalDatasetParameters +from .common import _CreateSandboxEnvironmentSnapshotRequestParameters from .common import _CreateSandboxEnvironmentTemplateRequestParameters from .common import _CustomJobParameters from .common import _CustomJobParameters @@ -52,6 +53,7 @@ from .common import _DeleteEvaluationMetricParameters from .common import _DeleteMultimodalDatasetRequestParameters from .common import _DeletePromptVersionRequestParameters +from .common import _DeleteSandboxEnvironmentSnapshotRequestParameters from .common import _DeleteSandboxEnvironmentTemplateRequestParameters from .common import _EvaluateInstancesRequestParameters from .common import _ExecuteCodeAgentEngineSandboxRequestParameters @@ -67,6 +69,7 @@ from .common import _GetAgentEngineRequestParameters from .common import _GetAgentEngineSandboxOperationParameters from .common import _GetAgentEngineSandboxRequestParameters +from .common import _GetAgentEngineSandboxSnapshotOperationParameters from .common import _GetAgentEngineSessionOperationParameters from .common import _GetAgentEngineSessionRequestParameters from .common import _GetAgentEngineTaskRequestParameters @@ -81,6 +84,7 @@ from .common import _GetEvaluationSetParameters from .common import _GetMultimodalDatasetOperationParameters from .common import _GetMultimodalDatasetParameters +from .common import _GetSandboxEnvironmentSnapshotRequestParameters from .common import _GetSandboxEnvironmentTemplateOperationParameters from .common import _GetSandboxEnvironmentTemplateRequestParameters from .common import _IngestEventsRequestParameters @@ -96,6 +100,7 @@ from .common import _ListDatasetVersionsRequestParameters from .common import _ListEvaluationMetricsParameters from .common import _ListMultimodalDatasetsRequestParameters +from .common import _ListSandboxEnvironmentSnapshotsRequestParameters from .common import _ListSandboxEnvironmentTemplatesRequestParameters from .common import _OptimizeRequestParameters from .common import _OptimizeRequestParameters @@ -146,6 +151,9 @@ from .common import AgentEngineSandboxOperation from .common import AgentEngineSandboxOperationDict from .common import AgentEngineSandboxOperationOrDict +from .common import AgentEngineSandboxSnapshotOperation +from .common import AgentEngineSandboxSnapshotOperationDict +from .common import AgentEngineSandboxSnapshotOperationOrDict from .common import AgentEngineSessionOperation from .common import AgentEngineSessionOperationDict from .common import AgentEngineSessionOperationOrDict @@ -243,6 +251,9 @@ from .common import CreateAgentEngineSandboxConfig from .common import CreateAgentEngineSandboxConfigDict from .common import CreateAgentEngineSandboxConfigOrDict +from .common import CreateAgentEngineSandboxSnapshotConfig +from .common import CreateAgentEngineSandboxSnapshotConfigDict +from .common import CreateAgentEngineSandboxSnapshotConfigOrDict from .common import CreateAgentEngineSessionConfig from .common import CreateAgentEngineSessionConfigDict from .common import CreateAgentEngineSessionConfigOrDict @@ -340,6 +351,12 @@ from .common import DeletePromptVersionOperation from .common import DeletePromptVersionOperationDict from .common import DeletePromptVersionOperationOrDict +from .common import DeleteSandboxEnvironmentSnapshotConfig +from .common import DeleteSandboxEnvironmentSnapshotConfigDict +from .common import DeleteSandboxEnvironmentSnapshotConfigOrDict +from .common import DeleteSandboxEnvironmentSnapshotOperation +from .common import DeleteSandboxEnvironmentSnapshotOperationDict +from .common import DeleteSandboxEnvironmentSnapshotOperationOrDict from .common import DeleteSandboxEnvironmentTemplateConfig from .common import DeleteSandboxEnvironmentTemplateConfigDict from .common import DeleteSandboxEnvironmentTemplateConfigOrDict @@ -565,6 +582,9 @@ from .common import GetPromptConfig from .common import GetPromptConfigDict from .common import GetPromptConfigOrDict +from .common import GetSandboxEnvironmentSnapshotConfig +from .common import GetSandboxEnvironmentSnapshotConfigDict +from .common import GetSandboxEnvironmentSnapshotConfigOrDict from .common import GetSandboxEnvironmentTemplateConfig from .common import GetSandboxEnvironmentTemplateConfigDict from .common import GetSandboxEnvironmentTemplateConfigOrDict @@ -659,6 +679,12 @@ from .common import ListReasoningEnginesSessionsResponse from .common import ListReasoningEnginesSessionsResponseDict from .common import ListReasoningEnginesSessionsResponseOrDict +from .common import ListSandboxEnvironmentSnapshotsConfig +from .common import ListSandboxEnvironmentSnapshotsConfigDict +from .common import ListSandboxEnvironmentSnapshotsConfigOrDict +from .common import ListSandboxEnvironmentSnapshotsResponse +from .common import ListSandboxEnvironmentSnapshotsResponseDict +from .common import ListSandboxEnvironmentSnapshotsResponseOrDict from .common import ListSandboxEnvironmentTemplatesConfig from .common import ListSandboxEnvironmentTemplatesConfigDict from .common import ListSandboxEnvironmentTemplatesConfigOrDict @@ -828,6 +854,7 @@ from .common import PointwiseMetricInstance from .common import PointwiseMetricInstanceDict from .common import PointwiseMetricInstanceOrDict +from .common import PostSnapshotAction from .common import Prompt from .common import PromptData from .common import PromptDataDict @@ -1058,6 +1085,9 @@ from .common import SandboxEnvironmentConnectionInfoOrDict from .common import SandboxEnvironmentDict from .common import SandboxEnvironmentOrDict +from .common import SandboxEnvironmentSnapshot +from .common import SandboxEnvironmentSnapshotDict +from .common import SandboxEnvironmentSnapshotOrDict from .common import SandboxEnvironmentSpec from .common import SandboxEnvironmentSpecCodeExecutionEnvironment from .common import SandboxEnvironmentSpecCodeExecutionEnvironmentDict @@ -2161,6 +2191,30 @@ "ListSandboxEnvironmentTemplatesResponse", "ListSandboxEnvironmentTemplatesResponseDict", "ListSandboxEnvironmentTemplatesResponseOrDict", + "CreateAgentEngineSandboxSnapshotConfig", + "CreateAgentEngineSandboxSnapshotConfigDict", + "CreateAgentEngineSandboxSnapshotConfigOrDict", + "SandboxEnvironmentSnapshot", + "SandboxEnvironmentSnapshotDict", + "SandboxEnvironmentSnapshotOrDict", + "AgentEngineSandboxSnapshotOperation", + "AgentEngineSandboxSnapshotOperationDict", + "AgentEngineSandboxSnapshotOperationOrDict", + "DeleteSandboxEnvironmentSnapshotConfig", + "DeleteSandboxEnvironmentSnapshotConfigDict", + "DeleteSandboxEnvironmentSnapshotConfigOrDict", + "DeleteSandboxEnvironmentSnapshotOperation", + "DeleteSandboxEnvironmentSnapshotOperationDict", + "DeleteSandboxEnvironmentSnapshotOperationOrDict", + "GetSandboxEnvironmentSnapshotConfig", + "GetSandboxEnvironmentSnapshotConfigDict", + "GetSandboxEnvironmentSnapshotConfigOrDict", + "ListSandboxEnvironmentSnapshotsConfig", + "ListSandboxEnvironmentSnapshotsConfigDict", + "ListSandboxEnvironmentSnapshotsConfigOrDict", + "ListSandboxEnvironmentSnapshotsResponse", + "ListSandboxEnvironmentSnapshotsResponseDict", + "ListSandboxEnvironmentSnapshotsResponseOrDict", "CreateAgentEngineSessionConfig", "CreateAgentEngineSessionConfigDict", "CreateAgentEngineSessionConfigOrDict", @@ -2470,6 +2524,7 @@ "MachineConfig", "Protocol", "DefaultContainerCategory", + "PostSnapshotAction", "Framework", "EvaluationItemType", "SamplingMethod", @@ -2560,6 +2615,11 @@ "_GetSandboxEnvironmentTemplateRequestParameters", "_ListSandboxEnvironmentTemplatesRequestParameters", "_GetSandboxEnvironmentTemplateOperationParameters", + "_CreateSandboxEnvironmentSnapshotRequestParameters", + "_DeleteSandboxEnvironmentSnapshotRequestParameters", + "_GetSandboxEnvironmentSnapshotRequestParameters", + "_ListSandboxEnvironmentSnapshotsRequestParameters", + "_GetAgentEngineSandboxSnapshotOperationParameters", "_CreateAgentEngineSessionRequestParameters", "_DeleteAgentEngineSessionRequestParameters", "_GetAgentEngineSessionRequestParameters", diff --git a/vertexai/_genai/types/common.py b/vertexai/_genai/types/common.py index b2b26b8573..c9724e5f81 100644 --- a/vertexai/_genai/types/common.py +++ b/vertexai/_genai/types/common.py @@ -344,6 +344,17 @@ class DefaultContainerCategory(_common.CaseInSensitiveEnum): """The default container image for Computer Use.""" +class PostSnapshotAction(_common.CaseInSensitiveEnum): + """Input only. Action to take on the source SandboxEnvironment after the snapshot is taken. This field is only used in CreateSandboxEnvironmentSnapshotRequest and it is not stored in the resource.""" + + POST_SNAPSHOT_ACTION_UNSPECIFIED = "POST_SNAPSHOT_ACTION_UNSPECIFIED" + """The default value. This value is unused.""" + RUNNING = "RUNNING" + """Sandbox environment will continue to run after snapshot is taken.""" + PAUSE = "PAUSE" + """Sandbox environment will be paused after snapshot is taken.""" + + class Framework(_common.CaseInSensitiveEnum): """Framework used to build the application.""" @@ -13055,6 +13066,471 @@ class _GetSandboxEnvironmentTemplateOperationParametersDict(TypedDict, total=Fal ] +class CreateAgentEngineSandboxSnapshotConfig(_common.BaseModel): + """Config for creating a Sandbox Environment Snapshot.""" + + http_options: Optional[genai_types.HttpOptions] = Field( + default=None, description="""Used to override HTTP request options.""" + ) + display_name: Optional[str] = Field( + default=None, description="""The display name of the sandbox snapshot.""" + ) + owner: Optional[str] = Field( + default=None, description="""The owner of the sandbox snapshot.""" + ) + ttl: Optional[str] = Field( + default=None, + description="""The TTL for this resource. The expiration time is computed: now + TTL.""", + ) + + +class CreateAgentEngineSandboxSnapshotConfigDict(TypedDict, total=False): + """Config for creating a Sandbox Environment Snapshot.""" + + http_options: Optional[genai_types.HttpOptionsDict] + """Used to override HTTP request options.""" + + display_name: Optional[str] + """The display name of the sandbox snapshot.""" + + owner: Optional[str] + """The owner of the sandbox snapshot.""" + + ttl: Optional[str] + """The TTL for this resource. The expiration time is computed: now + TTL.""" + + +CreateAgentEngineSandboxSnapshotConfigOrDict = Union[ + CreateAgentEngineSandboxSnapshotConfig, CreateAgentEngineSandboxSnapshotConfigDict +] + + +class _CreateSandboxEnvironmentSnapshotRequestParameters(_common.BaseModel): + """Parameters for creating a sandbox environment snapshot.""" + + source_sandbox_environment_name: Optional[str] = Field( + default=None, description="""Name of the sandbox environment to snapshot.""" + ) + config: Optional[CreateAgentEngineSandboxSnapshotConfig] = Field( + default=None, description="""""" + ) + + +class _CreateSandboxEnvironmentSnapshotRequestParametersDict(TypedDict, total=False): + """Parameters for creating a sandbox environment snapshot.""" + + source_sandbox_environment_name: Optional[str] + """Name of the sandbox environment to snapshot.""" + + config: Optional[CreateAgentEngineSandboxSnapshotConfigDict] + """""" + + +_CreateSandboxEnvironmentSnapshotRequestParametersOrDict = Union[ + _CreateSandboxEnvironmentSnapshotRequestParameters, + _CreateSandboxEnvironmentSnapshotRequestParametersDict, +] + + +class SandboxEnvironmentSnapshot(_common.BaseModel): + """A sandbox environment snapshot.""" + + display_name: Optional[str] = Field( + default=None, + description="""The display name of the sandbox environment snapshot.""", + ) + expire_time: Optional[datetime.datetime] = Field( + default=None, + description="""Expiration time of the sandbox environment snapshot. + """, + ) + create_time: Optional[datetime.datetime] = Field( + default=None, + description="""Output only. The timestamp when this SandboxEnvironmentSnapshot was created.""", + ) + name: Optional[str] = Field( + default=None, + description="""Identifier. The resource name of the SandboxEnvironmentSnapshot. Format: `projects/{project}/locations/{location}/reasoningEngines/{reasoning_engine}/sandboxEnvironmentSnapshots/{sandbox_environment_snapshot}`""", + ) + owner: Optional[str] = Field( + default=None, + description="""Optional. Owner information for this sandbox snapshot. Different owners will have isolations on snapshot storage and identity. If not set, snapshot will be created as the default owner.""", + ) + parent_snapshot: Optional[str] = Field( + default=None, + description="""Output only. The resource name of the parent SandboxEnvironmentSnapshot. Empty if this is a root Snapshot (the first snapshot from a newly created sandbox). Can be used to reconstruct the whole ancestry tree of snapshots.""", + ) + post_snapshot_action: Optional[PostSnapshotAction] = Field( + default=None, + description="""Optional. Input only. Action to take on the source SandboxEnvironment after the snapshot is taken. This field is only used in CreateSandboxEnvironmentSnapshotRequest and it is not stored in the resource.""", + ) + size_bytes: Optional[int] = Field( + default=None, + description="""Optional. Output only. Size of the snapshot data in bytes.""", + ) + source_sandbox_environment: Optional[str] = Field( + default=None, + description="""Required. The resource name of the source SandboxEnvironment this snapshot was taken from.""", + ) + ttl: Optional[str] = Field( + default=None, + description="""Optional. Input only. The TTL for the sandbox environment snapshot. The expiration time is computed: now + TTL.""", + ) + update_time: Optional[datetime.datetime] = Field( + default=None, + description="""Output only. The timestamp when this SandboxEnvironment was most recently updated.""", + ) + + +class SandboxEnvironmentSnapshotDict(TypedDict, total=False): + """A sandbox environment snapshot.""" + + display_name: Optional[str] + """The display name of the sandbox environment snapshot.""" + + expire_time: Optional[datetime.datetime] + """Expiration time of the sandbox environment snapshot. + """ + + create_time: Optional[datetime.datetime] + """Output only. The timestamp when this SandboxEnvironmentSnapshot was created.""" + + name: Optional[str] + """Identifier. The resource name of the SandboxEnvironmentSnapshot. Format: `projects/{project}/locations/{location}/reasoningEngines/{reasoning_engine}/sandboxEnvironmentSnapshots/{sandbox_environment_snapshot}`""" + + owner: Optional[str] + """Optional. Owner information for this sandbox snapshot. Different owners will have isolations on snapshot storage and identity. If not set, snapshot will be created as the default owner.""" + + parent_snapshot: Optional[str] + """Output only. The resource name of the parent SandboxEnvironmentSnapshot. Empty if this is a root Snapshot (the first snapshot from a newly created sandbox). Can be used to reconstruct the whole ancestry tree of snapshots.""" + + post_snapshot_action: Optional[PostSnapshotAction] + """Optional. Input only. Action to take on the source SandboxEnvironment after the snapshot is taken. This field is only used in CreateSandboxEnvironmentSnapshotRequest and it is not stored in the resource.""" + + size_bytes: Optional[int] + """Optional. Output only. Size of the snapshot data in bytes.""" + + source_sandbox_environment: Optional[str] + """Required. The resource name of the source SandboxEnvironment this snapshot was taken from.""" + + ttl: Optional[str] + """Optional. Input only. The TTL for the sandbox environment snapshot. The expiration time is computed: now + TTL.""" + + update_time: Optional[datetime.datetime] + """Output only. The timestamp when this SandboxEnvironment was most recently updated.""" + + +SandboxEnvironmentSnapshotOrDict = Union[ + SandboxEnvironmentSnapshot, SandboxEnvironmentSnapshotDict +] + + +class AgentEngineSandboxSnapshotOperation(_common.BaseModel): + """Operation that has an agent engine sandbox snapshot as a response.""" + + name: Optional[str] = Field( + default=None, + description="""The server-assigned name, which is only unique within the same service that originally returns it. If you use the default HTTP mapping, the `name` should be a resource name ending with `operations/{unique_id}`.""", + ) + metadata: Optional[dict[str, Any]] = Field( + default=None, + description="""Service-specific metadata associated with the operation. It typically contains progress information and common metadata such as create time. Some services might not provide such metadata. Any method that returns a long-running operation should document the metadata type, if any.""", + ) + done: Optional[bool] = Field( + default=None, + description="""If the value is `false`, it means the operation is still in progress. If `true`, the operation is completed, and either `error` or `response` is available.""", + ) + error: Optional[dict[str, Any]] = Field( + default=None, + description="""The error result of the operation in case of failure or cancellation.""", + ) + response: Optional[SandboxEnvironmentSnapshot] = Field( + default=None, description="""The Agent Engine Sandbox Snapshot.""" + ) + + +class AgentEngineSandboxSnapshotOperationDict(TypedDict, total=False): + """Operation that has an agent engine sandbox snapshot as a response.""" + + name: Optional[str] + """The server-assigned name, which is only unique within the same service that originally returns it. If you use the default HTTP mapping, the `name` should be a resource name ending with `operations/{unique_id}`.""" + + metadata: Optional[dict[str, Any]] + """Service-specific metadata associated with the operation. It typically contains progress information and common metadata such as create time. Some services might not provide such metadata. Any method that returns a long-running operation should document the metadata type, if any.""" + + done: Optional[bool] + """If the value is `false`, it means the operation is still in progress. If `true`, the operation is completed, and either `error` or `response` is available.""" + + error: Optional[dict[str, Any]] + """The error result of the operation in case of failure or cancellation.""" + + response: Optional[SandboxEnvironmentSnapshotDict] + """The Agent Engine Sandbox Snapshot.""" + + +AgentEngineSandboxSnapshotOperationOrDict = Union[ + AgentEngineSandboxSnapshotOperation, AgentEngineSandboxSnapshotOperationDict +] + + +class DeleteSandboxEnvironmentSnapshotConfig(_common.BaseModel): + """Config for deleting a Sandbox Environment Snapshot.""" + + http_options: Optional[genai_types.HttpOptions] = Field( + default=None, description="""Used to override HTTP request options.""" + ) + + +class DeleteSandboxEnvironmentSnapshotConfigDict(TypedDict, total=False): + """Config for deleting a Sandbox Environment Snapshot.""" + + http_options: Optional[genai_types.HttpOptionsDict] + """Used to override HTTP request options.""" + + +DeleteSandboxEnvironmentSnapshotConfigOrDict = Union[ + DeleteSandboxEnvironmentSnapshotConfig, DeleteSandboxEnvironmentSnapshotConfigDict +] + + +class _DeleteSandboxEnvironmentSnapshotRequestParameters(_common.BaseModel): + """Parameters for deleting sandbox environment snapshots.""" + + name: Optional[str] = Field( + default=None, + description="""Name of the sandbox environment snapshot to delete.""", + ) + config: Optional[DeleteSandboxEnvironmentSnapshotConfig] = Field( + default=None, description="""""" + ) + + +class _DeleteSandboxEnvironmentSnapshotRequestParametersDict(TypedDict, total=False): + """Parameters for deleting sandbox environment snapshots.""" + + name: Optional[str] + """Name of the sandbox environment snapshot to delete.""" + + config: Optional[DeleteSandboxEnvironmentSnapshotConfigDict] + """""" + + +_DeleteSandboxEnvironmentSnapshotRequestParametersOrDict = Union[ + _DeleteSandboxEnvironmentSnapshotRequestParameters, + _DeleteSandboxEnvironmentSnapshotRequestParametersDict, +] + + +class DeleteSandboxEnvironmentSnapshotOperation(_common.BaseModel): + """Operation for deleting sandbox environment snapshots.""" + + name: Optional[str] = Field( + default=None, + description="""The server-assigned name, which is only unique within the same service that originally returns it. If you use the default HTTP mapping, the `name` should be a resource name ending with `operations/{unique_id}`.""", + ) + metadata: Optional[dict[str, Any]] = Field( + default=None, + description="""Service-specific metadata associated with the operation. It typically contains progress information and common metadata such as create time. Some services might not provide such metadata. Any method that returns a long-running operation should document the metadata type, if any.""", + ) + done: Optional[bool] = Field( + default=None, + description="""If the value is `false`, it means the operation is still in progress. If `true`, the operation is completed, and either `error` or `response` is available.""", + ) + error: Optional[dict[str, Any]] = Field( + default=None, + description="""The error result of the operation in case of failure or cancellation.""", + ) + + +class DeleteSandboxEnvironmentSnapshotOperationDict(TypedDict, total=False): + """Operation for deleting sandbox environment snapshots.""" + + name: Optional[str] + """The server-assigned name, which is only unique within the same service that originally returns it. If you use the default HTTP mapping, the `name` should be a resource name ending with `operations/{unique_id}`.""" + + metadata: Optional[dict[str, Any]] + """Service-specific metadata associated with the operation. It typically contains progress information and common metadata such as create time. Some services might not provide such metadata. Any method that returns a long-running operation should document the metadata type, if any.""" + + done: Optional[bool] + """If the value is `false`, it means the operation is still in progress. If `true`, the operation is completed, and either `error` or `response` is available.""" + + error: Optional[dict[str, Any]] + """The error result of the operation in case of failure or cancellation.""" + + +DeleteSandboxEnvironmentSnapshotOperationOrDict = Union[ + DeleteSandboxEnvironmentSnapshotOperation, + DeleteSandboxEnvironmentSnapshotOperationDict, +] + + +class GetSandboxEnvironmentSnapshotConfig(_common.BaseModel): + """Config for getting a Sandbox Environment Snapshot.""" + + http_options: Optional[genai_types.HttpOptions] = Field( + default=None, description="""Used to override HTTP request options.""" + ) + + +class GetSandboxEnvironmentSnapshotConfigDict(TypedDict, total=False): + """Config for getting a Sandbox Environment Snapshot.""" + + http_options: Optional[genai_types.HttpOptionsDict] + """Used to override HTTP request options.""" + + +GetSandboxEnvironmentSnapshotConfigOrDict = Union[ + GetSandboxEnvironmentSnapshotConfig, GetSandboxEnvironmentSnapshotConfigDict +] + + +class _GetSandboxEnvironmentSnapshotRequestParameters(_common.BaseModel): + """Parameters for getting a sandbox environment snapshot.""" + + name: Optional[str] = Field( + default=None, description="""Name of the sandbox environment snapshot.""" + ) + config: Optional[GetSandboxEnvironmentSnapshotConfig] = Field( + default=None, description="""""" + ) + + +class _GetSandboxEnvironmentSnapshotRequestParametersDict(TypedDict, total=False): + """Parameters for getting a sandbox environment snapshot.""" + + name: Optional[str] + """Name of the sandbox environment snapshot.""" + + config: Optional[GetSandboxEnvironmentSnapshotConfigDict] + """""" + + +_GetSandboxEnvironmentSnapshotRequestParametersOrDict = Union[ + _GetSandboxEnvironmentSnapshotRequestParameters, + _GetSandboxEnvironmentSnapshotRequestParametersDict, +] + + +class ListSandboxEnvironmentSnapshotsConfig(_common.BaseModel): + """Config for listing sandbox environment snapshots.""" + + http_options: Optional[genai_types.HttpOptions] = Field( + default=None, description="""Used to override HTTP request options.""" + ) + page_size: Optional[int] = Field(default=None, description="""""") + page_token: Optional[str] = Field(default=None, description="""""") + filter: Optional[str] = Field( + default=None, + description="""An expression for filtering the results of the request.""", + ) + + +class ListSandboxEnvironmentSnapshotsConfigDict(TypedDict, total=False): + """Config for listing sandbox environment snapshots.""" + + http_options: Optional[genai_types.HttpOptionsDict] + """Used to override HTTP request options.""" + + page_size: Optional[int] + """""" + + page_token: Optional[str] + """""" + + filter: Optional[str] + """An expression for filtering the results of the request.""" + + +ListSandboxEnvironmentSnapshotsConfigOrDict = Union[ + ListSandboxEnvironmentSnapshotsConfig, ListSandboxEnvironmentSnapshotsConfigDict +] + + +class _ListSandboxEnvironmentSnapshotsRequestParameters(_common.BaseModel): + """Parameters for listing sandbox environment snapshots.""" + + name: Optional[str] = Field( + default=None, + description="""Name of the reasoning engine to list snapshots from.""", + ) + config: Optional[ListSandboxEnvironmentSnapshotsConfig] = Field( + default=None, description="""""" + ) + + +class _ListSandboxEnvironmentSnapshotsRequestParametersDict(TypedDict, total=False): + """Parameters for listing sandbox environment snapshots.""" + + name: Optional[str] + """Name of the reasoning engine to list snapshots from.""" + + config: Optional[ListSandboxEnvironmentSnapshotsConfigDict] + """""" + + +_ListSandboxEnvironmentSnapshotsRequestParametersOrDict = Union[ + _ListSandboxEnvironmentSnapshotsRequestParameters, + _ListSandboxEnvironmentSnapshotsRequestParametersDict, +] + + +class ListSandboxEnvironmentSnapshotsResponse(_common.BaseModel): + """Response for listing sandbox environment snapshots.""" + + sdk_http_response: Optional[genai_types.HttpResponse] = Field( + default=None, description="""Used to retain the full HTTP response.""" + ) + next_page_token: Optional[str] = Field(default=None, description="""""") + sandbox_environment_snapshots: Optional[list[SandboxEnvironmentSnapshot]] = Field( + default=None, description="""List of sandbox environment snapshots.""" + ) + + +class ListSandboxEnvironmentSnapshotsResponseDict(TypedDict, total=False): + """Response for listing sandbox environment snapshots.""" + + sdk_http_response: Optional[genai_types.HttpResponseDict] + """Used to retain the full HTTP response.""" + + next_page_token: Optional[str] + """""" + + sandbox_environment_snapshots: Optional[list[SandboxEnvironmentSnapshotDict]] + """List of sandbox environment snapshots.""" + + +ListSandboxEnvironmentSnapshotsResponseOrDict = Union[ + ListSandboxEnvironmentSnapshotsResponse, ListSandboxEnvironmentSnapshotsResponseDict +] + + +class _GetAgentEngineSandboxSnapshotOperationParameters(_common.BaseModel): + """Parameters for getting an operation with a sandbox snapshot as a response.""" + + operation_name: Optional[str] = Field( + default=None, description="""The server-assigned name for the operation.""" + ) + config: Optional[GetAgentEngineOperationConfig] = Field( + default=None, description="""Used to override the default configuration.""" + ) + + +class _GetAgentEngineSandboxSnapshotOperationParametersDict(TypedDict, total=False): + """Parameters for getting an operation with a sandbox snapshot as a response.""" + + operation_name: Optional[str] + """The server-assigned name for the operation.""" + + config: Optional[GetAgentEngineOperationConfigDict] + """Used to override the default configuration.""" + + +_GetAgentEngineSandboxSnapshotOperationParametersOrDict = Union[ + _GetAgentEngineSandboxSnapshotOperationParameters, + _GetAgentEngineSandboxSnapshotOperationParametersDict, +] + + class CreateAgentEngineSessionConfig(_common.BaseModel): """Config for creating a Session."""