Hitonabi
80 lines
3.4 KiB
Python
80 lines
3.4 KiB
Python
import os
|
|
import subprocess
|
|
import psutil
|
|
|
|
from fastapi import APIRouter
|
|
from pydantic import BaseModel
|
|
|
|
router = APIRouter(prefix="/api")
|
|
|
|
class RunSkillRequest(BaseModel):
|
|
skill_name: str
|
|
|
|
@router.get("/skills")
|
|
def list_skills():
|
|
"""Listet alle verfügbaren Skills aus dem deploy/skills Verzeichnis auf."""
|
|
# Pfad relativ zu dieser Datei (backend/routers/skills.py)
|
|
current_dir = os.path.dirname(os.path.abspath(__file__))
|
|
repo_root = os.path.dirname(os.path.dirname(current_dir))
|
|
skills_dir = os.path.join(repo_root, "deploy", "skills")
|
|
skills = []
|
|
|
|
# Check currently running hermes skill processes
|
|
running_skills = set()
|
|
for p in psutil.process_iter(['name', 'cmdline']):
|
|
try:
|
|
cmdline = p.info.get('cmdline')
|
|
if cmdline and any("hermes" in arg for arg in cmdline):
|
|
for arg in cmdline:
|
|
if "Führe den " in arg and " Skill aus" in arg:
|
|
# Extract skill name from "Führe den 'skill_name' Skill aus"
|
|
import re
|
|
match = re.search(r"Führe den '(.+?)' Skill aus", arg)
|
|
if match:
|
|
running_skills.add(match.group(1))
|
|
except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
|
|
pass
|
|
|
|
if os.path.exists(skills_dir) and os.path.isdir(skills_dir):
|
|
for entry in os.scandir(skills_dir):
|
|
if entry.is_dir():
|
|
# Suche nach SKILL.md für Metadaten
|
|
skill_md_path = os.path.join(entry.path, "SKILL.md")
|
|
description = ""
|
|
if os.path.exists(skill_md_path):
|
|
try:
|
|
with open(skill_md_path, "r", encoding="utf-8") as f:
|
|
# Lese erste Zeilen für Description (YAML Frontmatter)
|
|
lines = f.readlines()[:10]
|
|
for line in lines:
|
|
if "description:" in line.lower():
|
|
description = line.split(":", 1)[1].strip().strip('"\'')
|
|
break
|
|
elif "author:" in line.lower() and not description:
|
|
description = line.strip()
|
|
except Exception:
|
|
pass
|
|
|
|
skills.append({
|
|
"name": entry.name,
|
|
"description": description or "Keine Beschreibung verfügbar.",
|
|
"running": entry.name in running_skills
|
|
})
|
|
return {"skills": skills}
|
|
|
|
@router.post("/skills/run")
|
|
def run_skill(req: RunSkillRequest):
|
|
"""Startet einen autonomen Skill via Hermes im Hintergrund."""
|
|
hermes_bin = os.path.expanduser("~/.local/bin/hermes")
|
|
if not os.path.exists(hermes_bin):
|
|
# Fallback falls lokal (auf Windows) entwickelt wird
|
|
return {"ok": False, "err": "Hermes CLI nicht gefunden (~/.local/bin/hermes fehlt). Bist du lokal unterwegs?"}
|
|
|
|
# Entspricht: hermes "Führe den 'X' Skill aus"
|
|
cmd = [hermes_bin, f"Führe den '{req.skill_name}' Skill aus"]
|
|
try:
|
|
subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
return {"ok": True, "msg": f"Skill '{req.skill_name}' erfolgreich angestoßen."}
|
|
except Exception as e:
|
|
return {"ok": False, "err": str(e)}
|