mcp_server/server.py

"""
Evolva Codebase Memory MCP Server
FastAPI server exposing 6 MCP tools for codebase knowledge graph access.
Serves 54 indexed repositories across a 7-layer architecture.
"""

import json
import os
from pathlib import Path
from typing import Any, Dict, List, Optional

from fastapi import FastAPI, HTTPException, Query
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from fastapi.staticfiles import StaticFiles

# ── App setup ──────────────────────────────────────────────────────────────────
app = FastAPI(
    title="Evolva Codebase Memory MCP",
    description="MCP server providing codebase memory and knowledge graph access for 54 Evolva repositories",
    version="1.0.0",
    docs_url="/api/docs",
    redoc_url="/api/redoc",
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# ── Load knowledge graph ───────────────────────────────────────────────────────
KG_PATH = Path(__file__).parent.parent / "kg" / "knowledge_graph.json"

def load_kg() -> Dict[str, Any]:
    with open(KG_PATH, "r") as f:
        return json.load(f)

KG = load_kg()

def get_repos_list() -> List[Dict[str, Any]]:
    """Return repositories as a list, handling both dict and list formats."""
    repos = KG["repositories"]
    if isinstance(repos, dict):
        return list(repos.values())
    return repos

# ── Mount static docs ──────────────────────────────────────────────────────────
DOCS_PATH = Path(__file__).parent.parent / "docs"
if DOCS_PATH.exists():
    app.mount("/docs", StaticFiles(directory=str(DOCS_PATH), html=True), name="docs")

# ── Health endpoint ────────────────────────────────────────────────────────────
@app.get("/health")
async def health():
    """Health check endpoint."""
    return {
        "status": "up",
        "repos_indexed": KG["metadata"]["total_repos"],
        "version": KG["metadata"]["version"],
        "layers": KG["metadata"]["total_layers"],
    }

# ── MCP Tool definitions ───────────────────────────────────────────────────────
MCP_TOOLS = [
    {
        "name": "search_repos",
        "description": "Search repositories by name, topic, language, or layer. Returns matching repos with metadata.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "Search query (name, topic, or language)"},
                "layer": {"type": "integer", "description": "Filter by architecture layer (1-7)", "minimum": 1, "maximum": 7},
                "language": {"type": "string", "description": "Filter by programming language"},
                "limit": {"type": "integer", "description": "Max results to return", "default": 10},
            },
            "required": ["query"],
        },
    },
    {
        "name": "get_repo_details",
        "description": "Get detailed information about a specific repository including dependencies and relationships.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "repo_name": {"type": "string", "description": "Repository name to look up"},
            },
            "required": ["repo_name"],
        },
    },
    {
        "name": "get_architecture",
        "description": "Get the full 7-layer architecture overview with all repositories organized by layer.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "layer_id": {"type": "integer", "description": "Optional: get details for a specific layer (1-7)", "minimum": 1, "maximum": 7},
            },
        },
    },
    {
        "name": "get_dependencies",
        "description": "Get dependency graph for a repository — what it depends on and what depends on it.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "repo_name": {"type": "string", "description": "Repository name to get dependencies for"},
                "direction": {"type": "string", "enum": ["upstream", "downstream", "both"], "default": "both"},
            },
            "required": ["repo_name"],
        },
    },
    {
        "name": "get_stats",
        "description": "Get aggregate statistics about the codebase: language distribution, layer sizes, star counts, topic frequency.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "metric": {"type": "string", "enum": ["languages", "layers", "stars", "topics", "all"], "default": "all"},
            },
        },
    },
    {
        "name": "find_related",
        "description": "Find repositories related to a given repo based on shared topics, language, layer proximity, or dependency graph.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "repo_name": {"type": "string", "description": "Repository name to find related repos for"},
                "strategy": {"type": "string", "enum": ["topics", "language", "layer", "dependencies", "all"], "default": "all"},
                "limit": {"type": "integer", "description": "Max results to return", "default": 5},
            },
            "required": ["repo_name"],
        },
    },
]

# ── MCP endpoints ──────────────────────────────────────────────────────────────
@app.get("/mcp/tools")
async def list_tools():
    """Return all 6 MCP tool definitions."""
    return {
        "tools": MCP_TOOLS,
        "count": len(MCP_TOOLS),
        "server": "evolva-codebase-memory-mcp",
        "version": "1.0.0",
    }

