"""CLI-Anything catalog, install state, and safe CLI execution.""" from __future__ import annotations import json import os import re import shlex import shutil import subprocess import sys import time from dataclasses import dataclass from importlib import metadata as importlib_metadata from pathlib import Path from typing import Any from urllib.parse import urlparse import httpx from loguru import logger from nanobot.apps.protocol import app_manifest, compact_dict from nanobot.config.paths import get_runtime_subdir from nanobot.security.workspace_policy import is_path_within CLI_ANYTHING_REGISTRY_URL = "https://hkuds.github.io/CLI-Anything/registry.json" CLI_ANYTHING_PUBLIC_REGISTRY_URL = "https://hkuds.github.io/CLI-Anything/public_registry.json" CLI_ANYTHING_RAW_BASE = "https://raw.githubusercontent.com/HKUDS/CLI-Anything/main" NANOBOT_EXTENSION_REGISTRY_URL = "https://raw.githubusercontent.com/Re-bin/nanobot-extension/main/registry.json" NANOBOT_EXTENSION_RAW_BASE = "https://raw.githubusercontent.com/Re-bin/nanobot-extension/main" _CATALOG_SOURCES = ( ("harness", CLI_ANYTHING_REGISTRY_URL, CLI_ANYTHING_RAW_BASE, True), ("public", CLI_ANYTHING_PUBLIC_REGISTRY_URL, CLI_ANYTHING_RAW_BASE, True), ("extensions", NANOBOT_EXTENSION_REGISTRY_URL, NANOBOT_EXTENSION_RAW_BASE, False), ) _MAX_TOOL_OUTPUT_CHARS = 12_000 _MAX_ARTIFACT_SCAN_PATHS = 4_000 _MAX_ARTIFACT_REPORT = 12 _SAFE_NAME_RE = re.compile(r"[^a-z0-9_-]+") _SAFE_NPM_DIR_RE = re.compile(r"^[a-z0-9._-]+$", re.IGNORECASE) _MENTION_RE = re.compile(r"(^|[\s([{])@([a-z0-9_-]+)\b", re.IGNORECASE) _SHELL_META_CHARS = ("|", "&&", "||", ";", "$(", "`", ">", "<") _ENDORSEMENT_WORD_RE = re.compile(r"\bofficial\s+", re.IGNORECASE) _ARTIFACT_EXTENSIONS = frozenset({ ".csv", ".drawio", ".gif", ".html", ".jpeg", ".jpg", ".json", ".md", ".pdf", ".png", ".svg", ".txt", ".vsdx", ".webp", ".xml", }) _INLINE_ARTIFACT_EXTENSIONS = frozenset({".gif", ".jpeg", ".jpg", ".png", ".webp"}) _ARTIFACT_IGNORE_DIRS = frozenset({ ".git", ".hg", ".mypy_cache", ".nanobot", ".pytest_cache", ".ruff_cache", ".venv", "__pycache__", "build", "dist", "node_modules", "venv", }) class CliAppError(ValueError): """User-facing CLI Apps failure.""" def __init__(self, message: str, *, status: int = 400) -> None: super().__init__(message) self.message = message self.status = status @dataclass(slots=True) class CliAppsRuntimeConfig: """Runtime knobs for CLI Apps.""" install_timeout: int = 300 run_timeout: int = 60 catalog_ttl_seconds: int = 3600 _BRANDS: dict[str, tuple[str, str]] = { "1password-cli": ("1password", "#3B66BC"), "arcgis": ("arcgis", "#2C7AC3"), "arcgis-pro": ("arcgis", "#2C7AC3"), "audacity": ("audacity", "#0000CC"), "blender": ("blender", "#E87D0D"), "browser": ("googlechrome", "#4285F4"), "calibre": ("calibre", "#45B29D"), "chromadb": ("chroma", "#FFDE2D"), "comfyui": ("comfyui", "#111827"), "contentful": ("contentful", "#2478CC"), "dify": ("dify", "#155EEF"), "drawio": ("diagramsdotnet", "#F08705"), "elevenlabs": ("elevenlabs", "#000000"), "eth2-quickstart": ("ethereum", "#627EEA"), "firefly-iii": ("fireflyiii", "#CD5029"), "freecad": ("freecad", "#418FDE"), "generate-veo-video": ("googlegemini", "#8E75B2"), "gimp": ("gimp", "#5C5543"), "godot": ("godotengine", "#478CBF"), "hacker-feeds-cli": ("rss", "#FFA500"), "inkscape": ("inkscape", "#000000"), "intelwatch": ("intel", "#0071C5"), "iterm2": ("iterm2", "#000000"), "jimeng": ("bytedance", "#3C8CFF"), "joplin": ("joplin", "#1071D3"), "kdenlive": ("kdenlive", "#527EB2"), "krita": ("krita", "#3BABFF"), "libreoffice": ("libreoffice", "#18A303"), "mailchimp": ("mailchimp", "#FFE01B"), "mermaid": ("mermaid", "#FF3670"), "minimax": ("minimax", "#111827"), "musescore": ("musescore", "#1A70B8"), "n8n": ("n8n", "#EA4B71"), "notebooklm": ("googlenotebooklm", "#4285F4"), "obs-studio": ("obsstudio", "#302E31"), "obsidian": ("obsidian", "#7C3AED"), "ollama": ("ollama", "#000000"), "pm2": ("pm2", "#2B037A"), "qgis": ("qgis", "#589632"), "safari": ("safari", "#006CFF"), "sanity": ("sanity", "#F03E2F"), "sentry": ("sentry", "#362D59"), "sketch": ("sketch", "#F7B500"), "shopify": ("shopify", "#7AB55C"), "nsight-graphics": ("nvidia", "#76B900"), "unrealinsights": ("unrealengine", "#0E1128"), "ueatelier": ("unrealengine", "#0E1128"), "ve-twini": ("x", "#000000"), "wecom": ("wechat", "#07C160"), "suno": ("suno", "#000000"), "lldb": ("llvm", "#262D3A"), "android-cli": ("android", "#3DDC84"), "adguardhome": ("adguard", "#68BC71"), "zotero": ("zotero", "#CC2936"), "zoom": ("zoom", "#0B5CFF"), } _BRAND_DOMAINS: dict[str, tuple[str, str]] = { "3mf": ("3mf.io", "#00A1DE"), "anygen": ("anygen.io", "#111827"), "clibrowser": ("github.com/allthingssecurity/clibrowser", "#24292F"), "cloudanalyzer": ("github.com/rsasaki0109/CloudAnalyzer", "#2563EB"), "cloudcompare": ("cloudcompare.org", "#4D83C3"), "deployhq": ("deployhq.com", "#00A2D9"), "exa": ("exa.ai", "#111827"), "feishu": ("larksuite.com", "#00A5FF"), "inkstitch": ("inkstitch.org", "#222222"), "macrocli": ("github.com/HKUDS/CLI-Anything/tree/main/macrocli", "#24292F"), "mubu": ("mubu.com", "#16A085"), "nslogger": ("github.com/fpillet/NSLogger", "#24292F"), "novita": ("novita.ai", "#7C3AED"), "openscreen": ("openscreen.com", "#2563EB"), "py4csr": ("github.com/yanmingyu92/py4csr", "#24292F"), "quietshrink": ("github.com/achiya-automation/quietshrink", "#111827"), "renderdoc": ("renderdoc.org", "#2C7DB8"), "rms": ("rms.teltonika-networks.com", "#0054A6"), "sbox": ("sbox.game", "#F59E0B"), "seaclip": ("github.com/SeaClip-Lite/SeaClip", "#0284C7"), "shotcut": ("shotcut.org", "#3B82F6"), "slay-the-spire-ii": ("megacrit.com", "#B91C1C"), "stata": ("stata.com", "#1F4E79"), "unimol-tools": ("github.com/deepmodeling/Uni-Mol", "#4F46E5"), "videocaptioner": ("github.com/WEIFENG2333/VideoCaptioner", "#2563EB"), "wiremock": ("wiremock.org", "#FF6A00"), } _BRAND_ALIASES: dict[str, str] = { "1password": "1password-cli", "dify-workflow": "dify", "feishu-lark": "feishu", "lark-cli": "feishu", "minimax-cli": "minimax", "obsidian-cli": "obsidian", "slay-the-spire-2": "slay-the-spire-ii", "slay-the-spire-ii": "slay-the-spire-ii", "unimol-tools": "unimol-tools", "unimol": "unimol-tools", "veo": "generate-veo-video", } _BRAND_TRAILING_WORDS = ("cli", "workflow", "workflows", "app", "apps", "tool", "tools") def _now() -> float: return time.time() def _safe_skill_name(name: str) -> str: clean = _SAFE_NAME_RE.sub("-", name.lower()).strip("-") return f"cli-app-{clean or 'app'}" def _has_shell_meta(command: str) -> bool: return any(char in command for char in _SHELL_META_CHARS) def _command_exists(command: str) -> bool: try: parts = shlex.split(command) except ValueError: return False if not parts: return False return shutil.which(parts[0]) is not None def _is_pip_install_command(command: str) -> bool: try: tokens = shlex.split(command) except ValueError: return False return ( len(tokens) >= 3 and tokens[:2] == ["pip", "install"] ) or ( len(tokens) >= 5 and tokens[1:4] == ["-m", "pip", "install"] and tokens[0] in {"python", "python3", sys.executable} ) def _pip_uninstall_args_from_command(command: str) -> list[str] | None: if not command or _has_shell_meta(command): return None try: tokens = shlex.split(command) except ValueError: return None if tokens[:2] == ["pip", "uninstall"]: args = tokens[2:] elif ( len(tokens) >= 5 and tokens[1:4] == ["-m", "pip", "uninstall"] and tokens[0] in {"python", "python3", sys.executable} ): args = tokens[4:] else: return None packages = [arg for arg in args if arg not in {"-y", "--yes"}] if not packages or any(arg.startswith("-") for arg in packages): return None return packages def _console_script_distribution(entry_point: str) -> str | None: if not entry_point: return None try: distributions = importlib_metadata.distributions() except Exception: return None for distribution in distributions: try: entry_points = distribution.entry_points except Exception: continue for item in entry_points: if item.group != "console_scripts" or item.name != entry_point: continue try: name = distribution.metadata.get("Name") except Exception: name = None return str(name or getattr(distribution, "name", "") or "").strip() or None return None def _brand_key(value: str) -> str: return _SAFE_NAME_RE.sub("-", value.lower()).replace("_", "-").strip("-") def _brand_candidates(app: dict[str, Any]) -> list[str]: values = [ str(app.get("name") or ""), str(app.get("display_name") or ""), str(app.get("entry_point") or "").removeprefix("cli-anything-"), ] seen: set[str] = set() candidates: list[str] = [] for value in values: key = _brand_key(value) while key and key not in seen: seen.add(key) candidates.append(key) parts = key.split("-") if len(parts) <= 1 or parts[-1] not in _BRAND_TRAILING_WORDS: break key = "-".join(parts[:-1]) return candidates def _brand_payload(app: dict[str, Any]) -> tuple[str | None, str | None]: declared_logo = str(app.get("logo_url") or "").strip() if declared_logo.startswith(("https://", "/")): declared_color = str(app.get("brand_color") or "").strip() return declared_logo, declared_color or None brand = None domain_brand = None for candidate in _brand_candidates(app): key = _BRAND_ALIASES.get(candidate, candidate) brand = _BRANDS.get(key) if brand: break domain_brand = _BRAND_DOMAINS.get(key) if domain_brand: break if not brand: if not domain_brand: return None, None domain, color = domain_brand return f"https://www.google.com/s2/favicons?domain={domain}&sz=64", color slug, color = brand return f"https://cdn.simpleicons.org/{slug}/{color.lstrip('#')}", color def _read_json(path: Path) -> dict[str, Any] | None: try: data = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return None return data if isinstance(data, dict) else None def _write_json(path: Path, data: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) payload = json.dumps(data, indent=2, ensure_ascii=False) tmp_path = path.with_name(f".{path.name}.{os.getpid()}.{int(_now() * 1_000_000)}.tmp") try: tmp_path.write_text(payload, encoding="utf-8") tmp_path.replace(path) finally: if tmp_path.exists(): tmp_path.unlink() def _safe_skill_path(value: str) -> str | None: if not value.startswith("skills/"): return None parts = value.split("/") if any(part in {"", ".", ".."} for part in parts): return None return value if parts[-1] == "SKILL.md" else None def _skill_content_url(skill_md: str, *, raw_base: str = CLI_ANYTHING_RAW_BASE) -> str | None: safe_path = _safe_skill_path(skill_md) if safe_path: return f"{raw_base.rstrip('/')}/{safe_path}" parsed = urlparse(skill_md) if parsed.scheme != "https" or parsed.netloc != "raw.githubusercontent.com": return None raw_prefix = raw_base.rstrip("/") + "/" if not skill_md.startswith(raw_prefix): return None suffix = skill_md.removeprefix(raw_prefix) return skill_md if _safe_skill_path(suffix) else None def _truncate(text: str, limit: int = _MAX_TOOL_OUTPUT_CHARS) -> str: if len(text) <= limit: return text omitted = len(text) - limit return text[:limit] + f"\n\n... truncated {omitted} characters ..." def _catalog_description(app: dict[str, Any]) -> str: """Return catalog copy without implying vendor endorsement.""" description = str(app.get("description") or "") return _ENDORSEMENT_WORD_RE.sub("", description).strip() class CliAppManager: """Manage CLI-Anything registry entries and local install state.""" def __init__( self, *, workspace: Path, data_dir: Path | None = None, runtime: CliAppsRuntimeConfig | None = None, ) -> None: self.workspace = Path(workspace).expanduser() self.data_dir = Path(data_dir) if data_dir is not None else get_runtime_subdir("cli-apps") self.runtime = runtime or CliAppsRuntimeConfig() @property def installed_path(self) -> Path: return self.data_dir / "installed.json" def _cache_path(self, source: str) -> Path: return self.data_dir / f"{source}_registry_cache.json" def _cached_registry(self, cache_path: Path) -> tuple[dict[str, Any] | None, float]: cached = _read_json(cache_path) if not cached: return None, 0.0 data = cached.get("data") if not isinstance(data, dict): return None, 0.0 try: cached_at = float(cached.get("_cached_at", 0)) except (TypeError, ValueError): cached_at = 0.0 return data, cached_at def _load_installed(self) -> dict[str, Any]: data = _read_json(self.installed_path) or {} apps = data.get("apps") if isinstance(data.get("apps"), dict) else data return apps if isinstance(apps, dict) else {} def _save_installed(self, installed: dict[str, Any]) -> None: _write_json(self.installed_path, {"schema_version": 1, "apps": installed}) def installed_names(self) -> list[str]: """Return registry names explicitly installed through CLI Apps.""" return sorted(str(name) for name in self._load_installed()) def _fetch_registry( self, url: str, cache_path: Path, *, force_refresh: bool = False, ) -> dict[str, Any]: data, cached_at = self._cached_registry(cache_path) if ( not force_refresh and data is not None and _now() - cached_at < self.runtime.catalog_ttl_seconds ): return data try: response = httpx.get(url, timeout=15.0, follow_redirects=True) response.raise_for_status() fetched = response.json() if not isinstance(fetched, dict): raise ValueError("registry response must be an object") except Exception: if data is not None: return data raise _write_json(cache_path, {"_cached_at": _now(), "data": fetched}) return fetched async def _fetch_registry_async( self, url: str, cache_path: Path, *, force_refresh: bool = False, ) -> dict[str, Any]: data, cached_at = self._cached_registry(cache_path) if ( not force_refresh and data is not None and _now() - cached_at < self.runtime.catalog_ttl_seconds ): return data try: async with httpx.AsyncClient(timeout=15.0, follow_redirects=True) as client: response = await client.get(url) response.raise_for_status() fetched = response.json() if not isinstance(fetched, dict): raise ValueError("registry response must be an object") except Exception: if data is not None: return data raise _write_json(cache_path, {"_cached_at": _now(), "data": fetched}) return fetched async def refresh_catalog_cache(self, *, force_refresh: bool = False) -> None: for source, url, _raw_base, required in _CATALOG_SOURCES: try: await self._fetch_registry_async( url, self._cache_path(source), force_refresh=force_refresh, ) except Exception: if required: raise def catalog( self, *, force_refresh: bool = False, cache_only: bool = False, ) -> tuple[list[dict[str, Any]], str | None]: registries: list[tuple[str, str, dict[str, Any]]] = [] for source, url, raw_base, required in _CATALOG_SOURCES: try: cache_path = self._cache_path(source) if cache_only: registry, _ = self._cached_registry(cache_path) if registry is None: continue else: registry = self._fetch_registry( url, cache_path, force_refresh=force_refresh, ) except Exception: if required: raise continue registries.append((source, raw_base, registry)) apps_by_name: dict[str, dict[str, Any]] = {} updated_values: list[str] = [] for source, raw_base, registry in registries: meta = registry.get("meta") if isinstance(meta, dict) and isinstance(meta.get("updated"), str): updated_values.append(meta["updated"]) for row in registry.get("clis", []): if not isinstance(row, dict) or not row.get("name"): continue entry = dict(row) entry["_source"] = source entry["_raw_base"] = raw_base key = str(entry["name"]).lower() previous = apps_by_name.get(key) if previous: previous_source = str(previous.get("_source") or source) merged_source = ( previous_source if previous_source == source else f"{previous_source}+{source}" ) apps_by_name[key] = {**previous, **entry, "_source": merged_source} else: apps_by_name[key] = entry return list(apps_by_name.values()), max(updated_values) if updated_values else None def catalog_cache_fresh(self, *, include_optional: bool = False) -> bool: for source, _url, _raw_base, required in _CATALOG_SOURCES: if not required and not include_optional: continue data, cached_at = self._cached_registry(self._cache_path(source)) if data is None or _now() - cached_at >= self.runtime.catalog_ttl_seconds: return False return True def _manifest_source(self, app: dict[str, Any]) -> str: source = str(app.get("_source") or "harness") if source == "extensions": return "nanobot-extension" return f"cli-anything:{source}" def _trust_registry(self, app: dict[str, Any]) -> str: return "nanobot-extension" if str(app.get("_source") or "") == "extensions" else "cli-anything" def get_app(self, name: str, *, force_refresh: bool = False) -> dict[str, Any]: wanted = name.lower() for app in self.catalog(force_refresh=force_refresh)[0]: if str(app.get("name", "")).lower() == wanted: return app raise CliAppError(f"CLI app '{name}' not found", status=404) def mentioned_installed_apps(self, text: str) -> list[dict[str, str]]: """Return installed CLI Apps referenced as ``@name`` in user text.""" if "@" not in text: return [] installed = self._load_installed() if not installed: return [] installed_by_name = { str(name).lower(): (str(name), data if isinstance(data, dict) else {}) for name, data in installed.items() } seen: set[str] = set() mentions: list[dict[str, str]] = [] for match in _MENTION_RE.finditer(text): wanted = str(match.group(2)).lower() if wanted in seen or wanted not in installed_by_name: continue installed_name, data = installed_by_name[wanted] seen.add(wanted) entry_point = str(data.get("entry_point") or "") mentions.append( { "name": installed_name, "entry_point": entry_point, "source": str(data.get("source") or ""), "skill": f"skills/{_safe_skill_name(installed_name)}/SKILL.md", "tool": "run_cli_app", } ) return mentions def _strategy(self, app: dict[str, Any]) -> str: package_manager = str(app.get("package_manager") or "").lower() install_strategy = str(app.get("install_strategy") or "").lower() if package_manager == "bundled" or install_strategy == "bundled": return "bundled" if package_manager in {"npm", "brew", "uv", "pip"}: return package_manager if app.get("npm_package"): return "npm" install_cmd = str(app.get("install_cmd") or "") if _is_pip_install_command(install_cmd): return "pip" return "unsupported" def _install_supported(self, app: dict[str, Any]) -> bool: if self._strategy(app) == "unsupported": return False install_cmd = str(app.get("install_cmd") or "") return not _has_shell_meta(install_cmd) def _skill_path(self, name: str) -> Path: return self.workspace / "skills" / _safe_skill_name(name) / "SKILL.md" def _app_payload( self, app: dict[str, Any], installed: dict[str, Any], ) -> dict[str, Any]: name = str(app["name"]) entry_point = str(app.get("entry_point") or "") install_supported = self._install_supported(app) is_installed = name in installed available = bool(entry_point and shutil.which(entry_point)) if is_installed and available: status = "installed" elif is_installed: status = "missing" elif not install_supported: status = "unsupported" elif available: status = "available" else: status = "not_installed" logo_url, brand_color = _brand_payload(app) return { "name": name, "display_name": app.get("display_name") or name, "category": app.get("category") or "uncategorized", "description": _catalog_description(app), "requires": app.get("requires") or "", "source": app.get("_source") or "harness", "entry_point": entry_point, "install_supported": install_supported, "installed": is_installed, "available": available, "status": status, "logo_url": logo_url, "brand_color": brand_color, "skill_installed": self._skill_path(name).is_file(), "manifest": self._manifest_payload(app, logo_url=logo_url, brand_color=brand_color), } def _package_ref(self, app: dict[str, Any]) -> dict[str, Any] | None: strategy = self._strategy(app) name = "" if strategy == "pip": try: uninstall = self._pip_uninstall_argv(app) except CliAppError: uninstall = None name = uninstall[-1] if uninstall else "" elif strategy == "npm": name = str(app.get("npm_package") or "").strip() elif strategy in {"brew", "uv"}: try: uninstall = self._argv_for_action(app, "uninstall") except CliAppError: uninstall = None if uninstall: name = uninstall[-1] if not strategy or strategy in {"unsupported", "bundled"}: return None return compact_dict({"manager": strategy, "name": name}) def _manifest_payload( self, app: dict[str, Any], *, logo_url: str | None, brand_color: str | None, ) -> dict[str, Any]: name = str(app["name"]) entry_point = str(app.get("entry_point") or "") strategy = self._strategy(app) skill_path = f"skills/{_safe_skill_name(name)}/SKILL.md" capabilities = [ compact_dict({ "type": "cli", "entry_point": entry_point, "package": self._package_ref(app), }), {"type": "skill", "path": skill_path}, ] install_supported = self._install_supported(app) install = compact_dict({ "supported": install_supported, "strategy": strategy, "managed_paths": [skill_path], "verification": ["entry_point_available"] if entry_point else [], }) remove = compact_dict({ "supported": strategy != "unsupported", "strategy": strategy, "managed_paths": [skill_path], "verification": ( ["package_manager_ok", "entry_point_absent", "managed_paths_absent"] if strategy not in {"bundled", "unsupported"} else ["nanobot_state_absent", "managed_paths_absent"] ), }) return app_manifest( app_id=name, display_name=str(app.get("display_name") or name), version=str(app.get("version") or ""), description=_catalog_description(app), category=str(app.get("category") or "uncategorized"), source=self._manifest_source(app), logo_url=logo_url, brand_color=brand_color, capabilities=capabilities, install=install, remove=remove, trust={ "registry": self._trust_registry(app), "level": "catalog", "review_status": "catalog_entry", }, ) def payload(self, *, force_refresh: bool = False, cache_only: bool = False) -> dict[str, Any]: apps, updated = self.catalog(force_refresh=force_refresh, cache_only=cache_only) installed = self._load_installed() rows = [self._app_payload(app, installed) for app in apps] rows.sort(key=lambda item: (str(item["category"]), str(item["display_name"]).lower())) return { "apps": rows, "installed_count": sum(1 for item in rows if item["installed"]), "catalog_updated_at": updated, } def installed_payload(self) -> dict[str, Any]: installed = self._load_installed() rows = [] for name, raw_entry in sorted(installed.items()): entry = raw_entry if isinstance(raw_entry, dict) else {} strategy = str(entry.get("strategy") or "bundled") app = { "name": str(name), "display_name": str(entry.get("display_name") or name), "category": str(entry.get("category") or "installed"), "description": str(entry.get("description") or ""), "requires": str(entry.get("requires") or ""), "_source": str(entry.get("source") or "local"), "entry_point": str(entry.get("entry_point") or ""), "package_manager": strategy, } rows.append(self._app_payload(app, installed)) return { "apps": rows, "installed_count": len(rows), "catalog_updated_at": None, } def _pip_package_from_install(self, app: dict[str, Any]) -> str | None: install_cmd = str(app.get("install_cmd") or "") try: tokens = shlex.split(install_cmd) except ValueError: return None if tokens[:2] == ["pip", "install"]: args = tokens[2:] elif len(tokens) >= 5 and tokens[1:4] == ["-m", "pip", "install"]: args = tokens[4:] else: return None args = [arg for arg in args if not arg.startswith("-")] if len(args) != 1 or args[0].startswith("git+"): return None return args[0] @staticmethod def _pip_available() -> bool: """Return True if pip is importable for the current interpreter.""" from importlib.util import find_spec return find_spec("pip") is not None def _pip_install_argv(self, app: dict[str, Any], *, update: bool = False) -> list[str]: install_cmd = str(app.get("install_cmd") or "") if not _is_pip_install_command(install_cmd) or _has_shell_meta(install_cmd): raise CliAppError("unsupported pip install command") tokens = shlex.split(install_cmd) args = tokens[2:] if tokens[:2] == ["pip", "install"] else tokens[4:] pip_available = self._pip_available() if pip_available: prefix = [sys.executable, "-m", "pip", "install"] elif shutil.which("uv"): prefix = ["uv", "pip", "install", "--python", sys.executable] else: raise CliAppError("pip is not available and uv is not installed") if update: if pip_available: prefix.extend(["--upgrade", "--force-reinstall"]) else: prefix.extend(["--upgrade", "--reinstall"]) return prefix + args def _pip_uninstall_argv( self, app: dict[str, Any], installed_entry: dict[str, Any] | None = None, ) -> list[str]: if self._pip_available(): prefix = [sys.executable, "-m", "pip", "uninstall", "-y"] elif shutil.which("uv"): prefix = ["uv", "pip", "uninstall", "--python", sys.executable] else: raise CliAppError("pip is not available and uv is not installed") distribution = str((installed_entry or {}).get("pip_distribution") or "").strip() if distribution: return [*prefix, distribution] uninstall_cmd = str(app.get("uninstall_cmd") or "") packages = _pip_uninstall_args_from_command(uninstall_cmd) if packages: return [*prefix, *packages] package = str(app.get("pip_package") or "").strip() or self._pip_package_from_install(app) if not package: entry_point = str(app.get("entry_point") or "").strip() package = entry_point if entry_point.startswith("cli-anything-") else f"cli-anything-{_brand_key(str(app['name']))}" return [*prefix, package] def _npm_argv(self, app: dict[str, Any], action: str) -> list[str]: npm = shutil.which("npm") if not npm: raise CliAppError("npm is not installed") package = str(app.get("npm_package") or "") if not package: raise CliAppError("registry entry has no npm_package") if action == "install": return [npm, "install", "-g", package] if action == "update": return [npm, "install", "-g", package + "@latest"] return [npm, "uninstall", "-g", package] def _cleanup_stale_npm_install(self, app: dict[str, Any]) -> bool: npm = shutil.which("npm") package = str(app.get("npm_package") or "").strip() if not npm or not package or "/" in package or _SAFE_NPM_DIR_RE.match(package) is None: return False result = self._run_argv([npm, "root", "-g"], timeout=min(self.runtime.install_timeout, 30)) if result.returncode != 0: return False root = Path(result.stdout.strip()).expanduser() try: root = root.resolve(strict=True) except OSError: return False targets = [root / package, *root.glob(f".{package}-*")] removed = False for target in targets: try: resolved = target.resolve(strict=False) if not is_path_within(resolved, root) or not target.is_dir(): continue shutil.rmtree(target) removed = True except OSError: continue return removed def _retry_stale_npm_install( self, app: dict[str, Any], argv: list[str], result: subprocess.CompletedProcess[str], ) -> subprocess.CompletedProcess[str]: output = f"{result.stderr}\n{result.stdout}" if "ENOTEMPTY" not in output or "rename" not in output: return result if not self._cleanup_stale_npm_install(app): return result return self._run_argv(argv, timeout=self.runtime.install_timeout) def _split_safe_command(self, app: dict[str, Any], key: str, expected: str) -> list[str]: command = str(app.get(key) or "") if not command: raise CliAppError(f"no {key} is defined for {app['name']}") if _has_shell_meta(command): raise CliAppError("script-style install commands are disabled in this MVP") try: argv = shlex.split(command) except ValueError as exc: raise CliAppError(f"invalid command: {exc}") from exc if not argv or argv[0] != expected: raise CliAppError(f"unsupported {expected} command") return argv def _argv_for_action( self, app: dict[str, Any], action: str, installed_entry: dict[str, Any] | None = None, ) -> list[str] | None: strategy = self._strategy(app) if strategy == "pip": if action == "install": return self._pip_install_argv(app) if action == "update": return self._pip_install_argv(app, update=True) return self._pip_uninstall_argv(app, installed_entry=installed_entry) if strategy == "npm": return self._npm_argv(app, action) if strategy == "brew": key = {"install": "install_cmd", "update": "update_cmd", "uninstall": "uninstall_cmd"}[action] return self._split_safe_command(app, key, "brew") if strategy == "uv": key = {"install": "install_cmd", "update": "update_cmd", "uninstall": "uninstall_cmd"}[action] return self._split_safe_command(app, key, "uv") if strategy == "bundled": return None raise CliAppError("this CLI app uses an unsupported install strategy") def _run_argv(self, argv: list[str], *, timeout: int) -> subprocess.CompletedProcess[str]: command = subprocess.list2cmdline(argv) logger.info("CLI Apps: running {}", command) result = subprocess.run( argv, capture_output=True, text=True, timeout=timeout, ) logger.info("CLI Apps: command exited with code {}: {}", result.returncode, command) output = (result.stderr or result.stdout or "").strip() if output: logger.info("CLI Apps command output:\n{}", _truncate(output, 4000)) return result def _installed_entry(self, app: dict[str, Any]) -> dict[str, Any]: entry_point = str(app.get("entry_point") or "") strategy = self._strategy(app) entry: dict[str, Any] = { "version": app.get("version") or "unknown", "entry_point": entry_point, "source": app.get("_source") or "harness", "strategy": strategy, "installed_at": int(_now()), } resolved = shutil.which(entry_point) if entry_point else None if resolved: entry["entry_point_path"] = resolved if strategy == "pip": distribution = _console_script_distribution(entry_point) if distribution: entry["pip_distribution"] = distribution return entry def _fetch_skill_content(self, app: dict[str, Any]) -> str | None: skill_md = str(app.get("skill_md") or "").strip() if not skill_md: return None url = _skill_content_url(skill_md, raw_base=str(app.get("_raw_base") or CLI_ANYTHING_RAW_BASE)) if not url: return None try: response = httpx.get(url, timeout=15.0, follow_redirects=True) response.raise_for_status() text = response.text except Exception: return None if "SKILL.md" not in url and not text.lstrip().startswith("---"): return None return text if len(text) < 250_000 else None def _fallback_skill(self, app: dict[str, Any]) -> str: name = str(app.get("name") or "unknown") display = str(app.get("display_name") or name) entry = str(app.get("entry_point") or f"cli-anything-{name}") description = _catalog_description(app) or f"Use {display} from nanobot." return f"""--- name: {_safe_skill_name(name)} description: >- {description} --- # {display} Use this skill when the user asks nanobot to operate {display} through its installed CLI app. If the user attached `@{name}` in chat, treat that as the selected app for the current turn. ## Commands ```bash {entry} --help {entry} --json --help ``` Prefer machine-readable output when the CLI supports `--json`. """ def _with_nanobot_skill_note(self, content: str, app: dict[str, Any]) -> str: marker = "" if marker in content: return content name = str(app.get("name") or "unknown") note = f"""{marker} ## Nanobot execution Use the `run_cli_app` tool with `name="{name}"` for command execution. Do not invoke this CLI through shell unless the user explicitly asks. Prefer this skill when Runtime Context mentions `@{name}` as a CLI App Attachment. """ lines = content.splitlines(keepends=True) if lines and lines[0].strip() == "---": for index, line in enumerate(lines[1:], start=1): if line.strip() == "---": return "".join(lines[: index + 1]) + "\n" + note + "\n" + "".join(lines[index + 1 :]) return note + "\n" + content def install_skill(self, app: dict[str, Any]) -> Path: path = self._skill_path(str(app["name"])) path.parent.mkdir(parents=True, exist_ok=True) content = self._fetch_skill_content(app) or self._fallback_skill(app) content = self._with_nanobot_skill_note(content, app) path.write_text(content, encoding="utf-8") return path def remove_skill(self, name: str) -> None: skill_dir = self._skill_path(name).parent if skill_dir.is_dir(): shutil.rmtree(skill_dir) def _record_installed(self, app: dict[str, Any]) -> dict[str, Any]: installed = self._load_installed() entry = self._installed_entry(app) installed[str(app["name"])] = entry self._save_installed(installed) self.install_skill(app) return entry def install(self, name: str) -> dict[str, Any]: app = self.get_app(name) if not self._install_supported(app): raise CliAppError("this CLI app uses an unsupported install strategy") strategy = self._strategy(app) entry_point = str(app.get("entry_point") or "") if entry_point and shutil.which(entry_point): self._record_installed(app) return self.payload() | { "last_action": { "ok": True, "message": f"CLI for {app['display_name']} is already available.", "installed": True, "verification": ["entry_point_available", "state_recorded", "managed_paths_present"], } } if strategy == "bundled": detect_cmd = str(app.get("detect_cmd") or app.get("entry_point") or "") if detect_cmd and _command_exists(detect_cmd): self._record_installed(app) return self.payload() | { "last_action": { "ok": True, "message": f"CLI for {app['display_name']} is available.", "installed": True, "verification": ["entry_point_available", "state_recorded"], } } note = app.get("install_notes") or f"{app['display_name']} is bundled with its parent app." raise CliAppError(str(note)) argv = self._argv_for_action(app, "install") assert argv is not None result = self._run_argv(argv, timeout=self.runtime.install_timeout) if strategy == "npm" and result.returncode != 0: result = self._retry_stale_npm_install(app, argv, result) if result.returncode != 0: raise CliAppError(_truncate(result.stderr or result.stdout or "install failed"), status=500) self._record_installed(app) return self.payload() | { "last_action": { "ok": True, "message": f"Installed CLI for {app['display_name']}.", "installed": True, "verification": ["package_manager_ok", "state_recorded", "managed_paths_present"], } } def update(self, name: str) -> dict[str, Any]: app = self.get_app(name, force_refresh=True) if str(app["name"]) not in self._load_installed(): raise CliAppError("CLI app is not installed") if self._strategy(app) == "bundled": self._record_installed(app) return self.payload() | { "last_action": { "ok": True, "message": f"Checked {app['display_name']}.", "installed": True, "verification": ["state_recorded"], } } argv = self._argv_for_action(app, "update") assert argv is not None result = self._run_argv(argv, timeout=self.runtime.install_timeout) if result.returncode != 0: raise CliAppError(_truncate(result.stderr or result.stdout or "update failed"), status=500) self._record_installed(app) return self.payload() | { "last_action": { "ok": True, "message": f"Updated CLI for {app['display_name']}.", "installed": True, "verification": ["package_manager_ok", "state_recorded", "managed_paths_present"], } } def uninstall(self, name: str) -> dict[str, Any]: app = self.get_app(name) installed = self._load_installed() if str(app["name"]) not in installed: raise CliAppError("CLI app is not installed") raw_installed_entry = installed.get(str(app["name"])) installed_entry = raw_installed_entry if isinstance(raw_installed_entry, dict) else {} strategy = self._strategy(app) entry_point = str(app.get("entry_point") or "").strip() managed_entry_path = str(installed_entry.get("entry_point_path") or "").strip() if strategy != "bundled": argv = self._argv_for_action(app, "uninstall", installed_entry=installed_entry) assert argv is not None result = self._run_argv(argv, timeout=self.runtime.install_timeout) if result.returncode != 0: raise CliAppError(_truncate(result.stderr or result.stdout or "uninstall failed"), status=500) still_managed = bool(managed_entry_path and Path(managed_entry_path).exists()) still_available = bool(entry_point and shutil.which(entry_point)) if still_managed or (not managed_entry_path and still_available): reason = ( f"the recorded entry point at {managed_entry_path} still exists" if still_managed else f"{entry_point} is still available on PATH" ) message = ( f"Uninstall for {app['display_name']} completed, but {reason}, " "so nanobot kept it installed." ) return self.payload() | { "last_action": { "ok": False, "message": message, "removed": False, "still_available": True, "verification_failed": ["entry_point_absent"], } } else: still_available = bool(entry_point and shutil.which(entry_point)) installed.pop(str(app["name"]), None) self._save_installed(installed) self.remove_skill(str(app["name"])) if strategy == "bundled" and still_available: message = ( f"Removed {app['display_name']} from nanobot. {entry_point} " "is still available because it is managed outside nanobot." ) elif still_available: message = ( f"Uninstalled CLI for {app['display_name']}, but another {entry_point} " "is still available on PATH." ) else: message = f"Uninstalled CLI for {app['display_name']}." return self.payload() | { "last_action": { "ok": True, "message": message, "removed": True, "still_available": still_available, "verification": ["state_absent", "managed_paths_absent"] if still_available else ["entry_point_absent", "state_absent", "managed_paths_absent"], } } def test(self, name: str) -> dict[str, Any]: app = self.get_app(name) entry = str(app.get("entry_point") or "") resolved = shutil.which(entry) if not entry or not resolved: raise CliAppError(f"{entry or name} is not available on PATH") result = self._run_argv([resolved, "--help"], timeout=min(self.runtime.run_timeout, 30)) ok = result.returncode == 0 output = _truncate((result.stdout or result.stderr or "").strip(), 3000) return self.payload() | { "last_action": { "ok": ok, "message": f"{entry} --help exited {result.returncode}", "output": output, } } def _resolve_cwd( self, working_dir: str | None, *, restrict_to_workspace: bool, ) -> Path: cwd = Path(working_dir).expanduser() if working_dir else self.workspace cwd = cwd.resolve(strict=False) workspace = self.workspace.resolve(strict=False) if restrict_to_workspace and not is_path_within(cwd, workspace): raise CliAppError("working_dir is outside the configured workspace") return cwd def _iter_artifact_candidates(self, cwd: Path) -> list[Path]: if not cwd.is_dir(): return [] out: list[Path] = [] stack = [cwd] scanned = 0 while stack and scanned < _MAX_ARTIFACT_SCAN_PATHS: directory = stack.pop() try: entries = sorted(directory.iterdir(), key=lambda path: path.name.lower()) except OSError: continue for path in entries: if scanned >= _MAX_ARTIFACT_SCAN_PATHS: break scanned += 1 try: if path.is_dir() and not path.is_symlink(): if path.name not in _ARTIFACT_IGNORE_DIRS: stack.append(path) continue if path.is_file() and path.suffix.lower() in _ARTIFACT_EXTENSIONS: out.append(path.resolve(strict=False)) except OSError: continue return out def _artifact_snapshot(self, cwd: Path) -> dict[Path, tuple[int, int]]: snapshot: dict[Path, tuple[int, int]] = {} for path in self._iter_artifact_candidates(cwd): try: stat = path.stat() except OSError: continue snapshot[path] = (stat.st_mtime_ns, stat.st_size) return snapshot def _changed_artifacts( self, cwd: Path, before: dict[Path, tuple[int, int]], ) -> list[Path]: changed: list[tuple[int, Path]] = [] for path, stamp in self._artifact_snapshot(cwd).items(): if before.get(path) == stamp: continue changed.append((stamp[0], path)) changed.sort(key=lambda item: (item[0], item[1].name.lower())) return [path for _, path in changed[-_MAX_ARTIFACT_REPORT:]] def _format_artifact_path(self, cwd: Path, path: Path) -> str: try: return path.relative_to(cwd).as_posix() except ValueError: return path.name @staticmethod def _format_artifact_size(path: Path) -> str: try: size = path.stat().st_size except OSError: return "unknown size" if size < 1024: return f"{size} B" if size < 1024 * 1024: return f"{size / 1024:.1f} KB" return f"{size / (1024 * 1024):.1f} MB" def _format_artifact_lines(self, cwd: Path, paths: list[Path]) -> list[str]: lines: list[str] = [] for path in paths: rel = self._format_artifact_path(cwd, path) ext = path.suffix.lower() kind = ( "previewable image" if ext in _INLINE_ARTIFACT_EXTENSIONS else ext.lstrip(".") or "file" ) lines.append(f"- {rel} ({kind}, {self._format_artifact_size(path)})") return lines def run( self, name: str, args: list[str] | None = None, *, json_output: bool = False, working_dir: str | None = None, timeout: int | None = None, restrict_to_workspace: bool = False, ) -> str: app = self.get_app(name) installed = self._load_installed() if str(app["name"]) not in installed: raise CliAppError(f"CLI app '{name}' is not installed") cwd = self._resolve_cwd(working_dir, restrict_to_workspace=restrict_to_workspace) entry = str(installed[str(app["name"])].get("entry_point") or app.get("entry_point") or "") resolved = shutil.which(entry) if not entry or not resolved: raise CliAppError(f"{entry or name} is not available on PATH") clean_args = [str(arg) for arg in (args or [])] if json_output and "--json" not in clean_args: clean_args = ["--json", *clean_args] effective_timeout = max(1, min(timeout or self.runtime.run_timeout, 600)) artifact_snapshot = self._artifact_snapshot(cwd) try: result = subprocess.run( [resolved, *clean_args], cwd=str(cwd), capture_output=True, text=True, timeout=effective_timeout, env=os.environ.copy(), ) except subprocess.TimeoutExpired: return f"CLI app '{name}' timed out after {effective_timeout}s" output = [ f"CLI app '{name}' exited {result.returncode}.", f"Command: {entry} {' '.join(shlex.quote(arg) for arg in clean_args)}".rstrip(), ] if result.stdout: output.append("\nSTDOUT:\n" + result.stdout.rstrip()) if result.stderr: output.append("\nSTDERR:\n" + result.stderr.rstrip()) artifacts = self._changed_artifacts(cwd, artifact_snapshot) if artifacts: output.append( "\nArtifacts created or updated:\n" + "\n".join(self._format_artifact_lines(cwd, artifacts)) ) if any(path.suffix.lower() in _INLINE_ARTIFACT_EXTENSIONS for path in artifacts): output.append( "\nTo show a preview in WebUI, reference a raster artifact with Markdown " "using its workspace-relative path, for example `![diagram](diagram.png)`." ) return _truncate("\n".join(output))