Update mcp/graph_metrics.py

mcp/graph_metrics.py

@@ -1,20 +1,63 @@
+# mcp/graph_metrics.py
 """
-Minimal NetworkX helpers for MedGenesis graphs.
+Tiny NetworkX helpers with forgiving edge parsing.
+
+Why this exists
+---------------
+Your Streamlit-agraph edges carry keys ``source/target`` while other
+toolchains (PyVis, cytoscape, etc.) use ``from/to``.  A naïve
+``e["target"]`` lookup therefore raises **KeyError**.  The functions below
+accept either spelling and silently skip malformed edges.
 """
-import networkx as nx
+
+from __future__ import annotations
 from typing import List, Dict, Tuple
 
+import networkx as nx
+
+
+# ── private -----------------------------------------------------------------
+def _endpoints(edge: Dict) -> Tuple[str, str] | None:
+    """Return (src,dst) tuple or *None* if either end is missing."""
+    src = edge.get("source") or edge.get("from")
+    dst = edge.get("target") or edge.get("to")
+    return (src, dst) if src and dst else None
+
+
+# ── public helpers ----------------------------------------------------------
 def build_nx(nodes: List[Dict], edges: List[Dict]) -> nx.Graph:
+    """
+    Convert generic node/edge dictionaries into a NetworkX ``Graph``.
+
+    •   Nodes are added with an attribute ``label`` (fallback: id).  
+    •   Edges are added only when both src & dst exist.  
+    •   **No KeyError** even if edge-schema is mixed.
+    """
     G = nx.Graph()
+
+    # nodes
     for n in nodes:
-        G.add_node(n["id"])
+        try:
+            G.add_node(n["id"], label=n.get("label", n["id"]))
+        except KeyError:
+            # ignore malformed node dicts
+            continue
+
+    # edges
     for e in edges:
-        G.add_edge(e["source"], e["target"])
+        ends = _endpoints(e)
+        if ends:
+            G.add_edge(*ends)
+
     return G
 
-def get_density(G: nx.Graph) -> float:
-    return nx.density(G)
 
-def get_top_hubs(G: nx.Graph, k: int = 5) -> List[Tuple[str, float]]:
+def get_top_hubs(G: nx.Graph, k: int = 5):
+    """Return *k* nodes with highest degree-centrality."""
     dc = nx.degree_centrality(G)
     return sorted(dc.items(), key=lambda x: x[1], reverse=True)[:k]
+
+
+def get_density(G: nx.Graph) -> float:
+    """Graph density ∈ [0,1]."""
+    return nx.density(G)