#!/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 from functools import lru_cache from pathlib import Path from typing import Any import yaml from mcp.server.fastmcp import FastMCP CONTEXT_DIR = Path(os.environ.get("HOMELAB_CONTEXT_DIR", "/opt/homelab-context")) INVENTORY = CONTEXT_DIR / "inventory.yaml" HOSTS_DIR = CONTEXT_DIR / "hosts" # 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 --` 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/.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 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")