dev-intel-poc/mcp_server.py

"""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")