wsl-round-mcp:本机与 WSL OpenCode 互通的 CLI 工具

ocskill.py(skill 互通,零第三方依赖):
- list 运行时连接 WSL 的 OC(/api/skill 与 /skill 双端点,Bearer→Basic 自动降级),连不上则降级扫描 skills 目录
- push 接收 zip / 目录 / 单个 SKILL.md / base64(显式或按内容嗅探),先解码归一,再校验标准 skill 格式
  校验四项:有 SKILL.md、有 YAML frontmatter、frontmatter 有 name 且合法、有 description
  含 zip-slip 目录穿越防护;同名已存在需 --force(覆盖前自动备份)
- pull 反向把 WSL 里的 skill 拷回本机;dirs 查看两端目录与连接状态
- 写入 \\wsl.localhost\<发行版>\root\.config\opencode\skills\,OC 原生 watch 自动热加载,无需重启

mcpctl.py(MCP 配置对调开关,与插件同一套逻辑):
- status / on / off / mode / backup;读同一份 mcpcfg.txt,状态只看活文件是否含 mcp.servers
- 对调后调用 POST /api/location/reload 热重载;模式门控与插件一致(关闭一律拒绝)

附 6 个双击 bat(英文文案规避 GBK 编码问题,内部 chcp 65001)与中文 README。
This commit is contained in:
ypshidifu
2026-09-23 12:06:32 +08:00
commit 873d965ef3
10 changed files with 1277 additions and 0 deletions
+14
View File
@@ -0,0 +1,14 @@
# Python
__pycache__/
*.py[cod]
*.egg-info/
# 本地回滚/备份产物
*.bak-*
skills-backup/
# 编辑器与系统
.vscode/
.idea/
.DS_Store
Thumbs.db
+137
View File
@@ -0,0 +1,137 @@
# wsl-round-mcp — 本机 ⇄ WSL OpenCode 互通工具
打通**本机 OpenCode** 与 **WSL 里的 OpenCode**:skill 列表互查、skill 一键推送、MCP 配置对调开关。
全部零第三方依赖,只要装了 Python 3 就能跑。
---
## 一、双击就能用(小白版)
把整个 `wsl-round-mcp` 文件夹拷到任意位置,双击对应的 bat:
| 文件 | 作用 |
| --- | --- |
| `skill-list.bat` | 连接 WSL 的 OC,列出它当前有哪些 skill |
| `skill-push.bat` | 会让你粘贴一个 skill 路径(zip / 目录 / SKILL.md / base64),校验格式后写进 WSL 的 skill 目录 |
| `skill-pull.bat` | 会让你输入 skill 名,把它从 WSL 拷回本机 `skills-backup\` |
| `mcp-status.bat` | 查看 MCP 当前是开是关、第1行是什么模式 |
| `mcp-on.bat` | 把配置对调为「含 MCP」并热重载 |
| `mcp-off.bat` | 把配置对调为「安静版(无 MCP)」并热重载 |
> bat 内部会自动 `chcp 65001`,中文输出不会乱码。
> 没装 Python 或没加进 PATH,会提示 `Python not found in PATH`。
---
## 二、命令行用法
### 1. `ocskill.py` — skill 互通
```bash
# 列出 WSL OC 的 skill(API 连不上会自动退回目录扫描)
python ocskill.py list
# 推送:压缩包(自动解压)
python ocskill.py push .\my-skill.zip
# 推送:整个 skill 目录
python ocskill.py push .\my-skill
# 推送:单个 SKILL.md(会自动包成以 name 命名的目录)
python ocskill.py push .\SKILL.md
# 推送:base64 内容
python ocskill.py push <一长串base64> --base64
# 只校验格式、不写入
python ocskill.py push .\my-skill --dry-run
# 覆盖同名 skill(先自动备份成 .bak-时间戳)
python ocskill.py push .\my-skill --force
# 反向:把 WSL 里的 skill 拷回本机
python ocskill.py pull java-expert --out .\skills-backup
# 看两端目录与 API 连接状态
python ocskill.py dirs
```
**push 的完整流程**(对应「先解码 → 确认标准 skill 格式 → 传入 skill 目录」):
1. **解码归一**:zip 就解压(带 zip-slip 目录穿越防护),目录原样用,
单个 `.md` 会包成目录,base64 字符串先解成文件再判断是 zip 还是 md。
2. **定位 `SKILL.md`**:允许压缩包里多套一层父目录(最深 3 层)。
3. **校验标准 skill 格式**,四项全过才算通过:
- 找得到 `SKILL.md`
- 文件以 `---` 开头,有 YAML frontmatter
- frontmatter 里有 `name`,且只含 `字母/数字/._-`
- frontmatter 里有 `description`(大模型靠它决定何时调用)
4. **写入** `\\wsl.localhost\<发行版>\root\.config\opencode\skills\<name>\`,
OC 会 watch 这个目录(约 100ms 去抖后自动 rescan 并推 `skill.reload`),**不用重启**。
常用参数:
| 参数 | 默认 | 说明 |
| --- | --- | --- |
| `--distro` | `Ubuntu-24.04` | WSL 发行版名 |
| `--url` | 自动探测 | 显式指定 OC 地址,如 `http://localhost:4096` |
| `--port` | `4096` | OC 端口(等价于环境变量 `OC_PORT`) |
| `--password` | 自动读 | OC 的 `server.password`,先试 Bearer、401 再试 Basic |
### 2. `mcpctl.py` — MCP 配置对调开关
```bash
python mcpctl.py status # 看状态(只读)
python mcpctl.py on # 对调成含 MCP + 热重载
python mcpctl.py off # 对调成安静版 + 热重载
python mcpctl.py mode # 看第1行
python mcpctl.py mode 手动 # 改第1行(自动/手动/关闭,或 auto/manual/off)
python mcpctl.py status --target wsl # 操作 WSL 那一份配置
python mcpctl.py off --no-reload # 只对调文件,不调用重载
```
它读的正是 `mcpcfg.txt` 那三行:
```
自动 ← 第1行:模式
C:\...\opencode.json.1 ← 第2行:含 MCP 的原始配置
C:\...\opencode.json ← 第3行:删了 MCP 的安静配置
```
**状态判定只看活文件**(文件名为 `opencode.json` 的那份)里 `mcp.servers` 是否非空,
所以对调之后第2/3行的身份怎么漂移都不影响判断——这与 TUI 插件、server 插件完全一致。
模式语义(三个入口共用同一套规则):
| 模式 | `/mcpon` `/mcpoff` | 大模型(server 插件) | 本工具 `mcpctl.py` |
| --- | --- | --- | --- |
| 自动 | 可用(视为手动切换一次) | 可用 | 可用 |
| 手动 | **仅此可用** | 拒绝 | 可用(等价于用户输入命令) |
| 关闭 | 不注册 | 拒绝 | 拒绝(可用 `--ignore-mode` 强行) |
热重载打的是 `POST /api/location/reload`,会重建 location 并重读磁盘配置,MCP 立即生效;
失败时配置文件已经对调完成,重启 OC 一样生效。
---
## 三、连不上 OC 怎么办
1. 确认 WSL 里的 OC 起了:`opencode serve --port 4096`
2. 端口不对就加 `--port`,或设环境变量 `OC_PORT`
3. 提示 `鉴权失败 (401)` → 用 `--password` 传 `opencode.json` 里 `server.password` 的值
4. `localhost` 不通 → 工具会自动 `wsl hostname -I` 取 WSL 真实 IP 再试
5. 全都失败 → `list` 会自动降级为**直接扫目录**,照样能看到 skill 列表
---
## 四、和另外两个仓库的关系
本目录是独立 git 仓库,**分支名 = 文件夹名** `wsl-round-mcp`,远程与主仓库共用同一个根:
| 文件夹 | 分支 | 内容 |
| --- | --- | --- |
| `oc_session/`(主仓库) | `main` | TUI 插件 `oc-session-all`(`/sessions-all`、`/mcpon`、`/mcpoff`) |
| `oc_session/ai-mcp/` | `ai-mcp` | server 插件 `ai-mcp.ts`,给大模型用的 `mcp_switch` 工具 |
| `oc_session/wsl-round-mcp/` | `wsl-round-mcp` | 本目录:skill 互通 CLI + MCP 命令行开关 |
远程:`https://git.ypshidifu.cn/youpaishidifu/oc_session.git`
+15
View File
@@ -0,0 +1,15 @@
@echo off
chcp 65001 >nul
title MCP off (swap config + reload)
set "HERE=%~dp0"
where python >nul 2>nul
if errorlevel 1 (
echo [ERROR] Python not found in PATH.
pause
exit /b 1
)
python "%HERE%mcpctl.py" off
echo.
pause
+15
View File
@@ -0,0 +1,15 @@
@echo off
chcp 65001 >nul
title MCP on (swap config + reload)
set "HERE=%~dp0"
where python >nul 2>nul
if errorlevel 1 (
echo [ERROR] Python not found in PATH.
pause
exit /b 1
)
python "%HERE%mcpctl.py" on
echo.
pause
+15
View File
@@ -0,0 +1,15 @@
@echo off
chcp 65001 >nul
title MCP status
set "HERE=%~dp0"
where python >nul 2>nul
if errorlevel 1 (
echo [ERROR] Python not found in PATH.
pause
exit /b 1
)
python "%HERE%mcpctl.py" status
echo.
pause
+388
View File
@@ -0,0 +1,388 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
mcpctl —— 命令行版的 MCP 配置对调开关(与 TUI 插件 / server 插件同一套逻辑)
做什么:
status 查看当前模式与 MCP 开/关状态(只读,不改任何文件)
on 把配置对调为「含 MCP」的原始版,并调用 reload 热重载
off 把配置对调为「不含 MCP」的安静版,并调用 reload 热重载
mode 查看或修改 mcpcfg.txt 第1行(自动 / 手动 / 关闭)
模式语义(与插件完全一致):
自动 大模型可通过 server 插件切换;/mcpon /mcpoff 与本工具视为手动切换一次
手动 仅允许 /mcpon /mcpoff 与本工具(用户意图),大模型的切换请求会被拒绝
关闭 功能整体禁用,on/off 都会被拒绝
状态判定只看活文件(basename 为 opencode.json 的那份)里 mcp.servers 是否非空,
与 mcpcfg.txt 第2/3行的行语义无关,因此对调之后身份漂移也不影响判断。
"""
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 解析
def read_cfg(path: Path) -> dict:
if not path.exists():
return {
"mode": "off",
"error": f"缺少配置文件:{path}(配置缺失时功能视为关闭)",
}
try:
lines = path.read_text(encoding="utf-8", errors="replace").splitlines()
except Exception as exc:
return {"mode": "off", "error": f"读取 {path} 失败:{exc}"}
key = (lines[0] if lines else "").strip().lower()
mode = LABEL_MODE.get(key, "auto" if key == "" else None) or "auto"
on_path = (lines[1] if len(lines) > 1 else "").strip()
quiet_path = (lines[2] if len(lines) > 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,
"error": None,
}
if mode != "off" and not active:
result["error"] = "mcpcfg.txt 第2/3行必须有一个是 opencode.json 的完整路径"
elif mode != "off" and (not Path(on_path).exists() or not Path(quiet_path).exists()):
result["error"] = "mcpcfg.txt 第2/3行指向的文件不存在(请检查路径)"
return result
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 ""
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 urllib.request.urlopen(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 urllib.request.urlopen(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 第1行)")
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(f"拒绝:MCP 开关功能已关闭(mcpcfg.txt 第1行=关闭)。")
print("如需使用,请把第1行改成「自动」或「手动」。")
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 第2/3行路径。")
return 4
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
lines = path.read_text(encoding="utf-8", errors="replace").splitlines()
if not args.value:
cur = (lines[0] if lines else "").strip()
print(f"当前第1行:{cur or '(空,默认自动)'}")
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
while len(lines) < 3:
lines.append("")
lines[0] = LABEL_MODE[key]
path.write_text("\n".join(lines) + "\n", encoding="utf-8")
print(f"已把第1行设为:{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="忽略第1行的「关闭」限制(慎用)")
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 第1行的模式")
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())
+629
View File
@@ -0,0 +1,629 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
ocskill —— 打通本机与 WSL OpenCode 的 skill 互通
做什么:
list 运行时连接 WSL 里的 OpenCode,返回它的 skill 列表
push 把一个新的 skill 文件(压缩包 / 目录 / 单个 SKILL.md / base64)
先解码并校验是不是标准 skill 格式,然后写进 OC 的 skill 目录
(压缩包会额外做一次解压;OC 自带目录 watch,写完即热加载)
pull 反向操作:把 WSL 里的某个 skill 拷回本机
dirs 显示两端 skill 目录位置
零依赖:只用 Python 标准库(urllib / zipfile / shutil),无需 pip install。
"""
import argparse
import base64
import binascii
import json
import os
import re
import shutil
import subprocess
import sys
import tempfile
import urllib.error
import urllib.request
import zipfile
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()
# ---------------------------------------------------------------- 常量与默认值
DEFAULT_DISTRO = os.environ.get("OC_SKILL_DISTRO", "Ubuntu-24.04")
DEFAULT_PORT = int(os.environ.get("OC_PORT", "4096"))
SKILL_MD = "SKILL.md"
# 标准 skill 目录:OC 会 watch 每个 config entry 下的 skill/ 与 skills/
WINDOWS_WSL_ROOT = r"\\wsl.localhost\{distro}\root\.config\opencode\skills"
LINUX_SKILL_ROOT = "/root/.config/opencode/skills"
WIN_LOCAL_ROOT = str(Path.home() / ".config" / "opencode" / "skills")
def is_wsl_inside() -> bool:
"""当前是否已经跑在 WSL/Linux 里。"""
if os.name != "posix":
return False
if sys.platform.startswith("linux"):
try:
with open("/proc/version", "r", encoding="utf-8", errors="ignore") as fh:
if "microsoft" in fh.read().lower():
return True
except Exception:
pass
return True # 纯 Linux 也走这个分支
return False
def wsl_skill_root(distro: str) -> str:
"""WSL 侧 skill 目录(在 Windows 下用 UNC 路径直接读写,无需进 WSL)。"""
if is_wsl_inside():
return LINUX_SKILL_ROOT
return WINDOWS_WSL_ROOT.format(distro=distro)
# ---------------------------------------------------------------- OC 连接
def _read_server_password(config_dir: Path) -> str:
"""从 opencode.json 读 server.password(WSL 端配置里有这一项)。"""
cfg = config_dir / "opencode.json"
try:
data = json.loads(cfg.read_text(encoding="utf-8"))
pwd = (data.get("server") or {}).get("password")
return str(pwd) if pwd else ""
except Exception:
return ""
def _candidate_urls(explicit: str | None, distro: str, port: int) -> list[str]:
if explicit:
return [explicit.rstrip("/")]
urls = [
f"http://localhost:{port}",
f"http://127.0.0.1:{port}",
]
# Windows 下再试 WSL 的真实 IP(localhost 转发失效时兜底)
if os.name == "nt":
ip = _wsl_ip(distro)
if ip:
urls.append(f"http://{ip}:{port}")
seen, out = set(), []
for u in urls:
if u not in seen:
seen.add(u)
out.append(u)
return out
def _wsl_ip(distro: str) -> str:
try:
out = subprocess.run(
["wsl.exe", "-d", distro, "--", "bash", "-lc", "hostname -I"],
capture_output=True,
timeout=15,
)
text = (out.stdout or b"").decode("utf-8", errors="ignore")
for tok in text.split():
if re.fullmatch(r"\d{1,3}(\.\d{1,3}){3}", tok):
return tok
except Exception:
pass
return ""
def _build_request(url: str, password: str, method: str = "GET") -> urllib.request.Request:
req = urllib.request.Request(url, method=method)
req.add_header("accept", "application/json")
if password:
# OC 支持 Bearer 与 Basic 两种,先试 Bearer(401 时调用方会换 Basic 重试)
req.add_header("authorization", f"Bearer {password}")
return req
def _try_fetch(url: str, password: str, timeout: float) -> tuple[int, bytes]:
"""返回 (status, body)。401 时自动改用 Basic 重试一次。"""
req = _build_request(url, password)
try:
with urllib.request.urlopen(req, timeout=timeout) as resp:
return resp.status, resp.read()
except urllib.error.HTTPError as exc:
if exc.code == 401 and password:
raw = base64.b64encode(f":{password}".encode("utf-8")).decode("ascii")
req2 = _build_request(url, "")
req2.add_header("authorization", f"Basic {raw}")
try:
with urllib.request.urlopen(req2, timeout=timeout) as resp:
return resp.status, resp.read()
except urllib.error.HTTPError as exc2:
return exc2.code, exc2.read()
return exc.code, (exc.read() if hasattr(exc, "read") else b"")
except Exception:
return 0, b""
def connect_oc(args) -> tuple[str | None, object]:
"""
连接 OC,返回 (base_url, payload)。
依次尝试 /api/skill 与 /skill 两个已知端点;全失败则 payload=None。
"""
password = args.password or _read_server_password(Path(args.remote_config_dir))
last_err = ""
for base in _candidate_urls(args.url, args.distro, args.port):
for path in ("/api/skill", "/skill"):
code, body = _try_fetch(base + path, password, args.timeout)
if code == 200:
try:
return base, json.loads(body.decode("utf-8", errors="replace"))
except Exception as exc:
last_err = f"{base}{path} 返回非 JSON:{exc}"
elif code in (401, 403):
last_err = f"{base}{path} 鉴权失败(HTTP {code}),请用 --password 指定 server.password"
elif code:
last_err = f"{base}{path} 返回 HTTP {code}"
else:
last_err = f"{base} 连接失败(OC 可能没在运行,试试 opencode serve --port {args.port})"
return None, None if not last_err else last_err
# ---------------------------------------------------------------- skill 解析
def extract_skills(payload) -> list[dict]:
"""容错解析 skill 列表:兼容数组 / {skills:[...]} / {data:{...}} 等形态。"""
items = None
if isinstance(payload, list):
items = payload
elif isinstance(payload, dict):
for key in ("skills", "items", "data", "result", "list"):
val = payload.get(key)
if isinstance(val, list):
items = val
break
if isinstance(val, dict): # data: { skills: [...] }
for inner in val.values():
if isinstance(inner, list):
items = inner
break
if items is not None:
break
if items is None:
items = next((v for v in payload.values() if isinstance(v, list)), [])
if items is None:
return []
out = []
for it in items:
if isinstance(it, str):
out.append({"name": it, "description": ""})
elif isinstance(it, dict):
name = it.get("name") or it.get("id") or it.get("title") or ""
desc = it.get("description") or it.get("summary") or ""
out.append({"name": str(name), "description": str(desc)})
return [x for x in out if x["name"]]
def scan_skill_dir(root: str) -> list[dict]:
"""API 不可用时的兜底:直接扫目录。"""
out = []
base = Path(root)
if not base.is_dir():
return out
for child in sorted(base.iterdir()):
if not child.is_dir() or child.name.startswith("."):
continue
md = child / SKILL_MD
desc = ""
if md.is_file():
desc = _frontmatter(md.read_text(encoding="utf-8", errors="replace")).get("description", "")
out.append({"name": child.name, "description": desc})
return out
BLOCK_MARKERS = {">", "|", ">-", "|-", ">+", "|+", ">", "|"}
def _has_cjk(text: str) -> bool:
"""是否含中日韩字符(决定块标量折行时要不要补空格)。"""
return any("\u4e00" <= ch <= "\u9fff" or "\u3000" <= ch <= "\u303f" for ch in text)
def _frontmatter(text: str) -> dict:
"""
解析 SKILL.md 顶部的 YAML frontmatter(只取扁平 key: value,不引第三方库)。
会正确处理块标量(description: > / |)——否则描述会显示成一个 ">"。
"""
if not text.startswith("---"):
return {}
lines = text.splitlines()
if not lines or lines[0].strip() != "---":
return {}
data: dict = {}
pending_key: str | None = None # 正在收集块标量的 key
block_lines: list[str] = []
block_indent: int | None = None
def flush() -> None:
nonlocal pending_key, block_lines, block_indent
if pending_key is not None:
parts = [x.strip() for x in block_lines if x.strip()]
# 中文行之间不补空格,英文单词之间才补
joiner = "" if _has_cjk("".join(parts)) else " "
data[pending_key] = joiner.join(parts)
pending_key, block_lines, block_indent = None, [], None
for line in lines[1:]:
if line.strip() in ("---", "..."):
flush()
break
if pending_key is not None:
# 块标量的续行:空行或更深缩进的都算
if line.strip() == "" or (line[:1] in (" ", "\t")):
block_lines.append(line)
continue
flush()
m = re.match(r"^([A-Za-z0-9_-]+)\s*:\s*(.*)$", line)
if not m:
continue
key, val = m.group(1), m.group(2).strip()
if val in BLOCK_MARKERS or val.rstrip() in BLOCK_MARKERS:
pending_key = key
block_lines = []
block_indent = None
continue
data[key] = val.strip("'\"")
flush()
return data
# ---------------------------------------------------------------- 标准 skill 校验
NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$")
def find_skill_md(folder: Path, depth: int = 0) -> Path | None:
"""在目录里找 SKILL.md(兼容压缩包自带一层父目录的情况)。"""
direct = folder / SKILL_MD
if direct.is_file():
return direct
if depth >= 3:
return None
for child in sorted(folder.iterdir()):
if child.is_dir() and not child.name.startswith("."):
found = find_skill_md(child, depth + 1)
if found:
return found
return None
def validate_skill(folder: Path) -> tuple[bool, str, dict]:
"""确认是不是标准 skill 格式:有 SKILL.md + YAML frontmatter + name。"""
md = find_skill_md(folder)
if not md:
return False, f"不是标准 skill 格式:找不到 {SKILL_MD}", {}
text = md.read_text(encoding="utf-8", errors="replace")
if not text.startswith("---"):
return False, f"{SKILL_MD} 缺少 YAML frontmatter(文件必须以 --- 开头)", {}
meta = _frontmatter(text)
if not meta:
return False, f"{SKILL_MD} 的 frontmatter 为空或无法解析", {}
name = meta.get("name", "")
if not name:
return False, f"{SKILL_MD} frontmatter 缺少必填的 name 字段", {}
if not NAME_RE.match(name):
return False, f"非法 skill 名称「{name}」(只允许字母/数字/点/下划线/连字符,且以字母或数字开头)", meta
if not meta.get("description"):
return False, f"{SKILL_MD} frontmatter 缺少 description 字段(大模型靠它判断何时调用)", meta
return True, "", meta
def decode_input(src: str, work: Path, b64: bool) -> Path | None:
"""
把用户传进来的输入统一解码成一个本地路径:
- base64 字符串 -> 解出 zip 或 markdown
- 其余情况原样返回(文件/目录交给后续步骤处理)
"""
if not b64:
return None
raw = src.strip()
try:
decoded = base64.b64decode(raw, validate=True)
except (binascii.Error, ValueError) as exc:
raise SystemExit(f"base64 解码失败:{exc}")
if decoded[:2] == b"PK": # zipfile magic
target = work / "decoded.zip"
target.write_bytes(decoded)
return target
target = work / "decoded.md"
target.write_bytes(decoded)
return target
def stage_candidate(src: str, work: Path, b64: bool) -> Path:
"""归一化输入 -> 返回一个包含 SKILL.md 的候选目录。"""
resolved = decode_input(src, work, b64)
path = Path(resolved) if resolved else Path(os.path.expandvars(os.path.expanduser(src)))
if not path.exists():
# 允许直接传一整段 base64(自动嗅探)
stripped = src.strip()
if _looks_like_b64(stripped):
sniff = work / "sniffed.bin"
try:
sniff.write_bytes(base64.b64decode(stripped, validate=True))
path = sniff
except Exception as exc:
raise SystemExit(f"输入既不是存在的路径,也不是合法 base64:{exc}")
else:
raise SystemExit(f"路径不存在:{path}")
if path.is_dir():
return path
# --- 到这里是单个文件:按魔数 / 后缀 / 内容三重嗅探 ---
if zipfile.is_zipfile(path):
return _unzip(path, work)
suffix = path.suffix.lower()
if suffix in (".7z", ".rar", ".tar", ".gz", ".tgz"):
raise SystemExit(f"暂不支持该压缩格式:{path.suffix}(请先转成 .zip)")
# 魔数兜底(无后缀或后缀写错的 zip)
try:
if path.read_bytes()[:2] == b"PK":
return _unzip(path, work)
except OSError:
pass
text = path.read_text(encoding="utf-8", errors="replace")
compact = "".join(text.split())
is_b64_text = (
len(compact) >= 64
and len(compact) % 4 == 0
and re.fullmatch(r"[A-Za-z0-9+/=]+", compact) is not None
)
# base64:显式后缀,或者内容本身就是一段 base64
if suffix in (".b64", ".base64", ".enc") or is_b64_text:
try:
decoded = base64.b64decode(compact, validate=True)
except Exception:
decoded = b""
if decoded[:2] == b"PK":
zipped = work / "sniffed.zip"
zipped.write_bytes(decoded)
return _unzip(zipped, work)
if decoded:
md_file = work / "decoded.md"
md_file.write_bytes(decoded)
text = decoded.decode("utf-8", "replace")
# markdown:frontmatter 开头,或者至少是个标题
stripped = text.lstrip()
if stripped.startswith("---") or stripped.startswith("#"):
meta = _frontmatter(text)
name = meta.get("name") or path.stem
safe = re.sub(r"[^\w.-]+", "-", name).strip("-") or path.stem
folder = work / safe
folder.mkdir(parents=True, exist_ok=True)
(folder / SKILL_MD).write_text(text, encoding="utf-8")
return folder
raise SystemExit(f"无法识别的输入:{path}(支持 .zip / 目录 / SKILL.md / base64 字符串)")
def _unzip(path: Path, work: Path) -> Path:
dest = work / ("unzipped_" + (path.stem or "archive"))
dest.mkdir(parents=True, exist_ok=True)
with zipfile.ZipFile(path) as zf:
_safe_extract(zf, dest)
return dest
def _looks_like_b64(s: str) -> bool:
if len(s) < 64 or len(s) % 4 != 0:
return False
if not re.fullmatch(r"[A-Za-z0-9+/=\s]+", s):
return False
return True
def _safe_extract(zf: zipfile.ZipFile, dest: Path) -> None:
"""解压时防目录穿越(zip slip)。"""
for member in zf.infolist():
target = (dest / member.filename).resolve()
if not str(target).startswith(str(dest.resolve())):
raise SystemExit(f"压缩包内含非法路径,已拒绝解压:{member.filename}")
zf.extractall(dest)
# ---------------------------------------------------------------- 子命令
def cmd_list(args) -> int:
base, payload = connect_oc(args)
root = wsl_skill_root(args.distro)
if base:
skills = extract_skills(payload)
source = f"OpenCode API({base})"
else:
skills = scan_skill_dir(root)
reason = payload if isinstance(payload, str) else "未知原因"
source = f"目录扫描({root})—— API 不可用:{reason}"
print(f"skill 列表来源:{source}")
if not skills:
print("(没有找到任何 skill)")
return 1
width = max(len(s["name"]) for s in skills)
width = max(width, 4)
print(f"{'名称'.ljust(width)} 描述")
print("-" * (width + 40))
for s in skills:
desc = s["description"].replace("\n", " ")
if len(desc) > 70:
desc = desc[:67] + "..."
print(f"{s['name'].ljust(width)} {desc}")
print(f"\n共 {len(skills)} 个 skill。WSL skill 目录:{root}")
return 0
def cmd_push(args) -> int:
root = wsl_skill_root(args.distro)
root_path = Path(root)
if not root_path.is_dir():
if args.dry_run:
print(f"(dry-run)目标目录尚不存在:{root}")
else:
raise SystemExit(
f"找不到 WSL 的 skill 目录:{root}\n"
f"请确认 WSL 发行版名称(当前 {args.distro}),可用 --distro 指定。"
)
with tempfile.TemporaryDirectory(prefix="ocskill_") as tmp:
work = Path(tmp)
candidate = stage_candidate(args.source, work, args.base64)
ok, err, meta = validate_skill(candidate)
if not ok:
print(f"校验失败:{err}")
return 2
name = meta["name"]
desc = meta.get("description", "")
print(f"校验通过:标准 skill 格式")
print(f" 名称:{name}")
print(f" 描述:{desc[:100]}{'...' if len(desc) > 100 else ''}")
# SKILL.md 可能藏在压缩包的一层父目录里,定位到真正含 SKILL.md 的那层
md = find_skill_md(candidate)
real_dir = md.parent if md else candidate
dest = root_path / name
if args.dry_run:
print(f"dry-run:不会写入 {dest}")
return 0
if dest.exists():
if not args.force:
raise SystemExit(
f"目标已存在同名 skill:{dest}\n"
f"加 --force 覆盖(覆盖前会自动备份为 .bak-<时间戳>)。"
)
backup = dest.with_name(dest.name + f".bak-{_stamp()}")
shutil.move(str(dest), str(backup))
print(f"已备份旧版本 -> {backup}")
dest.parent.mkdir(parents=True, exist_ok=True)
shutil.copytree(real_dir, dest, dirs_exist_ok=True)
print(f"已写入:{dest}")
print("OC 会 watch skill 目录(约 100ms 去抖后自动重载),无需重启。")
return 0
def cmd_pull(args) -> int:
root = wsl_skill_root(args.distro)
src = Path(root) / args.name
if not src.is_dir():
raise SystemExit(f"WSL 端不存在该 skill:{src}")
out_dir = Path(args.out or ".")
out_dir.mkdir(parents=True, exist_ok=True)
dest = out_dir / args.name
if dest.exists():
if not args.force:
raise SystemExit(f"本地已存在:{dest}(用 --force 覆盖)")
shutil.rmtree(dest)
shutil.copytree(src, dest)
print(f"已取回:{dest}")
return 0
def cmd_dirs(args) -> int:
print(f"WSL skill 目录(push 目标):{wsl_skill_root(args.distro)}")
print(f"本机 skill 目录(pull 默认落到当前目录):{WIN_LOCAL_ROOT}")
base, _ = connect_oc(args)
print(f"OpenCode API:{base or '未连接'}")
return 0
def _stamp() -> str:
import datetime
return datetime.datetime.now().strftime("%Y%m%d%H%M%S")
# ---------------------------------------------------------------- 入口
def build_parser() -> argparse.ArgumentParser:
p = argparse.ArgumentParser(
prog="ocskill",
description="打通本机与 WSL OpenCode 的 skill 互通(list / push / pull / dirs)",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog=(
"示例:\n"
" python ocskill.py list\n"
" python ocskill.py push .\\my-skill.zip\n"
" python ocskill.py push .\\my-skill\\SKILL.md\n"
" python ocskill.py push <base64字符串> --base64\n"
" python ocskill.py push .\\my-skill --dry-run\n"
" python ocskill.py pull java-expert --out .\\skills-backup\n"
),
)
p.add_argument("--distro", default=DEFAULT_DISTRO, help=f"WSL 发行版名(默认 {DEFAULT_DISTRO})")
p.add_argument("--url", default=None, help="显式指定 OC 地址,如 http://localhost:4096")
p.add_argument("--port", type=int, default=DEFAULT_PORT, help=f"OC 端口(默认 {DEFAULT_PORT})")
p.add_argument("--timeout", type=float, default=4.0, help="HTTP 超时秒数")
p.add_argument("--password", default="", help="OC server.password(默认自动从 opencode.json 读)")
p.add_argument(
"--remote-config-dir",
default="/root/.config/opencode",
help="读取 server.password 用的配置目录(Windows 侧用 WSL 内路径时才需要改)",
)
sub = p.add_subparsers(dest="cmd")
sp = sub.add_parser("list", help="连接 WSL 的 OC,返回 skill 列表")
sp.set_defaults(func=cmd_list)
sp = sub.add_parser("push", help="推送 skill(zip / 目录 / SKILL.md / base64)到 WSL OC")
sp.add_argument("source", help="skill 文件路径、目录、zip 或 base64 字符串")
sp.add_argument("--base64", action="store_true", help="把 source 当作 base64 内容解码")
sp.add_argument("--force", action="store_true", help="覆盖同名 skill(先自动备份)")
sp.add_argument("--dry-run", action="store_true", help="只校验格式,不写入")
sp.set_defaults(func=cmd_push)
sp = sub.add_parser("pull", help="把 WSL 里的 skill 拷回本机")
sp.add_argument("name", help="skill 名称")
sp.add_argument("--out", default=None, help="输出目录(默认当前目录)")
sp.add_argument("--force", action="store_true", help="覆盖本地同名目录")
sp.set_defaults(func=cmd_pull)
sp = sub.add_parser("dirs", help="显示两端 skill 目录与 API 连接状态")
sp.set_defaults(func=cmd_dirs)
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())
+16
View File
@@ -0,0 +1,16 @@
@echo off
chcp 65001 >nul
title List WSL OpenCode skills
set "HERE=%~dp0"
where python >nul 2>nul
if errorlevel 1 (
echo [ERROR] Python not found in PATH.
echo Install Python 3 and tick "Add to PATH", then run this again.
pause
exit /b 1
)
python "%HERE%ocskill.py" list
echo.
pause
+23
View File
@@ -0,0 +1,23 @@
@echo off
chcp 65001 >nul
title Pull a skill from WSL OpenCode
set "HERE=%~dp0"
where python >nul 2>nul
if errorlevel 1 (
echo [ERROR] Python not found in PATH.
pause
exit /b 1
)
set /p "NAME=Skill name: "
if "%NAME%"=="" (
echo Nothing entered, cancelled.
pause
exit /b 0
)
set "OUT=%HERE%skills-backup"
python "%HERE%ocskill.py" pull "%NAME%" --out "%OUT%"
echo.
pause
+25
View File
@@ -0,0 +1,25 @@
@echo off
chcp 65001 >nul
title Push a skill to WSL OpenCode
set "HERE=%~dp0"
where python >nul 2>nul
if errorlevel 1 (
echo [ERROR] Python not found in PATH.
pause
exit /b 1
)
echo Accepts: a .zip file, a skill folder, a SKILL.md file, or base64 text.
echo Example: D:\skills\my-skill.zip
echo.
set /p "SRC=Path or content: "
if "%SRC%"=="" (
echo Nothing entered, cancelled.
pause
exit /b 0
)
python "%HERE%ocskill.py" push "%SRC%"
echo.
pause