Update graph-theory/mcp_output/mcp_plugin/mcp_service.py

graph-theory/mcp_output/mcp_plugin/mcp_service.py

@@ -1,582 +1,131 @@
-import os
-import sys
-from typing import Dict, Any, List, Optional, Tuple
-
-source_path = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), "source")
-if source_path not in sys.path:
-    sys.path.insert(0, source_path)
-
 from fastmcp import FastMCP
-from graph import Graph, Graph3D
 
+# Create the FastMCP service application
 mcp = FastMCP("graph_theory_service")
 
-# Store graphs by ID
-_graphs: Dict[str, Graph] = {}
-_graph3d: Dict[str, Graph3D] = {}
-
-
-@mcp.tool(name="get_library_info")
-def get_library_info() -> dict:
-    """
-    Get information about the graph-theory library.
-
-    Returns:
-        dict: Version and available functionality.
-    """
-    try:
-        from graph.version import __version__
-        
-        return {
-            "success": True,
-            "result": {
-                "library": "graph-theory",
-                "version": __version__,
-                "graph_types": ["Graph", "Graph3D"],
-                "features": [
-                    "shortest_path", "breadth_first_search", "depth_first_search",
-                    "maximum_flow", "minimum_cost_flow", "traveling_salesman",
-                    "topological_sort", "cycle_detection", "components",
-                    "critical_path", "adjacency_matrix"
-                ],
-            },
-            "error": None,
-        }
-    except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
-
-@mcp.tool(name="create_graph")
-def create_graph(
-    graph_id: str,
-    from_dict: Optional[Dict[str, Dict[str, float]]] = None,
-    from_list: Optional[List[Tuple]] = None
-) -> dict:
-    """
-    Create a new graph.
-
-    Parameters:
-        graph_id (str): Unique identifier for the graph.
-        from_dict (Optional[Dict]): Dictionary representation {node: {neighbor: distance}}.
-        from_list (Optional[List[Tuple]]): List of edges as [(n1, n2, distance), ...].
-
-    Returns:
-        dict: Success status and graph information.
-    """
-    try:
-        if graph_id in _graphs:
-            return {"success": False, "result": None, "error": f"Graph '{graph_id}' already exists"}
-        
-        g = Graph(from_dict=from_dict, from_list=from_list)
-        _graphs[graph_id] = g
-        
-        return {
-            "success": True,
-            "result": {
-                "graph_id": graph_id,
-                "nodes": list(g.nodes()),
-                "edges": list(g.edges()),
-                "num_nodes": len(list(g.nodes())),
-                "num_edges": len(list(g.edges())),
-            },
-            "error": None,
-        }
-    except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
-
-@mcp.tool(name="add_node")
-def add_node(graph_id: str, node_id: str, obj: Any = None) -> dict:
-    """
-    Add a node to a graph.
-
-    Parameters:
-        graph_id (str): ID of the graph.
-        node_id (str): Unique identifier for the node.
-        obj (Any): Optional object associated with the node.
-
-    Returns:
-        dict: Success status.
-    """
-    try:
-        if graph_id not in _graphs:
-            return {"success": False, "result": None, "error": f"Graph '{graph_id}' not found"}
-        
-        g = _graphs[graph_id]
-        g.add_node(node_id, obj)
-        
-        return {
-            "success": True,
-            "result": {"graph_id": graph_id, "node_id": node_id, "message": "Node added"},
-            "error": None,
-        }
-    except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
-
-@mcp.tool(name="add_edge")
-def add_edge(
-    graph_id: str,
-    node1: str,
-    node2: str,
-    distance: float = 1.0,
-    bidirectional: bool = False
-) -> dict:
+@mcp.tool(name="list_graph_methods", description="List all available graph methods")
+def list_graph_methods():
     """
-    Add an edge to a graph.
-
-    Parameters:
-        graph_id (str): ID of the graph.
-        node1 (str): Source node.
-        node2 (str): Destination node.
-        distance (float): Edge weight/distance.
-        bidirectional (bool): If True, creates edge in both directions.
-
-    Returns:
-        dict: Success status.
-    """
-    try:
-        if graph_id not in _graphs:
-            return {"success": False, "result": None, "error": f"Graph '{graph_id}' not found"}
-        
-        g = _graphs[graph_id]
-        g.add_edge(node1, node2, distance, bidirectional)
-        
-        return {
-            "success": True,
-            "result": {
-                "graph_id": graph_id,
-                "edge": (node1, node2),
-                "distance": distance,
-                "bidirectional": bidirectional,
-            },
-            "error": None,
-        }
-    except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
-
-@mcp.tool(name="shortest_path")
-def shortest_path(graph_id: str, start: str, end: str) -> dict:
-    """
-    Find the shortest path between two nodes.
-
-    Parameters:
-        graph_id (str): ID of the graph.
-        start (str): Start node.
-        end (str): End node.
-
-    Returns:
-        dict: Distance and path as list of nodes.
-    """
-    try:
-        if graph_id not in _graphs:
-            return {"success": False, "result": None, "error": f"Graph '{graph_id}' not found"}
-        
-        g = _graphs[graph_id]
-        distance, path = g.shortest_path(start, end)
-        
-        return {
-            "success": True,
-            "result": {
-                "distance": distance,
-                "path": path,
-                "num_nodes": len(path),
-            },
-            "error": None,
-        }
-    except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
-
-@mcp.tool(name="breadth_first_search")
-def breadth_first_search(graph_id: str, start: str, end: str) -> dict:
-    """
-    Find path with fewest nodes using BFS.
-
-    Parameters:
-        graph_id (str): ID of the graph.
-        start (str): Start node.
-        end (str): End node.
+    List all available methods in the graph-theory library.
 
     Returns:
-        dict: Number of nodes and path.
+        dict: A dictionary containing the list of methods and their descriptions.
     """
     try:
