#!/usr/bin/env python3 """XGM MCP server: exposes the XGM public API (https://xgm.ro/api) as MCP tools over stdio. No dependencies beyond the Python 3.9+ standard library. Every tool is read-only and calls the public API, so the same rate limits and target checks apply as on the website. Claude Code: claude mcp add xgm -- python3 /path/to/xgm_mcp.py Environment: XGM_API_BASE (default https://xgm.ro), XGM_API_TIMEOUT seconds (default 30) """ from __future__ import annotations import json import os import re import sys import urllib.error import urllib.parse import urllib.request SERVER_INFO = {"name": "xgm", "title": "XGM network and email checks", "version": "1.0.0"} PROTOCOL_VERSIONS = ["2025-06-18", "2025-03-26", "2024-11-05"] API_BASE = os.environ.get("XGM_API_BASE", "https://xgm.ro").rstrip("/") TIMEOUT = float(os.environ.get("XGM_API_TIMEOUT", "30")) DOMAIN = {"type": "string", "description": "Domain name, for example example.com", "minLength": 3, "maxLength": 253} IP = {"type": "string", "description": "IPv4 or IPv6 address", "minLength": 2, "maxLength": 45} TOOLS = [ {"name": "dmarc_check", "title": "DMARC check", "path": "email-security", "section": "dmarc", "arg": "domain", "schema": DOMAIN, "description": "Check a domain's DMARC policy (RFC 7489), as in the XGM Email Security tool: policy strength, pct, subdomain policy, report addresses and their authorisation. Returns findings with severity and copyable DNS fixes."}, {"name": "spf_check", "title": "SPF check", "path": "email-security", "section": "spf", "arg": "domain", "schema": DOMAIN, "description": "Check a domain's SPF record (RFC 7208), as in the XGM Email Security tool: resolves includes and redirects, counts DNS lookups against the limit of 10, flags +all, ?all, void lookups and syntax errors."}, {"name": "dns_lookup", "title": "DNS records", "path": "dns", "arg": "domain", "schema": DOMAIN, "description": "Look up A, AAAA, MX, NS, TXT and CNAME records for a domain (XGM DNS Lookup; other record types and reverse lookups are on the web tool)."}, {"name": "ssl_check", "title": "TLS certificate", "path": "ssl", "arg": "domain", "schema": DOMAIN, "description": "Read the TLS certificate served on port 443: issuer, subject, expiry, days remaining, protocol version and cipher."}, {"name": "http_headers_check", "title": "HTTP security headers", "path": "http-headers", "arg": "domain", "schema": DOMAIN, "description": "Fetch https:// and report which security headers (HSTS, CSP, X-Frame-Options, X-Content-Type-Options, Referrer-Policy, Permissions-Policy) are present, with all response headers."}, {"name": "redirect_check", "title": "Redirect chain", "path": "redirect", "arg": "domain", "schema": DOMAIN, "description": "Follow redirects starting at http:// and list each hop and the final URL (XGM Redirect Checker)."}, {"name": "whois_lookup", "title": "WHOIS", "path": "whois", "arg": "domain", "schema": DOMAIN, "description": "Registrar, creation/expiry dates, status, name servers and abuse contact from the registry (RDAP first, WHOIS as fallback)."}, {"name": "blacklist_check", "title": "DNS blocklists", "path": "blacklist", "arg": "target", "schema": {"type": "string", "description": "IPv4 address, or a domain (its name, A and MX addresses are checked)", "minLength": 3, "maxLength": 253}, "description": "Check an IPv4 address against 48 DNS blocklists (Spamhaus, SpamCop, Barracuda, UCEPROTECT, Mailspike and more); for a domain, also its A and MX addresses and the name on 12 domain lists (Spamhaus DBL, SURBL, URIBL). Refused queries are reported separately from listings."}, {"name": "ip_lookup", "title": "IP lookup", "path": "ip", "arg": "ip", "schema": IP, "description": "Reverse DNS, scope, network (ASN) and approximate geolocation for an IP address (XGM IP Intelligence)."}, ] TOOLS_BY_NAME = {tool["name"]: tool for tool in TOOLS} def tool_listing() -> list[dict]: return [ { "name": tool["name"], "title": tool["title"], "description": tool["description"], "inputSchema": {"type": "object", "properties": {tool["arg"]: tool["schema"]}, "required": [tool["arg"]], "additionalProperties": False}, "annotations": {"title": tool["title"], "readOnlyHint": True, "destructiveHint": False, "idempotentHint": True, "openWorldHint": True}, } for tool in TOOLS ] def call_api(path: str, target: str) -> tuple[int, dict]: url = f"{API_BASE}/api/v1/{path}/{urllib.parse.quote(target, safe='')}" request = urllib.request.Request(url, headers={"Accept": "application/json", "User-Agent": f"xgm-mcp/{SERVER_INFO['version']}"}) try: with urllib.request.urlopen(request, timeout=TIMEOUT) as response: return response.status, json.loads(response.read().decode("utf-8")) except urllib.error.HTTPError as exc: try: body = json.loads(exc.read().decode("utf-8")) except (ValueError, UnicodeDecodeError): body = {"detail": exc.reason} if exc.code == 429: body["detail"] = f"{body.get('detail', 'Rate limited')} Retry after {exc.headers.get('Retry-After', '60')} s." return exc.code, body def summarize(tool: str, data: dict) -> str: lines = [f"XGM {tool} for {data.get('target')}"] if data.get("verdict"): lines.append(f"Verdict: {data['verdict']}") for key in ("record", "policy", "lookups", "final_url", "status_code", "ip", "listed_count", "hostname"): if data.get(key) not in (None, "", [], {}): lines.append(f"{key}: {data[key]}") if data.get("records"): lines.extend(f"{kind}: {', '.join(values)}" for kind, values in data["records"].items() if values) if data.get("errors"): lines.append(f"lookup errors: {data['errors']}") if data.get("tls"): tls = data["tls"] lines.append(f"certificate: issuer {tls.get('issuer')}, expires {tls.get('expires_at')} ({tls.get('days_remaining')} days), {tls.get('version')}") if data.get("security_headers"): lines.append("security headers: " + ", ".join(f"{name} {'present' if present else 'MISSING'}" for name, present in data["security_headers"].items())) for hop in data.get("redirects") or []: lines.append(f"redirect: {hop.get('status_code')} {hop.get('url')} -> {hop.get('location')}") if data.get("summary"): lines.extend(f"{key}: {value}" for key, value in data["summary"].items() if value) for zone in data.get("zones") or []: lines.append(f"{zone['zone']}: {'LISTED' if zone['listed'] else 'query refused' if zone['refused'] else 'not listed'}") if data.get("geo", {}).get("success"): geo = data["geo"] lines.append(f"location: {', '.join(part for part in (geo.get('city'), geo.get('country')) if part)}; network: {geo.get('org') or geo.get('isp')}") for finding in data.get("findings") or []: lines.append(f"[{finding['severity'].upper()}] {finding['title']}: {finding['explanation']}") if finding.get("fix"): lines.append(f" fix ({finding['fix']['label']}): {finding['fix']['code']}") if data.get("include_tree"): lines.extend(["include tree:", data["include_tree"]]) lines.append(f"Details: {data.get('web_url')}") return "\n".join(lines) def call_tool(name: str, arguments: dict) -> dict: tool = TOOLS_BY_NAME.get(name) if not tool: raise JsonRpcError(-32602, f"Unknown tool: {name}") value = arguments.get(tool["arg"]) if isinstance(arguments, dict) else None if not isinstance(value, str) or not value.strip() or len(value) > tool["schema"]["maxLength"] or re.search(r"[\s/?#]", value.strip()): return {"content": [{"type": "text", "text": f"Invalid {tool['arg']}: pass a bare {tool['arg']} such as {'example.com' if tool['arg'] != 'ip' else '8.8.8.8'}."}], "isError": True} try: status, data = call_api(tool["path"], value.strip()) except (urllib.error.URLError, TimeoutError, OSError) as exc: return {"content": [{"type": "text", "text": f"The XGM API at {API_BASE} could not be reached: {exc}"}], "isError": True} if status != 200: return {"content": [{"type": "text", "text": f"XGM API error {status}: {data.get('detail', 'request failed')}"}], "isError": True} if tool.get("section"): # One API call returns SPF and DMARC; each MCP tool reports its own part. data = {**{key: value for key, value in data.items() if key not in ("spf", "dmarc", "verdict", "tone", "findings")}, **data[tool["section"]]} return {"content": [{"type": "text", "text": summarize(name, data)}], "structuredContent": data, "isError": False} class JsonRpcError(Exception): def __init__(self, code: int, message: str): super().__init__(message) self.code = code self.message = message def handle(message: dict): method = message.get("method") params = message.get("params") or {} if method == "initialize": requested = params.get("protocolVersion") return { "protocolVersion": requested if requested in PROTOCOL_VERSIONS else PROTOCOL_VERSIONS[0], "capabilities": {"tools": {"listChanged": False}}, "serverInfo": SERVER_INFO, "instructions": "Read-only checks for public domains and IPs via the XGM API. Findings include severity and DNS fixes. Private and internal targets are refused.", } if method == "ping": return {} if method == "tools/list": return {"tools": tool_listing()} if method == "tools/call": return call_tool(params.get("name", ""), params.get("arguments") or {}) raise JsonRpcError(-32601, f"Method not found: {method}") def main() -> None: for line in sys.stdin: line = line.strip() if not line: continue try: message = json.loads(line) except ValueError: reply = {"jsonrpc": "2.0", "id": None, "error": {"code": -32700, "message": "Parse error"}} else: if "id" not in message: continue # notification (for example notifications/initialized) try: reply = {"jsonrpc": "2.0", "id": message["id"], "result": handle(message)} except JsonRpcError as exc: reply = {"jsonrpc": "2.0", "id": message["id"], "error": {"code": exc.code, "message": exc.message}} except Exception as exc: # never crash the session on one bad call reply = {"jsonrpc": "2.0", "id": message["id"], "error": {"code": -32603, "message": f"Internal error: {exc.__class__.__name__}"}} sys.stdout.write(json.dumps(reply) + "\n") sys.stdout.flush() if __name__ == "__main__": main()