#!/usr/bin/env python3 """ MCP server that generates Lean proofs via OpenRouter's DeepSeek V4 Flash. Exposes tools: generate_lean_proof(context: str, line_no: int) -> str verify_lean_build(module: str, workdir: str) -> str Reads the OpenRouter API key from ~/.local/share/opencode/auth.json. """ from __future__ import annotations import json import os import subprocess import sys from pathlib import Path import urllib.request import urllib.error from mcp.server.fastmcp import FastMCP mcp = FastMCP("opencode-prover", log_level="WARNING") OPENROUTER_URL = "https://openrouter.ai/api/v1/chat/completions" MODEL = "deepseek/deepseek-v4-flash" def _get_api_key() -> str: """Read the OpenRouter API key from opencode's auth.json.""" auth_path = Path.home() / ".local" / "share" / "opencode" / "auth.json" try: auth = json.loads(auth_path.read_text()) return auth.get("openrouter", {}).get("key", "") except Exception: return os.environ.get("OPENROUTER_API_KEY", "") OPENER = ( "You are a Lean 4 theorem prover for a Research Stack project. " "Use Q16_16 fixed-point arithmetic (no Float). " "Output ONLY the proof block starting with `:= by`. " "No markdown fences. No explanation. No preamble." ) PROMPT_TPL = ( "Complete this Lean theorem by replacing the `sorry`.\n\n" "{context}\n\n" "Line {line_no} has the `sorry`. Generate the proof block." ) def _call_llm(prompt: str) -> tuple[str, str]: key = _get_api_key() if not key: return "# ERROR: No OpenRouter API key found", "no_key" body = json.dumps({ "model": MODEL, "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=headers, method="POST") try: with urllib.request.urlopen(req, timeout=180) as resp: data = json.loads(resp.read()) content = data["choices"][0]["message"]["content"] return content, "" except urllib.error.HTTPError as e: err = e.read().decode() return f"# HTTP {e.code}: {err[:200]}", f"http_{e.code}" except Exception as e: return f"# ERROR: {e}", str(e) @mcp.tool() def generate_lean_proof(theorem_context: str, line_no: int, pist_label: str = "", rrc_shape: str = "logogramProjection") -> str: """Send a Lean theorem context to DeepSeek V4 Flash and return the generated proof. The proof is validated through the RRC alignment watchdog before acceptance. Args: theorem_context: The Lean theorem text with context line_no: Line number of the sorry pist_label: Optional PIST label for RRC alignment gate rrc_shape: Optional RRC shape for alignment gate """ prompt = f"{OPENER}\n\n{PROMPT_TPL.format(context=theorem_context, line_no=line_no)}" for attempt in range(2): content, tag = _call_llm(prompt) if content.startswith("#"): if attempt == 0 and "401" in tag: return "# ERROR: OpenRouter auth failed — check key" return content proof = _extract_proof(content) if not proof: continue # RRC watchdog: classify the proof attempt if pist_label and os.path.exists(RRC_WATCHDOG): try: wd_result = subprocess.run( [RRC_WATCHDOG, "--pist-label", pist_label, "--exact-label", pist_label, "--rrc-shape", rrc_shape], capture_output=True, text=True, timeout=15, ) try: wd = json.loads(wd_result.stdout) score = wd.get("score", 0) if score < WATCHDOG_MIN_SCORE: # Low alignment — retry with more context if attempt == 0: continue except (json.JSONDecodeError, KeyError): pass # watchdog unavailable, accept anyway except (subprocess.TimeoutExpired, FileNotFoundError): pass # watchdog not built, accept anyway return proof return f"# Could not extract proof (watchdog rejected)\n\n{content[:500]}" RRC_WATCHDOG = os.path.join( os.environ.get("LAKE_WORKDIR", "/home/allaun/Research Stack/0-Core-Formalism/lean/Semantics"), ".lake/build/bin/rrc-watchdog" ) WATCHDOG_MIN_SCORE = 72 # compatibleStructuralProjection @mcp.tool() def classify_proof(pist_label: str = "", rrc_shape: str = "logogramProjection") -> str: """Run a proof through the RRC alignment watchdog and return JSON result. Args: pist_label: PIST structural label (e.g. 'LogogramProjection') rrc_shape: RRC semantic shape name (e.g. 'cognitiveLoadField', 'signalShapedRouteCompiler') """ try: result = subprocess.run( [RRC_WATCHDOG, "--pist-label", pist_label or "LogogramProjection", "--exact-label", pist_label or "LogogramProjection", "--rrc-shape", rrc_shape], capture_output=True, text=True, timeout=30, ) return result.stdout or f'{{"error": "empty", "stderr": "{result.stderr[:200]}"}}' except Exception as e: return json.dumps({"error": str(e)[:200]}) @mcp.tool() def verify_lean_build(module_name: str, workdir: str = "") -> str: """Run `lake build ` and return JSON result.""" wd = workdir or os.environ.get("LAKE_WORKDIR", "/home/allaun/Research Stack/0-Core-Formalism/lean/Semantics") try: result = subprocess.run( ["lake", "build", module_name], capture_output=True, text=True, timeout=240, cwd=wd, ) ok = result.returncode == 0 errors = "\n".join( l for l in (result.stdout + result.stderr).split("\n") if "error:" in l or "sorry" in l )[:1500] return json.dumps({"passed": ok, "returncode": result.returncode, "errors": errors}) except subprocess.TimeoutExpired: return json.dumps({"passed": False, "errors": "TIMEOUT"}) except Exception as e: return json.dumps({"passed": False, "errors": str(e)[:500]}) def _extract_proof(text: str) -> str: lines = text.split("\n") in_proof = False proof_lines: list[str] = [] for line in lines: s = line.strip() if s.startswith(":= by"): in_proof = True proof_lines = [s] continue if in_proof: if s.startswith("```"): continue if s == "" and len(proof_lines) < 2: continue 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 "" if __name__ == "__main__": mcp.run(transport="stdio")