Phase 1: cross-client homelab context + MCP scaffolding
Add the foundation for distributing homelab context to every client (LXCs, VMs, workstations including republic-laptop, mac-mini, ludo-mini) with a single source of truth, structured query layer (MCP), and per-client age-key issuance for secrets: - inventory.yaml — canonical topology (hosts, services, mesh addresses) - hosts/*.yaml — per-host identity files generated from inventory by mcp/build_host_files.py; do not edit by hand - AGENTS.md — orientation doc symlinked to /root/AGENTS.md on every client - bootstrap.sh — one-shot enroll (Linux + macOS), clones repo, fetches age key from issuance, installs sync timer/launchd job, drops the homelab CLI - bin/homelab — single-binary Python CLI: whoami, list, ssh, pct, logs, restart, open, status, secret, sync, mcp, client add/remove, nuke - mcp/server.py — FastMCP server: context tools + read-only management tools (no mutations exposed); shell-outs use mcp-reader restricted ssh key - mcp/deploy/ — claudio-monitor-style gitea webhook deploy scaffold for the MCP service on LXC 105 (ports 9810 mcp, 9811 webhook) - secrets-issuance/ — per-client age key auto-provisioning over the mesh; source-IP gated against inventory, with denylist for revoked clients (ports 9820 issue, 9821 webhook) - secrets/, .sops.yaml — SOPS recipient scaffolding; the operator fills in age public keys after Phase 3a generates them - scripts/sync/ — systemd timer (Linux) + launchd plist (macOS) pulling /opt/homelab-context every 5 min Mesh: both Netbird (preferred, 100.122.0.0/16) and Tailscale accepted during the in-flight migration; no client is gated on completing the move. Plan reference: /root/.claude/plans/lets-make-a-plan-fluttering-trinket.md Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
root
310
mcp/server.py
Executable file
310
mcp/server.py
Executable file
@@ -0,0 +1,310 @@
|
||||
#!/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"
|
||||
SSH_IDENTITY = os.environ.get("HOMELAB_MCP_SSH_KEY", "/etc/homelab-mcp/mcp-reader.key")
|
||||
SSH_USER = os.environ.get("HOMELAB_MCP_SSH_USER", "mcp-reader")
|
||||
SSH_TIMEOUT = int(os.environ.get("HOMELAB_MCP_SSH_TIMEOUT", "10"))
|
||||
|
||||
mcp = FastMCP("homelab")
|
||||
|
||||
|
||||
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 _host_ssh_target(name: str) -> str:
|
||||
"""Return the ssh target (user@host) the MCP reader uses for `name`."""
|
||||
inv = inventory()
|
||||
h = inv.get("hosts", {}).get(name)
|
||||
if h is None:
|
||||
raise ValueError(f"unknown host: {name}")
|
||||
mesh = h.get("mesh", {})
|
||||
# Netbird FQDN if available, then lan_ip, then tailscale.
|
||||
target = (
|
||||
mesh.get("netbird", {}).get("fqdn")
|
||||
or mesh.get("netbird", {}).get("ip")
|
||||
or h.get("lan_ip")
|
||||
or mesh.get("tailscale", {}).get("fqdn")
|
||||
)
|
||||
if not target:
|
||||
raise ValueError(f"no reachable address for host {name}")
|
||||
return f"{SSH_USER}@{target}"
|
||||
|
||||
|
||||
def _ssh(host: str, *cmd: str, timeout: int | None = None) -> subprocess.CompletedProcess:
|
||||
"""Run a command on a remote host as the restricted mcp-reader user."""
|
||||
target = _host_ssh_target(host)
|
||||
full = [
|
||||
"ssh", "-i", SSH_IDENTITY, "-o", "BatchMode=yes",
|
||||
"-o", "StrictHostKeyChecking=accept-new",
|
||||
"-o", f"ConnectTimeout={SSH_TIMEOUT}",
|
||||
target, *cmd,
|
||||
]
|
||||
return subprocess.run(full, capture_output=True, text=True,
|
||||
timeout=timeout or SSH_TIMEOUT * 3)
|
||||
|
||||
|
||||
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 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 = _ssh(host, "systemctl", "is-active", unit)
|
||||
enabled = _ssh(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 = _ssh(host, "journalctl", "-u", unit, "-n", str(lines), "--no-pager")
|
||||
return proc.stdout
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def list_lxcs() -> str:
|
||||
"""Run `pct list` on hubris."""
|
||||
return _ssh("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 = _ssh("hubris", "pct", "status", str(pve_id))
|
||||
cfg = _ssh("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="sse")
|
||||
Reference in New Issue
Block a user