NebulaGraph MCP Server

Instructions

Implementation Reference

• src/nebulagraph_mcp_server/server.py:285-296 (handler)

  The handler function for the 'find_neighbors' MCP tool, decorated with @mcp.tool() for registration. It receives parameters and delegates to the helper resource function.

  @mcp.tool()
  def find_neighbors(vertex: str, space: str, depth: int = 1) -> str:
      """Find the neighbors of the specified vertex
      Args:
          vertex: The vertex ID to query
          space: The space to use
          depth: The depth of the query, default is 1
      Returns:
          The neighbors of the specified vertex
      """
      return get_neighbors_resource(space, vertex, depth)

• src/nebulagraph_mcp_server/server.py:242-283 (helper)

  Core helper function implementing the neighbor query logic using NebulaGraph nGQL MATCH query to find vertices and edges connected to the given vertex up to specified depth.

  @mcp.resource("neighbors://space/{space}/vertex/{vertex}/depth/{depth}")
  def get_neighbors_resource(space: str, vertex: str, depth: int) -> str:
      """Get the neighbors of the specified vertex
      Args:
          space: The space to use
          vertex: The vertex ID to query
          depth: The depth of the query
      Returns:
          The neighbors of the specified vertex
      """
      pool = get_connection_pool()
      session = pool.get_session(
          os.getenv("NEBULA_USER", "root"), os.getenv("NEBULA_PASSWORD", "nebula")
      )
  
      try:
          session.execute(f"USE {space}")
  
          query = f"""
          MATCH (u)-[e*1..{depth}]-(v)
          WHERE id(u) == "{vertex}"
          RETURN DISTINCT v, e
          """
  
          result = session.execute(query)
          if result.is_succeeded():
              if result.row_size() > 0:
                  output = f"Vertex {vertex} neighbors (depth {depth}):\n\n"
                  for i in range(result.row_size()):
                      row = result.row_values(i)
                      neighbor_vertex = row[0]
                      edges = row[1]
                      output += (
                          f"Neighbor Vertex:\n{neighbor_vertex}\nEdges:\n{edges}\n\n"
                      )
                  return output
              return f"No neighbors found for vertex {vertex}"
          else:
              return f"Query failed: {result.error_msg()}"
      finally:
          session.release()