#!/usr/bin/env python3 """SHiRE Agent Hub candidate: local-only registry, health, scoped chat and task queues.""" from __future__ import annotations import argparse import datetime as dt import hashlib import html import json import os from pathlib import Path import re import subprocess import threading import time import urllib.error import urllib.request from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from typing import Any ROOT = Path(os.environ.get("SHIRE_ROOT", "/home/shire3d/ARMOR")).resolve() REGISTRY_FILE = Path(os.environ.get("SHIRE_AGENT_REGISTRY", ROOT / "config/agents/registry.json")) AGENTS_ROOT = Path(os.environ.get("SHIRE_AGENTS_ROOT", ROOT / "agents/definitions")) DATA_ROOT = Path(os.environ.get("SHIRE_AGENT_DATA", ROOT / "data/agents")) VAULT_ROOT = Path(os.environ.get("SHIRE_VAULT", "/SHiREVault")) OLLAMA_URL = os.environ.get("OLLAMA_URL", "http://127.0.0.1:11434").rstrip("/") MARKETING_URL = os.environ.get("MARKETING_BOSS_URL", "http://127.0.0.1:4173").rstrip("/") HOST = os.environ.get("SHIRE_AGENT_HUB_HOST", "127.0.0.1") PORT = int(os.environ.get("SHIRE_AGENT_HUB_PORT", "8770")) MAX_BODY = 1_000_000 LOCK = threading.RLock() HUB_VERSION = "0.2.0-generic-health" HEALTH_CACHE_TTL_SECONDS = 3.0 HEALTH_CACHE: dict[str, tuple[float, dict[str, Any]]] = {} SENSITIVE_RE = re.compile(r"(?i)(password|secret|api[_ -]?key|access[_ -]?token|refresh[_ -]?token)\s*[:=]\s*\S+") EXTERNAL_ACTION_RE = re.compile( r"(?i)\b(publish|post live|send to customer|message customer|spend|buy ads?|launch ads?|" r"change credentials?|enter password|connect account|delete|refund|charge|email customers?)\b" ) def now_iso() -> str: return dt.datetime.now(dt.timezone.utc).isoformat() def read_json(path: Path, default: Any = None) -> Any: try: return json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return default def atomic_write_json(path: Path, payload: Any) -> None: path.parent.mkdir(parents=True, exist_ok=True) tmp = path.with_suffix(path.suffix + ".tmp") tmp.write_text(json.dumps(payload, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") os.replace(tmp, path) def registry() -> dict[str, Any]: value = read_json(REGISTRY_FILE, {}) return value if isinstance(value, dict) else {} def agents() -> list[dict[str, Any]]: value = registry().get("agents", []) return value if isinstance(value, list) else [] def agent_by_id(agent_id: str) -> dict[str, Any] | None: return next((item for item in agents() if item.get("id") == agent_id), None) def contract_for(agent_id: str) -> str: path = AGENTS_ROOT / agent_id / "SYSTEM_CONTRACT.md" try: return path.read_text(encoding="utf-8") except OSError: return "You are a scoped SHiRE specialist. Report missing evidence and stop before external actions." def http_json(url: str, method: str = "GET", payload: Any = None, timeout: float = 4.0) -> tuple[int, Any]: data = None headers = {"Accept": "application/json"} if payload is not None: data = json.dumps(payload).encode("utf-8") headers["Content-Type"] = "application/json" req = urllib.request.Request(url, data=data, method=method, headers=headers) try: with urllib.request.urlopen(req, timeout=timeout) as response: raw = response.read(MAX_BODY) return response.status, json.loads(raw.decode("utf-8")) except urllib.error.HTTPError as exc: raw = exc.read(MAX_BODY) try: body = json.loads(raw.decode("utf-8")) except Exception: body = {"error": raw.decode("utf-8", errors="replace")} return exc.code, body except Exception as exc: return 0, {"error": str(exc)} def ollama_models() -> list[str]: status, payload = http_json(f"{OLLAMA_URL}/api/tags", timeout=2.0) if status != 200: return [] return [str(item.get("name")) for item in payload.get("models", []) if item.get("name")] def model_available(requested: str, installed: list[str]) -> bool: if requested in installed: return True base = requested.split(":", 1)[0] return any(name == base or name.startswith(base + ":") for name in installed) def vault_status() -> dict[str, Any]: mounted = False writable = False filesystem = "" try: proc = subprocess.run(["findmnt", "-n", "-o", "FSTYPE", "-T", str(VAULT_ROOT)], capture_output=True, text=True, timeout=3) mounted = proc.returncode == 0 filesystem = proc.stdout.strip() except Exception: mounted = False if mounted and VAULT_ROOT.is_dir(): probe = VAULT_ROOT / ".shire-agent-hub-write-probe" try: content = f"probe:{now_iso()}".encode() probe.write_bytes(content) writable = hashlib.sha256(probe.read_bytes()).digest() == hashlib.sha256(content).digest() except OSError: writable = False finally: try: probe.unlink() except OSError: pass return {"path": str(VAULT_ROOT), "mounted": mounted, "writable": writable, "filesystem": filesystem} def service_health() -> dict[str, Any]: installed = ollama_models() return { "ok": True, "service": "shire-agent-hub", "version": HUB_VERSION, "boundTo": f"{HOST}:{PORT}", "publicInboundPorts": False, "registry": str(REGISTRY_FILE), "agentCount": len(agents()), "ollama": {"url": OLLAMA_URL, "online": bool(installed), "models": installed}, "shireVault": vault_status(), "time": now_iso(), } def application_health(item: dict[str, Any]) -> dict[str, Any]: """Probe a registered Agent's supported health endpoints.""" agent_id = str(item.get("id") or "") local_url = str(item.get("localUrl") or "").strip().rstrip("/") if not local_url: return { "online": False, "health": None, "healthEndpoint": None, "error": "No localUrl is registered.", "attempts": [], } current = time.monotonic() with LOCK: cached = HEALTH_CACHE.get(agent_id) if cached and current - cached[0] < HEALTH_CACHE_TTL_SECONDS: result = dict(cached[1]) result["cached"] = True return result endpoints: list[str] = [] if agent_id == "marketing-boss": endpoints.append(f"{MARKETING_URL}/api/health") endpoints.extend( [ f"{local_url}/health", f"{local_url}/api/health", f"{local_url}/api/v1/health", ] ) unique_endpoints: list[str] = [] for endpoint in endpoints: if endpoint not in unique_endpoints: unique_endpoints.append(endpoint) attempts: list[dict[str, Any]] = [] selected_health: dict[str, Any] | None = None selected_endpoint: str | None = None last_error: str | None = None online = False for endpoint in unique_endpoints: code, value = http_json(endpoint, timeout=2.0) attempts.append( { "endpoint": endpoint, "httpStatus": code, } ) if isinstance(value, dict) and value.get("error"): last_error = str(value.get("error")) if code == 200 and isinstance(value, dict): selected_health = value selected_endpoint = endpoint # Some existing SHiRE services pre-date the explicit `ok` # field. A valid JSON health document is healthy unless it # explicitly reports ok=false. online = value.get("ok") is not False if online: break result = { "online": online, "health": selected_health, "healthEndpoint": selected_endpoint, "error": ( None if online else last_error or "No supported health endpoint returned healthy JSON." ), "attempts": attempts, "cached": False, } with LOCK: HEALTH_CACHE[agent_id] = (current, result) return dict(result) def agent_status(item: dict[str, Any]) -> dict[str, Any]: installed = ollama_models() fast = item.get("brain", {}).get("fastModel") or item.get("fastModel") deep = item.get("brain", {}).get("deepModel") or item.get("deepModel") status = { "id": item.get("id"), "name": item.get("name"), "declaredStatus": item.get("status"), "enabled": bool(item.get("enabled")), "route": item.get("route"), "localUrl": item.get("localUrl"), "brain": { "fastModel": fast, "fastAvailable": bool(fast and model_available(str(fast), installed)), "deepModel": deep, "deepAvailable": bool(deep and model_available(str(deep), installed)), }, "memory": { "root": str(DATA_ROOT / str(item.get("id"))), "isolated": True, }, "checkedAt": now_iso(), } status["application"] = application_health(item) return status def queue_task(agent_id: str, payload: dict[str, Any]) -> dict[str, Any]: message = str(payload.get("task") or payload.get("message") or "").strip() if not message: raise ValueError("A task is required.") clean = SENSITIVE_RE.sub(r"\1=[REDACTED]", message)[:20_000] requires_approval = bool(EXTERNAL_ACTION_RE.search(clean)) or bool(payload.get("requiresExternalAction")) task_id = f"TASK-{dt.datetime.now(dt.timezone.utc).strftime('%Y%m%dT%H%M%SZ')}-{hashlib.sha256(os.urandom(16)).hexdigest()[:8].upper()}" task = { "taskId": task_id, "agentId": agent_id, "task": clean, "state": "AWAITING_RAY_APPROVAL" if requires_approval else "QUEUED", "requiresExternalActionApproval": requires_approval, "createdAt": now_iso(), "source": "shire-agent-hub", } atomic_write_json(DATA_ROOT / agent_id / "tasks" / f"{task_id}.json", task) return task def chat(agent: dict[str, Any], payload: dict[str, Any]) -> dict[str, Any]: message = str(payload.get("message") or "").strip() if not message: raise ValueError("A message is required.") mode = "deep" if str(payload.get("mode", "fast")).lower() == "deep" else "fast" brain = agent.get("brain", {}) model = brain.get("deepModel" if mode == "deep" else "fastModel") or agent.get("deepModel" if mode == "deep" else "fastModel") if not model: raise RuntimeError("This Agent does not have a model profile.") system = contract_for(str(agent["id"])) if EXTERNAL_ACTION_RE.search(message): system += "\nThe current request appears to involve an external, financial, credential-sensitive, destructive or public action. Prepare a proposal only and explicitly mark it AWAITING_RAY_APPROVAL. Do not claim the action happened." status, result = http_json( f"{OLLAMA_URL}/api/chat", method="POST", payload={ "model": model, "stream": False, "messages": [ {"role": "system", "content": system}, {"role": "user", "content": message[:40_000]}, ], "options": {"temperature": 0.35}, }, timeout=180.0, ) if status != 200: raise RuntimeError(f"Ollama request failed: {result.get('error', 'unknown error')}") text = str((result.get("message") or {}).get("content") or "").strip() log = { "at": now_iso(), "agentId": agent["id"], "mode": mode, "model": model, "user": SENSITIVE_RE.sub(r"\1=[REDACTED]", message)[:20_000], "assistant": text[:80_000], } path = DATA_ROOT / str(agent["id"]) / "memory" / "conversation.jsonl" path.parent.mkdir(parents=True, exist_ok=True) with LOCK, path.open("a", encoding="utf-8") as handle: handle.write(json.dumps(log, ensure_ascii=False) + "\n") return {"agentId": agent["id"], "mode": mode, "model": model, "response": text, "externalActionApprovalRequired": bool(EXTERNAL_ACTION_RE.search(message))} def dashboard() -> str: cards = [] for item in agents(): status = html.escape(str(item.get("status", "UNKNOWN"))) enabled = bool(item.get("enabled")) name = html.escape(str(item.get("name", item.get("id")))) desc = html.escape(str(item.get("description", ""))) aid = html.escape(str(item.get("id"))) url = item.get("localUrl") open_button = f'Open workspace' if url else 'Workspace staged' cards.append(f'''

