#!/usr/bin/env python3 """ Humanizer web API + MCP server. Standard library only. Wraps humanizer.py (the guide's scorer) without changing it. python3 server.py HTTP on 127.0.0.1:8089 (put nginx in front) python3 server.py --port 9000 other port python3 server.py --stdio MCP over stdin/stdout for local clients HTTP: GET /api/health GET /api/rules POST /api/check {"text": "...", "channel": "email"} POST /api/rewrite {"text": "...", "channel": "email"} (rule-based rewrite, then re-scored) POST /mcp (JSON-RPC 2.0, MCP tools check_text, rewrite_text and get_rules) """ import json import os import sys from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import humanizer as H # noqa: E402 import rewrite as R # noqa: E402 MAX_TEXT = 20000 CHANNELS = [c for c in H.CHANNEL_WORD_LIMITS] def rules(): return { "banned_words": sorted(set(H.BANNED_WORDS)), "banned_phrase_patterns": H.BANNED_PHRASES, "channel_word_limits": H.CHANNEL_WORD_LIMITS, "preferred_contractions": [{"expanded": a.strip(), "contracted": b} for a, b in H.CONTRACTION_FORMS], "pass_rule": "No hard issues and score >= 70. Score starts at 100 and loses points per issue.", "hard_checks": [ "em-dash, en-dash or spaced hyphen", "banned words and phrases", "3+ item comma lists", "sentence and paragraph burstiness (too uniform, all long, all short)", "same two-word sentence opener 3+ times", "sloppy grammar, missing apostrophes, lowercase i, lowercase sentence starts", "channel word limit (whatsapp 55, dm 80, email 120)", "email address inside the body", ], "soft_checks": ["expanded forms that people usually contract", "body not ending in . ! or ?"], "note": "Stdlib heuristic only. No neural detector and no live GPTZero check is run.", } def check(text, channel="generic"): if not isinstance(text, str) or not text.strip(): raise ValueError("text must be a non-empty string") if len(text) > MAX_TEXT: raise ValueError(f"text too long (max {MAX_TEXT} characters)") channel = (channel or "generic").lower() if channel not in CHANNELS: raise ValueError("channel must be one of: " + ", ".join(CHANNELS)) return H.evaluate_text(text, channel=channel) def do_rewrite(text, channel="generic"): check(text, channel) # same validation as check return R.rewrite(text, (channel or "generic").lower()) TOOLS = [ { "name": "check_text", "description": "Score a draft (cold email, DM, script or web copy) with the Humanizer stdlib heuristic. Returns passed, verdict (HUMAN_HEURISTIC or AI_OR_RULE_VIOLATION), a 0-100 heuristic score, burstiness stats, the issues to fix and soft warnings. It is a pre-filter, not a guarantee against commercial AI detectors.", "inputSchema": { "type": "object", "properties": { "text": {"type": "string", "description": "The draft to check"}, "channel": {"type": "string", "enum": CHANNELS, "description": "Sets the word cap: whatsapp 55, dm 80, email 120, generic none"}, }, "required": ["text"], }, }, { "name": "rewrite_text", "description": "Rewrite a draft to remove the tells check_text flags: banned words and phrases, dashes, three-item lists, expanded contractions, sloppy apostrophes and capitalization, over-long sentences. Rule-based, so it can shift meaning slightly: read the result. Returns the rewritten text, the score before and after, every change made, and any issues it could not fix.", "inputSchema": { "type": "object", "properties": { "text": {"type": "string", "description": "The draft to rewrite"}, "channel": {"type": "string", "enum": CHANNELS, "description": "Sets the word cap used for the re-check"}, }, "required": ["text"], }, }, { "name": "get_rules", "description": "Return the rules the checker uses: banned words, banned phrase patterns, channel word limits, preferred contractions, and how the score and pass rule work. Read this before rewriting a draft.", "inputSchema": {"type": "object", "properties": {}}, }, ] def tool_result(obj, err=False): return {"content": [{"type": "text", "text": json.dumps(obj, ensure_ascii=False, indent=2)}], "isError": err} def rpc(msg): """Handle one JSON-RPC message; return a response dict or None for notifications.""" if not isinstance(msg, dict) or "id" not in msg: return None mid, method, params = msg["id"], msg.get("method", ""), msg.get("params") or {} try: if method == "initialize": res = { "protocolVersion": params.get("protocolVersion", "2024-11-05"), "capabilities": {"tools": {}}, "serverInfo": {"name": "humanizer", "version": "1.0.0"}, } elif method == "ping": res = {} elif method == "tools/list": res = {"tools": TOOLS} elif method == "tools/call": name, args = params.get("name"), params.get("arguments") or {} if name == "check_text": try: res = tool_result(check(args.get("text"), args.get("channel", "generic"))) except ValueError as e: res = tool_result({"error": str(e)}, True) elif name == "rewrite_text": try: res = tool_result(do_rewrite(args.get("text"), args.get("channel", "generic"))) except ValueError as e: res = tool_result({"error": str(e)}, True) elif name == "get_rules": res = tool_result(rules()) else: return {"jsonrpc": "2.0", "id": mid, "error": {"code": -32602, "message": f"unknown tool {name}"}} else: return {"jsonrpc": "2.0", "id": mid, "error": {"code": -32601, "message": f"method not found: {method}"}} except Exception as e: # never crash the server on a bad request return {"jsonrpc": "2.0", "id": mid, "error": {"code": -32603, "message": str(e)}} return {"jsonrpc": "2.0", "id": mid, "result": res} class Handler(BaseHTTPRequestHandler): server_version = "humanizer/1.0" def log_message(self, *a): pass def send(self, code, obj=None): body = b"" if obj is None else json.dumps(obj, ensure_ascii=False).encode() self.send_response(code) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") self.send_header("Access-Control-Allow-Origin", "*") self.end_headers() self.wfile.write(body) def do_OPTIONS(self): self.send_response(204) self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type") self.end_headers() def do_GET(self): if self.path == "/api/health": self.send(200, {"ok": True, "name": "humanizer", "version": "1.0.0"}) elif self.path == "/api/rules": self.send(200, rules()) else: self.send(404, {"error": "not found"}) def do_POST(self): if self.path not in ("/api/check", "/api/rewrite", "/mcp"): return self.send(404, {"error": "not found"}) try: n = int(self.headers.get("Content-Length", "0")) except ValueError: n = -1 if n < 0 or n > 65536: return self.send(413, {"error": "body too large (max 64 KB)"}) try: data = json.loads(self.rfile.read(n) or b"null") except (ValueError, UnicodeDecodeError): return self.send(400, {"error": "body must be JSON"}) if self.path == "/mcp": r = rpc(data) return self.send(202) if r is None else self.send(200, r) if not isinstance(data, dict): return self.send(400, {"error": 'send {"text": "...", "channel": "email"}'}) try: fn = do_rewrite if self.path == "/api/rewrite" else check self.send(200, fn(data.get("text"), data.get("channel", "generic"))) except ValueError as e: self.send(400, {"error": str(e)}) def stdio(): for line in sys.stdin: line = line.strip() if not line: continue try: r = rpc(json.loads(line)) except ValueError: continue if r is not None: print(json.dumps(r, ensure_ascii=False), flush=True) if __name__ == "__main__": if "--stdio" in sys.argv: stdio() else: port = int(sys.argv[sys.argv.index("--port") + 1]) if "--port" in sys.argv else 8089 print(f"humanizer listening on 127.0.0.1:{port}", file=sys.stderr, flush=True) ThreadingHTTPServer(("127.0.0.1", port), Handler).serve_forever()