import datetime
import json
import logging
import shlex
import uuid
from typing import Dict, List, Optional, Union, overload

import httpx
from packaging.version import Version
from typing_extensions import Self, Unpack

from e2b.api.client.types import Unset
from e2b.connection_config import ApiParams, ConnectionConfig
from e2b.envd.api import ENVD_API_HEALTH_ROUTE, handle_envd_api_exception
from e2b.envd.versions import ENVD_DEBUG_FALLBACK
from e2b.exceptions import (
    TemplateException,
    format_request_timeout_error,
)
from e2b.sandbox.main import SandboxOpts
from e2b.sandbox.sandbox_api import (
    McpServer,
    SandboxLifecycle,
    SandboxMetrics,
    SandboxNetworkOpts,
    SandboxNetworkUpdate,
    SnapshotInfo,
)
from e2b.sandbox.utils import class_method_variant
from e2b.sandbox_sync.commands.command import Commands
from e2b.sandbox_sync.commands.pty import Pty
from e2b.sandbox_sync.filesystem.filesystem import Filesystem
from e2b.sandbox_sync.git import Git
from e2b.sandbox_sync.sandbox_api import SandboxApi, SandboxInfo
from e2b.sandbox_sync.paginator import SnapshotPaginator
from e2b.api.client.models import SandboxVolumeMount as SandboxVolumeMountAPI
from e2b.volume.volume_sync import Volume

logger = logging.getLogger(__name__)

SandboxVolumeMount = Dict[str, Union[Volume, str]]


