3256e4e7c3
- find_path: BFS traversal to trace relationship chains between two files (max N hops) - get_file_signatures: lightweight context mode returning just function/type names - db.find_path(): bidirectional BFS with shortest-path tracking - db.get_all_files(): list all files with docs and staleness - db.get_file_signatures(): return functions list without full doc payload Inspired by Octocode's GraphRAG path-finding pattern. Addresses Mike/Dmitry feedback on usable roll-up summaries.
205 lines
7.2 KiB
Python
205 lines
7.2 KiB
Python
"""MCP server for Developer Intelligence POC. Queries SQLite, serves to Claude Code."""
|
|
|
|
import os
|
|
import sys
|
|
import json
|
|
from pathlib import Path
|
|
|
|
# Load .env if present
|
|
_env_file = Path(__file__).parent / ".env"
|
|
if _env_file.exists():
|
|
for line in _env_file.read_text().splitlines():
|
|
line = line.strip()
|
|
if line and not line.startswith("#") and "=" in line:
|
|
key, _, val = line.partition("=")
|
|
os.environ.setdefault(key.strip(), val.strip())
|
|
|
|
sys.path.insert(0, os.path.dirname(__file__))
|
|
|
|
from db import GraphDB
|
|
from mcp.server.fastmcp import FastMCP
|
|
|
|
mcp = FastMCP("Developer Intelligence")
|
|
|
|
|
|
def _get_db():
|
|
return GraphDB()
|
|
|
|
|
|
@mcp.tool()
|
|
def get_file_doc(path: str) -> str:
|
|
"""Get the generated documentation for a source file. Pass the relative file path (e.g. 'echo.go' or 'middleware/compress.go')."""
|
|
db = _get_db()
|
|
f = db.get_file(path)
|
|
db.close()
|
|
if not f:
|
|
return f"File not found: {path}"
|
|
staleness = " [STALE]" if f["staleness"] == "stale" else ""
|
|
prev = ""
|
|
if f.get("prev_documentation"):
|
|
prev = f"\n\n--- Previous version ---\n{f['prev_documentation']}"
|
|
return f"{f['documentation']}{staleness}\n\n(commit: {f['last_commit']}, updated: {f['updated_at']}){prev}"
|
|
|
|
|
|
@mcp.tool()
|
|
def get_relationship(file_a: str, file_b: str) -> str:
|
|
"""Get the documentation for the import relationship between two files."""
|
|
db = _get_db()
|
|
rel = db.get_relationship(file_a, file_b)
|
|
db.close()
|
|
if not rel:
|
|
return f"No relationship found between {file_a} and {file_b}"
|
|
doc = rel["documentation"] or "(no relationship documentation generated yet)"
|
|
staleness = " [STALE]" if rel["staleness"] == "stale" else ""
|
|
return f"{doc}{staleness}"
|
|
|
|
|
|
@mcp.tool()
|
|
def get_repo_overview() -> str:
|
|
"""Get the repo-level documentation summary — a high-level overview of the entire project."""
|
|
db = _get_db()
|
|
repo = db.get_repo()
|
|
db.close()
|
|
if not repo:
|
|
return "No repo found"
|
|
staleness = " [STALE]" if repo["staleness"] == "stale" else ""
|
|
return f"# {repo['name']}{staleness}\n\n{repo['documentation']}"
|
|
|
|
|
|
@mcp.tool()
|
|
def get_dependents(path: str) -> str:
|
|
"""Get all files that import/depend on the given file. Shows what breaks if you change this file."""
|
|
db = _get_db()
|
|
deps = db.get_dependents(path)
|
|
db.close()
|
|
if not deps:
|
|
return f"No files depend on {path}"
|
|
lines = [f"Files that depend on {path} ({len(deps)} total):\n"]
|
|
for d in deps:
|
|
staleness = " [STALE]" if d["rel_staleness"] == "stale" else ""
|
|
doc = d["rel_doc"] or "(no relationship doc)"
|
|
lines.append(f" {d['from_file']}{staleness}")
|
|
lines.append(f" Relationship: {doc}")
|
|
lines.append(f" File: {d['file_doc'][:150]}...")
|
|
return "\n".join(lines)
|
|
|
|
|
|
@mcp.tool()
|
|
def get_dependencies(path: str) -> str:
|
|
"""Get all files that the given file imports/depends on."""
|
|
db = _get_db()
|
|
deps = db.get_dependencies(path)
|
|
db.close()
|
|
if not deps:
|
|
return f"{path} has no tracked dependencies"
|
|
lines = [f"Dependencies of {path} ({len(deps)} total):\n"]
|
|
for d in deps:
|
|
staleness = " [STALE]" if d["rel_staleness"] == "stale" else ""
|
|
doc = d["rel_doc"] or "(no relationship doc)"
|
|
lines.append(f" {d['to_file']}{staleness}")
|
|
lines.append(f" Relationship: {doc}")
|
|
return "\n".join(lines)
|
|
|
|
|
|
@mcp.tool()
|
|
def search_docs(query: str) -> str:
|
|
"""Search across all file documentation by keyword. Use to find files related to a concept (e.g. 'routing', 'middleware', 'authentication')."""
|
|
db = _get_db()
|
|
results = db.search_docs(query)
|
|
db.close()
|
|
if not results:
|
|
return f"No files found matching '{query}'"
|
|
lines = [f"Files matching '{query}' ({len(results)} results):\n"]
|
|
for r in results:
|
|
staleness = " [STALE]" if r["staleness"] == "stale" else ""
|
|
doc = r["documentation"][:200] + "..." if len(r["documentation"]) > 200 else r["documentation"]
|
|
lines.append(f" {r['path']}{staleness}: {doc}")
|
|
return "\n".join(lines)
|
|
|
|
|
|
@mcp.tool()
|
|
def get_stale_docs() -> str:
|
|
"""List all entities and relationships with stale (outdated) documentation."""
|
|
db = _get_db()
|
|
stale_rels = db.get_stale_relationships()
|
|
stale_repos = db.get_stale_repos()
|
|
stats = db.get_stats()
|
|
db.close()
|
|
|
|
lines = ["Stale documentation:\n"]
|
|
if stale_repos:
|
|
lines.append(f" Repos ({len(stale_repos)}):")
|
|
for r in stale_repos:
|
|
lines.append(f" {r['name']}")
|
|
lines.append(f" Files: {stats['stale_files']} stale")
|
|
if stale_rels:
|
|
lines.append(f" Relationships ({len(stale_rels)}):")
|
|
for r in stale_rels[:20]: # Cap output
|
|
lines.append(f" {r['from_file']} -> {r['to_file']}")
|
|
if len(stale_rels) > 20:
|
|
lines.append(f" ... and {len(stale_rels) - 20} more")
|
|
if stats["stale_files"] == 0 and stats["stale_relationships"] == 0:
|
|
lines.append(" Everything is fresh!")
|
|
return "\n".join(lines)
|
|
|
|
|
|
@mcp.tool()
|
|
def find_path(source: str, target: str, max_depth: int = 4) -> str:
|
|
"""Trace the relationship chain between two files. Shows how file A connects to file B through imports/dependencies. Useful for understanding impact radius and architectural coupling."""
|
|
db = _get_db()
|
|
paths = db.find_path(source, target, max_depth)
|
|
db.close()
|
|
if not paths:
|
|
return f"No connection found between {source} and {target} within {max_depth} hops."
|
|
lines = [f"Connection paths from {source} → {target} ({len(paths)} found):\n"]
|
|
for i, path in enumerate(paths[:5]): # Cap at 5 paths
|
|
chain = " → ".join(path)
|
|
lines.append(f" Path {i+1} ({len(path)-1} hops): {chain}")
|
|
if len(paths) > 5:
|
|
lines.append(f" ... and {len(paths) - 5} more paths")
|
|
return "\n".join(lines)
|
|
|
|
|
|
@mcp.tool()
|
|
def get_file_signatures(path: str) -> str:
|
|
"""Get just the function/type signatures for a file — lightweight context without full documentation. Useful when you need a quick map of what a file exports."""
|
|
db = _get_db()
|
|
f = db.get_file_signatures(path)
|
|
db.close()
|
|
if not f:
|
|
return f"File not found: {path}"
|
|
import json
|
|
try:
|
|
funcs = json.loads(f["functions"]) if f["functions"] else []
|
|
except (json.JSONDecodeError, TypeError):
|
|
funcs = []
|
|
staleness = " [STALE]" if f["staleness"] == "stale" else ""
|
|
lines = [f"Signatures for {path}{staleness} ({f['language']}):\n"]
|
|
if funcs:
|
|
for fn in funcs:
|
|
lines.append(f" • {fn}")
|
|
else:
|
|
lines.append(" (no functions extracted)")
|
|
return "\n".join(lines)
|
|
|
|
|
|
@mcp.tool()
|
|
def get_graph_stats() -> str:
|
|
"""Get overall knowledge graph statistics — file count, relationship count, staleness."""
|
|
db = _get_db()
|
|
stats = db.get_stats()
|
|
repo = db.get_repo()
|
|
db.close()
|
|
return json.dumps({
|
|
"repo": repo["name"] if repo else None,
|
|
"files": stats["files"],
|
|
"relationships": stats["relationships"],
|
|
"stale_files": stats["stale_files"],
|
|
"stale_relationships": stats["stale_relationships"],
|
|
}, indent=2)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print("Starting Developer Intelligence MCP Server (stdio)...")
|
|
mcp.run(transport="stdio")
|