agi-env API

agi_env provides the shared environment and path-resolution layer used by both AGILab and the core runtime packages.

Usage Example

Instanciation

import asyncio
from pathlib import Path

from agi_cluster.agi_distributor import AGI
from agi_env import AgiEnv


AGILAB_PATH = Path((Path.home() / ".local/share/agilab/.agilab-path").read_text().strip())
APPS_PATH = AGILAB_PATH / "apps"
APP = "mycode_project"


async def main():
    app_env = AgiEnv(apps_path=APPS_PATH, app=APP, verbose=1)
    res = await AGI.get_distrib(app_env)
    print(res)
    return res


if __name__ == "__main__":
    asyncio.get_event_loop().run_until_complete(main())

Note

AgiEnv behaves as a singleton. Repeated instantiation updates the same environment instance. Call AgiEnv.reset() before configuring a new environment, or AgiEnv.current() to retrieve the active one.

Share directory resolution

AgiEnv exposes the resolved data root through AgiEnv.agi_share_path_abs. The path is derived from environment settings using the following precedence:

1. AGI_SHARE_DIR from the current process environment, then .env. This is the user-facing override.

2. AGI_CLUSTER_SHARE from the current process environment (e.g. ssh session, service unit), then .env.

3. AGI_LOCAL_SHARE (default ~/localshare) when cluster mode is disabled.

This ordering lets operators set one shared-data knob in the UI/installer while still allowing remote workers to honour host-specific cluster-share settings. The behaviour differs depending on whether cluster mode is enabled:

• Cluster mode enabled – AgiEnv uses AGI_CLUSTER_SHARE. Relative inputs are expanded against AgiEnv.home_abs on manager/developer shells. The configured share must be mounted and writable, and it must be distinct from AGI_LOCAL_SHARE. Missing or read-only shares now raise immediately instead of silently degrading to a local path.

• Cluster mode disabled – AgiEnv uses AGI_LOCAL_SHARE for local datasets and outputs.

• Remote workers – the configured cluster-share value remains relative (for example, clustershare/<app>) and is not created automatically. Workers never fall back to per-user local paths; the configured share path must already be mounted and writable on the remote host. When an absolute path under /Users/<user> or /home/<user> is provided, the leading segments are stripped so the worker can re-root the remainder under its own home directory or mount point.

Because the worker value stays relative, it will fail fast if agi_share_path is not mounted. This makes data provenance explicit and avoids hidden copies of datasets on remote machines.

Reference

AGILab environment bootstrapper and utility helpers.

The module exposes the AgiEnv class which orchestrates project discovery, virtual-environment management, packaging helpers, and convenience utilities used by installers as well as runtime workers. Supporting free functions provide small parsing and path utilities leveraged during setup.

Notes on singleton and pre‑init behavior

• AgiEnv behaves as a true singleton. Instance attributes are the source of truth; class attribute reads proxy to the singleton instance when initialised. Methods and descriptors are never shadowed by the delegation.

• A small subset of helpers is pre‑init safe and can be used before constructing an instance: AgiEnv.set_env_var(), AgiEnv.read_agilab_path(), AgiEnv._build_env, and AgiEnv.log_info(). These functions avoid hard failures when the shared logger/environment has not been configured yet. Logging in that mode is best‑effort and may fall back to print.

class agi_env.agi_env.AgiEnv(apps_path=None, app=None, verbose=None, debug=False, python_variante='', **kwargs)

Bases: object

Encapsulates filesystem and configuration state for AGILab deployments.

Singleton access

• Repeated instantiation reuses the same instance. Use AgiEnv.reset() to drop it, or AgiEnv.current() to retrieve it.

• Reading AgiEnv.attr proxies to the singleton’s attribute when the instance exists; callables/properties are always returned from the class.

EXTRA_INDEX_URL = 'https://pypi.org/simple'

GUI_SAMPLING = None

INDEX_URL = 'https://test.pypi.org/simple'

TABLE_MAX_ROWS = None

__init__(apps_path=None, app=None, verbose=None, debug=False, python_variante='', **kwargs)

active(target)

Switch app to target if it differs from the current one.

app = None

apps_path = None

benchmark = None

change_app(app)

static check_internet()

clone_directory(source_dir, dest_dir, rename_map, spec, source_root)

Recursively copy + rename directories, files, and contents.

clone_project(target_project, dest_project)

Clone a project by copying files, applying renames, and final cleanup.

