cang-ying/skill/skill_center.py

77 lines
2.6 KiB
Python
Raw Normal View History

#!/usr/bin/env python3
"""苍耳技能中心:技能注册表 + 提示词加载 + 工具执行
用法python3 skill_center.py [list|load <id>|run <id> [args...]]"""
import json, os, subprocess, sys
SKILL_DIR = os.path.dirname(os.path.abspath(__file__))
INDEX = os.path.join(SKILL_DIR, "index.json")
def _index():
with open(INDEX, encoding="utf-8") as f:
return json.load(f)
def list_skills():
return _index().get("skills", [])
def load(skill_id):
"""返回技能定义 + 提示词全文prompt_file 存在时)。"""
for s in list_skills():
if s["id"] == skill_id:
data = dict(s)
pf = data.get("prompt_file")
if pf:
p = os.path.normpath(os.path.join(SKILL_DIR, pf))
try:
with open(p, encoding="utf-8") as f:
data["prompt"] = f.read()
except Exception as e:
data["prompt"] = ""
data["prompt_error"] = str(e)
return data
return None
def run(skill_id, args=None):
"""执行工具型技能,返回 (ok, output)。"""
s = load(skill_id)
if not s:
return False, "技能不存在: " + skill_id
if s.get("type") != "tool":
return False, f"技能「{s.get('name','')}」是提示词型,请注入提示词使用"
tool = s.get("tool")
if not tool:
return False, "技能无执行命令"
cmd = tool.split() + (args or [])
try:
r = subprocess.run(cmd, capture_output=True, text=True, timeout=600)
return r.returncode == 0, (r.stdout or r.stderr)[-3000:]
except subprocess.TimeoutExpired:
return False, "执行超时10分钟"
except Exception as e:
return False, str(e)
if __name__ == "__main__":
if len(sys.argv) < 2:
print(json.dumps(list_skills(), ensure_ascii=False, indent=2)); sys.exit(0)
cmd = sys.argv[1]
if cmd == "list":
print(json.dumps({"skills": list_skills()}, ensure_ascii=False, indent=2)); sys.exit(0)
elif cmd == "load":
s = load(sys.argv[2]) if len(sys.argv) > 2 else None
if s:
print(json.dumps({"name": s.get("name"), "type": s.get("type"),
"prompt_len": len(s.get("prompt", "")),
"tool": s.get("tool")}, ensure_ascii=False))
else:
print("技能不存在"); sys.exit(1)
elif cmd == "run":
ok, out = run(sys.argv[2], sys.argv[3:])
print(("OK\n" if ok else "FAIL\n") + out)
sys.exit(0 if ok else 1)
else:
print("未知命令: " + cmd); sys.exit(1)