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

graph-theory/mcp_output/mcp_plugin/mcp_service.py

@@ -1,131 +1,582 @@
+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")
 
-@mcp.tool(name="list_graph_methods", description="List all available graph methods")
-def list_graph_methods():
+# 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:
     """
-    List all available methods in the graph-theory library.
+    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.
 
     Returns:
-        dict: A dictionary containing the list of methods and their descriptions.
+        dict: Number of nodes and path.
     """
     try:
-        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}
+        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,
+        }
     except Exception as e:
-        return {"success": False, "error": str(e)}
+        return {"success": False, "result": None, "error": str(e)}
 
-@mcp.tool(name="create_graph", description="Create a new graph")
-def create_graph():
+
+@mcp.tool(name="depth_first_search")
+def depth_first_search(graph_id: str, start: str, end: str) -> dict:
     """
-    Create a new graph instance.
+    Find a path using DFS.
+
+    Parameters:
+        graph_id (str): ID of the graph.
+        start (str): Start node.
+        end (str): End node.
 
     Returns:
-        dict: A dictionary with the graph instance.
+        dict: Path as list of nodes.
     """
     try:
-        from graph.base import BasicGraph
-        graph = BasicGraph()
-        return {"success": True, "graph": graph}
+        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,
+        }
     except Exception as e:
-        return {"success": False, "error": str(e)}
+        return {"success": False, "result": None, "error": str(e)}
 
-@mcp.tool(name="add_node", description="Add a node to the graph")
-def add_node(graph, node, obj=None):
+
+@mcp.tool(name="maximum_flow")
+def maximum_flow(graph_id: str, source: str, sink: str) -> dict:
     """
-    Add a node to the graph.
+    Calculate maximum flow from source to sink.
 
-    Args:
-        graph (BasicGraph): The graph instance.
-        node (hashable): The node to add.
-        obj (optional): An object to associate with the node.
+    Parameters:
+        graph_id (str): ID of the graph.
+        source (str): Source node.
+        sink (str): Sink node.
 
     Returns:
-        dict: A dictionary with the success status.
+        dict: Maximum flow value and flow graph.
     """
     try:
-        graph.add_node(node, obj)
-        return {"success": True}
+        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,
+        }
     except Exception as e:
-        return {"success": False, "error": str(e)}
+        return {"success": False, "result": None, "error": str(e)}
+
 
-@mcp.tool(name="add_edge", description="Add an edge to the graph")
-def add_edge(graph, node1, node2, value=1, bidirectional=False):
+@mcp.tool(name="solve_tsp")
+def solve_tsp(graph_id: str, method: str = "2023") -> dict:
     """
-    Add an edge to the graph.
+    Solve the Traveling Salesman Problem.
 
-    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.
+    Parameters:
+        graph_id (str): ID of the graph.
+        method (str): Algorithm to use ('greedy', 'bnb', '2023').
 
     Returns:
-        dict: A dictionary with the success status.
+        dict: Tour length and path.
     """
     try:
-        graph.add_edge(node1, node2, value, bidirectional)
-        return {"success": True}
+        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,
+        }
     except Exception as e:
-        return {"success": False, "error": str(e)}
+        return {"success": False, "result": None, "error": str(e)}
 
-@mcp.tool(name="shortest_path", description="Find the shortest path between two nodes")
-def shortest_path(graph, start, end):
+
+@mcp.tool(name="has_cycles")
+def has_cycles(graph_id: str) -> dict:
     """
-    Find the shortest path between two nodes in the graph.
+    Check if the graph has cycles.
 
-    Args:
-        graph (BasicGraph): The graph instance.
-        start (hashable): The starting node.
-        end (hashable): The ending node.
+    Parameters:
+        graph_id (str): ID of the graph.
 
     Returns:
-        dict: A dictionary with the shortest path and its distance.
+        dict: Boolean indicating if cycles exist.
     """
     try:
-        distance, path = graph.shortest_path(start, end)
-        return {"success": True, "distance": distance, "path": path}
+        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,
+        }
     except Exception as e:
-        return {"success": False, "error": str(e)}
+        return {"success": False, "result": None, "error": str(e)}
+
 
-@mcp.tool(name="visualize_graph", description="Visualize the graph")
-def visualize_graph(graph):
+@mcp.tool(name="components")
+def components(graph_id: str) -> dict:
     """
-    Visualize the graph using matplotlib.
+    Find connected components in the graph.
 
-    Args:
-        graph (BasicGraph): The graph instance.
+    Parameters:
+        graph_id (str): ID of the graph.
 
     Returns:
-        dict: A dictionary with the visualization status.
+        dict: List of components (each component is a set of nodes).
     """
     try:
-        from graph.visuals import plot_graph
-        plot_graph(graph)
-        return {"success": True}
+        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,
+        }
     except Exception as e:
-        return {"success": False, "error": str(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
 
-# Add more tools here following the same pattern
 
-# Run the MCP service
 if __name__ == "__main__":
-    mcp.run()
+    mcp.run(transport="http", host="0.0.0.0", port=8000)