diff --git a/.gitignore b/.gitignore
index ce100be4..b4d9d1fb 100644
--- a/.gitignore
+++ b/.gitignore
@@ -118,6 +118,9 @@ venv.bak/
 # Rope project settings
 .ropeproject
 
+# PyCharm project settings
+.idea/
+
 # mkdocs documentation
 /site
 
diff --git a/docs/index.rst b/docs/index.rst
index b444aa93..ed5c0679 100644
--- a/docs/index.rst
+++ b/docs/index.rst
@@ -33,6 +33,7 @@ Sections
    api/nucleus/index
    api/nucleus/metrics/index
    api/nucleus/modelci/index
+   api/nucleus/deploy/index
 
 
 Index
diff --git a/nucleus/deploy/README.md b/nucleus/deploy/README.md
new file mode 100644
index 00000000..a4a33c3f
--- /dev/null
+++ b/nucleus/deploy/README.md
@@ -0,0 +1,27 @@
+Currently, Scale Deploy is still being built out, so the contents of this library are subject to change.
+
+# Scale Deploy
+
+Moving an ML model from experiment to production requires significant engineering lift. 
+Scale Deploy provides ML engineers a simple Python interface for turning a local code snippet into a production service.
+A ML engineer needs to call a few functions from Scale's SDK, which quickly spins up a production-ready service.
+The service efficiently utilizes compute resources and automatically scales according to traffic.
+
+# Deploying your model via Scale Deploy
+
+Central to Scale Deploy are the notions of a `ModelBundle` and a `ModelEndpoint`. 
+A `ModelBundle` consists of a trained model as well as the surrounding preprocessing and postprocessing code.
+A `ModelEndpoint` is the compute layer that takes in a `ModelBundle`, and is able to carry out inference requests
+by using the `ModelBundle` to carry out predictions. The `ModelEndpoint` also knows infrastructure-level details,
+such as how many GPUs are needed, what type they are, how much memory, etc. The `ModelEndpoint` automatically handles
+infrastructure level details such as autoscaling and task queueing.
+
+Steps to deploy your model via Scale Deploy:
+
+1. First, you create and upload a `ModelBundle`. 
+
+2. Then, you create a `ModelEndpoint`.
+
+3. Lastly, you make requests to the `ModelEndpoint`.
+
+TODO: link some example colab notebook
diff --git a/nucleus/deploy/__init__.py b/nucleus/deploy/__init__.py
new file mode 100644
index 00000000..546417d9
--- /dev/null
+++ b/nucleus/deploy/__init__.py
@@ -0,0 +1,52 @@
+"""
+
+Moving an ML model from experiment to production requires significant engineering lift.
+Scale Deploy provides ML engineers a simple Python interface for turning a local code snippet into a production service.
+A ML engineer simply needs to call a few functions from Scale's SDK, which quickly spins up a production-ready service.
+The service efficiently utilizes compute resources and automatically scales according to traffic.
+
+
+Central to Scale Deploy are the notions of a `ModelBundle` and a `ModelEndpoint`.
+
+A `ModelBundle` consists of a trained model as well as the surrounding preprocessing and postprocessing code.
+Specifically, a `ModelBundle` consists of two Python objects, a (`model` or `load_model`), and a `load_predict_fn`; such that
+
+
+    load_predict_fn(model)
+
+
+or
+
+
+    load_predict_fn(load_model())
+
+
+returns a function `predict_fn` that takes in one argument representing model input,
+and outputs one argument representing model output.
+
+Typically, a `model` would be a Pytorch nn.Module or Tensorflow Keras model.
+
+TODO should we include a specific example here?
+
+A `ModelEndpoint` is the compute layer that takes in a `ModelBundle`, and is able to carry out inference requests
+by using the `ModelBundle` to carry out predictions. The `ModelEndpoint` also knows infrastructure-level details,
+such as how many GPUs are needed, what type they are, how much memory, etc. The `ModelEndpoint` automatically handles
+infrastructure level details such as autoscaling and task queueing.
+
+Steps to deploy your model via Scale Deploy:
+
+1. First, you create and upload a `ModelBundle`. Pass your trained model as well as pre-/post-processing code to
+the Scale Deploy Python SDK, and we'll create a model bundle based on the code and store it in our Bundle Store.
+
+2. Then, you create a `ModelEndpoint`. Pass a `ModelBundle` as well as infrastructure settings such as #GPUs to our SDK.
+This provisions resources on Scale's cluster dedicated to your `ModelEndpoint`.
+
+3. Lastly, you make requests to the `ModelEndpoint`. You can make requests through the Python SDK, or make HTTP requests directly
+to Scale.
+
+TODO: link some example colab notebook
+"""
+
+from .client import DeployClient
+from .model_bundle import ModelBundle
+from .model_endpoint import ModelEndpoint, ModelEndpointAsyncJob
diff --git a/nucleus/deploy/client.py b/nucleus/deploy/client.py
new file mode 100644
index 00000000..335206fa
--- /dev/null
+++ b/nucleus/deploy/client.py
@@ -0,0 +1,271 @@
+import logging
+from typing import Any, Callable, Dict, List, Optional, TypeVar
+
+import cloudpickle
+import requests
+
+from nucleus.connection import Connection
+from nucleus.deploy.find_packages import find_packages_from_imports
+from nucleus.deploy.model_bundle import ModelBundle
+from nucleus.deploy.model_endpoint import ModelEndpoint
+
+SCALE_DEPLOY_ENDPOINT = "https://api.scale.com/v1/hosted_inference"
+DEFAULT_NETWORK_TIMEOUT_SEC = 120
+
+logger = logging.getLogger(__name__)
+logging.basicConfig()
+
+DeployModel_T = TypeVar("DeployModel_T")
+
+
+class DeployClient:
+    """Scale Deploy Python Client extension."""
+
+    def __init__(self, api_key: str, endpoint: str = SCALE_DEPLOY_ENDPOINT):
+        """
+        Initializes a Scale Deploy Client.
+
+        Parameters:
+            api_key: Your Scale API key
+            endpoint: The Scale Deploy Endpoint (this should not need to be changed)
+        """
+        self.connection = Connection(api_key, endpoint)
+
+    def __repr__(self):
+        return f"DeployClient(connection='{self.connection}')"
+
+    def __eq__(self, other):
+        return self.connection == other.connection
+
+    def create_model_bundle(
+        self,
+        model_bundle_name: str,
+        model: DeployModel_T,
+        load_predict_fn: Callable[[DeployModel_T], Callable[[Any], Any]],
+    ) -> ModelBundle:
+        """
+        Grabs a s3 signed url and uploads a model bundle to Scale Deploy.
+        A model bundle consists of a "model" and a "load_predict_fn", such that
+        load_predict_fn(model) returns a function predict_fn that takes in model input and returns model output.
+        Pre/post-processing code can be included inside load_predict_fn/model.
+
+        Parameters:
+            model_bundle_name: Name of model bundle you want to create. This acts as a unique identifier.
+            model: Typically a trained Neural Network, e.g. a Pytorch module
+            load_predict_fn: Function that when called with model, returns a function that carries out inference
+        """
+        # Grab a signed url to make upload to
+        model_bundle_s3_url = self.connection.post({}, "model_bundle_upload")
+        if "signedUrl" not in model_bundle_s3_url:
+            raise Exception(
+                "Error in server request, no signedURL found"
+            )  # TODO code style broad exception
+        s3_path = model_bundle_s3_url["signedUrl"]
+        raw_s3_url = f"s3://{model_bundle_s3_url['bucket']}/{model_bundle_s3_url['key']}"
+
+        # Make bundle upload
+        bundle = dict(model=model, load_predict_fn=load_predict_fn)
+        serialized_bundle = cloudpickle.dumps(bundle)
+        requests.put(s3_path, data=serialized_bundle)
+
+        self.connection.post(
+            payload=dict(id=model_bundle_name, location=raw_s3_url),
+            route="model_bundle",
+        )  # TODO use return value somehow
+        # resp["data"]["bundle_name"] should equal model_bundle_name
+        # TODO check that a model bundle was created and no name collisions happened
+        return ModelBundle(model_bundle_name)
+
+    def create_model_endpoint(
+        self,
+        service_name: str,
+        model_bundle: ModelBundle,
+        cpus: int,
+        memory: str,
+        gpus: int,
+        min_workers: int,
+        max_workers: int,
+        per_worker: int,
+        env_params: Dict[str, str],
+        requirements: Optional[List[str]] = None,
+        gpu_type: Optional[str] = None,
+    ) -> ModelEndpoint:
+        """
+        Creates a Model Endpoint that is able to serve requests
+
+        Parameters:
+            service_name: Name of model endpoint. Must be unique.
+            model_bundle: The ModelBundle that you want your Model Endpoint to serve
+            cpus: Number of cpus each worker should get, e.g. 1, 2, etc.
+            memory: Amount of memory each worker should get, e.g. "4Gi", "512Mi", etc.
+            gpus: Number of gpus each worker should get, e.g. 0, 1, etc.
+            min_workers: Minimum number of workers for model endpoint
+            max_workers: Maximum number of workers for model endpoint
+            per_worker: An autoscaling parameter. Use this to make a tradeoff between latency and costs,
+                a lower per_worker will mean more workers are created for a given workload
+            requirements: A list of python package requirements, e.g.
+                ["tensorflow==2.3.0", "tensorflow-hub==0.11.0"]. If no list has been passed, will default to the currently
+                imported list of packages.
+            env_params: A dictionary that dictates environment information e.g.
+                the use of pytorch or tensorflow, which cuda/cudnn versions to use.
+                Specifically, the dictionary should contain the following keys:
+                "framework_type": either "tensorflow" or "pytorch".
+                "pytorch_version": Version of pytorch, e.g. "1.5.1", "1.7.0", etc. Only applicable if framework_type is pytorch
+                "cuda_version": Version of cuda used, e.g. "11.0".
+                "cudnn_version" Version of cudnn used, e.g. "cudnn8-devel".
+                "tensorflow_version": Version of tensorflow, e.g. "2.3.0". Only applicable if framework_type is tensorflow
+            gpu_type: If specifying a non-zero number of gpus, this controls the type of gpu requested. Current options are
+                "nvidia-tesla-t4" for NVIDIA T4s, or "nvidia-tesla-v100" for NVIDIA V100s.
+
+        Returns:
+             A ModelEndpoint object that can be used to make requests to the endpoint.
+
+        """
+        if requirements is None:
+            requirements_inferred = find_packages_from_imports(globals())
+            requirements = [
+                f"{key}=={value}"
+                for key, value in requirements_inferred.items()
+            ]
+            logger.info(
+                "Using \n%s\n for model endpoint %s",
+                requirements,
+                service_name,
+            )
+            # TODO test
+        payload = dict(
+            service_name=service_name,
+            env_params=env_params,
+            bundle_name=model_bundle.name,
+            cpus=cpus,
+            memory=memory,
+            gpus=gpus,
+            gpu_type=gpu_type,
+            min_workers=min_workers,
+            max_workers=max_workers,
+            per_worker=per_worker,
+            requirements=requirements,
+        )
+        if gpus == 0:
+            del payload["gpu_type"]
+        elif gpus > 0 and gpu_type is None:
+            raise ValueError("If nonzero gpus, must provide gpu_type")
+        resp = self.connection.post(payload, "endpoints")
+        endpoint_creation_task_id = resp["data"][
+            "endpoint_id"
+        ]  # Serverside needs updating
+        logger.info(
+            "Endpoint creation task id is %s", endpoint_creation_task_id
+        )
+        return ModelEndpoint(endpoint_id=service_name, client=self)
+
+    # Relatively small wrappers around http requests
+
+    def list_bundles(self) -> List[ModelBundle]:
+        """
+        Returns a list of model bundles that the user owns.
+        TODO this route doesn't exist serverside
+        """
+        # resp = self.connection.get("model_bundle")
+        raise NotImplementedError
+
+    def list_model_endpoints(self) -> List[ModelEndpoint]:
+        """
+        Lists all model endpoints that the user owns.
+        TODO: single get_model_endpoint(self)? route doesn't exist serverside I think
+
+        Returns:
+            A list of ModelEndpoint objects
+        """
+        resp = self.connection.get("endpoints")
+        return [
+            ModelEndpoint(endpoint_id=endpoint_id, client=self)
+            for endpoint_id in resp
+        ]
+
+    def sync_request(self, endpoint_id: str, s3url: str) -> str:
+        """
+        Makes a request to the Model Endpoint at endpoint_id, and blocks until request completion or timeout.
+
+        Parameters:
+            endpoint_id: The id of the endpoint to make the request to
+            s3url: A url that points to a file containing model input.
+                Must be accessible by Scale Deploy, hence it needs to either be public or a signedURL.
+
+        Returns:
+            A signedUrl that contains a cloudpickled Python object, the result of running inference on the model input
+            Example output:
+                `https://foo.s3.us-west-2.amazonaws.com/bar/baz/qux?xyzzy`
+        """
+        resp = self.connection.post(
+            payload=dict(url=s3url), route=f"task/{endpoint_id}"
+        )
+        return resp["data"]["result_url"]
+
+    def async_request(self, endpoint_id: str, s3url: str) -> str:
+        """
+        Makes a request to the Model Endpoint at endpoint_id, and immediately returns a key that can be used to retrieve
+        the result of inference at a later time.
+
+        Parameters:
+            endpoint_id: The id of the endpoint to make the request to
+            s3url: A url that points to a file containing model input.
+                Must be accessible by Scale Deploy, hence it needs to either be public or a signedURL.
+
+        Returns:
+            An id/key that can be used to fetch inference results at a later time.
+            Example output:
+                `abcabcab-cabc-abca-0123456789ab`
+        """
+        resp = self.connection.post(
+            payload=dict(url=s3url), route=f"task_async/{endpoint_id}"
+        )
+        return resp["data"]["task_id"]
+
+    def get_async_response(self, async_task_id: str) -> str:
+        """
+        Gets inference results from a previously created task.
+
+        Parameters:
+            async_task_id: The id/key returned from a previous invocation of async_request.
+
+        Returns:
+            A dictionary that contains task status and optionally a result url if the task has completed.
+            Dictionary's keys are as follows:
+            state: 'PENDING' or 'SUCCESS' or 'FAILURE'
+            result_url: a url pointing to inference results. This url is accessible for 12 hours after the request has been made.
+            Example output:
+                `{'state': 'SUCCESS', 'result_url': 'https://foo.s3.us-west-2.amazonaws.com/bar/baz/qux?xyzzy'}`
+        TODO: do we want to read the results from here as well? i.e. translate result_url into a python object
+        """
+
+        resp = self.connection.get(route=f"task/result/{async_task_id}")
+        return resp["data"]
+
+    def batch_async_request(self, endpoint_id: str, s3urls: List[str]):
+        """
+        Sends a batch inference request to the Model Endpoint at endpoint_id, returns a key that can be used to retrieve
+        the results of inference at a later time.
+
+        Parameters:
+            endpoint_id: The id of the endpoint to make the request to
+            s3urls: A list of urls, each pointing to a file containing model input.
+                Must be accessible by Scale Deploy, hence urls need to either be public or signedURLs.
+
+        Returns:
+            An id/key that can be used to fetch inference results at a later time
+        """
+        raise NotImplementedError
+
+    def get_batch_async_response(self, batch_async_task_id: str):
+        """
+        TODO not sure about how the batch task returns an identifier for the batch.
+        Gets inference results from a previously created batch task.
+
+        Parameters:
+            batch_async_task_id: An id representing the batch task job
+
+        Returns:
+            TODO Something similar to a list of signed s3URLs
+        """
+        raise NotImplementedError
diff --git a/nucleus/deploy/find_packages.py b/nucleus/deploy/find_packages.py
new file mode 100644
index 00000000..aa6bfbcc
--- /dev/null
+++ b/nucleus/deploy/find_packages.py
@@ -0,0 +1,330 @@
+#!/usr/bin/env python
+
+# Copyright 2019 Atalaya Tech, Inc.
+
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+
+# http://www.apache.org/licenses/LICENSE-2.0
+
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import ast
+import logging
+import os
+import pkgutil
+import sys
+import types
+import zipfile
+import zipimport
+from typing import Dict
+
+EPP_NO_ERROR = 0
+EPP_PKG_NOT_EXIST = 1
+EPP_PKG_VERSION_MISMATCH = 2
+
+ZIPIMPORT_DIR = "zipimports"
+
+__mm = None
+
+
+logger = logging.getLogger(__name__)
+
+
+def parse_requirement_string(rs):
+    name, _, version = rs.partition("==")
+    return name, version
+
+
+def verify_pkg(pkg_req):
+    global __mm  # pylint: disable=global-statement
+    if __mm is None:
+        __mm = ModuleManager()
+    return __mm.verify_pkg(pkg_req)
+
+
+def seek_pip_packages(target_py_file_path):
+    global __mm  # pylint: disable=global-statement
+    if __mm is None:
+        __mm = ModuleManager()
+    return __mm.seek_pip_packages(target_py_file_path)
+
+
+def seek_pip_packages_from_imports(import_set):
+    global __mm  # pylint: disable=global-statement
+    if __mm is None:
+        __mm = ModuleManager()
+    return __mm.seek_in_import_set(import_set)
+
+
+def get_pkg_version(pkg_name):
+    global __mm  # pylint: disable=global-statement
+    if __mm is None:
+        __mm = ModuleManager()
+    return __mm.pip_pkg_map.get(pkg_name, None)
+
+
+def get_zipmodules():
+    global __mm  # pylint: disable=global-statement
+    if __mm is None:
+        __mm = ModuleManager()
+    return __mm.zip_modules
+
+
+def get_all_pip_installed_modules():
+    global __mm  # pylint: disable=global-statement
+    if __mm is None:
+        __mm = ModuleManager()
+
+    installed_modules = list(
+        # local modules are the ones imported from current directory, either from a
+        # module.py file or a module directory that contains a `__init__.py` file
+        filter(lambda m: not m.is_local, __mm.searched_modules.values())
+    )
+    return list(map(lambda m: m.name, installed_modules))
+
+
+class ModuleInfo:
+    def __init__(self, name, path, is_local, is_pkg):
+        super().__init__()
+        self.name = name
+        self.path = path
+        self.is_local = is_local
+        self.is_pkg = is_pkg
+
+
+class ModuleManager:
+    def __init__(self):
+        super().__init__()
+        self.pip_pkg_map = {}
+        self.pip_module_map = {}
+        self.setuptools_module_set = set()
+        self.nonlocal_package_path = set()
+
+        import pkg_resources
+
+        # yixu: this populates either self.pip_pkg_map or self.nonlocal_package_path
+        # pkg_resources.working_set is basically a snapshot of sys.path, i.e. the packages that are imported
+        for (
+            dist
+        ) in pkg_resources.working_set:  # pylint: disable=not-an-iterable
+            module_path = dist.module_path or dist.location
+            if not module_path:
+                # Skip if no module path was found for pkg distribution
+                continue
+
+            if os.path.realpath(module_path) != os.getcwd():
+                # add to nonlocal_package path only if it's not current directory
+                self.nonlocal_package_path.add(module_path)
+
+            self.pip_pkg_map[dist._key] = dist._version
+            for mn in dist._get_metadata("top_level.txt"):
+                if dist._key != "setuptools":
+                    self.pip_module_map.setdefault(mn, []).append(
+                        (dist._key, dist._version)
+                    )
+                else:
+                    self.setuptools_module_set.add(mn)
+
+        # yixu: searched_modules is basically just pkgutil.iter_modules
+        self.searched_modules = {}
+        self.zip_modules: Dict[str, zipimport.zipimporter] = {}
+        for m in pkgutil.iter_modules():
+            if m.name not in self.searched_modules:
+                if isinstance(m.module_finder, zipimport.zipimporter):
+                    print(f"Detected zipimporter {m.module_finder}")
+                    path = m.module_finder.archive
+                    self.zip_modules[path] = m.module_finder
+                else:
+                    path = m.module_finder.path
+                is_local = self.is_local_path(path)
+                self.searched_modules[m.name] = ModuleInfo(
+                    m.name, path, is_local, m.ispkg
+                )
+
+    def verify_pkg(self, pkg_req):
+        if pkg_req.name not in self.pip_pkg_map:
+            # package does not exist in the current python session
+            return EPP_PKG_NOT_EXIST
+
+        if self.pip_pkg_map[pkg_req.name] not in pkg_req.specifier:
+            # package version being used in the current python session does not meet
+            # the specified package version requirement
+            return EPP_PKG_VERSION_MISMATCH
+
+        return EPP_NO_ERROR
+
+    def seek_pip_packages(self, target_py_file_path):
+        print("target py file path: %s", target_py_file_path)
+        work = DepSeekWork(self)
+        work.do(target_py_file_path)
+        requirements = {}
+        for _, pkg_info_list in work.dependencies.items():
+            for pkg_name, pkg_version in pkg_info_list:
+                requirements[pkg_name] = pkg_version
+
+        return requirements, work.unknown_module_set
+
+    def seek_in_import_set(self, import_set):
+        work = DepSeekWork(self)
+        work.do_import_set(import_set)
+        requirements = {}
+        for _, pkg_info_list in work.dependencies.items():
+            for pkg_name, pkg_version in pkg_info_list:
+                requirements[pkg_name] = pkg_version
+
+        return requirements, work.unknown_module_set
+
+    def is_local_path(self, path):
+        if path in self.nonlocal_package_path:
+            return False
+
+        dir_name = os.path.split(path)[1]
+        # pylint: disable=too-many-boolean-expressions
+        if (
+            "site-packages" in path
+            or "anaconda" in path
+            or path.endswith("packages")
+            or dir_name == "bin"
+            or dir_name.startswith("lib")
+            or dir_name.startswith("python")
+            or dir_name.startswith("plat")
+        ):
+            self.nonlocal_package_path.add(path)
+            return False
+
+        return True
+
+
+class DepSeekWork:
+    def __init__(self, module_manager):
+        super().__init__()
+        self.module_manager = module_manager
+
+        self.dependencies = {}
+        self.unknown_module_set = set()
+        self.parsed_module_set = set()
+
+    def do(self, target_py_file_path):
+        self.seek_in_file(target_py_file_path)
+
+    def do_import_set(self, import_set):
+        self.seek_in_import_set(import_set)
+
+    def seek_in_file(self, file_path):
+        try:
+            with open(file_path) as f:  # pylint: disable=unspecified-encoding
+                content = f.read()
+        except UnicodeDecodeError:
+            with open(file_path, encoding="utf-8") as f:
+                content = f.read()
+        self.seek_in_source(content)
+
+    def seek_in_import_set(self, import_set):
+        for module_name in import_set:
+            if module_name == "nucleus":
+                continue
+            if module_name in self.parsed_module_set:
+                continue
+            self.parsed_module_set.add(module_name)
+
+            if module_name in self.module_manager.searched_modules:
+                m = self.module_manager.searched_modules[module_name]
+                if m.is_local:
+                    # Recursively search dependencies in sub-modules
+                    if m.path in self.module_manager.zip_modules:
+                        self.seek_in_zip(m.path)
+                    elif m.is_pkg:
+                        self.seek_in_dir(os.path.join(m.path, m.name))
+                    else:
+                        self.seek_in_file(os.path.join(m.path, f"{m.name}.py"))
+                else:
+                    # check if the package has already been added to the list
+                    if (
+                        module_name in self.module_manager.pip_module_map
+                        and module_name not in self.dependencies
+                        and module_name
+                        not in self.module_manager.setuptools_module_set
+                    ):
+                        self.dependencies[
+                            module_name
+                        ] = self.module_manager.pip_module_map[module_name]
+            else:
+                if module_name in self.module_manager.pip_module_map:
+                    if module_name not in self.dependencies:
+                        # In some special cases, the pip-installed module can not
+                        # be located in the searched_modules
+                        self.dependencies[
+                            module_name
+                        ] = self.module_manager.pip_module_map[module_name]
+                else:
+                    if module_name not in sys.builtin_module_names:
+                        self.unknown_module_set.add(module_name)
+
+    def seek_in_source(self, content):
+        # Extract all dependency modules by searching through the trees of the Python
+        # abstract syntax grammar with Python's built-in ast module
+        tree = ast.parse(content)
+        import_set = set()
+        for node in ast.walk(tree):
+            if isinstance(node, ast.Import):
+                for name in node.names:
+                    import_set.add(name.name.partition(".")[0])
+            elif isinstance(node, ast.ImportFrom):
+                if node.module is not None and node.level == 0:
+                    import_set.add(node.module.partition(".")[0])
+        self.seek_in_import_set(import_set)
+
+    def seek_in_dir(self, dir_path):
+        for path, dir_list, file_list in os.walk(dir_path):
+            for file_name in file_list:
+                if not file_name.endswith(".py"):
+                    continue
+                self.seek_in_file(os.path.join(path, file_name))
+            for dir_name in dir_list:
+                if dir_name == "__pycache__":
+                    continue
+                self.seek_in_dir(os.path.join(path, dir_name))
+
+    def seek_in_zip(self, zip_path):
+        with zipfile.ZipFile(zip_path) as zf:
+            for module_path in zf.infolist():
+                filename = module_path.filename
+                if filename.endswith(".py"):
+                    logger.debug("Seeking modules in zip %s", filename)
+                    content = self.module_manager.zip_modules[
+                        zip_path
+                    ].get_source(filename.replace(".py", ""))
+                    self.seek_in_source(content)
+
+
+def find_packages_from_path(path: str):
+    """
+    Call this on another python file.
+    """
+    reqs, _ = seek_pip_packages(path)
+    return reqs
+
+
+def find_packages_from_imports(globals_copy):
+    """
+    Call this from a python notebook to get the current notebook's packages.
+    """
+    imports = _get_imports(globals_copy)
+    reqs, _ = seek_pip_packages_from_imports(imports)
+    return reqs
+
+
+def _get_imports(globals_copy):
+    """"""
+
+    imports = set()
+    for _, val in globals_copy.items():
+        if isinstance(val, types.ModuleType):
+            imports.add(val.__name__)
+    return imports
diff --git a/nucleus/deploy/model_bundle.py b/nucleus/deploy/model_bundle.py
new file mode 100644
index 00000000..8fdb0e84
--- /dev/null
+++ b/nucleus/deploy/model_bundle.py
@@ -0,0 +1,8 @@
+class ModelBundle:
+    """
+    Represents a ModelBundle.
+    TODO fill this out with more than just a name potentially.
+    """
+
+    def __init__(self, name):
+        self.name = name
diff --git a/nucleus/deploy/model_endpoint.py b/nucleus/deploy/model_endpoint.py
new file mode 100644
index 00000000..d8a81749
--- /dev/null
+++ b/nucleus/deploy/model_endpoint.py
@@ -0,0 +1,147 @@
+from typing import Dict, Optional, Sequence
+
+
+class ModelEndpoint:
+    """
+    A higher level abstraction for a Model Endpoint.
+    """
+
+    def __init__(self, endpoint_id: str, client):
+        """
+        Parameters:
+            endpoint_id: The unique name of the ModelEndpoint
+            client: A DeployClient object
+        """
+        self.endpoint_id = endpoint_id
+        self.client = client
+
+    def __str__(self):
+        return f"ModelEndpoint <endpoint_id:{self.endpoint_id}>"
+
+    def predict(
+        self,
+        s3urls: Sequence[str],
+    ) -> "ModelEndpointAsyncJob":
+        """
+        Runs inference on the data items specified by s3urls. Returns a ModelEndpointAsyncJob.
+
+        Parameters:
+            s3urls: The list of s3URLs that should have inference run on them.
+
+        Returns:
+            a ModelEndpointAsyncJob keeping track of the inference requests made
+        """
+        # Make inference requests to the endpoint,
+        # if batches are possible make this aware you can pass batches
+        # TODO add batch support once those are out
+
+        request_ids = {}  # Dict of s3url -> request id
+
+        for s3url in s3urls:
+            # TODO make these requests in parallel instead of making them serially
+            inference_request = self.client.async_request(
+                endpoint_id=self.endpoint_id,
+                s3url=s3url,
+            )
+            request_ids[s3url] = inference_request["task_id"]
+            # make the request to the endpoint (in parallel or something)
+
+        return ModelEndpointAsyncJob(
+            self.client,
+            request_ids=request_ids,
+        )
+
+    def status(self):
+        """Gets the status of the ModelEndpoint.
+        TODO this functionality currently does not exist on the server.
+        """
+        raise NotImplementedError
+
+    def sync_request(self, s3url: str) -> str:
+        """Makes a single request to the endpoint
+
+        Parameters:
+            s3url: A url that points to a file containing model input.
+                Must be accessible by Scale Deploy, hence it needs to either be public or a signedURL.
+
+        Returns:
+            A signedUrl that contains a cloudpickled Python object, the result of running inference on the model input
+            Example output:
+                `https://foo.s3.us-west-2.amazonaws.com/bar/baz/qux?xyzzy`
+        """
+        return self.client.sync_request(self.endpoint_id, s3url)
+
+    async def async_request(self, s3url: str) -> str:
+        """
+        Makes an async request to the endpoint. Polls the endpoint under the hood, but provides async/await semantics
+        on top.
+
+        Parameters:
+            s3url: A url that points to a file containing model input.
+                Must be accessible by Scale Deploy, hence it needs to either be public or a signedURL.
+
+        Returns:
+            A signedUrl that contains a cloudpickled Python object, the result of running inference on the model input
+            Example output:
+                `https://foo.s3.us-west-2.amazonaws.com/bar/baz/qux?xyzzy`
+        """
+        raise NotImplementedError
+
+
+class ModelEndpointAsyncJob:
+    """
+    Currently represents a list of async inference requests to a specific endpoint. Keeps track of the requests made,
+    and gives a way to poll for their status.
+
+    Invariant: set keys for self.request_ids and self.responses are equal
+
+    idk about this abstraction tbh, could use a redesign maybe?
+
+    Also batch inference sort of removes the need for much of the complication in here
+
+    """
+
+    def __init__(
+        self,
+        client,
+        request_ids: Dict[str, str],
+    ):
+
+        self.client = client
+        self.request_ids = request_ids.copy()  # s3url -> task_id
+        self.responses: Dict[str, Optional[str]] = {
+            s3url: None for s3url in request_ids.keys()
+        }
+
+    def poll_endpoints(self):
+        """
+        Runs one round of polling the endpoint for async task results
+        """
+
+        # TODO: replace with batch endpoint, or make requests in parallel
+        for s3url, request_id in self.request_ids.items():
+            current_response = self.responses[s3url]
+            if current_response is None:
+                response = self.client.get_async_response(request_id)
+                print(response)
+                if (
+                    "result_url" not in response
+                ):  # TODO this doesn't handle any task states other than Pending or Success
+                    continue
+                self.responses[s3url] = response["result_url"]
+
+    def is_done(self, poll=True) -> bool:
+        """
+        Checks if all the tasks from this round of requests are done, according to
+        the internal state of this object.
+        Optionally polls the endpoints to pick up new tasks that may have finished.
+        """
+        # TODO: make some request to some endpoint
+        if poll:
+            self.poll_endpoints()
+        return all(resp is not None for resp in self.responses.values())
+
+    def get_responses(self) -> Dict[str, Optional[str]]:
+        if not self.is_done(poll=False):
+            raise ValueError("Not all responses are done")
+        return self.responses.copy()
diff --git a/nucleus/deploy/nucleus_integration.py b/nucleus/deploy/nucleus_integration.py
new file mode 100644
index 00000000..20652501
--- /dev/null
+++ b/nucleus/deploy/nucleus_integration.py
@@ -0,0 +1,156 @@
+# This file contains all the Core Nucleus <-> HMI integrations
+import logging
+from typing import Dict, List, Tuple
+
+import cloudpickle
+import smart_open
+from boto3 import Session
+
+import nucleus
+from nucleus import Dataset, DatasetItem
+from nucleus.dataset_item import DatasetItemType
+from nucleus.deploy.model_endpoint import ModelEndpoint, ModelEndpointAsyncJob
+
+logger = logging.getLogger(__name__)
+logging.basicConfig()
+
+
+class NucleusDatasetInferenceRun:
+    """
+    This class is temporary, long-term we want to move to Nucleus backend calling HMI directly
+    """
+
+    # For the demo, we will need our Nucleus Dataset to have `image_location`s in s3://scale-ml-hosted-model-inference
+    def __init__(
+        self,
+        hmi_async_job: ModelEndpointAsyncJob,
+        nucleus_client,
+        s3url_to_dataset_map,
+        dataset,
+    ):
+        self.hmi_async_job = hmi_async_job
+        self.nucleus_client = nucleus_client
+        self.s3url_to_dataset_map = s3url_to_dataset_map
+        self.dataset = dataset
+
+    def is_done(self, poll=True):
+        return self.hmi_async_job.is_done(poll=poll)
+
+    def poll(self):
+        return self.hmi_async_job.poll_endpoints()
+
+    def upload_to_nucleus(
+        self, model_run_name, model=None, model_name=None, model_ref_id=None
+    ):
+        """Upload model run responses to Nucleus."""
+        # TODO untested
+        # TODO we eventually want to move away from client side upload
+
+        if not self.is_done(poll=False):
+            raise ValueError("Not all responses are done")
+
+        # Create a Nucleus model if we don't have one
+        if model is None:
+            assert (
+                model_name is not None and model_ref_id is not None
+            ), "If you don't pass a nucleus model you better pass a model name and reference id"
+            model = self.nucleus_client.add_model(
+                name=model_name, reference_id=model_ref_id
+            )
+
+        # Create a Nucleus model run
+        model_run = model.create_run(
+            name=model_run_name, dataset=self.dataset, predictions=[]
+        )
+        prediction_items = []
+        for s3url, dataset_item in self.s3url_to_dataset_map.items():
+            item_link = self.hmi_async_job.responses[s3url]
+            print(f"item_link={item_link}")
+            # e.g. s3://scale-ml/tmp/hosted-model-inference-outputs/a224499e-50ac-4b08-ad0c-c18e74c14184.pkl
+            kwargs = {
+                "transport_params": {
+                    "session": Session(profile_name="ml-worker")
+                }
+            }
+
+            with smart_open.open(item_link, "rb", **kwargs) as bundle_pkl:
+                inference_result = cloudpickle.load(bundle_pkl)
+                ref_id = dataset_item.reference_id
+                for box in inference_result:
+                    # TODO assuming box is a list of (x, y, w, h, label). This is almost certainly not the case.
+                    # We will have to use a ModelEndpoint/ModelBundle that returns boxes in this format.
+                    # Also, label is probably returned as an integer instead of a label that makes semantic sense
+                    pred_item = nucleus.BoxPrediction(
+                        label=box["label"],
+                        x=box["left"],
+                        y=box["top"],
+                        width=box["width"],
+                        height=box["height"],
+                        reference_id=ref_id,
+                    )
+                    prediction_items.append(pred_item)
+
+        job = model_run.predict(prediction_items, asynchronous=True)
+        job.sleep_until_complete()
+        job.errors()
+
+
+def create_nucleus_dataset_inference_run(
+    hmi_endpoint: ModelEndpoint, nucleus_client, dataset: Dataset
+):
+    """
+    Returns a NucleusDatasetInferenceRun, client will need to periodically call poll on this in order to upload
+    """
+    s3urls, s3url_to_dataset_map = _nucleus_ds_to_s3url_list(dataset)
+    async_job = hmi_endpoint.predict(s3urls)
+    return NucleusDatasetInferenceRun(
+        async_job, nucleus_client, s3url_to_dataset_map, dataset
+    )
+
+
+def _nucleus_ds_to_s3url_list(
+    dataset: Dataset,
+) -> Tuple[List[str], Dict[str, DatasetItem]]:
+    # TODO I'm not sure if dataset items are necessarily s3URLs. Does this matter?
+    # TODO support lidar point clouds
+    if len(dataset.items) == 0:
+        logger.warning("Passed a dataset of length 0")
+        return [], {}  # TODO return type?
+    dataset_item_type = dataset.items[0].type
+    if not all(data.type == dataset_item_type for data in dataset.items):
+        logger.warning("Dataset has multiple item types")
+        raise Exception  # TODO (code style) too broad exception
+
+    s3url_to_dataset_map = {}
+    # Do we need to keep track of nucleus ids?
+    if dataset_item_type == DatasetItemType.IMAGE:
+        s3Urls = [
+            data.image_location
+            for data in dataset.items
+            if data.image_location is not None
+        ]
+        s3url_to_dataset_map = {
+            data.image_location: data
+            for data in dataset.items
+            if data.image_location is not None
+        }
+    elif dataset_item_type == DatasetItemType.POINTCLOUD:
+        s3Urls = [
+            data.pointcloud_location
+            for data in dataset.items
+            if data.pointcloud_location is not None
+        ]
+        s3url_to_dataset_map = {
+            data.pointcloud_location: data
+            for data in dataset.items
+            if data.pointcloud_location is not None
+        }
+    else:
+        raise NotImplementedError(
+            f"Dataset Item Type {dataset_item_type} not implemented"
+        )
+    # TODO for demo
+    return (
+        s3Urls,
+        s3url_to_dataset_map,
+    )  # TODO duplicated data in returned values
diff --git a/pyproject.toml b/pyproject.toml
index 73250d92..fcdc2e34 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -31,6 +31,10 @@ repository = "https://github.com/scaleapi/nucleus-python-client"
 documentation = "https://dashboard.scale.com/nucleus/docs/api"
 packages = [{include="nucleus"}]
 