{name}

{'ACTIVE' if enabled else 'STAGED'}

{desc}

{status}
{open_button}Status JSON
''') return f'''SHiRE Agents

SHiRE Specialist Agents

Dedicated scoped brains coordinated by SHiRE Central. Private Tailnet access enabled.

{''.join(cards)}
''' class Handler(BaseHTTPRequestHandler): server_version = f"SHiREAgentHub/{HUB_VERSION}" def log_message(self, fmt: str, *args: Any) -> None: print(f"{self.address_string()} - {fmt % args}") def _send_json(self, status: int, payload: Any) -> None: body = json.dumps(payload, ensure_ascii=False, indent=2).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") self.send_header("X-Content-Type-Options", "nosniff") self.end_headers() self.wfile.write(body) def _read_json(self) -> dict[str, Any]: try: length = int(self.headers.get("Content-Length", "0")) except ValueError: raise ValueError("Invalid Content-Length.") if length < 1 or length > MAX_BODY: raise ValueError("Request body size is invalid.") raw = self.rfile.read(length) value = json.loads(raw.decode("utf-8")) if not isinstance(value, dict): raise ValueError("JSON body must be an object.") return value def do_GET(self) -> None: # noqa: N802 path = self.path.split("?", 1)[0] if path in ("/", "/agents"): body = dashboard().encode("utf-8") self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(body) return if path == "/health": return self._send_json(200, service_health()) if path == "/api/v1/agents": return self._send_json(200, {"agents": [agent_status(item) for item in agents()]}) match = re.fullmatch(r"/api/v1/agents/([a-z0-9-]+)(?:/status)?", path) if match: item = agent_by_id(match.group(1)) if not item: return self._send_json(404, {"error": "Agent not found."}) return self._send_json(200, agent_status(item)) return self._send_json(404, {"error": "Not found."}) def do_POST(self) -> None: # noqa: N802 path = self.path.split("?", 1)[0] match = re.fullmatch(r"/api/v1/agents/([a-z0-9-]+)/(chat|tasks)", path) if not match: return self._send_json(404, {"error": "Not found."}) item = agent_by_id(match.group(1)) if not item: return self._send_json(404, {"error": "Agent not found."}) try: payload = self._read_json() result = chat(item, payload) if match.group(2) == "chat" else queue_task(str(item["id"]), payload) return self._send_json(200, result) except ValueError as exc: return self._send_json(400, {"error": str(exc)}) except RuntimeError as exc: return self._send_json(503, {"error": str(exc)}) except Exception as exc: return self._send_json(500, {"error": f"Agent Hub error: {exc}"}) def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--check", action="store_true", help="validate registry and exit") args = parser.parse_args() if not REGISTRY_FILE.is_file(): raise SystemExit(f"Missing Agent registry: {REGISTRY_FILE}") missing = [item.get("id") for item in agents() if not (AGENTS_ROOT / str(item.get("id")) / "SYSTEM_CONTRACT.md").is_file()] if missing: raise SystemExit(f"Missing Agent contracts: {', '.join(map(str, missing))}") DATA_ROOT.mkdir(parents=True, exist_ok=True) if args.check: print(json.dumps({"ok": True, "agents": len(agents()), "registry": str(REGISTRY_FILE)}, indent=2)) return print(f"SHiRE Agent Hub {HUB_VERSION} listening on http://{HOST}:{PORT}") ThreadingHTTPServer((HOST, PORT), Handler).serve_forever() if __name__ == "__main__": main()