#!/usr/bin/env python3 # atlas-mcp — ATLAS Model Context Protocol server. Pure stdlib, zero deps. # # Serves THIS project's orientation map (ATLAS.md §0-1), the SKILL/SCARS # playbooks, and the measured token reduction to ANY MCP client — Claude Code, # Cursor, OpenClaw, Codex, Gemini, … — so the agent orients from the map instead # of grepping. stdio transport by default: the OS process boundary is the auth, # so no token is needed (or wanted) locally. # # The deep-query tools (atlas_graph / atlas_deepsearch / atlas_recall) appear # ONLY when ATLAS_MCP_BACKEND_URL is set; they proxy to a graph+vector+memory # backend (e.g. FuseGraph / FuseRAG) with an optional bearer token. With nothing # configured, ATLAS stays 100% local + free — the backend is a bring-your-own # upsell, never a dependency. # # Env: # ATLAS_PROJECT project root to serve (default: cwd) # ATLAS_BIN path to the atlas CLI (default: sibling ./atlas) # ATLAS_VERSION reported serverInfo.version (default: hardcoded below) # ATLAS_MCP_BACKEND_URL opt-in deep backend base URL — lights up graph/deepsearch/recall # ATLAS_MCP_BACKEND_TOKEN bearer token sent to that backend # # Launch: atlas mcp (stdio — what Claude Code/Cursor/OpenClaw spawn) # Spec: docs/SPEC.md · part of the ATLAS standard (https://github.com/Abbasi-Alain/atlas) import sys, os, re, json, glob, subprocess, shutil, urllib.request, http.server VERSION = "0.1.11" DEFAULT_PROTO = "2025-06-18" def log(*a): print("[atlas-mcp]", *a, file=sys.stderr, flush=True) def proj(): return os.environ.get("ATLAS_PROJECT") or os.getcwd() def _read(path, limit=200000): try: with open(path, encoding="utf-8", errors="replace") as f: return f.read(limit) except Exception: return "" def _atlas_md(): return os.path.join(proj(), "ATLAS.md") def _scars_md(): return os.path.join(proj(), "SCARS.md") def _skill_md(): hits = glob.glob(os.path.join(proj(), ".agents", "skill", "*", "SKILL.md")) return hits[0] if hits else "" def _until(text, stop_re): """Lines from the top until (exclusive) a line matching stop_re.""" out = [] for line in text.splitlines(): if re.match(stop_re, line): break out.append(line) return "\n".join(out).strip() def _toc(text): """The '## Table of contents' block of a SKILL/SCARS file, if present.""" grab, out = False, [] for line in text.splitlines(): if re.match(r'^##\s+Table of contents', line, re.I): grab = True continue if grab and re.match(r'^##\s+\S', line) and "contents" not in line.lower(): break if grab: out.append(line) return "\n".join(out).strip() # ---- tools: local (zero-infra) ------------------------------------------- _STOP = set("the a an of to in at on for and or is are be this that it with you your i we our how " "where what when why add fix get set use run make new via from into can does not".split()) def _tok(s): return set(w for w in re.findall(r"[a-z0-9_]+", s.lower()) if len(w) > 2 and w not in _STOP) def _rank(lines, qt, top): """Rank lines by keyword overlap with the task tokens (zero-infra relevance).""" scored = sorted(((len(_tok(ln) & qt), ln.strip()) for ln in lines if ln.strip()), key=lambda x: -x[0]) out, seen = [], set() for s, ln in scored: if s and ln not in seen: seen.add(ln) out.append(ln) if len(out) >= top: break return out def t_orient(args): a = _read(_atlas_md()) if not a: return "No ATLAS.md in %s — run `atlas init` to scaffold the map." % proj() task = (args.get("task") or "").strip() if not task: parts = ["# ATLAS — where things live (read this before you grep)\n", _until(a, r'^## 2\.')] # §0 quick-orientation + §1 top-level map sk = _read(_skill_md()) if sk: parts += ["\n## SKILL.md — task playbook (stable anchors)\n", _toc(sk)] sc = _read(_scars_md()) if sc: parts += ["\n## SCARS.md — failure memory (don't repeat)\n", _toc(sc)] return "\n".join(p for p in parts if p) # task-aware: return only the RELEVANT slice — map entries + playbook + the # SCARS that bite *this* task — ranked by keyword overlap. No one else does this. qt = _tok(task) map_rows = [l for l in a.splitlines() if "](" in l or ("|" in l and l.count("|") >= 2)] sk = _read(_skill_md()) skill_anchors = [l for l in sk.splitlines() if l.startswith("### ") or l.lstrip().startswith("- [")] sc = _read(_scars_md()) scar_anchors = [b.splitlines()[0] for b in re.split(r"(?=^### §)", sc, flags=re.M) if b.strip().startswith("### §")] out = ["# ATLAS — oriented for: %s\n" % task, "## Where to look (most relevant map entries)"] out += (_rank(map_rows, qt, 8) or ["(no direct match — start from ATLAS.md §0)"]) sh = _rank(skill_anchors, qt, 5) if sh: out += ["", "## Relevant playbook (SKILL anchors)"] + sh ch = _rank(scar_anchors, qt, 5) if ch: out += ["", "## Watch out for (relevant SCARS — what breaks here)"] + ch out += ["", "_(full map: call atlas_orient with no task, or read ATLAS.md §0)_"] return "\n".join(out) def t_find(args): q = (args.get("query") or "").strip() if not q: return "Pass a `query` (e.g. 'release pipeline')." rx = re.compile(re.escape(q), re.I) hits = [] for label, path in (("ATLAS", _atlas_md()), ("SKILL", _skill_md()), ("SCARS", _scars_md())): for i, line in enumerate(_read(path).splitlines(), 1): if line.strip() and rx.search(line): hits.append("%s:%d %s" % (label, i, line.strip()[:160])) if not hits: return "No map entry matches %r — it may be undocumented; consider adding it to ATLAS.md." % q return "Map entries matching %r:\n\n%s" % (q, "\n".join(hits[:40])) def t_scars(args): sc = _read(_scars_md()) if not sc: return "No SCARS.md in this project." blocks = [b for b in re.split(r'(?=^### §)', sc, flags=re.M) if b.strip().startswith("### §")] q = (args.get("query") or "").strip() if q: rx = re.compile(re.escape(q), re.I) blocks = [b for b in blocks if rx.search(b)] if not blocks: return "No SCARS anchor matches %r — see `atlas anchors` for the full list." % q return ("".join(blocks))[:8000] or "No anchors found." def t_measure(_args): atlas_bin = os.environ.get("ATLAS_BIN") or os.path.join( os.path.dirname(os.path.abspath(__file__)), "atlas") try: r = subprocess.run([atlas_bin, "measure"], cwd=proj(), capture_output=True, text=True, timeout=30) out = re.sub(r'\x1b\[[0-9;]*m', '', (r.stdout or "") + (r.stderr or "")) return out.strip() or "measure produced no output." except Exception as e: return "measure unavailable: %s" % e # ---- tools: optional backend (FuseGraph / FuseRAG) ----------------------- def _backend(path, payload): base = os.environ["ATLAS_MCP_BACKEND_URL"].rstrip("/") req = urllib.request.Request(base + path, data=json.dumps(payload).encode(), headers={"Content-Type": "application/json"}) tok = os.environ.get("ATLAS_MCP_BACKEND_TOKEN") if tok: req.add_header("Authorization", "Bearer " + tok) with urllib.request.urlopen(req, timeout=60) as r: return r.read().decode("utf-8", "replace") def _which(*names): for n in names: if shutil.which(n): return n return None def _route_deep(kind, query): """ATLAS as conductor: prefer the configured backend (FuseGraph/FuseRAG); else route to an installed ecosystem tool (graphify, CodeGraphContext). Free + local first.""" if os.environ.get("ATLAS_MCP_BACKEND_URL"): return _backend({"search": "/search", "graph": "/graph", "recall": "/recall"}[kind], {"project": proj(), "query": query}) gf = _which("graphify") if gf and kind in ("search", "graph"): return _shell([gf, "query", query], "graphify") cgc = _which("cgc", "codegraphcontext") if cgc and kind == "graph": return _shell([cgc, "query", query], "CodeGraphContext") return ("No deep-%s backend available. Either set ATLAS_MCP_BACKEND_URL (e.g. FuseGraph/FuseRAG), " "or install graphify / CodeGraphContext — ATLAS routes to whatever's present." % kind) def _shell(cmd, name): try: r = subprocess.run(cmd, cwd=proj(), capture_output=True, text=True, timeout=120) return (r.stdout or r.stderr or "").strip() or "(%s returned nothing)" % name except Exception as e: return "%s error: %s" % (name, e) def t_graph(args): return _route_deep("graph", args.get("query", "")) def t_deepsearch(args): return _route_deep("search", args.get("query", "")) def t_recall(args): return _route_deep("recall", args.get("query", "")) # ---- visual map (shareable Mermaid / HTML) ------------------------------- def _cell_name(s): m = re.search(r"`([^`]+)`", s) or re.search(r"\[([^\]]+)\]", s) return (m.group(1) if m else s).strip().rstrip("/") def _mid(name): return re.sub(r"[^A-Za-z0-9]", "_", name) or "n" def _render_ascii(title, nodes, edges): """A terminal-friendly Unicode graph: fan-out adjacency + standalone modules.""" out_adj, indeg, seen = {}, {nid: 0 for nid in nodes}, set() for s, d in edges: if (s, d) in seen: continue seen.add((s, d)) out_adj.setdefault(s, []).append(d) indeg[d] = indeg.get(d, 0) + 1 L = [" \U0001f5fa %s — module map" % title, " " + "─" * 46 + " from ATLAS.md §1", ""] any_edge = False for nid in nodes: tgts = out_adj.get(nid) if not tgts: continue any_edge = True if len(tgts) == 1: L.append(" %s ──▶ %s" % (nodes[nid], nodes[tgts[0]])) else: L.append(" %s" % nodes[nid]) for i, t in enumerate(tgts): conn = "└─▶" if i == len(tgts) - 1 else "├─▶" L.append(" %s %s" % (conn, nodes[t])) iso = [nodes[nid] for nid in nodes if nid not in out_adj and indeg.get(nid, 0) == 0] if iso: if any_edge: L.append("") L.append(" ○ standalone: " + " ".join(iso)) L += ["", " %d modules · %d links (--mermaid for Markdown · --html for a page)" % (len(nodes), len(seen))] return "\n".join(L) def render_map(html=False, fmt=None): """Render the repo graph from ATLAS.md §1's | Node | Role | Talks-to | table. fmt: 'ascii' (terminal), 'mermaid' (Markdown), or 'html'. html=True ⇒ 'html'.""" if fmt is None: fmt = "html" if html else "mermaid" a = _read(_atlas_md()) if not a: return "No ATLAS.md in %s — run `atlas init`." % proj() nodes, edges, in_tbl = {}, [], False for line in a.splitlines(): if re.match(r"^\|\s*Node\s*\|", line): in_tbl = True continue if in_tbl: if not line.lstrip().startswith("|"): break cells = [c.strip() for c in line.strip().strip("|").split("|")] if not cells or set("".join(cells)) <= set("-: "): continue name = _cell_name(cells[0]) if not name or name.lower() == "node": continue nid = _mid(name) nodes[nid] = name for ref in re.split(r"[,;]", cells[2] if len(cells) >= 3 else ""): rn = _cell_name(ref) if rn and " " not in rn and not rn.startswith("§") and rn.lower() not in ("—", "-", "n/a", "", "(none)", "none", "tbd"): rid = _mid(rn) nodes.setdefault(rid, rn) if rid != nid: edges.append((nid, rid)) if not nodes: return "No module table in ATLAS.md §1 — add a | Node | Role | Talks-to | table, then re-run." title = os.path.basename(proj().rstrip("/")) or "project" if fmt == "ascii": return _render_ascii(title, nodes, edges) mer = ["flowchart LR"] mer += [' %s["%s"]' % (nid, lab.replace('"', "'")) for nid, lab in nodes.items()] mer += [" %s --> %s" % (s, d) for (s, d) in dict.fromkeys(edges)] mermaid = "\n".join(mer) if fmt != "html": return "```mermaid\n%s\n```" % mermaid return _MAP_HTML % {"title": title, "mermaid": mermaid} _MAP_HTML = """ %(title)s — ATLAS map

