#!/usr/bin/env python3 import sys import json import os from lib.security import get_safe_path, PROJECTS_DIR, STATE_DIR from lib.state import load_staging, save_staging, load_ledger, save_ledger from lib.compiler import compile_page from lib.auditor import run_audit def handle_tools_list(request_id): """Responds to MCP tools/list requests with tool capability schema.""" response = { "jsonrpc": "2.0", "result": { "tools": [ { "name": "geoscaper", "description": "Unified design-and-build automation server for managing staging tokens, compilation, and page auditing.", "inputSchema": { "type": "object", "properties": { "action": {"type": "string", "description": "The command action: init, update_staging, queue_task, compile, audit, write_file, read_file, read_template_manifest"}, "project_name": {"type": "string", "description": "Name of the targeted website project"}, "page_id": {"type": "string", "description": "The specific page being compiled or checked"}, "data": {"type": "object", "description": "Staging configuration payload"}, "task": {"type": "object", "description": "Task registration metadata"}, "file_path": {"type": "string", "description": "Relative path to file (e.g. index.html, styles.css)"}, "content": {"type": "string", "description": "Content to write to file"} }, "required": ["action", "project_name"] } } ] }, "id": request_id } sys.stdout.write(json.dumps(response) + "\n") sys.stdout.flush() def handle_tools_call(request_id, params): """Executes target operations and returns outputs wrapped in MCP results.""" name = params.get("name") arguments = params.get("arguments", {}) if name != "geoscaper": response = { "jsonrpc": "2.0", "error": {"code": -32601, "message": f"Method not found: {name}"}, "id": request_id } sys.stdout.write(json.dumps(response) + "\n") sys.stdout.flush() return action = arguments.get("action") project_name = arguments.get("project_name") if not action or not project_name: result_text = json.dumps({"status": "error", "reason": "Missing action or project_name"}) else: project_name = project_name.lower().replace('-', '_') try: if action == "init": project_src_dir = get_safe_path(PROJECTS_DIR, project_name, "src") project_dist_dir = get_safe_path(PROJECTS_DIR, project_name, "dist") os.makedirs(project_src_dir, exist_ok=True) os.makedirs(project_dist_dir, exist_ok=True) if not os.path.exists(get_safe_path(STATE_DIR, f"{project_name}_staging.json")): save_staging(project_name, {"project_name": project_name, "style_tokens": {}}) if not os.path.exists(get_safe_path(STATE_DIR, f"{project_name}_ledger.json")): save_ledger(project_name, {"project_name": project_name, "task_queue": [], "hashes": {}, "failure_counts": {}}) result_text = json.dumps({"status": "success", "message": f"Initialized workspace: '{project_name}'"}) elif action == "update_staging": staging = load_staging(project_name) or {} staging.update(arguments.get("data", {})) save_staging(project_name, staging) result_text = json.dumps({"status": "success", "message": "Staging cache updated."}) elif action == "queue_task": ledger = load_ledger(project_name) task = arguments.get("task", {}) if "page_id" in task and "filename" in task: ledger.setdefault("task_queue", []).append(task) save_ledger(project_name, ledger) result_text = json.dumps({"status": "success", "message": f"Task '{task['page_id']}' queued."}) else: ledger.setdefault("dead_letter_queue", []).append(task) save_ledger(project_name, ledger) result_text = json.dumps({"status": "error", "reason": "Task must include 'page_id' and 'filename'. Malformed task appended to dead_letter_queue."}) elif action == "compile": result = compile_page(project_name, arguments.get("page_id")) result_text = json.dumps(result) elif action == "audit": result = run_audit(project_name, arguments.get("page_id")) result_text = json.dumps(result) elif action == "write_file": file_path = arguments.get("file_path") content = arguments.get("content", "") if not file_path: result_text = json.dumps({"status": "error", "reason": "Missing file_path"}) elif not file_path.lower().endswith(('.html', '.css', '.js')): result_text = json.dumps({"status": "error", "reason": "File extension not permitted. Allowed: .html, .css, .js"}) else: full_path = get_safe_path(PROJECTS_DIR, project_name, "src", file_path) os.makedirs(os.path.dirname(full_path), exist_ok=True) with open(full_path, "w", encoding="utf-8") as f: f.write(content) result_text = json.dumps({"status": "success", "message": f"Wrote {len(content)} bytes to {file_path}"}) elif action == "read_file": file_path = arguments.get("file_path") if not file_path: result_text = json.dumps({"status": "error", "reason": "Missing file_path"}) elif not file_path.lower().endswith(('.html', '.css', '.js')): result_text = json.dumps({"status": "error", "reason": "File extension not permitted. Allowed: .html, .css, .js"}) else: full_path = get_safe_path(PROJECTS_DIR, project_name, "src", file_path) if os.path.exists(full_path): with open(full_path, "r", encoding="utf-8") as f: content = f.read() result_text = json.dumps({"status": "success", "content": content}) else: result_text = json.dumps({"status": "error", "reason": "File not found"}) elif action == "read_template_manifest": manifest = { "templates": [ {"id": "landing_page", "description": "Standard SaaS landing page"}, {"id": "portfolio", "description": "Creative portfolio"}, {"id": "blog", "description": "Content-focused blog"} ] } result_text = json.dumps({"status": "success", "manifest": manifest}) else: result_text = json.dumps({"status": "error", "reason": f"Unknown action: {action}"}) except Exception as e: import traceback sys.stderr.write(f"[GEOSCAPER TOOL ERROR] {action} failed: {str(e)}\n{traceback.format_exc()}\n") sys.stderr.flush() result_text = json.dumps({"status": "error", "reason": f"Execution error: {str(e)}"}) # Formats result to match standard MCP tools/call return envelope response = { "jsonrpc": "2.0", "result": { "content": [ { "type": "text", "text": result_text } ] }, "id": request_id } sys.stdout.write(json.dumps(response) + "\n") sys.stdout.flush() def main(): """Persistent stdio event loop processing JSON-RPC packets.""" for line in sys.stdin: line = line.strip() if not line: continue try: payload = json.loads(line) request_id = payload.get("id") method = payload.get("method") params = payload.get("params", {}) if method == "tools/list": handle_tools_list(request_id) elif method == "tools/call": handle_tools_call(request_id, params) else: response = { "jsonrpc": "2.0", "error": {"code": -32601, "message": f"Method not found: {method}"}, "id": request_id } sys.stdout.write(json.dumps(response) + "\n") sys.stdout.flush() except Exception as e: import traceback sys.stderr.write(f"[GEOSCAPER PARSE ERROR] {str(e)}\n{traceback.format_exc()}\n") sys.stderr.flush() response = { "jsonrpc": "2.0", "error": {"code": -32700, "message": f"Parse error: {str(e)}"}, "id": None } sys.stdout.write(json.dumps(response) + "\n") sys.stdout.flush() if __name__ == "__main__": main()