#!/usr/bin/env python3 """Expose The Ultimate Critic to AI agents over MCP stdio. The server has no third-party dependencies. Configure TUC_API_KEY in the MCP client's environment for paid evaluation and usage tools. The deterministic preview tool works without a key and does not retain submitted artifacts. """ from __future__ import annotations import json import os import sys import urllib.error import urllib.request from typing import Any from urllib.parse import urlsplit SERVER_NAME = "the-ultimate-critic" SERVER_VERSION = "1.0.0" LATEST_PROTOCOL = "2025-11-25" SUPPORTED_PROTOCOLS = { "2025-11-25", "2025-06-18", "2025-03-26", "2024-11-05", } DEFAULT_API_BASE = "https://ultimate-critic.fly.dev/api" MAX_ARGUMENT_BYTES = 9 * 1024 * 1024 MAX_RESPONSE_BYTES = 8 * 1024 * 1024 class ToolFailure(ValueError): """A recoverable tool-level error that the calling model can act on.""" EVALUATION_REQUEST_SCHEMA: dict[str, Any] = { "type": "object", "additionalProperties": False, "properties": { "artifact_type": { "type": "string", "enum": ["text", "code", "image", "ui", "game_scene", "json", "mixed"], }, "artifact": { "type": "object", "description": "The work to evaluate. Use content for text/JSON and files for evidence or image payloads.", "properties": { "content": {}, "files": {"type": "array", "maxItems": 32, "items": {"type": "object"}}, "metadata": {"type": "object"}, }, "additionalProperties": False, }, "intent": { "type": "object", "properties": { "prompt": {"type": "string"}, "requirements": {"type": "array", "items": {"type": "string"}}, "constraints": {"type": "array", "items": {"type": "string"}}, "success_criteria": {"type": "array", "items": {"type": "string"}}, }, "additionalProperties": False, }, "domain": { "type": "string", "enum": ["general", "software", "game", "unity", "unreal", "design", "legal", "medical", "education", "architecture"], "default": "general", }, "evaluation_profile": { "type": "string", "description": "Built-in profile or the name of a customer custom profile.", "default": "default", }, "rubric": {"type": "array", "items": {}}, "reference_materials": {"type": "array", "items": {}}, "output_format": {"type": "string", "enum": ["json", "report", "both"], "default": "json"}, "evidence_policy": {"type": "string", "enum": ["standard", "strict_release"], "default": "standard"}, }, "required": ["artifact_type", "artifact"], } TOOLS: list[dict[str, Any]] = [ { "name": "tuc.evaluate", "title": "Evaluate work with TUC", "description": ( "Evaluate an AI-produced artifact against its task, requirements, constraints, and evidence. " "This uses 1, 5, or 10 review credits based on package size and media. Route fail findings to repair, " "collect evidence for insufficient_evidence, and release only on pass under your own policy." ), "inputSchema": { "type": "object", "additionalProperties": False, "properties": {"request": EVALUATION_REQUEST_SCHEMA}, "required": ["request"], }, "annotations": { "readOnlyHint": False, "destructiveHint": False, "idempotentHint": False, "openWorldHint": True, }, }, { "name": "tuc.preview", "title": "Preview TUC evaluation", "description": ( "Run the narrow no-key deterministic preview on one non-sensitive text, code, or JSON artifact. " "The preview is not retained, makes no external model call, and does not consume paid entitlement." ), "inputSchema": { "type": "object", "additionalProperties": False, "properties": { "artifact_type": {"type": "string", "enum": ["text", "code", "json"]}, "artifact": {"type": "string", "minLength": 1, "maxLength": 10000}, "task": {"type": "string", "minLength": 10, "maxLength": 2000}, "requirements": { "type": "array", "maxItems": 12, "items": {"type": "string", "minLength": 1, "maxLength": 500}, "default": [], }, }, "required": ["artifact_type", "artifact", "task"], }, "annotations": { "readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": True, }, }, { "name": "tuc.quote", "title": "Quote TUC review credits", "description": "Return the standard, extended, or full package tier and credit charge without running an evaluation.", "inputSchema": { "type": "object", "additionalProperties": False, "properties": {"request": EVALUATION_REQUEST_SCHEMA}, "required": ["request"], }, "annotations": { "readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": True, }, }, { "name": "tuc.usage", "title": "Get TUC usage", "description": "Return the configured tenant's review-credit usage and entitlement limits without submitting work.", "inputSchema": {"type": "object", "additionalProperties": False}, "annotations": { "readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": True, }, }, ] def _api_base() -> str: value = os.getenv("TUC_API_URL", DEFAULT_API_BASE).strip().rstrip("/") parsed = urlsplit(value) local_http = parsed.scheme == "http" and parsed.hostname in {"127.0.0.1", "localhost", "::1"} if (parsed.scheme != "https" and not local_http) or not parsed.netloc: raise ToolFailure("TUC_API_URL must be an absolute HTTPS URL; HTTP is allowed only for localhost") if parsed.username or parsed.password or parsed.query or parsed.fragment: raise ToolFailure("TUC_API_URL must not contain credentials, a query, or a fragment") return value def _timeout() -> float: raw = os.getenv("TUC_TIMEOUT_SECONDS", "60").strip() try: value = float(raw) except ValueError as exc: raise ToolFailure("TUC_TIMEOUT_SECONDS must be a number") from exc if not 1 <= value <= 120: raise ToolFailure("TUC_TIMEOUT_SECONDS must be between 1 and 120") return value def _encoded_size(value: Any) -> int: try: return len(json.dumps(value, separators=(",", ":"), ensure_ascii=False).encode("utf-8")) except (TypeError, ValueError) as exc: raise ToolFailure("tool arguments must be valid JSON values") from exc def _read_response(response: Any) -> dict[str, Any]: content_length = response.headers.get("Content-Length") if content_length: try: declared_length = int(content_length) except ValueError: declared_length = 0 if declared_length > MAX_RESPONSE_BYTES: raise ToolFailure("TUC response exceeded the MCP bridge limit") raw = response.read(MAX_RESPONSE_BYTES + 1) if len(raw) > MAX_RESPONSE_BYTES: raise ToolFailure("TUC response exceeded the MCP bridge limit") try: value = json.loads(raw) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise ToolFailure("TUC returned an invalid JSON response") from exc if not isinstance(value, dict): raise ToolFailure("TUC returned a non-object response") return value def _http_json(method: str, path: str, *, payload: dict[str, Any] | None = None, api_key: str = "") -> dict[str, Any]: body = None if payload is None else json.dumps(payload, separators=(",", ":"), ensure_ascii=False).encode("utf-8") headers = { "Accept": "application/json", "User-Agent": f"TUC-MCP/{SERVER_VERSION}", } if body is not None: headers["Content-Type"] = "application/json; charset=utf-8" if api_key: headers["X-API-Key"] = api_key request = urllib.request.Request(f"{_api_base()}{path}", data=body, headers=headers, method=method) try: with urllib.request.urlopen(request, timeout=_timeout()) as response: return _read_response(response) except urllib.error.HTTPError as exc: raw = exc.read(4096) detail = "" try: error_value = json.loads(raw) if isinstance(error_value, dict) and isinstance(error_value.get("detail"), str): detail = error_value["detail"] except (UnicodeDecodeError, json.JSONDecodeError): pass suffix = f": {detail[:1000]}" if detail else "" raise ToolFailure(f"TUC request failed with HTTP {exc.code}{suffix}") from None except urllib.error.URLError as exc: raise ToolFailure(f"TUC request could not be completed: {str(exc.reason)[:1000]}") from None except TimeoutError: raise ToolFailure("TUC request timed out") from None def _tenant_key() -> str: key = os.getenv("TUC_API_KEY", "").strip() if not key: raise ToolFailure("TUC_API_KEY is required for this tool; add it to the MCP server environment") return key def _tool_result(value: dict[str, Any]) -> dict[str, Any]: return { "content": [{"type": "text", "text": json.dumps(value, indent=2, sort_keys=True, ensure_ascii=False)}], "structuredContent": value, "isError": False, } def _tool_error(message: str) -> dict[str, Any]: return {"content": [{"type": "text", "text": message}], "isError": True} def _validate_preview(arguments: dict[str, Any]) -> dict[str, Any]: allowed = {"artifact_type", "artifact", "task", "requirements"} if set(arguments) - allowed: raise ToolFailure("tuc.preview received unsupported arguments") artifact_type = arguments.get("artifact_type") artifact = arguments.get("artifact") task = arguments.get("task") requirements = arguments.get("requirements", []) if artifact_type not in {"text", "code", "json"}: raise ToolFailure("artifact_type must be text, code, or json") if not isinstance(artifact, str) or not 1 <= len(artifact) <= 10_000: raise ToolFailure("artifact must contain 1 to 10,000 characters") if not isinstance(task, str) or not 10 <= len(task.strip()) <= 2_000: raise ToolFailure("task must contain 10 to 2,000 characters") if not isinstance(requirements, list) or len(requirements) > 12: raise ToolFailure("requirements must contain at most 12 items") if any(not isinstance(item, str) or not item.strip() or len(item) > 500 for item in requirements): raise ToolFailure("each requirement must contain 1 to 500 characters") if artifact_type == "json": try: parsed = json.loads(artifact) except json.JSONDecodeError as exc: raise ToolFailure("artifact must be valid JSON when artifact_type is json") from exc if not isinstance(parsed, (dict, list)): raise ToolFailure("JSON preview artifacts must contain an object or array") return { "artifact_type": artifact_type, "artifact": artifact, "task": task.strip(), "requirements": [item.strip() for item in requirements], } def _call_tool(name: str, arguments: dict[str, Any]) -> dict[str, Any]: try: if _encoded_size(arguments) > MAX_ARGUMENT_BYTES: raise ToolFailure("tool arguments exceed the 9 MiB MCP bridge limit") if name == "tuc.preview": payload = _validate_preview(arguments) value = _http_json("POST", "/v1/public/preview", payload=payload) if value.get("mode") != "deterministic_no_retention" or value.get("retained") is not False: raise ToolFailure("TUC preview did not return the no-retention contract") return _tool_result(value) if name == "tuc.evaluate": if set(arguments) != {"request"} or not isinstance(arguments.get("request"), dict): raise ToolFailure("tuc.evaluate requires exactly one request object") request = arguments["request"] if not isinstance(request.get("artifact_type"), str) or not isinstance(request.get("artifact"), dict): raise ToolFailure("request requires artifact_type and artifact") return _tool_result(_http_json("POST", "/v1/evaluate", payload=request, api_key=_tenant_key())) if name == "tuc.quote": if set(arguments) != {"request"} or not isinstance(arguments.get("request"), dict): raise ToolFailure("tuc.quote requires exactly one request object") request = arguments["request"] if not isinstance(request.get("artifact_type"), str) or not isinstance(request.get("artifact"), dict): raise ToolFailure("request requires artifact_type and artifact") return _tool_result(_http_json("POST", "/v1/evaluate/credits", payload=request, api_key=_tenant_key())) if name == "tuc.usage": if arguments: raise ToolFailure("tuc.usage does not accept arguments") return _tool_result(_http_json("GET", "/v1/usage", api_key=_tenant_key())) raise ToolFailure(f"Unknown tool: {name}") except ToolFailure as exc: return _tool_error(str(exc)) def _success(request_id: Any, result: dict[str, 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(message: Any) -> dict[str, Any] | None: if not isinstance(message, dict) or message.get("jsonrpc") != "2.0": return _error(message.get("id") if isinstance(message, dict) else None, -32600, "Invalid Request") request_id = message.get("id") method = message.get("method") params = message.get("params", {}) if not isinstance(method, str): return _error(request_id, -32600, "Invalid Request") if request_id is None: return None if method == "initialize": if not isinstance(params, dict): return _error(request_id, -32602, "Invalid initialize parameters") requested = params.get("protocolVersion") protocol = requested if requested in SUPPORTED_PROTOCOLS else LATEST_PROTOCOL return _success(request_id, { "protocolVersion": protocol, "capabilities": {"tools": {"listChanged": False}}, "serverInfo": {"name": SERVER_NAME, "version": SERVER_VERSION}, "instructions": ( "Use tuc.preview only for sanitized non-sensitive trials. Use tuc.quote before tuc.evaluate to see the 1, 5, or 10-credit charge; " "treat pass, fail, and insufficient_evidence as distinct branches." ), }) if method == "ping": return _success(request_id, {}) if method == "tools/list": if not isinstance(params, dict): return _error(request_id, -32602, "Invalid tools/list parameters") return _success(request_id, {"tools": TOOLS}) if method == "tools/call": if not isinstance(params, dict) or not isinstance(params.get("name"), str): return _error(request_id, -32602, "Invalid tools/call parameters") arguments = params.get("arguments", {}) if not isinstance(arguments, dict): return _error(request_id, -32602, "Tool arguments must be an object") return _success(request_id, _call_tool(params["name"], arguments)) return _error(request_id, -32601, "Method not found") def main() -> int: # MCP stdio is always UTF-8, independent of the host OS locale/code page. if hasattr(sys.stdin, "reconfigure"): sys.stdin.reconfigure(encoding="utf-8", errors="strict") if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(encoding="utf-8", errors="strict") for line in sys.stdin: if not line.strip(): continue try: # A conforming MCP client does not add a BOM, but Windows shell # pipelines sometimes prefix the first UTF-8 line with one. message = json.loads(line.lstrip("\ufeff")) except json.JSONDecodeError: response = _error(None, -32700, "Parse error") else: response = handle_message(message) if response is not None: sys.stdout.write(json.dumps(response, separators=(",", ":"), ensure_ascii=False) + "\n") sys.stdout.flush() return 0 if __name__ == "__main__": raise SystemExit(main())