"""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 collections.abc import Iterable from dataclasses import dataclass from importlib import metadata as importlib_metadata from pathlib import Path from typing import Any, cast from urllib.parse import urlparse import httpx import yaml 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" AGENT_PLUGIN_SCHEMA = "https://agent-plugins.org/schemas/1.0.0/plugin.schema.json" 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) _SKILL_FRONTMATTER_RE = re.compile(r"^---\s*\r?\n(.*?)\r?\n---\s*\r?\n?", re.DOTALL) _SKILL_NAME_LINE_RE = re.compile(r"^name\s*:.*$", re.MULTILINE) _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", "obsidian-agent": "obsidian", "obsidian-agent-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 _as_object_dict(value: object) -> dict[str, Any] | None: """Narrow a JSON-like object to the string-keyed mapping used by this module.""" return cast(dict[str, Any], value) if isinstance(value, dict) else None def _safe_skill_name(name: str) -> str: clean = _SAFE_NAME_RE.sub("-", name.lower()).replace("_", "-").strip("-") return f"cli-app-{clean or 'app'}" def _legacy_skill_name(name: str) -> str: """Return the workspace skill name emitted before Agent Plugins support.""" clean = _SAFE_NAME_RE.sub("-", name.lower()).strip("-") return f"cli-app-{clean or 'app'}" def _plugin_skill_relative_path(name: str) -> str: skill_name = _safe_skill_name(name) return f"plugins/{skill_name}/skills/{skill_name}/SKILL.md" 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: object = cast(Any, distribution.metadata).get("Name") except Exception: name = None fallback_name = cast(object, getattr(distribution, "name", "")) return str(name or fallback_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: object = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return None return _as_object_dict(data) 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 = _as_object_dict(cached.get("data")) if data is None: 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 = _as_object_dict(data.get("apps")) return apps if apps is not None else data 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 = _as_object_dict(response.json()) if fetched is None: 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 = _as_object_dict(response.json()) if fetched is None: 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 = _as_object_dict(registry.get("meta")) if meta is not None and isinstance(meta.get("updated"), str): updated_values.append(meta["updated"]) for row in cast(Iterable[object], registry.get("clis", [])): entry = _as_object_dict(row) if entry is None or not entry.get("name"): continue entry = dict(entry) 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), _as_object_dict(data) or {}) 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": self.skill_relative_path(installed_name), "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: skill_name = _safe_skill_name(name) return self.workspace / "plugins" / skill_name / "skills" / skill_name / "SKILL.md" def _legacy_skill_path(self, name: str) -> Path: return self.workspace / "skills" / _legacy_skill_name(name) / "SKILL.md" def _installed_skill_path(self, name: str) -> Path: path = self._skill_path(name) legacy_path = self._legacy_skill_path(name) return legacy_path if not path.is_file() and legacy_path.is_file() else path def skill_relative_path(self, name: str) -> str: """Return the existing skill path, falling back to the canonical plugin path.""" return self._installed_skill_path(name).relative_to(self.workspace).as_posix() 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._installed_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 = _plugin_skill_relative_path(name) plugin_path = f"plugins/{_safe_skill_name(name)}" 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": [plugin_path], "verification": ["entry_point_available"] if entry_point else [], }) remove = compact_dict({ "supported": strategy != "unsupported", "strategy": strategy, "managed_paths": [plugin_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() cached_apps, _ = self.catalog(cache_only=True) cached_by_name = { str(app.get("name") or "").lower(): app for app in cached_apps if app.get("name") } rows: list[dict[str, Any]] = [] for name, raw_entry in sorted(installed.items()): entry = _as_object_dict(raw_entry) if entry is None: entry = {} strategy = str(entry.get("strategy") or "bundled") cached_app = cached_by_name.get(str(name).lower(), {}) app: dict[str, Any] = { "name": str(name), "display_name": str( cached_app.get("display_name") or entry.get("display_name") or name ), "category": str(cached_app.get("category") or entry.get("category") or "installed"), "description": str(cached_app.get("description") or entry.get("description") or ""), "requires": str(cached_app.get("requires") or entry.get("requires") or ""), "_source": str(entry.get("source") or "local"), "entry_point": str(entry.get("entry_point") or ""), "package_manager": strategy, "logo_url": cached_app.get("logo_url") or entry.get("logo_url"), "brand_color": cached_app.get("brand_color") or entry.get("brand_color"), } 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, encoding="utf-8", errors="replace", 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()), } for field in ( "display_name", "category", "description", "requires", "logo_url", "brand_color", ): value = app.get(field) if value not in (None, ""): entry[field] = value 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.")[:1024] return f"""--- name: {_safe_skill_name(name)} description: {json.dumps(description, ensure_ascii=False)} --- # {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 _normalise_skill(self, content: str, app: dict[str, Any]) -> str: """Give a catalog skill the identity required by its plugin directory.""" match = _SKILL_FRONTMATTER_RE.match(content) if match is None: return self._fallback_skill(app) try: metadata = _as_object_dict(cast(object, yaml.safe_load(match.group(1)))) except yaml.YAMLError: return self._fallback_skill(app) description = metadata.get("description") if metadata is not None else None if not isinstance(description, str) or not 1 <= len(description.strip()) <= 1024: return self._fallback_skill(app) name = _safe_skill_name(str(app["name"])) frontmatter, replaced = _SKILL_NAME_LINE_RE.subn(f"name: {name}", match.group(1), count=1) if not replaced: frontmatter = f"name: {name}\n{frontmatter}" body = content[match.end():].lstrip() return f"---\n{frontmatter.strip()}\n---\n\n{body}" 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._normalise_skill(content, app) content = self._with_nanobot_skill_note(content, app) path.write_text(content, encoding="utf-8") plugin_root = path.parents[2] manifest = compact_dict({ "$schema": AGENT_PLUGIN_SCHEMA, "name": _safe_skill_name(str(app["name"])), "version": str(app.get("version") or ""), "description": _catalog_description(app), }) _write_json(plugin_root / "plugin.json", manifest) legacy_dir = self._legacy_skill_path(str(app["name"])).parent if legacy_dir.is_dir(): shutil.rmtree(legacy_dir) return path def remove_skill(self, name: str) -> None: plugin_root = self._skill_path(name).parents[2] if plugin_root.is_dir(): shutil.rmtree(plugin_root) legacy_dir = self._legacy_skill_path(name).parent if legacy_dir.is_dir(): shutil.rmtree(legacy_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 = _as_object_dict(raw_installed_entry) if installed_entry is None: installed_entry = {} 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, encoding="utf-8", errors="replace", 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))