@app.post("/mcp/call")
async def call_tool(body: Dict[str, Any]):
    """Execute an MCP tool call."""
    tool_name = body.get("name")
    args = body.get("arguments", {})

    if tool_name == "search_repos":
        return await _search_repos(args)
    elif tool_name == "get_repo_details":
        return await _get_repo_details(args)
    elif tool_name == "get_architecture":
        return await _get_architecture(args)
    elif tool_name == "get_dependencies":
        return await _get_dependencies(args)
    elif tool_name == "get_stats":
        return await _get_stats(args)
    elif tool_name == "find_related":
        return await _find_related(args)
    else:
        raise HTTPException(status_code=404, detail=f"Tool '{tool_name}' not found")

# ── REST API endpoints ─────────────────────────────────────────────────────────
@app.get("/v1/repos")
async def list_repos(
    layer: Optional[int] = Query(None, ge=1, le=7),
    language: Optional[str] = None,
    topic: Optional[str] = None,
):
    """Return all 54 repositories with optional filters."""
    repos = get_repos_list()
    if layer:
        repos = [r for r in repos if r["layer"] == layer]
    if language:
        repos = [r for r in repos if r["language"].lower() == language.lower()]
    if topic:
        repos = [r for r in repos if topic.lower() in [t.lower() for t in r.get("topics", [])]]
    return {
        "repos": repos,
        "count": len(repos),
        "total": KG["metadata"]["total_repos"],
    }

@app.get("/v1/repos/{repo_name}")
async def get_repo(repo_name: str):
    """Get a specific repository by name."""
    repos = get_repos_list()
    repo = next((r for r in repos if r["name"].lower() == repo_name.lower()), None)
    if not repo:
        raise HTTPException(status_code=404, detail=f"Repository '{repo_name}' not found")
    # Enrich with relationships
    relationships = KG.get("relationships", [])
    upstream = [r["to"] for r in relationships if r["from"] == repo["name"]]
    downstream = [r["from"] for r in relationships if r["to"] == repo["name"]]
    return {**repo, "depends_on": upstream, "depended_by": downstream}

@app.get("/v1/architecture")
async def get_architecture_endpoint(layer_id: Optional[int] = Query(None, ge=1, le=7)):
    """Return the 7-layer architecture."""
    layers = KG["architecture"]["layers"]
    if layer_id:
        layer = next((l for l in layers if l["id"] == layer_id), None)
        if not layer:
            raise HTTPException(status_code=404, detail=f"Layer {layer_id} not found")
        return layer
    return {
        "layers": layers,
        "total_layers": len(layers),
        "total_repos": KG["metadata"]["total_repos"],
    }

@app.get("/v1/stats")
async def get_stats_endpoint():
    """Return aggregate statistics."""
    repos = get_repos_list()
    languages: Dict[str, int] = {}
    topics: Dict[str, int] = {}
    layer_counts: Dict[int, int] = {}
    total_stars = 0

    for repo in repos:
        lang = repo.get("language", "Unknown")
        languages[lang] = languages.get(lang, 0) + 1
        for t in repo.get("topics", []):
            topics[t] = topics.get(t, 0) + 1
        layer = repo.get("layer", 0)
        layer_counts[layer] = layer_counts.get(layer, 0) + 1
        total_stars += repo.get("stars", 0)

    return {
        "total_repos": len(repos),
        "total_stars": total_stars,
        "languages": languages,
        "layer_distribution": layer_counts,
        "top_topics": dict(sorted(topics.items(), key=lambda x: x[1], reverse=True)[:20]),
    }

@app.get("/v1/relationships")
async def get_relationships():
    """Return all repository relationships."""
    return {
        "relationships": KG.get("relationships", []),
        "count": len(KG.get("relationships", [])),
    }

# ── Tool implementations ───────────────────────────────────────────────────────
async def _search_repos(args: Dict[str, Any]) -> Dict[str, Any]:
    query = args.get("query", "").lower()
    layer = args.get("layer")
    language = args.get("language", "").lower()
    limit = args.get("limit", 10)
    repos = get_repos_list()
    results = []
    for repo in repos:
        score = 0
        if query in repo["name"].lower():
            score += 3
        if query in repo.get("description", "").lower():
            score += 2
        if any(query in t.lower() for t in repo.get("topics", [])):
            score += 2
        if layer and repo["layer"] == layer:
            score += 1
        if language and repo.get("language", "").lower() == language:
            score += 1
        if score > 0:
            results.append({**repo, "_score": score})
    results.sort(key=lambda x: x["_score"], reverse=True)
    return {"results": results[:limit], "count": len(results[:limit]), "query": query}

async def _get_repo_details(args: Dict[str, Any]) -> Dict[str, Any]:
    repo_name = args.get("repo_name", "")
    repos = get_repos_list()
    repo = next((r for r in repos if r["name"].lower() == repo_name.lower()), None)
    if not repo:
        return {"error": f"Repository '{repo_name}' not found", "available_repos": [r["name"] for r in repos[:10]]}
    relationships = KG.get("relationships", [])
    upstream = [r["to"] for r in relationships if r["from"] == repo["name"]]
    downstream = [r["from"] for r in relationships if r["to"] == repo["name"]]
    layer_info = next((l for l in KG["architecture"]["layers"] if l["id"] == repo["layer"]), {})
    return {**repo, "depends_on": upstream, "depended_by": downstream, "layer_info": layer_info}

