The environment class

sweagent.environment.swe_env.SWEEnv

SWEEnv(*, deployment: AbstractDeployment, repo: Repo | RepoConfig | None, post_startup_commands: list[str], post_startup_command_timeout: int = 500, hooks: list[EnvHook] | None = None, name: str = 'main')

This class represents the environment in which we solve the tasks.

Parameters:

Name	Type	Description	Default
deployment	AbstractDeployment	SWE-ReX deployment instance	required
repo	Repo | RepoConfig | None	Repository configuration object, or anything following the Repo protocol	required
post_startup_commands	list[str]	Commands to execute before starting the agent	required
hooks	list[EnvHook] | None	Environment hooks (used to inject custom functionality) Equivalent to calling add_hook for each hook after initialization.	None
name	str	Name of the environment	'main'

Source code in sweagent/environment/swe_env.py

def __init__(
    self,
    *,
    deployment: AbstractDeployment,
    repo: Repo | RepoConfig | None,
    post_startup_commands: list[str],
    post_startup_command_timeout: int = 500,
    hooks: list[EnvHook] | None = None,
    name: str = "main",
):
    """This class represents the environment in which we solve the tasks.

    Args:
        deployment: SWE-ReX deployment instance
        repo: Repository configuration object, or anything following the `Repo` protocol
        post_startup_commands: Commands to execute before starting the agent
        hooks: Environment hooks (used to inject custom functionality)
            Equivalent to calling `add_hook` for each hook after initialization.
        name: Name of the environment
    """
    super().__init__()
    self.deployment = deployment
    self.repo = repo
    self._post_startup_commands = post_startup_commands
    self.post_startup_command_timeout = post_startup_command_timeout
    self.logger = get_logger("swea-env", emoji="���")
    self.name = name
    self.clean_multi_line_functions = lambda x: x
    self._chook = CombinedEnvHooks()
    for hook in hooks or []:
        self.add_hook(hook)

clean_multi_line_functions instance-attribute

clean_multi_line_functions = lambda x: x

deployment instance-attribute

deployment = deployment

logger instance-attribute

logger = get_logger('swea-env', emoji='���')

name instance-attribute

name = name

post_startup_command_timeout instance-attribute

post_startup_command_timeout = post_startup_command_timeout

repo instance-attribute

repo = repo

add_hook

add_hook(hook: EnvHook) -> None

Add EnvHook to the environment.

This allows to inject custom functionality at different stages of the environment lifecycle, in particular to connect SWE-agent to a new interface (like a GUI).

Source code in sweagent/environment/swe_env.py

def add_hook(self, hook: EnvHook) -> None:
    """Add `EnvHook` to the environment.

    This allows to inject custom functionality at different stages of the environment
    lifecycle, in particular to connect SWE-agent to a new interface (like a GUI).
    """
    hook.on_init(env=self)
    self._chook.add_hook(hook)

close

close() -> None

Shutdown SWE-ReX deployment etc.

Source code in sweagent/environment/swe_env.py

def close(self) -> None:
    """Shutdown SWE-ReX deployment etc."""
    self.logger.info("Beginning environment shutdown...")
    asyncio.run(self.deployment.stop())
    self._chook.on_close()

communicate

communicate(input: str, timeout: int | float = 25, *, check: Literal['warn', 'ignore', 'raise'] = 'ignore', error_msg: str = 'Command failed') -> str

Executes a command in the running shell. The details of this are handled by the SWE-ReX deployment/runtime.

Parameters:

Name	Type	Description	Default
input	str	input to send to container	required
timeout_duration		duration to wait for output	required
check	Literal['warn', 'ignore', 'raise']	ignore: do not extract exit code (more stable), warn: extract exit code and log error if exit code is non-zero, raise: raise error if exit code is non-zero	'ignore'
error_msg	str	error message to raise if the command fails	'Command failed'

Returns:

Name	Type	Description
output	str	output from container

Source code in sweagent/environment/swe_env.py

def communicate(
    self,
    input: str,
    timeout: int | float = 25,
    *,
    check: Literal["warn", "ignore", "raise"] = "ignore",
    error_msg: str = "Command failed",
) -> str:
    """Executes a command in the running shell. The details of this are handled by
    the SWE-ReX deployment/runtime.

    Args:
        input: input to send to container
        timeout_duration: duration to wait for output
        check: `ignore`: do not extract exit code (more stable), `warn`: extract exit code and log error if
            exit code is non-zero, `raise`: raise error if exit code is non-zero
        error_msg: error message to raise if the command fails

    Returns:
        output: output from container
    """
    self.logger.log(logging.TRACE, "Input:\n%s", input)  # type: ignore
    rex_check = "silent" if check else "ignore"
    r = asyncio.run(
        self.deployment.runtime.run_in_session(BashAction(command=input, timeout=timeout, check=rex_check))
    )
    output = r.output
    self.logger.log(logging.TRACE, "Output:\n%s", output)  # type: ignore
    if check != "ignore" and r.exit_code != 0:
        self.logger.error(f"{error_msg}:\n{output}")
        msg = f"Command {input!r} failed ({r.exit_code=}): {error_msg}"
        self.logger.error(msg)
        if check == "raise":
            self.close()
            raise RuntimeError(msg)
    return output

