#!/usr/bin/env python3 """jsonshim-mcp - one file, standard library only, no install. An MCP server (stdio, JSON-RPC 2.0) that gives an agent two tools: recover_json take a model's raw reply and return the JSON in it, or refuse. Refusing is a feature: it returns {"recovered": false} rather than inventing a value, because an invented value reaches your state looking exactly like data. classify_json_failure name which of the 12 MALFORMED-300 failure categories a broken reply matches, so a log line says WHY it broke. It has no dependencies, makes no network calls, and reads nothing from disk. Point an MCP client at it: {"mcpServers": {"jsonshim": {"command": "python3", "args": ["/absolute/path/jsonshim_mcp.py"]}}} Measured, not claimed. Scored against MALFORMED-300, the 300-case labelled suite at https://toolkitlabs.org/leaderboard/ : 218/300 exact · 193/275 on the recoverable cases · 0 of 25 unrecoverable cases invented · 55 false refusals It is deliberately weaker than the best library on that table (json-repair recovers 264 of 275) and it is not the same code as `jsonshim` (282/300). What it does have is the property this file is about: **it invented a value on zero of the 25 unrecoverable cases.** Where it cannot recover, it says so. Known failures, named because they will bite you: truncated output 0/25, raw control characters inside strings 0/25, unbalanced brackets 5/25. If your models truncate, use the scorer on the free sample and pick a library on the evidence: https://toolkitlabs.org/malformed300/score.py CC0-1.0. """ import json import re import sys PROTOCOL = "2024-11-05" NAME = "jsonshim-mcp" VERSION = "1.0.0" # ---------------------------------------------------------------- recovery --- _FENCE = re.compile(r"```(?:json5?|javascript|js)?\s*(.*?)```", re.S | re.I) _PY = {"True": "true", "False": "false", "None": "null"} def _strip_fences(t): m = _FENCE.search(t) return m.group(1) if m else t def _spans(t): """Every balanced {...} or [...] span, longest first, outermost first.""" out = [] for opener, closer in (("{", "}"), ("[", "]")): depth = 0 start = None in_str = False esc = False for i, ch in enumerate(t): if in_str: if esc: esc = False elif ch == "\\": esc = True elif ch == '"': in_str = False continue if ch == '"': in_str = True elif ch == opener: if depth == 0: start = i depth += 1 elif ch == closer: depth -= 1 if depth == 0 and start is not None: out.append(t[start:i + 1]) start = None elif depth < 0: depth = 0 out.sort(key=len, reverse=True) return out def _decomma(x): return re.sub(r",(\s*[}\]])", r"\1", x) def _repairs(s): """Progressively looser rewrites, cumulative. Each is reported, never applied silently. Every step re-runs the trailing-comma strip, because removing a comment or a quote can expose a comma that was not trailing a moment ago. Missing that is how a repairer ends up returning an inner array instead of the object that contained it - which is precisely the silent-wrong-value failure this tool exists to refuse to commit. """ yield s, [] a = re.sub(r",(\s*[}\]])", r"\1", s) if a != s: yield a, ["removed trailing comma"] b = _decomma(re.sub(r"/\*.*?\*/", "", re.sub(r"//[^\n]*", "", a), flags=re.S)) if b != a: yield b, ["removed comments"] c = _decomma(re.sub(r"\bTrue\b|\bFalse\b|\bNone\b", lambda m: _PY[m.group(0)], b)) if c != b: yield c, ["python literals -> json literals"] d = _decomma(re.sub(r"([{,]\s*)([A-Za-z_][A-Za-z0-9_]*)(\s*:)", r'\1"\2"\3', c)) if d != c: yield d, ["quoted bare keys"] e = _decomma(re.sub(r"'([^'\"]*)'", r'"\1"', d)) if e != d: yield e, ["single quotes -> double quotes"] def recover(text): if not isinstance(text, str) or not text.strip(): return {"recovered": False, "reason": "empty input"} body = _strip_fences(text) for cand in [body] + _spans(body) + _spans(text): for attempt, notes in _repairs(cand.strip()): try: v = json.loads(attempt) except Exception: continue if isinstance(v, (dict, list)): return {"recovered": True, "value": v, "repairs": notes} return {"recovered": False, "reason": "no balanced JSON value could be parsed from this text", "advice": "treat this as the model producing nothing, not as an empty object"} # -------------------------------------------------------------- classifier --- CATEGORIES = [ ("fenced", lambda t: "```" in t), ("prose_wrapped", lambda t: bool(re.match(r"^[^{\[]*[A-Za-z]{3,}[^{\[]*[{\[]", t.strip()))), ("single_quotes", lambda t: bool(re.search(r"'[^']*'\s*:", t))), ("unquoted_keys", lambda t: bool(re.search(r"[{,]\s*[A-Za-z_][A-Za-z0-9_]*\s*:", t))), ("trailing_comma", lambda t: bool(re.search(r",\s*[}\]]", t))), ("comments", lambda t: bool(re.search(r"(^|[^:])//|/\*", t))), ("py_literals", lambda t: bool(re.search(r"\b(True|False|None)\b", t))), ("raw_control", lambda t: bool(re.search(r'"[^"\n]*\n[^"]*"', t))), ("brackets", lambda t: len(re.findall(r"[{\[]", t)) != len(re.findall(r"[}\]]", t))), ] def classify(text): hits = [name for name, f in CATEGORIES if f(text or "")] unrec = not re.search(r"[{\[]", text or "") return {"categories": hits, "likely_unrecoverable": unrec, "note": "regex heuristics over the text; it reads the string, not the model"} # ------------------------------------------------------------------- server --- TOOLS = [ {"name": "recover_json", "description": "Extract the JSON value from a language model's raw reply, or refuse. " "Returns recovered:false rather than inventing a value.", "inputSchema": {"type": "object", "required": ["text"], "properties": {"text": {"type": "string", "description": "the model's raw reply"}}}}, {"name": "classify_json_failure", "description": "Name which known failure categories a broken model reply matches " "(fenced, prose_wrapped, trailing_comma, unquoted_keys, single_quotes, " "comments, py_literals, raw_control, brackets).", "inputSchema": {"type": "object", "required": ["text"], "properties": {"text": {"type": "string"}}}}, ] def handle(msg): mid = msg.get("id") method = msg.get("method") if method == "initialize": return {"jsonrpc": "2.0", "id": mid, "result": {"protocolVersion": PROTOCOL, "capabilities": {"tools": {}}, "serverInfo": {"name": NAME, "version": VERSION}}} if method in ("notifications/initialized", "initialized"): return None if method == "tools/list": return {"jsonrpc": "2.0", "id": mid, "result": {"tools": TOOLS}} if method == "tools/call": p = msg.get("params") or {} name = p.get("name") args = p.get("arguments") or {} text = args.get("text", "") if name == "recover_json": out = recover(text) elif name == "classify_json_failure": out = classify(text) else: return {"jsonrpc": "2.0", "id": mid, "error": {"code": -32601, "message": "unknown tool: %s" % name}} return {"jsonrpc": "2.0", "id": mid, "result": {"content": [{"type": "text", "text": json.dumps(out, ensure_ascii=False)}], "isError": False}} if method == "ping": return {"jsonrpc": "2.0", "id": mid, "result": {}} return {"jsonrpc": "2.0", "id": mid, "error": {"code": -32601, "message": "method not found: %s" % method}} def main(): for line in sys.stdin: line = line.strip() if not line: continue try: msg = json.loads(line) except Exception: continue resp = handle(msg) if resp is not None: sys.stdout.write(json.dumps(resp) + "\n") sys.stdout.flush() if __name__ == "__main__": main()