#!/usr/bin/env python3 # /// script # requires-python = ">=3.11" # dependencies = [ # "pyqpanda3", # "commentjson>=0.9.0", # "tomli-w>=1.2.0", # ] # /// """ prepare_qpanda3_ai_kits.py — Prepare QPanda3 dev environment for AI coding tools """ import atexit import json as stdjson import os import re import shutil import subprocess import sys import tarfile import tempfile import time import urllib.error import urllib.parse import urllib.request import zipfile import commentjson import tomli_w import tomllib RED = "\033[0;31m" GREEN = "\033[0;32m" YELLOW = "\033[1;33m" BLUE = "\033[0;34m" NC = "\033[0m" SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) TMP_ZIPS: list[str] = [] TMP_TGZ = None TMP_DIR = None def cleanup(): for p in TMP_ZIPS: if os.path.isfile(p): os.unlink(p) if TMP_TGZ and os.path.isfile(TMP_TGZ): os.unlink(TMP_TGZ) if TMP_DIR and os.path.isdir(TMP_DIR): shutil.rmtree(TMP_DIR, ignore_errors=True) atexit.register(cleanup) def log_info(msg): print(f"{BLUE}[INFO]{NC} {msg}") def log_ok(msg): print(f"{GREEN}[OK]{NC} {msg}") def log_warn(msg): print(f"{YELLOW}[WARN]{NC} {msg}") def log_error(msg): print(f"{RED}[ERROR]{NC} {msg}") # ══════════════════════ Default Configuration ══════════════════════ DEFAULT_MCP_URL = "https://qcloud.originqc.com.cn/qpanda/runtime/mcp" # 技能库下载地址列表 —— 每个链接对应一个独立技能库 zip，脚本逐个下载并安装为 # 各 AI 工具 skills/ 下独立的 <技能名>/ 目录（技能名 = URL 文件名去扩展名）， # 彼此不合并；uninstall 按同一列表批量卸载。新增技能库时在此追加链接即可。 # 可用配置文件（-k config.json）的 skill_url 字段覆盖（字符串或字符串列表）。 DEFAULT_SKILL_URLS = [ "https://show.originqc.com.cn/prod/qcloudai/qpanda3-skill_MCP/qpanda3-programming.zip", "https://show.originqc.com.cn/prod/qcloudai/qpanda3-skill_MCP/pyvqnet-skills.zip", ] # 已知技能名（安装目录名 / 卸载目标），由默认列表推导。 KNOWN_SKILL_NAMES = [ os.path.splitext(os.path.basename(u))[0] for u in DEFAULT_SKILL_URLS ] # dsh-qpanda3 插件单文件（.tgz）的默认下载地址 —— `prepare dsh` 默认下载该文件后 # 再执行 `dsh plugin --profile web add <本地tgz>`，完成 skill 安装与 MCP 配置。 # 地址按插件版本固定（dsh-qpanda3-.tgz），发布新版本时由维护者同步更新； # 可用环境变量 DSH_QPANDA3_PLUGIN_URL 覆盖（测试可指向本地服务器 / file://）。 DEFAULT_DSH_PLUGIN_TGZ_URL = "https://show.originqc.com.cn/prod/qcloudai/qpanda3-skill_MCP/dsh-qpanda3-1.1.0-0.tgz" def skill_name_from_url(url: str) -> str: """Skill install dir name: URL path basename minus the .zip extension.""" return os.path.splitext(os.path.basename(urllib.parse.urlparse(url).path))[0] def normalize_skill_urls(skill_url) -> list[str]: """Normalize a config file's skill_url value to a list of zip URLs. Accepts: absent/None -> built-in default list; a single string (legacy single-skill configs) -> one-element list; a list of strings -> as-is. """ if skill_url is None: return list(DEFAULT_SKILL_URLS) if isinstance(skill_url, str): return [skill_url] if isinstance(skill_url, list) and all(isinstance(u, str) for u in skill_url): return list(skill_url) log_warn(f"Invalid skill_url value: {skill_url!r}, using the built-in skill list") return list(DEFAULT_SKILL_URLS) def resolve_k_param(k_value: str): """ Resolve -k parameter: - If a file with that name exists next to the script, parse its JSON for mcp config + skill_url - Otherwise treat as API key string, use global defaults Returns (mcp_cfg: dict | None, skill_urls: list[str]) """ filepath = os.path.join(SCRIPT_DIR, k_value) if os.path.isfile(filepath): try: with open(filepath, encoding="utf-8") as f: data = stdjson.load(f) mcp_cfg = data.get("mcp") skill_urls = normalize_skill_urls(data.get("skill_url")) log_info(f"Read config from: {filepath}") return mcp_cfg, skill_urls except Exception as e: log_warn(f"Failed to parse config ({e}), falling back to API key mode") return None, normalize_skill_urls(None) def generate_mcp_entry(apikey: str, mcp_url: str) -> dict: """Generate a generic MCP server entry (same format as config file mcp field)""" return {"qpanda3-runtime": {"url": mcp_url, "headers": {"X-Api-Key": apikey}}} def wrap_mcp_entry(tool: str, entry: dict) -> dict: """Wrap a generic MCP server entry into each tool's MCP JSON format""" if tool == "opencode": servers = {} for name, cfg in entry.items(): servers[name] = dict(cfg) servers[name].setdefault("type", "remote") return {"mcp": servers} elif tool == "claude": servers = {} for name, cfg in entry.items(): servers[name] = dict(cfg) servers[name].setdefault("type", "http") servers[name].setdefault("enabled", True) servers[name].setdefault("transport", "streamable-http") return {"mcpServers": servers} elif tool == "codex": return {"mcp_servers": entry} elif tool == "gemini": return {"mcpServers": entry} elif tool == "codebuddy": return {"mcpServers": entry} elif tool == "cline": servers = {} for name, cfg in entry.items(): servers[name] = { "disabled": False, "autoApprove": [], "transport": { "type": "streamableHttp", "url": cfg["url"], "headers": cfg.get("headers", {}), }, } return {"mcpServers": servers} elif tool == "codewhale": servers = {} for name, cfg in entry.items(): servers[name] = { "url": cfg["url"], "headers": cfg.get("headers", {}), "command": None, "args": [], "env": {}, "connect_timeout": None, "execute_timeout": None, "read_timeout": None, "disabled": False, "enabled": True, "required": False, "enabled_tools": [], "disabled_tools": [], } return {"servers": servers} elif tool == "zcode": servers = {} for name, cfg in entry.items(): servers[name] = { "type": "http", "url": cfg["url"], "headers": cfg.get("headers", {}), } return {"mcp": {"servers": servers}} return {} USER_CONFIG_DIRS = { "opencode": ["~/.config/opencode"], "claude": ["~/.claude", "~/.claude.json"], "codex": ["~/.codex"], "gemini": ["~/.gemini"], "codebuddy": ["~/.codebuddy"], "cline": ["~/.cline"], "codewhale": ["~/.codewhale"], "zcode": ["~/.zcode"], "dsh": ["~/.dsh"], } TOOL_COMMANDS = { "opencode": "opencode", "claude": "claude", "codex": "codex", "gemini": "gemini", "codebuddy": "codebuddy", "cline": "cline", "zcode": "zcode", "dsh": "dsh", } def check_tool_usable(tool: str, scope: str) -> bool: if tool == "cline" and scope == "project": log_error( "Cline only supports user-level configuration. Use 'prepare cline -k KEY' without -p." ) return False if tool == "codewhale" and scope == "project": log_error( "Codewhale only supports user-level configuration. Use 'prepare codewhale -k KEY' without -p." ) return False if tool == "zcode" and scope == "project": log_error( "ZCode only supports user-level configuration. Use 'prepare zcode -k KEY' without -p." ) return False if tool == "dsh" and scope == "project": log_error( "dsh only supports user-level configuration (the web profile lives under ~/.dsh). " "Use 'prepare dsh -k KEY' without -p." ) return False if scope == "project": cmd = TOOL_COMMANDS.get(tool) if cmd and shutil.which(cmd): return True log_warn( f"{SUPPORTED_TOOLS[tool]['name']} command not found in PATH ({cmd}), skipped" ) return False paths = USER_CONFIG_DIRS.get(tool, []) for p in paths: if os.path.exists(os.path.expanduser(p)): return True log_warn( f"{SUPPORTED_TOOLS[tool]['name']} not detected ({', '.join(USER_CONFIG_DIRS.get(tool, []))}), skipped" ) return False def detect_installed_tools(scope: str = "user") -> list[str]: return [t for t in SUPPORTED_TOOLS if check_tool_usable(t, scope)] SUPPORTED_TOOLS = { "opencode": {"name": "OpenCode"}, "claude": {"name": "Claude Code"}, "codex": {"name": "Codex"}, "gemini": {"name": "Gemini CLI"}, "codebuddy": {"name": "CodeBuddy"}, "cline": {"name": "Cline"}, "codewhale": {"name": "Codewhale"}, "zcode": {"name": "ZCode Agent"}, "dsh": {"name": "DeepSeek Harness"}, } # ══════════════════════ Step 1: pyqpanda3 ══════════════════════ def run_step1(): log_info("Step 1: Verify pyqpanda3") try: import pyqpanda3 log_ok("pyqpanda3 ready") except ImportError: log_error("pyqpanda3 unavailable") # ══════════════════════ Shared Helper Functions ══════════════════════ def shared_download_zip(url): """Download one skill zip; return (skill_name, local_path). Temp files are registered in TMP_ZIPS and removed by the atexit cleanup. """ tmp = tempfile.NamedTemporaryFile(prefix="_skill.", suffix=".zip", delete=False) path = tmp.name tmp.close() TMP_ZIPS.append(path) log_info(f"Downloading: {url}") try: urllib.request.urlretrieve(url, path) except Exception as e: log_error(f"Download failed: {url} ({e})") sys.exit(1) log_ok("Download complete") shared_validate_zip(path) return skill_name_from_url(url), path def shared_install_skills(skills, base): """Install every (name, zip) as its own independent // directory. Skills never merge: each zip keeps its own install dir, so a tool loads them as separate skills. Returns the installed [(name, target_dir)] list. """ installed = [] for name, zip_path in skills: target = os.path.join(base, name) shared_extract_and_install(zip_path, target, f"{target}") installed.append((name, target)) return installed def shared_skill_banner(installed, scope_label, restart_hint): """Verify SKILL.md in every installed skill dir, then print the banner.""" ok = True for name, target in installed: if os.path.isfile(os.path.join(target, "SKILL.md")): continue log_error(f"Skill {name} verification failed: no SKILL.md under {target}") ok = False if not ok: sys.exit(1) log_ok(f"{len(installed)} skill(s) installed: {', '.join(n for n, _ in installed)}") print("┌────────────────────────────────────────") for name, target in installed: print(f"│ Skill: {name}") print(f"│ {scope_label}: {target}{os.sep}") print(f"│ Tip: {restart_hint}") print("└────────────────────────────────────────") def shared_validate_zip(zp): try: with zipfile.ZipFile(zp) as z: r = z.testzip() if r: log_error(f"Corrupt ZIP: {r}") sys.exit(1) except Exception as e: log_error(f"Invalid ZIP: {e}") sys.exit(1) s = os.path.getsize(zp) log_ok(f"Valid ZIP ({s // 1024}K)" if s >= 1024 else f"Valid ZIP ({s}B)") def shared_extract_and_install(zip_path, target, desc): global TMP_DIR os.makedirs(target, exist_ok=True) td = tempfile.mkdtemp(prefix="_skill_extract.") TMP_DIR = td try: with zipfile.ZipFile(zip_path) as z: z.extractall(td) _items = os.listdir(td) if len(_items) == 1 and os.path.isdir(os.path.join(td, _items[0])): _inner = os.path.join(td, _items[0]) for _item in os.listdir(_inner): shutil.move(os.path.join(_inner, _item), os.path.join(td, _item)) os.rmdir(_inner) if shutil.which("rsync"): subprocess.run( [ "rsync", "-a", "--exclude=build_index.py", "--exclude=verify_sync.py", f"{td}/", f"{target}/", ], check=True, ) else: for item in os.listdir(td): s = os.path.join(td, item) d = os.path.join(target, item) if item in ("build_index.py", "verify_sync.py"): continue if os.path.isdir(s): shutil.copytree( s, d, ignore=shutil.ignore_patterns( "build_index.py", "verify_sync.py" ), dirs_exist_ok=True, ) else: shutil.copy2(s, d) log_ok(f"Installed: {desc}") finally: if os.path.isdir(td): shutil.rmtree(td, ignore_errors=True) TMP_DIR = None def shared_test_connection(url, headers=None, skip_env_vars=False): log_info(f"Testing connection: {url}") req = urllib.request.Request( url, stdjson.dumps({"jsonrpc": "2.0", "id": 1, "method": "ping"}).encode("utf-8"), method="POST", ) req.add_header("Content-Type", "application/json") req.add_header("Accept", "application/json") if headers: for k, v in headers.items(): if skip_env_vars and "${" in str(v): continue req.add_header(k, v) try: with urllib.request.urlopen(req, timeout=30) as resp: body = resp.read().decode("utf-8") if resp.status == 200: try: d = stdjson.loads(body) r = d.get("result", "") result = ( r if r else ( "Reachable (Method not found, expected)" if d.get("error", {}).get("code") == -32601 else "Connected" ) ) except: result = "Response parsed" log_ok(f"Connection successful! {result}") else: log_error(f"Connection failed (HTTP {resp.status})") except urllib.error.HTTPError as e: log_error(f"Connection failed (HTTP {e.code})") except urllib.error.URLError as e: log_error(f"Connection failed: {e.reason}") # ══════════════════════ Shared Uninstall Helpers ══════════════════════ def uninstall_json_entry(fp, mcp_key, server_name): if not os.path.isfile(fp): return None try: with open(fp, "r", encoding="utf-8") as f: ct = f.read() if not ct.strip(): return None try: data = commentjson.loads(ct) except Exception: try: data = stdjson.loads(ct) except: return None se = data.get(mcp_key, {}) if server_name not in se: return "NOT_FOUND" del se[server_name] data[mcp_key] = se with open(fp, "w", encoding="utf-8", newline="\n") as f: stdjson.dump(data, f, indent=2, ensure_ascii=False) f.write("\n") return "REMOVED" except Exception as e: return f"ERROR: {e}" def uninstall_mcp_arrays(fp, server_name, array_keys): if not os.path.isfile(fp): return None try: with open(fp, "r", encoding="utf-8") as f: ct = f.read() if not ct.strip(): return None try: data = commentjson.loads(ct) except: try: data = stdjson.loads(ct) except: return None changed = False for k in array_keys: arr = data.get(k) if isinstance(arr, list) and server_name in arr: arr.remove(server_name) changed = True if changed: with open(fp, "w", encoding="utf-8", newline="\n") as f: stdjson.dump(data, f, indent=2, ensure_ascii=False) f.write("\n") return "REMOVED" return "NOT_FOUND" except Exception as e: return f"ERROR: {e}" def uninstall_skill_dir(path): if os.path.isdir(path): shutil.rmtree(path) return True return False def uninstall_known_skills(base, label): """Batch-remove every known skill dir under a tool's skills base. Each skill keeps its own independent directory; the known names are derived from the built-in DEFAULT_SKILL_URLS list, so uninstall needs no install-time bookkeeping. """ for name in KNOWN_SKILL_NAMES: t = os.path.join(base, name) if uninstall_skill_dir(t): log_ok(f"Removed: {label}/{name}") else: log_warn(f"{label}/{name} not found") def uninstall_print_result(label, r): if r is None: log_warn(f"{label} not found, skipped") elif r == "NOT_FOUND": log_warn(f"{label} not found") elif r == "REMOVED": log_ok(f"Removed from {label}") def uninstall_toml_entry(fp, mcp_key, server_name): if not os.path.isfile(fp): return None try: with open(fp, "rb") as f: data = tomllib.load(f) servers = data.get(mcp_key, {}) if server_name not in servers: return "NOT_FOUND" del servers[server_name] with open(fp, "wb") as f: tomli_w.dump(data, f) return "REMOVED" except Exception as e: return f"ERROR: {e}" # ══════════════════════ Tool Modules ══════════════════════ # ---------- OpenCode ---------- def opencode_mcp_merge(): os.makedirs(OC_CONFIG_DIR, exist_ok=True) if os.path.isfile(OC_CONFIG_FILE): with open(OC_CONFIG_FILE, encoding="utf-8") as f: ct = f.read() try: ex = commentjson.loads(ct) if ct.strip() else {} except: ex = stdjson.loads(ct) if ct.strip() else {} nd = stdjson.loads(OC_NORMALIZED_JSON) se = ex.get("mcp") or ex.get("mcpServers", {}) for n, e in nd["mcp"].items(): se[n] = e ex["mcp"] = se ex.pop("mcpServers", None) merged = stdjson.dumps(ex, indent=2, ensure_ascii=False) else: merged = OC_NORMALIZED_JSON with open(OC_CONFIG_FILE, "w", encoding="utf-8", newline="\n") as f: f.write(merged) log_ok(f"Written: {OC_CONFIG_FILE}") def run_prepare_opencode(mcp_cfg: dict, skills: list, scope: str): global OC_CONFIG_DIR, OC_CONFIG_FILE, OC_NORMALIZED_JSON log_info("Configuring OpenCode MCP") if scope == "project": OC_CONFIG_DIR = os.path.join(SCRIPT_DIR, ".opencode") else: OC_CONFIG_DIR = os.path.expanduser("~/.config/opencode") OC_CONFIG_FILE = os.path.join(OC_CONFIG_DIR, "opencode.jsonc") OC_NORMALIZED_JSON = stdjson.dumps( wrap_mcp_entry("opencode", mcp_cfg), indent=2, ensure_ascii=False ) opencode_mcp_merge() log_info("Installing QPanda3 Skills") if scope == "project": base = os.path.join(SCRIPT_DIR, ".opencode", "skills") label = "project" else: base = os.path.expanduser("~/.config/opencode/skills") label = "global" installed = shared_install_skills(skills, base) shared_skill_banner(installed, label, "restart OpenCode") def run_uninstall_opencode(): sn = "qpanda3-runtime" log_info("Uninstalling OpenCode MCP configuration") for p, lb in [ (os.path.expanduser("~/.config/opencode/opencode.jsonc"), "OpenCode global"), (os.path.join(SCRIPT_DIR, ".opencode", "opencode.jsonc"), "OpenCode project"), ]: uninstall_print_result(lb, uninstall_json_entry(p, "mcp", sn)) print() log_info("Uninstalling Skills") for b, lb in [ (os.path.expanduser("~/.config/opencode/skills"), "global"), (os.path.join(SCRIPT_DIR, ".opencode", "skills"), "project"), ]: uninstall_known_skills(b, lb) print() log_ok("Uninstall complete! Restart OpenCode.") # ---------- Claude Code ---------- def claude_mcp_merge(): os.makedirs(CL_CONFIG_DIR, exist_ok=True) if os.path.isfile(CL_CONFIG_FILE): with open(CL_CONFIG_FILE, encoding="utf-8") as f: ct = f.read() ex = stdjson.loads(ct) if ct.strip() else {} nd = stdjson.loads(CL_NORMALIZED_JSON) if "mcpServers" not in ex: ex["mcpServers"] = {} for n, e in nd["mcpServers"].items(): ex["mcpServers"][n] = e merged = stdjson.dumps(ex, indent=2, ensure_ascii=False) else: merged = CL_NORMALIZED_JSON with open(CL_CONFIG_FILE, "w", encoding="utf-8", newline="\n") as f: f.write(merged) log_ok(f"Written: {CL_CONFIG_FILE}") def run_prepare_claude(mcp_cfg: dict, skills: list, scope: str): global CL_CONFIG_DIR, CL_CONFIG_FILE, CL_NORMALIZED_JSON log_info("Configuring Claude Code MCP") if scope == "project": CL_CONFIG_DIR = SCRIPT_DIR CL_CONFIG_FILE = os.path.join(CL_CONFIG_DIR, ".mcp.json") else: CL_CONFIG_DIR = os.path.expanduser("~") CL_CONFIG_FILE = os.path.join(CL_CONFIG_DIR, ".claude.json") CL_NORMALIZED_JSON = stdjson.dumps( wrap_mcp_entry("claude", mcp_cfg), indent=2, ensure_ascii=False ) claude_mcp_merge() log_info("Installing QPanda3 Skills") if scope == "project": base = os.path.join(SCRIPT_DIR, ".claude", "skills") label = "project" else: base = os.path.expanduser("~/.claude/skills") label = "global" installed = shared_install_skills(skills, base) shared_skill_banner(installed, label, "restart Claude Code") def run_uninstall_claude(): sn = "qpanda3-runtime" log_info("Uninstalling Claude Code MCP configuration") uninstall_print_result( "Claude global", uninstall_json_entry(os.path.expanduser("~/.claude.json"), "mcpServers", sn), ) for base in (SCRIPT_DIR,): p = os.path.join(base, ".mcp.json") uninstall_print_result( "Claude project", uninstall_json_entry(p, "mcpServers", sn) ) for base in (SCRIPT_DIR,): ls = os.path.join(base, ".claude", "settings.local.json") r = uninstall_mcp_arrays( ls, sn, ["enabledMcpjsonServers", "disabledMcpjsonServers"] ) if r == "REMOVED": log_ok(f"Removed {sn} from settings.local.json") print() log_info("Uninstalling Skills") for b, lb in [ (os.path.expanduser("~/.claude/skills"), "global"), (os.path.join(SCRIPT_DIR, ".claude", "skills"), "project"), ]: uninstall_known_skills(b, lb) print() log_ok("Uninstall complete! Restart Claude Code.") # ---------- Codex ---------- def codex_mcp_merge(): os.makedirs(CDX_CONFIG_DIR, exist_ok=True) nd = stdjson.loads(CDX_NORMALIZED_JSON) servers = nd.get("mcp_servers", {}) existing = {} if os.path.isfile(CDX_CONFIG_FILE): try: with open(CDX_CONFIG_FILE, "rb") as f: existing = tomllib.load(f) except FileNotFoundError: pass ms = existing.get("mcp_servers", {}) for n, e in servers.items(): ms[n] = e existing["mcp_servers"] = ms for sn in existing.get("mcp_servers", {}): cfg = existing["mcp_servers"][sn] if "headers" in cfg: cfg["http_headers"] = cfg.pop("headers") for skip in ("type", "transport"): cfg.pop(skip, None) with open(CDX_CONFIG_FILE, "wb") as f: tomli_w.dump(existing, f) log_ok(f"Written: {CDX_CONFIG_FILE}") def run_prepare_codex(mcp_cfg: dict, skills: list, scope: str): global CDX_CONFIG_DIR, CDX_CONFIG_FILE, CDX_NORMALIZED_JSON log_info("Configuring Codex MCP") if scope == "project": CDX_CONFIG_DIR = os.path.join(SCRIPT_DIR, ".codex") else: CDX_CONFIG_DIR = os.path.expanduser("~/.codex") CDX_CONFIG_FILE = os.path.join(CDX_CONFIG_DIR, "config.toml") CDX_NORMALIZED_JSON = stdjson.dumps( wrap_mcp_entry("codex", mcp_cfg), indent=2, ensure_ascii=False ) codex_mcp_merge() log_info("Installing QPanda3 Skills") if scope == "project": base = os.path.join(SCRIPT_DIR, ".codex", "skills") label = "project" else: base = os.path.expanduser("~/.codex/skills") label = "global" installed = shared_install_skills(skills, base) shared_skill_banner(installed, label, "restart Codex") def run_uninstall_codex(): sn = "qpanda3-runtime" log_info("Uninstalling Codex MCP configuration") for p, lb in [ (os.path.expanduser("~/.codex/config.toml"), "Codex global"), (os.path.join(SCRIPT_DIR, ".codex", "config.toml"), "Codex project"), ]: uninstall_print_result(lb, uninstall_toml_entry(p, "mcp_servers", sn)) print() log_info("Uninstalling Skills") for b, lb in [ (os.path.expanduser("~/.codex/skills"), "global"), (os.path.join(SCRIPT_DIR, ".codex", "skills"), "project"), ]: uninstall_known_skills(b, lb) print() log_ok("Uninstall complete! Restart Codex.") # ---------- Gemini ---------- def gemini_mcp_merge(): os.makedirs(GM_CONFIG_DIR, exist_ok=True) if os.path.isfile(GM_CONFIG_FILE): with open(GM_CONFIG_FILE, encoding="utf-8") as f: ct = f.read() ex = stdjson.loads(ct) if ct.strip() else {} nd = stdjson.loads(GM_NORMALIZED_JSON) if "mcpServers" not in ex: ex["mcpServers"] = {} for n, e in nd["mcpServers"].items(): ex["mcpServers"][n] = e merged = stdjson.dumps(ex, indent=2, ensure_ascii=False) else: merged = GM_NORMALIZED_JSON with open(GM_CONFIG_FILE, "w", encoding="utf-8", newline="\n") as f: f.write(merged) log_ok(f"Written: {GM_CONFIG_FILE}") def run_prepare_gemini(mcp_cfg: dict, skills: list, scope: str): global GM_CONFIG_DIR, GM_CONFIG_FILE, GM_NORMALIZED_JSON log_info("Configuring Gemini CLI MCP") if scope == "project": GM_CONFIG_DIR = os.path.join(SCRIPT_DIR, ".gemini") else: GM_CONFIG_DIR = os.path.expanduser("~/.gemini") GM_CONFIG_FILE = os.path.join(GM_CONFIG_DIR, "settings.json") GM_NORMALIZED_JSON = stdjson.dumps( wrap_mcp_entry("gemini", mcp_cfg), indent=2, ensure_ascii=False ) gemini_mcp_merge() log_info("Installing QPanda3 Skills") if scope == "project": base = os.path.join(SCRIPT_DIR, ".gemini", "skills") label = "project" else: base = os.path.expanduser("~/.gemini/skills") label = "user" installed = shared_install_skills(skills, base) shared_skill_banner(installed, label, "restart Gemini CLI") def run_uninstall_gemini(): sn = "qpanda3-runtime" log_info("Uninstalling Gemini MCP configuration") for p, lb in [ (os.path.expanduser("~/.gemini/settings.json"), "Gemini user"), (os.path.join(SCRIPT_DIR, ".gemini", "settings.json"), "Gemini project"), ]: uninstall_print_result(lb, uninstall_json_entry(p, "mcpServers", sn)) print() log_info("Uninstalling Skills") for b, lb in [ (os.path.expanduser("~/.gemini/skills"), "user"), (os.path.join(SCRIPT_DIR, ".gemini", "skills"), "project"), ]: uninstall_known_skills(b, lb) print() log_ok("Uninstall complete! Restart Gemini CLI.") # ---------- CodeBuddy ---------- def cdbd_find_user(): for f in ["~/.codebuddy/.mcp.json", "~/.codebuddy/mcp.json", "~/.codebuddy.json"]: p = os.path.expanduser(f) if os.path.isfile(p): return p return None def cdbd_find_project(): for f in [ os.path.join(SCRIPT_DIR, ".mcp.json"), os.path.join(SCRIPT_DIR, "mcp.json"), ]: if os.path.isfile(f): return f return None def cdbd_mcp_merge(): os.makedirs(CB_CONFIG_DIR, exist_ok=True) if os.path.isfile(CB_CONFIG_FILE): with open(CB_CONFIG_FILE, encoding="utf-8") as f: ct = f.read() try: ex = commentjson.loads(ct) if ct.strip() else {} except: ex = stdjson.loads(ct) if ct.strip() else {} nd = stdjson.loads(CB_NORMALIZED_JSON) if "mcpServers" not in ex: ex["mcpServers"] = {} for n, e in nd["mcpServers"].items(): ex["mcpServers"][n] = e merged = stdjson.dumps(ex, indent=2, ensure_ascii=False) else: merged = CB_NORMALIZED_JSON with open(CB_CONFIG_FILE, "w", encoding="utf-8", newline="\n") as f: f.write(merged) log_ok(f"Written: {CB_CONFIG_FILE}") def run_prepare_codebuddy(mcp_cfg: dict, skills: list, scope: str): global CB_CONFIG_DIR, CB_CONFIG_FILE, CB_NORMALIZED_JSON log_info("Configuring CodeBuddy MCP") if scope == "project": f = cdbd_find_project() CB_CONFIG_FILE = f if f else os.path.join(SCRIPT_DIR, ".mcp.json") else: f = cdbd_find_user() CB_CONFIG_FILE = f if f else os.path.expanduser("~/.codebuddy/.mcp.json") CB_CONFIG_DIR = os.path.dirname(CB_CONFIG_FILE) CB_NORMALIZED_JSON = stdjson.dumps( wrap_mcp_entry("codebuddy", mcp_cfg), indent=2, ensure_ascii=False ) cdbd_mcp_merge() log_info("Installing QPanda3 Skills") if scope == "project": base = os.path.join(SCRIPT_DIR, ".codebuddy", "skills") label = "project" else: base = os.path.expanduser("~/.codebuddy/skills") label = "user" installed = shared_install_skills(skills, base) shared_skill_banner(installed, label, "restart CodeBuddy") def run_uninstall_codebuddy(): sn = "qpanda3-runtime" log_info("Uninstalling CodeBuddy MCP configuration") for p, lb in [ (os.path.expanduser("~/.codebuddy/.mcp.json"), "CodeBuddy"), (os.path.expanduser("~/.codebuddy/mcp.json"), "CodeBuddy"), (os.path.expanduser("~/.codebuddy.json"), "CodeBuddy"), (os.path.join(SCRIPT_DIR, ".mcp.json"), "CodeBuddy project"), (os.path.join(SCRIPT_DIR, "mcp.json"), "CodeBuddy project"), ]: r = uninstall_json_entry(p, "mcpServers", sn) if r == "REMOVED": log_ok(f"Removed {sn} from {lb}") print() log_info("Uninstalling Skills") for b, lb in [ (os.path.expanduser("~/.codebuddy/skills"), "user"), (os.path.join(SCRIPT_DIR, ".codebuddy", "skills"), "project"), ]: uninstall_known_skills(b, lb) print() log_ok("Uninstall complete! Restart CodeBuddy.") # ---------- Cline ---------- def cline_mcp_merge(): os.makedirs(CLN_CONFIG_DIR, exist_ok=True) if os.path.isfile(CLN_CONFIG_FILE): with open(CLN_CONFIG_FILE, encoding="utf-8") as f: ct = f.read() ex = stdjson.loads(ct) if ct.strip() else {} nd = stdjson.loads(CLN_NORMALIZED_JSON) if "mcpServers" not in ex: ex["mcpServers"] = {} for n, e in nd["mcpServers"].items(): ex["mcpServers"][n] = e merged = stdjson.dumps(ex, indent=2, ensure_ascii=False) else: merged = CLN_NORMALIZED_JSON with open(CLN_CONFIG_FILE, "w", encoding="utf-8", newline="\n") as f: f.write(merged) log_ok(f"Written: {CLN_CONFIG_FILE}") def run_prepare_cline(mcp_cfg: dict, skills: list, scope: str): global CLN_CONFIG_DIR, CLN_CONFIG_FILE, CLN_NORMALIZED_JSON log_info("Configuring Cline MCP") CLN_CONFIG_DIR = os.path.expanduser("~/.cline/data/settings") CLN_CONFIG_FILE = os.path.join(CLN_CONFIG_DIR, "cline_mcp_settings.json") CLN_NORMALIZED_JSON = stdjson.dumps( wrap_mcp_entry("cline", mcp_cfg), indent=2, ensure_ascii=False ) cline_mcp_merge() log_info("Installing QPanda3 Skills") base = os.path.expanduser("~/.cline/skills") installed = shared_install_skills(skills, base) shared_skill_banner(installed, "user", "restart Cline") def run_uninstall_cline(): sn = "qpanda3-runtime" log_info("Uninstalling Cline MCP configuration") uninstall_print_result( "Cline", uninstall_json_entry( os.path.expanduser("~/.cline/data/settings/cline_mcp_settings.json"), "mcpServers", sn, ), ) print() log_info("Uninstalling Skills") uninstall_known_skills(os.path.expanduser("~/.cline/skills"), "user") print() log_ok("Uninstall complete! Restart Cline.") # ---------- Codewhale ---------- def codewhale_mcp_merge(): os.makedirs(CW_CONFIG_DIR, exist_ok=True) nd = stdjson.loads(CW_NORMALIZED_JSON) servers = nd.get("servers", {}) existing = {} if os.path.isfile(CW_CONFIG_FILE): with open(CW_CONFIG_FILE, encoding="utf-8") as f: ct = f.read() if ct.strip(): existing = stdjson.loads(ct) ms = existing.get("servers", {}) for n, e in servers.items(): ms[n] = e existing["servers"] = ms with open(CW_CONFIG_FILE, "w", encoding="utf-8", newline="\n") as f: stdjson.dump(existing, f, indent=2, ensure_ascii=False) f.write("\n") log_ok(f"Written: {CW_CONFIG_FILE}") def codewhale_ensure_priority(): """Ensure '预加载技能库和mcp工具列表' exists in constitution.json priorities.""" cw_dir = os.path.expanduser("~/.codewhale") constitution_file = os.path.join(cw_dir, "constitution.json") if not os.path.isfile(constitution_file): log_error( f"constitution.json not found at {constitution_file}, " "required for Codewhale configuration" ) sys.exit(1) try: with open(constitution_file, encoding="utf-8") as f: data = stdjson.load(f) priorities = data.get("priorities", []) target = "预加载技能库和mcp工具列表" if target in priorities: log_ok("Priority '预加载技能库和mcp工具列表' already exists") return priorities.append(target) data["priorities"] = priorities with open(constitution_file, "w", encoding="utf-8", newline="\n") as f: stdjson.dump(data, f, indent=2, ensure_ascii=False) f.write("\n") log_ok("Added priority '预加载技能库和mcp工具列表' to constitution.json") except Exception as e: log_error(f"Failed to update constitution.json: {e}") sys.exit(1) def run_prepare_codewhale(mcp_cfg: dict, skills: list, scope: str): global CW_CONFIG_DIR, CW_CONFIG_FILE, CW_NORMALIZED_JSON log_info("Configuring Codewhale MCP") CW_CONFIG_DIR = os.path.expanduser("~/.codewhale") CW_CONFIG_FILE = os.path.join(CW_CONFIG_DIR, "mcp.json") CW_NORMALIZED_JSON = stdjson.dumps( wrap_mcp_entry("codewhale", mcp_cfg), indent=2, ensure_ascii=False ) codewhale_mcp_merge() codewhale_ensure_priority() log_info("Installing QPanda3 Skills") base = os.path.expanduser("~/.codewhale/skills") installed = shared_install_skills(skills, base) shared_skill_banner(installed, "user", "restart Codewhale") def run_uninstall_codewhale(): sn = "qpanda3-runtime" log_info("Uninstalling Codewhale MCP configuration") uninstall_print_result( "Codewhale", uninstall_json_entry( os.path.expanduser("~/.codewhale/mcp.json"), "servers", sn, ), ) print() log_info("Uninstalling Skills") uninstall_known_skills(os.path.expanduser("~/.codewhale/skills"), "user") print() log_ok("Uninstall complete! Restart Codewhale.") # ---------- ZCode Agent ---------- def zcode_mcp_merge(): os.makedirs(ZC_CONFIG_DIR, exist_ok=True) existing = {} if os.path.isfile(ZC_CONFIG_FILE): with open(ZC_CONFIG_FILE, encoding="utf-8") as f: ct = f.read() if ct.strip(): existing = stdjson.loads(ct) nd = stdjson.loads(ZC_NORMALIZED_JSON) mcp_cfg = nd.get("mcp", {}) mcp = existing.get("mcp", {}) servers = mcp.get("servers", {}) for n, e in mcp_cfg.get("servers", {}).items(): servers[n] = e mcp["servers"] = servers existing["mcp"] = mcp with open(ZC_CONFIG_FILE, "w", encoding="utf-8", newline="\n") as f: stdjson.dump(existing, f, indent=2, ensure_ascii=False) f.write("\n") log_ok(f"Written: {ZC_CONFIG_FILE}") def run_prepare_zcode(mcp_cfg: dict, skills: list, scope: str): global ZC_CONFIG_DIR, ZC_CONFIG_FILE, ZC_NORMALIZED_JSON log_info("Configuring ZCode Agent MCP") ZC_CONFIG_DIR = os.path.expanduser("~/.zcode/cli") ZC_CONFIG_FILE = os.path.join(ZC_CONFIG_DIR, "config.json") ZC_NORMALIZED_JSON = stdjson.dumps( wrap_mcp_entry("zcode", mcp_cfg), indent=2, ensure_ascii=False ) zcode_mcp_merge() log_info("Installing QPanda3 Skills") base = os.path.expanduser("~/.zcode/skills") installed = shared_install_skills(skills, base) shared_skill_banner(installed, "user", "restart ZCode Agent") def run_uninstall_zcode(): sn = "qpanda3-runtime" log_info("Uninstalling ZCode Agent MCP configuration") r = uninstall_nested_json_entry( os.path.expanduser("~/.zcode/cli/config.json"), ["mcp", "servers"], sn ) uninstall_print_result("ZCode global", r) print() log_info("Uninstalling Skills") uninstall_known_skills(os.path.expanduser("~/.zcode/skills"), "user") print() log_ok("Uninstall complete! Restart ZCode Agent.") # ---------- DeepSeek Harness (dsh) ---------- def dsh_key_file_path() -> str: """The user-level key file dsh-qpanda3 reads: $DSH_HOME/qpanda3.env (default ~/.dsh/qpanda3.env).""" home = os.environ.get("DSH_HOME") or os.path.join(os.path.expanduser("~"), ".dsh") return os.path.join(home, "qpanda3.env") def dsh_qpanda3_dir() -> str: """Fixed QPanda3 asset dir inside the dsh home: $DSH_HOME/qpanda3 (default ~/.dsh/qpanda3). The plugin tarball is cached here, per version, for the installed lifetime of the plugin: pnpm records the dependency spec as the absolute `file:` path it was given, so the file must stay resolvable for later pnpm operations (uninstall / update / reinstall). On uninstall the cached tarballs are deleted after the plugin is removed, restoring the pre-install state. The path is computed from the DSH_HOME env var (with ~/.dsh fallback) instead of a hardcoded absolute path, keeping it stable per machine and free of repository-location coupling. """ home = os.environ.get("DSH_HOME") or os.path.join(os.path.expanduser("~"), ".dsh") return os.path.join(home, "qpanda3") def dsh_plugin_cache_path(version: str) -> str: """The fixed, versioned cache path of the plugin tarball under the dsh home.""" return os.path.join(dsh_qpanda3_dir(), f"dsh-qpanda3-{version}.tgz") # dsh-qpanda3 plugin source. # # 插件版本 X.Y.Z-n 的第 4 位 n 为纯打包修订号, 与技能库版本无关 # (技能库自带 frontmatter version:, 由 generate_skill/scripts_sh/step5_package.py # 打包生成不含技能内容的单文件 plugins/dsh-qpanda3-.tgz)。 # # `prepare dsh` 的安装流程: # 1. 与其他工具共用同一批已下载的技能库 zip (skills 列表); # 2. 解析插件来源并统一物化到 dsh 安装目录内的固定缓存 # $DSH_HOME/qpanda3/dsh-qpanda3-.tgz —— URL 下载 (默认, # env DSH_QPANDA3_PLUGIN_URL 可覆盖)、env DSH_QPANDA3_PLUGIN_DIR 指向的 # 本地 .tgz (复制进缓存, 不改动用户文件)、本地插件目录 (打成 tgz 进缓存, # 不改动源目录; URL 失败时的离线回退也走这里); # 3. 把已下载的技能库 zip 注入缓存插件的 skills/ 目录 —— 逐个解包为 # package/skills/<技能名>/, 同名技能先删旧树再写入 (替换语义, 无残留), # 插件的 cordis.patch.yml 将整个 skills/ 注册为技能根, 注入即生效, # 无需改清单; # 4. `dsh plugin --profile web add <缓存路径>` 一次完成 MCP + 全部技能安装。 # 缓存文件在插件安装期间保留 —— pnpm 会把依赖记录为该 `file:` 绝对路径, # 卸载/更新/重装时需能重新解析它, 提前删除会导致不一致 (ENOENT); 卸载 # (uninstall dsh) 时在插件移除成功后删除缓存, 恢复安装前状态。 # 路径由 DSH_HOME 环境变量 (回退 ~/.dsh) 计算, 不写死机器绝对路径。 DSH_QPANDA3_PLUGIN_LOCAL_DIR = os.path.abspath( os.path.join(SCRIPT_DIR, "..", "plugins", "dsh-qpanda3") ) def dsh_plugin_source() -> tuple[str, str]: """Return (kind, source) of the dsh-qpanda3 plugin. Priority: 1. env DSH_QPANDA3_PLUGIN_DIR -> ('file', ) or ('dir', ) 2. default -> ('url', DSH_QPANDA3_PLUGIN_URL or DEFAULT_DSH_PLUGIN_TGZ_URL); the caller materializes it into the fixed per-version cache under the dsh home (kept, so the recorded `file:` spec stays resolvable) and runs `dsh plugin --profile web add `; on download failure the caller falls back to ('dir', DSH_QPANDA3_PLUGIN_LOCAL_DIR). """ override = os.environ.get("DSH_QPANDA3_PLUGIN_DIR") if override: p = os.path.normpath(override) if p.lower().endswith((".tgz", ".tar.gz")): return "file", p return "dir", p return "url", os.environ.get("DSH_QPANDA3_PLUGIN_URL") or DEFAULT_DSH_PLUGIN_TGZ_URL def dsh_plugin_dir_version(dirpath: str) -> str: """Read the plugin version X.Y.Z-n from a local plugin dir's package.json ("" when unknown).""" pj = os.path.join(dirpath, "package.json") if os.path.isfile(pj): try: with open(pj, encoding="utf-8") as f: return stdjson.load(f).get("version", "") or "" except Exception: return "" return "" def dsh_pack_dir_to_tgz(dirpath: str, out_path: str) -> None: """Pack a local plugin directory into an npm-style tgz (package/ prefix). Mirrors the tar fallback of generate_skill/scripts_sh/step5_package.py so the layout matches what `npm pack` produces for the same directory. """ td = tempfile.mkdtemp(prefix="_dsh_pack.") try: pkg = os.path.join(td, "package") os.makedirs(pkg) for fname in ("package.json", "cordis.patch.yml", "README.md", "LICENSE"): src = os.path.join(dirpath, fname) if os.path.isfile(src): shutil.copy2(src, os.path.join(pkg, fname)) # tgz 不携带技能内容：仅保留空的 skills/ 目录占位，技能在安装时注入 os.makedirs(os.path.join(pkg, "skills"), exist_ok=True) with tarfile.open(out_path, "w:gz") as tf: tf.add(pkg, arcname="package") finally: shutil.rmtree(td, ignore_errors=True) def dsh_add_skill_members(dst_tar: tarfile.TarFile, name: str, zip_path: str) -> None: """Append one skill zip to a plugin tarball as package/skills//. A single top-level directory inside the zip is unwrapped first (same rule as shared_extract_and_install), so the skill lands directly in its dir. """ td = tempfile.mkdtemp(prefix="_dsh_skill.") try: with zipfile.ZipFile(zip_path) as z: z.extractall(td) items = os.listdir(td) root = td if len(items) == 1 and os.path.isdir(os.path.join(td, items[0])): root = os.path.join(td, items[0]) mtime = int(time.time()) for cur, dirs, files in os.walk(root): dirs.sort() files.sort() rel = os.path.relpath(cur, root).replace(os.sep, "/") base = f"package/skills/{name}" if rel == "." else f"package/skills/{name}/{rel}" for d in dirs: info = tarfile.TarInfo(f"{base}/{d}") info.type = tarfile.DIRTYPE info.mode = 0o755 info.mtime = mtime dst_tar.addfile(info) for fn in files: fp = os.path.join(cur, fn) info = tarfile.TarInfo(f"{base}/{fn}") info.size = os.path.getsize(fp) info.mode = 0o644 info.mtime = mtime with open(fp, "rb") as f: dst_tar.addfile(info, f) finally: shutil.rmtree(td, ignore_errors=True) def zip_skill_version(zip_path: str) -> str: """Read a skill zip's own frontmatter version (X.Y.Z) from its SKILL.md; "" when unknown.""" try: with zipfile.ZipFile(zip_path) as z: for entry in z.namelist(): if entry.split("/")[-1] == "SKILL.md": with z.open(entry) as fh: head = fh.read(65536).decode("utf-8", errors="replace") m = re.search(r"(?m)^version:\s*(\d+\.\d+\.\d+)\s*$", head) if m: return m.group(1) except Exception: pass return "" def dsh_inject_skills_into_tgz(tgz_path: str, skills: list) -> None: """Rewrite the plugin tarball so package/skills// carries the skills. Replace semantics per injected name: existing package/skills// members are dropped first, then each downloaded skill zip is unpacked into package/skills//. Members outside the injected trees pass through unchanged. The rewrite is atomic (temp file + os.replace), keeping the cached tarball a valid npm package for pnpm resolution at all times. """ injected = {name for name, _ in skills} prefix = "package/skills/" fd, tmp_out = tempfile.mkstemp(prefix="_dsh_inject.", suffix=".tgz") os.close(fd) try: with tarfile.open(tgz_path, "r:gz") as src, tarfile.open(tmp_out, "w:gz") as dst: has_skills_dir = False for m in src: if m.name.startswith(prefix) and m.name[len(prefix):].split("/")[0] in injected: continue if m.name.rstrip("/") == "package/skills": has_skills_dir = True if m.isreg(): dst.addfile(m, src.extractfile(m)) else: dst.addfile(m) if not has_skills_dir: info = tarfile.TarInfo("package/skills") info.type = tarfile.DIRTYPE info.mode = 0o755 dst.addfile(info) for name, zip_path in skills: dsh_add_skill_members(dst, name, zip_path) os.replace(tmp_out, tgz_path) except BaseException: if os.path.isfile(tmp_out): os.unlink(tmp_out) raise def dsh_plugin_install_source(skills: list) -> tuple[str, str, str]: """Resolve the plugin source, materialize it into the fixed per-version cache under the dsh home, inject the downloaded skill zips into the cached tarball's skills/ directory, and return (kind, source, version). kind is always 'file': every source kind materializes into the cached tgz (kept, so the recorded `file:` dependency spec stays resolvable for later pnpm operations; uninstall removes it after plugin removal) — - 'url' (default): downloaded from DSH_QPANDA3_PLUGIN_URL or DEFAULT_DSH_PLUGIN_TGZ_URL; download failure falls back to packing the local plugin dir DSH_QPANDA3_PLUGIN_LOCAL_DIR (offline). - 'file' (env DSH_QPANDA3_PLUGIN_DIR -> .tgz): copied into the cache so the user's file is never mutated. - 'dir' (env DSH_QPANDA3_PLUGIN_DIR -> directory): packed into the cache so the source directory is never mutated. source is what `dsh plugin add` should consume (the injected cache tgz); version is the plugin version X.Y.Z-n shown in the success banner ("" / "local" when unknown for dir sources without a package.json). """ kind, source = dsh_plugin_source() if kind == "url": global TMP_TGZ name = os.path.basename(source) version = name.removeprefix("dsh-qpanda3-").removesuffix(".tgz") cached = dsh_plugin_cache_path(version) os.makedirs(dsh_qpanda3_dir(), exist_ok=True) tmp = tempfile.NamedTemporaryFile(prefix="_dsh_qpanda3.", suffix=".tgz", delete=False) TMP_TGZ = tmp.name tmp.close() log_info(f"Downloading plugin tarball: {source}") try: urllib.request.urlretrieve(source, TMP_TGZ) except Exception as e: local = os.path.normpath(DSH_QPANDA3_PLUGIN_LOCAL_DIR) log_warn(f"Plugin tarball download failed ({e}); falling back to local plugin dir") kind, source = "dir", local else: if not tarfile.is_tarfile(TMP_TGZ): local = os.path.normpath(DSH_QPANDA3_PLUGIN_LOCAL_DIR) log_warn("Downloaded file is not a valid tarball; falling back to local plugin dir") kind, source = "dir", local else: # 每次安装都重新下载，并原子覆盖固定缓存路径。文件在插件安装期间保留 # (pnpm 记录的 file: 依赖 spec 指向它，卸载/更新时需能解析)，卸载成功后删除。 os.replace(TMP_TGZ, cached) TMP_TGZ = None log_ok(f"Cached plugin tarball: {cached}") if kind == "file": if not os.path.isfile(source): log_error(f"dsh-qpanda3 plugin tarball not found: {source}") sys.exit(1) name = os.path.basename(source) version = name.removeprefix("dsh-qpanda3-").removesuffix(".tgz") cached = dsh_plugin_cache_path(version) if os.path.normpath(source) != os.path.normpath(cached): # 复制进固定缓存，不改动用户文件；缓存保留供 pnpm 解析，卸载时清理。 os.makedirs(dsh_qpanda3_dir(), exist_ok=True) shutil.copyfile(source, cached) log_ok(f"Cached plugin tarball: {cached}") if kind == "dir": if not os.path.isdir(source): log_error(f"dsh-qpanda3 plugin dir not found: {source}") sys.exit(1) version = dsh_plugin_dir_version(source) or "local" cached = dsh_plugin_cache_path(version) os.makedirs(dsh_qpanda3_dir(), exist_ok=True) dsh_pack_dir_to_tgz(source, cached) log_ok(f"Cached plugin tarball (packed from dir): {cached}") if skills: log_info( f"Injecting {len(skills)} skill(s) into the plugin: " + ", ".join(n for n, _ in skills) ) dsh_inject_skills_into_tgz(cached, skills) log_ok(f"Plugin skills injected: {cached}") return "file", cached, version def dsh_command() -> list[str] | None: """Resolve the `dsh` CLI to an executable command list (cross-platform). On Windows `dsh` ships as dsh.cmd / dsh.ps1; CreateProcess cannot run a batch file directly, so a resolved .cmd/.bat is invoked through `cmd /c` (and .ps1 through `powershell -File`). On POSIX `dsh` is a plain executable (node shebang) and is invoked directly. Returns None when the CLI is not on PATH. """ exe = shutil.which("dsh") if not exe: return None if os.name == "nt": ext = os.path.splitext(exe)[1].lower() if ext == ".cmd" or ext == ".bat": return ["cmd", "/c", exe] if ext == ".ps1": return ["powershell", "-NoProfile", "-File", exe] return [exe] def dsh_extract_credentials(mcp_cfg: dict) -> tuple[str, str, str]: """Pull (apikey, mcp_url, backend_url) out of the generic MCP entry. Mirrors the other AI-coding-tool paths (resolve_k_param/generate_mcp_entry): - apikey: required, from headers X-Api-Key (or the -k KEY string) - mcp_url: MCP server endpoint, cfg url or DEFAULT_MCP_URL - backend_url: optional custom backend channel, only when the config file explicitly sets an X-Qpanda3-Server-Url header — a custom `url` only changes the endpoint, it is never re-sent as a header (mirrors the other AI-coding-tool configs). """ apikey = "" mcp_url = DEFAULT_MCP_URL backend_url = "" for name, cfg in (mcp_cfg or {}).items(): if not isinstance(cfg, dict): continue hdrs = cfg.get("headers") or {} if isinstance(hdrs, dict): if hdrs.get("X-Api-Key"): apikey = hdrs["X-Api-Key"] if hdrs.get("X-Qpanda3-Server-Url"): backend_url = hdrs["X-Qpanda3-Server-Url"] cfg_url = cfg.get("url") if cfg_url: mcp_url = cfg_url return apikey, mcp_url, backend_url def run_prepare_dsh(mcp_cfg: dict, skills: list, scope: str): # dsh 走插件通道：把已下载的技能库注入 dsh-qpanda3 插件的 skills/ 目录后， # 一条 `dsh plugin add` 完成 MCP 配置与全部技能的统一安装。 dsh_cli = dsh_command() if not dsh_cli: log_error("dsh command not found in PATH — install DeepSeek Harness first") sys.exit(1) dsh_home = os.environ.get("DSH_HOME") or os.path.join(os.path.expanduser("~"), ".dsh") if not os.path.isdir(dsh_home): log_error(f"DSH home not found: {dsh_home}") sys.exit(1) log_info("Configuring DeepSeek Harness (dsh)") apikey, mcp_url, backend_url = dsh_extract_credentials(mcp_cfg) if not apikey: log_error("No API key found (pass -k KEY or a config file with X-Api-Key)") sys.exit(1) # 1. Write the user key file the plugin patch reads at boot. # newline="\n" keeps LF on Windows too, matching the patch's split("\n") parser. # Only QPANDA3_API_KEY is written by default; QPANDA3_MCP_URL / QPANDA3_SERVER_URL # are written only when the user explicitly configured a custom MCP endpoint / # backend channel (same rule as the other AI tools, where X-Qpanda3-Server-Url # is optional and the default endpoint is never re-sent as a header). key_file = dsh_key_file_path() lines = [f"QPANDA3_API_KEY={apikey}"] if mcp_url and mcp_url != DEFAULT_MCP_URL: lines.append(f"QPANDA3_MCP_URL={mcp_url}") if backend_url: lines.append(f"QPANDA3_SERVER_URL={backend_url}") with open(key_file, "w", encoding="utf-8", newline="\n") as f: f.write("\n".join(lines) + "\n") log_ok(f"Wrote API key to: {key_file}") # 2. Install the plugin bundle into the web profile (idempotent: pnpm skips an existing dep). # The downloaded skill zips (the same batch the other tools install) are # injected into the plugin tarball's skills/ directory first, then one # `dsh plugin add` installs MCP + all skills together. Source: URL # (default) or a local dir / .tgz via DSH_QPANDA3_PLUGIN_DIR — every # kind materializes into the fixed cache tgz (user files / repo dirs are # never mutated); URL failure falls back to packing the local plugin dir. plugin_kind, plugin_source, plugin_version = dsh_plugin_install_source(skills) log_info(f"Running: dsh plugin --profile web add {plugin_source}") r = subprocess.run( [*dsh_cli, "plugin", "--profile", "web", "add", plugin_source], capture_output=False, ) if r.returncode != 0: log_error(f"dsh plugin add failed (exit {r.returncode})") sys.exit(r.returncode) log_ok("dsh-qpanda3 plugin added to profile 'web'") print("┌────────────────────────────────────────") print("│ dsh-qpanda3 plugin installed") if plugin_version: print(f"│ version: {plugin_version}") if skills: print("│ skills:") for n, zp in skills: v = zip_skill_version(zp) print(f"│ - {n}" + (f" (version: {v})" if v else "")) print("│ (下载的技能库已注入插件 skills/ 目录)") print(f"│ key file: {key_file}") print("│ Tip: restart dsh web (dsh web) so the") print("│ plugin patch and MCP reconnect") print("└────────────────────────────────────────") def run_uninstall_dsh(): sn = "qpanda3-runtime" log_info("Uninstalling dsh-qpanda3 from DeepSeek Harness") removed = False # 1. Remove the plugin bundle from the web profile (drops its MCP + skill # contributions). The cached plugin tarball must still be present here: # pnpm's recorded `file:` dependency spec points at it and has to resolve # for `dsh plugin remove` to succeed. dsh_cli = dsh_command() if dsh_cli: log_info(f"Running: dsh plugin --profile web remove dsh-qpanda3 (bundle for {sn})") r = subprocess.run( [*dsh_cli, "plugin", "--profile", "web", "remove", "dsh-qpanda3"], capture_output=False, ) if r.returncode == 0: removed = True else: log_warn(f"dsh plugin remove exited {r.returncode} (plugin may not be installed)") else: log_warn("dsh command not found in PATH — cannot run 'dsh plugin remove'") # 2. Only after the plugin is removed: delete the cached plugin tarballs so # the dsh home is restored to its pre-install state. If removal failed, # keep the cache — the recorded `file:` spec may still point at it and # deleting it would leave pnpm unable to resolve the dependency. if removed: qdir = dsh_qpanda3_dir() deleted = False if os.path.isdir(qdir): for f in sorted(os.listdir(qdir)): if f.startswith("dsh-qpanda3-") and f.endswith(".tgz"): os.unlink(os.path.join(qdir, f)) deleted = True if deleted: log_ok(f"Removed cached plugin tarballs: {qdir}") try: os.rmdir(qdir) # only succeeds when empty except OSError: pass else: log_info(f"No plugin tarball cache to remove: {qdir}") else: log_warn("dsh-qpanda3 not removed from profile; kept cached plugin tarballs for pnpm resolution") # 3. Remove the user key file so no credential lingers. key_file = dsh_key_file_path() if os.path.isfile(key_file): os.unlink(key_file) log_ok(f"Removed key file: {key_file}") else: log_warn(f"Key file not found: {key_file}") print() log_ok("Uninstall complete! Restart dsh web (dsh web).") def uninstall_nested_json_entry(fp, keys: list, server_name: str): """Remove server_name from a nested JSON key path like data['mcp']['servers'].""" if not os.path.isfile(fp): return None try: with open(fp, encoding="utf-8") as f: ct = f.read() if not ct.strip(): return None data = stdjson.loads(ct) d = data for k in keys[:-1]: d = d.get(k, {}) parent_key = keys[-1] se = d.get(parent_key, {}) if server_name not in se: return "NOT_FOUND" del se[server_name] d[parent_key] = se with open(fp, "w", encoding="utf-8", newline="\n") as f: stdjson.dump(data, f, indent=2, ensure_ascii=False) f.write("\n") return "REMOVED" except Exception as e: return f"ERROR: {e}" # ══════════════════════ Main Entry ══════════════════════ def cmd_list(): print("Supported AI coding tools:") for key, info in SUPPORTED_TOOLS.items(): print(f" {key:12s} {info['name']}") def cmd_help(): print("QPanda3 AI Coding Tools Setup") print("=" * 40) print("This tool configures the QPanda3 Runtime MCP server and") print("installs the QPanda3 skills for AI coding tools. Every skill") print("zip is installed as an independent skills// directory;") print("the script installs / uninstalls them as a batch.") print() print("Supported tools:") for key, info in SUPPORTED_TOOLS.items(): print(f" {key:12s} {info['name']}") print() print("Usage: prepare_qpanda3_ai_kits.py [options]") print() print("commands:") print(" list List supported AI coding tools") print(" prepare [tool] -k KEY [-p] Prepare environment") print(" uninstall [tool] Uninstall environment") print(" help Show this help") print() print("prepare options:") print(" tool AI coding tool name (from list), auto-detect if omitted") print(" -k KEY API key string, or a config file path whose JSON may") print(" set 'skill_url' to one zip URL string or a list of") print(" strings, overriding the built-in skill list") print(" -p, --project Project-level config (script dir as project root)") print() print("uninstall:") print(" tool AI coding tool name, uninstall all if omitted") print() print("Examples:") print(" uv run prepare_qpanda3_ai_kits.py list") print(" uv run prepare_qpanda3_ai_kits.py prepare opencode -k mykey") print(" uv run prepare_qpanda3_ai_kits.py prepare -k mykey") print(" uv run prepare_qpanda3_ai_kits.py prepare codex -k mykey -p") print(" uv run prepare_qpanda3_ai_kits.py prepare dsh -k mykey") print(" uv run prepare_qpanda3_ai_kits.py uninstall claude") print(" uv run prepare_qpanda3_ai_kits.py uninstall dsh") print(" uv run prepare_qpanda3_ai_kits.py uninstall") def main(): import argparse parser = argparse.ArgumentParser( prog="prepare_qpanda3_ai_kits.py", description="Prepare QPanda3 dev environment for AI coding tools", add_help=False, ) parser.add_argument( "command", nargs="?", help="subcommand: list, prepare, uninstall, help" ) parser.add_argument("tool", nargs="?", help="AI coding tool name") parser.add_argument("-k", "--apikey", help="API key or config file path") parser.add_argument( "-p", "--project", action="store_true", help="project-level config" ) args = parser.parse_args() cmd = args.command or "help" if cmd == "list": cmd_list() elif cmd == "help": cmd_help() elif cmd == "prepare": if not args.apikey: log_error("-k/--apikey is required") sys.exit(1) mcp_cfg, skill_urls = resolve_k_param(args.apikey) if mcp_cfg is None: mcp_cfg = generate_mcp_entry(args.apikey, DEFAULT_MCP_URL) scope = "project" if args.project else "user" tools = [args.tool] if args.tool else detect_installed_tools() if not tools: log_error("No tool specified and no installed tools detected") sys.exit(1) # Download every skill zip once, up front; each skill is then installed # as an independent skills// directory for every target tool. skills = [] seen_names = set() for u in skill_urls: name, path = shared_download_zip(u) if name in seen_names: log_warn( f"Duplicate skill name '{name}' from {u}; " "the later install overwrites the earlier one" ) seen_names.add(name) skills.append((name, path)) for t in tools: if t not in SUPPORTED_TOOLS: log_warn(f"Unknown tool: {t}, skipped") continue if not check_tool_usable(t, scope): continue prep_fn = { "opencode": run_prepare_opencode, "claude": run_prepare_claude, "codex": run_prepare_codex, "gemini": run_prepare_gemini, "codebuddy": run_prepare_codebuddy, "cline": run_prepare_cline, "codewhale": run_prepare_codewhale, "zcode": run_prepare_zcode, "dsh": run_prepare_dsh, }.get(t) if prep_fn: print(f"\n{'=' * 44}") print(f" {SUPPORTED_TOOLS[t]['name']} Environment Setup") print(f"{'=' * 44}") prep_fn(mcp_cfg, skills, scope) elif cmd == "uninstall": tools = [args.tool] if args.tool else list(SUPPORTED_TOOLS.keys()) for t in tools: if t not in SUPPORTED_TOOLS: log_warn(f"Unknown tool: {t}, skipped") continue uninstall_fn = { "opencode": run_uninstall_opencode, "claude": run_uninstall_claude, "codex": run_uninstall_codex, "gemini": run_uninstall_gemini, "codebuddy": run_uninstall_codebuddy, "cline": run_uninstall_cline, "codewhale": run_uninstall_codewhale, "zcode": run_uninstall_zcode, "dsh": run_uninstall_dsh, }.get(t) if uninstall_fn: print(f"\n{'=' * 44}") print(f" Uninstalling {SUPPORTED_TOOLS[t]['name']}") print(f"{'=' * 44}") uninstall_fn() else: log_error(f"Unknown command: {cmd}") cmd_help() sys.exit(1) if __name__ == "__main__": OC_NORMALIZED_JSON = "" OC_CONFIG_DIR = "" OC_CONFIG_FILE = "" OC_SKILL_IG = False OC_SKILL_IP = False OC_SKILL_PR = "" CL_NORMALIZED_JSON = "" CL_CONFIG_DIR = "" CL_CONFIG_FILE = "" CL_SKILL_IG = False CL_SKILL_IP = False CL_SKILL_PR = "" CDX_NORMALIZED_JSON = "" CDX_CONFIG_DIR = "" CDX_CONFIG_FILE = "" CDX_SKILL_IG = False CDX_SKILL_IP = False CDX_SKILL_PR = "" GM_NORMALIZED_JSON = "" GM_CONFIG_DIR = "" GM_CONFIG_FILE = "" GM_SKILL_IG = False GM_SKILL_IP = False GM_SKILL_PR = "" CB_NORMALIZED_JSON = "" CB_CONFIG_DIR = "" CB_CONFIG_FILE = "" CB_SKILL_IG = False CB_SKILL_IP = False CB_SKILL_PR = "" CLN_NORMALIZED_JSON = "" CLN_CONFIG_DIR = "" CLN_CONFIG_FILE = "" CW_NORMALIZED_JSON = "" CW_CONFIG_DIR = "" CW_CONFIG_FILE = "" ZC_NORMALIZED_JSON = "" ZC_CONFIG_DIR = "" ZC_CONFIG_FILE = "" ZC_SKILL_IG = False ZC_SKILL_IP = False ZC_SKILL_PR = "" main()