wehub-resource-sync
1694 lines
73 KiB
Python
1694 lines
73 KiB
Python
# MCP stdio server - exposes graph query tools to Claude and other agents
|
|
from __future__ import annotations
|
|
import json
|
|
import math
|
|
import re
|
|
import sys
|
|
from array import array
|
|
from pathlib import Path
|
|
import networkx as nx
|
|
from networkx.readwrite import json_graph
|
|
from graphify.security import sanitize_label, check_graph_file_size_cap
|
|
from graphify.build import edge_data
|
|
from graphify.paths import default_graph_json as _default_graph_json
|
|
|
|
try:
|
|
import jieba as _jieba # type: ignore[import-untyped]
|
|
except ImportError:
|
|
_jieba = None
|
|
|
|
|
|
def _load_graph(graph_path: str) -> nx.Graph:
|
|
try:
|
|
resolved = Path(graph_path).resolve()
|
|
if resolved.suffix != ".json":
|
|
raise ValueError(f"Graph path must be a .json file, got: {graph_path!r}")
|
|
if not resolved.exists():
|
|
raise FileNotFoundError(f"Graph file not found: {resolved}")
|
|
check_graph_file_size_cap(resolved)
|
|
safe = resolved
|
|
data = json.loads(safe.read_text(encoding="utf-8"))
|
|
if "links" not in data and "edges" in data:
|
|
data = dict(data, links=data["edges"])
|
|
data = {**data, "directed": True}
|
|
try:
|
|
from graphify.build import graph_has_legacy_ids as _legacy
|
|
if _legacy(data.get("nodes", [])):
|
|
print(
|
|
"[graphify] note: this graph uses the pre-#1504 node-ID scheme; "
|
|
"rebuild with `graphify extract --force` for path-qualified IDs.",
|
|
file=sys.stderr,
|
|
)
|
|
except Exception:
|
|
pass
|
|
try:
|
|
G = json_graph.node_link_graph(data, edges="links")
|
|
except TypeError:
|
|
G = json_graph.node_link_graph(data)
|
|
# Attach the work-memory overlay (derived sidecar next to graph.json) so
|
|
# the query/MCP read surface can annotate NODE lines display-only. Empty
|
|
# when no sidecar exists, leaving un-annotated output byte-identical.
|
|
try:
|
|
from graphify.reflect import load_learning_overlay as _llo
|
|
G.graph["_learning_overlay"] = _llo(resolved)
|
|
except Exception:
|
|
G.graph["_learning_overlay"] = {}
|
|
return G
|
|
except (ValueError, FileNotFoundError) as exc:
|
|
print(f"error: {exc}", file=sys.stderr)
|
|
sys.exit(1)
|
|
except json.JSONDecodeError as exc:
|
|
print(f"error: graph.json is corrupted ({exc}). Re-run /graphify to rebuild.", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
|
|
def _communities_from_graph(G: nx.Graph) -> dict[int, list[str]]:
|
|
"""Reconstruct community dict from community property stored on nodes."""
|
|
communities: dict[int, list[str]] = {}
|
|
for node_id, data in G.nodes(data=True):
|
|
cid = data.get("community")
|
|
if cid is not None:
|
|
communities.setdefault(int(cid), []).append(node_id)
|
|
return communities
|
|
|
|
|
|
def _strip_diacritics(text: str | None) -> str:
|
|
import unicodedata
|
|
if not isinstance(text, str):
|
|
text = "" if text is None else str(text)
|
|
nfkd = unicodedata.normalize("NFKD", text)
|
|
return "".join(c for c in nfkd if not unicodedata.combining(c))
|
|
|
|
|
|
def _search_tokens(text: str) -> list[str]:
|
|
"""Split text into word tokens, stripping punctuation and diacritics."""
|
|
return re.findall(r"\w+", _strip_diacritics(str(text)).lower())
|
|
|
|
|
|
def _has_chinese(text: str) -> bool:
|
|
return any("一" <= ch <= "鿿" for ch in text)
|
|
|
|
|
|
def _segment_chinese(text: str) -> list[str]:
|
|
"""Segment Chinese text and keep the original term for exact matching."""
|
|
if _jieba is not None:
|
|
segments = [w for w in _jieba.cut(text) if len(w.strip()) > 0]
|
|
else:
|
|
segments = [text[i:i + 2] for i in range(len(text) - 1)] or [text]
|
|
if len(text) > 1 and text not in segments:
|
|
segments.append(text)
|
|
return segments
|
|
|
|
|
|
def _is_searchable(term: str) -> bool:
|
|
"""True if term is Chinese, non-English, or an English word longer than 2 chars."""
|
|
if all("a" <= ch <= "z" for ch in term):
|
|
return len(term) > 2
|
|
return True
|
|
|
|
|
|
# English question/filler words dropped from query terms so content words drive
|
|
# BFS seeding. Without this, "how does the frontier cache work" seeds on "how"/
|
|
# "the"/"work" (which prefix-match prose labels like "Working Principles" at 100x)
|
|
# instead of "frontier"/"cache", and lands in the wrong part of the graph. Applied
|
|
# to query terms only — node text is never filtered, so a symbol literally named
|
|
# `work` stays findable via explain/path. `work`/`works`/`working` are included
|
|
# because "how does X work" / "how X works" is the most common question phrasing.
|
|
_QUERY_STOPWORDS = frozenset({
|
|
"how", "what", "why", "when", "where", "which", "who", "whom", "whose",
|
|
"does", "did", "is", "are", "was", "were", "be", "been", "being",
|
|
"can", "could", "should", "would", "will", "shall", "may", "might", "must",
|
|
"has", "have", "had", "the", "and", "but", "not", "for", "from", "with",
|
|
"without", "into", "onto", "off", "that", "this", "these", "those", "there",
|
|
"here", "its", "their", "them", "they", "about", "any", "all", "some",
|
|
"work", "works", "working",
|
|
})
|
|
|
|
|
|
def _query_terms(question: str) -> list[str]:
|
|
"""Split a query into searchable terms, segmenting Chinese text, then drop
|
|
English question/filler words (`_QUERY_STOPWORDS`) so content words drive
|
|
seeding. Falls back to the unfiltered terms if the query is all stopwords, so
|
|
a question like "how does it work" still seeds on something."""
|
|
terms: list[str] = []
|
|
for raw in question.split():
|
|
if _has_chinese(raw):
|
|
for seg in _segment_chinese(raw.lower().strip()):
|
|
seg = seg.strip()
|
|
if seg and _is_searchable(seg):
|
|
terms.append(seg)
|
|
else:
|
|
# Strip punctuation without touching Unicode characters (avoid NFKD mangling non-Latin scripts)
|
|
for tok in re.findall(r"\w+", raw.lower()):
|
|
if _is_searchable(tok):
|
|
terms.append(tok)
|
|
content = [t for t in terms if t not in _QUERY_STOPWORDS]
|
|
return content or terms
|
|
|
|
|
|
_EXACT_MATCH_BONUS = 1000.0
|
|
_PREFIX_MATCH_BONUS = 100.0
|
|
_SUBSTRING_MATCH_BONUS = 1.0
|
|
_SOURCE_MATCH_BONUS = 0.5
|
|
|
|
|
|
def _compute_idf(G: nx.Graph, terms: list[str]) -> dict[str, float]:
|
|
"""IDF weights for query terms, cached in G.graph['_idf_cache'].
|
|
|
|
Common terms like 'error' or 'exception' that match hundreds of nodes get
|
|
low weights; rare identifiers like 'FooBarService' get high weights.
|
|
Cache is stored on the graph object itself so it auto-invalidates when
|
|
a hot-reload replaces G with a new object.
|
|
"""
|
|
cache: dict[str, float] = G.graph.setdefault("_idf_cache", {})
|
|
N = G.number_of_nodes() or 1
|
|
uncached = [t for t in terms if t not in cache]
|
|
if uncached:
|
|
df: dict[str, int] = {t: 0 for t in uncached}
|
|
for _, data in G.nodes(data=True):
|
|
norm_label = (
|
|
data.get("norm_label") or _strip_diacritics(data.get("label") or "")
|
|
).lower()
|
|
for t in uncached:
|
|
if t in norm_label:
|
|
df[t] += 1
|
|
for t in uncached:
|
|
cache[t] = math.log(1 + N / (1 + df[t]))
|
|
return {t: cache.get(t, math.log(1 + N)) for t in terms}
|
|
|
|
|
|
def _trigrams(text: str) -> set[str]:
|
|
"""Character trigrams of `text`; for <3-char text the whole string is the key."""
|
|
if len(text) < 3:
|
|
return {text} if text else set()
|
|
return {text[i:i + 3] for i in range(len(text) - 2)}
|
|
|
|
|
|
def _node_search_text(data: dict, nid: str) -> str:
|
|
"""Concatenate every field _score_nodes / _find_node match a query against, so
|
|
one trigram index over this text is a complete candidate generator for both.
|
|
|
|
- `norm_label` and `source_file` feed _score_nodes' per-term substring tiers.
|
|
- `label_tokens` (the space-joined token form) feeds _find_node's
|
|
`term in label_tokens` branch, where a multi-word `term` can span a token
|
|
boundary that punctuation hides in `norm_label` (e.g. query "foo bar" matches
|
|
label "foo.bar" only via its tokenized form).
|
|
- `source_tokens` feeds _find_node's exact source-file path lookup, where a
|
|
query like "app/api/example/route.ts" tokenizes to "app api example route ts".
|
|
- `nid` feeds the whole-query `joined == nid_lower` tier.
|
|
|
|
NUL separators stop a trigram from spanning two fields (a query never contains
|
|
NUL, so a cross-field trigram can never be a real match).
|
|
"""
|
|
norm_label = data.get("norm_label") or _strip_diacritics(data.get("label") or "").lower()
|
|
label_tokens = " ".join(_search_tokens(data.get("label") or ""))
|
|
source = (data.get("source_file") or "").lower()
|
|
source_tokens = " ".join(_search_tokens(data.get("source_file") or ""))
|
|
return "\x00".join((norm_label, label_tokens, str(nid).lower(), source, source_tokens))
|
|
|
|
|
|
def _get_trigram_index(G: nx.Graph) -> dict:
|
|
"""Lazily build and cache a trigram -> node-position postings map on the graph.
|
|
|
|
Cached on `G.graph` so it auto-invalidates when a hot-reload swaps in a
|
|
fresh graph object, exactly like `_idf_cache`. `set_cache` memoizes per-trigram
|
|
id-sets across queries within one graph generation.
|
|
"""
|
|
idx = G.graph.get("_trigram_index")
|
|
if idx is not None:
|
|
return idx
|
|
ids = list(G.nodes())
|
|
postings: dict[str, array] = {}
|
|
for i, nid in enumerate(ids):
|
|
for g in _trigrams(_node_search_text(G.nodes[nid], nid)):
|
|
bucket = postings.get(g)
|
|
if bucket is None:
|
|
bucket = array("i")
|
|
postings[g] = bucket
|
|
bucket.append(i)
|
|
idx = {"ids": ids, "postings": postings, "set_cache": {}}
|
|
G.graph["_trigram_index"] = idx
|
|
return idx
|
|
|
|
|
|
def _trigram_candidates(G: nx.Graph, needles: list[str], *, guard_frac: float = 0.10) -> list[str] | None:
|
|
"""Node IDs whose text could contain any `needle` as a substring, via the
|
|
trigram index — a *superset* the caller then re-scores with the exact predicates.
|
|
|
|
Returns candidates in graph-iteration order (so order-sensitive callers like
|
|
_find_node stay byte-identical to a full scan), or **None** when the index isn't
|
|
worth it — a needle is too short to trigram, or its rarest trigram is still
|
|
common enough that the candidate set would approach the whole graph. The caller
|
|
falls back to the full scan, preserving the never-worse contract. The guard is
|
|
cheap: postings-length lookups only, no set intersection.
|
|
"""
|
|
idx = _get_trigram_index(G)
|
|
ids, postings, set_cache = idx["ids"], idx["postings"], idx["set_cache"]
|
|
n = len(ids)
|
|
if n == 0:
|
|
return []
|
|
needles = [s for s in needles if s]
|
|
thresh = int(n * guard_frac)
|
|
for s in needles:
|
|
tgs = _trigrams(s)
|
|
if not tgs or any(len(g) < 3 for g in tgs):
|
|
return None # too short to trigram-filter
|
|
present = [len(postings[g]) for g in tgs if g in postings]
|
|
if not present:
|
|
continue # this needle matches nothing — contributes no candidates
|
|
if min(present) > thresh:
|
|
return None # rarest trigram still too common -> not worth the index
|
|
cand: set[int] = set()
|
|
for s in needles:
|
|
sets: list[set] | None = []
|
|
for g in _trigrams(s):
|
|
bucket = postings.get(g)
|
|
if bucket is None:
|
|
sets = None # a trigram absent everywhere -> needle matches nothing
|
|
break
|
|
cached = set_cache.get(g)
|
|
if cached is None:
|
|
cached = set(bucket)
|
|
set_cache[g] = cached
|
|
sets.append(cached)
|
|
if not sets:
|
|
continue
|
|
sets.sort(key=len) # intersect smallest-first
|
|
hit = set(sets[0])
|
|
for other in sets[1:]:
|
|
hit &= other
|
|
if not hit:
|
|
break
|
|
cand |= hit
|
|
return [ids[i] for i in sorted(cand)]
|
|
|
|
|
|
def _score_nodes(G: nx.Graph, terms: list[str]) -> list[tuple[float, str]]:
|
|
scored = []
|
|
# Dedupe tokens, order-preserving (as _pick_seeds already does): a repeated
|
|
# query word must not double-count every tier, and with coverage scaling
|
|
# below it would also inflate the matched-term ratio (#1602).
|
|
norm_terms = list(dict.fromkeys(tok for t in terms for tok in _search_tokens(t)))
|
|
n_terms = len(norm_terms)
|
|
idf = _compute_idf(G, norm_terms)
|
|
# Whole-query string for full-label matching (mirrors _find_node's `term`).
|
|
joined = " ".join(norm_terms)
|
|
# Weight the full-query bonus by the rarest constituent term so a specific
|
|
# multi-word label still outweighs common-token noise; floor at 1.0.
|
|
joined_w = max((idf.get(t, 1.0) for t in norm_terms), default=1.0)
|
|
# Trigram prefilter: score only nodes whose text could match a term, falling
|
|
# back to the whole graph when the index isn't selective. The result is
|
|
# identical either way — the per-node scoring below is unchanged and a
|
|
# non-candidate node always scores 0. (IDF above stays a whole-graph statistic.)
|
|
candidate_ids = _trigram_candidates(G, norm_terms + ([joined] if joined else []))
|
|
node_iter = (
|
|
G.nodes(data=True) if candidate_ids is None
|
|
else ((nid, G.nodes[nid]) for nid in candidate_ids)
|
|
)
|
|
for nid, data in node_iter:
|
|
norm_label = data.get("norm_label") or _strip_diacritics(data.get("label") or "").lower()
|
|
bare_label = norm_label.rstrip("()")
|
|
# Tokenized form of the label (punctuation stripped, same transform as the
|
|
# query). norm_label may still carry punctuation like ':' or '-', which a
|
|
# tokenized query can never equal; comparing token-joined forms on both
|
|
# sides makes "uoce: dehumidifier driver" match query "uoce dehumidifier
|
|
# driver".
|
|
label_tokens = " ".join(_search_tokens(data.get("label") or ""))
|
|
source = (data.get("source_file") or "").lower()
|
|
score = 0.0
|
|
# Full-query tier: a multi-word query that equals (or prefixes) the whole
|
|
# label must dominate the per-token bag-of-words sums below, so `path`/
|
|
# `query` resolve the same node `explain` does (via _find_node). Without
|
|
# this, no single token equals a multi-word label, the per-token exact
|
|
# tier never fires, and every node sharing the token set ties -> arbitrary
|
|
# node-id sort -> wrong/disconnected endpoint -> false "No path found".
|
|
if joined:
|
|
nid_lower = nid.lower()
|
|
if joined in (norm_label, bare_label, label_tokens, nid_lower):
|
|
score += _EXACT_MATCH_BONUS * 10 * joined_w
|
|
elif (
|
|
norm_label.startswith(joined)
|
|
or bare_label.startswith(joined)
|
|
or label_tokens.startswith(joined)
|
|
):
|
|
score += _PREFIX_MATCH_BONUS * 10 * joined_w
|
|
# Term coverage (#1602): scale the per-term exact/prefix tiers by the
|
|
# squared fraction of query terms the node's LABEL matches, so a lone
|
|
# generic word that happens to equal a short label (query term "home"
|
|
# vs. a home() leaf) cannot bury nodes that match several of the
|
|
# query's terms. Squaring matters because the exact tier is 10x the
|
|
# prefix tier: at linear coverage a 1-of-10-terms exact match still
|
|
# outscores a 3-of-10 prefix+substring match. Single-term and
|
|
# full-coverage queries are unchanged (coverage == 1), so identifier
|
|
# lookups keep exact-match dominance. Source-file hits score but do
|
|
# not count as coverage: a colliding leaf whose directory shares
|
|
# tokens with the query (common near the intended target) must not
|
|
# win back its exact tier via path fragments. The substring/source
|
|
# bonuses and the full-query tier above stay unscaled.
|
|
matched = 0
|
|
tiered = 0.0
|
|
for t in norm_terms:
|
|
w = idf.get(t, 1.0)
|
|
# Three-tier precedence: exact > prefix > substring (take the
|
|
# strongest tier per term so a single term cannot double-count).
|
|
if t == norm_label or t == bare_label:
|
|
tiered += _EXACT_MATCH_BONUS * w
|
|
matched += 1
|
|
elif norm_label.startswith(t) or bare_label.startswith(t):
|
|
tiered += _PREFIX_MATCH_BONUS * w
|
|
matched += 1
|
|
elif t in norm_label:
|
|
score += _SUBSTRING_MATCH_BONUS * w
|
|
matched += 1
|
|
if t in source:
|
|
score += _SOURCE_MATCH_BONUS * w
|
|
if tiered:
|
|
score += tiered * (matched / n_terms) ** 2
|
|
if score > 0:
|
|
scored.append((score, nid))
|
|
# Sort by score desc; break ties toward the shorter label so a concise exact
|
|
# match beats a longer superset that happens to share the same score.
|
|
scored.sort(key=lambda s: (-s[0], len(G.nodes[s[1]].get("label") or s[1]), s[1]))
|
|
return scored
|
|
|
|
|
|
def _pick_scored_endpoint(G: nx.Graph, scored: list[tuple[float, str]], query: str) -> str:
|
|
"""Pick a path endpoint from a _score_nodes result, preferring full-token matches.
|
|
|
|
The full-query tier in _score_nodes only fires when the query equals or
|
|
prefixes a label, so a query that is a token *subset* of the intended label
|
|
(query "Reject-everything judge" vs. label "Degenerate Reject-Everything
|
|
Judge") gets no bonus, and a node prefix-matching one rare token (label
|
|
"Rejection Summary") can out-score it on IDF alone. Committing to scored[0]
|
|
then anchors the path on an unrelated — often disconnected — node and yields
|
|
a false "No path found". Scan the score-ordered list and take the first
|
|
candidate whose label contains EVERY query token; when the top candidate
|
|
already full-matches, or no candidate does, this is exactly scored[0].
|
|
|
|
`scored` must be non-empty (both callers return early on no match).
|
|
"""
|
|
qtokens = set(_search_tokens(query))
|
|
if not qtokens:
|
|
return scored[0][1]
|
|
for _score, nid in scored:
|
|
if qtokens <= set(_search_tokens(G.nodes[nid].get("label") or nid)):
|
|
return nid
|
|
return scored[0][1]
|
|
|
|
|
|
def _pick_seeds(
|
|
scored: list[tuple[float, str]],
|
|
max_k: int = 3,
|
|
gap_ratio: float = 0.2,
|
|
*,
|
|
G: "nx.Graph | None" = None,
|
|
terms: list[str] | None = None,
|
|
) -> list[str]:
|
|
"""Select BFS seed nodes, stopping when score drops too far below the top.
|
|
|
|
Prevents high-frequency noise terms (error, exception) from stealing seed
|
|
slots from a dominant identifier match. When FooBarService scores 1000 and
|
|
error nodes score 1.0, only FooBarService is seeded — the score gap is 99.9%
|
|
which is well above the 20% threshold that would allow additional seeds.
|
|
|
|
That same gap_ratio cutoff has a failure mode on multi-term natural-language
|
|
queries: if one term happens to hit an EXACT label match on a node that is
|
|
otherwise unrelated to the query's intent (e.g. a common word that is also
|
|
used as an unrelated identifier or field name elsewhere in the corpus), it
|
|
can outscore every SUBSTRING match on the query's other, actually-relevant
|
|
terms by ~1000x (see `_EXACT_MATCH_BONUS` vs. `_SUBSTRING_MATCH_BONUS`).
|
|
The 20%-gap cutoff then silently discards all of those substring-tier
|
|
seeds, so the BFS traversal only ever explores the neighborhood of the one
|
|
unrelated exact match — see #1445.
|
|
|
|
When `G` and `terms` are supplied, this guarantees at least one seed per
|
|
distinct query term that has any match at all, so one term's incidental
|
|
collision cannot starve out the others. Ties within a term are broken by
|
|
graph degree (structural centrality), so an isolated incidental match
|
|
doesn't out-rank a real, well-connected hub for that term.
|
|
|
|
Coverage scaling in _score_nodes (#1602) now dampens a lone collision's
|
|
exact tier on multi-term queries, which brings label-matching relevant
|
|
nodes back inside the gap window; this per-term guarantee remains
|
|
load-bearing for relevant nodes matched only via substrings, whose flat
|
|
scores a dampened collision can still exceed.
|
|
"""
|
|
if not scored:
|
|
return []
|
|
top_score = scored[0][0]
|
|
seeds = []
|
|
for score, nid in scored[:max_k]:
|
|
if seeds and score < top_score * gap_ratio:
|
|
break
|
|
seeds.append(nid)
|
|
|
|
if G is not None and terms:
|
|
norm_terms = sorted({tok for t in terms for tok in _search_tokens(t)})
|
|
for term in norm_terms:
|
|
term_scored = _score_nodes(G, [term])
|
|
if not term_scored:
|
|
continue
|
|
best_score = term_scored[0][0]
|
|
tied = [nid for s, nid in term_scored if s == best_score]
|
|
best_nid = max(tied, key=lambda n: G.degree(n)) if len(tied) > 1 else term_scored[0][1]
|
|
if best_nid not in seeds:
|
|
seeds.append(best_nid)
|
|
return seeds
|
|
|
|
|
|
_CONTEXT_HINTS: tuple[tuple[str, tuple[str, ...]], ...] = (
|
|
("call", ("call", "calls", "called", "invoke", "invokes", "invoked")),
|
|
("import", ("import", "imports", "imported", "module", "modules")),
|
|
("field", ("field", "fields", "member", "members", "property", "properties")),
|
|
("parameter_type", ("parameter", "parameters", "param", "params", "argument", "arguments")),
|
|
("return_type", ("return", "returns", "returned")),
|
|
("generic_arg", ("generic", "generics", "template", "templates")),
|
|
)
|
|
|
|
|
|
_CONTEXT_FILTER_ALIASES: dict[str, str] = {
|
|
"param": "parameter_type",
|
|
"params": "parameter_type",
|
|
"parameter": "parameter_type",
|
|
"parameters": "parameter_type",
|
|
"argument": "parameter_type",
|
|
"arguments": "parameter_type",
|
|
"arg": "parameter_type",
|
|
"args": "parameter_type",
|
|
"return": "return_type",
|
|
"returns": "return_type",
|
|
"returned": "return_type",
|
|
"generic": "generic_arg",
|
|
"generics": "generic_arg",
|
|
"template": "generic_arg",
|
|
"templates": "generic_arg",
|
|
"annotation": "attribute",
|
|
"annotations": "attribute",
|
|
"decorator": "attribute",
|
|
"decorators": "attribute",
|
|
"calls": "call",
|
|
"called": "call",
|
|
"invoke": "call",
|
|
"invocation": "call",
|
|
"fields": "field",
|
|
"property": "field",
|
|
"properties": "field",
|
|
"member": "field",
|
|
"members": "field",
|
|
"imports": "import",
|
|
"imported": "import",
|
|
"module": "import",
|
|
"modules": "import",
|
|
"exports": "export",
|
|
"exported": "export",
|
|
}
|
|
|
|
|
|
def _normalize_context_filters(filters: list[str] | None) -> list[str]:
|
|
if not filters:
|
|
return []
|
|
normalized: list[str] = []
|
|
seen: set[str] = set()
|
|
for value in filters:
|
|
key = _strip_diacritics(str(value)).strip().lower()
|
|
if not key:
|
|
continue
|
|
key = _CONTEXT_FILTER_ALIASES.get(key, key)
|
|
if key not in seen:
|
|
seen.add(key)
|
|
normalized.append(key)
|
|
return normalized
|
|
|
|
|
|
def _infer_context_filters(question: str) -> list[str]:
|
|
lowered = {
|
|
_strip_diacritics(token).lower()
|
|
for token in question.replace("?", " ").replace(",", " ").split()
|
|
}
|
|
inferred: list[str] = []
|
|
for context, hints in _CONTEXT_HINTS:
|
|
if any(hint in lowered for hint in hints):
|
|
inferred.append(context)
|
|
return inferred
|
|
|
|
|
|
def _resolve_context_filters(question: str, explicit_filters: list[str] | None = None) -> tuple[list[str], str | None]:
|
|
normalized = _normalize_context_filters(explicit_filters)
|
|
if normalized:
|
|
return normalized, "explicit"
|
|
inferred = _infer_context_filters(question)
|
|
if inferred:
|
|
return inferred, "heuristic"
|
|
return [], None
|
|
|
|
|
|
def _filter_graph_by_context(G: nx.Graph, context_filters: list[str] | None) -> nx.Graph:
|
|
filters = set(_normalize_context_filters(context_filters))
|
|
if not filters:
|
|
return G
|
|
H = G.__class__()
|
|
H.add_nodes_from(G.nodes(data=True))
|
|
if isinstance(G, (nx.MultiGraph, nx.MultiDiGraph)):
|
|
for u, v, key, data in G.edges(keys=True, data=True):
|
|
if data.get("context") in filters:
|
|
H.add_edge(u, v, key=key, **data)
|
|
else:
|
|
for u, v, data in G.edges(data=True):
|
|
if data.get("context") in filters:
|
|
H.add_edge(u, v, **data)
|
|
return H
|
|
|
|
|
|
def _bfs(G: nx.Graph, start_nodes: list[str], depth: int) -> tuple[set[str], list[tuple]]:
|
|
# Compute hub threshold: nodes above this degree are not expanded as transit.
|
|
# p99 of degree distribution, floored at 50 to avoid over-blocking small graphs.
|
|
degrees = [G.degree(n) for n in G.nodes()]
|
|
if degrees:
|
|
degrees_sorted = sorted(degrees)
|
|
p99_idx = int(len(degrees_sorted) * 0.99)
|
|
hub_threshold = max(50, degrees_sorted[p99_idx])
|
|
else:
|
|
hub_threshold = 50
|
|
seed_set = set(start_nodes)
|
|
visited: set[str] = set(start_nodes)
|
|
frontier = set(start_nodes)
|
|
edges_seen: list[tuple] = []
|
|
for _ in range(depth):
|
|
next_frontier: set[str] = set()
|
|
for n in frontier:
|
|
# Don't expand through high-degree hubs (except seeds - a hub that
|
|
# is the starting node should still be explored).
|
|
if n not in seed_set and G.degree(n) >= hub_threshold:
|
|
continue
|
|
for neighbor in G.neighbors(n):
|
|
if neighbor not in visited:
|
|
next_frontier.add(neighbor)
|
|
edges_seen.append((n, neighbor))
|
|
visited.update(next_frontier)
|
|
frontier = next_frontier
|
|
return visited, edges_seen
|
|
|
|
|
|
def _dfs(G: nx.Graph, start_nodes: list[str], depth: int) -> tuple[set[str], list[tuple]]:
|
|
degrees = [G.degree(n) for n in G.nodes()]
|
|
if degrees:
|
|
degrees_sorted = sorted(degrees)
|
|
p99_idx = int(len(degrees_sorted) * 0.99)
|
|
hub_threshold = max(50, degrees_sorted[p99_idx])
|
|
else:
|
|
hub_threshold = 50
|
|
seed_set = set(start_nodes)
|
|
visited: set[str] = set()
|
|
edges_seen: list[tuple] = []
|
|
stack = [(n, 0) for n in reversed(start_nodes)]
|
|
while stack:
|
|
node, d = stack.pop()
|
|
if node in visited or d > depth:
|
|
continue
|
|
visited.add(node)
|
|
if node not in seed_set and G.degree(node) >= hub_threshold:
|
|
continue
|
|
for neighbor in G.neighbors(node):
|
|
if neighbor not in visited:
|
|
stack.append((neighbor, d + 1))
|
|
edges_seen.append((node, neighbor))
|
|
return visited, edges_seen
|
|
|
|
|
|
def _subgraph_to_text(G: nx.Graph, nodes: set[str], edges: list[tuple], token_budget: int = 2000, *, seeds: list[str] | None = None) -> str:
|
|
"""Render subgraph as text, cutting at token_budget (approx 3 chars/token).
|
|
|
|
seeds: exact-match nodes rendered first before the degree-sorted expansion,
|
|
so the queried symbol always appears at the top of the output.
|
|
"""
|
|
char_budget = token_budget * 3
|
|
lines = []
|
|
# Work-memory overlay (derived sidecar) stashed on the graph at load time.
|
|
# Empty when no sidecar exists, so un-annotated output stays byte-identical.
|
|
overlay = getattr(G, "graph", {}).get("_learning_overlay", {}) or {}
|
|
seed_set = set(seeds or [])
|
|
ordered = [n for n in (seeds or []) if n in nodes] + \
|
|
sorted(nodes - seed_set, key=lambda n: G.degree(n), reverse=True)
|
|
for nid in ordered:
|
|
d = G.nodes[nid]
|
|
# Every LLM-derived field passes through sanitize_label before being
|
|
# concatenated into MCP tool output (F-010): an attacker who controls a
|
|
# corpus document can otherwise inject ANSI escapes, fake graphify-out
|
|
# log lines, or prompt-injection markup into the model's context via
|
|
# source_file / source_location / community.
|
|
# The learning= suffix is appended INSIDE the bracket and BEFORE the
|
|
# budget check below, so it counts in char_budget accounting.
|
|
entry = overlay.get(str(nid))
|
|
learning_suffix = ""
|
|
if entry:
|
|
status = sanitize_label(str(entry.get("status", "")))
|
|
if status:
|
|
learning_suffix = f" learning={status}{':stale' if entry.get('stale') else ''}"
|
|
line = (
|
|
f"NODE {sanitize_label(d.get('label', nid))} "
|
|
f"[src={sanitize_label(str(d.get('source_file', '')))} "
|
|
f"loc={sanitize_label(str(d.get('source_location', '')))} "
|
|
f"community={sanitize_label(str(d.get('community_name') or d.get('community', '')))}"
|
|
f"{learning_suffix}]"
|
|
)
|
|
lines.append(line)
|
|
for u, v in edges:
|
|
if u in nodes and v in nodes:
|
|
raw = G[u][v]
|
|
d = next(iter(raw.values()), {}) if isinstance(G, (nx.MultiGraph, nx.MultiDiGraph)) else raw
|
|
context = d.get("context")
|
|
context_suffix = f" context={sanitize_label(str(context))}" if context else ""
|
|
line = (
|
|
f"EDGE {sanitize_label(G.nodes[u].get('label', u))} "
|
|
f"--{sanitize_label(str(d.get('relation', '')))} "
|
|
f"[{sanitize_label(str(d.get('confidence', '')))}{context_suffix}]--> "
|
|
f"{sanitize_label(G.nodes[v].get('label', v))}"
|
|
)
|
|
lines.append(line)
|
|
output = "\n".join(lines)
|
|
if len(output) > char_budget:
|
|
cut_at = output[:char_budget].rfind("\n")
|
|
cut_at = cut_at if cut_at > 0 else char_budget
|
|
total_nodes = sum(1 for l in lines if l.startswith("NODE "))
|
|
shown_nodes = output[:cut_at].count("\nNODE ") + (1 if output.startswith("NODE ") else 0)
|
|
cut_count = total_nodes - shown_nodes
|
|
output = (
|
|
output[:cut_at]
|
|
+ f"\n... (truncated — {cut_count} more nodes cut by ~{token_budget}-token budget."
|
|
f" Narrow with context_filter=['call'] or use get_node for a specific symbol)"
|
|
)
|
|
return output
|
|
|
|
|
|
def _query_graph_text(
|
|
G: nx.Graph,
|
|
question: str,
|
|
*,
|
|
mode: str = "bfs",
|
|
depth: int = 3,
|
|
token_budget: int = 2000,
|
|
context_filters: list[str] | None = None,
|
|
) -> str:
|
|
terms = _query_terms(question)
|
|
scored = _score_nodes(G, terms)
|
|
start_nodes = _pick_seeds(scored, G=G, terms=terms)
|
|
if not start_nodes:
|
|
return "No matching nodes found."
|
|
resolved_filters, filter_source = _resolve_context_filters(question, context_filters)
|
|
traversal_graph = _filter_graph_by_context(G, resolved_filters)
|
|
nodes, edges = _dfs(traversal_graph, start_nodes, depth) if mode == "dfs" else _bfs(traversal_graph, start_nodes, depth)
|
|
header_parts = [
|
|
f"Traversal: {mode.upper()} depth={depth}",
|
|
f"Start: {[G.nodes[n].get('label', n) for n in start_nodes]}",
|
|
]
|
|
if resolved_filters:
|
|
header_parts.append(f"Context: {', '.join(resolved_filters)} ({filter_source})")
|
|
header_parts.append(f"{len(nodes)} nodes found")
|
|
header = " | ".join(header_parts) + "\n\n"
|
|
return header + _subgraph_to_text(traversal_graph, nodes, edges, token_budget)
|
|
|
|
|
|
def _find_node(G: nx.Graph, label: str) -> list[str]:
|
|
"""Return node IDs whose label or ID matches the search term (diacritic-insensitive).
|
|
|
|
Results are ordered by precedence: exact source-file path match first, then
|
|
exact (label/ID) match, then prefix match, then substring match. Node-ID exact
|
|
matches are grouped with label exact matches.
|
|
"""
|
|
term = " ".join(_search_tokens(label))
|
|
if not term:
|
|
return []
|
|
# Punctuation-preserving normalized query. `term` tokenizes on \w+ (so
|
|
# "blockStream.ts" -> "blockstream ts", space where the '.' was), but a node's
|
|
# stored `norm_label` keeps punctuation ("blockstream.ts"). Matching only via
|
|
# `term`/`label_tokens` works when the node label tokenizes the same way, but is
|
|
# fragile if `label` and `norm_label` diverge. `norm_query` matches `norm_label`
|
|
# symmetrically so an exactly-typed punctuated label always resolves (#1704).
|
|
norm_query = _strip_diacritics(str(label)).lower().strip()
|
|
source_exact: list[str] = []
|
|
exact: list[str] = []
|
|
prefix: list[str] = []
|
|
substring: list[str] = []
|
|
# Trigram prefilter (graph-iteration order preserved so exact/prefix/substring
|
|
# ordering — and thus matches[0] — is byte-identical to the full scan).
|
|
candidate_ids = _trigram_candidates(G, [term, norm_query])
|
|
node_iter = (
|
|
G.nodes(data=True) if candidate_ids is None
|
|
else ((nid, G.nodes[nid]) for nid in candidate_ids)
|
|
)
|
|
for nid, d in node_iter:
|
|
norm_label = d.get("norm_label") or _strip_diacritics(d.get("label") or "").lower()
|
|
bare_label = norm_label.rstrip("()")
|
|
label_tokens = " ".join(_search_tokens(d.get("label") or ""))
|
|
source_tokens = " ".join(_search_tokens(d.get("source_file") or ""))
|
|
nid_lower = nid.lower()
|
|
if term == source_tokens:
|
|
source_exact.append(nid)
|
|
elif (
|
|
term == norm_label or term == bare_label or term == label_tokens or term == nid_lower
|
|
or norm_query == norm_label or norm_query == bare_label
|
|
):
|
|
exact.append(nid)
|
|
elif (
|
|
norm_label.startswith(term)
|
|
or bare_label.startswith(term)
|
|
or label_tokens.startswith(term)
|
|
or nid_lower.startswith(term)
|
|
or norm_label.startswith(norm_query)
|
|
or bare_label.startswith(norm_query)
|
|
):
|
|
prefix.append(nid)
|
|
elif term in norm_label or term in label_tokens or norm_query in norm_label:
|
|
substring.append(nid)
|
|
|
|
if source_exact:
|
|
query_basename = _strip_diacritics(Path(label).name).lower()
|
|
preferred = [
|
|
nid
|
|
for nid in source_exact
|
|
if str(G.nodes[nid].get("source_location", "")) == "L1"
|
|
and _strip_diacritics(str(G.nodes[nid].get("label") or "")).lower()
|
|
== query_basename
|
|
]
|
|
if len(preferred) == 1:
|
|
source_exact = preferred + [nid for nid in source_exact if nid != preferred[0]]
|
|
|
|
return source_exact + exact + prefix + substring
|
|
|
|
|
|
def _filter_blank_stdin() -> None:
|
|
"""Filter blank lines from stdin before MCP reads it.
|
|
|
|
Some MCP clients (Claude Desktop, etc.) send blank lines between JSON
|
|
messages. The MCP stdio transport tries to parse every line as a
|
|
JSONRPCMessage, so a bare newline triggers a Pydantic ValidationError.
|
|
This installs an OS-level pipe that relays stdin while dropping blanks.
|
|
"""
|
|
import os
|
|
import threading
|
|
|
|
r_fd, w_fd = os.pipe()
|
|
saved_fd = os.dup(sys.stdin.fileno())
|
|
|
|
def _relay() -> None:
|
|
try:
|
|
with open(saved_fd, "rb") as src, open(w_fd, "wb") as dst:
|
|
for line in src:
|
|
if line.strip():
|
|
dst.write(line)
|
|
dst.flush()
|
|
except Exception:
|
|
pass
|
|
|
|
threading.Thread(target=_relay, daemon=True).start()
|
|
os.dup2(r_fd, sys.stdin.fileno())
|
|
os.close(r_fd)
|
|
sys.stdin = open(0, "r", closefd=False)
|
|
|
|
|
|
def _community_header(cid: int, community_name) -> str:
|
|
# Header for get_community: "Community N — Name", matching get_node / query
|
|
# output which read the community_name attribute to_json writes onto nodes.
|
|
# Skip the name when it is just the "Community N" placeholder (written for
|
|
# unnamed communities) so the header never reads "Community 12 — Community 12";
|
|
# also falls back to the bare id when there is no name. Name is sanitised
|
|
# (F-010) like every other LLM-derived field.
|
|
base = f"Community {cid}"
|
|
if community_name:
|
|
clean = sanitize_label(str(community_name))
|
|
if clean and clean != base:
|
|
return f"{base} — {clean}"
|
|
return base
|
|
|
|
|
|
def _build_server(graph_path: str):
|
|
"""Build the configured low-level MCP Server (shared by every transport).
|
|
|
|
All graph query tools and resources are registered here over a single
|
|
``mcp.server.Server`` instance; the caller picks the transport (stdio or
|
|
Streamable HTTP) and runs it. Hot-reload of graph.json works the same way
|
|
regardless of transport, since reloads happen inside the tool handlers.
|
|
"""
|
|
import threading
|
|
|
|
try:
|
|
from mcp.server import Server
|
|
from mcp import types
|
|
from mcp.types import AnyUrl
|
|
except ImportError as e:
|
|
raise ImportError('mcp not installed. Run: pip install "graphifyy[mcp]"') from e
|
|
|
|
from graphify import paths as _paths
|
|
|
|
# Per-graph context cache: resolved graph.json path -> {key, G, communities}.
|
|
# The server's default graph is just the first entry; a tool call carrying a
|
|
# project_path adds its own. Routing every graph through one cache means the
|
|
# eager trigram index and the mtime+size hot-reload behave identically for
|
|
# the default graph and for any project graph.
|
|
_default_graph_path = graph_path
|
|
_ctx_lock = threading.Lock()
|
|
_ctx_cache: dict[str, dict] = {}
|
|
|
|
def _load_ctx(path: str):
|
|
"""Return (G, communities) for a graph.json path, reusing a cached
|
|
context until the file's (mtime, size) changes and then transparently
|
|
rebuilding it. Unlike ``_load_graph`` it never exits the process on a
|
|
missing/corrupt file — it raises, so a bad project_path surfaces as a
|
|
tool error instead of killing a server that is happily serving other
|
|
projects."""
|
|
try:
|
|
s = Path(path).stat()
|
|
key = (s.st_mtime_ns, s.st_size)
|
|
except FileNotFoundError:
|
|
raise FileNotFoundError(f"graph.json not found: {path}")
|
|
ent = _ctx_cache.get(path)
|
|
if ent is not None and ent["key"] == key:
|
|
return ent["G"], ent["communities"]
|
|
with _ctx_lock:
|
|
ent = _ctx_cache.get(path)
|
|
if ent is not None and ent["key"] == key:
|
|
return ent["G"], ent["communities"] # another thread built it
|
|
try:
|
|
new_G = _load_graph(path)
|
|
except SystemExit as e: # _load_graph exits on missing/corrupt file
|
|
raise RuntimeError(f"could not load graph.json at {path}") from e
|
|
# Warm the trigram index before exposing the graph so the first query
|
|
# against it is fast (same rationale as the original startup warm-up).
|
|
_get_trigram_index(new_G)
|
|
comm = _communities_from_graph(new_G)
|
|
_ctx_cache[path] = {"key": key, "G": new_G, "communities": comm}
|
|
return new_G, comm
|
|
|
|
def _resolve_graph_path(project_path) -> str:
|
|
"""Map an optional project_path to a concrete graph.json path. ``None``
|
|
keeps the server's default graph (backward-compatible); a project_path
|
|
resolves to ``<project_path>/<GRAPHIFY_OUT>/graph.json``, honouring the
|
|
GRAPHIFY_OUT override so worktree/shared-output setups keep working."""
|
|
if not project_path:
|
|
return _default_graph_path
|
|
return str(Path(project_path) / _paths.GRAPHIFY_OUT / "graph.json")
|
|
|
|
# Active per-request context, rebound by _select_graph() and read by the tool
|
|
# handlers below. No lock needed on the hot path: _select_graph and the
|
|
# handler run in one synchronous stretch of each call_tool coroutine (no
|
|
# await between them), so a concurrent call never observes a half-applied
|
|
# swap.
|
|
active_graph_path = _default_graph_path
|
|
try:
|
|
G, communities = _load_ctx(_default_graph_path)
|
|
except (FileNotFoundError, RuntimeError):
|
|
# No default graph at startup → run as a pure multi-project server. Tools
|
|
# then require project_path; a call without one gets a clear error rather
|
|
# than the process refusing to start (which is what _load_graph would do).
|
|
G, communities = None, {}
|
|
|
|
def _select_graph(project_path) -> None:
|
|
nonlocal G, communities, active_graph_path
|
|
path = _resolve_graph_path(project_path)
|
|
G, communities = _load_ctx(path)
|
|
active_graph_path = path
|
|
|
|
server = Server("graphify")
|
|
|
|
@server.list_tools()
|
|
async def list_tools() -> list[types.Tool]:
|
|
_tools = [
|
|
types.Tool(
|
|
name="query_graph",
|
|
description="Search the knowledge graph using BFS or DFS. Returns relevant nodes and edges as text context.",
|
|
inputSchema={
|
|
"type": "object",
|
|
"properties": {
|
|
"question": {"type": "string", "description": "Natural language question or keyword search"},
|
|
"mode": {"type": "string", "enum": ["bfs", "dfs"], "default": "bfs",
|
|
"description": "bfs=broad context, dfs=trace a specific path"},
|
|
"depth": {"type": "integer", "default": 3, "description": "Traversal depth (1-6)"},
|
|
"token_budget": {"type": "integer", "default": 2000, "description": "Max output tokens"},
|
|
"context_filter": {
|
|
"type": "array",
|
|
"items": {"type": "string"},
|
|
"description": "Optional explicit edge-context filter, e.g. ['call', 'field']",
|
|
},
|
|
},
|
|
"required": ["question"],
|
|
},
|
|
),
|
|
types.Tool(
|
|
name="get_node",
|
|
description="Get full details for a specific node by label or ID.",
|
|
inputSchema={
|
|
"type": "object",
|
|
"properties": {"label": {"type": "string", "description": "Node label or ID to look up"}},
|
|
"required": ["label"],
|
|
},
|
|
),
|
|
types.Tool(
|
|
name="get_neighbors",
|
|
description="Get all direct neighbors of a node with edge details.",
|
|
inputSchema={
|
|
"type": "object",
|
|
"properties": {
|
|
"label": {"type": "string"},
|
|
"relation_filter": {"type": "string", "description": "Optional: filter by relation type"},
|
|
},
|
|
"required": ["label"],
|
|
},
|
|
),
|
|
types.Tool(
|
|
name="get_community",
|
|
description="Get all nodes in a community by community ID.",
|
|
inputSchema={
|
|
"type": "object",
|
|
"properties": {"community_id": {"type": "integer", "description": "Community ID (0-indexed by size)"}},
|
|
"required": ["community_id"],
|
|
},
|
|
),
|
|
types.Tool(
|
|
name="god_nodes",
|
|
description="Return the most connected nodes - the core abstractions of the knowledge graph.",
|
|
inputSchema={"type": "object", "properties": {"top_n": {"type": "integer", "default": 10}}},
|
|
),
|
|
types.Tool(
|
|
name="graph_stats",
|
|
description="Return summary statistics: node count, edge count, communities, confidence breakdown.",
|
|
inputSchema={"type": "object", "properties": {}},
|
|
),
|
|
types.Tool(
|
|
name="shortest_path",
|
|
description="Find the shortest path between two concepts in the knowledge graph.",
|
|
inputSchema={
|
|
"type": "object",
|
|
"properties": {
|
|
"source": {"type": "string", "description": "Source concept label or keyword"},
|
|
"target": {"type": "string", "description": "Target concept label or keyword"},
|
|
"max_hops": {"type": "integer", "default": 8, "description": "Maximum hops to consider"},
|
|
},
|
|
"required": ["source", "target"],
|
|
},
|
|
),
|
|
types.Tool(
|
|
name="list_prs",
|
|
description=(
|
|
"List open GitHub PRs with CI status, review state, and graph impact "
|
|
"(which communities each PR touches, blast radius). Use this before starting "
|
|
"work to check if a PR already covers the area you're about to change."
|
|
),
|
|
inputSchema={
|
|
"type": "object",
|
|
"properties": {
|
|
"base": {"type": "string", "description": "Base branch to filter PRs by (auto-detected if omitted)"},
|
|
"repo": {"type": "string", "description": "GitHub repo (owner/repo). Defaults to current repo."},
|
|
},
|
|
},
|
|
),
|
|
types.Tool(
|
|
name="get_pr_impact",
|
|
description=(
|
|
"Get detailed graph impact for a specific PR: which files it changes, "
|
|
"which knowledge-graph communities are affected, and how many nodes are touched. "
|
|
"Use this to assess merge risk or check for overlap with your current work."
|
|
),
|
|
inputSchema={
|
|
"type": "object",
|
|
"properties": {
|
|
"pr_number": {"type": "integer", "description": "PR number to analyse"},
|
|
"repo": {"type": "string", "description": "GitHub repo (owner/repo). Defaults to current repo."},
|
|
},
|
|
"required": ["pr_number"],
|
|
},
|
|
),
|
|
types.Tool(
|
|
name="triage_prs",
|
|
description=(
|
|
"Return all actionable open PRs (correct base, not stale) with full graph impact data "
|
|
"so you can reason about review priority, merge order, and conflict risk. "
|
|
"Call this when the user asks 'what PRs should I review?' or 'what's ready to merge?'"
|
|
),
|
|
inputSchema={
|
|
"type": "object",
|
|
"properties": {
|
|
"base": {"type": "string", "description": "Base branch to filter PRs by (auto-detected if omitted)"},
|
|
"repo": {"type": "string", "description": "GitHub repo (owner/repo). Defaults to current repo."},
|
|
},
|
|
},
|
|
),
|
|
]
|
|
# Multi-project support: every tool accepts an optional project_path.
|
|
# Injected here (rather than repeated in 11 literal schemas) so the set
|
|
# stays in lockstep as tools are added. Omitting it keeps the historical
|
|
# single-graph behaviour, so this is purely additive for existing callers.
|
|
for _t in _tools:
|
|
_t.inputSchema.setdefault("properties", {})["project_path"] = {
|
|
"type": "string",
|
|
"description": (
|
|
"Absolute path to a project directory containing "
|
|
"graphify-out/graph.json. Optional — defaults to the graph "
|
|
"this server was started with."
|
|
),
|
|
}
|
|
return _tools
|
|
|
|
def _tool_query_graph(arguments: dict) -> str:
|
|
import time as _time
|
|
from graphify import querylog
|
|
question = arguments["question"]
|
|
mode = arguments.get("mode", "bfs")
|
|
depth = min(int(arguments.get("depth", 3)), 6)
|
|
budget = int(arguments.get("token_budget", 2000))
|
|
context_filter = arguments.get("context_filter")
|
|
_t0 = _time.perf_counter()
|
|
result = _query_graph_text(
|
|
G,
|
|
question,
|
|
mode=mode,
|
|
depth=depth,
|
|
token_budget=budget,
|
|
context_filters=context_filter,
|
|
)
|
|
querylog.log_query(
|
|
kind="mcp_query",
|
|
question=question,
|
|
corpus=str(active_graph_path),
|
|
result=result,
|
|
mode=mode,
|
|
depth=depth,
|
|
token_budget=budget,
|
|
duration_ms=(_time.perf_counter() - _t0) * 1000,
|
|
)
|
|
return result
|
|
|
|
def _tool_get_node(arguments: dict) -> str:
|
|
label = arguments["label"].lower()
|
|
matches = [(nid, d) for nid, d in G.nodes(data=True)
|
|
if label in (d.get("label") or "").lower() or label == nid.lower()]
|
|
if not matches:
|
|
return f"No node matching '{label}' found."
|
|
nid, d = matches[0]
|
|
# Sanitise every LLM-derived field before concatenation (F-010).
|
|
return "\n".join([
|
|
f"Node: {sanitize_label(d.get('label', nid))}",
|
|
f" ID: {sanitize_label(nid)}",
|
|
f" Source: {sanitize_label(str(d.get('source_file', '')))} {sanitize_label(str(d.get('source_location', '')))}",
|
|
f" Type: {sanitize_label(str(d.get('file_type', '')))}",
|
|
f" Community: {sanitize_label(str(d.get('community_name') or d.get('community', '')))}",
|
|
f" Degree: {G.degree(nid)}",
|
|
])
|
|
|
|
def _tool_get_neighbors(arguments: dict) -> str:
|
|
label = arguments["label"].lower()
|
|
rel_filter = arguments.get("relation_filter", "").lower()
|
|
matches = _find_node(G, label)
|
|
if not matches:
|
|
return f"No node matching '{label}' found."
|
|
nid = matches[0]
|
|
lines = [f"Neighbors of {sanitize_label(G.nodes[nid].get('label', nid))}:"]
|
|
for nb in G.successors(nid):
|
|
d = edge_data(G, nid, nb)
|
|
rel = d.get("relation", "")
|
|
if rel_filter and rel_filter not in rel.lower():
|
|
continue
|
|
lines.append(
|
|
f" --> {sanitize_label(G.nodes[nb].get('label', nb))} "
|
|
f"[{sanitize_label(str(rel))}] [{sanitize_label(str(d.get('confidence', '')))}]"
|
|
)
|
|
for nb in G.predecessors(nid):
|
|
d = edge_data(G, nb, nid)
|
|
rel = d.get("relation", "")
|
|
if rel_filter and rel_filter not in rel.lower():
|
|
continue
|
|
lines.append(
|
|
f" <-- {sanitize_label(G.nodes[nb].get('label', nb))} "
|
|
f"[{sanitize_label(str(rel))}] [{sanitize_label(str(d.get('confidence', '')))}]"
|
|
)
|
|
return "\n".join(lines)
|
|
|
|
def _tool_get_community(arguments: dict) -> str:
|
|
cid = int(arguments["community_id"])
|
|
nodes = communities.get(cid, [])
|
|
if not nodes:
|
|
return f"Community {cid} not found."
|
|
header = _community_header(cid, G.nodes[nodes[0]].get("community_name"))
|
|
lines = [f"{header} ({len(nodes)} nodes):"]
|
|
for n in nodes:
|
|
d = G.nodes[n]
|
|
# Sanitise label and source_file (F-010).
|
|
lines.append(
|
|
f" {sanitize_label(d.get('label', n))} "
|
|
f"[{sanitize_label(str(d.get('source_file', '')))}]"
|
|
)
|
|
return "\n".join(lines)
|
|
|
|
def _tool_god_nodes(arguments: dict) -> str:
|
|
from graphify.analyze import god_nodes as _god_nodes
|
|
nodes = _god_nodes(G, top_n=int(arguments.get("top_n", 10)))
|
|
lines = ["God nodes (most connected):"]
|
|
lines += [f" {i}. {n['label']} - {n['degree']} edges" for i, n in enumerate(nodes, 1)]
|
|
return "\n".join(lines)
|
|
|
|
def _tool_graph_stats(_: dict) -> str:
|
|
confs = [d.get("confidence", "EXTRACTED") for _, _, d in G.edges(data=True)]
|
|
total = len(confs) or 1
|
|
return (
|
|
f"Nodes: {G.number_of_nodes()}\n"
|
|
f"Edges: {G.number_of_edges()}\n"
|
|
f"Communities: {len(communities)}\n"
|
|
f"EXTRACTED: {round(confs.count('EXTRACTED')/total*100)}%\n"
|
|
f"INFERRED: {round(confs.count('INFERRED')/total*100)}%\n"
|
|
f"AMBIGUOUS: {round(confs.count('AMBIGUOUS')/total*100)}%\n"
|
|
)
|
|
|
|
def _tool_shortest_path(arguments: dict) -> str:
|
|
src_scored = _score_nodes(G, [t.lower() for t in arguments["source"].split()])
|
|
tgt_scored = _score_nodes(G, [t.lower() for t in arguments["target"].split()])
|
|
if not src_scored:
|
|
return f"No node matching source '{arguments['source']}' found."
|
|
if not tgt_scored:
|
|
return f"No node matching target '{arguments['target']}' found."
|
|
src_nid = _pick_scored_endpoint(G, src_scored, arguments["source"])
|
|
tgt_nid = _pick_scored_endpoint(G, tgt_scored, arguments["target"])
|
|
# Ambiguity guard: when both queries resolve to the same node, the
|
|
# shortest path is trivially zero hops, which is almost never what the
|
|
# caller wanted (see bug #828).
|
|
if src_nid == tgt_nid:
|
|
return (
|
|
f"'{arguments['source']}' and '{arguments['target']}' both resolved to "
|
|
f"the same node '{src_nid}'. Use a more specific label or the exact node ID."
|
|
)
|
|
warnings: list[str] = []
|
|
for name, scored, nid in (
|
|
("source", src_scored, src_nid),
|
|
("target", tgt_scored, tgt_nid),
|
|
):
|
|
# Only meaningful when the raw score head is what got picked — a
|
|
# full-token override was chosen on token coverage, not score.
|
|
if len(scored) >= 2 and nid == scored[0][1]:
|
|
top, runner = scored[0][0], scored[1][0]
|
|
if top > 0 and (top - runner) / top < 0.10:
|
|
warnings.append(
|
|
f"warning: {name} match was ambiguous "
|
|
f"(top score {top:g}, runner-up {runner:g})"
|
|
)
|
|
max_hops = int(arguments.get("max_hops", 8))
|
|
try:
|
|
# Use undirected view for path-finding (works regardless of query src/tgt order)
|
|
path_nodes = nx.shortest_path(G.to_undirected(as_view=True), src_nid, tgt_nid)
|
|
except (nx.NetworkXNoPath, nx.NodeNotFound):
|
|
return f"No path found between '{G.nodes[src_nid].get('label', src_nid)}' and '{G.nodes[tgt_nid].get('label', tgt_nid)}'."
|
|
hops = len(path_nodes) - 1
|
|
if hops > max_hops:
|
|
return f"Path exceeds max_hops={max_hops} ({hops} hops found)."
|
|
segments = []
|
|
for i in range(len(path_nodes) - 1):
|
|
u, v = path_nodes[i], path_nodes[i + 1]
|
|
if G.has_edge(u, v):
|
|
edata = edge_data(G, u, v)
|
|
forward = True
|
|
else:
|
|
edata = edge_data(G, v, u)
|
|
forward = False
|
|
rel = edata.get("relation", "")
|
|
conf = edata.get("confidence", "")
|
|
conf_str = f" [{conf}]" if conf else ""
|
|
if i == 0:
|
|
segments.append(G.nodes[u].get("label", u))
|
|
if forward:
|
|
segments.append(f"--{rel}{conf_str}--> {G.nodes[v].get('label', v)}")
|
|
else:
|
|
segments.append(f"<--{rel}{conf_str}-- {G.nodes[v].get('label', v)}")
|
|
prefix = ("\n".join(warnings) + "\n") if warnings else ""
|
|
return prefix + f"Shortest path ({hops} hops):\n " + " ".join(segments)
|
|
|
|
def _tool_list_prs(arguments: dict) -> str:
|
|
from graphify.prs import fetch_prs, fetch_worktrees, format_prs_text, _detect_default_branch
|
|
repo = arguments.get("repo") or None
|
|
base = arguments.get("base") or _detect_default_branch(repo)
|
|
try:
|
|
prs = fetch_prs(repo=repo, base=base)
|
|
except RuntimeError as e:
|
|
return f"Error: {e}"
|
|
worktrees = fetch_worktrees()
|
|
for pr in prs:
|
|
pr.worktree_path = worktrees.get(pr.branch)
|
|
return format_prs_text(prs, base)
|
|
|
|
def _tool_get_pr_impact(arguments: dict) -> str:
|
|
from graphify.prs import fetch_pr_files, compute_pr_impact, _gh, _parse_ci
|
|
number = int(arguments["pr_number"])
|
|
repo = arguments.get("repo") or None
|
|
# Use gh pr view directly — works for any base branch, not just the default
|
|
view_args = ["pr", "view", str(number), "--json",
|
|
"title,headRefName,baseRefName,author,isDraft,reviewDecision,statusCheckRollup,updatedAt"]
|
|
if repo:
|
|
view_args += ["--repo", repo]
|
|
pr_data = _gh(*view_args)
|
|
if pr_data is None:
|
|
return f"PR #{number} not found or gh not authenticated."
|
|
files = fetch_pr_files(number, repo)
|
|
if not files:
|
|
return f"PR #{number}: no changed files found (may require gh auth)."
|
|
comms, nodes = compute_pr_impact(files, G)
|
|
ci = _parse_ci(pr_data.get("statusCheckRollup") or [])
|
|
lines = [
|
|
f"PR #{number}: {pr_data['title']}",
|
|
f"CI: {ci} Review: {pr_data.get('reviewDecision') or 'none'}",
|
|
f"Base: {pr_data['baseRefName']} Author: {(pr_data.get('author') or {}).get('login', '?')}",
|
|
f"\nGraph impact: {nodes} nodes across {len(comms)} communities",
|
|
f"Communities touched: {comms}",
|
|
f"Files changed ({len(files)}):",
|
|
]
|
|
lines += [f" {f}" for f in files[:20]]
|
|
if len(files) > 20:
|
|
lines.append(f" … and {len(files) - 20} more")
|
|
return "\n".join(lines)
|
|
|
|
def _tool_triage_prs(arguments: dict) -> str:
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
from graphify.prs import fetch_prs, fetch_worktrees, fetch_pr_files, compute_pr_impact, _STATUS_ORDER, _detect_default_branch
|
|
repo = arguments.get("repo") or None
|
|
base = arguments.get("base") or _detect_default_branch(repo)
|
|
try:
|
|
prs = fetch_prs(repo=repo, base=base)
|
|
except RuntimeError as e:
|
|
return f"Error: {e}"
|
|
worktrees = fetch_worktrees()
|
|
for pr in prs:
|
|
pr.worktree_path = worktrees.get(pr.branch)
|
|
actionable = [p for p in prs if p.base_branch == base and p.status not in ("WRONG-BASE", "STALE")]
|
|
if not actionable:
|
|
return f"No actionable PRs targeting {base}."
|
|
# Fetch diffs concurrently then compute graph impact using in-memory G
|
|
workers = min(8, len(actionable))
|
|
with ThreadPoolExecutor(max_workers=workers) as pool:
|
|
future_to_pr = {pool.submit(fetch_pr_files, pr.number, repo): pr for pr in actionable}
|
|
for fut in as_completed(future_to_pr):
|
|
pr = future_to_pr[fut]
|
|
try:
|
|
files = fut.result()
|
|
except Exception:
|
|
files = []
|
|
if files:
|
|
pr.files_changed = files
|
|
pr.communities_touched, pr.nodes_affected = compute_pr_impact(files, G)
|
|
header = (
|
|
f"Actionable PRs targeting {base}: {len(actionable)}\n"
|
|
"Rank these by review priority. Higher blast_radius = more graph communities affected = higher merge risk.\n"
|
|
)
|
|
lines = [header]
|
|
for p in sorted(actionable, key=lambda x: (_STATUS_ORDER.index(x.status) if x.status in _STATUS_ORDER else 99)):
|
|
impact = f" blast_radius={p.blast_radius}" if p.blast_radius else ""
|
|
wt = f" worktree={p.worktree_path}" if p.worktree_path else ""
|
|
lines.append(
|
|
f"PR #{p.number} [{p.status}] CI={p.ci_status} review={p.review_decision or 'none'} "
|
|
f"age={p.days_old}d author={p.author}{impact}{wt}\n title: {p.title}"
|
|
)
|
|
return "\n\n".join(lines)
|
|
|
|
_handlers = {
|
|
"query_graph": _tool_query_graph,
|
|
"get_node": _tool_get_node,
|
|
"get_neighbors": _tool_get_neighbors,
|
|
"get_community": _tool_get_community,
|
|
"god_nodes": _tool_god_nodes,
|
|
"graph_stats": _tool_graph_stats,
|
|
"shortest_path": _tool_shortest_path,
|
|
"list_prs": _tool_list_prs,
|
|
"get_pr_impact": _tool_get_pr_impact,
|
|
"triage_prs": _tool_triage_prs,
|
|
}
|
|
|
|
def _load_community_labels() -> dict[int, str]:
|
|
labels_path = Path(active_graph_path).parent / ".graphify_labels.json"
|
|
if labels_path.exists():
|
|
try:
|
|
return {int(k): v for k, v in json.loads(labels_path.read_text(encoding="utf-8")).items()}
|
|
except Exception:
|
|
pass
|
|
return {cid: f"Community {cid}" for cid in communities}
|
|
|
|
@server.list_resources()
|
|
async def list_resources() -> list[types.Resource]:
|
|
return [
|
|
types.Resource(uri=AnyUrl("graphify://report"), name="Graph Report", description="Full GRAPH_REPORT.md", mimeType="text/markdown"),
|
|
types.Resource(uri=AnyUrl("graphify://stats"), name="Graph Stats", description="Node/edge/community counts and confidence breakdown", mimeType="text/plain"),
|
|
types.Resource(uri=AnyUrl("graphify://god-nodes"), name="God Nodes", description="Top 10 most-connected nodes", mimeType="text/plain"),
|
|
types.Resource(uri=AnyUrl("graphify://surprises"), name="Surprising Connections", description="Cross-community surprising connections", mimeType="text/plain"),
|
|
types.Resource(uri=AnyUrl("graphify://audit"), name="Confidence Audit", description="EXTRACTED/INFERRED/AMBIGUOUS edge breakdown", mimeType="text/plain"),
|
|
types.Resource(uri=AnyUrl("graphify://questions"), name="Suggested Questions", description="Suggested questions for this codebase", mimeType="text/plain"),
|
|
]
|
|
|
|
@server.read_resource()
|
|
async def read_resource(uri: AnyUrl) -> str:
|
|
_select_graph(None) # resources read the server's default graph
|
|
uri_str = str(uri)
|
|
if uri_str == "graphify://report":
|
|
report_path = Path(active_graph_path).parent / "GRAPH_REPORT.md"
|
|
if report_path.exists():
|
|
return report_path.read_text(encoding="utf-8")
|
|
return "GRAPH_REPORT.md not found. Run graphify extract first."
|
|
if uri_str == "graphify://stats":
|
|
return _tool_graph_stats({})
|
|
if uri_str == "graphify://god-nodes":
|
|
return _tool_god_nodes({"top_n": 10})
|
|
if uri_str == "graphify://surprises":
|
|
try:
|
|
from graphify.analyze import surprising_connections
|
|
surprises = surprising_connections(G, communities, top_n=10)
|
|
if not surprises:
|
|
return "No surprising connections found."
|
|
lines = ["Surprising cross-community connections:"]
|
|
for s in surprises:
|
|
lines.append(f" {s.get('source', '')} <-> {s.get('target', '')} [{s.get('relation', '')}]")
|
|
return "\n".join(lines)
|
|
except Exception as exc:
|
|
return f"Could not compute surprising connections: {exc}"
|
|
if uri_str == "graphify://audit":
|
|
confs = [d.get("confidence", "EXTRACTED") for _, _, d in G.edges(data=True)]
|
|
total = len(confs) or 1
|
|
return (
|
|
f"Total edges: {total}\n"
|
|
f"EXTRACTED: {confs.count('EXTRACTED')} ({round(confs.count('EXTRACTED')/total*100)}%)\n"
|
|
f"INFERRED: {confs.count('INFERRED')} ({round(confs.count('INFERRED')/total*100)}%)\n"
|
|
f"AMBIGUOUS: {confs.count('AMBIGUOUS')} ({round(confs.count('AMBIGUOUS')/total*100)}%)\n"
|
|
)
|
|
if uri_str == "graphify://questions":
|
|
try:
|
|
from graphify.analyze import suggest_questions
|
|
community_labels = _load_community_labels()
|
|
questions = suggest_questions(G, communities, community_labels, top_n=10)
|
|
if not questions:
|
|
return "No suggested questions available."
|
|
lines = ["Suggested questions:"]
|
|
for q in questions:
|
|
if isinstance(q, dict):
|
|
lines.append(f" - {q.get('question', '')}")
|
|
else:
|
|
lines.append(f" - {q}")
|
|
return "\n".join(lines)
|
|
except Exception as exc:
|
|
return f"Could not generate questions: {exc}"
|
|
raise ValueError(f"Unknown resource: {uri_str}")
|
|
|
|
@server.call_tool()
|
|
async def call_tool(name: str, arguments: dict) -> list[types.TextContent]:
|
|
arguments = dict(arguments or {})
|
|
project_path = arguments.pop("project_path", None)
|
|
handler = _handlers.get(name)
|
|
if not handler:
|
|
return [types.TextContent(type="text", text=f"Unknown tool: {name}")]
|
|
try:
|
|
_select_graph(project_path) # bind G/communities to the target graph
|
|
return [types.TextContent(type="text", text=handler(arguments))]
|
|
except Exception as exc:
|
|
return [types.TextContent(type="text", text=f"Error executing {name}: {exc}")]
|
|
|
|
return server
|
|
|
|
|
|
def serve(graph_path: str | None = None) -> None:
|
|
"""Start the MCP server over stdio (the default, per-developer transport)."""
|
|
graph_path = graph_path or _default_graph_json()
|
|
try:
|
|
from mcp.server.stdio import stdio_server
|
|
except ImportError as e:
|
|
raise ImportError('mcp not installed. Run: pip install "graphifyy[mcp]"') from e
|
|
import asyncio
|
|
|
|
server = _build_server(graph_path)
|
|
|
|
async def main() -> None:
|
|
async with stdio_server() as streams:
|
|
await server.run(streams[0], streams[1], server.create_initialization_options())
|
|
|
|
_filter_blank_stdin()
|
|
asyncio.run(main())
|
|
|
|
|
|
class _MCPASGIApp:
|
|
"""Raw-ASGI wrapper around the Streamable HTTP session manager.
|
|
|
|
Passed to a Starlette ``Route`` as a class instance (not a function) so
|
|
Starlette treats it as an ASGI app: it serves the exact mount path for all
|
|
methods (GET/POST/DELETE) with no request/response wrapping and no
|
|
trailing-slash redirect — mirroring how FastMCP mounts the same manager.
|
|
"""
|
|
|
|
def __init__(self, manager) -> None:
|
|
self._manager = manager
|
|
|
|
async def __call__(self, scope, receive, send) -> None:
|
|
await self._manager.handle_request(scope, receive, send)
|
|
|
|
|
|
class _ApiKeyMiddleware:
|
|
"""Pure-ASGI API-key gate for the HTTP transport.
|
|
|
|
Implemented as raw ASGI (not Starlette's BaseHTTPMiddleware) on purpose:
|
|
BaseHTTPMiddleware buffers responses and breaks the Streamable HTTP SSE
|
|
stream. This short-circuits with 401 before the request ever reaches the
|
|
session manager, leaving the streaming path untouched for authorized calls.
|
|
"""
|
|
|
|
def __init__(self, app, api_key: str) -> None:
|
|
self.app = app
|
|
self._expected = api_key.encode("utf-8")
|
|
|
|
async def __call__(self, scope, receive, send) -> None:
|
|
if scope["type"] != "http":
|
|
await self.app(scope, receive, send)
|
|
return
|
|
import hmac
|
|
headers = dict(scope.get("headers") or [])
|
|
provided = headers.get(b"x-api-key")
|
|
if provided is None:
|
|
# RFC 6750: the auth scheme token is case-insensitive.
|
|
scheme, _, token = headers.get(b"authorization", b"").partition(b" ")
|
|
if scheme.lower() == b"bearer" and token:
|
|
provided = token.strip()
|
|
# Constant-time compare; reject when no key was supplied at all.
|
|
if provided is None or not hmac.compare_digest(provided, self._expected):
|
|
body = b'{"error": "unauthorized"}'
|
|
await send({
|
|
"type": "http.response.start",
|
|
"status": 401,
|
|
"headers": [
|
|
(b"content-type", b"application/json"),
|
|
(b"content-length", str(len(body)).encode("ascii")),
|
|
],
|
|
})
|
|
await send({"type": "http.response.body", "body": body})
|
|
return
|
|
await self.app(scope, receive, send)
|
|
|
|
|
|
def _build_http_app(
|
|
graph_path: str,
|
|
*,
|
|
host: str = "127.0.0.1",
|
|
port: int = 8080,
|
|
api_key: str | None = None,
|
|
path: str = "/mcp",
|
|
json_response: bool = False,
|
|
stateless: bool = False,
|
|
session_timeout: float | None = 3600.0,
|
|
):
|
|
"""Build the Starlette ASGI app for the Streamable HTTP transport.
|
|
|
|
Split out from :func:`serve_http` (which blocks on uvicorn) so the wiring
|
|
can be exercised with an in-process ASGI test client.
|
|
|
|
``session_timeout`` reaps stateful sessions idle for that many seconds so a
|
|
long-running shared server does not leak memory when IDE clients disconnect
|
|
without sending a DELETE. ``None`` (or <= 0) disables reaping; it is forced
|
|
to ``None`` in stateless mode, which has no sessions to reap.
|
|
"""
|
|
try:
|
|
import contextlib
|
|
|
|
from starlette.applications import Starlette
|
|
from starlette.middleware import Middleware
|
|
from starlette.routing import Route
|
|
|
|
from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
|
|
from mcp.server.transport_security import TransportSecuritySettings
|
|
except ImportError as e:
|
|
raise ImportError(
|
|
'HTTP transport needs the mcp extra (mcp + starlette + uvicorn). '
|
|
'Run: pip install "graphifyy[mcp]"'
|
|
) from e
|
|
|
|
# A blank key (e.g. --api-key "" or an empty GRAPHIFY_API_KEY) must not be
|
|
# mistaken for "auth on" — normalize it to None so the gate is unambiguous.
|
|
api_key = (api_key or "").strip() or None
|
|
|
|
server = _build_server(graph_path)
|
|
|
|
# DNS-rebinding protection. When the operator binds a wildcard address they
|
|
# are intentionally exposing the server, so accept any Host header; for a
|
|
# loopback/specific bind, restrict Host to that address (with and without
|
|
# the port) plus the localhost aliases.
|
|
if host in ("0.0.0.0", "::", ""):
|
|
security = TransportSecuritySettings(enable_dns_rebinding_protection=False)
|
|
else:
|
|
allowed = {host, "localhost", "127.0.0.1"}
|
|
allowed |= {f"{h}:{port}" for h in list(allowed)}
|
|
security = TransportSecuritySettings(allowed_hosts=sorted(allowed))
|
|
|
|
# The SDK rejects a non-positive timeout and forbids one in stateless mode.
|
|
idle_timeout = None if (stateless or not session_timeout or session_timeout <= 0) else session_timeout
|
|
|
|
manager = StreamableHTTPSessionManager(
|
|
app=server,
|
|
json_response=json_response,
|
|
stateless=stateless,
|
|
security_settings=security,
|
|
session_idle_timeout=idle_timeout,
|
|
)
|
|
|
|
@contextlib.asynccontextmanager
|
|
async def lifespan(_app):
|
|
# The session manager owns an anyio task group that must wrap the whole
|
|
# server lifetime, so enter it here rather than per-request.
|
|
async with manager.run():
|
|
yield
|
|
|
|
middleware = []
|
|
if api_key:
|
|
middleware.append(Middleware(_ApiKeyMiddleware, api_key=api_key))
|
|
|
|
return Starlette(
|
|
routes=[Route(path, endpoint=_MCPASGIApp(manager))],
|
|
middleware=middleware,
|
|
lifespan=lifespan,
|
|
)
|
|
|
|
|
|
def serve_http(
|
|
graph_path: str | None = None,
|
|
*,
|
|
host: str = "127.0.0.1",
|
|
port: int = 8080,
|
|
api_key: str | None = None,
|
|
path: str = "/mcp",
|
|
json_response: bool = False,
|
|
stateless: bool = False,
|
|
session_timeout: float | None = 3600.0,
|
|
) -> None:
|
|
"""Start the MCP server over Streamable HTTP (MCP spec 2025-03-26).
|
|
|
|
Serves the same tools/resources as the stdio transport, so a single shared
|
|
process can host the graph for a whole team. Clients point their IDE MCP
|
|
config at ``http://<host>:<port><path>`` (default ``/mcp``).
|
|
|
|
``api_key`` (or the ``GRAPHIFY_API_KEY`` env var) enables a simple header
|
|
check (``Authorization: Bearer <key>`` or ``X-API-Key: <key>``). OAuth is a
|
|
deliberate follow-up. Binding ``0.0.0.0`` exposes the server beyond
|
|
localhost — set an api_key when you do.
|
|
"""
|
|
graph_path = graph_path or _default_graph_json()
|
|
try:
|
|
import uvicorn
|
|
except ImportError as e:
|
|
raise ImportError(
|
|
'HTTP transport needs the mcp extra (mcp + starlette + uvicorn). '
|
|
'Run: pip install "graphifyy[mcp]"'
|
|
) from e
|
|
|
|
api_key = (api_key or "").strip() or None
|
|
|
|
app = _build_http_app(
|
|
graph_path,
|
|
host=host,
|
|
port=port,
|
|
api_key=api_key,
|
|
path=path,
|
|
json_response=json_response,
|
|
stateless=stateless,
|
|
session_timeout=session_timeout,
|
|
)
|
|
|
|
auth_note = "api-key required" if api_key else "no auth (set --api-key to require one)"
|
|
print(
|
|
f"graphify MCP server (streamable-http) on http://{host}:{port}{path} - {auth_note}",
|
|
file=sys.stderr,
|
|
)
|
|
if host in ("0.0.0.0", "::", "") and not api_key:
|
|
print(
|
|
f"WARNING: binding {host or '0.0.0.0'} with no api-key exposes the graph "
|
|
"unauthenticated on the network. Set --api-key (or GRAPHIFY_API_KEY).",
|
|
file=sys.stderr,
|
|
)
|
|
uvicorn.run(app, host=host, port=port)
|
|
|
|
|
|
def _main(argv: list[str] | None = None) -> None:
|
|
import argparse
|
|
import os
|
|
|
|
parser = argparse.ArgumentParser(
|
|
prog="python -m graphify.serve",
|
|
description="Serve a graphify knowledge graph over MCP (stdio or Streamable HTTP).",
|
|
)
|
|
parser.add_argument(
|
|
"graph_path",
|
|
nargs="?",
|
|
default=None,
|
|
help="Path to graph.json (default: graphify-out/graph.json)",
|
|
)
|
|
parser.add_argument(
|
|
"--graph",
|
|
dest="graph_flag",
|
|
default=None,
|
|
metavar="PATH",
|
|
help="Path to graph.json — alias for the positional argument",
|
|
)
|
|
parser.add_argument(
|
|
"--transport",
|
|
choices=["stdio", "http"],
|
|
default="stdio",
|
|
help="Transport to serve on (default: stdio)",
|
|
)
|
|
parser.add_argument("--host", default="127.0.0.1", help="HTTP bind host (default: 127.0.0.1)")
|
|
parser.add_argument("--port", type=int, default=8080, help="HTTP bind port (default: 8080)")
|
|
parser.add_argument(
|
|
"--api-key",
|
|
default=os.environ.get("GRAPHIFY_API_KEY"),
|
|
help="Require this key on the HTTP transport (env: GRAPHIFY_API_KEY)",
|
|
)
|
|
parser.add_argument("--path", default="/mcp", help="HTTP mount path (default: /mcp)")
|
|
parser.add_argument(
|
|
"--json-response",
|
|
action="store_true",
|
|
help="Return plain JSON responses instead of SSE streams",
|
|
)
|
|
parser.add_argument(
|
|
"--stateless",
|
|
action="store_true",
|
|
help="Run without per-session state (for load-balanced / CI deployments)",
|
|
)
|
|
parser.add_argument(
|
|
"--session-timeout",
|
|
type=float,
|
|
default=3600.0,
|
|
help="Reap stateful sessions idle this many seconds (default: 3600; 0 disables)",
|
|
)
|
|
args = parser.parse_args(argv)
|
|
graph_path = args.graph_flag or args.graph_path or _default_graph_json()
|
|
|
|
if args.transport == "http":
|
|
serve_http(
|
|
graph_path,
|
|
host=args.host,
|
|
port=args.port,
|
|
api_key=args.api_key,
|
|
path=args.path,
|
|
json_response=args.json_response,
|
|
stateless=args.stateless,
|
|
session_timeout=args.session_timeout,
|
|
)
|
|
else:
|
|
serve(graph_path)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
_main()
|