+[[tool.poetry.source]]
+name = "pytorch-wheel-source"
+url = "https://vxlabs.com/pypi//simple/"
+
 [tool.poetry.dependencies]
 python = "^3.6.2"
 requests = "^2.23.0"
@@ -44,6 +48,8 @@ isort = "^5.10.1"
 numpy = "^1.19.5"
 scipy = "^1.5.4"
 Shapely = "^1.8.0"
+cloudpickle = "^1.6.0"
+
 
 [tool.poetry.dev-dependencies]
 poetry = "^1.1.5"
@@ -61,6 +67,9 @@ sphinx-autobuild = "^2021.3.14"
 furo = "^2021.10.9"
 sphinx-autoapi = "^1.8.4"
 pytest-xdist = "^2.5.0"
+torch = "1.7.1" # Test uploading model bundles only
+smart_open = "^1.9.0" # Temporarily for uploading model bundles directly to s3
+boto3 = "^1.12.32" # Temporarily for uploading model bundles directly to s3
 
 
 [tool.pytest.ini_options]
diff --git a/temp_test_e2e.py b/temp_test_e2e.py
new file mode 100644
index 00000000..b7b57ddf
--- /dev/null
+++ b/temp_test_e2e.py
@@ -0,0 +1,170 @@
+import os
+import time
+
+import nucleus
+from nucleus.deploy.client import DeployClient
+from nucleus.deploy.find_packages import find_packages_from_imports
+from nucleus.deploy.model_bundle import ModelBundle
+
+# TODO Don't include this file in final pr
+from nucleus.deploy.model_endpoint import ModelEndpoint
+from nucleus.deploy.nucleus_integration import (
+    create_nucleus_dataset_inference_run,
+)
+
+
+def create_dummy_bundle(hmi_client):
+    def returns_returns_1(x):
+        def returns_1(y):
+            return 1
+
+        return returns_1
+
+    model = None
+    load_predict_func = returns_returns_1
+    hmi_client.create_model_bundle("return1", model, load_predict_func)
+
+
+def create_endpoint(hmi_client):
+    env_params = {
+        "framework_type": "pytorch",
+        "pytorch_version": "1.7.0",
+        "cuda_version": "11.0",
+        "cudnn_version": "cudnn8-devel",
+    }
+
+    mb = ModelBundle("abc123")
+
+    # TODO out of date lol
+    args = {
+        "service_name": "seantest",
+        "env_params": env_params,
+        # "bundle_id": "abc123",
+        "model_bundle": mb,
+        "cpus": 1,
+        "memory": "4Gi",
+        "gpus": 1,
+        "gpu_type": "nvidia-tesla-t4",
+        "min_workers": 1,
+        "max_workers": 1,
+        "per_worker": 1,
+        "requirements": [],
+    }
+
+    model_endpoint = hmi_client.create_model_endpoint(**args)
+    print(model_endpoint.endpoint_id)
+
+
+def make_task_call(
+    endpoint_name: str, dataset_id: str, upload_to_nucleus=True
+):
+    print("Need to export NUCLEUS_API_KEY=live_<your live api key here>")
+    client = nucleus.NucleusClient(os.environ["NUCLEUS_API_KEY"])
+
+    dataset = client.get_dataset(dataset_id)
+
+    model_endpoint = ModelEndpoint(
+        endpoint_id=endpoint_name, client=hmi_client
+    )
+
+    inference_run = create_nucleus_dataset_inference_run(
+        model_endpoint, client, dataset
+    )
+
+    while not inference_run.is_done(poll=True):
+        print("Waiting for predictions to finish...")
+        time.sleep(5)
+
+    print("Predictions complete!")
+    predictions = inference_run.hmi_async_job.get_responses()
+    print(predictions)
+
+    if upload_to_nucleus:
+        ts = str(time.time())
+        inference_run.upload_to_nucleus(
+            model_run_name=ts,
+            model_name="Test HMI upload",
+            model_ref_id=f"test_hmi_upload_{ts}",
+        )
+
+
+def temp_clone_pandaset():
+    print("Need to export NUCLEUS_API_KEY=live_<your live api key here>")
+    public_dataset_client = nucleus.NucleusClient(
+        os.environ["NUCLEUS_PUBLIC_DATASET_API_KEY"]
+    )
+
+    dataset = public_dataset_client.get_dataset(
+        "ds_bwhjbyfb8mjj0ykagxf0"
+    )  # Public Pandaset Dataset id
+    dataset_items = dataset.items
+
+    client = nucleus.NucleusClient(os.environ["NUCLEUS_API_KEY"])
+    cloned_dataset = client.create_dataset("Pandaset clone - small")
+    cloned_dataset_items = [
+        nucleus.DatasetItem(
+            image_location=x.image_location,
+            reference_id=x.reference_id,
+            metadata=x.metadata,
+        )
+        for x in dataset_items[:1]
+    ]
+    cloned_dataset.append(cloned_dataset_items)
+
+
+# temp_clone_pandaset()
+
+if __name__ == "__main__":
+    # make_task_call(
+    #     endpoint_name="yi-tf-test",
+    #     #dataset_id="ds_c4wht080x81g060m0nfg",
+    #     dataset_id="ds_c4x8s3m6n260060cngs0",
+    #     upload_to_nucleus=True
+    # )
+    # temp_clone_pandaset()
+
+    packages = find_packages_from_imports(globals())
+    print(packages)
+    packages = find_packages_from_imports(globals())
+    print(packages)
+
+    hmi_client = DeployClient(api_key=os.environ["NUCLEUS_API_KEY"])
+    img_url = (
+        "https://scale.com/_next/static/media/dashboard-hero.ab478d39.png"
+    )
+    # hmi_client.create_endpoint()
+    # print(hmi_client.connection.post({}, "model_bundle_upload"))
+    # create_dummy_bundle(hmi_client)
+    env_params = {
+        "framework_type": "pytorch",
+        "pytorch_version": "1.7.0",
+        "cuda_version": "11.0",
+        "cudnn_version": "cudnn8-devel",
+    }
+    # me2 = hmi_client.create_model_endpoint(
+    #     service_name="seantest2",
+    #     env_params=env_params,
+    #     model_bundle=ModelBundle(name="return1"),
+    #     cpus=1,
+    #     memory="4Gi",
+    #     gpus=0,
+    #     gpu_type="Doesn't Matter",
+    #     min_workers=1,
+    #     max_workers=1,
+    #     per_worker=1,
+    #     requirements=[],
+    # )
+    # print(me2)
+
+    mes = hmi_client.list_model_endpoints()
+    for me in mes:
+        print(me)
+    m1 = mes[0]
+    print(hmi_client.sync_request("seantest2", img_url))
+    async_task = hmi_client.async_request("seantest2", img_url)
+    print(async_task)
+    print(hmi_client.get_async_response(async_task))
+    for i in range(10):
+        time.sleep(1)
+        print(hmi_client.get_async_response(async_task))
+    # print(hmi_client.get_bundles())
diff --git a/tests/experimental/__init__.py b/tests/experimental/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/tests/experimental/conftest.py b/tests/experimental/conftest.py
new file mode 100644
index 00000000..ba006ae5
--- /dev/null
+++ b/tests/experimental/conftest.py
@@ -0,0 +1,12 @@
+import os
+
+import pytest
+
+from nucleus.deploy import DeployClient
+
+API_KEY = os.environ["NUCLEUS_PYTEST_API_KEY"]
+
+
+@pytest.fixture(scope="module")
+def DEPLOY_CLIENT():
+    return DeployClient(api_key=API_KEY)
diff --git a/tests/experimental/test_model_bundle.py b/tests/experimental/test_model_bundle.py
new file mode 100644
index 00000000..7bcfe06b
--- /dev/null
+++ b/tests/experimental/test_model_bundle.py
@@ -0,0 +1,60 @@
+import time
+
+import pytest
+import torch
+
+from nucleus.deploy.nucleus_integration import _nucleus_ds_to_s3url_list
+
+
+@pytest.mark.integration
+def test_add_model_bundle(DEPLOY_CLIENT):
+    # perhaps this belongs in some other test script? This is a pretty heavy/end-to-end test
+    # Tests both client and server functionality
+    # TODO does it make sense to use an actual user to make the requests?
+
+    client = DEPLOY_CLIENT  # TODO set up nucleus pytest api key
+
+    model_name = f"TestModel_{int(time.time())}"  # Unfortunately we don't have a way of deleting this from the database
+    model = torch.nn.Linear(1, 1)  # probably should be something pytorch
+    model.weight[0, 0] = 42
+    model.bias[0] = 43
+
+    def load_predict_fn(model):
+        if torch.cuda.is_available():
+            print("Using GPU for inference")
+            device = torch.device("cuda")
+            model.cuda()
+        else:
+            print("Using CPU for inference")
+            device = torch.device("cpu")
+
+        model.eval()
+
+        def predict(preprocess_output):
+            # 'model' refers to the model we got back from get_model()
+            with torch.no_grad():
+                model_output = model(preprocess_output.to(device))
+
+            return model_output
+
+        return predict
+
+    model_bundle = client.create_model_bundle(
+        model_bundle_name=model_name,
+        model=model,
+        load_predict_fn=load_predict_fn,
+    )
+    assert model_bundle.name == model_name, "Model bundle name is not correct"
+
+    # TODO more granular tests?
+
+
+def test_nucleus_ds_to_s3url_list(dataset):
+    # This is a weird test, verification is roughly as hard as the actual code
+    s3urls, _ = _nucleus_ds_to_s3url_list(dataset)
+    assert len(s3urls) == len(
+        dataset.items
+    ), "number of s3urls isn't number of dataset.items"
+    assert s3urls == [
+        item.image_location for item in dataset.items
+    ], "S3URLs are different from items' image_locations"