dtoro/oikos
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

cleanup: reflect Go rewrite reality — remove legacy Python artifacts

Browse Source 
Removed:
- bin/hermes (9.3MB compiled binary accidentally committed to git)
- oikos/console/ (Flask web console — replaced by Go REST API + SSE)
- mcp/server.py (Python MCP server — replaced by internal/mcp/)
- oikos/policy.yaml, oikos/ontology.yaml (duplicates of seeds/)

Kept:
- oikos/*.py kernel files (12 files) — still imported by bin/homelab
  for operational CLI commands (ssh, pct, logs, restart, status,
  open, secret, client, sync, mcp). Will be removed when bin/homelab
  is ported to Go.
- mcp/build_host_files.py — generates hosts/*.yaml from inventory,
  still operational. Will be ported to Go.
- bin/homelab — active Python CLI, still operational.

Updated:
- .gitignore: added bin/hermes, cleaned up legacy comments
- plans/index.md: listed all 4 active plans with accurate statuses
This commit is contained in:
authentik Default Admin
2026-07-07 17:43:10 +02:00
parent dcd35b6315
commit c8b27b2c51
23 changed files with 13 additions and 1573 deletions
Download Patch File Download Diff File Expand all files Collapse all files

411
mcp/server.py
View File

@@ -1,411 +0,0 @@
#!/usr/bin/env python3
"""
Homelab MCP server.
Reads the canonical state from /opt/homelab-context/ (a git clone of
dtoro/Homelab-Docs) and exposes structured tools to any MCP-capable agent.
Two tool groups:
- Context (pure read of the clone, no shell-outs)
- Management (read-only ssh/pct/systemctl/curl, NO mutations)
Mutations live in the `homelab` CLI on each client, behind operator
confirmation. The MCP server never restarts, edits, or executes arbitrary
commands.
"""
from __future__ import annotations
import os
import re
import shlex
import subprocess
import sys
from functools import lru_cache
from pathlib import Path
from typing import Any
import yaml
from mcp.server.fastmcp import FastMCP # noqa: E402 — must precede the sys.path
# insert below: CONTEXT_DIR contains its
# own top-level "mcp/" directory, which
# would shadow the real `mcp` package if
# inserted first.
CONTEXT_DIR = Path(os.environ.get("HOMELAB_CONTEXT_DIR", "/opt/homelab-context"))
INVENTORY = CONTEXT_DIR / "inventory.yaml"
HOSTS_DIR = CONTEXT_DIR / "hosts"
CARDS_DIR = CONTEXT_DIR / "oikos" / "cards"
sys.path.insert(0, str(CONTEXT_DIR))
from oikos import ledger as oikos_ledger # noqa: E402
from oikos import policy as oikos_policy # noqa: E402
from oikos import relations as oikos_relations # noqa: E402
# All management tools proxy through hubris (the Proxmox host) via a single
# restricted-shell SSH connection. The wrapper at mcp/mcp-reader-shell on
# hubris validates each command against a strict read-only allowlist.
SSH_IDENTITY = os.environ.get("HOMELAB_MCP_SSH_KEY", "/etc/homelab-mcp/mcp-reader.key")
SSH_USER = os.environ.get("HOMELAB_MCP_SSH_USER", "root")
SSH_KNOWN_HOSTS = os.environ.get("HOMELAB_MCP_SSH_KNOWN_HOSTS",
"/etc/homelab-mcp/known_hosts")
HUBRIS_HOST = os.environ.get("HOMELAB_MCP_HUBRIS_HOST", "192.168.8.77")
SSH_TIMEOUT = int(os.environ.get("HOMELAB_MCP_SSH_TIMEOUT", "10"))
mcp = FastMCP("homelab")
mcp.settings.host = os.environ.get("HOMELAB_MCP_HOST", "0.0.0.0")
mcp.settings.port = int(os.environ.get("HOMELAB_MCP_PORT", "9810"))
# FastMCP's DNS-rebinding protection only whitelists 127.0.0.1 / localhost / [::1]
# by default, which breaks any LAN/mesh client. We're already mesh+LAN-gated at
# nftables and the browser-attack threat doesn't apply to mesh-only services.
mcp.settings.transport_security.enable_dns_rebinding_protection = False
def _load_inventory() -> dict:
if not INVENTORY.exists():
raise RuntimeError(f"inventory not found: {INVENTORY}")
return yaml.safe_load(INVENTORY.read_text())
@lru_cache(maxsize=1)
def _inventory_cache_key() -> int:
"""Cache buster keyed on mtime so we re-read after each sync pull."""
return INVENTORY.stat().st_mtime_ns
def inventory() -> dict:
_inventory_cache_key.cache_clear()
_ = _inventory_cache_key() # warm
return _load_inventory()
def _run_via_hubris(remote_host: str, cmd: list[str],
timeout: int | None = None) -> subprocess.CompletedProcess:
"""SSH to hubris (single channel, restricted-shell key) and run `cmd`.
If remote_host is an LXC, the command is wrapped in `pct exec <id> --`
so hubris executes it inside the LXC. The wrapper at /usr/local/bin/
mcp-reader-shell on hubris validates the final command against an
allowlist before execution.
"""
inv_hosts = inventory().get("hosts", {})
if remote_host == "hubris":
full = cmd
else:
host_entry = inv_hosts.get(remote_host)
if host_entry is None:
raise ValueError(f"unknown host: {remote_host}")
pve_id = host_entry.get("pve_id")
if pve_id is None:
raise ValueError(f"{remote_host} has no pve_id; can't pct-exec into it")
full = ["pct", "exec", str(pve_id), "--", *cmd]
# The restricted shell on hubris reads the joined command from
# $SSH_ORIGINAL_COMMAND and validates it as a single string.
joined = " ".join(full)
ssh_args = [
"ssh", "-i", SSH_IDENTITY, "-o", "BatchMode=yes",
# The systemd unit runs with ProtectHome=true so ~/.ssh is unreachable.
# Use a pre-populated known_hosts in /etc/homelab-mcp/.
"-o", f"UserKnownHostsFile={SSH_KNOWN_HOSTS}",
"-o", "StrictHostKeyChecking=yes",
"-o", f"ConnectTimeout={SSH_TIMEOUT}",
f"{SSH_USER}@{HUBRIS_HOST}",
joined,
]
proc = subprocess.run(
ssh_args, capture_output=True, text=True,
timeout=timeout or SSH_TIMEOUT * 3,
)
if proc.returncode != 0:
import logging as _l
_l.getLogger("homelab-mcp").warning(
"ssh failed rc=%s host=%s cmd=%r stderr=%r",
proc.returncode, remote_host, joined, proc.stderr.strip(),
)
return proc
def _service_to_host(service: str) -> str:
"""Resolve a service name to its backend host name."""
svc = inventory().get("services", {}).get(service)
if not svc:
raise ValueError(f"unknown service: {service}")
return svc["backend"]
# ---------------------------------------------------------------------------
# Context tools
# ---------------------------------------------------------------------------
@mcp.tool()
def get_host(name: str) -> dict:
"""Return the structured record for a host (LXC, VM, workstation, or hubris)."""
inv = inventory()
entry = inv.get("hosts", {}).get(name)
if entry is None:
raise ValueError(f"unknown host: {name}")
out = {"name": name, **entry}
yaml_path = HOSTS_DIR / f"{name}.yaml"
if yaml_path.exists():
out["host_yaml_path"] = str(yaml_path)
return out
@mcp.tool()
def list_services() -> dict:
"""List every service registered in inventory.yaml."""
return inventory().get("services", {})
@mcp.tool()
def find_service(query: str) -> list[dict]:
"""Find services by name substring, role, or backend host name."""
q = query.lower()
inv = inventory()
out = []
for name, entry in inv.get("services", {}).items():
haystack = " ".join([
name,
str(entry.get("backend", "")),
str(entry.get("role", "")),
str(entry.get("note", "")),
]).lower()
if q in haystack:
out.append({"service": name, **entry})
return out
@mcp.tool()
def get_topology() -> dict:
"""Return the full inventory (hosts + services + mesh)."""
return inventory()
@mcp.tool()
def search_docs(query: str, max_results: int = 10) -> list[dict]:
"""Ripgrep the markdown wiki for query, return file:line hits."""
rg = subprocess.run(
["rg", "--no-heading", "-n", "-i", "--type", "md", "-m", "5", query, str(CONTEXT_DIR)],
capture_output=True, text=True, timeout=10,
)
results = []
for line in rg.stdout.splitlines()[: max_results * 5]:
m = re.match(r"^(.+?):(\d+):(.*)$", line)
if m:
results.append({
"path": str(Path(m.group(1)).relative_to(CONTEXT_DIR)),
"line": int(m.group(2)),
"text": m.group(3).strip(),
})
if len(results) >= max_results:
break
return results
@mcp.tool()
def get_page(path: str) -> str:
"""Return a markdown page verbatim. Path is relative to the repo root."""
full = (CONTEXT_DIR / path).resolve()
if not str(full).startswith(str(CONTEXT_DIR.resolve())):
raise ValueError("path escapes repo")
if not full.exists():
raise ValueError(f"no such page: {path}")
return full.read_text()
@mcp.tool()
def get_changelog(path: str, since: str | None = None) -> str:
"""Extract the trailing ## Changelog section from a page; optionally filter."""
text = get_page(path)
m = re.search(r"^##\s+Changelog\s*\n(.*)", text, re.MULTILINE | re.DOTALL)
if not m:
return ""
body = m.group(1).strip()
if since:
kept = []
for entry in re.split(r"^###\s+", body, flags=re.MULTILINE):
if not entry.strip():
continue
head = entry.split("\n", 1)[0]
date_m = re.match(r"(\d{4}-\d{2}-\d{2})", head)
if not date_m or date_m.group(1) >= since:
kept.append("### " + entry.rstrip())
return "\n\n".join(kept).strip()
return body
@mcp.tool()
def whoami(hostname: str) -> dict:
"""Given a hostname, return that host's full yaml record from hosts/<name>.yaml."""
candidate = HOSTS_DIR / f"{hostname}.yaml"
if not candidate.exists():
raise ValueError(f"no host record for {hostname}")
return yaml.safe_load(candidate.read_text())
@mcp.tool()
def explain(service: str) -> str:
"""Return the compact context card for a service: identity, blast
radius, safe actions + risk class, doc pointer, recent ledger history.
Card-first — cheaper for agent orientation than search_docs + get_page.
"""
card = CARDS_DIR / f"service-{service}.md"
if not card.exists():
raise ValueError(f"no context card for {service} — has gen-topology.py run?")
return card.read_text()
@mcp.tool()
def preflight(service: str) -> dict:
"""Dry-run report before mutating a service: risk class, approval
requirement, current health, config repo, and the verification command
to run after the change."""
inv_svc = inventory().get("services", {}).get(service)
if not inv_svc:
raise ValueError(f"unknown service: {service}")
risk = (oikos_policy.classify_action("tracked-config-edit", service)
if inv_svc.get("config_repo")
else oikos_policy.classify_action("service-restart", service)) or "config_mutation"
url = inv_svc.get("url") or inv_svc.get("endpoint")
return {
"service": service,
"risk_class": risk,
"approval": oikos_policy.approval_for(risk),
"config_repo": inv_svc.get("config_repo"),
"risk_notes": inv_svc.get("risk_notes"),
"verification": f"curl -sf {url}" if url else f"tail_log({service!r})",
}
@mcp.tool()
def get_relations(entity: str) -> list[dict]:
"""Walk the ontology graph both directions for a host or service name:
what it impacts, what affects it, and its full transitive blast radius.
"""
return oikos_relations.relations_for_name(entity)
@mcp.tool()
def get_state_snapshot() -> dict:
"""The Week-3 scheduler's last Observe-pass snapshot (service health,
host disk usage, drift-finding count, generated_at timestamp). This is
what makes cache-first reads work from ANY client, not just the one the
scheduler runs on: the CLI's local-file cache only helps on that host;
agents elsewhere should call this tool instead of assuming a local
oikos/state.json exists."""
from oikos import scheduler as oikos_scheduler
state = oikos_scheduler.read_state()
if state is None:
raise ValueError("no scheduler snapshot yet — has oikos-scheduler.timer run?")
return state
@mcp.tool()
def get_change_history(entity: str, limit: int = 20) -> list[dict]:
"""Ledger entries for `entity` (e.g. "service:jellyfin", "host:strong"),
newest first."""
return oikos_ledger.history(entity, limit=limit)
@mcp.tool()
def list_my_secrets(caller_pubkey: str) -> list[str]:
"""Return the names of secrets the caller (identified by age pubkey) can decrypt.
Metadata only — the server never returns plaintext. The client decrypts
locally with its own /etc/age/key.txt.
"""
if not caller_pubkey:
return []
secrets_dir = CONTEXT_DIR / "secrets"
if not secrets_dir.exists():
return []
out = []
for path in secrets_dir.glob("*.yaml"):
try:
data = yaml.safe_load(path.read_text()) or {}
except yaml.YAMLError:
continue
recipients = (
data.get("sops", {})
.get("age", [])
)
for r in recipients:
if r.get("recipient") == caller_pubkey:
out.append(path.stem)
break
return sorted(out)
# ---------------------------------------------------------------------------
# Management tools (read-only)
# ---------------------------------------------------------------------------
@mcp.tool()
def get_service_status(service: str) -> dict:
"""systemctl is-active + is-enabled for the named service on its backend host."""
host = _service_to_host(service)
inv_svc = inventory()["services"][service]
unit = inv_svc.get("systemd_unit", service)
active = _run_via_hubris(host, ["systemctl", "is-active", unit])
enabled = _run_via_hubris(host, ["systemctl", "is-enabled", unit])
return {
"service": service,
"host": host,
"unit": unit,
"active": active.stdout.strip(),
"enabled": enabled.stdout.strip(),
}
@mcp.tool()
def tail_log(service: str, lines: int = 200) -> str:
"""Last N journalctl lines for the named service on its backend host."""
host = _service_to_host(service)
inv_svc = inventory()["services"][service]
unit = inv_svc.get("systemd_unit", service)
proc = _run_via_hubris(host, ["journalctl", "-u", unit,
"-n", str(lines), "--no-pager"])
return proc.stdout
@mcp.tool()
def list_lxcs() -> str:
"""Run `pct list` on hubris."""
return _run_via_hubris("hubris", ["pct", "list"]).stdout
@mcp.tool()
def get_lxc_state(lxc: str) -> dict:
"""`pct status` and a quick `pct config` snapshot for the named LXC."""
inv = inventory().get("hosts", {})
if lxc not in inv:
raise ValueError(f"unknown lxc: {lxc}")
pve_id = inv[lxc].get("pve_id")
if pve_id is None:
raise ValueError(f"{lxc} has no pve_id (is it actually an LXC?)")
status = _run_via_hubris("hubris", ["pct", "status", str(pve_id)])
cfg = _run_via_hubris("hubris", ["pct", "config", str(pve_id)])
return {"lxc": lxc, "pve_id": pve_id,
"status": status.stdout.strip(),
"config": cfg.stdout}
@mcp.tool()
def ping_service(service: str) -> dict:
"""HTTP check against the service's URL (from inventory)."""
inv_svc = inventory().get("services", {}).get(service)
if not inv_svc:
raise ValueError(f"unknown service: {service}")
url = inv_svc.get("url") or inv_svc.get("endpoint") or inv_svc.get("backend_url")
if not url:
return {"service": service, "ok": False, "reason": "no URL in inventory"}
proc = subprocess.run(
["curl", "-sS", "-o", "/dev/null", "-w", "%{http_code} %{time_total}",
"--max-time", "5", url],
capture_output=True, text=True,
)
code, t = (proc.stdout.strip().split() + ["", ""])[:2]
return {"service": service, "url": url, "http_code": code, "time_s": t,
"ok": code.startswith("2") or code.startswith("3")}
if __name__ == "__main__":
mcp.run(transport="streamable-http")