From 88e137d947131a268e53643e7629809327ff0d0e Mon Sep 17 00:00:00 2001 From: drjones Date: Mon, 21 Sep 2026 15:02:16 +0000 Subject: [PATCH] VEGA tool host: 22 API-first developer tools --- .gitignore | 3 + README.md | 17 +++ app.py | 338 +++++++++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 358 insertions(+) create mode 100644 .gitignore create mode 100644 README.md create mode 100644 app.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..d0ee3b1 --- /dev/null +++ b/.gitignore @@ -0,0 +1,3 @@ +venv/ +__pycache__/ +*.pyc diff --git a/README.md b/README.md new file mode 100644 index 0000000..370f4e3 --- /dev/null +++ b/README.md @@ -0,0 +1,17 @@ +# VEGA — API-first utility/developer tool host + +22 deterministic developer tools, each exposed as one REST endpoint. +Phase 1 of the paid-MCP toolbox (wired into Hyperion + Omninexus). + +## Run + + python3 -m venv venv && ./venv/bin/pip install flask + ./venv/bin/python app.py + +## API + + GET /health + GET /api/tools + GET|POST /api/? + +Public: https://vega.thetempleofdoom.com diff --git a/app.py b/app.py new file mode 100644 index 0000000..9a03817 --- /dev/null +++ b/app.py @@ -0,0 +1,338 @@ +#!/usr/bin/env python3 +""" +VEGA — API-first utility/developer tool host. + +Every capability is a plain function registered in TOOLS and auto-exposed as: + GET /api/tools -> catalog {name: {label, category, params, desc}} + GET /api/? -> run one tool, return JSON {tool, result} + POST /api/ -> same, params as JSON body + GET /health -> liveness + tool count + +Design goals: + - API-first: the web UI (later) and the MCP wrapper (later) both consume these + exact endpoints — no separate logic paths. + - Deterministic + stdlib-only for the core: no keys, no network, fast, safe. + - One registry -> N surfaces (REST now, MCP + Hyperion metering next). + +This is the Phase-1 slice (developer tools). Later phases add internet/infra +(re-using osint.py's 454), files/media, monitoring, and AI/LLM categories. +""" +import base64 +import binascii +import csv +import hashlib +import html +import io +import json +import re +import time +import unicodedata +import uuid as _uuid +from urllib.parse import parse_qs, quote, unquote, urlparse + +from flask import Flask, jsonify, request + +app = Flask(__name__) +TOOLS = {} + + +def tool(name, category, params, desc): + """Register a function as a tool. params = list of input key names.""" + def deco(fn): + TOOLS[name] = {"fn": fn, "category": category, "params": params, + "desc": desc, "label": name.replace("_", " ").title()} + return fn + return deco + + +# --------------------------------------------------------------------------- +# JSON / data +# --------------------------------------------------------------------------- +@tool("json_format", "data", ["json"], "Validate + pretty-print JSON.") +def json_format(json="{}"): + d = json_loads(json) + return {"valid": True, "pretty": json.dumps(d, indent=2, sort_keys=True), + "type": type(d).__name__} + + +@tool("json_to_csv", "data", ["json"], "Flatten a JSON array of objects to CSV.") +def json_to_csv(json="[]"): + d = json_loads(json) + if not isinstance(d, list): + return {"error": "expected a JSON array of objects"} + out = io.StringIO() + w = csv.DictWriter(out, fieldnames=sorted({k for o in d if isinstance(o, dict) for k in o})) + w.writeheader() + for o in d: + if isinstance(o, dict): + w.writerow(o) + return {"csv": out.getvalue()} + + +@tool("json_diff", "data", ["a", "b"], "Diff two JSON values (added/removed/changed keys).") +def json_diff(a="{}", b="{}"): + da, db = json_loads(a), json_loads(b) + return diff(da, db) + + +@tool("jwt_decode", "data", ["token"], "Decode a JWT header+payload (no signature verification).") +def jwt_decode(token=""): + parts = token.split(".") + if len(parts) < 2: + return {"error": "not a JWT (need header.payload.signature)"} + out = {} + for name, p in (("header", parts[0]), ("payload", parts[1])): + try: + pad = p + "=" * (-len(p) % 4) + out[name] = json.loads(base64.urlsafe_b64decode(pad).decode("utf-8", "replace")) + except Exception as e: + out[name] = {"error": str(e)} + return out + + +# --------------------------------------------------------------------------- +# encoding / crypto / ids +# --------------------------------------------------------------------------- +@tool("hash", "encoding", ["text", "algo"], + "Hash a string. algo = md5/sha1/sha256/sha512.") +def hash_text(text="", algo="sha256"): + h = hashlib.new(algo if algo in hashlib.algorithms_available else "sha256") + h.update(text.encode()) + return {"algo": h.name, "hex": h.hexdigest(), "bytes": h.digest_size} + + +@tool("base64_encode", "encoding", ["text"], "Base64-encode text.") +def b64_encode(text=""): + return {"encoded": base64.b64encode(text.encode()).decode()} + + +@tool("base64_decode", "encoding", ["data"], "Base64-decode (returns text + hex).") +def b64_decode(data=""): + raw = base64.b64decode(data + "=" * (-len(data) % 4)) + return {"text": raw.decode("utf-8", "replace"), "hex": raw.hex()} + + +@tool("url_encode", "encoding", ["text"], "Percent-encode a string.") +def url_encode(text=""): + return {"encoded": quote(text, safe="")} + + +@tool("url_decode", "encoding", ["data"], "Percent-decode a string.") +def url_decode(data=""): + return {"decoded": unquote(data)} + + +@tool("url_parse", "encoding", ["url"], "Parse a URL into scheme/host/path/query/fragment.") +def url_parse(url=""): + p = urlparse(url) + return {"scheme": p.scheme, "netloc": p.netloc, "host": p.hostname, + "port": p.port, "path": p.path, "query": parse_qs(p.query), + "fragment": p.fragment} + + +@tool("uuid", "ids", [], "Generate a random UUID v4.") +def uuid_gen(): + return {"uuid": str(_uuid.uuid4()), "hex": _uuid.uuid4().hex} + + +@tool("hex_to_text", "encoding", ["hex"], "Decode hex bytes to text.") +def hex_to_text(hex=""): + try: + raw = bytes.fromhex(hex) + except ValueError as e: + return {"error": str(e)} + return {"text": raw.decode("utf-8", "replace"), "bytes": len(raw)} + + +@tool("text_to_hex", "encoding", ["text"], "Encode text to hex.") +def text_to_hex(text=""): + return {"hex": text.encode().hex()} + + +# --------------------------------------------------------------------------- +# text / regex / string utils +# --------------------------------------------------------------------------- +@tool("regex_test", "text", ["pattern", "text"], "Test a regex; return matches + groups.") +def regex_test(pattern="", text=""): + try: + rx = re.compile(pattern) + except re.error as e: + return {"error": f"bad regex: {e}"} + matches = [{"match": m.group(0), "groups": list(m.groups()), "span": m.span()} + for m in rx.finditer(text)] + return {"valid": True, "count": len(matches), "matches": matches[:50]} + + +@tool("cron_explain", "text", ["expr"], "Explain a 5-field cron expression.") +def cron_explain(expr=""): + fields = expr.split() + if len(fields) != 5: + return {"error": "need 5 fields: min hour dom mon dow"} + names = ["minute", "hour", "day-of-month", "month", "day-of-week"] + return {names[i]: f for i, f in enumerate(fields)} + + +@tool("slugify", "text", ["text"], "Slugify text (lowercase, dashes, ascii).") +def slugify(text=""): + s = unicodedata.normalize("NFKD", text).encode("ascii", "ignore").decode() + s = re.sub(r"[^a-zA-Z0-9]+", "-", s).strip("-").lower() + return {"slug": s} + + +@tool("case_convert", "text", ["text", "to"], + "Convert case: camel/snake/kebab/pascal/title/upper/lower.") +def case_convert(text="", to="snake"): + words = [w for w in re.split(r"[^a-zA-Z0-9]+|(?<=[a-z0-9])(?=[A-Z])", text) if w] + if to == "camel": + return {"result": words[0].lower() + "".join(w.capitalize() for w in words[1:]) if words else ""} + if to == "pascal": + return {"result": "".join(w.capitalize() for w in words)} + if to == "snake": + return {"result": "_".join(w.lower() for w in words)} + if to == "kebab": + return {"result": "-".join(w.lower() for w in words)} + if to == "upper": + return {"result": text.upper()} + if to == "lower": + return {"result": text.lower()} + if to == "title": + return {"result": text.title()} + return {"error": f"unknown 'to': {to}"} + + +@tool("string_metrics", "text", ["text"], "Length/word/line/char counts + Shannon entropy.") +def string_metrics(text=""): + if not text: + return {"chars": 0, "words": 0, "lines": 0, "bytes": 0, "entropy": 0.0} + import math + freq = {} + for c in text: + freq[c] = freq.get(c, 0) + 1 + n = len(text) + entropy = -sum((v / n) * math.log2(v / n) for v in freq.values()) + return {"chars": n, "words": len(text.split()), "lines": text.count("\n") + 1, + "bytes": len(text.encode()), "unique_chars": len(freq), + "entropy": round(entropy, 3)} + + +@tool("html_decode", "text", ["html"], "Decode HTML entities.") +def html_decode(html=""): + return {"text": html.unescape(html)} + + +# --------------------------------------------------------------------------- +# numbers / time +# --------------------------------------------------------------------------- +@tool("base_convert", "numbers", ["value", "from_base", "to_base"], + "Convert a number between bases 2-36.") +def base_convert(value="", from_base="10", to_base="16"): + try: + n = int(value, int(from_base)) + except ValueError: + return {"error": "invalid value for base"} + chars = "0123456789abcdefghijklmnopqrstuvwxyz" + if not 2 <= int(to_base) <= 36: + return {"error": "to_base must be 2-36"} + if n == 0: + return {"result": "0"} + neg = n < 0 + n = abs(n) + out = "" + while n: + out = chars[n % int(to_base)] + out + n //= int(to_base) + return {"result": ("-" if neg else "") + out, + "decimal": int(value, int(from_base))} + + +@tool("epoch", "time", ["ts"], "Convert a unix epoch to human-readable UTC + relative.") +def epoch(ts=""): + try: + t = int(float(ts)) + except ValueError: + return {"error": "invalid epoch"} + return {"utc": time.strftime("%Y-%m-%d %H:%M:%S", time.gmtime(t)), + "local": time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(t)), + "age_seconds": int(time.time() - t)} + + +@tool("now", "time", [], "Current unix time (UTC).") +def now(): + t = time.time() + return {"epoch": int(t), "utc": time.strftime("%Y-%m-%d %H:%M:%S", time.gmtime(t))} + + +# --------------------------------------------------------------------------- +# helpers +# --------------------------------------------------------------------------- +def json_loads(s): + try: + return json.loads(s) + except Exception as e: + return {"error": f"invalid JSON: {e}"} + + +def diff(a, b, path=""): + out = {"added": [], "removed": [], "changed": []} + if type(a) is not type(b): + out["changed"].append(path or "") + return out + if isinstance(a, dict): + for k in set(a) | set(b): + p = f"{path}.{k}" if path else k + if k not in b: + out["removed"].append(p) + elif k not in a: + out["added"].append(p) + elif a[k] != b[k]: + d = diff(a[k], b[k], p) + out["added"] += d["added"]; out["removed"] += d["removed"]; out["changed"] += d["changed"] + elif isinstance(a, list): + if len(a) != len(b): + out["changed"].append(f"{path or ''}[length {len(a)}->{len(b)}]") + for i, (x, y) in enumerate(zip(a, b)): + if x != y: + d = diff(x, y, f"{path}[{i}]") + out["added"] += d["added"]; out["removed"] += d["removed"]; out["changed"] += d["changed"] + elif a != b: + out["changed"].append(path or "") + return out + + +# --------------------------------------------------------------------------- +# routes +# --------------------------------------------------------------------------- +@app.route("/health") +def health(): + return jsonify({"ok": True, "tools": len(TOOLS), + "categories": sorted({t["category"] for t in TOOLS.values()})}) + + +@app.route("/api/tools") +def api_catalog(): + return jsonify({name: {"label": t["label"], "category": t["category"], + "params": t["params"], "desc": t["desc"]} + for name, t in TOOLS.items()}) + + +@app.route("/api/", methods=["GET", "POST"]) +def api_run(name): + if name not in TOOLS: + return jsonify({"error": f"unknown tool: {name}"}), 404 + t = TOOLS[name] + if request.method == "POST": + args = request.get_json(silent=True) or {} + else: + args = {k: v for k, v in request.args.items()} + args = {k: args.get(k, "") for k in t["params"]} + try: + result = t["fn"](**args) + except Exception as e: + return jsonify({"tool": name, "error": f"{type(e).__name__}: {e}"}), 400 + if isinstance(result, dict) and "error" in result: + return jsonify({"tool": name, **result}), 200 + return jsonify({"tool": name, "result": result}) + + +if __name__ == "__main__": + app.run(host="0.0.0.0", port=8080)