#!/usr/bin/env python3 """ Agentic OS — FastAPI Backend Multi-agent orchestration server for opencode, Hermes, Gemini CLI """ import argparse import json import os import shutil import subprocess import tarfile import time import uuid from datetime import datetime, timezone from pathlib import Path from typing import Optional from fastapi import FastAPI, HTTPException, Query from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import HTMLResponse, JSONResponse, Response from fastapi.staticfiles import StaticFiles from pydantic import BaseModel app = FastAPI(title="Agentic OS", version="1.1.0") # Load OpenRouter API key from Hermes .env HERMES_ENV = Path.home() / ".hermes" / ".env" if HERMES_ENV.exists(): for line in HERMES_ENV.read_text().splitlines(): line = line.strip() if line and not line.startswith("#") and "=" in line: k, v = line.split("=", 1) if k == "OPENROUTER_API_KEY": os.environ[k] = v # last value wins (matches shell sourcing) # CORS for local dev app.add_middleware( CORSMiddleware, allow_origins=["http://127.0.0.1:8080", "http://localhost:8080"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) BASE_DIR = Path(__file__).parent.resolve() # ─── Models ─────────────────────────────────────────────────────── class BrainUpdate(BaseModel): content: str class SkillRunRequest(BaseModel): input: Optional[str] = "" agent: Optional[str] = "auto" class ScheduleJobRequest(BaseModel): name: str skill: str cron: str enabled: bool = True class SettingsUpdate(BaseModel): settings: dict class BackupRestoreRequest(BaseModel): file: str class ChatRequest(BaseModel): agent: str message: str # ─── Helper Functions ───────────────────────────────────────────── def read_file(path: Path): if not path.exists(): return "" return path.read_text(encoding="utf-8") def write_file(path: Path, content: str): path.write_text(content, encoding="utf-8") return True def list_dir(path: Path): if not path.exists(): return [] return sorted([p.name for p in path.iterdir() if not p.name.startswith(".")]) def get_timestamp(): return datetime.now(timezone.utc).isoformat() def append_audit(entry: dict): audit_file = BASE_DIR / "audit" / "audit.log" entry["timestamp"] = get_timestamp() entry["id"] = str(uuid.uuid4())[:8] with open(audit_file, "a") as f: f.write(json.dumps(entry) + "\n") # ─── Agent Discovery (instant filesystem checks) ──────────────────── def check_agent(name: str) -> dict: """Instant filesystem-based check. No subprocess needed.""" try: if name == "opencode": exists = shutil.which("opencode") is not None status = "online" if exists else "offline" elif name == "hermes": exists = shutil.which("hermes") is not None status = "online" if exists else "offline" elif name == "gemini": # Gemini has valid OAuth tokens logged in oauth = Path.home() / ".gemini" / "oauth_creds.json" exists = shutil.which("gemini") is not None logged_in = oauth.exists() and "ya29" in oauth.read_text() status = "online" if exists and logged_in else "offline" if not exists else "warning" else: status = "offline" except Exception: status = "offline" return {"name": name, "status": status} # ─── Routes: Status ─────────────────────────────────────────────── @app.get("/api/status") def get_status(): agents = [check_agent(a) for a in ["opencode", "hermes", "gemini"]] skills = list_dir(BASE_DIR / "skills") return { "status": "healthy", "agents": agents, "skills_count": len(skills), "uptime": time.time(), } # ─── Routes: Brain ──────────────────────────────────────────────── @app.get("/api/brain") def list_brain(): files = list_dir(BASE_DIR / "brain") brain_data = {} for f in files: path = BASE_DIR / "brain" / f brain_data[f] = read_file(path) return brain_data @app.get("/api/brain/{file_name}") def get_brain_file(file_name: str): path = BASE_DIR / "brain" / file_name if not path.exists() or path.is_dir(): raise HTTPException(404, "File not found") return {"name": file_name, "content": read_file(path)} @app.put("/api/brain/{file_name}") def update_brain_file(file_name: str, data: BrainUpdate): path = BASE_DIR / "brain" / file_name write_file(path, data.content) append_audit({"action": "brain_update", "file": file_name}) return {"status": "ok", "file": file_name} # ─── Routes: Skills ─────────────────────────────────────────────── @app.get("/api/skills") def list_skills(): skills = [] for d in sorted((BASE_DIR / "skills").iterdir()): if d.is_dir() and not d.name.startswith("_"): skill_md = read_file(d / "SKILL.md") learnings = read_file(d / "learnings.md") eval_data = {} eval_path = d / "eval.json" if eval_path.exists(): eval_data = json.loads(eval_path.read_text()) score_history = [] score_path = d / "score-history.json" if score_path.exists(): score_history = json.loads(score_path.read_text()) skills.append({ "name": d.name, "description": skill_md[:200] if skill_md else "", "has_learnings": bool(learnings), "eval_criteria": eval_data.get("criteria", []), "scores": score_history, }) return skills @app.get("/api/skills/{name}") def get_skill(name: str): path = BASE_DIR / "skills" / name if not path.exists(): raise HTTPException(404, "Skill not found") return { "name": name, "skill": read_file(path / "SKILL.md"), "learnings": read_file(path / "learnings.md"), "eval": json.loads((path / "eval.json").read_text()) if (path / "eval.json").exists() else {}, "score_history": json.loads((path / "score-history.json").read_text()) if (path / "score-history.json").exists() else [], "context": [f.name for f in (path / "context").iterdir()] if (path / "context").exists() else [], } @app.post("/api/skills/{name}/run") def run_skill(name: str, req: Optional[SkillRunRequest] = None): path = BASE_DIR / "skills" / name if not path.exists(): raise HTTPException(404, "Skill not found") agent_choice = req.agent if req else "auto" skill_input = req.input if req else "" # Read skill files skill_md = read_file(path / "SKILL.md") learnings = read_file(path / "learnings.md") # Determine which agent based on skill type if agent_choice == "auto": devops_keywords = ["devops", "audit", "deploy", "k8s", "gcp", "infra", "terraform"] research_keywords = ["research", "synthesis", "analyze", "search", "compare"] if any(k in name for k in devops_keywords): agent_choice = "opencode" elif any(k in name for k in research_keywords): agent_choice = "gemini" else: # Check SKILL.md for explicit agent assignment for line in skill_md.split('\n'): line = line.strip() if "Primary:" in line: candidate = line.split(":")[-1].strip().lower() if candidate in ("opencode", "hermes", "gemini"): agent_choice = candidate break if agent_choice == "auto": agent_choice = "opencode" # Build prompt from skill instructions + learnings + user input prompt = f"Execute the '{name}' skill.\n\n" if skill_md: prompt += f"## Skill Instructions\n{skill_md}\n\n" if learnings and learnings.strip(): prompt += f"## Past Learnings\n{learnings}\n\n" if skill_input: prompt += f"## User Input\n{skill_input}" run_id = str(uuid.uuid4())[:8] # Execute via agent try: response_text = execute_agent(agent_choice, prompt) except subprocess.TimeoutExpired: response_text = f"⏱ Skill '{name}' timed out on agent '{agent_choice}'." except FileNotFoundError: response_text = f"⚠ Agent '{agent_choice}' CLI not installed. Install it and try again." except Exception as e: response_text = f"⚠ Error executing skill: {str(e)}" # Save output to learnings.md timestamp = get_timestamp()[:10] existing = read_file(path / "learnings.md") new_entry = ( f"\n## {timestamp} (Run {run_id})\n" f"- Agent: {agent_choice}\n" f"- Input: {skill_input or '(none)'}\n" f"- Output: {response_text[:500]}\n" ) write_file(path / "learnings.md", existing + new_entry) # Log execution append_audit({ "action": "skill_run", "skill": name, "agent": agent_choice, "run_id": run_id, "output_preview": response_text[:100], }) return { "status": "completed", "run_id": run_id, "skill": name, "agent": agent_choice, "output": response_text, "message": f"Skill '{name}' completed via {agent_choice}", } @app.get("/api/skills/{name}/eval") def get_skill_eval(name: str): path = BASE_DIR / "skills" / name / "score-history.json" if not path.exists(): return {"scores": []} return {"scores": json.loads(path.read_text())} # ─── Routes: Scheduler ──────────────────────────────────────────── @app.get("/api/scheduler/jobs") def list_jobs(): jobs_dir = BASE_DIR / "scheduler" / "jobs" jobs = [] for f in sorted(jobs_dir.glob("*.json")): jobs.append(json.loads(f.read_text())) return jobs @app.post("/api/scheduler/jobs") def create_job(job: ScheduleJobRequest): jobs_dir = BASE_DIR / "scheduler" / "jobs" jobs_dir.mkdir(parents=True, exist_ok=True) job_data = { "id": str(uuid.uuid4())[:8], "name": job.name, "skill": job.skill, "cron": job.cron, "enabled": job.enabled, "created": get_timestamp(), "last_run": None, "next_run": None, } (jobs_dir / f"{job.name.replace(' ', '_')}.json").write_text( json.dumps(job_data, indent=2) ) append_audit({"action": "job_created", "job": job.name}) return job_data @app.delete("/api/scheduler/jobs/{job_id}") def delete_job(job_id: str): jobs_dir = BASE_DIR / "scheduler" / "jobs" for f in jobs_dir.glob("*.json"): data = json.loads(f.read_text()) if data.get("id") == job_id: f.unlink() append_audit({"action": "job_deleted", "job_id": job_id}) return {"status": "deleted"} raise HTTPException(404, "Job not found") # ─── Routes: Audit ──────────────────────────────────────────────── @app.get("/api/audit") def get_audit(limit: int = Query(100, le=500)): audit_file = BASE_DIR / "audit" / "audit.log" if not audit_file.exists(): return {"entries": []} lines = audit_file.read_text().strip().split("\n") entries = [json.loads(l) for l in lines if l.strip()] return {"entries": entries[-limit:]} # ─── Routes: Cost Analytics ─────────────────────────────────────── @app.get("/api/cost") def get_cost(): cost_file = BASE_DIR / "data" / "cost-history.json" if not cost_file.exists(): return {"entries": [], "daily_totals": {}, "monthly_projection": 0, "free_tier_alerts": []} return json.loads(cost_file.read_text()) @app.post("/api/cost/record") def record_cost(data: dict): cost_file = BASE_DIR / "data" / "cost-history.json" cost_data = json.loads(cost_file.read_text()) if cost_file.exists() else \ {"entries": [], "daily_totals": {}, "monthly_projection": 0, "free_tier_alerts": []} cost_data["entries"].append({ "timestamp": get_timestamp(), "agent": data.get("agent", "unknown"), "tokens": data.get("tokens", 0), "cost": data.get("cost", 0.0), "model": data.get("model", "unknown"), }) cost_file.write_text(json.dumps(cost_data, indent=2)) return {"status": "recorded"} # ─── Routes: Registry/Plugins ───────────────────────────────────── @app.get("/api/plugins") def list_plugins(): reg_file = BASE_DIR / "registry" / "plugins.json" if not reg_file.exists(): return {"plugins": []} return json.loads(reg_file.read_text()) @app.post("/api/plugins/install") def install_plugin(data: dict): name = data.get("name", "").strip() if not name: raise HTTPException(400, "Plugin name required") reg_file = BASE_DIR / "registry" / "plugins.json" reg = json.loads(reg_file.read_text()) if reg_file.exists() else {"plugins": []} if any(p["name"] == name for p in reg["plugins"]): return {"status": "already_installed"} reg["plugins"].append({ "name": name, "installed": get_timestamp(), "version": "1.0.0", }) reg_file.write_text(json.dumps(reg, indent=2)) append_audit({"action": "plugin_installed", "plugin": name}) return {"status": "installed", "plugin": name} # ─── Routes: Backup ─────────────────────────────────────────────── @app.get("/api/backups") def list_backups(): backup_dir = BASE_DIR / "backups" backups = [] for f in sorted(backup_dir.glob("*.tar.gz"), reverse=True): backups.append({ "name": f.name, "size": f.stat().st_size, "created": datetime.fromtimestamp(f.stat().st_mtime).isoformat(), }) return backups @app.post("/api/backup") def create_backup(): backup_dir = BASE_DIR / "backups" backup_dir.mkdir(exist_ok=True) ts = datetime.now().strftime("%Y%m%d_%H%M%S") backup_file = backup_dir / f"agentic-os-{ts}.tar.gz" with tarfile.open(backup_file, "w:gz") as tar: for dir_name in ["brain", "skills", "agents", "registry", "standards", "prompts"]: d = BASE_DIR / dir_name if d.exists(): tar.add(d, arcname=dir_name) append_audit({"action": "backup_created", "file": backup_file.name}) return {"status": "ok", "file": backup_file.name, "size": backup_file.stat().st_size} @app.post("/api/backup/restore") def restore_backup(data: BackupRestoreRequest): backup_file = BASE_DIR / "backups" / data.file if not backup_file.exists(): raise HTTPException(404, "Backup file not found") with tarfile.open(backup_file, "r:gz") as tar: tar.extractall(path=BASE_DIR) append_audit({"action": "backup_restored", "file": data.file}) return {"status": "restored"} # ─── Routes: Prompts ────────────────────────────────────────────── @app.get("/api/prompts") def list_prompts(): prompts_dir = BASE_DIR / "prompts" prompts = {} for f in sorted(prompts_dir.glob("*.md")): prompts[f.stem] = read_file(f) return prompts # ─── Routes: Settings ───────────────────────────────────────────── @app.get("/api/settings") def get_settings(): sf = BASE_DIR / "data" / "settings.json" if not sf.exists(): return {} return json.loads(sf.read_text()) @app.put("/api/settings") def update_settings(data: SettingsUpdate): sf = BASE_DIR / "data" / "settings.json" # Merge with existing existing = json.loads(sf.read_text()) if sf.exists() else {} existing.update(data.settings) sf.write_text(json.dumps(existing, indent=2)) append_audit({"action": "settings_updated"}) return {"status": "ok"} # ─── Routes: Standards ──────────────────────────────────────────── @app.get("/api/standards") def list_standards(): std_dir = BASE_DIR / "standards" if not std_dir.exists(): return {"standards": []} standards = [] index_file = std_dir / "index.yml" index_content = read_file(index_file) for f in std_dir.glob("*.md"): standards.append({ "name": f.stem, "content": read_file(f), }) return {"standards": standards, "index": index_content} @app.post("/api/standards/discover") def discover_standards(): # Stub: scans codebase for patterns append_audit({"action": "standards_discovery_run"}) return {"status": "discovery_started", "message": "Scanning codebase for patterns..."} # ─── Routes: Chat ───────────────────────────────────────────────── CHAT_HISTORY_FILE = BASE_DIR / "data" / "chat-history.json" def load_chat_history(): if CHAT_HISTORY_FILE.exists(): return json.loads(CHAT_HISTORY_FILE.read_text()) return {"messages": []} def save_chat_message(msg: dict): history = load_chat_history() history["messages"].append(msg) if len(history["messages"]) > 200: history["messages"] = history["messages"][-200:] CHAT_HISTORY_FILE.write_text(json.dumps(history, indent=2)) def run_cli(args: list, timeout: int = 30) -> tuple: r = subprocess.run(args, capture_output=True, text=True, timeout=timeout) return r.returncode, r.stdout, r.stderr def clean_hermes_output(raw: str) -> str: """Strip CLI metadata from Hermes output, returning only the AI response.""" if not raw: return "" lines = raw.split('\n') in_box = False content_lines = [] for line in lines: if '╭─' in line: in_box = True continue if '╰─' in line: in_box = False continue if in_box: # Remove ANSI escape codes and leading whitespace cleaned = line.strip() if cleaned: content_lines.append(cleaned) if content_lines: return '\n'.join(content_lines) # Fallback: if no box found, return last non-metadata line non_meta = [l.strip() for l in lines if l.strip() and not l.startswith(('Query:', 'Initializing', '──', 'Resume', 'Session:', 'Duration:', 'Messages:'))] return '\n'.join(non_meta[-5:]) or raw def execute_agent(agent: str, message: str) -> str: try: if agent == "opencode": try: code, out, err = run_cli(["opencode", "run", "--format", "json", message], timeout=30) except subprocess.TimeoutExpired: return f"⏱ Agent 'opencode' timed out.\n\nOpenCode's model is taking too long. Try running `opencode run \"{message[:60]}\"` directly in your terminal.\n\n**Message:** {message[:100]}" if code == 0: response_text = "" for line in (out or "").split('\n'): line = line.strip() if not line: continue try: event = json.loads(line) if event.get("type") == "text": text = event.get("part", {}).get("text", "") if text: response_text += text + "\n" except (json.JSONDecodeError, KeyError): continue if response_text: return response_text.strip() return f"**opencode**\n\nProcessed your message.\n\n**Message:** {message[:100]}" err_msg = (err or "").strip() return err_msg or f"opencode returned exit code {code}" elif agent == "hermes": try: code, out, err = run_cli(["hermes", "chat", "-q", message], timeout=180) except subprocess.TimeoutExpired: return f"⏱ Hermes timed out.\n\nThe model took too long to respond. Try a shorter query or check your OpenRouter rate limits.\n\n**Message:** {message[:100]}" if code == 0: cleaned = clean_hermes_output(out or "") if cleaned: return cleaned # Empty response from model - return useful fallback return f"**Hermes**\n\nReceived your message but the model returned an empty response. Try rephrasing your query.\n\n**Message:** {message}" err_msg = (err or "").strip() if "invalid choice" in err_msg or "usage:" in err_msg: return f"**Hermes needs setup**\n\nRun `hermes setup` or check your config.\n\n**Details:** {err_msg[:200]}" return err_msg or f"hermes returned exit code {code}" elif agent == "gemini": for attempt, (args, to) in enumerate([ (["-y", "-m", "gemini-2.5-flash"], 60), (["-y"], 40), ]): try: code, out, err = run_cli(["gemini", *args, message], timeout=to) except subprocess.TimeoutExpired: if attempt == 0: continue return f"⏱ Gemini timed out.\n\nTry running `gemini \"{message[:60]}\"` directly.\n\n**Message:** {message[:100]}" if code == 0: return (out or "").strip() or f"**Gemini CLI**\n\nProcessed your query.\n\n**Message:** {message}" err_msg = (err or "").strip() if attempt == 0 and ("model" in err_msg.lower() or "not found" in err_msg.lower()): continue if "auth" in err_msg.lower() or "login" in err_msg.lower(): return f"**Gemini needs re-auth**\n\nRun `gemini auth login` to re-authenticate.\n\n**Details:** {err_msg[:200]}" return err_msg or f"gemini returned exit code {code}" return "Gemini CLI did not return a response." else: return f"Unknown agent: {agent}" except subprocess.TimeoutExpired: return f"⏱ Agent '{agent}' timed out.\n\nRun `{agent} --help` in your terminal for CLI usage.\n\n**Message:** {message[:100]}" except FileNotFoundError: return f"⚠ Agent '{agent}' CLI not installed. Install it and try again." except Exception as e: return f"⚠ Error communicating with {agent}: {str(e)}" @app.post("/api/chat") def chat(req: ChatRequest): agent = req.agent.lower().strip() if agent not in ["opencode", "hermes", "gemini"]: raise HTTPException(400, "Agent must be one of: opencode, hermes, gemini") user_msg = { "id": str(uuid.uuid4())[:8], "role": "user", "agent": agent, "content": req.message, "timestamp": get_timestamp(), } save_chat_message(user_msg) response_text = execute_agent(agent, req.message) agent_msg = { "id": str(uuid.uuid4())[:8], "role": "assistant", "agent": agent, "content": response_text, "timestamp": get_timestamp(), } save_chat_message(agent_msg) append_audit({"action": "chat_message", "agent": agent, "msg_preview": req.message[:50]}) return {"status": "ok", "response": agent_msg} @app.get("/api/chat/history") def get_chat_history(): return load_chat_history() # ─── Routes: Dashboard Static Files ────────────────────────────── dashboard_dir = BASE_DIR / "dashboard" if dashboard_dir.exists(): app.mount("/dashboard", StaticFiles(directory=str(dashboard_dir)), name="dashboard") @app.get("/", response_class=HTMLResponse) def index(): html_file = BASE_DIR / "dashboard" / "index.html" if html_file.exists(): content = html_file.read_text() content = content.replace('href="styles.css"', 'href="/dashboard/styles.css"') content = content.replace('src="utils.js"', 'src="/dashboard/utils.js"') content = content.replace('src="api.js"', 'src="/dashboard/api.js"') content = content.replace('src="app.js"', 'src="/dashboard/app.js"') content = content.replace('pages/', '/dashboard/pages/') return HTMLResponse(content=content) return HTMLResponse("

Agentic OS

Dashboard not built yet. Run ./install.sh first.

") # ─── Favicon ────────────────────────────────────────────────────── FAVICON_SVG = '' @app.get("/favicon.ico") def favicon(): return Response(content=FAVICON_SVG, media_type="image/svg+xml") @app.get("/favicon.svg") def favicon_svg(): return Response(content=FAVICON_SVG, media_type="image/svg+xml") # ─── Main ───────────────────────────────────────────────────────── if __name__ == "__main__": import uvicorn parser = argparse.ArgumentParser() parser.add_argument("--port", type=int, default=8080) parser.add_argument("--host", type=str, default="127.0.0.1") args = parser.parse_args() uvicorn.run(app, host=args.host, port=args.port)