Local

swerex.runtime.local.LocalRuntime

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

Bases: AbstractRuntime

A Runtime that runs locally and actually executes commands in a shell. If you are deploying to Modal/Fargate/etc., this class will be running within the docker container on Modal/Fargate/etc.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Keyword arguments (see LocalRuntimeConfig for details).	{}

Source code in swerex/runtime/local.py

def __init__(self, *, logger: logging.Logger | None = None, **kwargs: Any):
    """A Runtime that runs locally and actually executes commands in a shell.
    If you are deploying to Modal/Fargate/etc., this class will be running within the docker container
    on Modal/Fargate/etc.

    Args:
        **kwargs: Keyword arguments (see `LocalRuntimeConfig` for details).
    """
    self._config = LocalRuntimeConfig(**kwargs)
    self._sessions: dict[str, Session] = {}
    self.logger = logger or get_logger("rex-runtime")

logger instance-attribute

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

sessions property

sessions: dict[str, Session]

close async

close() -> CloseResponse

Closes the runtime.

Source code in swerex/runtime/local.py

async def close(self) -> CloseResponse:
    """Closes the runtime."""
    for session in self.sessions.values():
        await session.close()
    return CloseResponse()

close_session async

close_session(request: CloseSessionRequest) -> CloseSessionResponse

Closes a shell session.

Source code in swerex/runtime/local.py

async def close_session(self, request: CloseSessionRequest) -> CloseSessionResponse:
    """Closes a shell session."""
    if request.session not in self.sessions:
        msg = f"session {request.session!r} does not exist"
        raise SessionDoesNotExistError(msg)
    out = await self.sessions[request.session].close()
    del self.sessions[request.session]
    return out

create_session async

create_session(request: CreateSessionRequest) -> CreateSessionResponse

Creates a new session.

Source code in swerex/runtime/local.py

async def create_session(self, request: CreateSessionRequest) -> CreateSessionResponse:
    """Creates a new session."""
    if request.session in self.sessions:
        msg = f"session {request.session} already exists"
        raise SessionExistsError(msg)
    if isinstance(request, CreateBashSessionRequest):
        session = BashSession(request)
    else:
        msg = f"unknown session type: {request!r}"
        raise ValueError(msg)
    self.sessions[request.session] = session
    return await session.start()

execute async

execute(command: Command) -> CommandResponse

Executes a command (independent of any shell session).

Raises:

Type	Description
CommandTimeoutError	If the command times out.
NonZeroExitCodeError	If the command has a non-zero exit code and check is True.

Source code in swerex/runtime/local.py

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

    Raises:
        CommandTimeoutError: If the command times out.
        NonZeroExitCodeError: If the command has a non-zero exit code and `check` is True.
    """
    try:
        result = subprocess.run(
            command.command,
            shell=command.shell,
            timeout=command.timeout,
            env=command.env,
            capture_output=True,
            cwd=command.cwd,
        )
        r = CommandResponse(
            stdout=result.stdout.decode(errors="backslashreplace"),
            stderr=result.stderr.decode(errors="backslashreplace"),
            exit_code=result.returncode,
        )
    except subprocess.TimeoutExpired as e:
        msg = f"Timeout ({command.timeout}s) exceeded while running command"
        raise CommandTimeoutError(msg) from e
    if command.check and result.returncode != 0:
        msg = (
            f"Command {command.command!r} failed with exit code {result.returncode}. "
            f"Stdout:\n{r.stdout!r}\nStderr:\n{r.stderr!r}"
        )
        if command.error_msg:
            msg = f"{command.error_msg}: {msg}"
        raise NonZeroExitCodeError(msg)
    return r

from_config classmethod

from_config(config: LocalRuntimeConfig) -> Self

Source code in swerex/runtime/local.py

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

is_alive async

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

Checks if the runtime is alive.

Source code in swerex/runtime/local.py

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

read_file async

read_file(request: ReadFileRequest) -> ReadFileResponse

Reads a file

Source code in swerex/runtime/local.py

async def read_file(self, request: ReadFileRequest) -> ReadFileResponse:
    """Reads a file"""
    content = Path(request.path).read_text(encoding=request.encoding, errors=request.errors)
    return ReadFileResponse(content=content)

run_in_session async

run_in_session(action: Action) -> Observation

Runs a command in a session.

Source code in swerex/runtime/local.py

async def run_in_session(self, action: Action) -> Observation:
    """Runs a command in a session."""
    if action.session not in self.sessions:
        msg = f"session {action.session!r} does not exist"
        raise SessionDoesNotExistError(msg)
    return await self.sessions[action.session].run(action)

upload async

upload(request: UploadRequest) -> UploadResponse

Uploads a file

Source code in swerex/runtime/local.py

async def upload(self, request: UploadRequest) -> UploadResponse:
    """Uploads a file"""
    if Path(request.source_path).is_dir():
        shutil.copytree(request.source_path, request.target_path)
    else:
        shutil.copy(request.source_path, request.target_path)
    return UploadResponse()

write_file async

write_file(request: WriteFileRequest) -> WriteFileResponse

Writes a file

Source code in swerex/runtime/local.py

async def write_file(self, request: WriteFileRequest) -> WriteFileResponse:
    """Writes a file"""
    Path(request.path).parent.mkdir(parents=True, exist_ok=True)
    Path(request.path).write_text(request.content)
    return WriteFileResponse()

SWE-agent SWE-smith SWE-bench sb-cli