# 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 functools
import json
import logging
import mimetypes
from typing import Any, Iterator, 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 google.genai.pagers import Pager

from . import _agent_engines_utils
from . import types


logger = logging.getLogger("vertexai_genai.sandboxes")

logger.setLevel(logging.INFO)


def _CreateAgentEngineSandboxConfig_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, ["description"]) is not None:
        setv(parent_object, ["description"], getv(from_object, ["description"]))

    if getv(from_object, ["ttl"]) is not None:
        setv(parent_object, ["ttl"], getv(from_object, ["ttl"]))

    return to_object


def _CreateAgentEngineSandboxRequestParameters_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, ["spec"]) is not None:
        setv(to_object, ["spec"], getv(from_object, ["spec"]))

    if getv(from_object, ["config"]) is not None:
        setv(
            to_object,
            ["config"],
            _CreateAgentEngineSandboxConfig_to_vertex(
                getv(from_object, ["config"]), to_object
            ),
        )

    return to_object


def _DeleteAgentEngineSandboxRequestParameters_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:
        setv(to_object, ["config"], getv(from_object, ["config"]))

    return to_object


def _ExecuteCodeAgentEngineSandboxRequestParameters_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, ["inputs"]) is not None:
        setv(to_object, ["inputs"], [item for item in getv(from_object, ["inputs"])])

    if getv(from_object, ["config"]) is not None:
        setv(to_object, ["config"], getv(from_object, ["config"]))

    return to_object


def _GetAgentEngineSandboxOperationParameters_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"])
        )

    if getv(from_object, ["config"]) is not None:
        setv(to_object, ["config"], getv(from_object, ["config"]))

    return to_object


def _GetAgentEngineSandboxRequestParameters_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:
        setv(to_object, ["config"], getv(from_object, ["config"]))

    return to_object


def _ListAgentEngineSandboxesConfig_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 _ListAgentEngineSandboxesRequestParameters_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:
        setv(
            to_object,
            ["config"],
            _ListAgentEngineSandboxesConfig_to_vertex(
                getv(from_object, ["config"]), to_object
            ),
        )

    return to_object


