478 lines
18 KiB
Python
478 lines
18 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"
|
||
|
||
|
||
# ---------------------------------------------------------------- mcpcfg 解析
|
||
# 与 tui.js / ai-mcp.ts 的 readMcpCfg 保持一致：
|
||
# 优先按键值解析（模式: / 含MCP的配置: / 安静配置:，认中英文冒号），
|
||
# 解析不到再回退旧「裸三行」（第1行模式、第2行含MCP、第3行安静）。
|
||
def read_cfg(path: Path) -> 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()
|
||
|
||
active = shadow = None
|
||
if _basename(on_path) == "opencode.json":
|
||
active, shadow = on_path, quiet_path
|
||
elif _basename(quiet_path) == "opencode.json":
|
||
active, shadow = quiet_path, on_path
|
||
|
||
result = {
|
||
"mode": mode,
|
||
"on_path": on_path,
|
||
"quiet_path": quiet_path,
|
||
"active": active,
|
||
"shadow": shadow,
|
||
"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()):
|
||
result["error"] = "mcpcfg.txt 指向的配置文件不存在（请检查「含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:
|
||
if os.name != "nt":
|
||
return ""
|
||
try:
|
||
import subprocess
|
||
|
||
out = subprocess.run(
|
||
["wsl.exe", "-d", distro, "--", "bash", "-lc", "hostname -I"],
|
||
capture_output=True,
|
||
timeout=15,
|
||
)
|
||
for tok in (out.stdout or b"").decode("utf-8", "ignore").split():
|
||
if re.fullmatch(r"\d{1,3}(\.\d{1,3}){3}", tok):
|
||
return tok
|
||
except Exception:
|
||
pass
|
||
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) -> tuple[int, str]:
|
||
req = urllib.request.Request(url, method="POST", data=b"{}")
|
||
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="POST", data=b"{}")
|
||
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 _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 配置）'}")
|
||
print(f"活配置：{cfg.get('active') or '（未识别）'}")
|
||
if cfg.get("error"):
|
||
print(f"警告：{cfg['error']}")
|
||
return has
|
||
|
||
|
||
def cmd_status(args) -> int:
|
||
path = resolve_cfg_file(args.target, args.distro)
|
||
print(f"配置文件：{path}")
|
||
cfg = read_cfg(path)
|
||
print_state(cfg)
|
||
return 0 if not cfg.get("error") else 2
|
||
|
||
|
||
def _switch(args, want_on: bool) -> int:
|
||
path = resolve_cfg_file(args.target, args.distro)
|
||
print(f"配置文件：{path}")
|
||
cfg = read_cfg(path)
|
||
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 '关闭'}」。")
|
||
|
||
if args.no_reload:
|
||
print("已按 --no-reload 跳过热重载，需重启 OC 生效。")
|
||
return 0
|
||
|
||
ok, msg = do_reload(args.target, args)
|
||
if ok:
|
||
print(f"{msg}，MCP 配置立即生效。")
|
||
return 0
|
||
print(f"热重载失败：{msg}")
|
||
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:
|
||
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)
|
||
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:
|
||
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
|
||
|
||
|
||
# ---------------------------------------------------------------- 入口
|
||
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:
|
||
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())
|