admin/riscv-ulp-survey
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
Files
main
riscv-ulp-survey/rtl-mcp/mcp_server.py
T

254 lines
9.1 KiB
Python
Raw Permalink Blame History

#!/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()
Reference in New Issue View Git Blame Copy Permalink