execute_command

execute_command(command: str, shell: bool = True, check: bool = False, env: dict[str, str] | None = None, cwd: str | None = None) -> None

Execute a command in the environment independent of the session (i.e., as a subprocess)

Source code in sweagent/environment/swe_env.py

def execute_command(
    self,
    command: str,
    shell: bool = True,
    check: bool = False,
    env: dict[str, str] | None = None,
    cwd: str | None = None,
) -> None:
    """Execute a command in the environment independent of the session (i.e., as a subprocess)"""
    asyncio.run(
        self.deployment.runtime.execute(RexCommand(command=command, shell=shell, check=check, env=env, cwd=cwd))
    )

from_config classmethod

from_config(config: EnvironmentConfig) -> Self

Create an environment instance from a configuration object. This is the recommended way to create an environment instance, unless you need more flexibility.

Source code in sweagent/environment/swe_env.py

@classmethod
def from_config(cls, config: EnvironmentConfig) -> Self:
    """Create an environment instance from a configuration object.
    This is the recommended way to create an environment instance, unless you need
    more flexibility.
    """
    # Always copy config to avoid shared state between different instances
    config = config.model_copy(deep=True)
    return cls(
        deployment=get_deployment(config.deployment),
        repo=config.repo,
        post_startup_commands=config.post_startup_commands,
        post_startup_command_timeout=config.post_startup_command_timeout,
        name=config.name,
    )

hard_reset

hard_reset()

Resets the environment and deployment, i.e., completely restarts the deployment.

Source code in sweagent/environment/swe_env.py

def hard_reset(self):
    """Resets the environment and deployment, i.e., completely restarts the
    deployment.
    """
    self.close()
    self.start()

interrupt_session

interrupt_session()

Source code in sweagent/environment/swe_env.py

def interrupt_session(self):
    self.logger.info("Interrupting session")
    asyncio.run(self.deployment.runtime.run_in_session(BashInterruptAction()))

read_file

read_file(path: str | PurePath, encoding: str | None = None, errors: str | None = None) -> str

Read file contents from container

Parameters:

Name	Type	Description	Default
path	str | PurePath	Absolute path to file	required
encoding	str | None	Encoding to use when reading the file. None means default encoding. This is the same as the encoding argument of Path.read_text()	None
errors	str | None	Error handling to use when reading the file. None means default error handling. This is the same as the errors argument of Path.read_text()	None

Returns:

Name	Type	Description
file_contents	str	Contents of file as string

Source code in sweagent/environment/swe_env.py

def read_file(self, path: str | PurePath, encoding: str | None = None, errors: str | None = None) -> str:
    """Read file contents from container

    Args:
        path: Absolute path to file
        encoding: Encoding to use when reading the file. None means default encoding.
            This is the same as the `encoding` argument of `Path.read_text()`
        errors: Error handling to use when reading the file. None means default error handling.
            This is the same as the `errors` argument of `Path.read_text()`

    Returns:
        file_contents: Contents of file as string
    """
    r = asyncio.run(
        self.deployment.runtime.read_file(ReadFileRequest(path=str(path), encoding=encoding, errors=errors))
    )
    return r.content

reset

reset()

Reset the environment to a clean state. Gets called by start, but can also be called independently to reset the environment to a clean state before a new attempt.

Returns:

Name	Type	Description
observation		output from container
info		additional information (e.g. debugging information)

Source code in sweagent/environment/swe_env.py

def reset(self):
    """Reset the environment to a clean state.
    Gets called by `start`, but can also be called independently to reset the
    environment to a clean state before a new attempt.

    Returns:
        observation: output from container
        info: additional information (e.g. debugging information)
    """
    self.communicate(input="cd /", check="raise")
    self._copy_repo()
    self._reset_repository()
    self._chook.on_environment_startup()

set_env_variables

set_env_variables(env_variables: dict[str, str]) -> None

Set environment variables in the environment.

Source code in sweagent/environment/swe_env.py

def set_env_variables(self, env_variables: dict[str, str]) -> None:
    """Set environment variables in the environment."""
    if not env_variables:
        self.logger.debug("No environment variables to set")
        return
    _env_setters = [f"export {k}={shlex.quote(str(v))}" for k, v in env_variables.items()]
    command = " && ".join(_env_setters)
    self.communicate(command, check="raise")

start

start() -> None

Start the environment and reset it to a clean state.

Source code in sweagent/environment/swe_env.py

def start(self) -> None:
    """Start the environment and reset it to a clean state."""
    self._init_deployment()
    self.reset()
    for command in self._post_startup_commands:
        self.communicate(command, check="raise", timeout=self.post_startup_command_timeout)

write_file

write_file(path: str | PurePath, content: str) -> None

Write content to file in container

Source code in sweagent/environment/swe_env.py

def write_file(self, path: str | PurePath, content: str) -> None:
    """Write content to file in container"""
    asyncio.run(self.deployment.runtime.write_file(WriteFileRequest(path=str(path), content=content)))