feat(apps): unify CLI apps and MCP (#3991)

* refactor(cli): load bundled apps from catalog

* feat(plugins): unify CLI and MCP settings

* feat(plugins): add settings category filter

* style(plugins): refine settings catalog

* refactor(cli): load nanobot apps from repo catalog

* feat(store): add capability store entry

* feat(apps): rename capability store

* fix(apps): verify clean app removal

* fix(apps): keep main sidebar on apps view

* feat(apps): add shared app manifest protocol

* fix(apps): dismiss app status message

* refactor(apps): move CLI adapter under apps

* refactor(apps): drop legacy cli apps package
This commit is contained in:
Xubin Ren
2026-05-25 20:07:02 +08:00
committed by GitHub
parent 179acfe104
commit 418cb23da2
32 changed files with 2026 additions and 903 deletions
+1 -1
View File
@@ -13,7 +13,7 @@ from nanobot.agent.skills import SkillsLoader
from nanobot.agent.tools import mcp as mcp_tools
from nanobot.agent.tools.registry import ToolRegistry
from nanobot.bus.events import InboundMessage
from nanobot.cli_apps import utils as cli_app_utils
from nanobot.apps.cli import utils as cli_app_utils
from nanobot.session.goal_state import goal_state_runtime_lines
from nanobot.utils.helpers import (
current_time_str,
+1 -1
View File
@@ -9,7 +9,7 @@ from pydantic import Field
from nanobot.agent.tools.base import Tool, tool_parameters
from nanobot.agent.tools.schema import ArraySchema, BooleanSchema, IntegerSchema, StringSchema, tool_parameters_schema
from nanobot.cli_apps import CliAppError, CliAppManager, CliAppsRuntimeConfig
from nanobot.apps.cli import CliAppError, CliAppManager, CliAppsRuntimeConfig
from nanobot.config.schema import Base
+5
View File
@@ -0,0 +1,5 @@
"""Shared app protocol helpers."""
from nanobot.apps.protocol import APP_PROTOCOL_SCHEMA, app_manifest
__all__ = ["APP_PROTOCOL_SCHEMA", "app_manifest"]
@@ -1,6 +1,6 @@
"""CLI Apps integration helpers."""
"""CLI app adapter for the unified Apps domain."""
from nanobot.cli_apps.service import (
from nanobot.apps.cli.service import (
CliAppError,
CliAppManager,
CliAppsRuntimeConfig,
@@ -11,12 +11,14 @@ 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 nanobot.apps.protocol import app_manifest, compact_dict
from nanobot.config.paths import get_runtime_subdir
CLI_ANYTHING_REGISTRY_URL = "https://hkuds.github.io/CLI-Anything/registry.json"
@@ -139,7 +141,7 @@ _BRANDS: dict[str, tuple[str, str]] = {
_BRAND_DOMAINS: dict[str, tuple[str, str]] = {
"3mf": ("3mf.io", "#00A1DE"),
"anygen": ("anygen.com", "#111827"),
"anygen": ("anygen.io", "#111827"),
"clibrowser": ("github.com/allthingssecurity/clibrowser", "#24292F"),
"cloudanalyzer": ("github.com/rsasaki0109/CloudAnalyzer", "#2563EB"),
"cloudcompare": ("cloudcompare.org", "#4D83C3"),
@@ -244,6 +246,29 @@ def _pip_uninstall_args_from_command(command: str) -> list[str] | 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("-")
@@ -540,8 +565,86 @@ class CliAppManager:
"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=str(app.get("description") or ""),
category=str(app.get("category") or "uncategorized"),
source=f"cli-anything:{app.get('_source') or 'harness'}",
logo_url=logo_url,
brand_color=brand_color,
capabilities=capabilities,
install=install,
remove=remove,
trust={
"registry": "cli-anything",
"level": "catalog",
"review_status": "catalog_entry",
},
)
def payload(self, *, force_refresh: bool = False) -> dict[str, Any]:
apps, updated = self.catalog(force_refresh=force_refresh)
installed = self._load_installed()
@@ -581,7 +684,14 @@ class CliAppManager:
prefix.extend(["--upgrade", "--force-reinstall"])
return prefix + args
def _pip_uninstall_argv(self, app: dict[str, Any]) -> list[str]:
def _pip_uninstall_argv(
self,
app: dict[str, Any],
installed_entry: dict[str, Any] | None = None,
) -> list[str]:
distribution = str((installed_entry or {}).get("pip_distribution") or "").strip()
if distribution:
return [sys.executable, "-m", "pip", "uninstall", "-y", distribution]
uninstall_cmd = str(app.get("uninstall_cmd") or "")
packages = _pip_uninstall_args_from_command(uninstall_cmd)
if packages:
@@ -619,14 +729,19 @@ class CliAppManager:
raise CliAppError(f"unsupported {expected} command")
return argv
def _argv_for_action(self, app: dict[str, Any], action: str) -> list[str] | None:
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)
return self._pip_uninstall_argv(app, installed_entry=installed_entry)
if strategy == "npm":
return self._npm_argv(app, action)
if strategy == "brew":
@@ -648,13 +763,23 @@ class CliAppManager:
)
def _installed_entry(self, app: dict[str, Any]) -> dict[str, Any]:
return {
entry_point = str(app.get("entry_point") or "")
strategy = self._strategy(app)
entry: dict[str, Any] = {
"version": app.get("version") or "unknown",
"entry_point": app.get("entry_point") or "",
"entry_point": entry_point,
"source": app.get("_source") or "harness",
"strategy": self._strategy(app),
"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()
@@ -730,11 +855,13 @@ Use the `run_cli_app` tool with `name="{name}"` for command execution. Do not in
if skill_dir.is_dir():
shutil.rmtree(skill_dir)
def _record_installed(self, app: dict[str, Any]) -> None:
def _record_installed(self, app: dict[str, Any]) -> dict[str, Any]:
installed = self._load_installed()
installed[str(app["name"])] = self._installed_entry(app)
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)
@@ -745,7 +872,14 @@ Use the `run_cli_app` tool with `name="{name}"` for command execution. Do not in
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."}}
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")
@@ -754,7 +888,14 @@ Use the `run_cli_app` tool with `name="{name}"` for command execution. Do not in
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']}."}}
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)
@@ -762,30 +903,94 @@ Use the `run_cli_app` tool with `name="{name}"` for command execution. Do not in
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']}."}}
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']}."}}
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")
if self._strategy(app) != "bundled":
argv = self._argv_for_action(app, "uninstall")
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"]))
return self.payload() | {"last_action": {"ok": True, "message": f"Uninstalled CLI for {app['display_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)
@@ -46,7 +46,7 @@ def _cli_app_runtime_lines(
if "@" not in text:
return []
try:
from nanobot.cli_apps import CliAppManager
from nanobot.apps.cli import CliAppManager
mentions = CliAppManager(workspace=workspace).mentioned_installed_apps(text)
except Exception:
+56
View File
@@ -0,0 +1,56 @@
"""Neutral manifest shape for settings-managed agent apps.
The manifest is intentionally descriptive. Installers still live in their
own adapters, while this protocol gives the WebUI and future registries one
small vocabulary for capabilities, trust, and verified install/remove plans.
"""
from __future__ import annotations
from typing import Any
APP_PROTOCOL_SCHEMA = "agent-app.v1"
def compact_dict(values: dict[str, Any]) -> dict[str, Any]:
"""Drop empty optional values while preserving explicit booleans and zeros."""
return {
key: value
for key, value in values.items()
if value is not None and value != "" and value != [] and value != {}
}
def app_manifest(
*,
app_id: str,
display_name: str,
description: str,
category: str,
source: str,
capabilities: list[dict[str, Any]],
install: dict[str, Any],
remove: dict[str, Any],
trust: dict[str, Any],
version: str | None = None,
logo_url: str | None = None,
brand_color: str | None = None,
docs_url: str | None = None,
) -> dict[str, Any]:
"""Build a stable app manifest dictionary."""
return compact_dict({
"schema": APP_PROTOCOL_SCHEMA,
"id": app_id,
"display_name": display_name,
"version": version,
"description": description,
"category": category,
"source": source,
"logo_url": logo_url,
"brand_color": brand_color,
"docs_url": docs_url,
"capabilities": capabilities,
"install": install,
"remove": remove,
"trust": trust,
})
+7 -1
View File
@@ -10,10 +10,11 @@ import pydantic
from loguru import logger
from pydantic import BaseModel
from nanobot.config.schema import Config
from nanobot.config.schema import Config, _resolve_tool_config_refs
# Global variable to store current config path (for multi-instance support)
_current_config_path: Path | None = None
_schema_refs_ready = False
def set_config_path(path: Path) -> None:
@@ -39,6 +40,11 @@ def load_config(config_path: Path | None = None) -> Config:
Returns:
Loaded configuration object.
"""
global _schema_refs_ready
if not _schema_refs_ready:
_resolve_tool_config_refs()
_schema_refs_ready = True
path = config_path or get_config_path()
config = Config()
+1 -1
View File
@@ -5,7 +5,7 @@ from __future__ import annotations
import re
from typing import Any
from nanobot.cli_apps import CliAppError, CliAppManager, CliAppsRuntimeConfig
from nanobot.apps.cli import CliAppError, CliAppManager, CliAppsRuntimeConfig
from nanobot.config.loader import load_config
QueryParams = dict[str, list[str]]
+149 -3
View File
@@ -16,6 +16,7 @@ from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Literal, Mapping
from nanobot.apps.protocol import app_manifest, compact_dict
from nanobot.agent.tools.registry import ToolRegistry
from nanobot.config.loader import load_config, resolve_config_env_vars, save_config
from nanobot.config.paths import get_runtime_subdir
@@ -475,6 +476,20 @@ def _with_managed_stdio_cwd(name: str, cfg: MCPServerConfig) -> MCPServerConfig:
return cfg
def _remove_managed_stdio_cwd(name: str, cfg: MCPServerConfig | None) -> bool:
if cfg is None or not cfg.cwd:
return False
cwd = Path(cfg.cwd).expanduser().resolve(strict=False)
managed = (get_runtime_subdir("mcp") / name).resolve(strict=False)
if cwd != managed or not cwd.exists():
return False
if cwd.is_symlink() or cwd.is_file():
cwd.unlink()
else:
shutil.rmtree(cwd)
return True
def _url_with_param(url: str, key: str, value: str) -> str:
parsed = urllib.parse.urlsplit(url)
query = urllib.parse.parse_qsl(parsed.query, keep_blank_values=True)
@@ -647,10 +662,108 @@ def _tool_allowlist(cfg: MCPServerConfig | None) -> list[str]:
return list(cfg.enabled_tools)
def _managed_mcp_path(name: str, cfg: MCPServerConfig | None) -> list[str]:
if cfg is None or not cfg.command:
return []
return [f"runtime:mcp/{name}"]
def _preset_manifest(preset: McpPreset, *, logo_url: str) -> dict[str, Any]:
server = preset.server
managed_paths = _managed_mcp_path(preset.name, server)
field_specs = [
compact_dict({
"name": field.name,
"target": field.target[0],
"required": field.required,
"secret": field.secret,
"env_var": field.env_var,
})
for field in preset.fields
]
capabilities = [
compact_dict({
"type": "mcp",
"transport": preset.transport,
"command": server.command if server and server.command else None,
"args": list(server.args) if server and server.command else None,
"url": _connection_summary(server) if server and server.url else None,
"fields": field_specs,
})
]
return app_manifest(
app_id=preset.name,
display_name=preset.display_name,
description=preset.description,
category=preset.category,
source="mcp-preset",
docs_url=preset.docs_url,
logo_url=logo_url,
brand_color=preset.brand_color,
capabilities=capabilities,
install=compact_dict({
"supported": preset.install_supported,
"strategy": "config",
"managed_paths": managed_paths,
"verification": ["config_present", "dependency_available"],
}),
remove=compact_dict({
"supported": True,
"strategy": "config",
"managed_paths": managed_paths,
"verification": ["config_absent", "managed_paths_absent"] if managed_paths else ["config_absent"],
}),
trust={
"registry": "mcp-presets",
"level": "builtin",
"review_status": "builtin_preset",
},
)
def _custom_manifest(name: str, cfg: MCPServerConfig) -> dict[str, Any]:
transport = cfg.type or ("stdio" if cfg.command else "streamableHttp")
managed_paths: list[str] = []
return app_manifest(
app_id=name,
display_name=name,
description="Custom MCP server from nanobot config.",
category="custom",
source="mcp-custom",
brand_color="#64748B",
capabilities=[
compact_dict({
"type": "mcp",
"transport": transport,
"command": cfg.command or None,
"url": _connection_summary(cfg) if cfg.url else None,
})
],
install=compact_dict({
"supported": True,
"strategy": "config",
"managed_paths": managed_paths,
"verification": ["config_present", "dependency_available"],
}),
remove=compact_dict({
"supported": True,
"strategy": "config",
"managed_paths": managed_paths,
"verification": ["config_absent", "managed_paths_absent"] if managed_paths else ["config_absent"],
}),
trust={
"registry": "user-config",
"level": "user",
"review_status": "user_managed",
},
)
def _preset_payload(preset: McpPreset, configured_servers: dict[str, MCPServerConfig]) -> dict[str, Any]:
cfg = configured_servers.get(preset.name)
status = _status_for(preset, cfg)
configured = cfg is not None and status not in {"missing_credentials"}
logo_url = _favicon_url(preset.brand_domain)
return {
"name": preset.name,
"display_name": preset.display_name,
@@ -665,12 +778,13 @@ def _preset_payload(preset: McpPreset, configured_servers: dict[str, MCPServerCo
"configured": configured,
"available": configured and _config_available(cfg),
"status": status,
"logo_url": _favicon_url(preset.brand_domain),
"logo_url": logo_url,
"brand_color": preset.brand_color,
"required_fields": [_field_payload(field, cfg) for field in preset.fields],
"connection_summary": _connection_summary(cfg),
"enabled_tools": _tool_allowlist(cfg),
"source": "preset",
"manifest": _preset_manifest(preset, logo_url=logo_url),
}
@@ -705,6 +819,7 @@ def _custom_payload(
"enabled_tools": _tool_allowlist(cfg),
"tool_names": tool_names or [],
"source": "custom",
"manifest": _custom_manifest(name, cfg),
}
@@ -744,10 +859,17 @@ def _action_message(action: str, preset: McpPreset, *, ok: bool = True) -> dict[
"remove": "Removed",
"test": "Checked",
}.get(action, "Updated")
return {
payload: dict[str, Any] = {
"ok": ok,
"message": f"{verb} MCP preset for {preset.display_name}.",
}
if action == "enable":
payload["installed"] = True
payload["verification"] = ["config_present"]
elif action == "remove":
payload["removed"] = True
payload["verification"] = ["config_absent"]
return payload
def _server_action_message(action: str, name: str, *, ok: bool = True) -> dict[str, Any]:
@@ -758,10 +880,17 @@ def _server_action_message(action: str, name: str, *, ok: bool = True) -> dict[s
"tools": "Updated tools for",
"remove": "Removed",
}.get(action, "Updated")
return {
payload: dict[str, Any] = {
"ok": ok,
"message": f"{verb} MCP server {name}.",
}
if action in {"custom", "import", "import-cursor"}:
payload["installed"] = True
payload["verification"] = ["config_present"]
elif action == "remove":
payload["removed"] = True
payload["verification"] = ["config_absent"]
return payload
def _scrub_test_error(text: str) -> str:
@@ -1113,7 +1242,14 @@ def mcp_presets_action(action: str, query: QueryParams) -> dict[str, Any]:
if action == "remove":
if preset is None and name not in config.tools.mcp_servers:
raise McpPresetError("unknown MCP server", status=404)
removed_runtime_files = False
cleanup_error = ""
if name in config.tools.mcp_servers:
existing_cfg = config.tools.mcp_servers[name]
try:
removed_runtime_files = _remove_managed_stdio_cwd(name, existing_cfg)
except OSError as exc:
cleanup_error = str(exc)
del config.tools.mcp_servers[name]
save_config(config)
last_action = (
@@ -1121,6 +1257,16 @@ def mcp_presets_action(action: str, query: QueryParams) -> dict[str, Any]:
if preset is not None
else _server_action_message(action, name)
)
if removed_runtime_files:
last_action["message"] = f"{last_action['message']} Removed managed runtime files."
last_action["managed_paths_removed"] = [f"runtime:mcp/{name}"]
last_action["verification"] = ["config_absent", "managed_paths_absent"]
if cleanup_error:
last_action["ok"] = False
last_action["message"] = (
f"{last_action['message']} Could not remove managed runtime files: {cleanup_error}"
)
last_action["verification_failed"] = ["managed_paths_absent"]
payload = mcp_presets_payload(last_action=last_action)
payload["requires_restart"] = True
return payload