class Sandbox(SandboxApi):
    """
    E2B cloud sandbox is a secure and isolated cloud environment.

    The sandbox allows you to:
    - Access Linux OS
    - Create, list, and delete files and directories
    - Run commands
    - Run isolated code
    - Access the internet

    Check docs [here](https://e2b.dev/docs).

    Use the `Sandbox.create()` to create a new sandbox.

    Example:
    ```python
    from e2b import Sandbox

    sandbox = Sandbox.create()
    ```
    """

    @property
    def files(self) -> Filesystem:
        """
        Module for interacting with the sandbox filesystem.
        """
        return self._filesystem

    @property
    def commands(self) -> Commands:
        """
        Module for running commands in the sandbox.
        """
        return self._commands

    @property
    def pty(self) -> Pty:
        """
        Module for interacting with the sandbox pseudo-terminal.
        """
        return self._pty

    @property
    def git(self) -> Git:
        """
        Module for running git operations in the sandbox.
        """
        return self._git

    def __init__(self, **opts: Unpack[SandboxOpts]):
        """
        :deprecated: This constructor is deprecated

        Use `Sandbox.create()` to create a new sandbox instead.
        """
        super().__init__(**opts)

        self._filesystem = Filesystem(
            self.envd_api_url,
            self._envd_version,
            self.connection_config,
        )
        self._commands = Commands(
            self.envd_api_url,
            self.connection_config,
            self._envd_version,
        )
        self._pty = Pty(
            self.envd_api_url,
            self.connection_config,
            self._envd_version,
        )
        self._git = Git(self._commands)

    @property
    def _envd_api(self) -> httpx.Client:
        return self._filesystem._envd_api

    def is_running(self, request_timeout: Optional[float] = None) -> bool:
        """
        Check if the sandbox is running.

        :param request_timeout: Timeout for the request in **seconds**

        :return: `True` if the sandbox is running, `False` otherwise

        Example
        ```python
        sandbox = Sandbox.create()
        sandbox.is_running() # Returns True

        sandbox.kill()
        sandbox.is_running() # Returns False
        ```
        """
        try:
            r = self._envd_api.get(
                ENVD_API_HEALTH_ROUTE,
                timeout=self.connection_config.get_request_timeout(request_timeout),
            )

            if r.status_code == 502:
                return False

            err = handle_envd_api_exception(r)

            if err:
                raise err

        except httpx.TimeoutException:
            raise format_request_timeout_error()

        return True

    @classmethod
    def create(
        cls,
        template: Optional[str] = None,
        timeout: Optional[int] = None,
        metadata: Optional[Dict[str, str]] = None,
        envs: Optional[Dict[str, str]] = None,
        secure: bool = True,
        allow_internet_access: bool = True,
        mcp: Optional[McpServer] = None,
        network: Optional[SandboxNetworkOpts] = None,
        lifecycle: Optional[SandboxLifecycle] = None,
        volume_mounts: Optional[SandboxVolumeMount] = None,
        logger: Optional[logging.Logger] = None,
        **opts: Unpack[ApiParams],
    ) -> Self:
        """
        Create a new sandbox.

        By default, the sandbox is created from the default `base` sandbox template.

        :param template: Sandbox template name or ID
        :param timeout: Timeout for the sandbox in **seconds**, default to 300 seconds. The maximum time a sandbox can be kept alive is 24 hours (86_400 seconds) for Pro users and 1 hour (3_600 seconds) for Hobby users.
        :param metadata: Custom metadata for the sandbox
        :param envs: Custom environment variables for the sandbox
        :param secure: Envd is secured with access token and cannot be used without it, defaults to `True`.
        :param allow_internet_access: Allow sandbox to access the internet, defaults to `True`. If set to `False`, it works the same as setting network `deny_out` to `[0.0.0.0/0]`.
        :param mcp: MCP server to enable in the sandbox
        :param network: Sandbox network configuration. ``allow_out``/``deny_out`` may also be a callable receiving a :class:`SandboxNetworkSelectorContext` (``ctx.all_traffic``, ``ctx.rules``) and returning a list of strings. Per-host transform rules are nested under ``network.rules``.
        :param lifecycle: Sandbox lifecycle configuration — ``on_timeout``: ``"kill"`` (default) or ``"pause"``, or an object ``{"action": "pause"|"kill", "keep_memory": bool}`` where ``keep_memory`` (default ``True``) set to ``False`` makes a timeout auto-pause filesystem-only (cold-boots on resume; cannot be combined with ``auto_resume``); ``auto_resume``: ``False`` (default) or ``True`` (only when ``on_timeout`` action is ``"pause"``). Example: ``{"on_timeout": {"action": "pause", "keep_memory": False}}``
        :param volume_mounts: Dictionary mapping mount paths to Volume instances or volume names
        :param logger: Logger used for request and response logging for this sandbox. Accepts any standard library `logging.Logger`. When omitted, no request/response logging is emitted.

        :return: A Sandbox instance for the new sandbox

        Use this method instead of using the constructor to create a new sandbox.
        """
        if not template and mcp is not None:
            template = cls.default_mcp_template
        elif not template:
            template = cls.default_template

        transformed_mounts = None
        if volume_mounts:
            transformed_mounts = [
                SandboxVolumeMountAPI(
                    name=vol.name if isinstance(vol, Volume) else vol,
                    path=path,
                )
                for path, vol in volume_mounts.items()
            ]

        sandbox = cls._create(
            template=template,
            timeout=timeout,
            metadata=metadata,
            envs=envs,
            secure=secure,
            allow_internet_access=allow_internet_access,
            mcp=mcp,
            network=network,
            lifecycle=lifecycle,
            volume_mounts=transformed_mounts,
            logger=logger,
            **opts,
        )

        if mcp is not None:
            token = str(uuid.uuid4())
            sandbox._mcp_token = token

            res = sandbox.commands.run(
                f"mcp-gateway --config {shlex.quote(json.dumps(mcp))}",
                user="root",
                envs={"GATEWAY_ACCESS_TOKEN": token},
            )
            if res.exit_code != 0:
                raise Exception(f"Failed to start MCP gateway: {res.stderr}")

        return sandbox

    @overload
    def connect(
        self,
        timeout: Optional[int] = None,
        **opts: Unpack[ApiParams],
    ) -> Self:
        """
        Connect to a sandbox. If the sandbox is paused, it will be automatically resumed.
        Sandbox must be either running or be paused.

        With sandbox ID you can connect to the same sandbox from different places or environments (serverless functions, etc).

        :param timeout: Timeout for the sandbox in **seconds**
            For running sandboxes, the timeout will update only if the new timeout is longer than the existing one.
        :return: A running sandbox instance

        @example
        ```python
        sandbox = Sandbox.create()
        sandbox.pause()

        # Another code block
        same_sandbox = sandbox.connect()

        :return: A running sandbox instance
        """
        ...

    @overload
    @staticmethod
    def connect(
        sandbox_id: str,
        timeout: Optional[int] = None,
        logger: Optional[logging.Logger] = None,
        **opts: Unpack[ApiParams],
    ) -> "Sandbox":
        """
        Connect to a sandbox. If the sandbox is paused, it will be automatically resumed.
        Sandbox must be either running or be paused.

        With sandbox ID you can connect to the same sandbox from different places or environments (serverless functions, etc).

        :param sandbox_id: Sandbox ID
        :param timeout: Timeout for the sandbox in **seconds**.
            For running sandboxes, the timeout will update only if the new timeout is longer than the existing one.
        :param logger: Logger used for request and response logging for this sandbox. Accepts any standard library `logging.Logger`. When omitted, no request/response logging is emitted.
        :return: A running sandbox instance

        @example
        ```python
        sandbox = Sandbox.create()
        Sandbox.pause(sandbox.sandbox_id)

        # Another code block
        same_sandbox = Sandbox.connect(sandbox.sandbox_id)
        ```
        """
        ...

    @class_method_variant("_cls_connect_sandbox")
    def connect(
        self,
        timeout: Optional[int] = None,
        **opts: Unpack[ApiParams],
    ) -> Self:
        """
        Connect to a sandbox. If the sandbox is paused, it will be automatically resumed.
        Sandbox must be either running or be paused.

        With sandbox ID you can connect to the same sandbox from different places or environments (serverless functions, etc).

        :param timeout: Timeout for the sandbox in **seconds**.
            For running sandboxes, the timeout will update only if the new timeout is longer than the existing one.
        :return: A running sandbox instance

        @example
        ```python
        sandbox = Sandbox.create()
        sandbox.pause()

        # Another code block
        same_sandbox = sandbox.connect()
        ```
        """
        if self.connection_config.debug:
            # Skip connecting to the sandbox in debug mode
            return self

        SandboxApi._cls_connect(
            sandbox_id=self.sandbox_id,
            timeout=timeout,
            **self.connection_config.get_api_params(**opts),
        )

        return self

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        self.kill()

    @overload
    def kill(
        self,
        **opts: Unpack[ApiParams],
    ) -> bool:
        """
        Kill the sandbox.

        :return: `True` if the sandbox was killed, `False` if the sandbox was not found
        """
        ...

    @overload
    @staticmethod
    def kill(
        sandbox_id: str,
        **opts: Unpack[ApiParams],
    ) -> bool:
        """
        Kill the sandbox specified by sandbox ID.

        :param sandbox_id: Sandbox ID

        :return: `True` if the sandbox was killed, `False` if the sandbox was not found
        """
        ...

    @class_method_variant("_cls_kill")
    def kill(
        self,
        **opts: Unpack[ApiParams],
    ) -> bool:
        """
        Kill the sandbox specified by sandbox ID.

        :return: `True` if the sandbox was killed, `False` if the sandbox was not found
        """
        if self.connection_config.debug:
            # Skip killing the sandbox in debug mode
            return True

        return SandboxApi._cls_kill(
            sandbox_id=self.sandbox_id,
            **self.connection_config.get_api_params(**opts),
        )

    @overload
    def set_timeout(
        self,
        timeout: int,
        **opts: Unpack[ApiParams],
    ) -> None:
        """
        Set the timeout of the sandbox.
        This method can extend or reduce the sandbox timeout set when creating the sandbox or from the last call to `.set_timeout`.

        The maximum time a sandbox can be kept alive is 24 hours (86_400 seconds) for Pro users and 1 hour (3_600 seconds) for Hobby users.

        :param timeout: Timeout for the sandbox in **seconds**
        """
        ...

    @overload
    @staticmethod
    def set_timeout(
        sandbox_id: str,
        timeout: int,
        **opts: Unpack[ApiParams],
    ) -> None:
        """
        Set the timeout of the sandbox specified by sandbox ID.
        This method can extend or reduce the sandbox timeout set when creating the sandbox or from the last call to `.set_timeout`.

        The maximum time a sandbox can be kept alive is 24 hours (86_400 seconds) for Pro users and 1 hour (3_600 seconds) for Hobby users.

        :param sandbox_id: Sandbox ID
        :param timeout: Timeout for the sandbox in **seconds**
        """
        ...

    @class_method_variant("_cls_set_timeout")
    def set_timeout(
        self,
        timeout: int,
        **opts: Unpack[ApiParams],
    ) -> None:
        """
        Set the timeout of the sandbox.
        This method can extend or reduce the sandbox timeout set when creating the sandbox or from the last call to `.set_timeout`.

        The maximum time a sandbox can be kept alive is 24 hours (86_400 seconds) for Pro users and 1 hour (3_600 seconds) for Hobby users.

        :param timeout: Timeout for the sandbox in **seconds**

        """

        SandboxApi._cls_set_timeout(
            sandbox_id=self.sandbox_id,
            timeout=timeout,
            **self.connection_config.get_api_params(**opts),
        )

    @overload
    def update_network(
        self,
        network: SandboxNetworkUpdate,
        **opts: Unpack[ApiParams],
    ) -> None:
        """
        Update the network configuration of the sandbox.

        Replaces the current egress configuration atomically — fields that are
        omitted are cleared on the server.

        :param network: New network configuration.
        """
        ...

    @overload
    @staticmethod
    def update_network(
        sandbox_id: str,
        network: SandboxNetworkUpdate,
        **opts: Unpack[ApiParams],
    ) -> None:
        """
        Update the network configuration of the sandbox specified by sandbox ID.

        Replaces the current egress configuration atomically — fields that are
        omitted are cleared on the server.

        :param sandbox_id: Sandbox ID.
        :param network: New network configuration.
        """
        ...

    @class_method_variant("_cls_update_network")
    def update_network(
        self,
        network: SandboxNetworkUpdate,
        **opts: Unpack[ApiParams],
    ) -> None:
        """
        Update the network configuration of the sandbox.

        Replaces the current egress configuration atomically — fields that are
        omitted are cleared on the server.

        :param network: New network configuration.
        """

        SandboxApi._cls_update_network(
            sandbox_id=self.sandbox_id,
            network=network,
            **self.connection_config.get_api_params(**opts),
        )

    @overload
    def get_info(
        self,
        **opts: Unpack[ApiParams],
    ) -> SandboxInfo:
        """
        Get sandbox information like sandbox ID, template, metadata, started at/end at date.

        :return: Sandbox info
        """
        ...

    @overload
    @staticmethod
    def get_info(
        sandbox_id: str,
        **opts: Unpack[ApiParams],
    ) -> SandboxInfo:
        """
        Get sandbox information like sandbox ID, template, metadata, started at/end at date.

        :param sandbox_id: Sandbox ID

        :return: Sandbox info
        """
        ...

    @class_method_variant("_cls_get_info")
    def get_info(
        self,
        **opts: Unpack[ApiParams],
    ) -> SandboxInfo:
        """
        Get sandbox information like sandbox ID, template, metadata, started at/end at date.

        :return: Sandbox info
        """
        return SandboxApi._cls_get_info(
            sandbox_id=self.sandbox_id,
            **self.connection_config.get_api_params(**opts),
        )

    @overload
    def get_metrics(
        self,
        start: Optional[datetime.datetime] = None,
        end: Optional[datetime.datetime] = None,
        **opts: Unpack[ApiParams],
    ) -> List[SandboxMetrics]:
        """
        Get the metrics of the current sandbox.

        :param start: Start time for the metrics, defaults to the start of the sandbox
        :param end: End time for the metrics, defaults to the current time

        :return: List of sandbox metrics containing CPU, memory and disk usage information
        """
        ...

    @overload
    @staticmethod
    def get_metrics(
        sandbox_id: str,
        start: Optional[datetime.datetime] = None,
        end: Optional[datetime.datetime] = None,
        **opts: Unpack[ApiParams],
    ) -> List[SandboxMetrics]:
        """
        Get the metrics of the sandbox specified by sandbox ID.

        :param sandbox_id: Sandbox ID
        :param start: Start time for the metrics, defaults to the start of the sandbox
        :param end: End time for the metrics, defaults to the current time

        :return: List of sandbox metrics containing CPU, memory and disk usage information
        """
        ...

    @class_method_variant("_cls_get_metrics")
    def get_metrics(
        self,
        start: Optional[datetime.datetime] = None,
        end: Optional[datetime.datetime] = None,
        **opts: Unpack[ApiParams],
    ) -> List[SandboxMetrics]:
        """
        Get the metrics of the current sandbox.

        :param start: Start time for the metrics, defaults to the start of the sandbox
        :param end: End time for the metrics, defaults to the current time

        :return: List of sandbox metrics containing CPU, memory and disk usage information
        """
        if self.connection_config.debug:
            # Skip getting the metrics in debug mode
            return []

        if self._envd_version < Version("0.1.5"):
            raise TemplateException(
                "You need to update the template to use the new SDK."
            )

        if self._envd_version < Version("0.2.4"):
            logger.warning(
                "Disk metrics are not supported in this version of the sandbox, please rebuild the template to get disk metrics."
            )

        return SandboxApi._cls_get_metrics(
            sandbox_id=self.sandbox_id,
            start=start,
            end=end,
            **self.connection_config.get_api_params(**opts),
        )

    @overload
    def pause(
        self,
        keep_memory: bool = True,
        **opts: Unpack[ApiParams],
    ) -> bool:
        """
        Pause the sandbox.

        :param keep_memory: When `False`, the in-memory state is dropped and only the filesystem is persisted (no memory snapshot); resuming such a sandbox cold-boots (reboots) it from disk. Defaults to `True`.

        :return: `True` if the sandbox got paused, `False` if the sandbox was already paused
        """
        ...

    @overload
    @staticmethod
    def pause(
        sandbox_id: str,
        keep_memory: bool = True,
        **opts: Unpack[ApiParams],
    ) -> bool:
        """
        Pause the sandbox specified by sandbox ID.

        :param sandbox_id: Sandbox ID
        :param keep_memory: When `False`, the in-memory state is dropped and only the filesystem is persisted (no memory snapshot); resuming such a sandbox cold-boots (reboots) it from disk. Defaults to `True`.

        :return: `True` if the sandbox got paused, `False` if the sandbox was already paused
        """
        ...

    @class_method_variant("_cls_pause")
    def pause(
        self,
        keep_memory: bool = True,
        **opts: Unpack[ApiParams],
    ) -> bool:
        """
        Pause the sandbox.

        :param keep_memory: When `False`, the in-memory state is dropped and only the filesystem is persisted (no memory snapshot); resuming such a sandbox cold-boots (reboots) it from disk, losing running processes and open connections. Defaults to `True` (full memory snapshot).

        :return: `True` if the sandbox got paused, `False` if the sandbox was already paused
        """

        return SandboxApi._cls_pause(
            sandbox_id=self.sandbox_id,
            keep_memory=keep_memory,
            **self.connection_config.get_api_params(**opts),
        )

    @overload
    def beta_pause(
        self,
        keep_memory: bool = True,
        **opts: Unpack[ApiParams],
    ) -> bool: ...

    @overload
    @staticmethod
    def beta_pause(
        sandbox_id: str,
        keep_memory: bool = True,
        **opts: Unpack[ApiParams],
    ) -> bool: ...

    @class_method_variant("_cls_pause")
    def beta_pause(
        self,
        keep_memory: bool = True,
        **opts: Unpack[ApiParams],
    ) -> bool:
        """
        :deprecated: Use `pause()` instead.

        :return: `True` if the sandbox got paused, `False` if the sandbox was already paused
        """
        return self.pause(keep_memory=keep_memory, **opts)

    @overload
    def create_snapshot(
        self,
        name: Optional[str] = None,
        **opts: Unpack[ApiParams],
    ) -> SnapshotInfo:
        """
        Create a snapshot of the sandbox's current state.

        The sandbox will be paused while the snapshot is being created.
        The snapshot can be used to create new sandboxes with the same filesystem and state.
        Snapshots are persistent and survive sandbox deletion.

        Use the returned `snapshot_id` with `Sandbox.create(snapshot_id)` to create a new sandbox from the snapshot.

        :param name: Optional name for the snapshot template. If a snapshot template with this name already exists, a new build will be assigned to the existing template instead of creating a new one.

        :return: Snapshot information including the snapshot ID and names
        """
        ...

    @overload
    @staticmethod
    def create_snapshot(
        sandbox_id: str,
        name: Optional[str] = None,
        **opts: Unpack[ApiParams],
    ) -> SnapshotInfo:
        """
        Create a snapshot from the sandbox specified by sandbox ID.

        The sandbox will be paused while the snapshot is being created.

        :param sandbox_id: Sandbox ID
        :param name: Optional name for the snapshot template. If a snapshot template with this name already exists, a new build will be assigned to the existing template instead of creating a new one.

        :return: Snapshot information including the snapshot ID and names
        """
        ...

    @class_method_variant("_cls_create_snapshot")
    def create_snapshot(
        self,
        name: Optional[str] = None,
        **opts: Unpack[ApiParams],
    ) -> SnapshotInfo:
        """
        Create a snapshot of the sandbox's current state.

        The sandbox will be paused while the snapshot is being created.
        The snapshot can be used to create new sandboxes with the same filesystem and state.
        Snapshots are persistent and survive sandbox deletion.

        Use the returned `snapshot_id` with `Sandbox.create(snapshot_id)` to create a new sandbox from the snapshot.

        :param name: Optional name for the snapshot template. If a snapshot template with this name already exists, a new build will be assigned to the existing template instead of creating a new one.

        :return: Snapshot information including the snapshot ID and names
        """
        return SandboxApi._cls_create_snapshot(
            sandbox_id=self.sandbox_id,
            name=name,
            **self.connection_config.get_api_params(**opts),
        )

    @overload
    def list_snapshots(
        self,
        limit: Optional[int] = None,
        next_token: Optional[str] = None,
        **opts: Unpack[ApiParams],
    ) -> SnapshotPaginator:
        """
        List snapshots for this sandbox.

        :param limit: Maximum number of snapshots to return per page
        :param next_token: Token for pagination

        :return: Paginator for listing snapshots
        """
        ...

    @overload
    @staticmethod
    def list_snapshots(
        sandbox_id: Optional[str] = None,
        limit: Optional[int] = None,
        next_token: Optional[str] = None,
        **opts: Unpack[ApiParams],
    ) -> SnapshotPaginator:
        """
        List all snapshots.

        :param sandbox_id: Filter snapshots by source sandbox ID
        :param limit: Maximum number of snapshots to return per page
        :param next_token: Token for pagination

        :return: Paginator for listing snapshots
        """
        ...

    @class_method_variant("_cls_list_snapshots")
    def list_snapshots(
        self,
        limit: Optional[int] = None,
        next_token: Optional[str] = None,
        **opts: Unpack[ApiParams],
    ) -> SnapshotPaginator:
        """
        List snapshots for this sandbox.

        :param limit: Maximum number of snapshots to return per page
        :param next_token: Token for pagination

        :return: Paginator for listing snapshots
        """
        return SnapshotPaginator(
            sandbox_id=self.sandbox_id,
            limit=limit,
            next_token=next_token,
            **self.connection_config.get_api_params(**opts),
        )

    @staticmethod
    def _cls_list_snapshots(
        sandbox_id: Optional[str] = None,
        limit: Optional[int] = None,
        next_token: Optional[str] = None,
        **opts: Unpack[ApiParams],
    ) -> SnapshotPaginator:
        return SnapshotPaginator(
            sandbox_id=sandbox_id,
            limit=limit,
            next_token=next_token,
            **opts,
        )

    @staticmethod
    def delete_snapshot(
        snapshot_id: str,
        **opts: Unpack[ApiParams],
    ) -> bool:
        """
        Delete a snapshot.

        :param snapshot_id: Snapshot ID
        :return: `True` if the snapshot was deleted, `False` if it was not found
        """
        return SandboxApi._cls_delete_snapshot(
            snapshot_id=snapshot_id,
            **opts,
        )

    def get_mcp_token(self) -> Optional[str]:
        """
        Get the MCP token for the sandbox.

        :return: MCP token for the sandbox, or None if MCP is not enabled.
        """
        if not self._mcp_token:
            self._mcp_token = self.files.read("/etc/mcp-gateway/.token", user="root")
        return self._mcp_token

    @classmethod
    def _cls_connect_sandbox(
        cls,
        sandbox_id: str,
        timeout: Optional[int] = None,
        logger: Optional[logging.Logger] = None,
        **opts: Unpack[ApiParams],
    ) -> Self:
        debug = ConnectionConfig(**opts).debug
        if debug:
            sandbox_domain = None
            envd_version = ENVD_DEBUG_FALLBACK
            envd_access_token = None
            traffic_access_token = None
        else:
            sandbox = SandboxApi._cls_connect(
                sandbox_id=sandbox_id,
                timeout=timeout,
                logger=logger,
                **opts,
            )

            sandbox_id = sandbox.sandbox_id
            sandbox_domain = sandbox.sandbox_domain
            envd_version = Version(sandbox.envd_version)
            envd_access_token = sandbox.envd_access_token
            traffic_access_token = sandbox.traffic_access_token

        sandbox_headers = {
            "E2b-Sandbox-Id": sandbox_id,
            "E2b-Sandbox-Port": str(ConnectionConfig.envd_port),
        }
        if envd_access_token is not None and not isinstance(envd_access_token, Unset):
            sandbox_headers["X-Access-Token"] = envd_access_token

        connection_config = ConnectionConfig(
            extra_sandbox_headers=sandbox_headers,
            logger=logger,
            **opts,
        )

        return cls(
            sandbox_id=sandbox_id,
            sandbox_domain=sandbox_domain,
            envd_version=envd_version,
            envd_access_token=envd_access_token,
            traffic_access_token=traffic_access_token,
            connection_config=connection_config,
        )

    @classmethod
    def _create(
        cls,
        template: Optional[str],
        timeout: Optional[int],
        metadata: Optional[Dict[str, str]],
        envs: Optional[Dict[str, str]],
        secure: bool,
        allow_internet_access: bool,
        mcp: Optional[McpServer] = None,
        network: Optional[SandboxNetworkOpts] = None,
        lifecycle: Optional[SandboxLifecycle] = None,
        volume_mounts: Optional[list] = None,
        logger: Optional[logging.Logger] = None,
        **opts: Unpack[ApiParams],
    ) -> Self:
        extra_sandbox_headers = {}

        debug = ConnectionConfig(**opts).debug
        if debug:
            sandbox_id = "debug_sandbox_id"
            sandbox_domain = None
            envd_version = ENVD_DEBUG_FALLBACK
            envd_access_token = None
            traffic_access_token = None
        else:
            response = SandboxApi._create_sandbox(
                template=template or cls.default_template,
                timeout=timeout or cls.default_sandbox_timeout,
                metadata=metadata,
                env_vars=envs,
                secure=secure,
                allow_internet_access=allow_internet_access,
                mcp=mcp,
                network=network,
                lifecycle=lifecycle,
                volume_mounts=volume_mounts,
                logger=logger,
                **opts,
            )

            sandbox_id = response.sandbox_id
            sandbox_domain = response.sandbox_domain
            envd_version = Version(response.envd_version)
            envd_access_token = response.envd_access_token
            traffic_access_token = response.traffic_access_token

            if envd_access_token is not None and not isinstance(
                envd_access_token, Unset
            ):
                extra_sandbox_headers["X-Access-Token"] = envd_access_token

        extra_sandbox_headers["E2b-Sandbox-Id"] = sandbox_id
        extra_sandbox_headers["E2b-Sandbox-Port"] = str(ConnectionConfig.envd_port)

        connection_config = ConnectionConfig(
            extra_sandbox_headers=extra_sandbox_headers,
            logger=logger,
            **opts,
        )

        return cls(
            sandbox_id=sandbox_id,
            sandbox_domain=sandbox_domain,
            envd_version=envd_version,
            envd_access_token=envd_access_token,
            traffic_access_token=traffic_access_token,
            connection_config=connection_config,
        )
