#!/usr/bin/env python3 """Minimal stdio MCP server for Moltgate paid task intake. This server intentionally uses only the Python standard library so it can be used from Claude Code, Codex, Cursor Agent, or any MCP-capable runner without installing extra packages. """ from __future__ import annotations import json import os import sys import urllib.error import urllib.parse import urllib.request from typing import Any SERVER_NAME = "moltgate" SERVER_VERSION = "0.1.0" DEFAULT_BASE_URL = "https://moltgate.com" TOOLS = [ { "name": "moltgate_list_paid_tasks", "description": "List paid Moltgate tasks from the recipient inbox.", "inputSchema": { "type": "object", "properties": { "status": { "type": "string", "description": "Inbox status filter.", "enum": ["NEW", "IN_PROGRESS", "NEEDS_REVIEW", "DELIVERED", "ARCHIVED"], "default": "NEW", }, "offer_id": { "type": "string", "description": "Optional offer UUID filter.", }, "is_read": { "type": "boolean", "description": "Optional read filter. false maps to NEW.", }, }, }, }, { "name": "moltgate_get_paid_task", "description": "Get full detail for one paid Moltgate task, including offer contract fields.", "inputSchema": { "type": "object", "properties": { "id": {"type": "string", "description": "Message/task id."}, }, "required": ["id"], }, }, { "name": "moltgate_mark_task_status", "description": "Mark a paid Moltgate task NEW, IN_PROGRESS, NEEDS_REVIEW, DELIVERED, or ARCHIVED.", "inputSchema": { "type": "object", "properties": { "id": {"type": "string", "description": "Message/task id."}, "inbox_status": { "type": "string", "enum": ["NEW", "IN_PROGRESS", "NEEDS_REVIEW", "DELIVERED", "ARCHIVED"], }, "internal_notes": { "type": "string", "description": "Optional internal note to save with the task.", }, }, "required": ["id", "inbox_status"], }, }, { "name": "moltgate_list_offers", "description": "List the authenticated recipient's Moltgate offers, including deliverables, input requirements, output format, exclusions, and timing.", "inputSchema": {"type": "object", "properties": {}}, }, ] def _base_url() -> str: return os.environ.get("MOLTGATE_BASE_URL", DEFAULT_BASE_URL).rstrip("/") def _api_key() -> str: api_key = os.environ.get("MOLTGATE_API_KEY", "").strip() if not api_key: raise RuntimeError("MOLTGATE_API_KEY is not set") return api_key def _http_request(method: str, path: str, body: dict[str, Any] | None = None) -> Any: url = f"{_base_url()}{path}" data = None headers = { "Authorization": f"Bearer {_api_key()}", "Accept": "application/json", } if body is not None: data = json.dumps(body).encode("utf-8") headers["Content-Type"] = "application/json" request = urllib.request.Request(url, data=data, headers=headers, method=method) try: with urllib.request.urlopen(request, timeout=30) as response: raw = response.read().decode("utf-8") return json.loads(raw) if raw else {} except urllib.error.HTTPError as exc: raw = exc.read().decode("utf-8", errors="replace") raise RuntimeError(f"Moltgate API returned HTTP {exc.code}: {raw}") from exc except urllib.error.URLError as exc: raise RuntimeError(f"Could not reach Moltgate API: {exc.reason}") from exc def _tool_result(payload: Any) -> dict[str, Any]: return { "content": [ { "type": "text", "text": json.dumps(payload, indent=2, sort_keys=True), } ] } def _call_tool(name: str, args: dict[str, Any]) -> dict[str, Any]: if name == "moltgate_list_paid_tasks": params: dict[str, str] = {} if args.get("status"): params["status"] = str(args["status"]).upper() if args.get("offer_id"): params["offer_id"] = str(args["offer_id"]) if "is_read" in args: params["is_read"] = "true" if args["is_read"] else "false" query = f"?{urllib.parse.urlencode(params)}" if params else "" return _tool_result(_http_request("GET", f"/api/inbox/messages/{query}")) if name == "moltgate_get_paid_task": task_id = str(args["id"]) return _tool_result(_http_request("GET", f"/api/inbox/messages/{task_id}/")) if name == "moltgate_mark_task_status": task_id = str(args["id"]) body = {"inbox_status": str(args["inbox_status"]).upper()} if "internal_notes" in args: body["internal_notes"] = str(args["internal_notes"]) return _tool_result( _http_request("PATCH", f"/api/inbox/messages/{task_id}/update_status/", body) ) if name == "moltgate_list_offers": return _tool_result(_http_request("GET", "/api/offers/")) raise RuntimeError(f"Unknown tool: {name}") def _read_message() -> dict[str, Any] | None: headers: dict[str, str] = {} while True: line = sys.stdin.buffer.readline() if not line: return None if line in (b"\r\n", b"\n"): break key, _, value = line.decode("ascii").partition(":") headers[key.strip().lower()] = value.strip() length = int(headers.get("content-length", "0")) if length <= 0: return None raw = sys.stdin.buffer.read(length).decode("utf-8") return json.loads(raw) def _write_message(message: dict[str, Any]) -> None: raw = json.dumps(message, separators=(",", ":")).encode("utf-8") sys.stdout.buffer.write(f"Content-Length: {len(raw)}\r\n\r\n".encode("ascii")) sys.stdout.buffer.write(raw) sys.stdout.buffer.flush() def _response(request_id: Any, result: Any) -> dict[str, Any]: return {"jsonrpc": "2.0", "id": request_id, "result": result} def _error(request_id: Any, code: int, message: str) -> dict[str, Any]: return {"jsonrpc": "2.0", "id": request_id, "error": {"code": code, "message": message}} def _handle(message: dict[str, Any]) -> dict[str, Any] | None: method = message.get("method") request_id = message.get("id") if method == "initialize": return _response( request_id, { "protocolVersion": "2024-11-05", "capabilities": {"tools": {}}, "serverInfo": {"name": SERVER_NAME, "version": SERVER_VERSION}, }, ) if method == "notifications/initialized": return None if method == "tools/list": return _response(request_id, {"tools": TOOLS}) if method == "tools/call": params = message.get("params") or {} try: return _response( request_id, _call_tool(str(params.get("name")), params.get("arguments") or {}), ) except Exception as exc: # noqa: BLE001 - return tool errors to the client. return _response( request_id, { "isError": True, "content": [{"type": "text", "text": str(exc)}], }, ) if method and method.startswith("notifications/"): return None return _error(request_id, -32601, f"Method not found: {method}") def main() -> None: while True: message = _read_message() if message is None: break response = _handle(message) if response is not None: _write_message(response) if __name__ == "__main__": main()