\U0001f5fa️ %(title)s — ATLAS map

the repo's brain, from ATLAS.md §1 · generated by atlas map · ATLAS
%(mermaid)s
""" # ---- registry ------------------------------------------------------------ def _obj(props, required=()): return {"type": "object", "properties": props, "required": list(required)} Q = {"query": {"type": "string", "description": "natural-language query"}} LOCAL_TOOLS = [ ("atlas_orient", "Where things live. Call this FIRST to orient instead of grepping. Pass an optional `task` to get just the RELEVANT slice — the map entries, playbook anchors, and the SCARS that bite *this* task; omit it for the full §0 map + ToCs.", _obj({"task": {"type": "string", "description": "the task you're about to do — returns only the relevant slice of the map"}}), t_orient), ("atlas_find", "Find where something lives in the project map (files/modules/anchors) by keyword.", _obj(Q, ["query"]), t_find), ("atlas_scars", "Hard-won failure lessons (SCARS anchors) — what breaks and how to avoid it. Optionally filter by query.", _obj(Q), t_scars), ("atlas_measure", "The measured orientation-token reduction ATLAS buys this repo (vs a smart skim and a whole-repo dump).", _obj({}), t_measure), ] BACKEND_TOOLS = [ ("atlas_graph", "Deep structural query (callers/callees, call chains, dead code). ATLAS routes to the configured backend (FuseGraph) or an installed graph tool (CodeGraphContext / graphify).", _obj(Q, ["query"]), t_graph), ("atlas_deepsearch", "Semantic code/doc search. Routes to the configured vector backend (FuseRAG) or an installed tool (graphify).", _obj(Q, ["query"]), t_deepsearch), ("atlas_recall", "Recall relevant context from prior sessions (needs a configured backend, e.g. FuseRAG).", _obj(Q, ["query"]), t_recall), ] def _router_available(): return bool(os.environ.get("ATLAS_MCP_BACKEND_URL")) or bool(_which("graphify", "cgc", "codegraphcontext")) def tools(): t = list(LOCAL_TOOLS) if _router_available(): # deep tools appear when a backend OR an ecosystem tool is present t += BACKEND_TOOLS return t # ---- JSON-RPC / MCP ------------------------------------------------------ def _ok(i, result): return {"jsonrpc": "2.0", "id": i, "result": result} def _err(i, code, m): return {"jsonrpc": "2.0", "id": i, "error": {"code": code, "message": m}} def handle(msg): method, i = msg.get("method"), msg.get("id") if method == "initialize": proto = (msg.get("params") or {}).get("protocolVersion") or DEFAULT_PROTO return _ok(i, {"protocolVersion": proto, "capabilities": {"tools": {}}, "serverInfo": {"name": "atlas", "version": os.environ.get("ATLAS_VERSION", VERSION)}}) if method in ("notifications/initialized", "initialized"): return None if method == "ping": return _ok(i, {}) if method == "tools/list": return _ok(i, {"tools": [{"name": n, "description": d, "inputSchema": s} for (n, d, s, _f) in tools()]}) if method == "tools/call": p = msg.get("params") or {} fn = dict((n, f) for (n, _d, _s, f) in tools()).get(p.get("name")) if not fn: return _err(i, -32602, "unknown tool: %s" % p.get("name")) try: return _ok(i, {"content": [{"type": "text", "text": fn(p.get("arguments") or {})}], "isError": False}) except Exception as e: return _ok(i, {"content": [{"type": "text", "text": "error: %s" % e}], "isError": True}) if i is None: return None # unknown notification — ignore return _err(i, -32601, "method not found: %s" % method) def serve_stdio(): log("ATLAS MCP (stdio) v%s — project: %s%s" % ( os.environ.get("ATLAS_VERSION", VERSION), proj(), " [deep backend: ON]" if os.environ.get("ATLAS_MCP_BACKEND_URL") else "")) for line in sys.stdin: line = line.strip() if not line: continue try: msg = json.loads(line) except Exception as e: log("bad JSON-RPC line:", e) continue for one in (msg if isinstance(msg, list) else [msg]): resp = handle(one) if resp is not None: sys.stdout.write(json.dumps(resp) + "\n") sys.stdout.flush() def serve_http(host, port, token): # Minimal MCP-over-HTTP (Streamable HTTP, non-streaming): POST JSON-RPC → # JSON response. Optional bearer-token auth — opt-in, for team/remote use. ver = os.environ.get("ATLAS_VERSION", VERSION) class Handler(http.server.BaseHTTPRequestHandler): protocol_version = "HTTP/1.1" def _send(self, code, body=b"", ctype="application/json"): if isinstance(body, str): body = body.encode() self.send_response(code) if body: self.send_header("Content-Type", ctype) self.send_header("Content-Length", str(len(body))) self.end_headers() if body: self.wfile.write(body) def _authed(self): return (not token) or self.headers.get("Authorization", "") == "Bearer " + token def do_GET(self): if self.path.rstrip("/") in ("/health", "/healthz"): return self._send(200, json.dumps({"status": "ok", "server": "atlas", "version": ver})) return self._send(405, json.dumps({"error": "POST JSON-RPC to this endpoint"})) def do_POST(self): if not self._authed(): self.send_response(401) self.send_header("WWW-Authenticate", "Bearer") self.send_header("Content-Length", "0") self.end_headers() return n = int(self.headers.get("Content-Length") or 0) raw = self.rfile.read(n).decode("utf-8", "replace") if n else "" try: msg = json.loads(raw) except Exception as e: return self._send(400, json.dumps(_err(None, -32700, "parse error: %s" % e))) out = [r for r in (handle(o) for o in (msg if isinstance(msg, list) else [msg])) if r is not None] if not out: return self._send(202) # notification(s) only return self._send(200, json.dumps(out if isinstance(msg, list) else out[0])) def log_message(self, *a): pass # quiet; the startup line already went to stderr httpd = http.server.ThreadingHTTPServer((host, port), Handler) log("ATLAS MCP (http) v%s on http://%s:%d project: %s [auth: %s]" % ( ver, host, port, proj(), "ON" if token else "OFF")) if not token and host not in ("127.0.0.1", "localhost", "::1"): log("WARNING: bound to %s with no --token — anyone on the network can query it." % host) try: httpd.serve_forever() except KeyboardInterrupt: pass def main(): args = sys.argv[1:] if args and args[0] == "--orient": # one-shot CLI: `atlas orient [task]` print(t_orient({"task": " ".join(args[1:])})) return if args and args[0] == "--map": # one-shot CLI: `atlas map [--ascii|--mermaid|--html]` rest = args[1:] fmt = "html" if "--html" in rest else "ascii" if "--ascii" in rest else "mermaid" print(render_map(fmt=fmt)) return if "--http" in args: serve_http(os.environ.get("ATLAS_MCP_HOST", "127.0.0.1"), int(os.environ.get("ATLAS_MCP_PORT", "7332")), os.environ.get("ATLAS_MCP_TOKEN") or None) else: serve_stdio() if __name__ == "__main__": main()