From 349c5944abedc5fa5e71e91d8b5f0aefba1c2a79 Mon Sep 17 00:00:00 2001 From: allaun Date: Tue, 16 Jun 2026 20:04:40 -0500 Subject: [PATCH] =?UTF-8?q?feat(infra):=20token-saver=20MCP=20=E2=80=94=20?= =?UTF-8?q?free=20local=20compute=20for=20any=20MCP=20client?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 8 tools, all routing to free compute first: verify_build, classify_proof, compile_check, linter_check, generate_proof (neon → paid fallback), format_code, doc_lookup, health Any MCP-compatible agent (Claude Code, OpenCode, Cursor) can call these tools and avoid burning API tokens on compilation/verification. --- .../tools-scripts/mcp/token_saver_mcp.py | 345 +++++++++++------- 1 file changed, 213 insertions(+), 132 deletions(-) diff --git a/5-Applications/tools-scripts/mcp/token_saver_mcp.py b/5-Applications/tools-scripts/mcp/token_saver_mcp.py index 2a0366f0..cf3f97b8 100644 --- a/5-Applications/tools-scripts/mcp/token_saver_mcp.py +++ b/5-Applications/tools-scripts/mcp/token_saver_mcp.py @@ -1,18 +1,22 @@ #!/usr/bin/env python3 """ -TokenSaver MCP — routes requests to free local compute first, falls through to paid API. +TokenSaver MCP — intercepts token-burning operations, routes to free local compute. -Free tier (no tokens): - - verify_build: qfox-1 build server (HTTP) - - classify_proof: rrc-watchdog (local Lean exe) - - compile_check: local lean --run via SSH build server - - generate_proof: neon's local Ollama models (CPU, free) +Free tier (runs on neon CPU or qfox-1 build server, 0 token cost): + - verify_build lake build verification + - classify_proof RRC alignment classification + - compile_check Lean snippet compilation + - linter_check Lean linter pass + - generate_proof DeepSeek-Prover on neon (CPU, free) + - format_code Lean code formatting (lake fmt) + - doc_lookup Mathlib doc search (local cache) -Paid fallback (uses tokens): - - generate_proof → OpenRouter DeepSeek V4 Flash (only if local models fail) +Paid fallback (only if free tier fails): + - generate_proof → OpenRouter DeepSeek V4 Flash -Usage: - python3 token_saver_mcp.py +Usage: python3 token_saver_mcp.py +Register in opencode.json as a local MCP server. +Any MCP-compatible agent (Claude Code, OpenCode, Cursor) can call it. """ from __future__ import annotations @@ -21,161 +25,94 @@ import json import os import subprocess import sys -import urllib.request import urllib.error +import urllib.request from pathlib import Path from mcp.server.fastmcp import FastMCP mcp = FastMCP("token-saver", log_level="WARNING") -# ── Free resources ────────────────────────────────────────────────────── +# ── Resource URLs (configurable via env) ──────────────────────────────── BUILD_SERVER = os.environ.get("BUILD_SERVER_URL", "http://100.88.57.96:8765") -NEON_OLLAMA = os.environ.get("NEON_OLLAMA_URL", "http://100.92.88.64:11434") -NEON_MODEL = os.environ.get("NEON_MODEL", "hf.co/irmma/DeepSeek-Prover-V2-7B-Q4_K_M-GGUF") -LAKE_WORKDIR = os.environ.get("LAKE_WORKDIR", +NEON_OLLAMA = os.environ.get("NEON_OLLAMA_URL", "http://100.92.88.64:11434") +NEON_MODEL = os.environ.get("NEON_MODEL", + "hf.co/irmma/DeepSeek-Prover-V2-7B-Q4_K_M-GGUF") +LAKE_WORKDIR = os.environ.get("LAKE_WORKDIR", "/home/allaun/Research Stack/0-Core-Formalism/lean/Semantics") -RRC_WATCHDOG = os.path.join(LAKE_WORKDIR, ".lake/build/bin/rrc-watchdog") -# ── Paid fallback ─────────────────────────────────────────────────────── +# ── Registered tools ──────────────────────────────────────────────────── -OPENROUTER_URL = "https://openrouter.ai/api/v1/chat/completions" -OPENROUTER_MODEL = "deepseek/deepseek-v4-flash" +TOOL_MANIFEST = """ +All tools in this server run on free local compute (neon CPU / qfox-1 build server). +Zero API tokens consumed unless explicitly marked [PAID]. -def _get_openrouter_key() -> str: - auth_path = Path.home() / ".local" / "share" / "opencode" / "auth.json" + verify_build(module) → build server (free) + classify_proof(label,shape) → rrc-watchdog (free) + compile_check(code) → lake env lean (free) + linter_check(code) → lake build --warnings (free) + generate_proof(context) → neon Ollama → OpenRouter fallback [free → PAID] + format_code(code) → lake fmt (free) + doc_lookup(query) → local mathlib doc cache (free) +""" + + +def _call_build_server(module: str, timeout: int = 600) -> str: + """Call the local build server. 0 tokens.""" try: - auth = json.loads(auth_path.read_text()) - return auth.get("openrouter", {}).get("key", "") - except Exception: - return os.environ.get("OPENROUTER_API_KEY", "") - -# ── Tools: Free tier ──────────────────────────────────────────────────── - -@mcp.tool() -def verify_build(module: str) -> str: - """Verify a Lean module compiles. Uses local build server (free, 0 tokens).""" - try: - resp = urllib.request.urlopen(f"{BUILD_SERVER}/{module}", timeout=600) + resp = urllib.request.urlopen(f"{BUILD_SERVER}/{module}", timeout=timeout) return resp.read().decode() except Exception as e: return json.dumps({"passed": False, "error": str(e)[:200]}) -@mcp.tool() -def classify_proof(pist_label: str = "", rrc_shape: str = "logogramProjection") -> str: - """Classify a proof through the RRC alignment gate. Local exe (free, 0 tokens).""" - if not os.path.exists(RRC_WATCHDOG): - return json.dumps({"error": "rrc-watchdog not built", "score": 0}) - try: - result = subprocess.run( - [RRC_WATCHDOG, "--pist-label", pist_label or "none", - "--exact-label", pist_label or "none", "--rrc-shape", rrc_shape], - capture_output=True, text=True, timeout=30, - ) - return result.stdout or json.dumps({"error": "empty", "score": 0}) - except Exception as e: - return json.dumps({"error": str(e)[:200], "score": 0}) - - -@mcp.tool() -def compile_check(code: str) -> str: - """Compile-check a Lean snippet. Uses build server (free, 0 tokens). - Writes code to a temp file and runs `lake build` on it. - """ - import tempfile - tmp = Path(tempfile.mkdtemp()) / "check.lean" - tmp.write_text(code) - try: - result = subprocess.run( - ["lake", "env", "lean", str(tmp)], - capture_output=True, text=True, timeout=120, - cwd=LAKE_WORKDIR, - ) - ok = result.returncode == 0 - errors = "\n".join(l for l in (result.stdout + result.stderr).split("\n") - if "error:" in l)[:1000] - return json.dumps({"passed": ok, "errors": errors}) - except Exception as e: - return json.dumps({"passed": False, "errors": str(e)[:200]}) - finally: - tmp.unlink(missing_ok=True) - - -@mcp.tool() -def generate_proof(theorem_context: str, line_no: int = 0) -> str: - """Generate a Lean proof. Tries free local models first, falls back to paid API. - - Free tier: neon's DeepSeek-Prover-V2 (CPU, 0 token cost) - Paid fallback: OpenRouter DeepSeek V4 Flash (uses API tokens) - """ - # Try free: neon Ollama (raw completion mode) - proof = _try_neon(theorem_context) - if proof and not proof.startswith("#"): - return proof - - # Fallback: paid OpenRouter - return _try_openrouter(theorem_context, line_no) - - -# ── Internals ─────────────────────────────────────────────────────────── - -def _try_neon(theorem_context: str) -> str: - """Try generating a proof using neon's free local model.""" - prompt = f"theorem {theorem_context} := by" +def _call_neon(prompt: str, raw: bool = True) -> str: + """Call neon's local Ollama model. 0 tokens.""" body = json.dumps({ "model": NEON_MODEL, "prompt": prompt, - "raw": True, + "raw": raw, "stream": False, "options": {"temperature": 0, "num_predict": 512, "num_thread": 18}, }).encode() - try: req = urllib.request.Request( f"{NEON_OLLAMA}/api/generate", data=body, - headers={"Content-Type": "application/json"}, - method="POST", + headers={"Content-Type": "application/json"}, method="POST", ) with urllib.request.urlopen(req, timeout=300) as resp: - data = json.loads(resp.read()) - text = data.get("response", "") - return _extract_proof(text) + return json.loads(resp.read()).get("response", "") except Exception as e: return f"# neon error: {e}" -def _try_openrouter(theorem_context: str, line_no: int) -> str: - """Fallback: generate proof via OpenRouter (token cost).""" - key = _get_openrouter_key() +def _call_openrouter(prompt: str) -> str: + """Fallback: paid OpenRouter API.""" + auth_path = Path.home() / ".local" / "share" / "opencode" / "auth.json" + key = "" + try: + auth = json.loads(auth_path.read_text()) + key = auth.get("openrouter", {}).get("key", "") + except Exception: + key = os.environ.get("OPENROUTER_API_KEY", "") if not key: - return "# No OpenRouter key configured" - - prompt = ( - "You are a Lean 4 theorem prover. " - "Output ONLY the proof block starting with `:= by`. " - "No markdown fences. No explanation.\n\n" - f"Complete this Lean theorem:\n\n{theorem_context}" - ) + return "# No API key configured" body = json.dumps({ - "model": OPENROUTER_MODEL, + "model": "deepseek/deepseek-v4-flash", "messages": [{"role": "user", "content": prompt}], "temperature": 0.3, "max_tokens": 4096, }).encode() + headers = {"Content-Type": "application/json", "Authorization": f"Bearer {key}"} - req = urllib.request.Request( - OPENROUTER_URL, data=body, - headers={"Content-Type": "application/json", - "Authorization": f"Bearer {key}"}, - method="POST", - ) try: + req = urllib.request.Request( + "https://openrouter.ai/api/v1/chat/completions", data=body, + headers=headers, method="POST", + ) with urllib.request.urlopen(req, timeout=180) as resp: - data = json.loads(resp.read()) - content = data["choices"][0]["message"]["content"] - return _extract_proof(content) or content[:500] + return json.loads(resp.read())["choices"][0]["message"]["content"] except Exception as e: return f"# OpenRouter error: {e}" @@ -190,17 +127,161 @@ def _extract_proof(text: str) -> str: in_proof = True proof_lines = [s] continue - if in_proof: - if s.startswith("```"): - continue + if in_proof and not s.startswith("```"): proof_lines.append(s) - if proof_lines: - return "\n".join(proof_lines) - for line in lines: - s = line.strip() - if s.startswith(":= by") or s.startswith("by "): - return s - return "" + return "\n".join(proof_lines) if proof_lines else "" + + +# ═══════════════════════════════════════════════════════════════════════════ +# Tools +# ═══════════════════════════════════════════════════════════════════════════ + +@mcp.tool() +def verify_build(module: str) -> str: + """[FREE] Verify a Lean module compiles via local build server.""" + return _call_build_server(module) + + +@mcp.tool() +def classify_proof(pist_label: str = "", rrc_shape: str = "logogramProjection") -> str: + """[FREE] Classify a proof through RRC alignment gate (local rrc-watchdog).""" + watchdog = os.path.join(LAKE_WORKDIR, ".lake/build/bin/rrc-watchdog") + if not os.path.exists(watchdog): + return json.dumps({"error": "rrc-watchdog not built", "score": 0}) + try: + result = subprocess.run( + [watchdog, "--pist-label", pist_label or "none", + "--exact-label", pist_label or "none", "--rrc-shape", rrc_shape], + capture_output=True, text=True, timeout=30, + ) + return result.stdout or json.dumps({"error": "empty", "score": 0}) + except Exception as e: + return json.dumps({"error": str(e)[:200], "score": 0}) + + +@mcp.tool() +def compile_check(code: str) -> str: + """[FREE] Compile-check a Lean snippet locally. 0 tokens.""" + import tempfile, shutil + tmp_dir = Path(tempfile.mkdtemp()) + try: + tmp_file = tmp_dir / "check.lean" + tmp_file.write_text(code) + result = subprocess.run( + ["lake", "env", "lean", str(tmp_file)], + capture_output=True, text=True, timeout=120, cwd=LAKE_WORKDIR, + ) + ok = result.returncode == 0 + errors = "\n".join( + l for l in (result.stdout + result.stderr).split("\n") + if "error:" in l or "warning:" in l + )[:2000] + return json.dumps({"passed": ok, "errors": errors}) + finally: + shutil.rmtree(tmp_dir, ignore_errors=True) + + +@mcp.tool() +def linter_check(code: str) -> str: + """[FREE] Run Lean linter on a snippet. 0 tokens.""" + import tempfile, shutil + tmp_dir = Path(tempfile.mkdtemp()) + try: + tmp_file = tmp_dir / "check.lean" + tmp_file.write_text(code) + result = subprocess.run( + ["lake", "env", "lean", f"--lint={str(tmp_file)}"], + capture_output=True, text=True, timeout=120, cwd=LAKE_WORKDIR, + ) + return json.dumps({ + "passed": result.returncode == 0, + "warnings": "\n".join( + l for l in (result.stdout + result.stderr).split("\n") + if "warning:" in l + )[:2000], + }) + finally: + shutil.rmtree(tmp_dir, ignore_errors=True) + + +@mcp.tool() +def generate_proof(theorem_context: str, line_no: int = 0) -> str: + """[FREE → PAID] Generate a Lean proof. Tries neon's CPU model first (0 tokens), + falls back to OpenRouter DeepSeek V4 Flash (token cost).""" + # Free tier: neon's DeepSeek-Prover with raw completion + prompt = f"theorem {theorem_context} := by" + result = _call_neon(prompt) + proof = _extract_proof(result) + + if proof: + return proof + + # Paid fallback: OpenRouter + paid_prompt = ( + "You are a Lean 4 theorem prover. " + "Output ONLY the proof block starting with `:= by`. " + "No markdown fences.\n\n" + f"Complete:\n\n{theorem_context}" + ) + result = _call_openrouter(paid_prompt) + proof = _extract_proof(result) + return proof or result[:500] + + +@mcp.tool() +def format_code(code: str) -> str: + """[FREE] Format Lean code using lake fmt. 0 tokens.""" + import tempfile, shutil + tmp_dir = Path(tempfile.mkdtemp()) + try: + tmp_file = tmp_dir / "check.lean" + tmp_file.write_text(code) + result = subprocess.run( + ["lake", "fmt", str(tmp_file)], + capture_output=True, text=True, timeout=30, cwd=LAKE_WORKDIR, + ) + if result.returncode == 0: + return tmp_file.read_text() + return code # Return original if fmt fails + finally: + shutil.rmtree(tmp_dir, ignore_errors=True) + + +@mcp.tool() +def doc_lookup(query: str) -> str: + """[FREE] Look up a Lean/mathlib symbol in local docs. 0 tokens.""" + try: + result = subprocess.run( + ["lake", "env", "lean", "--run", "-e", + f"#check {query}"], + capture_output=True, text=True, timeout=30, cwd=LAKE_WORKDIR, + ) + output = (result.stdout + result.stderr)[:1000] + return output or f"Symbol '{query}' not found" + except Exception as e: + return f"Error: {e}" + + +@mcp.tool() +def health() -> str: + """Check which free resources are reachable.""" + results = {} + # Build server + try: + urllib.request.urlopen(f"{BUILD_SERVER}/Semantics", timeout=5) + results["build_server"] = "up" + except Exception: + results["build_server"] = "down" + # Neon Ollama + try: + urllib.request.urlopen(f"{NEON_OLLAMA}/api/tags", timeout=5) + results["neon_ollama"] = "up" + except Exception: + results["neon_ollama"] = "down" + # rrc-watchdog + watchdog = os.path.join(LAKE_WORKDIR, ".lake/build/bin/rrc-watchdog") + results["rrc_watchdog"] = "present" if os.path.exists(watchdog) else "missing" + return json.dumps(results) if __name__ == "__main__":