ypshidifu
f7b7695aa7
- fs_view：mcpcfg 内 POSIX 路径在 Windows 操作 WSL 目标时翻译为 \\wsl.localhost UNC 视图，根除「两键路径不存在/mcpcfg.txt不存在」 假错；报错与展示保留原文；read_cfg 返回 active_raw 等 4 个 raw 字段 - _hoist_shared：parse 前把共享 flag 提到最前，mcpctl/ocskill 子命令 前后都能写 --target/--distro 等，根除 unrecognized arguments - wsl_gate 统一门控：读操作 VM 未跑秒回提示，写操作显式打印后唤醒 （45s 超时）；status/mode/backup/_switch/push/pull 全接入 - _wsl_safe_run 零管道执行（stdin=DEVNULL+stdout/stderr 落临时文件） + _wsl_running 先查 --running 绝不无提示冷启动，_wsl_ip 全部改走， 根除 wslrelay 继承管道句柄致 kill 后 EOF 永久阻塞（skill 菜单卡死 必须 ctrl+C 的根因） - mcpctl on 加 push_mcp_config：reload 前 PATCH /config 推送 mcp， 双保险；ocskill list/dirs 连接前打进度行（flush） - oc.py run_safe 护栏：子命令 KeyboardInterrupt/SystemExit/异常 一律回菜单，不再杀穿整个菜单进程 - README 补稳定性设计表与推送双保险链路说明
695 lines
26 KiB
Python
695 lines
26 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
mcpctl —— 命令行版的 MCP 配置对调开关（与 TUI 插件 / server 插件同一套逻辑）
|
||
|
||
做什么：
|
||
status 查看当前模式与 MCP 开/关状态（只读，不改任何文件）
|
||
on 把配置对调为「含 MCP」的原始版，并调用 reload 热重载
|
||
off 把配置对调为「不含 MCP」的安静版，并调用 reload 热重载
|
||
mode 查看或修改 mcpcfg.txt 模式键（自动 / 手动 / 关闭）
|
||
|
||
模式语义（与插件完全一致）：
|
||
自动 大模型可通过 server 插件切换；/mcpon /mcpoff 与本工具视为手动切换一次
|
||
手动 仅允许 /mcpon /mcpoff 与本工具（用户意图），大模型的切换请求会被拒绝
|
||
关闭 功能整体禁用，on/off 都会被拒绝
|
||
|
||
状态判定只看活文件（basename 为 opencode.json 的那份）里 mcp.servers 是否非空，
|
||
与 mcpcfg.txt「含MCP的配置」/「安静配置」两键的行语义无关，因此对调之后身份漂移也不影响判断。
|
||
"""
|
||
|
||
import argparse
|
||
import base64
|
||
import datetime
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
import urllib.error
|
||
import urllib.request
|
||
from pathlib import Path
|
||
|
||
# ---------------------------------------------------------------- 控制台编码
|
||
def _fix_console() -> None:
|
||
for stream in (sys.stdout, sys.stderr):
|
||
try:
|
||
stream.reconfigure(encoding="utf-8", errors="replace") # type: ignore[attr-defined]
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
_fix_console()
|
||
|
||
MODE_LABEL = {"auto": "自动", "manual": "手动", "off": "关闭"}
|
||
LABEL_MODE = {
|
||
"自动": "auto",
|
||
"auto": "auto",
|
||
"audo": "auto", # 用户手误拼写，保持与插件一致
|
||
"手动": "manual",
|
||
"manual": "manual",
|
||
"hand": "manual",
|
||
"on": "manual",
|
||
"关闭": "off",
|
||
"off": "off",
|
||
"disable": "off",
|
||
"disabled": "off",
|
||
}
|
||
|
||
DEFAULT_DISTRO = os.environ.get("OC_SKILL_DISTRO", "Ubuntu-24.04")
|
||
DEFAULT_PORT = int(os.environ.get("OC_PORT", "4096"))
|
||
|
||
|
||
# ---------------------------------------------------------------- 路径工具
|
||
def _basename(path: str) -> str:
|
||
"""跨平台取文件名：同时认 / 与 \\，避免在 Windows 上解析 WSL 路径出错。"""
|
||
return re.split(r"[\\/]", path.strip())[-1]
|
||
|
||
|
||
def _home_config() -> Path:
|
||
return Path.home() / ".config" / "opencode"
|
||
|
||
|
||
def wsl_config_dir(distro: str) -> Path:
|
||
if os.name == "nt":
|
||
return Path(f"\\\\wsl.localhost\\{distro}\\root\\.config\\opencode")
|
||
return Path("/root/.config/opencode")
|
||
|
||
|
||
def resolve_cfg_file(target: str, distro: str) -> Path:
|
||
if target == "wsl":
|
||
return wsl_config_dir(distro) / "mcpcfg.txt"
|
||
return _home_config() / "mcpcfg.txt"
|
||
|
||
|
||
# ---------------------------------------------------------------- WSL 安全访问
|
||
# 1) wsl.exe 子进程绝不留管道：stdout/stderr 落临时文件。否则 wslrelay 等
|
||
# 后代进程继承管道句柄，client 被超时 kill 后 communicate() 等 EOF 会
|
||
# 永久阻塞（skill 目录“卡死必须 ctrl+C”的根因）。
|
||
# 2) wsl.exe 重活之前先查 --running：VM 没跑绝不触发无提示冷启动。
|
||
def _wsl_safe_run(argv: list, timeout: float = 10.0) -> tuple[int, bytes]:
|
||
if os.name != "nt":
|
||
return -1, b""
|
||
import subprocess
|
||
import tempfile
|
||
|
||
with tempfile.TemporaryFile() as fo, tempfile.TemporaryFile() as fe:
|
||
try:
|
||
cp = subprocess.run(
|
||
argv,
|
||
stdin=subprocess.DEVNULL,
|
||
stdout=fo,
|
||
stderr=fe,
|
||
timeout=timeout,
|
||
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
|
||
)
|
||
fo.seek(0)
|
||
fe.seek(0)
|
||
return cp.returncode, fo.read() + fe.read()
|
||
except subprocess.TimeoutExpired:
|
||
return -1, b""
|
||
except Exception as exc:
|
||
return -2, str(exc).encode("utf-8", "replace")
|
||
|
||
|
||
def _wsl_running(distro: str) -> bool:
|
||
if os.name != "nt":
|
||
return True
|
||
code, out = _wsl_safe_run(["wsl.exe", "-l", "-q", "--running"], timeout=8)
|
||
if code != 0 or not out:
|
||
return False
|
||
# wsl.exe 列表输出是 UTF-16LE（常无 BOM）
|
||
text = (
|
||
out.decode("utf-16le", "ignore")
|
||
if out.count(b"\x00") > len(out) // 4
|
||
else out.decode("utf-8", "ignore")
|
||
)
|
||
return distro in text
|
||
|
||
|
||
def fs_view(p: str, target: str, distro: str) -> str:
|
||
"""mcpcfg 内部路径 -> 当前系统可访问的视图。
|
||
Windows 操作 WSL 目标时，/root/... 必须换成 \\\\wsl.localhost\\<distro>\\...，
|
||
否则 Path("/root/...") 会被解析成 C:\\root → 误报「两键路径不存在」
|
||
（用户看到的“键值不存在 / mcpcfg.txt不存在”就是这个假错）。"""
|
||
if target == "wsl" and os.name == "nt" and p.startswith("/"):
|
||
return "\\\\wsl.localhost\\%s%s" % (distro, p.replace("/", "\\"))
|
||
return p
|
||
|
||
|
||
def wsl_gate(args, need_write: bool = False) -> str | None:
|
||
"""WSL 目标统一门（返回 None=可继续；返回字符串=错误信息）：
|
||
- 读操作（status/mode查看/backup）：VM 没跑 → 秒回清晰提示，绝不撞 UNC 冷启动
|
||
- 写操作（on/off/mode设置）：VM 没跑 → 显式提示后唤醒，唤醒失败才报错"""
|
||
if getattr(args, "target", "local") != "wsl" or os.name != "nt":
|
||
return None
|
||
if _wsl_running(args.distro):
|
||
return None
|
||
if not need_write:
|
||
return (
|
||
f"WSL（{args.distro}）未运行，读不到它的配置。"
|
||
f"先启动：wsl.exe -d {args.distro}"
|
||
)
|
||
print(f"WSL（{args.distro}）未运行，正在唤醒（可能需要十几秒）...", flush=True)
|
||
_wsl_safe_run(["wsl.exe", "-d", args.distro, "--", "true"], timeout=45)
|
||
if _wsl_running(args.distro):
|
||
print("WSL 已就绪。", flush=True)
|
||
return None
|
||
return (
|
||
f"WSL（{args.distro}）唤醒失败，无法访问其配置。"
|
||
f"请手动执行 wsl.exe -d {args.distro} 启动后重试。"
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------- mcpcfg 解析
|
||
# 与 tui.js / ai-mcp.ts 的 readMcpCfg 保持一致：
|
||
# 优先按键值解析（模式: / 含MCP的配置: / 安静配置:，认中英文冒号），
|
||
# 解析不到再回退旧「裸三行」（第1行模式、第2行含MCP、第3行安静）。
|
||
def read_cfg(path: Path, target: str = "local", distro: str = DEFAULT_DISTRO) -> dict:
|
||
if not path.exists():
|
||
return {
|
||
"mode": "off",
|
||
"error": f"缺少配置文件：{path}（配置缺失时功能视为关闭）",
|
||
}
|
||
try:
|
||
raw_text = path.read_text(encoding="utf-8", errors="replace")
|
||
except Exception as exc:
|
||
return {"mode": "off", "error": f"读取 {path} 失败：{exc}"}
|
||
|
||
lines = raw_text.splitlines()
|
||
|
||
kv: dict[str, str] = {}
|
||
for ln in lines:
|
||
line = ln.strip()
|
||
if not line or line.startswith("#") or line.startswith("//"):
|
||
continue
|
||
m = re.match(r"^([^:：]+)[:：]\s*(.*)$", line)
|
||
if m:
|
||
kv[m.group(1).strip().lower()] = m.group(2).strip()
|
||
|
||
mode_val = kv.get("模式", kv.get("mode", kv.get("第1行")))
|
||
on_val = kv.get("含mcp的配置", kv.get("含mcp配置", kv.get("含mcp", kv.get("on", kv.get("原始配置", kv.get("第2行"))))))
|
||
quiet_val = kv.get("安静配置", kv.get("安静", kv.get("quiet", kv.get("备份", kv.get("backup", kv.get("第3行"))))))
|
||
|
||
is_legacy = mode_val is None and on_val is None and quiet_val is None
|
||
if not is_legacy:
|
||
mode = LABEL_MODE.get((mode_val or "自动").strip().lower(), "auto")
|
||
on_path = (on_val or "").strip()
|
||
quiet_path = (quiet_val or "").strip()
|
||
else:
|
||
bare = [l.strip() for l in lines if l.strip() and not l.strip().startswith("#") and not l.strip().startswith("//")]
|
||
mode = LABEL_MODE.get((bare[0] if bare else "").strip().lower(), "auto" if not bare or bare[0] == "" else "auto")
|
||
on_path = (bare[1] if len(bare) > 1 else "").strip()
|
||
quiet_path = (bare[2] if len(bare) > 2 else "").strip()
|
||
|
||
# 先按原文识别身份，再统一转成“可访问视图”（Windows→WSL 换 UNC）；
|
||
# raw 保留在结果里用于展示与报错（与用户在 txt 里看到的一致）
|
||
on_raw, quiet_raw = on_path, quiet_path
|
||
active_raw = shadow_raw = None
|
||
if _basename(on_path) == "opencode.json":
|
||
active_raw, shadow_raw = on_path, quiet_path
|
||
elif _basename(quiet_path) == "opencode.json":
|
||
active_raw, shadow_raw = quiet_path, on_path
|
||
|
||
on_path = fs_view(on_raw, target, distro)
|
||
quiet_path = fs_view(quiet_raw, target, distro)
|
||
active = fs_view(active_raw, target, distro) if active_raw else None
|
||
shadow = fs_view(shadow_raw, target, distro) if shadow_raw else None
|
||
|
||
result = {
|
||
"mode": mode,
|
||
"on_path": on_path,
|
||
"quiet_path": quiet_path,
|
||
"active": active,
|
||
"shadow": shadow,
|
||
"on_raw": on_raw,
|
||
"quiet_raw": quiet_raw,
|
||
"active_raw": active_raw,
|
||
"shadow_raw": shadow_raw,
|
||
"legacy": is_legacy,
|
||
"error": None,
|
||
}
|
||
if mode != "off" and not active:
|
||
result["error"] = "mcpcfg.txt 两行路径都不是 opencode.json（「含MCP的配置」或「安静配置」需指向 opencode.json）"
|
||
elif mode != "off" and (not Path(on_path).exists() or not Path(quiet_path).exists()):
|
||
gone = [
|
||
raw
|
||
for raw, view in ((on_raw, on_path), (quiet_raw, quiet_path))
|
||
if not Path(view).exists()
|
||
]
|
||
result["error"] = (
|
||
"mcpcfg.txt 指向的配置文件不存在: "
|
||
+ "、".join(gone)
|
||
+ "（请检查「含MCP的配置」/「安静配置」两键的路径）"
|
||
)
|
||
return result
|
||
|
||
|
||
def write_cfg(path: Path, mode: str, on_path: str, quiet_path: str) -> None:
|
||
"""按键值格式写回（模式/含MCP的配置/安静配置），带注释头。"""
|
||
mode_cn = {"auto": "自动", "manual": "手动", "off": "关闭"}.get(mode, "自动")
|
||
body = "\n".join([
|
||
"# MCP 开关配置：改完保存，重启 OC 生效（第1个键=模式）",
|
||
"# 模式可填：自动（大模型可切换） / 手动（仅 /mcpon /mcpoff） / 关闭（禁用本功能）",
|
||
f"模式: {mode_cn}",
|
||
f"含MCP的配置: {on_path}",
|
||
f"安静配置: {quiet_path}",
|
||
"",
|
||
])
|
||
path.write_text(body, encoding="utf-8")
|
||
|
||
|
||
def set_mode_in_file(path: Path, mode: str) -> None:
|
||
"""只改模式键：键值文件按键替换；旧三行文件替换第1行（插件下次读会自动迁移）。"""
|
||
raw = path.read_text(encoding="utf-8", errors="replace")
|
||
lines = raw.splitlines()
|
||
mode_cn = {"auto": "自动", "manual": "手动", "off": "关闭"}.get(mode, "自动")
|
||
|
||
kv_pattern = re.compile(r"^(模式|mode)\s*[:：]\s*.*$", re.IGNORECASE)
|
||
hit = False
|
||
for i, ln in enumerate(lines):
|
||
if kv_pattern.match(ln.strip()):
|
||
prefix = ":" if "模式" in ln.split(":")[0] or "：" in ln else ":"
|
||
lines[i] = f"模式{prefix} {mode_cn}"
|
||
hit = True
|
||
break
|
||
if not hit:
|
||
# 旧三行（或键缺失）：第1行位放模式；若本来是键值缺模式键，插到最前面
|
||
if any(re.match(r"^([^:：]+)[:：]", l.strip()) for l in lines if l.strip() and not l.strip().startswith("#")):
|
||
lines.insert(0, f"模式: {mode_cn}")
|
||
else:
|
||
while len(lines) < 3:
|
||
lines.append("")
|
||
lines[0] = mode_cn
|
||
path.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
||
|
||
|
||
def flag_path(cfg_dir_target: str, distro: str) -> Path:
|
||
"""显式开启标记 mcp.on.flag 的落点（与插件共用；wsl 目标落在 WSL 目录）。"""
|
||
base = wsl_config_dir(distro) if cfg_dir_target == "wsl" else _home_config()
|
||
return base / "mcp.on.flag"
|
||
|
||
|
||
def set_on_flag(cfg_dir_target: str, distro: str, on: bool) -> None:
|
||
try:
|
||
p = flag_path(cfg_dir_target, distro)
|
||
if on:
|
||
p.write_text(str(os.getpid()), encoding="utf-8")
|
||
elif p.exists():
|
||
p.unlink()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def active_has_mcp(active: str | None) -> bool:
|
||
try:
|
||
if not active:
|
||
return False
|
||
data = json.loads(Path(active).read_text(encoding="utf-8"))
|
||
servers = (data.get("mcp") or {}).get("servers") or {}
|
||
return bool(servers)
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
def swap_files(a: str, b: str) -> None:
|
||
pa, pb = Path(a), Path(b)
|
||
ca, cb = pa.read_bytes(), pb.read_bytes()
|
||
pa.write_bytes(cb)
|
||
pb.write_bytes(ca)
|
||
|
||
|
||
# ---------------------------------------------------------------- reload
|
||
def _read_password(cfg_dir: Path) -> str:
|
||
try:
|
||
data = json.loads((cfg_dir / "opencode.json").read_text(encoding="utf-8"))
|
||
pwd = (data.get("server") or {}).get("password")
|
||
return str(pwd) if pwd else ""
|
||
except Exception:
|
||
return ""
|
||
|
||
|
||
def _wsl_ip(distro: str) -> str:
|
||
"""取 WSL IP；VM 没跑直接返回空（绝不无提示冷启动），
|
||
且走 _wsl_safe_run（零管道）——原 capture_output 版在 VM 被清理后
|
||
会因 wslrelay 继承管道句柄导致 kill 后 communicate 永久阻塞。"""
|
||
if os.name != "nt" or not _wsl_running(distro):
|
||
return ""
|
||
code, out = _wsl_safe_run(
|
||
["wsl.exe", "-d", distro, "--", "bash", "-lc", "hostname -I"], timeout=15
|
||
)
|
||
if code != 0:
|
||
return ""
|
||
for tok in out.decode("utf-8", "ignore").split():
|
||
if re.fullmatch(r"\d{1,3}(\.\d{1,3}){3}", tok):
|
||
return tok
|
||
return ""
|
||
|
||
|
||
# 关键：本机/WSL 的 OC 绝不能走系统代理。
|
||
# 环境里若有 HTTP_PROXY/HTTPS_PROXY，urlopen 会把 localhost 请求丢给代理，
|
||
# 结果是代理回 502 —— 看起来像 OC 挂了，其实是被劫走了。
|
||
_NO_PROXY_OPENER = urllib.request.build_opener(urllib.request.ProxyHandler({}))
|
||
if os.environ.get("MCPCTL_USE_PROXY"):
|
||
_NO_PROXY_OPENER = urllib.request.build_opener()
|
||
|
||
|
||
def _post(
|
||
url: str,
|
||
password: str,
|
||
timeout: float,
|
||
data: bytes = b"{}",
|
||
method: str = "POST",
|
||
) -> tuple[int, str]:
|
||
req = urllib.request.Request(url, method=method, data=data)
|
||
req.add_header("content-type", "application/json")
|
||
req.add_header("accept", "application/json")
|
||
if password:
|
||
req.add_header("authorization", f"Bearer {password}")
|
||
try:
|
||
with _NO_PROXY_OPENER.open(req, timeout=timeout) as resp:
|
||
return resp.status, resp.read().decode("utf-8", "replace")
|
||
except urllib.error.HTTPError as exc:
|
||
if exc.code == 401 and password:
|
||
raw = base64.b64encode(f":{password}".encode()).decode("ascii")
|
||
req2 = urllib.request.Request(url, method=method, data=data)
|
||
req2.add_header("content-type", "application/json")
|
||
req2.add_header("authorization", f"Basic {raw}")
|
||
try:
|
||
with _NO_PROXY_OPENER.open(req2, timeout=timeout) as resp:
|
||
return resp.status, resp.read().decode("utf-8", "replace")
|
||
except urllib.error.HTTPError as exc2:
|
||
return exc2.code, str(exc2)
|
||
return exc.code, str(exc)
|
||
except Exception as exc:
|
||
return 0, str(exc)
|
||
|
||
|
||
def do_reload(target: str, args) -> tuple[bool, str]:
|
||
password = args.password or _read_password(
|
||
wsl_config_dir(args.distro) if target == "wsl" else _home_config()
|
||
)
|
||
hosts = [f"http://localhost:{args.port}", f"http://127.0.0.1:{args.port}"]
|
||
if target == "wsl" and os.name == "nt":
|
||
ip = _wsl_ip(args.distro)
|
||
if ip:
|
||
hosts.append(f"http://{ip}:{args.port}")
|
||
|
||
last = ""
|
||
for host in hosts:
|
||
code, body = _post(host + "/api/location/reload", password, args.timeout)
|
||
if 200 <= code < 300:
|
||
return True, f"热重载成功（{host}/api/location/reload）"
|
||
if code in (401, 403):
|
||
last = f"{host} 鉴权失败（HTTP {code}），请用 --password 指定 server.password"
|
||
elif code == 404:
|
||
last = f"{host} 未提供 /api/location/reload（可能需要重启 OC 生效）"
|
||
elif code:
|
||
last = f"{host} 返回 HTTP {code}：{body[:200]}"
|
||
else:
|
||
last = f"{host} 连不上（OC 可能没在运行）"
|
||
return False, last
|
||
|
||
|
||
def push_mcp_config(args) -> tuple[bool, str]:
|
||
"""reload 之前的兜底：把活配置里的 mcp 直接 PATCH 进运行中的 OC。
|
||
与 tui.js / ai-mcp.ts 的 pushMcpConfig 同一套链（SDK PATCH /config 无 /api 前缀）。
|
||
只在「开启」方向推送；失败不阻断，继续走 reload 单通道。"""
|
||
path = resolve_cfg_file(args.target, args.distro)
|
||
cfg = read_cfg(path, args.target, args.distro)
|
||
if cfg.get("error") or not cfg.get("active"):
|
||
return False, cfg.get("error") or "未识别活配置"
|
||
try:
|
||
data = json.loads(Path(cfg["active"]).read_text(encoding="utf-8"))
|
||
except Exception as exc:
|
||
return False, f"读活配置失败：{exc}"
|
||
mcp = data.get("mcp")
|
||
if not isinstance(mcp, dict) or not ((mcp.get("servers") or {})):
|
||
return False, "活配置里没有 mcp.servers，无可推送内容"
|
||
|
||
password = args.password or _read_password(
|
||
wsl_config_dir(args.distro) if args.target == "wsl" else _home_config()
|
||
)
|
||
hosts = [f"http://localhost:{args.port}", f"http://127.0.0.1:{args.port}"]
|
||
if args.target == "wsl" and os.name == "nt":
|
||
ip = _wsl_ip(args.distro)
|
||
if ip:
|
||
hosts.append(f"http://{ip}:{args.port}")
|
||
|
||
body = json.dumps({"mcp": mcp}).encode("utf-8")
|
||
last = ""
|
||
for host in hosts:
|
||
code, resp = _post(
|
||
host + "/config", password, args.timeout, data=body, method="PATCH"
|
||
)
|
||
if 200 <= code < 300:
|
||
return True, f"已推送（{host}/config）"
|
||
last = f"{host} HTTP {code}：{resp[:160]}"
|
||
return False, last
|
||
|
||
|
||
# ---------------------------------------------------------------- 子命令
|
||
def _stamp() -> str:
|
||
return datetime.datetime.now().strftime("%Y%m%d%H%M%S")
|
||
|
||
|
||
def print_state(cfg: dict) -> bool:
|
||
has = active_has_mcp(cfg.get("active"))
|
||
print(f"模式：{MODE_LABEL.get(cfg['mode'], cfg['mode'])}（mcpcfg.txt「模式」键）")
|
||
print(f"MCP ：{'已开启（配置含 mcp.servers）' if has else '已关闭（安静，无 mcp 配置）'}")
|
||
# 展示用原文（用户在 mcpcfg.txt 里看到的路径），内部访问已走视图翻译
|
||
shown = cfg.get("active_raw") or cfg.get("active") or "（未识别）"
|
||
print(f"活配置：{shown}")
|
||
if cfg.get("error"):
|
||
print(f"警告：{cfg['error']}")
|
||
return has
|
||
|
||
|
||
def cmd_status(args) -> int:
|
||
err = wsl_gate(args, need_write=False)
|
||
if err:
|
||
print(err)
|
||
return 2
|
||
path = resolve_cfg_file(args.target, args.distro)
|
||
print(f"配置文件：{path}")
|
||
cfg = read_cfg(path, args.target, args.distro)
|
||
print_state(cfg)
|
||
return 0 if not cfg.get("error") else 2
|
||
|
||
|
||
def _switch(args, want_on: bool) -> int:
|
||
# 写操作：WSL 没跑先唤醒（gate 在任何 UNC .exists() 之前）
|
||
err = wsl_gate(args, need_write=True)
|
||
if err:
|
||
print(err)
|
||
return 2
|
||
path = resolve_cfg_file(args.target, args.distro)
|
||
print(f"配置文件：{path}")
|
||
cfg = read_cfg(path, args.target, args.distro)
|
||
if cfg.get("error"):
|
||
print(f"失败：{cfg['error']}")
|
||
return 2
|
||
|
||
action = "开启" if want_on else "关闭"
|
||
if cfg["mode"] == "off" and not args.ignore_mode:
|
||
print("拒绝：MCP 开关功能已关闭（mcpcfg.txt「模式: 关闭」）。")
|
||
print("如需使用，请把模式改成「自动」或「手动」。")
|
||
return 3
|
||
|
||
has = active_has_mcp(cfg["active"])
|
||
if has == want_on:
|
||
print(f"无需切换：MCP 已是{'开启' if want_on else '关闭（安静）'}状态。")
|
||
print_state(cfg)
|
||
return 0
|
||
|
||
swap_files(cfg["on_path"], cfg["quiet_path"])
|
||
after = active_has_mcp(cfg["active"])
|
||
if after != want_on:
|
||
print("失败：配置对调后状态校验不符，请检查 mcpcfg.txt「含MCP的配置」/「安静配置」路径。")
|
||
return 4
|
||
|
||
# 显式意图落标记：开写 mcp.on.flag、关删标记（与 tui.js / ai-mcp 共用，
|
||
# 避免自动模式的“启动归位”把刚开启的状态又换回去）
|
||
set_on_flag(args.target, args.distro, want_on)
|
||
|
||
print(f"配置已对调，MCP 现在是「{'开启' if after else '关闭'}」。")
|
||
|
||
# 兜底推送（仅开启方向）：reload 之前先把 mcp 配置送进运行中的 OC，
|
||
# 推送成功 = 双保险；推送失败不阻断，继续走 reload 兜底
|
||
pushed = None
|
||
pmsg = ""
|
||
if want_on and not args.no_reload:
|
||
pushed, pmsg = push_mcp_config(args)
|
||
if pushed:
|
||
print(f"mcp 配置已先推送进 OC：{pmsg}")
|
||
else:
|
||
print(f"配置推送未生效（继续走 reload 兜底）：{pmsg}")
|
||
|
||
if args.no_reload:
|
||
print("已按 --no-reload 跳过热重载，需重启 OC 生效。")
|
||
return 0
|
||
|
||
ok, msg = do_reload(args.target, args)
|
||
if ok:
|
||
if pushed:
|
||
print(f"{msg}，mcp 配置已先送入 + 热重载（双保险），MCP 百分百生效。")
|
||
else:
|
||
print(f"{msg}，MCP 配置立即生效。")
|
||
return 0
|
||
print(f"热重载失败：{msg}")
|
||
if pushed:
|
||
print("但 mcp 配置已提前推送进 OC，应已生效；重启 OpenCode 亦可完全同步。")
|
||
return 0
|
||
print("配置文件已经对调完成，重启 OpenCode 也能生效。")
|
||
return 5
|
||
|
||
|
||
def cmd_on(args) -> int:
|
||
return _switch(args, True)
|
||
|
||
|
||
def cmd_off(args) -> int:
|
||
return _switch(args, False)
|
||
|
||
|
||
def cmd_mode(args) -> int:
|
||
# 查看=读门（VM 没跑秒回提示）；设置=写门（唤醒）
|
||
err = wsl_gate(args, need_write=bool(args.value))
|
||
if err:
|
||
print(err)
|
||
return 2
|
||
path = resolve_cfg_file(args.target, args.distro)
|
||
if not path.exists():
|
||
print(f"配置文件不存在：{path}")
|
||
return 2
|
||
|
||
if not args.value:
|
||
cfg = read_cfg(path, args.target, args.distro)
|
||
cur = MODE_LABEL.get(cfg.get("mode", "auto"), cfg.get("mode", "auto"))
|
||
print(f"当前模式：{cur}")
|
||
print("可选值：自动 / 手动 / 关闭（也支持 auto / manual / off）")
|
||
return 0
|
||
|
||
key = args.value.strip().lower()
|
||
if key not in LABEL_MODE:
|
||
print(f"无法识别的模式：{args.value}（可选：自动 / 手动 / 关闭 / auto / manual / off）")
|
||
return 2
|
||
|
||
set_mode_in_file(path, LABEL_MODE[key])
|
||
print(f"已把模式设为：{MODE_LABEL[LABEL_MODE[key]]}（{path}）")
|
||
return 0
|
||
|
||
|
||
def cmd_backup(args) -> int:
|
||
err = wsl_gate(args, need_write=False)
|
||
if err:
|
||
print(err)
|
||
return 2
|
||
path = resolve_cfg_file(args.target, args.distro)
|
||
if not path.exists():
|
||
print(f"配置文件不存在：{path}")
|
||
return 2
|
||
bak = path.with_name(path.name + f".bak-{_stamp()}")
|
||
bak.write_bytes(path.read_bytes())
|
||
print(f"已备份：{bak}")
|
||
return 0
|
||
|
||
|
||
# ---------------------------------------------------------------- 参数顺序自愈
|
||
# 共享 flag 定义在主解析器（子命令之前），`mcpctl status --target wsl` 这种
|
||
# 写法会被 argparse 判为 unrecognized。这里在 parse 之前把共享 flag 从任意
|
||
# 位置提到最前，让用户不必记参数顺序（与 oc.py 菜单直传兼容）。
|
||
_SHARED_VALUE_FLAGS = {"--target", "--distro", "--port", "--timeout", "--password"}
|
||
_SHARED_BOOL_FLAGS = {"--no-reload", "--ignore-mode"}
|
||
|
||
|
||
def _hoist_shared(argv: list[str]) -> list[str]:
|
||
head: list[str] = []
|
||
rest: list[str] = []
|
||
i = 0
|
||
n = len(argv)
|
||
while i < n:
|
||
tok = argv[i]
|
||
key = tok.split("=", 1)[0]
|
||
if key in _SHARED_VALUE_FLAGS:
|
||
if "=" in tok:
|
||
head.append(tok)
|
||
i += 1
|
||
else:
|
||
head.append(tok)
|
||
if i + 1 < n:
|
||
head.append(argv[i + 1])
|
||
i += 2
|
||
else:
|
||
i += 1
|
||
elif key in _SHARED_BOOL_FLAGS:
|
||
head.append(tok)
|
||
i += 1
|
||
else:
|
||
rest.append(tok)
|
||
i += 1
|
||
return head + rest
|
||
|
||
|
||
# ---------------------------------------------------------------- 入口
|
||
def build_parser() -> argparse.ArgumentParser:
|
||
p = argparse.ArgumentParser(
|
||
prog="mcpctl",
|
||
description="MCP 配置对调开关（status / on / off / mode），与插件共用同一份 mcpcfg.txt",
|
||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||
epilog=(
|
||
"示例：\n"
|
||
" python mcpctl.py status\n"
|
||
" python mcpctl.py on\n"
|
||
" python mcpctl.py off --no-reload\n"
|
||
" python mcpctl.py mode 手动\n"
|
||
" python mcpctl.py status --target wsl\n"
|
||
),
|
||
)
|
||
p.add_argument("--target", choices=["local", "wsl"], default="local", help="操作本机还是 WSL 的配置（默认 local）")
|
||
p.add_argument("--distro", default=DEFAULT_DISTRO, help=f"WSL 发行版名（默认 {DEFAULT_DISTRO}）")
|
||
p.add_argument("--port", type=int, default=DEFAULT_PORT, help=f"OC 端口（默认 {DEFAULT_PORT}）")
|
||
p.add_argument("--timeout", type=float, default=5.0, help="HTTP 超时秒数")
|
||
p.add_argument("--password", default="", help="OC server.password（默认自动从 opencode.json 读）")
|
||
p.add_argument("--no-reload", action="store_true", help="只对调文件，不调用热重载")
|
||
p.add_argument("--ignore-mode", action="store_true", help="忽略「模式: 关闭」限制强制切换（慎用）")
|
||
|
||
sub = p.add_subparsers(dest="cmd")
|
||
|
||
sp = sub.add_parser("status", help="查看当前模式与 MCP 开关状态")
|
||
sp.set_defaults(func=cmd_status)
|
||
|
||
sp = sub.add_parser("on", help="对调为含 MCP 的配置并热重载")
|
||
sp.set_defaults(func=cmd_on)
|
||
|
||
sp = sub.add_parser("off", help="对调为安静配置（删 MCP）并热重载")
|
||
sp.set_defaults(func=cmd_off)
|
||
|
||
sp = sub.add_parser("mode", help="查看或修改 mcpcfg.txt「模式」键（自动/手动/关闭）")
|
||
sp.add_argument("value", nargs="?", help="要设置的模式：自动 / 手动 / 关闭（不传则只查看）")
|
||
sp.set_defaults(func=cmd_mode)
|
||
|
||
sp = sub.add_parser("backup", help="备份当前 mcpcfg.txt")
|
||
sp.set_defaults(func=cmd_backup)
|
||
|
||
return p
|
||
|
||
|
||
def main(argv: list[str] | None = None) -> int:
|
||
if argv is None:
|
||
argv = sys.argv[1:]
|
||
argv = _hoist_shared(list(argv))
|
||
parser = build_parser()
|
||
args = parser.parse_args(argv)
|
||
if not getattr(args, "cmd", None):
|
||
parser.print_help()
|
||
return 0
|
||
try:
|
||
return args.func(args)
|
||
except KeyboardInterrupt:
|
||
print("\n已取消。")
|
||
return 130
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|