-        if graph_id not in _graphs:
-            return {"success": False, "result": None, "error": f"Graph '{graph_id}' not found"}
-        
-        g = _graphs[graph_id]
-        nodes, path = g.breadth_first_search(start, end)
-        
-        return {
-            "success": True,
-            "result": {
-                "num_nodes": nodes,
-                "path": path,
-            },
-            "error": None,
-        }
+        methods = [
+            "add_node",
+            "add_edge",
+            "del_node",
+            "del_edge",
+            "shortest_path",
+            "breadth_first_search",
+            "depth_first_search",
+            "maximum_flow",
+            "topological_sort",
+            "critical_path",
+            "network_size",
+            "distance_map",
+            "degree_of_separation",
+            "visualize_graph",
+        ]
+        return {"success": True, "methods": methods}
     except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
+        return {"success": False, "error": str(e)}
 
-
-@mcp.tool(name="depth_first_search")
-def depth_first_search(graph_id: str, start: str, end: str) -> dict:
+@mcp.tool(name="create_graph", description="Create a new graph")
+def create_graph():
     """
-    Find a path using DFS.
-
-    Parameters:
-        graph_id (str): ID of the graph.
-        start (str): Start node.
-        end (str): End node.
+    Create a new graph instance.
 
     Returns:
-        dict: Path as list of nodes.
+        dict: A dictionary with the graph instance.
     """
     try:
-        if graph_id not in _graphs:
-            return {"success": False, "result": None, "error": f"Graph '{graph_id}' not found"}
-        
-        g = _graphs[graph_id]
-        path = g.depth_first_search(start, end)
-        
-        return {
-            "success": True,
-            "result": {"path": path},
-            "error": None,
-        }
+        from graph.base import BasicGraph
+        graph = BasicGraph()
+        return {"success": True, "graph": graph}
     except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
+        return {"success": False, "error": str(e)}
 
-
-@mcp.tool(name="maximum_flow")
-def maximum_flow(graph_id: str, source: str, sink: str) -> dict:
+@mcp.tool(name="add_node", description="Add a node to the graph")
+def add_node(graph, node, obj=None):
     """
-    Calculate maximum flow from source to sink.
+    Add a node to the graph.
 
-    Parameters:
-        graph_id (str): ID of the graph.
-        source (str): Source node.
-        sink (str): Sink node.
+    Args:
+        graph (BasicGraph): The graph instance.
+        node (hashable): The node to add.
+        obj (optional): An object to associate with the node.
 
     Returns:
-        dict: Maximum flow value and flow graph.
+        dict: A dictionary with the success status.
     """
     try:
-        if graph_id not in _graphs:
-            return {"success": False, "result": None, "error": f"Graph '{graph_id}' not found"}
-        
-        g = _graphs[graph_id]
-        flow_value, flow_graph = g.maximum_flow(source, sink)
-        
-        return {
-            "success": True,
-            "result": {
-                "max_flow": flow_value,
-                "flow_edges": list(flow_graph.edges()) if flow_graph else [],
-            },
-            "error": None,
-        }
+        graph.add_node(node, obj)
+        return {"success": True}
     except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
+        return {"success": False, "error": str(e)}
 
