BaseEnvironment

Abstract base class for containerized execution environments

Overview

The BaseEnvironment abstract class defines the interface for containerized environments where agents execute tasks. Harbor supports multiple environment backends including Docker, Daytona, E2B, Modal, and GKE.

Import: from harbor.environments.base import BaseEnvironment

Class Attributes

environment_dir · Path · required

Path to the environment directory containing definition files (e.g., docker-compose.yaml).

environment_name · str · required

The name of the environment, typically the task name.

session_id · str · required

Unique session identifier for this environment instance, typically the trial name.

trial_paths · TrialPaths · required

Path configuration for the trial.

task_env_config · EnvironmentConfig · required

Environment configuration from the task definition.

logger · logging.Logger · required

Logger instance for the environment.

Constructor

def __init__(
    self,
    environment_dir: Path,
    environment_name: str,
    session_id: str,
    trial_paths: TrialPaths,
    task_env_config: EnvironmentConfig,
    logger: logging.Logger | None = None,
    override_cpus: int | None = None,
    override_memory_mb: int | None = None,
    override_storage_mb: int | None = None,
    override_gpus: int | None = None,
    suppress_override_warnings: bool = False,
    *args,
    **kwargs,
)

environment_dir · Path · required

Path to the environment directory.

environment_name · str · required

Name of the environment.

session_id · str · required

Session ID for this instance.

trial_paths · TrialPaths · required

Trial paths configuration.

task_env_config · EnvironmentConfig · required

Environment configuration from task.

logger · logging.Logger | None

Optional logger instance.

override_cpus · int | None

Override CPU allocation. Warning: May disqualify from leaderboards.

override_memory_mb · int | None

Override memory allocation in MB. Warning: May disqualify from leaderboards.

override_storage_mb · int | None

Override storage allocation in MB. Warning: May disqualify from leaderboards.

override_gpus · int | None

Override GPU allocation. Warning: May disqualify from leaderboards.

suppress_override_warnings · bool · default: "False"

Suppress warnings about resource overrides.

Abstract Methods

type

@staticmethod
@abstractmethod
def type() -> EnvironmentType

Returns the environment type (e.g., DOCKER, DAYTONA, MODAL).

type · EnvironmentType · required

The environment type identifier.

Abstract Properties

@property
@abstractmethod
def is_mounted(self) -> bool

is_mounted · bool · required

Whether the environment mounts the logging directories.

@property
@abstractmethod
def supports_gpus(self) -> bool

supports_gpus · bool · required

Whether this environment type supports GPU allocation.

@property
@abstractmethod
def can_disable_internet(self) -> bool

can_disable_internet · bool · required

Whether this environment type supports disabling internet access.

start

@abstractmethod
async def start(self, force_build: bool) -> None

Starts the environment and optionally forces a rebuild.

force_build · bool · required

Whether to force a rebuild of the environment.

stop

@abstractmethod
async def stop(self, delete: bool) -> None

Stops the environment and optionally deletes it.

delete · bool · required

Whether to delete the environment after stopping.

File Operations

upload_file

@abstractmethod
async def upload_file(self, source_path: Path | str, target_path: str) -> None

source_path · Path | str · required

Local file path to upload.

target_path · str · required

Target path in the environment.

upload_dir

@abstractmethod
async def upload_dir(self, source_dir: Path | str, target_dir: str) -> None

source_dir · Path | str · required

Local directory to upload.

target_dir · str · required

Target directory in the environment.

download_file

@abstractmethod
async def download_file(self, source_path: str, target_path: Path | str) -> None

source_path · str · required

File path in the environment.

target_path · Path | str · required

Local target path.

download_dir

@abstractmethod
async def download_dir(self, source_dir: str, target_dir: Path | str) -> None

Downloads a directory from the environment. Overwrites existing files.

source_dir · str · required

Directory path in the environment.

target_dir · Path | str · required

Local target directory.

exec

@abstractmethod
async def exec(
    self,
    command: str,
    cwd: str | None = None,
    env: dict[str, str] | None = None,
    timeout_sec: int | None = None,
) -> ExecResult

Executes a command in the environment.

command · str · required

The command to execute.

cwd · str | None

Working directory for command execution.

env · dict[str, str] | None

Environment variables to set.

timeout_sec · int | None

Command timeout in seconds.

ExecResult · ExecResult

Result containing stdout, stderr, and return code.

Concrete Methods

is_dir

async def is_dir(self, path: str) -> bool

Checks if a remote path is a directory.

path · str · required

Path to check in the environment.

is_directory · bool

True if path exists and is a directory.

is_file

async def is_file(self, path: str) -> bool

Checks if a remote path is a regular file.

path · str · required

Path to check in the environment.

is_file · bool

True if path exists and is a regular file.

attach

async def attach(self) -> None

Attaches to the environment using os.execvp. Not supported by all environment types.

Raises: NotImplementedError if the environment doesn’t support attaching.

ExecResult Model

class ExecResult(BaseModel):
    stdout: str | None = None
    stderr: str | None = None
    return_code: int

stdout · str | None

Standard output from the command.

stderr · str | None

Standard error from the command.

return_code · int · required

Exit code from the command.

Example Implementation

from pathlib import Path
from harbor.environments.base import BaseEnvironment, ExecResult
from harbor.models.environment_type import EnvironmentType
from harbor.models.task.config import EnvironmentConfig
from harbor.models.trial.paths import TrialPaths

class MyCustomEnvironment(BaseEnvironment):
    """Custom environment implementation."""
    
    @staticmethod
    def type() -> EnvironmentType:
        return EnvironmentType.CUSTOM
    
    @property
    def is_mounted(self) -> bool:
        return True
    
    @property
    def supports_gpus(self) -> bool:
        return False
    
    @property
    def can_disable_internet(self) -> bool:
        return True
    
    def _validate_definition(self):
        # Check that required environment files exist
        if not (self.environment_dir / "Dockerfile").exists():
            raise FileNotFoundError("Dockerfile not found")
    
    async def start(self, force_build: bool) -> None:
        # Start the environment
        self.logger.info(f"Starting environment {self.session_id}")
        # Implementation details...
    
    async def stop(self, delete: bool) -> None:
        # Stop and optionally delete the environment
        self.logger.info(f"Stopping environment {self.session_id}")
        # Implementation details...
    
    async def exec(
        self,
        command: str,
        cwd: str | None = None,
        env: dict[str, str] | None = None,
        timeout_sec: int | None = None,
    ) -> ExecResult:
        # Execute command in environment
        # Implementation details...
        return ExecResult(
            stdout="command output",
            stderr=None,
            return_code=0
        )
    
    async def upload_file(self, source_path: Path | str, target_path: str):
        # Upload file to environment
        pass
    
    async def upload_dir(self, source_dir: Path | str, target_dir: str):
        # Upload directory to environment
        pass
    
    async def download_file(self, source_path: str, target_path: Path | str):
        # Download file from environment
        pass
    
    async def download_dir(self, source_dir: str, target_dir: Path | str):
        # Download directory from environment
        pass

Related Types

• EnvironmentConfig - Environment configuration
• BaseAgent - Agent interface