#!/usr/bin/env python3 """An MCP server over the nomore404 API, for an assistant to read. #387 Model Context Protocol on stdin and stdout: JSON-RPC, one message per line. No dependencies, like the CLI beside it, and it borrows that file for the parts that are the same, so there is one place that knows where a token comes from and one that knows how to follow a cursor. N404_TOKEN=n404_... ./nomore404_mcp.py Configure it wherever your assistant keeps its servers, as a command with that environment. A `nomore404.env` beside these two files works as well. **It reads and does not write.** An assistant is a caller that can be talked into things, by a web page it summarised or by a mistake in its own reasoning, and the useful questions here are read-mostly anyway: what is down, why did this open, how has it been this month. A tool that deletes a monitor and its history is one prompt away from being used, and the API is still there with a token that may write for somebody who wants exactly that. So the tools are the API's GET operations, and a test holds them to that list rather than to a list somebody maintains. """ from __future__ import annotations import json import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) import nomore404 as api # noqa: E402 the file beside this one # What we answer `initialize` with when the client asks for something we do # not know. Echoing back an unknown version is how a client ends up assuming # a capability that is not there. PROTOCOL = "2025-06-18" KNOWN_PROTOCOLS = (PROTOCOL, "2025-03-26", "2024-11-05") NAME = "nomore404" # JSON-RPC's own codes, for the errors that are about the protocol rather than # about monitoring. A tool that fails answers a result with isError instead, # which is what lets an assistant read the reason and try something else. PARSE_ERROR = -32700 INVALID_REQUEST = -32600 METHOD_NOT_FOUND = -32601 INTERNAL_ERROR = -32603 UUID = {"type": "string", "description": "The id from a listing, a UUID."} DAYS = {"type": "integer", "minimum": 1, "maximum": 365, "description": "How many days back to look. 30 by default."} def _monitors(_: dict) -> object: return list(api.every("/monitors")) def _monitor(args: dict) -> object: return api.call("GET", f"/monitors/{args['id']}") def _status(args: dict) -> object: return api.call("GET", f"/monitors/{args['id']}/status", {"days": args.get("days")}) def _incidents(args: dict) -> object: return list(api.every("/incidents", { "state": args.get("state", "all"), "days": args.get("days"), "monitor": args.get("monitor")})) def _incident(args: dict) -> object: return api.call("GET", f"/incidents/{args['id']}") def _maintenance(_: dict) -> object: return list(api.every("/maintenance")) def _window(args: dict) -> object: return api.call("GET", f"/maintenance/{args['id']}") def _groups(_: dict) -> object: return list(api.every("/groups")) def _group(args: dict) -> object: return api.call("GET", f"/groups/{args['id']}") def _whoami(_: dict) -> object: return api.call("GET", "/me") # name, what it is for, its arguments, and the function that answers it. The # descriptions are written for something that has to choose between them # without trying one first, which is why each says when it is the right tool # rather than only what it returns. TOOLS = [ { "name": "whoami", "description": "Which nomore404 organisation this token acts for, " "and what it may do. Run this first when anything is " "unexpected: the answer names the account, so it tells " "you whether you are looking at the right one.", "inputSchema": {"type": "object", "properties": {}}, "run": _whoami, }, { "name": "list_monitors", "description": "Every monitor this organisation watches, with its " "type, address, interval and whether it is paused. The " "id of each is what the other tools take.", "inputSchema": {"type": "object", "properties": {}}, "run": _monitors, }, { "name": "get_monitor", "description": "One monitor's settings. For what it is doing right " "now, use monitor_status instead.", "inputSchema": {"type": "object", "properties": {"id": UUID}, "required": ["id"]}, "run": _monitor, }, { "name": "monitor_status", "description": "Whether one monitor is up right now, and its uptime " "over a window: the percentage, and how many seconds " "it was down. This is the tool for 'is the site up' " "and for 'how has it been this month'.", "inputSchema": {"type": "object", "properties": {"id": UUID, "days": DAYS}, "required": ["id"]}, "run": _status, }, { "name": "list_incidents", "description": "Outages, newest first. Filter by state (all, open, " "resolved), by monitor, or by how far back to look. " "An outage that is still happening is returned " "whatever window was asked for.", "inputSchema": {"type": "object", "properties": { "state": {"type": "string", "enum": ["all", "open", "resolved"]}, "monitor": {"type": "string", "description": "Only this monitor's, by its id."}, "days": DAYS}}, "run": _incidents, }, { "name": "get_incident", "description": "One outage in full: what failed, when, whether it is " "over, which probe saw it and which one confirmed it " "from another continent, and any note written on it.", "inputSchema": {"type": "object", "properties": {"id": UUID}, "required": ["id"]}, "run": _incident, }, { "name": "list_maintenance", "description": "Scheduled maintenance windows and what they cover. " "A monitor inside one is still checked and does not " "alert, so this explains a quiet period that is not an " "outage.", "inputSchema": {"type": "object", "properties": {}}, "run": _maintenance, }, { "name": "get_maintenance_window", "description": "One maintenance window: when it starts and ends, " "whether it repeats, and how many monitors and groups " "it covers.", "inputSchema": {"type": "object", "properties": {"id": UUID}, "required": ["id"]}, "run": _window, }, { "name": "list_groups", "description": "The groups this organisation sorts its monitors " "into, with how many are in each. A group gives its " "monitors one status between them, so this is the " "shape of the estate rather than a list of checks.", "inputSchema": {"type": "object", "properties": {}}, "run": _groups, }, { "name": "get_group", "description": "One group by its id: its name, description and how " "many monitors it holds. A monitor's own group is on " "the monitor, so use list_monitors to see which are " "in it.", "inputSchema": {"type": "object", "properties": {"id": UUID}, "required": ["id"]}, "run": _group, }, ] BY_NAME = {tool["name"]: tool for tool in TOOLS} def described() -> list[dict]: """The tools as the protocol wants them, without our own `run`.""" return [{k: v for k, v in tool.items() if k != "run"} for tool in TOOLS] def call_tool(name: str, arguments: dict) -> dict: """One tool, and its answer as content. A refusal comes back as a result with isError rather than as a JSON-RPC error, which is the difference between "this request was malformed" and "your token may not do that": the second is something an assistant can read, explain and work around, and the API already writes that sentence. """ tool = BY_NAME.get(name) if tool is None: return {"content": [{"type": "text", "text": f"No tool called {name!r}."}], "isError": True} try: answer = tool["run"](arguments or {}) except api.Problem as e: return {"content": [{"type": "text", "text": str(e)}], "isError": True} return {"content": [{"type": "text", "text": json.dumps(answer, indent=2)}], "isError": False} def handle(message: dict) -> dict | None: """One JSON-RPC message in, one answer out, or None for a notification. A notification has no id and takes no reply, which is not a detail to skip: answering `notifications/initialized` with a result is a protocol error that some clients report as the server being broken. """ method = message.get("method") request_id = message.get("id") params = message.get("params") or {} if request_id is None: return None if method == "initialize": asked = (params.get("protocolVersion") or "").strip() return _result(request_id, { "protocolVersion": asked if asked in KNOWN_PROTOCOLS else PROTOCOL, "capabilities": {"tools": {"listChanged": False}}, "serverInfo": {"name": NAME, "version": "1"}, }) if method == "tools/list": return _result(request_id, {"tools": described()}) if method == "tools/call": return _result(request_id, call_tool(params.get("name", ""), params.get("arguments") or {})) if method == "ping": return _result(request_id, {}) return _error(request_id, METHOD_NOT_FOUND, f"No method {method!r}.") def _result(request_id, result: dict) -> dict: return {"jsonrpc": "2.0", "id": request_id, "result": result} def _error(request_id, code: int, message: str) -> dict: return {"jsonrpc": "2.0", "id": request_id, "error": {"code": code, "message": message}} def serve(stdin=None, stdout=None) -> int: """Read a message per line, answer on the next. Nothing is printed to stdout that is not a message: the transport is the stream, so a stray print is a parse error at the other end. Anything worth saying goes to stderr. """ stdin = stdin or sys.stdin stdout = stdout or sys.stdout for line in stdin: line = line.strip() if not line: continue try: message = json.loads(line) except json.JSONDecodeError: answer = _error(None, PARSE_ERROR, "That was not JSON.") else: if not isinstance(message, dict): answer = _error(None, INVALID_REQUEST, "Expected an object.") else: try: answer = handle(message) except Exception as e: # noqa: BLE001 the loop must not die print(f"nomore404-mcp: {e}", file=sys.stderr) answer = _error(message.get("id"), INTERNAL_ERROR, "Something went wrong in the server.") if answer is not None: stdout.write(json.dumps(answer) + "\n") stdout.flush() return 0 if __name__ == "__main__": sys.exit(serve())