Remote

swerex.runtime.remote.RemoteRuntime

RemoteRuntime(*, logger: Logger | None = None, **kwargs: Any)

Bases: AbstractRuntime

A runtime that connects to a remote server.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Keyword arguments to pass to the RemoteRuntimeConfig constructor.	{}

Source code in swerex/runtime/remote.py

def __init__(
    self,
    *,
    logger: logging.Logger | None = None,
    **kwargs: Any,
):
    """A runtime that connects to a remote server.

    Args:
        **kwargs: Keyword arguments to pass to the `RemoteRuntimeConfig` constructor.
    """
    self._config = RemoteRuntimeConfig(**kwargs)
    self.logger = logger or get_logger("rex-runtime")
    if not self._config.host.startswith("http"):
        self.logger.warning("Host %s does not start with http, adding http://", self._config.host)
        self._config.host = f"http://{self._config.host}"

logger instance-attribute

logger = logger or get_logger('rex-runtime')

close async

close() -> CloseResponse

Closes the runtime.

Source code in swerex/runtime/remote.py

async def close(self) -> CloseResponse:
    """Closes the runtime."""
    return await self._request("close", None, CloseResponse)

close_session async

close_session(request: CloseSessionRequest) -> CloseSessionResponse

Closes a shell session.

Source code in swerex/runtime/remote.py

async def close_session(self, request: CloseSessionRequest) -> CloseSessionResponse:
    """Closes a shell session."""
    return await self._request("close_session", request, CloseSessionResponse)

create_session async

create_session(request: CreateSessionRequest) -> CreateSessionResponse

Creates a new session.

Source code in swerex/runtime/remote.py

async def create_session(self, request: CreateSessionRequest) -> CreateSessionResponse:
    """Creates a new session."""
    return await self._request("create_session", request, CreateSessionResponse)

execute async

execute(command: Command) -> CommandResponse

Executes a command (independent of any shell session).

Source code in swerex/runtime/remote.py

async def execute(self, command: Command) -> CommandResponse:
    """Executes a command (independent of any shell session)."""
    return await self._request("execute", command, CommandResponse)

from_config classmethod

from_config(config: RemoteRuntimeConfig) -> Self

Source code in swerex/runtime/remote.py

@classmethod
def from_config(cls, config: RemoteRuntimeConfig) -> Self:
    return cls(**config.model_dump())

is_alive async

is_alive(*, timeout: float | None = None) -> IsAliveResponse

Checks if the runtime is alive.

Internal server errors are thrown, everything else just has us return False together with the message.

Source code in swerex/runtime/remote.py

async def is_alive(self, *, timeout: float | None = None) -> IsAliveResponse:
    """Checks if the runtime is alive.

    Internal server errors are thrown, everything else just has us return False
    together with the message.
    """
    try:
        async with aiohttp.ClientSession(connector=aiohttp.TCPConnector(force_close=True)) as session:
            timeout_value = self._get_timeout(timeout)
            async with session.get(
                f"{self._api_url}/is_alive",
                headers=self._headers,
                timeout=aiohttp.ClientTimeout(total=timeout_value),
            ) as response:
                if response.status == 200:
                    data = await response.json()
                    return IsAliveResponse(**data)
                elif response.status == 511:
                    data = await response.json()
                    exc_transfer = _ExceptionTransfer(**data["swerexception"])
                    self._handle_transfer_exception(exc_transfer)

                data = await response.json()
                msg = f"Status code {response.status} from {self._api_url}/is_alive. Message: {data.get('detail')}"
                return IsAliveResponse(is_alive=False, message=msg)
    except aiohttp.ClientError:
        msg = f"Failed to connect to {self._config.host}\n"
        msg += traceback.format_exc()
        return IsAliveResponse(is_alive=False, message=msg)
    except Exception:
        msg = f"Failed to connect to {self._config.host}\n"
        msg += traceback.format_exc()
        return IsAliveResponse(is_alive=False, message=msg)

read_file async

read_file(request: ReadFileRequest) -> ReadFileResponse

Reads a file

Source code in swerex/runtime/remote.py

async def read_file(self, request: ReadFileRequest) -> ReadFileResponse:
    """Reads a file"""
    return await self._request("read_file", request, ReadFileResponse)

run_in_session async

run_in_session(action: Action) -> Observation

Runs a command in a session.

Source code in swerex/runtime/remote.py

async def run_in_session(self, action: Action) -> Observation:
    """Runs a command in a session."""
    return await self._request("run_in_session", action, Observation)

upload async

upload(request: UploadRequest) -> UploadResponse

Uploads a file

Source code in swerex/runtime/remote.py

async def upload(self, request: UploadRequest) -> UploadResponse:
    """Uploads a file"""
    source = Path(request.source_path).resolve()
    self.logger.debug("Uploading file from %s to %s", request.source_path, request.target_path)

    async with aiohttp.ClientSession(connector=aiohttp.TCPConnector(force_close=True)) as session:
        if source.is_dir():
            # Ignore cleanup errors: See https://github.com/SWE-agent/SWE-agent/issues/1005
            with tempfile.TemporaryDirectory(ignore_cleanup_errors=True) as temp_dir:
                zip_path = Path(temp_dir) / "zipped_transfer.zip"
                shutil.make_archive(str(zip_path.with_suffix("")), "zip", source)
                self.logger.debug("Created zip file at %s", zip_path)

                with open(zip_path, "rb") as f:
                    data = aiohttp.FormData()
                    data.add_field("file", f, filename=zip_path.name, content_type="application/zip")
                    data.add_field("target_path", request.target_path)
                    data.add_field("unzip", "true")

                    async with session.post(
                        f"{self._api_url}/upload", data=data, headers=self._headers
                    ) as response:
                        await self._handle_response_errors(response)
                        return UploadResponse(**(await response.json()))
        elif source.is_file():
            self.logger.debug("Uploading file from %s to %s", source, request.target_path)

            with open(source, "rb") as f:
                data = aiohttp.FormData()
                data.add_field("file", f, filename=source.name)
                data.add_field("target_path", request.target_path)
                data.add_field("unzip", "false")

                async with session.post(f"{self._api_url}/upload", data=data, headers=self._headers) as response:
                    await self._handle_response_errors(response)
                    return UploadResponse(**(await response.json()))
        else:
            msg = f"Source path {source} is not a file or directory"
            raise ValueError(msg)

wait_until_alive async

wait_until_alive(*, timeout: float = 60.0)

Source code in swerex/runtime/remote.py

async def wait_until_alive(self, *, timeout: float = 60.0):
    return await _wait_until_alive(self.is_alive, timeout=timeout)

write_file async

write_file(request: WriteFileRequest) -> WriteFileResponse

Writes a file

Source code in swerex/runtime/remote.py

async def write_file(self, request: WriteFileRequest) -> WriteFileResponse:
    """Writes a file"""
    return await self._request("write_file", request, WriteFileResponse)

SWE-agent Mini-SWE-Agent SWE-smith SWE-bench sb-cli