from __future__ import annotations from .common import ( ENV_NAME_PATTERN, MIN_PYTHON, REQUIRED_PROJECT_FILES, SERVER_NAME, Any, Check, InstallerError, Path, Runner, Which, _load_json_object, _project_root, asdict, json, os, shutil, subprocess, sys, tempfile, ) 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 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 venv_path = root / ".venv" command = [uv_bin, "sync", "--locked", "--no-dev", "--python", f"{MIN_PYTHON[0]}.{MIN_PYTHON[1]}"] command_environment = os.environ.copy() command_environment["UV_PROJECT_ENVIRONMENT"] = str(venv_path) try: result = runner( command, cwd=str(root), env=command_environment, 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-runtime-only", "environmentPath": str(venv_path), } 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 render_client_config( project_root: str | Path | None = None, *, enable_writes: bool = False, which: Which = shutil.which, os_name: str | None = None, ) -> dict[str, Any]: root = _project_root(project_root) nb_bin = which("nb") or "nb" if os.path.isabs(nb_bin): nb_bin = os.path.abspath(nb_bin) python_bin = _venv_python_path(root, os_name=os_name) args = [ "-m", "nocobase_crm_mcp.server", "--config", str(root / "config.json"), ] if enable_writes: args.append("--enable-writes") return { "mcpServers": { SERVER_NAME: { "command": str(python_bin), "args": args, "env": {"NOCOBASE_MCP_NB_BIN": nb_bin}, } } } def _venv_python_path(root: Path, *, os_name: str | None = None) -> Path: if (os_name or os.name) == "nt": return root / ".venv" / "Scripts" / "python.exe" return root / ".venv" / "bin" / "python"