mcp/graph_metrics.py

# mcp/graph_metrics.py
from typing import List, Dict, Tuple
import networkx as nx

def _edge_endpoints(e: Dict) -> Tuple[str,str] | None:
    src = e.get("source") or e.get("from")
    dst = e.get("target") or e.get("to")
    return (src,dst) if src and dst else None

def build_nx(nodes: List[Dict], edges: List[Dict]) -> nx.Graph:
    G = nx.Graph()
    for n in nodes:
        nid = n.get("id")
        if nid:
            G.add_node(nid, label=n.get("label", nid))
    for e in edges:
        pts = _edge_endpoints(e)
        if pts:
            G.add_edge(*pts)
    return G

def get_top_hubs(G: nx.Graph, k: int = 5):
    dc = nx.degree_centrality(G)
    return sorted(dc.items(), key=lambda x: x[1], reverse=True)[:k]

def get_density(G: nx.Graph) -> float:
    return nx.density(G)