#!/usr/bin/env python3 """rtl-signal-mcp — MCP server exposing RTL signal-flow tracing to AI agents. Protocol: MCP over stdio (JSON-RPC 2.0). No third-party MCP SDK required. Tools (token-efficient by design — every answer is a compact graph, not source): build_index build/replace the index from file/dir roots list_modules module names + instantiation counts module_summary ports / instances / net count of one module find_signal regex search over all signals -> module + file:line trace_drivers who drives this signal (recursively, across module boundaries) trace_loads what does this signal drive (recursively, up and down) get_source small excerpt of a file by line range hierarchy instantiation tree under a module Config: env RTL_MCP_ROOTS="/path/a:/path/b" pre-builds at startup. """ import os, sys, json, io sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", write_through=True) sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8", write_through=True) from rtl_index import RtlIndex IDX = None # global index def _need_index(): if IDX is None or not IDX.db: raise RuntimeError("index empty: call build_index first") # ---------------- tool implementations ---------------- def t_build_index(args): global IDX roots = args.get("roots") or [r for r in os.environ.get("RTL_MCP_ROOTS", "").split(":") if r] if not roots: return dict(error="no roots given (pass roots[] or set RTL_MCP_ROOTS)") t0 = os.times() IDX = RtlIndex.build(roots) n_always = sum(len(m["always"]) for m in IDX.db.values()) n_assign = sum(len(m["assigns"]) for m in IDX.db.values()) return dict(modules=len(IDX.db), cross_edges=len(IDX.edges), assign_facts=n_assign, always_facts=n_always, note="index in memory; use find_signal to locate, trace_drivers/trace_loads to follow") def t_list_modules(args): _need_index() pat = args.get("pattern") import re rx = re.compile(pat, re.I) if pat else None out = [] for name, m in IDX.db.items(): if rx and not rx.search(name): continue out.append(dict(module=name, ports=len(m["ports"]), insts=len(m["insts"]), assigns=len(m["assigns"]), always=len(m["always"]))) return dict(modules=out[:80], total=len(out)) def t_module_summary(args): _need_index() s = IDX.module_summary(args["module"]) if s is None: return dict(error=f"module {args['module']} not found") return s def t_find_signal(args): _need_index() hits = IDX.search(args["pattern"], limit=int(args.get("limit", 20))) return dict(hits=hits) def _fmt_trace(rows): lines = [] for r in rows: rhs = ", ".join(r["rhs"][:6]) lines.append(f"[{r['kind']}] {r['where']} :: {r['sig']} <- {rhs}") return lines def t_trace_drivers(args): _need_index() rows = IDX.drivers(args["module"], args["signal"], max_depth=int(args.get("depth", 4))) if not rows: return dict(result=[], text=f"no driver found for {args['module']}.{args['signal']} " "(top input / testbench-driven / not indexed)") lines = _fmt_trace(rows) return dict(result=rows, text="\n".join(lines), note="read chain bottom-up: leaf entries are the ultimate drivers; " "use get_source for file:line details") def t_trace_loads(args): _need_index() rows = IDX.loads(args["module"], args["signal"], max_depth=int(args.get("depth", 3))) if not rows: return dict(result=[], text=f"no loads found for {args['module']}.{args['signal']}") lines = _fmt_trace(rows) return dict(result=rows, text="\n".join(lines)) _FMAP = None def _file_map(): """basename -> absolute path, built lazily by walking indexed roots.""" global _FMAP if _FMAP is None: _FMAP = {} for r in getattr(IDX, "_roots", []) or []: if os.path.isfile(r): _FMAP.setdefault(os.path.basename(r), r) continue for dp, _dns, fns in os.walk(r): for fn in fns: _FMAP.setdefault(fn, os.path.join(dp, fn)) return _FMAP def t_get_source(args): path, start, end = args["file"], int(args["start"]), int(args["end"]) end = min(end, start + 200) if not os.path.isabs(path): # resolve by basename against indexed roots (files may sit in subdirs) cand = _file_map().get(os.path.basename(path)) if cand: path = cand if not os.path.exists(path): return dict(error=f"file not found: {path}") out = [] with open(path, encoding="utf-8", errors="replace") as f: for i, line in enumerate(f, 1): if start <= i <= end: out.append(f"{i:5d}| {line.rstrip()}") if i > end: break return dict(text="\n".join(out)) def t_hierarchy(args): _need_index() top = args["module"] depth = int(args.get("depth", 2)) lines = [] def rec(mod, indent, d, seen): if d > depth or (mod, indent) in seen or len(lines) > 80: return seen.add((mod, indent)) m = IDX.db.get(mod) if not m: return for (itype, iname, conns) in m["insts"]: lines.append(f"{' ' * (indent + 1)}└─ {iname} : {itype}") rec(itype, indent + 1, d + 1, seen) lines.append(top) rec(top, 0, 0, set()) return dict(text="\n".join(lines)) TOOLS = { "build_index": (t_build_index, "Build the signal-flow index. roots: list of files or directories to scan (.v/.sv)."), "list_modules": (t_list_modules, "List indexed modules (name, ports, insts, fact counts). pattern: optional regex."), "module_summary": (t_module_summary, "Compact summary of one module: port list, instances with connected nets, net names."), "find_signal": (t_find_signal, "Find signals by regex. Returns module + file:line for each hit."), "trace_drivers": (t_trace_drivers, "Trace WHO drives a signal — recursive across module boundaries (Verdi-style). Args: module, signal, depth=4."), "trace_loads": (t_trace_loads, "Trace WHAT a signal drives — loads in same module, flows out output ports, feeds child instances. Args: module, signal, depth=3."), "get_source": (t_get_source, "Read a small source excerpt (file, start, end; max 200 lines). Use after tracing to see the exact logic."), "hierarchy": (t_hierarchy, "Instantiation tree under a module. Args: module, depth=2."), } TOOLSPEC = [dict(name=n, description=d, inputSchema=dict(type="object", properties={}, additionalProperties=True)) for n, (f, d) in TOOLS.items()] # ---------------- JSON-RPC / MCP plumbing ---------------- def reply(msg_id, result): sys.stdout.write(json.dumps(dict(jsonrpc="2.0", id=msg_id, result=result)) + "\n") def reply_error(msg_id, code, message): sys.stdout.write(json.dumps(dict(jsonrpc="2.0", id=msg_id, error=dict(code=code, message=message))) + "\n") def handle(req): method = req.get("method") msg_id = req.get("id") if method == "initialize": reply(msg_id, dict(protocolVersion="2024-11-05", capabilities=dict(tools={}), serverInfo=dict(name="rtl-signal-mcp", version="0.1.0"))) elif method == "notifications/initialized": pass elif method == "tools/list": reply(msg_id, dict(tools=TOOLSPEC)) elif method == "tools/call": params = req.get("params", {}) name = params.get("name") args = params.get("arguments", {}) or {} fn = TOOLS.get(name) if fn is None: reply_error(msg_id, -32602, f"unknown tool {name}") return try: res = fn[0](args) text = res.pop("text", None) or json.dumps(res, ensure_ascii=False) reply(msg_id, dict(content=[dict(type="text", text=text)], structuredContent=res, isError=False)) except Exception as e: reply(msg_id, dict(content=[dict(type="text", text=f"error: {e}")], isError=True)) elif method == "ping": reply(msg_id, {}) elif msg_id is not None: reply_error(msg_id, -32601, f"method not supported: {method}") def main(): if os.environ.get("RTL_MCP_ROOTS"): try: res = t_build_index({}) print(f"[rtl-signal-mcp] prebuilt index: {res}", file=sys.stderr) except Exception as e: print(f"[rtl-signal-mcp] prebuild failed: {e}", file=sys.stderr) for line in sys.stdin: line = line.strip() if not line: continue try: req = json.loads(line) except Exception: continue try: handle(req) except Exception as e: if req.get("id") is not None: reply_error(req["id"], -32603, str(e)) if __name__ == "__main__": main()