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}"

_api_url property

_api_url: str

_config instance-attribute

_config = RemoteRuntimeConfig(**kwargs)

_headers property

_headers: dict[str, str]

Request headers to use for authentication.

logger instance-attribute

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

_get_timeout

_get_timeout(timeout: float | None = None) -> float

Source code in swerex/runtime/remote.py

def _get_timeout(self, timeout: float | None = None) -> float:
    if timeout is None:
        return self._config.timeout
    return timeout

_handle_response_errors

_handle_response_errors(response: Response) -> None

Raise exceptions found in the request response.

Source code in swerex/runtime/remote.py

def _handle_response_errors(self, response: requests.Response) -> None:
    """Raise exceptions found in the request response."""
    if response.status_code == 511:
        exc_transfer = _ExceptionTransfer(**response.json()["swerexception"])
        self._handle_transfer_exception(exc_transfer)
    try:
        response.raise_for_status()
    except Exception:
        self.logger.critical("Received error response: %s", response.json())
        raise

_handle_transfer_exception

_handle_transfer_exception(exc_transfer: _ExceptionTransfer) -> None

Reraise exceptions that were thrown on the remote.

Source code in swerex/runtime/remote.py

def _handle_transfer_exception(self, exc_transfer: _ExceptionTransfer) -> None:
    """Reraise exceptions that were thrown on the remote."""
    if exc_transfer.traceback:
        self.logger.critical("Traceback: \n%s", exc_transfer.traceback)
    module, _, exc_name = exc_transfer.class_path.rpartition(".")
    print(module, exc_name)
    if module == "builtins":
        module_obj = __builtins__
    else:
        if module not in sys.modules:
            self.logger.debug("Module %s not in sys.modules, trying to import it", module)
            try:
                __import__(module)
            except ImportError:
                self.logger.debug("Failed to import module %s", module)
                exc = SwerexException(exc_transfer.message)
                raise exc from None
        module_obj = sys.modules[module]
    try:
        if isinstance(module_obj, dict):
            # __builtins__, sometimes
            exception = module_obj[exc_name](exc_transfer.message)
        else:
            exception = getattr(module_obj, exc_name)(exc_transfer.message)
    except (AttributeError, TypeError):
        self.logger.error(
            f"Could not initialize transferred exception: {exc_transfer.class_path!r}. "
            f"Transfer object: {exc_transfer}"
        )
        exception = SwerexException(exc_transfer.message)
    exception.extra_info = exc_transfer.extra_info
    raise exception from None

_request

_request(endpoint: str, request: BaseModel | None, output_class: Any)

Small helper to make requests to the server and handle errors and output.

Source code in swerex/runtime/remote.py

def _request(self, endpoint: str, request: BaseModel | None, output_class: Any):
    """Small helper to make requests to the server and handle errors and output."""
    response = requests.post(
        f"{self._api_url}/{endpoint}", json=request.model_dump() if request else None, headers=self._headers
    )
    self._handle_response_errors(response)
    return output_class(**response.json())

close async

close() -> CloseResponse

Closes the runtime.

Source code in swerex/runtime/remote.py

async def close(self) -> CloseResponse:
    """Closes the runtime."""
    return 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 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 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 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:
        response = requests.get(
            f"{self._api_url}/is_alive", headers=self._headers, timeout=self._get_timeout(timeout)
        )
        if response.status_code == 200:
            return IsAliveResponse(**response.json())
        elif response.status_code == 511:
            exc_transfer = _ExceptionTransfer(**response.json()["swerexception"])
            self._handle_transfer_exception(exc_transfer)
        msg = (
            f"Status code {response.status_code} from {self._api_url}/is_alive. "
            f"Message: {response.json().get('detail')}"
        )
        return IsAliveResponse(is_alive=False, message=msg)
    except requests.RequestException:
        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 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 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)
    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)
            files = {"file": zip_path.open("rb")}
            data = {"target_path": request.target_path, "unzip": "true"}
            response = requests.post(f"{self._api_url}/upload", files=files, data=data, headers=self._headers)
            self._handle_response_errors(response)
            return UploadResponse(**response.json())
    elif source.is_file():
        self.logger.debug("Uploading file from %s to %s", source, request.target_path)
        files = {"file": source.open("rb")}
        data = {"target_path": request.target_path, "unzip": "false"}
        response = requests.post(f"{self._api_url}/upload", files=files, data=data, headers=self._headers)
        self._handle_response_errors(response)
        return UploadResponse(**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 self._request("write_file", request, WriteFileResponse)