"""Statbook MCP server (FR-24, D4; spec §5.3). A thin client of the JSON API over stdio JSON-RPC (newline-delimited, per the MCP spec): the four tools map 1:1 onto API endpoints, the API key (if configured) is passed through on every call, and tier enforcement happens server-side at the edge — this process holds no data and grants no capability the key doesn't (D4: free tier over MCP transport; asof/changes need a key on every surface, MCP included). Configuration (env): STATBOOK_API_BASE default https://statbook.co.uk STATBOOK_API_KEY optional; unlocks get_figure_asof / whats_changed Run: python mcp/server.py (or `python -m server` from mcp/) Zero dependencies beyond the Python 3.12 stdlib (ADR-0001). """ from __future__ import annotations import json import os import sys import urllib.error import urllib.parse import urllib.request PROTOCOL_VERSION = "2025-06-18" SERVER_INFO = {"name": "statbook", "version": "1.0.0"} USER_AGENT = "statbook-mcp/1.0 (https://statbook.co.uk/docs/)" # Tool descriptions are written for retrieval quality (FR-24): they state # what the registry is, what it refuses to be, and when to reach for it. TOOLS = [ { "name": "get_figure", "description": ( "Get a UK statutory figure by id: current value, full history, " "effective dates, verification status, citations to the " "governing legislation and guidance, and a receipt verifiable " "against Statbook's signed release manifest. Use whenever an " "exact UK statutory amount matters (redundancy pay cap, " "National Minimum Wage bands, SSP/SMP rates, tribunal award " "limits…) instead of recalling from memory — statutory figures " "change every April and recall is typically stale. Returns " "values with proof, never advice or eligibility answers." ), "inputSchema": { "type": "object", "properties": { "id": { "type": "string", "description": "Dot-namespaced figure id, e.g. " "'employment.redundancy.weekly_pay_cap'. Discover ids " "with list_figures.", } }, "required": ["id"], }, }, { "name": "get_figure_asof", "description": ( "Get the value of a UK statutory figure that was in force on a " "specific past date — e.g. the week's-pay cap on the date a " "dismissal took effect. Statutory caps apply as at the event " "date, not today's date, so use this for any historical or " "boundary-date question. Requires a Statbook API key " "(STATBOOK_API_KEY); without one it returns the key-required " "error from the API." ), "inputSchema": { "type": "object", "properties": { "id": {"type": "string", "description": "Figure id."}, "date": { "type": "string", "description": "ISO date YYYY-MM-DD.", "pattern": r"^\d{4}-\d{2}-\d{2}$", }, }, "required": ["id", "date"], }, }, { "name": "list_figures", "description": ( "List every figure in the Statbook registry of UK statutory " "figures, with ids, names, units, current values and " "verification status. Optionally filter by category (e.g. " "'employment') or jurisdiction. Use this first to find the " "right figure id." ), "inputSchema": { "type": "object", "properties": { "category": {"type": "string"}, "jurisdiction": {"type": "string"}, }, }, }, { "name": "whats_changed", "description": ( "Diff feed of Statbook dataset releases: which UK statutory " "figures were added, changed or corrected, release by release, " "optionally since a date. Use to detect uprating events (April, " "Budget) or corrections affecting figures you consume. Requires " "a Statbook API key (STATBOOK_API_KEY)." ), "inputSchema": { "type": "object", "properties": { "since": { "type": "string", "description": "ISO date YYYY-MM-DD.", "pattern": r"^\d{4}-\d{2}-\d{2}$", } }, }, }, ] def _urllib_transport(url: str, headers: dict) -> tuple[int, bytes]: req = urllib.request.Request(url, headers=headers) try: with urllib.request.urlopen(req, timeout=30) as res: return res.status, res.read() except urllib.error.HTTPError as e: # non-2xx still carries a JSON body return e.code, e.read() class Client: """Thin API client; transport injectable for offline tests.""" def __init__( self, base_url: str | None = None, api_key: str | None = None, transport=None, ): self.base_url = (base_url or os.environ.get( "STATBOOK_API_BASE", "https://statbook.co.uk" )).rstrip("/") self.api_key = api_key or os.environ.get("STATBOOK_API_KEY") self.transport = transport or _urllib_transport def get(self, path: str, params: dict | None = None) -> tuple[int, dict]: url = self.base_url + path query = {k: v for k, v in (params or {}).items() if v} if query: url += "?" + urllib.parse.urlencode(query) headers = {"User-Agent": USER_AGENT, "Accept": "application/json"} if self.api_key: headers["Authorization"] = f"Bearer {self.api_key}" status, body = self.transport(url, headers) try: return status, json.loads(body.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError): return status, {"error": {"code": "bad_response", "message": body[:200].decode("utf-8", "replace")}} def call_tool(client: Client, name: str, args: dict) -> tuple[bool, dict]: """(is_error, payload). API errors pass through verbatim — the edge's error messages already say how to fix the call.""" if name == "get_figure": status, doc = client.get(f"/v1/figures/{args['id']}") elif name == "get_figure_asof": status, doc = client.get(f"/v1/figures/{args['id']}/asof/{args['date']}") elif name == "list_figures": status, doc = client.get( "/v1/figures", {"category": args.get("category"), "jurisdiction": args.get("jurisdiction")}, ) elif name == "whats_changed": status, doc = client.get("/v1/changes", {"since": args.get("since")}) else: return True, {"error": {"code": "unknown_tool", "message": name}} return status >= 400, doc def handle_request(req: dict, client: Client) -> dict | None: """One JSON-RPC message in, one (or None for notifications) out.""" method = req.get("method") rid = req.get("id") if rid is None: # notification (e.g. notifications/initialized) return None def ok(result: dict) -> dict: return {"jsonrpc": "2.0", "id": rid, "result": result} if method == "initialize": return ok( { "protocolVersion": PROTOCOL_VERSION, "capabilities": {"tools": {}}, "serverInfo": SERVER_INFO, "instructions": ( "Statbook is a registry of UK employment statutory " "figures — cited, cross-checked, re-verified daily, " "each with an explicit assurance status. It answers " "'what is/was figure X, as of date D, with proof'; it " "never answers eligibility or gives advice." ), } ) if method == "ping": return ok({}) if method == "tools/list": return ok({"tools": TOOLS}) if method == "tools/call": params = req.get("params", {}) name = params.get("name") args = params.get("arguments", {}) try: is_error, doc = call_tool(client, name, args) except KeyError as e: is_error, doc = True, { "error": {"code": "missing_argument", "message": str(e)} } return ok( { "content": [ {"type": "text", "text": json.dumps(doc, ensure_ascii=False, indent=2)} ], "isError": is_error, } ) return { "jsonrpc": "2.0", "id": rid, "error": {"code": -32601, "message": f"method not found: {method}"}, } def serve(stdin=None, stdout=None, client: Client | None = None) -> None: stdin = stdin or sys.stdin stdout = stdout or sys.stdout client = client or Client() for line in stdin: line = line.strip() if not line: continue try: req = json.loads(line) except json.JSONDecodeError: resp = { "jsonrpc": "2.0", "id": None, "error": {"code": -32700, "message": "parse error"}, } else: resp = handle_request(req, client) if resp is None: continue stdout.write(json.dumps(resp, ensure_ascii=False) + "\n") stdout.flush() if __name__ == "__main__": serve()