# This file was auto-generated by Fern from our API Definition.

import typing

from ..core.client_wrapper import AsyncClientWrapper, SyncClientWrapper
from ..core.request_options import RequestOptions
from ..types.response import Response
from ..types.response_active_shell_sessions_result import ResponseActiveShellSessionsResult
from ..types.response_shell_command_result import ResponseShellCommandResult
from ..types.response_shell_create_session_response import ResponseShellCreateSessionResponse
from ..types.response_shell_kill_result import ResponseShellKillResult
from ..types.response_shell_view_result import ResponseShellViewResult
from ..types.response_shell_wait_result import ResponseShellWaitResult
from ..types.response_shell_write_result import ResponseShellWriteResult
from ..types.response_str import ResponseStr
from .raw_client import AsyncRawShellClient, RawShellClient

# this is used as the default value for optional parameters
OMIT = typing.cast(typing.Any, ...)


class ShellClient:
    def __init__(self, *, client_wrapper: SyncClientWrapper):
        self._raw_client = RawShellClient(client_wrapper=client_wrapper)

    @property
    def with_raw_response(self) -> RawShellClient:
        """
        Retrieves a raw implementation of this client that returns raw responses.

        Returns
        -------
        RawShellClient
        """
        return self._raw_client

    def exec_command(
        self,
        *,
        command: str,
        id: typing.Optional[str] = OMIT,
        exec_dir: typing.Optional[str] = OMIT,
        async_mode: typing.Optional[bool] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> ResponseShellCommandResult:
        """
        Execute command in the specified shell session
        Supports SSE streaming if Accept header contains 'text/event-stream'

        Parameters
        ----------
        command : str
            Shell command to execute

        id : typing.Optional[str]
            Unique identifier of the target shell session, if not provided, one will be automatically created

        exec_dir : typing.Optional[str]
            Working directory for command execution (must use absolute path)

        async_mode : typing.Optional[bool]
            Whether to execute command asynchronously (default: False for async, False for synchronous execution)

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        ResponseShellCommandResult
            Successful Response

        Examples
        --------
        from agent_sandbox import Sandbox

        client = Sandbox(
            base_url="https://yourhost.com/path/to/api",
        )
        client.shell.exec_command(
            command="command",
        )
        """
        _response = self._raw_client.exec_command(
            command=command, id=id, exec_dir=exec_dir, async_mode=async_mode, request_options=request_options
        )
        return _response.data

    def view(self, *, id: str, request_options: typing.Optional[RequestOptions] = None) -> ResponseShellViewResult:
        """
        View output of the specified shell session
        Supports SSE streaming if Accept header contains 'text/event-stream'

        Parameters
        ----------
        id : str
            Unique identifier of the target shell session

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        ResponseShellViewResult
            Successful Response

        Examples
        --------
        from agent_sandbox import Sandbox

        client = Sandbox(
            base_url="https://yourhost.com/path/to/api",
        )
        client.shell.view(
            id="id",
        )
        """
        _response = self._raw_client.view(id=id, request_options=request_options)
        return _response.data

    def wait_for_process(
        self, *, id: str, seconds: typing.Optional[int] = OMIT, request_options: typing.Optional[RequestOptions] = None
    ) -> ResponseShellWaitResult:
        """
        Wait for the process in the specified shell session to return

        Parameters
        ----------
        id : str
            Unique identifier of the target shell session

        seconds : typing.Optional[int]
            Wait time (seconds)

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        ResponseShellWaitResult
            Successful Response

        Examples
        --------
        from agent_sandbox import Sandbox

        client = Sandbox(
            base_url="https://yourhost.com/path/to/api",
        )
        client.shell.wait_for_process(
            id="id",
        )
        """
        _response = self._raw_client.wait_for_process(id=id, seconds=seconds, request_options=request_options)
        return _response.data

    def write_to_process(
        self, *, id: str, input: str, press_enter: bool, request_options: typing.Optional[RequestOptions] = None
    ) -> ResponseShellWriteResult:
        """
        Write input to the process in the specified shell session

        Parameters
        ----------
        id : str
            Unique identifier of the target shell session

        input : str
            Input content to write to the process

        press_enter : bool
            Whether to press enter key after input

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        ResponseShellWriteResult
            Successful Response

        Examples
        --------
        from agent_sandbox import Sandbox

        client = Sandbox(
            base_url="https://yourhost.com/path/to/api",
        )
        client.shell.write_to_process(
            id="id",
            input="input",
            press_enter=True,
        )
        """
        _response = self._raw_client.write_to_process(
            id=id, input=input, press_enter=press_enter, request_options=request_options
        )
        return _response.data

    def kill_process(
        self, *, id: str, request_options: typing.Optional[RequestOptions] = None
    ) -> ResponseShellKillResult:
        """
        Terminate the process in the specified shell session

        Parameters
        ----------
        id : str
            Unique identifier of the target shell session

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        ResponseShellKillResult
            Successful Response

        Examples
        --------
        from agent_sandbox import Sandbox

        client = Sandbox(
            base_url="https://yourhost.com/path/to/api",
        )
        client.shell.kill_process(
            id="id",
        )
        """
        _response = self._raw_client.kill_process(id=id, request_options=request_options)
        return _response.data

    def create_session(
        self,
        *,
        id: typing.Optional[str] = OMIT,
        exec_dir: typing.Optional[str] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> ResponseShellCreateSessionResponse:
        """
        Create a new shell session and return its ID
        If id already exists, return the existing session

        Parameters
        ----------
        id : typing.Optional[str]
            Unique identifier for the shell session, auto-generated if not provided

        exec_dir : typing.Optional[str]
            Working directory for the new session (must use absolute path)

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        ResponseShellCreateSessionResponse
            Successful Response

        Examples
        --------
        from agent_sandbox import Sandbox

        client = Sandbox(
            base_url="https://yourhost.com/path/to/api",
        )
        client.shell.create_session()
        """
        _response = self._raw_client.create_session(id=id, exec_dir=exec_dir, request_options=request_options)
        return _response.data

    def get_terminal_url(self, *, request_options: typing.Optional[RequestOptions] = None) -> ResponseStr:
        """
        Create a new shell session and return the terminal URL

        Parameters
        ----------
        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        ResponseStr
            Successful Response

        Examples
        --------
        from agent_sandbox import Sandbox

        client = Sandbox(
            base_url="https://yourhost.com/path/to/api",
        )
        client.shell.get_terminal_url()
        """
        _response = self._raw_client.get_terminal_url(request_options=request_options)
        return _response.data

    def list_sessions(
        self, *, request_options: typing.Optional[RequestOptions] = None
    ) -> ResponseActiveShellSessionsResult:
        """
        List all active shell sessions

        Parameters
        ----------
        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        ResponseActiveShellSessionsResult
            Successful Response

        Examples
        --------
        from agent_sandbox import Sandbox

        client = Sandbox(
            base_url="https://yourhost.com/path/to/api",
        )
        client.shell.list_sessions()
        """
        _response = self._raw_client.list_sessions(request_options=request_options)
        return _response.data

    def cleanup_all_sessions(self, *, request_options: typing.Optional[RequestOptions] = None) -> Response:
        """
        Cleanup all active shell sessions

        Parameters
        ----------
        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        Response
            Successful Response

        Examples
        --------
        from agent_sandbox import Sandbox

        client = Sandbox(
            base_url="https://yourhost.com/path/to/api",
        )
        client.shell.cleanup_all_sessions()
        """
        _response = self._raw_client.cleanup_all_sessions(request_options=request_options)
        return _response.data

    def cleanup_session(self, session_id: str, *, request_options: typing.Optional[RequestOptions] = None) -> Response:
        """
        Manually cleanup a specific shell session

        Parameters
        ----------
        session_id : str

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        Response
            Successful Response

        Examples
        --------
        from agent_sandbox import Sandbox

        client = Sandbox(
            base_url="https://yourhost.com/path/to/api",
        )
        client.shell.cleanup_session(
            session_id="session_id",
        )
        """
        _response = self._raw_client.cleanup_session(session_id, request_options=request_options)
        return _response.data


class AsyncShellClient:
    def __init__(self, *, client_wrapper: AsyncClientWrapper):
        self._raw_client = AsyncRawShellClient(client_wrapper=client_wrapper)

    @property
    def with_raw_response(self) -> AsyncRawShellClient:
        """
        Retrieves a raw implementation of this client that returns raw responses.

        Returns
        -------
        AsyncRawShellClient
        """
        return self._raw_client

    async def exec_command(
        self,
        *,
        command: str,
        id: typing.Optional[str] = OMIT,
        exec_dir: typing.Optional[str] = OMIT,
        async_mode: typing.Optional[bool] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> ResponseShellCommandResult:
        """
        Execute command in the specified shell session
        Supports SSE streaming if Accept header contains 'text/event-stream'

        Parameters
        ----------
        command : str
            Shell command to execute

        id : typing.Optional[str]
            Unique identifier of the target shell session, if not provided, one will be automatically created

        exec_dir : typing.Optional[str]
            Working directory for command execution (must use absolute path)

        async_mode : typing.Optional[bool]
            Whether to execute command asynchronously (default: False for async, False for synchronous execution)

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        ResponseShellCommandResult
            Successful Response

        Examples
        --------
        import asyncio

        from agent_sandbox import AsyncSandbox

        client = AsyncSandbox(
            base_url="https://yourhost.com/path/to/api",
        )


        async def main() -> None:
            await client.shell.exec_command(
                command="command",
            )


        asyncio.run(main())
        """
        _response = await self._raw_client.exec_command(
            command=command, id=id, exec_dir=exec_dir, async_mode=async_mode, request_options=request_options
        )
        return _response.data

    async def view(
        self, *, id: str, request_options: typing.Optional[RequestOptions] = None
    ) -> ResponseShellViewResult:
        """
        View output of the specified shell session
        Supports SSE streaming if Accept header contains 'text/event-stream'

        Parameters
        ----------
        id : str
            Unique identifier of the target shell session

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        ResponseShellViewResult
            Successful Response

        Examples
        --------
        import asyncio

        from agent_sandbox import AsyncSandbox

        client = AsyncSandbox(
            base_url="https://yourhost.com/path/to/api",
        )


        async def main() -> None:
            await client.shell.view(
                id="id",
            )


        asyncio.run(main())
        """
        _response = await self._raw_client.view(id=id, request_options=request_options)
        return _response.data

    async def wait_for_process(
        self, *, id: str, seconds: typing.Optional[int] = OMIT, request_options: typing.Optional[RequestOptions] = None
    ) -> ResponseShellWaitResult:
        """
        Wait for the process in the specified shell session to return

        Parameters
        ----------
        id : str
            Unique identifier of the target shell session

        seconds : typing.Optional[int]
            Wait time (seconds)

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        ResponseShellWaitResult
            Successful Response

        Examples
        --------
        import asyncio

        from agent_sandbox import AsyncSandbox

        client = AsyncSandbox(
            base_url="https://yourhost.com/path/to/api",
        )


        async def main() -> None:
            await client.shell.wait_for_process(
                id="id",
            )


        asyncio.run(main())
        """
        _response = await self._raw_client.wait_for_process(id=id, seconds=seconds, request_options=request_options)
        return _response.data

    async def write_to_process(
        self, *, id: str, input: str, press_enter: bool, request_options: typing.Optional[RequestOptions] = None
    ) -> ResponseShellWriteResult:
        """
        Write input to the process in the specified shell session

        Parameters
        ----------
        id : str
            Unique identifier of the target shell session

        input : str
            Input content to write to the process

        press_enter : bool
            Whether to press enter key after input

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        ResponseShellWriteResult
            Successful Response

        Examples
        --------
        import asyncio

        from agent_sandbox import AsyncSandbox

        client = AsyncSandbox(
            base_url="https://yourhost.com/path/to/api",
        )


        async def main() -> None:
            await client.shell.write_to_process(
                id="id",
                input="input",
                press_enter=True,
            )


        asyncio.run(main())
        """
        _response = await self._raw_client.write_to_process(
            id=id, input=input, press_enter=press_enter, request_options=request_options
        )
        return _response.data

    async def kill_process(
        self, *, id: str, request_options: typing.Optional[RequestOptions] = None
    ) -> ResponseShellKillResult:
        """
        Terminate the process in the specified shell session

        Parameters
        ----------
        id : str
            Unique identifier of the target shell session

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        ResponseShellKillResult
            Successful Response

        Examples
        --------
        import asyncio

        from agent_sandbox import AsyncSandbox

        client = AsyncSandbox(
            base_url="https://yourhost.com/path/to/api",
        )


        async def main() -> None:
            await client.shell.kill_process(
                id="id",
            )


        asyncio.run(main())
        """
        _response = await self._raw_client.kill_process(id=id, request_options=request_options)
        return _response.data

    async def create_session(
        self,
        *,
        id: typing.Optional[str] = OMIT,
        exec_dir: typing.Optional[str] = OMIT,
        request_options: typing.Optional[RequestOptions] = None,
    ) -> ResponseShellCreateSessionResponse:
        """
        Create a new shell session and return its ID
        If id already exists, return the existing session

        Parameters
        ----------
        id : typing.Optional[str]
            Unique identifier for the shell session, auto-generated if not provided

        exec_dir : typing.Optional[str]
            Working directory for the new session (must use absolute path)

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        ResponseShellCreateSessionResponse
            Successful Response

        Examples
        --------
        import asyncio

        from agent_sandbox import AsyncSandbox

        client = AsyncSandbox(
            base_url="https://yourhost.com/path/to/api",
        )


        async def main() -> None:
            await client.shell.create_session()


        asyncio.run(main())
        """
        _response = await self._raw_client.create_session(id=id, exec_dir=exec_dir, request_options=request_options)
        return _response.data

    async def get_terminal_url(self, *, request_options: typing.Optional[RequestOptions] = None) -> ResponseStr:
        """
        Create a new shell session and return the terminal URL

        Parameters
        ----------
        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        ResponseStr
            Successful Response

        Examples
        --------
        import asyncio

        from agent_sandbox import AsyncSandbox

        client = AsyncSandbox(
            base_url="https://yourhost.com/path/to/api",
        )


        async def main() -> None:
            await client.shell.get_terminal_url()


        asyncio.run(main())
        """
        _response = await self._raw_client.get_terminal_url(request_options=request_options)
        return _response.data

    async def list_sessions(
        self, *, request_options: typing.Optional[RequestOptions] = None
    ) -> ResponseActiveShellSessionsResult:
        """
        List all active shell sessions

        Parameters
        ----------
        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        ResponseActiveShellSessionsResult
            Successful Response

        Examples
        --------
        import asyncio

        from agent_sandbox import AsyncSandbox

        client = AsyncSandbox(
            base_url="https://yourhost.com/path/to/api",
        )


        async def main() -> None:
            await client.shell.list_sessions()


        asyncio.run(main())
        """
        _response = await self._raw_client.list_sessions(request_options=request_options)
        return _response.data

    async def cleanup_all_sessions(self, *, request_options: typing.Optional[RequestOptions] = None) -> Response:
        """
        Cleanup all active shell sessions

        Parameters
        ----------
        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        Response
            Successful Response

        Examples
        --------
        import asyncio

        from agent_sandbox import AsyncSandbox

        client = AsyncSandbox(
            base_url="https://yourhost.com/path/to/api",
        )


        async def main() -> None:
            await client.shell.cleanup_all_sessions()


        asyncio.run(main())
        """
        _response = await self._raw_client.cleanup_all_sessions(request_options=request_options)
        return _response.data

    async def cleanup_session(
        self, session_id: str, *, request_options: typing.Optional[RequestOptions] = None
    ) -> Response:
        """
        Manually cleanup a specific shell session

        Parameters
        ----------
        session_id : str

        request_options : typing.Optional[RequestOptions]
            Request-specific configuration.

        Returns
        -------
        Response
            Successful Response

        Examples
        --------
        import asyncio

        from agent_sandbox import AsyncSandbox

        client = AsyncSandbox(
            base_url="https://yourhost.com/path/to/api",
        )


        async def main() -> None:
            await client.shell.cleanup_session(
                session_id="session_id",
            )


        asyncio.run(main())
        """
        _response = await self._raw_client.cleanup_session(session_id, request_options=request_options)
        return _response.data