async def _get_architecture(args: Dict[str, Any]) -> Dict[str, Any]:
    layer_id = args.get("layer_id")
    layers = KG["architecture"]["layers"]
    if layer_id:
        layer = next((l for l in layers if l["id"] == layer_id), None)
        if not layer:
            return {"error": f"Layer {layer_id} not found"}
        repos_in_layer = [r for r in get_repos_list() if r["layer"] == layer_id]
        return {**layer, "repositories": repos_in_layer}
    return {"layers": layers, "total_layers": len(layers), "total_repos": KG["metadata"]["total_repos"]}

async def _get_dependencies(args: Dict[str, Any]) -> Dict[str, Any]:
    repo_name = args.get("repo_name", "")
    direction = args.get("direction", "both")
    relationships = KG.get("relationships", [])
    result: Dict[str, Any] = {"repo": repo_name}
    if direction in ("upstream", "both"):
        result["depends_on"] = [r["to"] for r in relationships if r["from"] == repo_name]
    if direction in ("downstream", "both"):
        result["depended_by"] = [r["from"] for r in relationships if r["to"] == repo_name]
    return result

async def _get_stats(args: Dict[str, Any]) -> Dict[str, Any]:
    metric = args.get("metric", "all")
    repos = get_repos_list()
    languages: Dict[str, int] = {}
    topics: Dict[str, int] = {}
    layer_counts: Dict[int, int] = {}
    total_stars = 0
    for repo in repos:
        lang = repo.get("language", "Unknown")
        languages[lang] = languages.get(lang, 0) + 1
        for t in repo.get("topics", []):
            topics[t] = topics.get(t, 0) + 1
        layer = repo.get("layer", 0)
        layer_counts[layer] = layer_counts.get(layer, 0) + 1
        total_stars += repo.get("stars", 0)
    if metric == "languages":
        return {"languages": languages}
    elif metric == "layers":
        return {"layer_distribution": layer_counts}
    elif metric == "stars":
        return {"total_stars": total_stars, "avg_stars": round(total_stars / len(repos), 1)}
    elif metric == "topics":
        return {"topics": dict(sorted(topics.items(), key=lambda x: x[1], reverse=True)[:20])}
    return {
        "total_repos": len(repos),
        "total_stars": total_stars,
        "avg_stars": round(total_stars / len(repos), 1),
        "languages": languages,
        "layer_distribution": layer_counts,
        "top_topics": dict(sorted(topics.items(), key=lambda x: x[1], reverse=True)[:20]),
    }

async def _find_related(args: Dict[str, Any]) -> Dict[str, Any]:
    repo_name = args.get("repo_name", "")
    strategy = args.get("strategy", "all")
    limit = args.get("limit", 5)
    repos = get_repos_list()
    target = next((r for r in repos if r["name"].lower() == repo_name.lower()), None)
    if not target:
        return {"error": f"Repository '{repo_name}' not found"}
    scores: Dict[str, float] = {}
    for repo in repos:
        if repo["name"] == target["name"]:
            continue
        score = 0.0
        if strategy in ("topics", "all"):
            shared = set(target.get("topics", [])) & set(repo.get("topics", []))
            score += len(shared) * 2
        if strategy in ("language", "all"):
            if repo.get("language") == target.get("language"):
                score += 1
        if strategy in ("layer", "all"):
            layer_diff = abs(repo["layer"] - target["layer"])
            if layer_diff == 0:
                score += 2
            elif layer_diff == 1:
                score += 1
        if strategy in ("dependencies", "all"):
            relationships = KG.get("relationships", [])
            if any(r["from"] == target["name"] and r["to"] == repo["name"] for r in relationships):
                score += 3
            if any(r["from"] == repo["name"] and r["to"] == target["name"] for r in relationships):
                score += 3
        if score > 0:
            scores[repo["name"]] = score
    ranked = sorted(scores.items(), key=lambda x: x[1], reverse=True)[:limit]
    related = []
    for name, score in ranked:
        repo = next(r for r in repos if r["name"] == name)
        related.append({**repo, "_relevance_score": score})
    return {"repo": repo_name, "related": related, "strategy": strategy}

# ── Entry point ────────────────────────────────────────────────────────────────
if __name__ == "__main__":
    import uvicorn
    port = int(os.environ.get("PORT", 7860))
    uvicorn.run(app, host="0.0.0.0", port=port)