feat(infra): token-saver MCP — free local compute for any MCP client

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.
This commit is contained in:
allaun 2026-06-16 20:04:40 -05:00
parent f9951cbf07
commit 349c5944ab

View file

@ -1,18 +1,22 @@
#!/usr/bin/env python3 #!/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): Free tier (runs on neon CPU or qfox-1 build server, 0 token cost):
- verify_build: qfox-1 build server (HTTP) - verify_build lake build verification
- classify_proof: rrc-watchdog (local Lean exe) - classify_proof RRC alignment classification
- compile_check: local lean --run via SSH build server - compile_check Lean snippet compilation
- generate_proof: neon's local Ollama models (CPU, free) - 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): Paid fallback (only if free tier fails):
- generate_proof OpenRouter DeepSeek V4 Flash (only if local models fail) - generate_proof OpenRouter DeepSeek V4 Flash
Usage: Usage: python3 token_saver_mcp.py
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 from __future__ import annotations
@ -21,161 +25,94 @@ import json
import os import os
import subprocess import subprocess
import sys import sys
import urllib.request
import urllib.error import urllib.error
import urllib.request
from pathlib import Path from pathlib import Path
from mcp.server.fastmcp import FastMCP from mcp.server.fastmcp import FastMCP
mcp = FastMCP("token-saver", log_level="WARNING") 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") 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_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") NEON_MODEL = os.environ.get("NEON_MODEL",
LAKE_WORKDIR = os.environ.get("LAKE_WORKDIR", "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") "/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" TOOL_MANIFEST = """
OPENROUTER_MODEL = "deepseek/deepseek-v4-flash" 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: verify_build(module) build server (free)
auth_path = Path.home() / ".local" / "share" / "opencode" / "auth.json" 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: try:
auth = json.loads(auth_path.read_text()) resp = urllib.request.urlopen(f"{BUILD_SERVER}/{module}", timeout=timeout)
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)
return resp.read().decode() return resp.read().decode()
except Exception as e: except Exception as e:
return json.dumps({"passed": False, "error": str(e)[:200]}) return json.dumps({"passed": False, "error": str(e)[:200]})
@mcp.tool() def _call_neon(prompt: str, raw: bool = True) -> str:
def classify_proof(pist_label: str = "", rrc_shape: str = "logogramProjection") -> str: """Call neon's local Ollama model. 0 tokens."""
"""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"
body = json.dumps({ body = json.dumps({
"model": NEON_MODEL, "model": NEON_MODEL,
"prompt": prompt, "prompt": prompt,
"raw": True, "raw": raw,
"stream": False, "stream": False,
"options": {"temperature": 0, "num_predict": 512, "num_thread": 18}, "options": {"temperature": 0, "num_predict": 512, "num_thread": 18},
}).encode() }).encode()
try: try:
req = urllib.request.Request( req = urllib.request.Request(
f"{NEON_OLLAMA}/api/generate", data=body, f"{NEON_OLLAMA}/api/generate", data=body,
headers={"Content-Type": "application/json"}, headers={"Content-Type": "application/json"}, method="POST",
method="POST",
) )
with urllib.request.urlopen(req, timeout=300) as resp: with urllib.request.urlopen(req, timeout=300) as resp:
data = json.loads(resp.read()) return json.loads(resp.read()).get("response", "")
text = data.get("response", "")
return _extract_proof(text)
except Exception as e: except Exception as e:
return f"# neon error: {e}" return f"# neon error: {e}"
def _try_openrouter(theorem_context: str, line_no: int) -> str: def _call_openrouter(prompt: str) -> str:
"""Fallback: generate proof via OpenRouter (token cost).""" """Fallback: paid OpenRouter API."""
key = _get_openrouter_key() 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: if not key:
return "# No OpenRouter key configured" return "# No API 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}"
)
body = json.dumps({ body = json.dumps({
"model": OPENROUTER_MODEL, "model": "deepseek/deepseek-v4-flash",
"messages": [{"role": "user", "content": prompt}], "messages": [{"role": "user", "content": prompt}],
"temperature": 0.3, "max_tokens": 4096, "temperature": 0.3, "max_tokens": 4096,
}).encode() }).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: 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: with urllib.request.urlopen(req, timeout=180) as resp:
data = json.loads(resp.read()) return json.loads(resp.read())["choices"][0]["message"]["content"]
content = data["choices"][0]["message"]["content"]
return _extract_proof(content) or content[:500]
except Exception as e: except Exception as e:
return f"# OpenRouter error: {e}" return f"# OpenRouter error: {e}"
@ -190,17 +127,161 @@ def _extract_proof(text: str) -> str:
in_proof = True in_proof = True
proof_lines = [s] proof_lines = [s]
continue continue
if in_proof: if in_proof and not s.startswith("```"):
if s.startswith("```"):
continue
proof_lines.append(s) proof_lines.append(s)
if proof_lines: return "\n".join(proof_lines) if proof_lines else ""
return "\n".join(proof_lines)
for line in lines:
s = line.strip() # ═══════════════════════════════════════════════════════════════════════════
if s.startswith(":= by") or s.startswith("by "): # Tools
return s # ═══════════════════════════════════════════════════════════════════════════
return ""
@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__": if __name__ == "__main__":