#!/usr/bin/env python3 """Deterministic helpers for the agent-guided local installation flow. This script deliberately does not clone the repository, install global tools, authenticate to NocoBase, or edit an AI client's configuration. Those actions need environment-specific judgment or direct user participation and are governed by INSTALL.md. """ from __future__ import annotations import argparse import json import os import re import shutil import subprocess import sys import tempfile from collections.abc import Callable, Sequence from dataclasses import asdict, dataclass from pathlib import Path from typing import Any from urllib.parse import urlsplit, urlunsplit MIN_PYTHON = (3, 12) DEFAULT_ENV_NAME = "nb-mcp" SERVER_NAME = "nocobase-crm" REQUIRED_PROJECT_FILES = ("pyproject.toml", "uv.lock", "config.example.json") ENV_NAME_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$") ANSI_ESCAPE_PATTERN = re.compile(r"\x1b\[[0-?]*[ -/]*[@-~]") Which = Callable[[str], str | None] Runner = Callable[..., subprocess.CompletedProcess[str]] class InstallerError(RuntimeError): """Raised when a deterministic installation step cannot finish safely.""" @dataclass(frozen=True) class Check: name: str ok: bool required: bool detail: str def _project_root(path: str | Path | None) -> Path: candidate = Path(path) if path is not None else Path(__file__).resolve().parent.parent return candidate.expanduser().resolve() def inspect_project( project_root: str | Path | None = None, *, which: Which = shutil.which, python_version: tuple[int, int, int] | None = None, ) -> dict[str, Any]: root = _project_root(project_root) version = python_version or (sys.version_info.major, sys.version_info.minor, sys.version_info.micro) commands = {name: which(name) for name in ("git", "uv", "nb")} missing_files = [name for name in REQUIRED_PROJECT_FILES if not (root / name).is_file()] checks = [ Check( "python", version[:2] >= MIN_PYTHON, True, f"Python {version[0]}.{version[1]}.{version[2]} (requires >= {MIN_PYTHON[0]}.{MIN_PYTHON[1]})", ), Check("git", bool(commands["git"]), True, commands["git"] or "git was not found on PATH"), Check("uv", bool(commands["uv"]), True, commands["uv"] or "uv was not found on PATH"), Check("nb", bool(commands["nb"]), True, commands["nb"] or "nb was not found on PATH"), Check( "project_files", not missing_files, True, "required project files are present" if not missing_files else f"missing: {', '.join(missing_files)}", ), Check( "config", (root / "config.json").is_file(), False, "config.json is present" if (root / "config.json").is_file() else "config.json will be created by prepare", ), ] return { "ok": all(check.ok for check in checks if check.required), "projectRoot": str(root), "checks": [asdict(check) for check in checks], } def _require_prepare_prerequisites( root: Path, *, which: Which, python_version: tuple[int, int, int], ) -> str: inspection = inspect_project(root, which=which, python_version=python_version) required = {check["name"]: check for check in inspection["checks"] if check["required"]} blocking = [required[name]["detail"] for name in ("python", "uv", "project_files") if not required[name]["ok"]] if blocking: raise InstallerError("; ".join(blocking)) uv_bin = which("uv") if not uv_bin: raise InstallerError("uv was not found on PATH") return os.path.abspath(uv_bin) def _load_json_object(path: Path) -> dict[str, Any]: try: value = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as error: raise InstallerError(f"Could not read valid JSON from {path}: {error}") from error if not isinstance(value, dict): raise InstallerError(f"Expected a JSON object in {path}") return value def prepare_project( project_root: str | Path | None = None, *, runner: Runner = subprocess.run, which: Which = shutil.which, python_version: tuple[int, int, int] | None = None, ) -> dict[str, Any]: root = _project_root(project_root) version = python_version or (sys.version_info.major, sys.version_info.minor, sys.version_info.micro) uv_bin = _require_prepare_prerequisites(root, which=which, python_version=version) example_path = root / "config.example.json" config_path = root / "config.json" _load_json_object(example_path) created_config = False if config_path.exists(): _load_json_object(config_path) else: shutil.copy2(example_path, config_path) created_config = True command = [uv_bin, "sync", "--locked", "--python", f"{MIN_PYTHON[0]}.{MIN_PYTHON[1]}"] try: result = runner(command, cwd=str(root), capture_output=True, text=True, check=False) except OSError as error: raise InstallerError(f"Could not execute uv sync: {error}") from error if result.returncode != 0: raise InstallerError( f"uv sync --locked failed with exit code {result.returncode}; rerun it directly to inspect package-manager output" ) return { "ok": True, "projectRoot": str(root), "configPath": str(config_path), "configCreated": created_config, "dependencies": "locked", } def set_config_env(project_root: str | Path | None, env_name: str) -> dict[str, Any]: if not ENV_NAME_PATTERN.fullmatch(env_name): raise InstallerError( "Environment name must start with a letter or digit and contain only letters, digits, dot, underscore, or dash" ) root = _project_root(project_root) config_path = root / "config.json" config = _load_json_object(config_path) previous = config.get("env") if previous == env_name: return { "ok": True, "changed": False, "configPath": str(config_path), "env": env_name, "backupPath": None, } backup_path = root / "config.json.pre-agent-install.bak" if not backup_path.exists(): shutil.copy2(config_path, backup_path) config["env"] = env_name fd, temporary_name = tempfile.mkstemp(prefix=".config.json.", dir=root) temporary_path = Path(temporary_name) try: with os.fdopen(fd, "w", encoding="utf-8") as handle: json.dump(config, handle, ensure_ascii=False, indent=2) handle.write("\n") os.replace(temporary_path, config_path) finally: if temporary_path.exists(): temporary_path.unlink() return { "ok": True, "changed": True, "configPath": str(config_path), "env": env_name, "previousEnv": previous, "backupPath": str(backup_path), } def normalize_api_base_url(value: str) -> str: raw = value.strip() if not raw: raise InstallerError("NocoBase API base URL is required") parsed = urlsplit(raw) if parsed.scheme not in {"http", "https"} or not parsed.hostname: raise InstallerError("NocoBase URL must be an absolute http:// or https:// URL") if parsed.username or parsed.password or parsed.query or parsed.fragment: raise InstallerError("NocoBase URL must not contain credentials, a query string, or a fragment") hostname = parsed.hostname.lower() is_local = hostname == "localhost" or hostname == "::1" or hostname.startswith("127.") if parsed.scheme != "https" and not is_local: raise InstallerError("A remote NocoBase API Key may only be sent to an https:// endpoint") path = parsed.path.rstrip("/") if not path.endswith("/api"): path = f"{path}/api" if path else "/api" return urlunsplit((parsed.scheme, parsed.netloc, path, "", "")) def _parse_cli_json(output: str) -> dict[str, Any]: decoder = json.JSONDecoder() candidates: list[tuple[int, int, dict[str, Any]]] = [] for start, character in enumerate(output): if character != "{": continue try: value, end = decoder.raw_decode(output, start) except json.JSONDecodeError: continue if isinstance(value, dict): candidates.append((start, end, value)) outer_candidates = [ candidate for candidate in candidates if not any( other_start <= candidate[0] and other_end >= candidate[1] and (other_start, other_end) != (candidate[0], candidate[1]) for other_start, other_end, _value in candidates ) ] if len(outer_candidates) != 1: raise InstallerError("nb env info did not return one JSON object") return outer_candidates[0][2] def _nb_command_environment(nb_bin: str) -> dict[str, str]: environment = os.environ.copy() environment["NB_CLI_LOG_DISABLED"] = "1" environment["NB_SKIP_STARTUP_UPDATE"] = "1" nb_path = Path(nb_bin) if nb_path.is_absolute(): nb_dir = str(nb_path.parent) existing_path = environment.get("PATH", "") parts = [part for part in existing_path.split(os.pathsep) if part] if nb_dir not in parts: environment["PATH"] = nb_dir + (os.pathsep + existing_path if existing_path else "") return environment def _read_nb_environment( root: Path, env_name: str, *, nb_bin: str, runner: Runner, ) -> dict[str, Any] | None: command = [nb_bin, "env", "info", env_name, "--json"] try: result = runner( command, cwd=str(root), env=_nb_command_environment(nb_bin), capture_output=True, text=True, check=False, ) except OSError as error: raise InstallerError(f"Could not execute nb env info: {error}") from error if result.returncode == 0: return _parse_cli_json(result.stdout) diagnostic = ANSI_ESCAPE_PATTERN.sub("", f"{result.stdout}\n{result.stderr}") if "is not configured" in diagnostic: return None raise InstallerError( f"nb env info failed with exit code {result.returncode}; inspect the nb CLI diagnostic locally" ) def _api_details(payload: dict[str, Any]) -> tuple[str | None, str | None]: api = payload.get("api") auth = api.get("auth") if isinstance(api, dict) else None api_base_url = api.get("apiBaseUrl") if isinstance(api, dict) else None auth_type = auth.get("type") if isinstance(auth, dict) else None normalized_url = api_base_url.strip().rstrip("/") if isinstance(api_base_url, str) else None normalized_type = auth_type.strip() if isinstance(auth_type, str) else None return normalized_url, normalized_type def _suggest_env_name(api_base_url: str) -> str: hostname = urlsplit(api_base_url).hostname or "server" suffix = re.sub(r"[^a-z0-9]+", "-", hostname.lower()).strip("-")[:32] or "server" return f"{DEFAULT_ENV_NAME}-{suffix}" def connect_with_personal_key( project_root: str | Path | None, api_base_url: str, *, env_name: str = DEFAULT_ENV_NAME, runner: Runner = subprocess.run, which: Which = shutil.which, stdin_isatty: Callable[[], bool] = sys.stdin.isatty, ) -> dict[str, Any]: """Bind a personal API Key through nb's private TTY prompt. The key is never accepted by this function, placed in argv, inherited from the process environment, captured, printed, or returned. The child nb CLI reads it directly from the user-controlled terminal. """ if not ENV_NAME_PATTERN.fullmatch(env_name): raise InstallerError( "Environment name must start with a letter or digit and contain only letters, digits, dot, underscore, or dash" ) if not stdin_isatty(): raise InstallerError( "Secure API Key entry requires a user-controlled local terminal. " "Do not paste the Key into Agent chat or pass it with --access-token." ) root = _project_root(project_root) normalized_url = normalize_api_base_url(api_base_url) discovered_nb = which("nb") if not discovered_nb: raise InstallerError("nb was not found on PATH") nb_bin = os.path.abspath(discovered_nb) existing = _read_nb_environment(root, env_name, nb_bin=nb_bin, runner=runner) if existing is None: interactive_command = [ nb_bin, "env", "add", env_name, "--api-base-url", normalized_url, "--auth-type", "token", ] action = "created" else: existing_url, _auth_type = _api_details(existing) if existing_url != normalized_url.rstrip("/"): suggestion = _suggest_env_name(normalized_url) raise InstallerError( f"Environment {env_name!r} already targets {existing_url or 'an unconfigured endpoint'!r}; " f"use a new name such as {suggestion!r} instead of overwriting it" ) interactive_command = [nb_bin, "env", "auth", env_name, "--auth-type", "token"] action = "reauthenticated" try: result = runner( interactive_command, cwd=str(root), env=_nb_command_environment(nb_bin), text=True, check=False, ) except OSError as error: raise InstallerError(f"Could not execute the interactive nb Key prompt: {error}") from error if result.returncode != 0: raise InstallerError( f"Interactive API Key authentication failed with exit code {result.returncode}; " "revoke the exposed or invalid Key if necessary and retry through the installer" ) verified = _read_nb_environment(root, env_name, nb_bin=nb_bin, runner=runner) if verified is None: raise InstallerError("nb reported success but the saved environment could not be read back") verified_url, auth_type = _api_details(verified) if verified_url != normalized_url.rstrip("/") or auth_type != "token": raise InstallerError("The saved nb environment did not verify as token authentication for the requested URL") list_result = runner( [nb_bin, "env", "list"], cwd=str(root), env=_nb_command_environment(nb_bin), capture_output=True, text=True, check=False, ) if list_result.returncode != 0: raise InstallerError("API Key was saved, but nb env list readback failed") config_result = set_config_env(root, env_name) return { "ok": True, "action": action, "env": env_name, "apiBaseUrl": normalized_url, "authType": "token", "keyExposure": "private-tty-only", "configChanged": config_result["changed"], "configBackupPath": config_result["backupPath"], } def render_client_config( project_root: str | Path | None = None, *, enable_writes: bool = False, which: Which = shutil.which, ) -> dict[str, Any]: root = _project_root(project_root) uv_bin = which("uv") or "uv" nb_bin = which("nb") or "nb" if os.path.isabs(uv_bin): uv_bin = os.path.abspath(uv_bin) if os.path.isabs(nb_bin): nb_bin = os.path.abspath(nb_bin) args = [ "--directory", str(root), "run", "--no-sync", "nocobase-crm-mcp", "--config", str(root / "config.json"), ] if enable_writes: args.append("--enable-writes") return { "mcpServers": { SERVER_NAME: { "command": uv_bin, "args": args, "env": {"NOCOBASE_MCP_NB_BIN": nb_bin}, } } } def doctor_project( project_root: str | Path | None = None, *, runner: Runner = subprocess.run, which: Which = shutil.which, python_version: tuple[int, int, int] | None = None, ) -> dict[str, Any]: root = _project_root(project_root) inspection = inspect_project(root, which=which, python_version=python_version) checks = [Check(**item) for item in inspection["checks"]] config_path = root / "config.json" if config_path.is_file(): try: config = _load_json_object(config_path) env_name = config.get("env") config_ok = isinstance(env_name, str) and bool(env_name.strip()) detail = f"config env is {env_name!r}" if config_ok else "config.json has no non-empty env" except InstallerError as error: config_ok = False detail = str(error) else: config_ok = False detail = "config.json is missing; run prepare" checks.append(Check("configured_env", config_ok, True, detail)) uv_bin = which("uv") if uv_bin: commands = ( [uv_bin, "lock", "--check"], [ uv_bin, "run", "--no-sync", "python", "-c", "import mcp; import nocobase_crm_mcp", ], ) labels = ("lockfile", "python_environment") for label, command in zip(labels, commands, strict=True): try: result = runner(command, cwd=str(root), capture_output=True, text=True, check=False) ok = result.returncode == 0 detail = "passed" if ok else f"failed with exit code {result.returncode}" except OSError as error: ok = False detail = str(error) checks.append(Check(label, ok, True, detail)) else: checks.extend( [ Check("lockfile", False, True, "uv was not found on PATH"), Check("python_environment", False, True, "uv was not found on PATH"), ] ) return { "ok": all(check.ok for check in checks if check.required), "scope": "local-installation", "projectRoot": str(root), "checks": [asdict(check) for check in checks], "remainingVerification": [ "Have the installation Agent verify the saved env through nb CLI's default redacted info output.", "Register the generated stdio MCP configuration in the user's client.", "Call the MCP tool crm_health and require ok=true.", ], } def _print_result(payload: dict[str, Any], *, as_json: bool) -> None: if as_json: print(json.dumps(payload, ensure_ascii=False, indent=2)) return print(f"Overall: {'ready' if payload.get('ok') else 'action required'}") if "projectRoot" in payload: print(f"Project root: {payload['projectRoot']}") for check in payload.get("checks", []): marker = "PASS" if check["ok"] else ("FAIL" if check["required"] else "INFO") print(f"[{marker}] {check['name']}: {check['detail']}") def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="Prepare and inspect an agent-guided NocoBase CRM MCP installation.") parser.add_argument("--project-root", type=Path, default=None) subparsers = parser.add_subparsers(dest="command", required=True) for name in ("inspect", "prepare", "doctor"): command = subparsers.add_parser(name) command.add_argument("--json", action="store_true", dest="as_json") set_env = subparsers.add_parser("set-env") set_env.add_argument("env_name") set_env.add_argument("--json", action="store_true", dest="as_json") connect_key = subparsers.add_parser("connect-key") connect_key.add_argument("--api-base-url", required=True) connect_key.add_argument("--env-name", default=DEFAULT_ENV_NAME) connect_key.add_argument("--json", action="store_true", dest="as_json") client_config = subparsers.add_parser("client-config") client_config.add_argument("--enable-writes", action="store_true") return parser def main(argv: Sequence[str] | None = None) -> int: parser = build_parser() args = parser.parse_args(argv) try: if args.command == "inspect": payload = inspect_project(args.project_root) _print_result(payload, as_json=args.as_json) return 0 if payload["ok"] else 1 if args.command == "prepare": payload = prepare_project(args.project_root) _print_result(payload, as_json=args.as_json) return 0 if args.command == "set-env": payload = set_config_env(args.project_root, args.env_name) _print_result(payload, as_json=args.as_json) return 0 if args.command == "connect-key": payload = connect_with_personal_key( args.project_root, args.api_base_url, env_name=args.env_name, ) _print_result(payload, as_json=args.as_json) return 0 if args.command == "client-config": print(json.dumps(render_client_config(args.project_root, enable_writes=args.enable_writes), indent=2)) return 0 if args.command == "doctor": payload = doctor_project(args.project_root) _print_result(payload, as_json=args.as_json) return 0 if payload["ok"] else 1 except InstallerError as error: print(json.dumps({"ok": False, "error": {"code": "install_step_failed", "message": str(error)}})) return 1 raise AssertionError(f"Unhandled command: {args.command}") if __name__ == "__main__": raise SystemExit(main())