-@mcp.tool(name="solve_tsp")
-def solve_tsp(graph_id: str, method: str = "2023") -> dict:
+@mcp.tool(name="add_edge", description="Add an edge to the graph")
+def add_edge(graph, node1, node2, value=1, bidirectional=False):
     """
-    Solve the Traveling Salesman Problem.
+    Add an edge to the graph.
 
-    Parameters:
-        graph_id (str): ID of the graph.
-        method (str): Algorithm to use ('greedy', 'bnb', '2023').
+    Args:
+        graph (BasicGraph): The graph instance.
+        node1 (hashable): The starting node.
+        node2 (hashable): The ending node.
+        value (int or float): The weight of the edge.
+        bidirectional (bool): Whether the edge is bidirectional.
 
     Returns:
-        dict: Tour length and path.
+        dict: A dictionary with the success status.
     """
     try:
-        if graph_id not in _graphs:
-            return {"success": False, "result": None, "error": f"Graph '{graph_id}' not found"}
-        
-        g = _graphs[graph_id]
-        tour_length, path = g.solve_tsp(method=method)
-        
-        return {
-            "success": True,
-            "result": {
-                "tour_length": tour_length,
-                "path": path,
-                "method": method,
-            },
-            "error": None,
-        }
+        graph.add_edge(node1, node2, value, bidirectional)
+        return {"success": True}
     except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
+        return {"success": False, "error": str(e)}
 
-
-@mcp.tool(name="has_cycles")
-def has_cycles(graph_id: str) -> dict:
+@mcp.tool(name="shortest_path", description="Find the shortest path between two nodes")
+def shortest_path(graph, start, end):
     """
-    Check if the graph has cycles.
+    Find the shortest path between two nodes in the graph.
 
-    Parameters:
-        graph_id (str): ID of the graph.
+    Args:
+        graph (BasicGraph): The graph instance.
+        start (hashable): The starting node.
+        end (hashable): The ending node.
 
     Returns:
-        dict: Boolean indicating if cycles exist.
+        dict: A dictionary with the shortest path and its distance.
     """
     try:
-        if graph_id not in _graphs:
-            return {"success": False, "result": None, "error": f"Graph '{graph_id}' not found"}
-        
-        g = _graphs[graph_id]
-        has_cycle = g.has_cycles()
-        
-        return {
-            "success": True,
-            "result": {"has_cycles": has_cycle},
-            "error": None,
-        }
+        distance, path = graph.shortest_path(start, end)
+        return {"success": True, "distance": distance, "path": path}
     except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
+        return {"success": False, "error": str(e)}
 
-@mcp.tool(name="components")
-def components(graph_id: str) -> dict:
+@mcp.tool(name="visualize_graph", description="Visualize the graph")
+def visualize_graph(graph):
     """
-    Find connected components in the graph.
+    Visualize the graph using matplotlib.
 
-    Parameters:
-        graph_id (str): ID of the graph.
+    Args:
+        graph (BasicGraph): The graph instance.
 
     Returns:
-        dict: List of components (each component is a set of nodes).
+        dict: A dictionary with the visualization status.
     """
     try:
-        if graph_id not in _graphs:
-            return {"success": False, "result": None, "error": f"Graph '{graph_id}' not found"}
-        
-        g = _graphs[graph_id]
-        comps = g.components()
-        
-        return {
-            "success": True,
-            "result": {
-                "num_components": len(comps),
-                "components": [list(comp) for comp in comps],
-            },
-            "error": None,
-        }
+        from graph.visuals import plot_graph
+        plot_graph(graph)
+        return {"success": True}
     except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
