#!/usr/bin/env python3 """Hermes Agent Dashboard — Full Control Panel (no CLI dependency) See SKILL.md for architecture and deployment guide. Source: ~/docker/hermes-dashboard/app.py """ import os, json, subprocess, re, yaml, glob, sqlite3 from datetime import datetime, timezone from functools import wraps from flask import Flask, render_template, request, jsonify, redirect, url_for, session from werkzeug.security import generate_password_hash, check_password_hash app = Flask(__name__) app.secret_key = os.environ.get("DASHBOARD_SECRET_KEY", os.urandom(24).hex()) HERMES_HOME = os.environ.get("HERMES_HOME", os.path.expanduser("~/.hermes")) CONFIG_PATH = os.path.join(HERMES_HOME, "config.yaml") LOGS_DIR = os.path.join(HERMES_HOME, "logs") SKILLS_DIR = os.path.join(HERMES_HOME, "skills") PASSWORD_HASH = os.environ.get("DASHBOARD_PASSWORD_HASH") if not PASSWORD_HASH: pw = os.environ.get("DASHBOARD_PASSWORD", "hermes") PASSWORD_HASH = generate_password_hash(pw) # ---- Auth ---- def login_required(f): @wraps(f) def decorated(*args, **kwargs): if not session.get("authenticated"): return redirect(url_for("login")) return f(*args, **kwargs) return decorated @app.route("/login", methods=["GET", "POST"]) def login(): if request.method == "POST": if check_password_hash(PASSWORD_HASH, request.form.get("password", "")): session["authenticated"] = True return redirect(url_for("index")) return render_template("login.html", error="Invalid password") return render_template("login.html") @app.route("/logout") def logout(): session.clear() return redirect(url_for("login")) # ---- Helpers ---- def send_command(cmd, timeout=10): try: result = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout, shell=True) return result.stdout + result.stderr except: return "Command failed" def get_status(): status = {"gateway": "unknown", "api_server": "unknown", "cron_jobs": 0, "disk_usage": "unknown"} gw = send_command("systemctl --user is-active hermes-gateway 2>/dev/null || echo inactive", timeout=3) status["gateway"] = "running" if "active" in gw else "stopped" api = send_command("curl -s -o /dev/null -w '%{http_code}' http://localhost:8642/health 2>/dev/null || echo 000", timeout=3) status["api_server"] = "running" if api.strip() == "200" else "stopped" # Count cron jobs from jobs.json jobs_path = os.path.join(HERMES_HOME, "cron", "jobs.json") if os.path.exists(jobs_path): try: with open(jobs_path) as f: data = json.load(f) status["cron_jobs"] = len(data.get("jobs", [])) except: status["cron_jobs"] = "?" du = send_command(f"df -h {HERMES_HOME} | tail -1 | awk '{{print $5}}'", timeout=3) status["disk_usage"] = du.strip() return status def read_config(): try: with open(CONFIG_PATH) as f: return yaml.safe_load(f) except: return {} def write_config(data): try: with open(CONFIG_PATH, 'w') as f: yaml.dump(data, f, default_flow_style=False, sort_keys=False) return True except: return False def get_cron_jobs(): """Read cron jobs from jobs.json""" jobs_path = os.path.join(HERMES_HOME, "cron", "jobs.json") if not os.path.exists(jobs_path): return [] try: with open(jobs_path) as f: data = json.load(f) return data.get("jobs", []) except: return [] def get_skills_list(): skills = [] if os.path.exists(SKILLS_DIR): for root, dirs, files in os.walk(SKILLS_DIR): if "SKILL.md" in files: rel = os.path.relpath(root, SKILLS_DIR) skill_path = os.path.join(root, "SKILL.md") try: with open(skill_path) as f: content = f.read() name = desc = "" for line in content.split('\n'): if line.startswith("name:"): name = line.split(":",1)[1].strip().strip('"') elif line.startswith("description:"): desc = line.split(":",1)[1].strip().strip('"') if not name: name = rel except: name, desc = rel, "" skills.append({"name": name, "path": rel, "description": desc}) return sorted(skills, key=lambda s: s["name"].lower()) # ---- Routes ---- @app.route("/") @login_required def index(): return render_template("index.html", status=get_status()) @app.route("/config") @login_required def config_view(): cfg = read_config() def mask(d, path=""): if isinstance(d, dict): return {k: mask(v, f"{path}.{k}") for k,v in d.items()} if isinstance(d, list): return [mask(v, path) for v in d] if isinstance(d, str) and any(x in path.lower() for x in ["api_key","token","secret","password"]): if len(d) > 8: return d[:4]+"..."+d[-4:] return d return render_template("config.html", config=mask(cfg), raw=yaml.dump(cfg, default_flow_style=False, sort_keys=False)) @app.route("/config/edit", methods=["POST"]) @login_required def config_edit(): try: data = yaml.safe_load(request.form.get("yaml_content", "")) ok = write_config(data) return jsonify({"success": ok, "message": "Config updated. Restart gateway." if ok else "Failed"}) except yaml.YAMLError as e: return jsonify({"success": False, "message": f"YAML error: {e}"}) @app.route("/config/quick", methods=["POST"]) @login_required def config_quick(): key = request.form.get("key") value = request.form.get("value") if not key: return jsonify({"success": False, "message": "Key required"}) cfg = read_config() keys = key.split(".") target = cfg for k in keys[:-1]: if k not in target: target[k] = {} target = target[k] try: val = int(value) except ValueError: val = value.lower() == "true" if value.lower() in ("true","false") else value target[keys[-1]] = val return jsonify({"success": write_config(cfg), "message": f"Set {key} = {value}"}) @app.route("/models") @login_required def models_view(): cfg = read_config() return render_template("models.html", current_model=cfg.get("model",{}).get("default","unknown"), current_provider=cfg.get("model",{}).get("provider","unknown"), delegation_model=cfg.get("delegation",{}).get("model","")) @app.route("/models/switch", methods=["POST"]) @login_required def models_switch(): cfg = read_config() provider = request.form.get("provider") model = request.form.get("model") if provider: cfg.setdefault("model",{})["provider"] = provider if model: cfg.setdefault("model",{})["default"] = model return jsonify({"success": write_config(cfg), "message": "Model updated. Restart gateway."}) @app.route("/tools") @login_required def tools_view(): cfg = read_config() toolsets = cfg.get("toolsets", {}) all_known = ["web","terminal","file","search","browser","vision","image_gen","video","tts", "skills","memory","session_search","delegation","cronjob","clarify","messaging", "todo","spotify","homeassistant","discord","code_execution","kanban"] if isinstance(toolsets, list): enabled_names = set(toolsets) toolsets = {name: {"enabled": name in enabled_names} for name in all_known} enabled = [] for t in all_known: ts_cfg = toolsets.get(t, {}) is_enabled = ts_cfg.get("enabled", True) if isinstance(ts_cfg, dict) else True enabled.append({"name": t, "enabled": is_enabled}) return render_template("tools.html", tools=enabled) @app.route("/tools/toggle", methods=["POST"]) @login_required def tools_toggle(): name = request.form.get("name") action = request.form.get("action") cfg = read_config() cfg.setdefault("toolsets", {}).setdefault(name, {}) cfg["toolsets"][name]["enabled"] = (action == "enable") return jsonify({"success": write_config(cfg), "message": f"Toolset '{name}' {action}d"}) @app.route("/skills") @login_required def skills_view(): return render_template("skills.html", skills=get_skills_list()) @app.route("/cron") @login_required def cron_view(): return render_template("cron.html", jobs=get_cron_jobs()) @app.route("/cron/toggle", methods=["POST"]) @login_required def cron_toggle(): job_id = request.form.get("job_id") enabled = request.form.get("enabled") == "true" jobs_path = os.path.join(HERMES_HOME, "cron", "jobs.json") try: with open(jobs_path) as f: data = json.load(f) for job in data.get("jobs", []): if job["id"] == job_id: job["enabled"] = enabled break with open(jobs_path, 'w') as f: json.dump(data, f, indent=2) return jsonify({"success": True, "message": f"Job {'enabled' if enabled else 'paused'}"}) except Exception as e: return jsonify({"success": False, "message": str(e)}) @app.route("/cron/run", methods=["POST"]) @login_required def cron_run(): job_id = request.form.get("job_id") flag = os.path.join(HERMES_HOME, "cron", f".trigger_{job_id}") try: os.makedirs(os.path.dirname(flag), exist_ok=True) with open(flag, 'w') as f: f.write(datetime.now(timezone.utc).isoformat()) return jsonify({"success": True, "message": "Job triggered. Should run within 30s."}) except Exception as e: return jsonify({"success": False, "message": str(e)}) @app.route("/logs") @login_required def logs_view(): log_type = request.args.get("type", "gateway") lines = request.args.get("lines", "100") log_files = {"gateway": os.path.join(LOGS_DIR,"gateway.log"), "error": os.path.join(LOGS_DIR,"error.log")} content = "" log_path = log_files.get(log_type) if log_path and os.path.exists(log_path): content = send_command(f"tail -n {lines} {log_path}", timeout=5) else: available = [k for k,v in log_files.items() if os.path.exists(v)] content = f"Available: {', '.join(available)}" if available else "No log files found." return render_template("logs.html", content=content, log_type=log_type, lines=lines) @app.route("/api/status") @login_required def api_status(): return jsonify(get_status()) @app.route("/api/logs/tail") @login_required def api_logs_tail(): log_type = request.args.get("type", "gateway") log_files = {"gateway": os.path.join(LOGS_DIR,"gateway.log"), "error": os.path.join(LOGS_DIR,"error.log")} log_path = log_files.get(log_type) if log_path and os.path.exists(log_path): return jsonify({"content": send_command(f"tail -n 50 {log_path}", timeout=3)}) return jsonify({"content": "Log file not found"}) @app.route("/health") def health(): return jsonify({"status": "ok", "timestamp": datetime.now(timezone.utc).isoformat()}) if __name__ == "__main__": port = int(os.environ.get("DASHBOARD_PORT", 5000)) app.run(host="0.0.0.0", port=port, debug=False)