#!/usr/bin/env python3 """苍耳技能中心:技能注册表 + 提示词加载 + 工具执行 用法:python3 skill_center.py [list|load |run [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)