-
-@mcp.tool(name="topological_sort")
-def topological_sort(graph_id: str) -> dict:
-    """
-    Perform topological sort on a DAG.
-
-    Parameters:
-        graph_id (str): ID of the graph.
-
-    Returns:
-        dict: Topologically sorted list of nodes.
-    """
-    try:
-        if graph_id not in _graphs:
-            return {"success": False, "result": None, "error": f"Graph '{graph_id}' not found"}
-        
-        g = _graphs[graph_id]
-        sorted_nodes = list(g.topological_sort())
-        
-        return {
-            "success": True,
-            "result": {"sorted_nodes": sorted_nodes},
-            "error": None,
-        }
-    except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
-
-@mcp.tool(name="critical_path")
-def critical_path(graph_id: str) -> dict:
-    """
-    Find the critical path in a project network.
-
-    Parameters:
-        graph_id (str): ID of the graph.
-
-    Returns:
-        dict: Critical path information.
-    """
-    try:
-        if graph_id not in _graphs:
-            return {"success": False, "result": None, "error": f"Graph '{graph_id}' not found"}
-        
-        g = _graphs[graph_id]
-        result = g.critical_path()
-        
-        return {
-            "success": True,
-            "result": {"critical_path": result},
-            "error": None,
-        }
-    except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
-
-@mcp.tool(name="adjacency_matrix")
-def adjacency_matrix(graph_id: str) -> dict:
-    """
-    Get the adjacency matrix representation of the graph.
-
-    Parameters:
-        graph_id (str): ID of the graph.
-
-    Returns:
-        dict: Adjacency matrix as dictionary.
-    """
-    try:
-        if graph_id not in _graphs:
-            return {"success": False, "result": None, "error": f"Graph '{graph_id}' not found"}
-        
-        g = _graphs[graph_id]
-        matrix = g.adjacency_matrix()
-        
-        return {
-            "success": True,
-            "result": {"adjacency_matrix": matrix},
-            "error": None,
-        }
-    except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
-
-@mcp.tool(name="is_partite")
-def is_partite(graph_id: str, n: int = 2) -> dict:
-    """
-    Check if graph is n-partite.
-
-    Parameters:
-        graph_id (str): ID of the graph.
-        n (int): Number of partitions to check for.
-
-    Returns:
-        dict: Boolean and partitions if n-partite.
-    """
-    try:
-        if graph_id not in _graphs:
-            return {"success": False, "result": None, "error": f"Graph '{graph_id}' not found"}
-        
-        g = _graphs[graph_id]
-        is_n_partite, partitions = g.is_partite(n)
-        
-        return {
-            "success": True,
-            "result": {
-                "is_partite": is_n_partite,
-                "partitions": partitions,
-                "n": n,
-            },
-            "error": None,
-        }
-    except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
-
-@mcp.tool(name="get_graph_info")
-def get_graph_info(graph_id: str) -> dict:
-    """
-    Get information about a stored graph.
-
-    Parameters:
-        graph_id (str): ID of the graph.
-
-    Returns:
-        dict: Graph statistics and properties.
-    """
-    try:
-        if graph_id not in _graphs:
-            return {"success": False, "result": None, "error": f"Graph '{graph_id}' not found"}
-        
-        g = _graphs[graph_id]
-        nodes = list(g.nodes())
-        edges = list(g.edges())
-        
-        return {
-            "success": True,
-            "result": {
-                "graph_id": graph_id,
-                "num_nodes": len(nodes),
-                "num_edges": len(edges),
-                "nodes": nodes[:20],  # First 20 for brevity
-                "edges": edges[:20],  # First 20 for brevity
-                "has_cycles": g.has_cycles(),
-            },
-            "error": None,
-        }
-    except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
-
-@mcp.tool(name="list_graphs")
-def list_graphs() -> dict:
-    """
-    List all stored graphs.
-
-    Returns:
-        dict: List of graph IDs.
-    """
-    try:
-        return {
-            "success": True,
-            "result": {
-                "graphs": list(_graphs.keys()),
-                "graph3d": list(_graph3d.keys()),
-                "total": len(_graphs) + len(_graph3d),
-            },
-            "error": None,
-        }
-    except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
-
-@mcp.tool(name="delete_graph")
-def delete_graph(graph_id: str) -> dict:
-    """
-    Delete a stored graph.
-
-    Parameters:
-        graph_id (str): ID of the graph to delete.
-
-    Returns:
-        dict: Confirmation of deletion.
-    """
-    try:
-        if graph_id in _graphs:
-            del _graphs[graph_id]
-            return {
-                "success": True,
-                "result": {"message": f"Graph '{graph_id}' deleted"},
-                "error": None,
-            }
-        elif graph_id in _graph3d:
-            del _graph3d[graph_id]
-            return {
-                "success": True,
-                "result": {"message": f"Graph3D '{graph_id}' deleted"},
-                "error": None,
-            }
-        else:
-            return {"success": False, "result": None, "error": f"Graph '{graph_id}' not found"}
-    except Exception as e:
-        return {"success": False, "result": None, "error": str(e)}
-
-
-def create_app():
-    """Create and return FastMCP application instance"""
-    return mcp
+        return {"success": False, "error": str(e)}
 
+# Add more tools here following the same pattern
 
+# Run the MCP service
 if __name__ == "__main__":
-    mcp.run(transport="http", host="0.0.0.0", port=8000)
+    mcp.run()