copy_existing_projects(src_apps, dst_apps)

Copy *_project trees from src_apps into dst_apps if missing.

static create_junction_windows(source, dest)

Create a directory junction on Windows.

Parameters:

• source (Path) – The target directory path.

• dest (Path) – The destination junction path.

create_rename_map(target_project, dest_project)

Create a mapping of old → new names for cloning.

Return type:

dict

static create_symlink(src, dest)

static create_symlink_windows(source, dest)

Create a symbolic link on Windows, handling permissions and types.

Parameters:

• source (Path) – Source directory path.

• dest (Path) – Destination symlink path.

classmethod current()

Return the currently initialised environment instance.

Return type:

AgiEnv

debug = False

envars = {}

err_log = None

extract_base_info(base, import_mapping)

Return the base-class name and originating module for base nodes.

find_source_app_settings_file(app_name=None)

Return the versioned/source app_settings.toml for an app when available.

Return type:

Path | None

get_base_classes(module_path, class_name)

Inspect module_path AST to retrieve base classes of class_name.

get_base_worker_cls(module_path, class_name)

Return the base worker class name and module for class_name.

get_full_attribute_name(node)

Reconstruct the dotted attribute path represented by node.

get_import_mapping(source)

Build a mapping of names to modules from import statements in source.

get_projects(*paths)

Return the names of *_project directories beneath the provided paths.

static has_admin_rights()

Check if the current process has administrative rights on Windows.

Returns:

True if admin, False otherwise.

Return type:

bool

has_agilab_anywhere_under_home(path)

Return True when path sits under the user’s home agilab tree.

Return type:

bool

humanize_validation_errors(error)

Format pydantic-style validation error messages for human consumption.

hw_rapids_capable = None

init_done = False

init_envars_app(envars)

Cache frequently used environment variables and ensure directories exist.

install_type = None

static is_local(ip)

Parameters:

ip

Returns:

is_local_worker = False

is_source_env = False

is_valid_ip(ip)

Return True when ip is a syntactically valid IPv4 address.

Return type:

bool

is_worker_env = False

static locate_agi_installation(verbose=False)

Deprecated alias for locate_agilab_installation().

static locate_agilab_installation(verbose=False)

Attempt to locate the installed AGILab package path on disk.

static log_info(line)

Lightweight info logger retained for legacy hooks (e.g. pre_install scripts).

Return type:

None

logger = None

static mode2int(mode)

Convert an iterable of mode flags (p, c, d) to the bitmask int.

mode2str(mode)

Encode a bitmask mode into readable pcdr flag form.

out_log = None

pyvers_worker = None

static read_agilab_path(verbose=False)

Return the persisted AGILab installation path if previously recorded.

read_gitignore(gitignore_path)

Return type:

PathSpec

replace_content(txt, rename_map)

Return type:

str

classmethod reset()

Drop the cached singleton so a fresh environment can be bootstrapped.

Return type:

None

resolve_share_path(path)

Resolve path relative to the shared storage root.

None or "." returns the root itself; absolute inputs pass through unchanged.

Return type:

Path

resolve_user_app_settings_file(app_name=None, *, ensure_exists=True)

Return the per-user mutable app_settings.toml path for an app.

The workspace copy lives under ~/.agilab/apps/<app>/app_settings.toml and is seeded from the versioned source file on first use.

Return type:

Path

resources_path = PosixPath('/home/runner/.agilab')

async static run(cmd, venv, cwd=None, timeout=None, wait=True, log_callback=None)

Run a shell command inside a virtual environment.

async run_agi(code, log_callback=None, venv=None, type=None)

Asynchronous version of run_agi for use within an async context.

async static run_async(cmd, venv=None, cwd=None, timeout=None, log_callback=None)

Run a shell command asynchronously and return the last non-empty line.

static set_env_var(key, value)

Persist key/value in envars, os.environ and the .env file.

share_root_path()

Return the absolute path corresponding to agi_share_path.

Return type:

Path

skip_repo_links = False

snippet_tail = 'asyncio.get_event_loop().run_until_complete(main())'

target = None

unzip_data(archive_path, extract_to=None, *, force_extract=False)

uv = None

verbose = None

class agi_env.agi_env.ContentRenamer(rename_map)

Bases: ContentRenamer

Compatibility wrapper that binds the pure renamer to AgiEnv.logger.

__init__(rename_map)