Graphistry MCP

Instructions

Identify patterns, communities, and anomalies within graphs. Runs all supported analyses and returns a combined report.

Args:
    graph_id: ID of the graph to analyze
    ctx: MCP context for progress reporting

Returns:
    Dictionary with results from all analyses that succeeded. Keys may include:
        - degree_centrality
        - betweenness_centrality
        - closeness_centrality
        - communities (if community detection is available)
        - shortest_path (if path finding is possible)
        - path_length
        - anomalies (if anomaly detection is available)
        - errors (dict of analysis_type -> error message)

Implementation Reference

• src/graphistry_mcp_server/server.py:244-331 (handler)

  The detect_patterns tool handler. This async function performs graph pattern detection using NetworkX for centrality measures, community detection (Louvain), shortest paths, and simple anomaly detection (degree 1 nodes). It is registered via the @mcp.tool() decorator. The function signature and docstring define the input schema (graph_id: str, optional ctx) and output (dict of results).

  @mcp.tool()
  async def detect_patterns(graph_id: str, ctx: Optional[Context] = None) -> Dict[str, Any]:
      """
      Identify patterns, communities, and anomalies within graphs. Runs all supported analyses and returns a combined report.
  
      Args:
          graph_id: ID of the graph to analyze
          ctx: MCP context for progress reporting
  
      Returns:
          Dictionary with results from all analyses that succeeded. Keys may include:
              - degree_centrality
              - betweenness_centrality
              - closeness_centrality
              - communities (if community detection is available)
              - shortest_path (if path finding is possible)
              - path_length
              - anomalies (if anomaly detection is available)
              - errors (dict of analysis_type -> error message)
  
      """
      try:
          if graph_id not in graph_cache:
              raise ValueError(f"Graph not found: {graph_id}")
  
          if ctx:
              await ctx.info("Starting pattern detection (all analyses)...")
  
          graph_data = graph_cache[graph_id]
          nx_graph = graph_data["nx_graph"]
          edges_df = graph_data["edges_df"]
          source = graph_data["source"]
          destination = graph_data["destination"]
  
          # Convert to NetworkX graph if needed
          if nx_graph is None and edges_df is not None:
              nx_graph = nx.from_pandas_edgelist(edges_df, source=source, target=destination)
  
          if nx_graph is None:
              raise ValueError("Graph data not available for analysis")
  
          results = {}
          errors = {}
  
          # Centrality
          try:
              results["degree_centrality"] = nx.degree_centrality(nx_graph)
              results["betweenness_centrality"] = nx.betweenness_centrality(nx_graph)
              results["closeness_centrality"] = nx.closeness_centrality(nx_graph)
          except Exception as e:
              errors["centrality"] = str(e)
  
          # Community detection
          try:
              import community as community_louvain
              partition = community_louvain.best_partition(nx_graph)
              results["communities"] = partition
          except Exception as e:
              errors["community_detection"] = str(e)
  
          # Path finding (try between first two nodes if possible)
          try:
              nodes = list(nx_graph.nodes())
              if len(nodes) >= 2:
                  path = nx.shortest_path(nx_graph, nodes[0], nodes[1])
                  results["shortest_path"] = path
                  results["path_length"] = len(path) - 1
          except Exception as e:
              errors["path_finding"] = str(e)
  
          # Anomaly detection (placeholder)
          try:
              # Example: nodes with degree 1 as "anomalies"
              anomalies = [n for n, d in nx_graph.degree() if d == 1]
              results["anomalies"] = anomalies
          except Exception as e:
              errors["anomaly_detection"] = str(e)
  
          if errors:
              results["errors"] = errors
  
          if ctx:
              await ctx.info("Pattern detection complete!")
  
          return results
      except Exception as e:
          logger.error(f"Error in detect_patterns: {e}")
          raise

• src/graphistry_mcp_server/server.py:244-244 (registration)

  The @mcp.tool() decorator registers the detect_patterns function as an MCP tool.

  @mcp.tool()

• src/graphistry_mcp_server/server.py:246-264 (schema)

  Docstring providing detailed input/output schema for the detect_patterns tool.

  """
  Identify patterns, communities, and anomalies within graphs. Runs all supported analyses and returns a combined report.
  
  Args:
      graph_id: ID of the graph to analyze
      ctx: MCP context for progress reporting
  
  Returns:
      Dictionary with results from all analyses that succeeded. Keys may include:
          - degree_centrality
          - betweenness_centrality
          - closeness_centrality
          - communities (if community detection is available)
          - shortest_path (if path finding is possible)
          - path_length
          - anomalies (if anomaly detection is available)
          - errors (dict of analysis_type -> error message)
  
  """