|
1
|
""" |
|
2
|
Navegador MCP server — exposes graph context tools to AI coding agents. |
|
3
|
|
|
4
|
Run: |
|
5
|
navegador mcp --db .navegador/graph.db |
|
6
|
""" |
|
7
|
|
|
8
|
import json |
|
9
|
import logging |
|
10
|
from typing import Any |
|
11
|
|
|
12
|
logger = logging.getLogger(__name__) |
|
13
|
|
|
14
|
|
|
15
|
def create_mcp_server(store_factory, read_only: bool = False): |
|
16
|
""" |
|
17
|
Build and return an MCP server instance wired to a GraphStore factory. |
|
18
|
|
|
19
|
Args: |
|
20
|
store_factory: Callable[[], GraphStore] — called lazily on first request. |
|
21
|
read_only: When True, the ingest_repo tool is disabled and all |
|
22
|
query_graph queries are validated for write operations and |
|
23
|
injection patterns. Complexity checks apply to all modes. |
|
24
|
""" |
|
25
|
try: |
|
26
|
from mcp.server import Server # type: ignore[import] |
|
27
|
from mcp.types import TextContent, Tool # type: ignore[import] |
|
28
|
except ImportError as e: |
|
29
|
raise ImportError("Install mcp: pip install mcp") from e |
|
30
|
|
|
31
|
from navegador.context import ContextLoader |
|
32
|
from navegador.mcp.security import check_complexity, validate_cypher |
|
33
|
|
|
34
|
server = Server("navegador") |
|
35
|
_store: Any = None |
|
36
|
_loader: ContextLoader | None = None |
|
37
|
|
|
38
|
def _get_loader() -> ContextLoader: |
|
39
|
nonlocal _store, _loader |
|
40
|
if _loader is None: |
|
41
|
_store = store_factory() |
|
42
|
_loader = ContextLoader(_store) |
|
43
|
return _loader |
|
44
|
|
|
45
|
@server.list_tools() |
|
46
|
async def list_tools() -> list[Tool]: |
|
47
|
return [ |
|
48
|
Tool( |
|
49
|
name="ingest_repo", |
|
50
|
description="Parse and ingest a local code repository into the navegador graph.", |
|
51
|
inputSchema={ |
|
52
|
"type": "object", |
|
53
|
"properties": { |
|
54
|
"path": {"type": "string", "description": "Absolute path to the repo."}, |
|
55
|
"clear": { |
|
56
|
"type": "boolean", |
|
57
|
"description": "Clear existing graph before ingesting.", |
|
58
|
"default": False, |
|
59
|
}, |
|
60
|
}, |
|
61
|
"required": ["path"], |
|
62
|
}, |
|
63
|
), |
|
64
|
Tool( |
|
65
|
name="load_file_context", |
|
66
|
description="Return all symbols (functions, classes, imports) in a file.", |
|
67
|
inputSchema={ |
|
68
|
"type": "object", |
|
69
|
"properties": { |
|
70
|
"file_path": { |
|
71
|
"type": "string", |
|
72
|
"description": "Relative file path within the ingested repo.", |
|
73
|
}, |
|
74
|
"format": { |
|
75
|
"type": "string", |
|
76
|
"enum": ["json", "markdown"], |
|
77
|
"default": "markdown", |
|
78
|
}, |
|
79
|
}, |
|
80
|
"required": ["file_path"], |
|
81
|
}, |
|
82
|
), |
|
83
|
Tool( |
|
84
|
name="load_function_context", |
|
85
|
description="Return context for a function — what it calls and what calls it.", |
|
86
|
inputSchema={ |
|
87
|
"type": "object", |
|
88
|
"properties": { |
|
89
|
"name": {"type": "string", "description": "Function name."}, |
|
90
|
"file_path": {"type": "string", "description": "Relative file path."}, |
|
91
|
"depth": {"type": "integer", "default": 2}, |
|
92
|
"format": { |
|
93
|
"type": "string", |
|
94
|
"enum": ["json", "markdown"], |
|
95
|
"default": "markdown", |
|
96
|
}, |
|
97
|
}, |
|
98
|
"required": ["name", "file_path"], |
|
99
|
}, |
|
100
|
), |
|
101
|
Tool( |
|
102
|
name="load_class_context", |
|
103
|
description="Return context for a class — methods, inheritance, subclasses.", |
|
104
|
inputSchema={ |
|
105
|
"type": "object", |
|
106
|
"properties": { |
|
107
|
"name": {"type": "string", "description": "Class name."}, |
|
108
|
"file_path": {"type": "string", "description": "Relative file path."}, |
|
109
|
"format": { |
|
110
|
"type": "string", |
|
111
|
"enum": ["json", "markdown"], |
|
112
|
"default": "markdown", |
|
113
|
}, |
|
114
|
}, |
|
115
|
"required": ["name", "file_path"], |
|
116
|
}, |
|
117
|
), |
|
118
|
Tool( |
|
119
|
name="search_symbols", |
|
120
|
description="Fuzzy search for functions, classes, or methods by name.", |
|
121
|
inputSchema={ |
|
122
|
"type": "object", |
|
123
|
"properties": { |
|
124
|
"query": {"type": "string", "description": "Partial name to search."}, |
|
125
|
"limit": {"type": "integer", "default": 20}, |
|
126
|
}, |
|
127
|
"required": ["query"], |
|
128
|
}, |
|
129
|
), |
|
130
|
Tool( |
|
131
|
name="query_graph", |
|
132
|
description="Execute a raw Cypher query against the navegador graph.", |
|
133
|
inputSchema={ |
|
134
|
"type": "object", |
|
135
|
"properties": { |
|
136
|
"cypher": {"type": "string", "description": "Cypher query string."}, |
|
137
|
}, |
|
138
|
"required": ["cypher"], |
|
139
|
}, |
|
140
|
), |
|
141
|
Tool( |
|
142
|
name="graph_stats", |
|
143
|
description="Return node and edge counts for the current graph.", |
|
144
|
inputSchema={"type": "object", "properties": {}}, |
|
145
|
), |
|
146
|
Tool( |
|
147
|
name="get_rationale", |
|
148
|
description="Return rationale, alternatives, and status of a decision.", |
|
149
|
inputSchema={ |
|
150
|
"type": "object", |
|
151
|
"properties": { |
|
152
|
"name": {"type": "string", "description": "Decision name."}, |
|
153
|
"format": { |
|
154
|
"type": "string", |
|
155
|
"enum": ["json", "markdown"], |
|
156
|
"default": "markdown", |
|
157
|
}, |
|
158
|
}, |
|
159
|
"required": ["name"], |
|
160
|
}, |
|
161
|
), |
|
162
|
Tool( |
|
163
|
name="find_owners", |
|
164
|
description="Find people (owners, stakeholders) assigned to a node.", |
|
165
|
inputSchema={ |
|
166
|
"type": "object", |
|
167
|
"properties": { |
|
168
|
"name": {"type": "string", "description": "Node name."}, |
|
169
|
"file_path": { |
|
170
|
"type": "string", |
|
171
|
"description": "Narrow to a specific file.", |
|
172
|
"default": "", |
|
173
|
}, |
|
174
|
}, |
|
175
|
"required": ["name"], |
|
176
|
}, |
|
177
|
), |
|
178
|
Tool( |
|
179
|
name="search_knowledge", |
|
180
|
description="Search concepts, rules, decisions, and wiki pages.", |
|
181
|
inputSchema={ |
|
182
|
"type": "object", |
|
183
|
"properties": { |
|
184
|
"query": {"type": "string", "description": "Search query."}, |
|
185
|
"limit": {"type": "integer", "default": 20}, |
|
186
|
}, |
|
187
|
"required": ["query"], |
|
188
|
}, |
|
189
|
), |
|
190
|
Tool( |
|
191
|
name="blast_radius", |
|
192
|
description=( |
|
193
|
"Impact analysis: find all nodes and files affected by changing a symbol. " |
|
194
|
"Traverses CALLS, REFERENCES, INHERITS, IMPLEMENTS, ANNOTATES edges outward." |
|
195
|
), |
|
196
|
inputSchema={ |
|
197
|
"type": "object", |
|
198
|
"properties": { |
|
199
|
"name": {"type": "string", "description": "Symbol name to analyse."}, |
|
200
|
"file_path": { |
|
201
|
"type": "string", |
|
202
|
"description": "Narrow to a specific file (optional).", |
|
203
|
"default": "", |
|
204
|
}, |
|
205
|
"depth": { |
|
206
|
"type": "integer", |
|
207
|
"description": "Maximum traversal depth.", |
|
208
|
"default": 3, |
|
209
|
}, |
|
210
|
}, |
|
211
|
"required": ["name"], |
|
212
|
}, |
|
213
|
), |
|
214
|
] |
|
215
|
|
|
216
|
@server.call_tool() |
|
217
|
async def call_tool(name: str, arguments: dict) -> list[TextContent]: |
|
218
|
loader = _get_loader() |
|
219
|
|
|
220
|
if name == "ingest_repo": |
|
221
|
if read_only: |
|
222
|
return [ |
|
223
|
TextContent( |
|
224
|
type="text", |
|
225
|
text="Error: ingest_repo is disabled in read-only mode.", |
|
226
|
) |
|
227
|
] |
|
228
|
from navegador.ingestion import RepoIngester |
|
229
|
|
|
230
|
ingester = RepoIngester(loader.store) |
|
231
|
stats = ingester.ingest(arguments["path"], clear=arguments.get("clear", False)) |
|
232
|
return [TextContent(type="text", text=json.dumps(stats, indent=2))] |
|
233
|
|
|
234
|
elif name == "load_file_context": |
|
235
|
bundle = loader.load_file(arguments["file_path"]) |
|
236
|
fmt = arguments.get("format", "markdown") |
|
237
|
text = bundle.to_markdown() if fmt == "markdown" else bundle.to_json() |
|
238
|
return [TextContent(type="text", text=text)] |
|
239
|
|
|
240
|
elif name == "load_function_context": |
|
241
|
bundle = loader.load_function( |
|
242
|
arguments["name"], |
|
243
|
arguments["file_path"], |
|
244
|
depth=arguments.get("depth", 2), |
|
245
|
) |
|
246
|
fmt = arguments.get("format", "markdown") |
|
247
|
text = bundle.to_markdown() if fmt == "markdown" else bundle.to_json() |
|
248
|
return [TextContent(type="text", text=text)] |
|
249
|
|
|
250
|
elif name == "load_class_context": |
|
251
|
bundle = loader.load_class(arguments["name"], arguments["file_path"]) |
|
252
|
fmt = arguments.get("format", "markdown") |
|
253
|
text = bundle.to_markdown() if fmt == "markdown" else bundle.to_json() |
|
254
|
return [TextContent(type="text", text=text)] |
|
255
|
|
|
256
|
elif name == "search_symbols": |
|
257
|
results = loader.search(arguments["query"], limit=arguments.get("limit", 20)) |
|
258
|
lines = [f"- **{r.type}** `{r.name}` — `{r.file_path}`:{r.line_start}" for r in results] |
|
259
|
return [TextContent(type="text", text="\n".join(lines) or "No results.")] |
|
260
|
|
|
261
|
elif name == "query_graph": |
|
262
|
cypher = arguments["cypher"] |
|
263
|
if read_only: |
|
264
|
try: |
|
265
|
validate_cypher(cypher) |
|
266
|
except Exception as exc: |
|
267
|
return [TextContent(type="text", text=f"Error: {exc}")] |
|
268
|
try: |
|
269
|
check_complexity(cypher) |
|
270
|
except Exception as exc: |
|
271
|
return [TextContent(type="text", text=f"Error: {exc}")] |
|
272
|
result = loader.store.query(cypher) |
|
273
|
rows = result.result_set or [] |
|
274
|
text = json.dumps(rows, default=str, indent=2) |
|
275
|
return [TextContent(type="text", text=text)] |
|
276
|
|
|
277
|
elif name == "graph_stats": |
|
278
|
stats = { |
|
279
|
"nodes": loader.store.node_count(), |
|
280
|
"edges": loader.store.edge_count(), |
|
281
|
} |
|
282
|
return [TextContent(type="text", text=json.dumps(stats, indent=2))] |
|
283
|
|
|
284
|
elif name == "get_rationale": |
|
285
|
bundle = loader.load_decision(arguments["name"]) |
|
286
|
fmt = arguments.get("format", "markdown") |
|
287
|
text = bundle.to_markdown() if fmt == "markdown" else bundle.to_json() |
|
288
|
return [TextContent(type="text", text=text)] |
|
289
|
|
|
290
|
elif name == "find_owners": |
|
291
|
results = loader.find_owners( |
|
292
|
arguments["name"], file_path=arguments.get("file_path", "") |
|
293
|
) |
|
294
|
if not results: |
|
295
|
return [TextContent(type="text", text="No owners found.")] |
|
296
|
lines = [f"- **{r.name}** ({r.description})" for r in results] |
|
297
|
return [TextContent(type="text", text="\n".join(lines))] |
|
298
|
|
|
299
|
elif name == "search_knowledge": |
|
300
|
results = loader.search_knowledge(arguments["query"], limit=arguments.get("limit", 20)) |
|
301
|
if not results: |
|
302
|
return [TextContent(type="text", text="No results.")] |
|
303
|
lines = [f"- **{r.type}** `{r.name}` — {r.description or ''}" for r in results] |
|
304
|
return [TextContent(type="text", text="\n".join(lines))] |
|
305
|
|
|
306
|
elif name == "blast_radius": |
|
307
|
from navegador.analysis.impact import ImpactAnalyzer |
|
308
|
|
|
309
|
result = ImpactAnalyzer(loader.store).blast_radius( |
|
310
|
arguments["name"], |
|
311
|
file_path=arguments.get("file_path", ""), |
|
312
|
depth=arguments.get("depth", 3), |
|
313
|
) |
|
314
|
return [TextContent(type="text", text=json.dumps(result.to_dict(), indent=2))] |
|
315
|
|
|
316
|
return [TextContent(type="text", text=f"Unknown tool: {name}")] |
|
317
|
|
|
318
|
return server |
|
319
|
|