class Sandboxes(_api_module.BaseModule):

    def _create(
        self,
        *,
        name: str,
        spec: Optional[types.SandboxEnvironmentSpecOrDict] = None,
        config: Optional[types.CreateAgentEngineSandboxConfigOrDict] = None,
    ) -> types.AgentEngineSandboxOperation:
        """
        Creates a new sandbox in the Agent Engine.
        """

        parameter_model = types._CreateAgentEngineSandboxRequestParameters(
            name=name,
            spec=spec,
            config=config,
        )

        request_url_dict: Optional[dict[str, str]]
        if not self._api_client.vertexai:
            raise ValueError("This method is only supported in the Vertex AI client.")
        else:
            request_dict = _CreateAgentEngineSandboxRequestParameters_to_vertex(
                parameter_model
            )
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = "{name}/sandboxEnvironments".format_map(request_url_dict)
            else:
                path = "{name}/sandboxEnvironments"

        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.AgentEngineSandboxOperation._from_response(
            response=response_dict, kwargs=parameter_model.model_dump()
        )

        self._api_client._verify_response(return_value)
        return return_value

    def _delete(
        self,
        *,
        name: str,
        config: Optional[types.DeleteAgentEngineSandboxConfigOrDict] = None,
    ) -> types.DeleteAgentEngineSandboxOperation:
        """
        Delete an Agent Engine sandbox.

        Args:
            name (str):
                Required. The name of the Agent Engine sandbox to be deleted. Format:
                `projects/{project}/locations/{location}/reasoningEngines/{resource_id}/sandboxEnvironments/{sandbox}`.

        """

        parameter_model = types._DeleteAgentEngineSandboxRequestParameters(
            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 Vertex AI client.")
        else:
            request_dict = _DeleteAgentEngineSandboxRequestParameters_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.DeleteAgentEngineSandboxOperation._from_response(
            response=response_dict, kwargs=parameter_model.model_dump()
        )

        self._api_client._verify_response(return_value)
        return return_value

    def _execute_code(
        self,
        *,
        name: str,
        inputs: Optional[list[types.ChunkOrDict]] = None,
        config: Optional[types.ExecuteCodeAgentEngineSandboxConfigOrDict] = None,
    ) -> types.ExecuteSandboxEnvironmentResponse:
        """
        Execute code in an Agent Engine sandbox.
        """

        parameter_model = types._ExecuteCodeAgentEngineSandboxRequestParameters(
            name=name,
            inputs=inputs,
            config=config,
        )

        request_url_dict: Optional[dict[str, str]]
        if not self._api_client.vertexai:
            raise ValueError("This method is only supported in the Vertex AI client.")
        else:
            request_dict = _ExecuteCodeAgentEngineSandboxRequestParameters_to_vertex(
                parameter_model
            )
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = "{name}/:execute".format_map(request_url_dict)
            else:
                path = "{name}/:execute"

        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.ExecuteSandboxEnvironmentResponse._from_response(
            response=response_dict, kwargs=parameter_model.model_dump()
        )

        self._api_client._verify_response(return_value)
        return return_value

    def _get(
        self,
        *,
        name: str,
        config: Optional[types.GetAgentEngineSandboxConfigOrDict] = None,
    ) -> types.SandboxEnvironment:
        """
        Gets an agent engine sandbox.

        Args:
            name (str): Required. A fully-qualified resource name or ID such as
              "projects/123/locations/us-central1/reasoningEngines/456/sandboxEnvironments/789"
              or a shortened name such as "reasoningEngines/456/sandboxEnvironments/789".

        """

        parameter_model = types._GetAgentEngineSandboxRequestParameters(
            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 Vertex AI client.")
        else:
            request_dict = _GetAgentEngineSandboxRequestParameters_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.SandboxEnvironment._from_response(
            response=response_dict, kwargs=parameter_model.model_dump()
        )

        self._api_client._verify_response(return_value)
        return return_value

    def _list(
        self,
        *,
        name: str,
        config: Optional[types.ListAgentEngineSandboxesConfigOrDict] = None,
    ) -> types.ListAgentEngineSandboxesResponse:
        """
        Lists Agent Engine sandboxes.

        Args:
            name (str): Required. The name of the Agent Engine to list sessions for. Format:
                `projects/{project}/locations/{location}/reasoningEngines/{resource_id}`.
            config (ListAgentEngineSandboxesConfig):
                Optional. Additional configurations for listing the Agent Engine sandboxes.

        Returns:
            ListReasoningEnginesSandboxesResponse: The requested Agent Engine sandboxes.

        """

        parameter_model = types._ListAgentEngineSandboxesRequestParameters(
            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 Vertex AI client.")
        else:
            request_dict = _ListAgentEngineSandboxesRequestParameters_to_vertex(
                parameter_model
            )
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = "{name}/sandboxEnvironments".format_map(request_url_dict)
            else:
                path = "{name}/sandboxEnvironments"

        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.ListAgentEngineSandboxesResponse._from_response(
            response=response_dict, kwargs=parameter_model.model_dump()
        )

        self._api_client._verify_response(return_value)
        return return_value

    def _get_sandbox_operation(
        self,
        *,
        operation_name: str,
        config: Optional[types.GetAgentEngineOperationConfigOrDict] = None,
    ) -> types.AgentEngineSandboxOperation:
        parameter_model = types._GetAgentEngineSandboxOperationParameters(
            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 Vertex AI client.")
        else:
            request_dict = _GetAgentEngineSandboxOperationParameters_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.AgentEngineSandboxOperation._from_response(
            response=response_dict, kwargs=parameter_model.model_dump()
        )

        self._api_client._verify_response(return_value)
        return return_value

    def create(
        self,
        *,
        name: str,
        spec: Optional[types.SandboxEnvironmentSpecOrDict] = None,
        config: Optional[types.CreateAgentEngineSandboxConfigOrDict] = None,
    ) -> types.AgentEngineSandboxOperation:
        """Creates a new sandbox in the Agent Engine.

        Args:
            name (str):
                Required. The name of the agent engine to create sandbox for.
                projects/{project}/locations/{location}/reasoningEngines/{resource_id}
            spec (SandboxEnvironmentSpec):
                Optional. The specification for the sandbox to create.
            config (CreateAgentEngineSandboxConfigOrDict):
                Optional. The configuration for the sandbox.

        Returns:
            AgentEngineSandboxOperation: The operation for creating the sandbox.
        """
        operation = self._create(
            name=name,
            spec=spec,
            config=config,
        )
        if config is None:
            config = types.CreateAgentEngineSandboxConfig()
        elif isinstance(config, dict):
            config = types.CreateAgentEngineSandboxConfig.model_validate(config)
        if config.wait_for_completion:
            if not operation.done:
                operation = _agent_engines_utils._await_operation(
                    operation_name=operation.name,
                    get_operation_fn=self._get_sandbox_operation,
                    poll_interval_seconds=0.1,
                )
            # We need to make a call to get the sandbox because the operation
            # response might not contain the relevant fields.
            if not operation.response:
                raise ValueError("Error retrieving sandbox.")
            operation.response = self.get(name=operation.response.name)
        return operation

    def list(
        self,
        *,
        name: str,
        config: Optional[types.ListAgentEngineSandboxesConfigOrDict] = None,
    ) -> Iterator[types.SandboxEnvironment]:
        """Lists Agent Engine sandboxes.

        Args:
            name (str):
                Required. The name of the agent engine to list sandboxes for.
                projects/{project}/locations/{location}/reasoningEngines/{resource_id}/SandboxEnvironments/{sandbox_id}
            config (ListAgentEngineSandboxConfig):
                Optional. The configuration for the sandboxes to list.

        Returns:
            Iterable[SandboxEnvironment]: An iterable of agent engine sandboxes.
        """
        return Pager(
            "sandbox_environments",
            functools.partial(self._list, name=name),
            self._list(name=name, config=config),
            config,
        )

    def execute_code(
        self,
        *,
        name: str,
        input_data: dict[str, Any],
        config: Optional[types.ExecuteCodeAgentEngineSandboxConfigOrDict] = None,
    ) -> types.ExecuteSandboxEnvironmentResponse:
        """Executes code in the Agent Engine sandbox.

        Args:
            name (str):
                Required. The name of the agent engine sandbox to run code in.
                projects/{project}/locations/{location}/reasoningEngines/{resource_id}/SandboxEnvironments/{sandbox_id}
            input_data (dict[str, Any]):
                Required. The input to the code to execute.
            config (ExecuteCodeAgentEngineSandboxConfigOrDict):
                Optional. The configuration for the sandboxes to run code in.

        Returns:
            ExecuteSandboxEnvironmentResponse: The response from executing the code.
        """
        input_chunks = []

        if input_data.get("code") is not None:
            code = input_data.get("code", "")
            json_code = json.dumps({"code": code}).encode("utf-8")
            input_chunks.append(
                types.Chunk(
                    mime_type="application/json",
                    data=json_code,
                )
            )

        for file in input_data.get("files", []):
            file_name = file.get("name", "")
            input_chunks.append(
                types.Chunk(
                    mime_type=file.get("mimeType", ""),
                    data=file.get("content", b""),
                    metadata={"attributes": {"file_name": file_name.encode("utf-8")}},
                )
            )

        response = self._execute_code(
            name=name,
            inputs=input_chunks,
            config=config,
        )

        output_chunks = []
        for output in response.outputs:
            if output.mime_type is None:
                # if mime_type is not available, try to guess the mime_type from the file_name.
                if (
                    output.metadata is not None
                    and output.metadata.attributes is not None
                ):
                    file_name = output.metadata.attributes.get("file_name", b"").decode(
                        "utf-8"
                    )
                    mime_type, _ = mimetypes.guess_type(file_name)
                    output.mime_type = mime_type

            output_chunks.append(output)

        response = types.ExecuteSandboxEnvironmentResponse(outputs=output_chunks)

        return response

    def get(
        self,
        *,
        name: str,
        config: Optional[types.GetAgentEngineSandboxConfigOrDict] = None,
    ) -> types.SandboxEnvironment:
        """Gets an agent engine sandbox.
        Args:
          name (str):
              Required. A fully-qualified resource name or ID such as
              projects/{project}/locations/{location}/reasoningEngines/{resource_id}/SandboxEnvironments/{sandbox_id}
              or a shortened name such as "reasoningEngines/{resource_id}/sandboxEnvironments/{sandbox_id}".
          config (GetAgentEngineSandboxConfigOrDict):
              Optional. The configuration for the sandbox to get.

        """
        return self._get(name=name, config=config)

    def delete(
        self,
        *,
        name: str,
        config: Optional[types.DeleteAgentEngineSandboxConfigOrDict] = None,
    ) -> types.DeleteAgentEngineSandboxOperation:
        """Deletes an agent engine sandbox.
        Args:
          name (str):
              Required. A fully-qualified resource name or ID such as
              projects/{project}/locations/{location}/reasoningEngines/{resource_id}/SandboxEnvironments/{sandbox_id}
              or a shortened name such as "reasoningEngines/{resource_id}/sandboxEnvironments/{sandbox_id}".
          config (DeleteAgentEngineSandboxConfigOrDict):
              Optional. The configuration for the sandbox to delete.
        """
        return self._delete(name=name, config=config)


class AsyncSandboxes(_api_module.BaseModule):

    async def _create(
        self,
        *,
        name: str,
        spec: Optional[types.SandboxEnvironmentSpecOrDict] = None,
        config: Optional[types.CreateAgentEngineSandboxConfigOrDict] = None,
    ) -> types.AgentEngineSandboxOperation:
        """
        Creates a new sandbox in the Agent Engine.
        """

        parameter_model = types._CreateAgentEngineSandboxRequestParameters(
            name=name,
            spec=spec,
            config=config,
        )

        request_url_dict: Optional[dict[str, str]]
        if not self._api_client.vertexai:
            raise ValueError("This method is only supported in the Vertex AI client.")
        else:
            request_dict = _CreateAgentEngineSandboxRequestParameters_to_vertex(
                parameter_model
            )
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = "{name}/sandboxEnvironments".format_map(request_url_dict)
            else:
                path = "{name}/sandboxEnvironments"

        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.AgentEngineSandboxOperation._from_response(
            response=response_dict, kwargs=parameter_model.model_dump()
        )

        self._api_client._verify_response(return_value)
        return return_value

    async def _delete(
        self,
        *,
        name: str,
        config: Optional[types.DeleteAgentEngineSandboxConfigOrDict] = None,
    ) -> types.DeleteAgentEngineSandboxOperation:
        """
        Delete an Agent Engine sandbox.

        Args:
            name (str):
                Required. The name of the Agent Engine sandbox to be deleted. Format:
                `projects/{project}/locations/{location}/reasoningEngines/{resource_id}/sandboxEnvironments/{sandbox}`.

        """

        parameter_model = types._DeleteAgentEngineSandboxRequestParameters(
            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 Vertex AI client.")
        else:
            request_dict = _DeleteAgentEngineSandboxRequestParameters_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.DeleteAgentEngineSandboxOperation._from_response(
            response=response_dict, kwargs=parameter_model.model_dump()
        )

        self._api_client._verify_response(return_value)
        return return_value

    async def _execute_code(
        self,
        *,
        name: str,
        inputs: Optional[list[types.ChunkOrDict]] = None,
        config: Optional[types.ExecuteCodeAgentEngineSandboxConfigOrDict] = None,
    ) -> types.ExecuteSandboxEnvironmentResponse:
        """
        Execute code in an Agent Engine sandbox.
        """

        parameter_model = types._ExecuteCodeAgentEngineSandboxRequestParameters(
            name=name,
            inputs=inputs,
            config=config,
        )

        request_url_dict: Optional[dict[str, str]]
        if not self._api_client.vertexai:
            raise ValueError("This method is only supported in the Vertex AI client.")
        else:
            request_dict = _ExecuteCodeAgentEngineSandboxRequestParameters_to_vertex(
                parameter_model
            )
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = "{name}/:execute".format_map(request_url_dict)
            else:
                path = "{name}/:execute"

        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.ExecuteSandboxEnvironmentResponse._from_response(
            response=response_dict, kwargs=parameter_model.model_dump()
        )

        self._api_client._verify_response(return_value)
        return return_value

    async def _get(
        self,
        *,
        name: str,
        config: Optional[types.GetAgentEngineSandboxConfigOrDict] = None,
    ) -> types.SandboxEnvironment:
        """
        Gets an agent engine sandbox.

        Args:
            name (str): Required. A fully-qualified resource name or ID such as
              "projects/123/locations/us-central1/reasoningEngines/456/sandboxEnvironments/789"
              or a shortened name such as "reasoningEngines/456/sandboxEnvironments/789".

        """

        parameter_model = types._GetAgentEngineSandboxRequestParameters(
            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 Vertex AI client.")
        else:
            request_dict = _GetAgentEngineSandboxRequestParameters_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.SandboxEnvironment._from_response(
            response=response_dict, kwargs=parameter_model.model_dump()
        )

        self._api_client._verify_response(return_value)
        return return_value

    async def _list(
        self,
        *,
        name: str,
        config: Optional[types.ListAgentEngineSandboxesConfigOrDict] = None,
    ) -> types.ListAgentEngineSandboxesResponse:
        """
        Lists Agent Engine sandboxes.

        Args:
            name (str): Required. The name of the Agent Engine to list sessions for. Format:
                `projects/{project}/locations/{location}/reasoningEngines/{resource_id}`.
            config (ListAgentEngineSandboxesConfig):
                Optional. Additional configurations for listing the Agent Engine sandboxes.

        Returns:
            ListReasoningEnginesSandboxesResponse: The requested Agent Engine sandboxes.

        """

        parameter_model = types._ListAgentEngineSandboxesRequestParameters(
            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 Vertex AI client.")
        else:
            request_dict = _ListAgentEngineSandboxesRequestParameters_to_vertex(
                parameter_model
            )
            request_url_dict = request_dict.get("_url")
            if request_url_dict:
                path = "{name}/sandboxEnvironments".format_map(request_url_dict)
            else:
                path = "{name}/sandboxEnvironments"

        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.ListAgentEngineSandboxesResponse._from_response(
            response=response_dict, kwargs=parameter_model.model_dump()
        )

        self._api_client._verify_response(return_value)
        return return_value

    async def _get_sandbox_operation(
        self,
        *,
        operation_name: str,
        config: Optional[types.GetAgentEngineOperationConfigOrDict] = None,
    ) -> types.AgentEngineSandboxOperation:
        parameter_model = types._GetAgentEngineSandboxOperationParameters(
            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 Vertex AI client.")
        else:
            request_dict = _GetAgentEngineSandboxOperationParameters_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.AgentEngineSandboxOperation._from_response(
            response=response_dict, kwargs=parameter_model.model_dump()
        )

        self._api_client._verify_response(return_value)
        return return_value