import re, sys, pathlib, json
CODE = {".ts", ".js", ".mjs", ".py", ".go", ".rs"}
SKIP = ("node_modules", ".git", "dist", "build", "target", "__pycache__",
"test", "tests", "__tests__", "spec", "e2e", "examples")
EFFECT = re.compile(r"\b(pay|payment|charge|transfer|payout|refund|settle|invoice|spend|"
r"send|sms|email|dispatch|notify|publish|submit|order|purchase|buy|sell|"
r"swap|mint|withdraw|deposit|create|deploy|provision|launch|terminate|"
r"delete|destroy|update|write|execute|trigger|enroll|register)\b", re.I)
MONEY = re.compile(r"\b(pay|payment|charge|transfer|payout|refund|settle|invoice|spend|"
r"usdc|wallet|billing|purchase|buy|sell|swap|mint|withdraw|deposit|"
r"escrow|x402|budget|checkout)\b", re.I)
READ = re.compile(r"\b(get|list|read|fetch|search|query|lookup|resolve|retrieve|check|"
r"describe|show|view|find|count|status|inspect|validate|verify)\b", re.I)
WRITE_CALL = re.compile(
r"(\.post\s*\(|\.put\s*\(|\.patch\s*\(|\.delete\s*\(|"
r"requests\.(post|put|patch|delete)|axios\.(post|put|patch|delete)|"
r"httpx\.(post|put|patch|delete)|method\s*[:=][^,;\n]*(POST|PUT|PATCH|DELETE)|"
r"\.create\s*\(|\.send\s*\(|\.submit\s*\(|\.execute\s*\(|"
r"\.sign(AndSubmit|Transaction)?\s*\(|submitTransaction|sendTransaction|"
r"\.insert\s*\(|\.save\s*\()", re.I)
GUARD = re.compile(r"(idempot|dedup|alreadySent|already_sent|alreadyPaid|already_paid|"
r"alreadyProcessed|already_processed|exactly[- ]?once|effectfence|"
r"once\.run|seenKeys?|seen_keys?|seenIds?|seen_ids?|"
r"processedIds?|processed_ids?|replayProtect|replay_protect|"
r"requestId|request_id|clientToken|client_token|"
r"transactionKey|transaction_key|dedupeKey|dedupe_key)", re.I)
RETRY = re.compile(r"\b(retry|retries|backoff|max_?attempts|reattempt)\b", re.I)
STRLIT = re.compile(r'''"[^"]*"|\'[^\']*\'|`[^`]*`''')
TOOLNAME = re.compile(
r"""(?:name:\s*["']([a-z0-9_.\-]{3,60})["']"""
r"""|@mcp\.tool\(\s*\)?\s*(?:\n\s*)?def\s+([a-z0-9_]{3,60})"""
r"""|Tool\(\s*["']([a-z0-9_.\-]{3,60})["'])""", re.I)
def _own_block(txt: str, start: int, limit: int = 4000) -> str:
i = txt.find("{", start)
if i == -1:
return txt[start:start + 300]
depth, j, end = 0, i, min(len(txt), i + limit)
while j < end:
c = txt[j]
if c == "{":
depth += 1
elif c == "}":
depth -= 1
if depth == 0:
return txt[start:j + 1]
j += 1
return txt[start:end]
def scan(root: pathlib.Path):
tools, guards, retries, writes = {}, [], [], []
for p in root.rglob("*"):
if not p.is_file() or p.suffix not in CODE:
continue
if any(d in [q.lower() for q in p.parts] for d in SKIP):
continue
try:
txt = p.read_text(encoding="utf-8", errors="ignore")
except Exception:
continue
rel = str(p.relative_to(root))
for m in TOOLNAME.finditer(txt):
nm = next(g for g in m.groups() if g)
block = _own_block(txt, m.start())
if not re.search(r"(description|handler|inputSchema|input_schema|callback|execute)",
block, re.I):
continue
if nm in tools:
continue
tools[nm] = {
"tool": nm, "file": rel,
"line": txt[:m.start()].count("\n") + 1,
"block": block,
}
for i, line in enumerate(txt.splitlines(), 1):
code_only = STRLIT.sub("", line) if line.strip().startswith(("//", "#", "*", "/*", "/**")):
continue
if GUARD.search(code_only):
guards.append({"file": rel, "line": i, "code": line.strip()[:120]})
if RETRY.search(code_only):
retries.append({"file": rel, "line": i, "code": line.strip()[:120]})
if WRITE_CALL.search(line):
writes.append({"file": rel, "line": i, "code": line.strip()[:120]})
candidates = []
for t in tools.values():
blk, nm = t["block"], t["tool"]
if READ.search(nm) and not MONEY.search(nm):
continue
if not (EFFECT.search(nm) or EFFECT.search(blk)):
continue
t["writes_in_same_file"] = len([w for w in writes if w["file"] == t["file"]])
t["money"] = bool(MONEY.search(nm) or MONEY.search(blk))
t.pop("block")
candidates.append(t)
candidates.sort(key=lambda c: (c["money"], c["writes_in_same_file"]), reverse=True)
return tools, candidates, guards, retries, writes
if __name__ == "__main__":
if len(sys.argv) < 2:
sys.exit("usage: fencescan.py <path-to-repo>")
root = pathlib.Path(sys.argv[1])
tools, candidates, guards, retries, writes = scan(root)
print(json.dumps({
"repo": root.name,
"summary": {
"tools_declared": len(tools),
"candidates": len(candidates),
"write_sites_in_repo": len(writes),
"money_candidates": len([c for c in candidates if c["money"]]),
"guard_sites_found_in_this_repo": len(guards),
"retry_sites_found_in_this_repo": len(retries),
},
"candidates": candidates[:10],
"guards_found": guards[:5],
"retry_sites": retries[:5],
"write_sites": writes[:5],
"retry_note": ("Retry logic that does not branch on HTTP method will retry writes. "
"If these sites wrap a create, a timeout can land the effect twice."
if retries else None),
"unknowns_this_tool_CANNOT_see": [
"Whether the API being called deduplicates server-side.",
"Whether a guard lives in a sibling repo or a separate SDK "
"(CryptoAPIs' idempotency key lived in a different published package).",
"Whether a tool that looks like a write only returns a payload for "
"someone else to sign (XRPName did exactly this).",
"Whether the package is actually used by anyone — check downloads first.",
],
"how_to_use_this": [
"1. Open each candidate's file:line and confirm the handler really writes.",
"2. Ask whether a retried call reaches the same effect twice.",
"3. If a guard exists off-repo, this tool cannot see it — go look.",
"4. Assert only what is visibly client-side. Otherwise ask a question.",
],
}, indent=1))