mirror of
https://github.com/allaunthefox/Research-Stack.git
synced 2026-07-31 03:05:21 +00:00
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:
parent
568f8686c5
commit
45048db9cc
1 changed files with 213 additions and 132 deletions
|
|
@ -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__":
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue