diff --git a/mani_skill/agents/base_agent.py b/mani_skill/agents/base_agent.py
index 65171f319..be6142ece 100644
--- a/mani_skill/agents/base_agent.py
+++ b/mani_skill/agents/base_agent.py
@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import os
 from dataclasses import dataclass
 from typing import TYPE_CHECKING, Dict, List, Optional, Union
 
@@ -15,7 +16,7 @@
     PDJointPosControllerConfig,
 )
 from mani_skill.sensors.base_sensor import BaseSensor, BaseSensorConfig
-from mani_skill.utils import sapien_utils
+from mani_skill.utils import assets, download_asset, sapien_utils
 from mani_skill.utils.structs import Actor, Array, Articulation, Pose
 
 from .controllers.base_controller import (
@@ -157,6 +158,25 @@ def _load_articulation(self):
             sapien_utils.check_urdf_config(urdf_config)
             sapien_utils.apply_urdf_config(loader, urdf_config)
 
+        if not os.path.exists(asset_path):
+            print(f"Robot {self.uid} definition file not found at {asset_path}")
+            if len(assets.DATA_GROUPS[self.uid]) > 0:
+                response = download_asset.prompt_yes_no(
+                    f"Robot {self.uid} has assets available for download. Would you like to download them now?"
+                )
+                if response:
+                    for (
+                        asset_id
+                    ) in assets.expand_data_group_into_individual_data_source_ids(
+                        self.uid
+                    ):
+                        download_asset.download(assets.DATA_SOURCES[asset_id])
+                else:
+                    print(f"Exiting as assets for robot {self.uid} are not downloaded")
+                    exit()
+            else:
+                print(f"Exiting as assets for robot {self.uid} are not found")
+                exit()
         self.robot: Articulation = loader.load(asset_path)
         assert self.robot is not None, f"Fail to load URDF/MJCF from {asset_path}"
 
diff --git a/mani_skill/agents/registration.py b/mani_skill/agents/registration.py
index 99a57a74e..203909d1b 100644
--- a/mani_skill/agents/registration.py
+++ b/mani_skill/agents/registration.py
@@ -1,21 +1,21 @@
 from dataclasses import dataclass
-from typing import Dict
+from typing import Dict, List
 
 from mani_skill import logger
 from mani_skill.agents.base_agent import BaseAgent
+from mani_skill.utils import assets
 
 
 @dataclass
 class AgentSpec:
-    """Agent specifications. At the moment it is a simple wrapper around the agent_cls but the dataclass is used in case we may need additional metadata"""
-
     agent_cls: type[BaseAgent]
+    asset_download_ids: List[str]
 
 
 REGISTERED_AGENTS: Dict[str, AgentSpec] = {}
 
 
-def register_agent(override=False):
+def register_agent(asset_download_ids: List[str] = [], override=False):
     """A decorator to register agents into ManiSkill so they can be used easily by string uid.
 
     Args:
@@ -34,7 +34,10 @@ def _register_agent(agent_cls: type[BaseAgent]):
                 )
             return agent_cls
 
-        REGISTERED_AGENTS[agent_cls.uid] = AgentSpec(agent_cls=agent_cls)
+        REGISTERED_AGENTS[agent_cls.uid] = AgentSpec(
+            agent_cls=agent_cls, asset_download_ids=asset_download_ids
+        )
+        assets.DATA_GROUPS[agent_cls.uid] = asset_download_ids
         return agent_cls
 
     return _register_agent
diff --git a/mani_skill/agents/robots/anymal/anymal_c.py b/mani_skill/agents/robots/anymal/anymal_c.py
index c227d87e8..c487eab90 100644
--- a/mani_skill/agents/robots/anymal/anymal_c.py
+++ b/mani_skill/agents/robots/anymal/anymal_c.py
@@ -11,7 +11,7 @@
 from mani_skill.utils.structs.articulation import Articulation
 
 
-@register_agent()
+@register_agent(asset_download_ids=["anymal_c"])
 class ANYmalC(BaseAgent):
     uid = "anymal_c"
     urdf_path = f"{ASSET_DIR}/robots/anymal_c/urdf/anymal.urdf"
diff --git a/mani_skill/agents/robots/googlerobot/googlerobot.py b/mani_skill/agents/robots/googlerobot/googlerobot.py
index e3847c2a5..610828666 100644
--- a/mani_skill/agents/robots/googlerobot/googlerobot.py
+++ b/mani_skill/agents/robots/googlerobot/googlerobot.py
@@ -9,7 +9,7 @@
 
 
 # TODO (stao) (xuanlin): Add mobile base, model it properly based on real2sim
-@register_agent()
+@register_agent(asset_download_ids=["googlerobot"])
 class GoogleRobot(BaseAgent):
     uid = "googlerobot"
     urdf_path = (
diff --git a/mani_skill/agents/robots/stompy/stompy.py b/mani_skill/agents/robots/stompy/stompy.py
index 0f996219e..c2df133ca 100644
--- a/mani_skill/agents/robots/stompy/stompy.py
+++ b/mani_skill/agents/robots/stompy/stompy.py
@@ -9,7 +9,7 @@
 from mani_skill.sensors.camera import CameraConfig
 
 
-@register_agent()  # uncomment this if you want to register the agent so you can instantiate it by ID when creating environments
+@register_agent(asset_download_ids=["stompy"])
 class Stompy(BaseAgent):
     uid = "stompy"
     urdf_path = f"{ASSET_DIR}/robots/stompy/robot.urdf"
diff --git a/mani_skill/agents/robots/unitree_g1/g1.py b/mani_skill/agents/robots/unitree_g1/g1.py
index 9f7fbd9d0..2cd66bd70 100644
--- a/mani_skill/agents/robots/unitree_g1/g1.py
+++ b/mani_skill/agents/robots/unitree_g1/g1.py
@@ -8,7 +8,7 @@
 from mani_skill.sensors.camera import CameraConfig
 
 
-@register_agent()
+@register_agent(asset_download_ids=["unitree_g1"])
 class UnitreeG1(BaseAgent):
     uid = "unitree_g1"
     urdf_path = f"{ASSET_DIR}/robots/unitree_g1/g1.urdf"
@@ -19,10 +19,7 @@ class UnitreeG1(BaseAgent):
     keyframes = dict(
         standing=Keyframe(
             pose=sapien.Pose(p=[0, 0, 0.755]),
-            qpos=np.array(
-                [0.0] * 37
-            )
-            * 1,
+            qpos=np.array([0.0] * 37) * 1,
         )
     )
 
@@ -112,7 +109,7 @@ def is_fallen(self):
         return self.robot.pose.p[:, 2] < 0.3
 
 
-@register_agent()
+@register_agent(asset_download_ids=["unitree_g1"])
 class UnitreeG1Simplified(UnitreeG1):
     uid = "unitree_g1_simplified_legs"
     urdf_path = f"{ASSET_DIR}/robots/unitree_g1/g1_simplified_legs.urdf"
diff --git a/mani_skill/agents/robots/unitree_g1/g1_upper_body.py b/mani_skill/agents/robots/unitree_g1/g1_upper_body.py
index db7d95a8f..d0498fa21 100644
--- a/mani_skill/agents/robots/unitree_g1/g1_upper_body.py
+++ b/mani_skill/agents/robots/unitree_g1/g1_upper_body.py
@@ -11,7 +11,7 @@
 from mani_skill.utils.structs.actor import Actor
 
 
-@register_agent()
+@register_agent(asset_download_ids=["unitree_g1"])
 class UnitreeG1UpperBody(BaseAgent):
     uid = "unitree_g1_simplified_upper_body"
     urdf_path = f"{ASSET_DIR}/robots/unitree_g1/g1_simplified_upper_body.urdf"
@@ -194,7 +194,7 @@ def right_hand_is_grasping(self, object: Actor, min_force=0.5, max_angle=85):
         return torch.logical_and(lflag, rflag)
 
 
-@register_agent()
+@register_agent(asset_download_ids=["unitree_g1"])
 class UnitreeG1UpperBodyRightArm(UnitreeG1UpperBody):
     uid = "unitree_g1_simplified_upper_body_right_arm"
     urdf_path = f"{ASSET_DIR}/robots/unitree_g1/g1_simplified_upper_body.urdf"
diff --git a/mani_skill/agents/robots/unitree_go/unitree_go2.py b/mani_skill/agents/robots/unitree_go/unitree_go2.py
index 97b14d0cd..22ac303e6 100644
--- a/mani_skill/agents/robots/unitree_go/unitree_go2.py
+++ b/mani_skill/agents/robots/unitree_go/unitree_go2.py
@@ -8,7 +8,9 @@
 from mani_skill.sensors.camera import CameraConfig
 
 
-@register_agent()  # uncomment this if you want to register the agent so you can instantiate it by ID when creating environments
+@register_agent(
+    asset_download_ids=["unitree_go2"]
+)  # uncomment this if you want to register the agent so you can instantiate it by ID when creating environments
 class UnitreeGo2(BaseAgent):
     uid = "unitree_go2"
     urdf_path = f"{ASSET_DIR}/robots/unitree_go2/urdf/go2_description.urdf"  # You can use f"{PACKAGE_ASSET_DIR}" to reference a urdf file in the mani_skill /assets package folder
diff --git a/mani_skill/agents/robots/unitree_h1/h1.py b/mani_skill/agents/robots/unitree_h1/h1.py
index 99effc45b..726824038 100644
--- a/mani_skill/agents/robots/unitree_h1/h1.py
+++ b/mani_skill/agents/robots/unitree_h1/h1.py
@@ -8,7 +8,7 @@
 from mani_skill.sensors.camera import CameraConfig
 
 
-@register_agent()
+@register_agent(asset_download_ids=["unitree_h1"])
 class UnitreeH1(BaseAgent):
     uid = "unitree_h1"
     urdf_path = f"{ASSET_DIR}/robots/unitree_h1/urdf/h1.urdf"
@@ -114,7 +114,7 @@ def is_fallen(self):
         return self.robot.pose.p[:, 2] < 0.3
 
 
-@register_agent()
+@register_agent(asset_download_ids=["unitree_h1"])
 class UnitreeH1Simplified(UnitreeH1):
     uid = "unitree_h1_simplified"
     urdf_path = f"{ASSET_DIR}/robots/unitree_h1/urdf/h1_simplified.urdf"
diff --git a/mani_skill/agents/robots/ur_e/ur_10e.py b/mani_skill/agents/robots/ur_e/ur_10e.py
index 845c26c97..3da995896 100644
--- a/mani_skill/agents/robots/ur_e/ur_10e.py
+++ b/mani_skill/agents/robots/ur_e/ur_10e.py
@@ -7,7 +7,7 @@
 from mani_skill.agents.registration import register_agent
 
 
-@register_agent()
+@register_agent(asset_download_ids=["ur10e"])
 class UR10e(BaseAgent):
     uid = "ur_10e"
     mjcf_path = f"{ASSET_DIR}/robots/ur10e/ur10e.xml"
diff --git a/mani_skill/agents/robots/widowx/widowx.py b/mani_skill/agents/robots/widowx/widowx.py
index c340fa713..748edb355 100644
--- a/mani_skill/agents/robots/widowx/widowx.py
+++ b/mani_skill/agents/robots/widowx/widowx.py
@@ -9,7 +9,7 @@
 
 
 # TODO (stao) (xuanlin): model it properly based on real2sim
-@register_agent()
+@register_agent(asset_download_ids=["widowx250s"])
 class WidowX250S(BaseAgent):
     uid = "widowx250s"
     urdf_path = f"{ASSET_DIR}/robots/widowx250s/wx250s.urdf"
diff --git a/mani_skill/agents/robots/xmate3/xmate3.py b/mani_skill/agents/robots/xmate3/xmate3.py
index d65eae282..c66e68a61 100644
--- a/mani_skill/agents/robots/xmate3/xmate3.py
+++ b/mani_skill/agents/robots/xmate3/xmate3.py
@@ -15,7 +15,7 @@
 from mani_skill.utils.structs.actor import Actor
 
 
-@register_agent()
+@register_agent(asset_download_ids=["xmate3_robotiq"])
 class Xmate3Robotiq(BaseAgent):
     uid = "xmate3_robotiq"
     urdf_path = f"{ASSET_DIR}/robots/xmate3_robotiq/xmate3_robotiq.urdf"
diff --git a/mani_skill/envs/tasks/dexterity/rotate_single_object_in_hand.py b/mani_skill/envs/tasks/dexterity/rotate_single_object_in_hand.py
index 503bf2593..a37ccb17e 100644
--- a/mani_skill/envs/tasks/dexterity/rotate_single_object_in_hand.py
+++ b/mani_skill/envs/tasks/dexterity/rotate_single_object_in_hand.py
@@ -345,7 +345,11 @@ def __init__(self, *args, **kwargs):
         )
 
 
-@register_env("RotateSingleObjectInHandLevel2-v1", max_episode_steps=300)
+@register_env(
+    "RotateSingleObjectInHandLevel2-v1",
+    max_episode_steps=300,
+    asset_download_ids=["ycb"],
+)
 class RotateSingleObjectInHandLevel2(RotateSingleObjectInHand):
     def __init__(self, *args, **kwargs):
         super().__init__(
@@ -357,7 +361,11 @@ def __init__(self, *args, **kwargs):
         )
 
 
-@register_env("RotateSingleObjectInHandLevel3-v1", max_episode_steps=300)
+@register_env(
+    "RotateSingleObjectInHandLevel3-v1",
+    max_episode_steps=300,
+    asset_download_ids=["ycb"],
+)
 class RotateSingleObjectInHandLevel3(RotateSingleObjectInHand):
     def __init__(self, *args, **kwargs):
         super().__init__(
diff --git a/mani_skill/envs/tasks/mobile_manipulation/open_cabinet_drawer.py b/mani_skill/envs/tasks/mobile_manipulation/open_cabinet_drawer.py
index 32f12ed35..e6cef00a7 100644
--- a/mani_skill/envs/tasks/mobile_manipulation/open_cabinet_drawer.py
+++ b/mani_skill/envs/tasks/mobile_manipulation/open_cabinet_drawer.py
@@ -25,7 +25,11 @@
 
 # TODO (stao): we need to cut the meshes of all the cabinets in this dataset for gpu sim, there may be some wierd physics
 # that may happen although it seems okay for state based RL
-@register_env("OpenCabinetDrawer-v1", max_episode_steps=100)
+@register_env(
+    "OpenCabinetDrawer-v1",
+    asset_download_ids=["partnet_mobility_cabinet"],
+    max_episode_steps=100,
+)
 class OpenCabinetDrawerEnv(BaseEnv):
 
     SUPPORTED_ROBOTS = ["fetch"]
@@ -306,9 +310,9 @@ def compute_dense_reward(self, obs: Any, action: torch.Tensor, info: Dict):
             self.target_qpos - self.handle_link.joint.qpos, self.target_qpos
         )
         open_reward = 2 * (1 - amount_to_open_left)
-        reaching_reward[amount_to_open_left < 0.999] = (
-            2  # if joint opens even a tiny bit, we don't need reach reward anymore
-        )
+        reaching_reward[
+            amount_to_open_left < 0.999
+        ] = 2  # if joint opens even a tiny bit, we don't need reach reward anymore
         # print(open_reward.shape)
         open_reward[info["open_enough"]] = 3  # give max reward here
         reward = reaching_reward + open_reward
diff --git a/mani_skill/envs/tasks/tabletop/assembling_kits.py b/mani_skill/envs/tasks/tabletop/assembling_kits.py
index 84bee302a..9187f63d3 100644
--- a/mani_skill/envs/tasks/tabletop/assembling_kits.py
+++ b/mani_skill/envs/tasks/tabletop/assembling_kits.py
@@ -19,7 +19,9 @@
 from mani_skill.utils.structs.types import GPUMemoryConfig, SimConfig
 
 
-@register_env("AssemblingKits-v1", max_episode_steps=200)
+@register_env(
+    "AssemblingKits-v1", asset_download_ids=["assembling_kits"], max_episode_steps=200
+)
 class AssemblingKitsEnv(BaseEnv):
     SUPPORTED_REWARD_MODES = ["sparse", "none"]
     SUPPORTED_ROBOTS = ["panda_wristcam"]
@@ -64,7 +66,9 @@ def __init__(
 
     @property
     def _default_sim_config(self):
-        return SimConfig(gpu_memory_cfg=GPUMemoryConfig(max_rigid_contact_count=2**20))
+        return SimConfig(
+            gpu_memory_cfg=GPUMemoryConfig(max_rigid_contact_count=2**20)
+        )
 
     @property
     def _default_sensor_configs(self):
diff --git a/mani_skill/envs/tasks/tabletop/pick_clutter_ycb.py b/mani_skill/envs/tasks/tabletop/pick_clutter_ycb.py
index 387634413..7537ad78f 100644
--- a/mani_skill/envs/tasks/tabletop/pick_clutter_ycb.py
+++ b/mani_skill/envs/tasks/tabletop/pick_clutter_ycb.py
@@ -196,7 +196,11 @@ def compute_normalized_dense_reward(
         return self.compute_dense_reward(obs=obs, action=action, info=info) / max_reward
 
 
-@register_env("PickClutterYCB-v1", max_episode_steps=100)
+@register_env(
+    "PickClutterYCB-v1",
+    asset_download_ids=["ycb", "pick_clutter_ycb_configs"],
+    max_episode_steps=100,
+)
 class PickClutterYCBEnv(PickClutterEnv):
     DEFAULT_EPISODE_JSON = f"{ASSET_DIR}/tasks/pick_clutter/ycb_train_5k.json.gz"
 
diff --git a/mani_skill/envs/tasks/tabletop/pick_single_ycb.py b/mani_skill/envs/tasks/tabletop/pick_single_ycb.py
index 7bb9e6b97..f9fe62973 100644
--- a/mani_skill/envs/tasks/tabletop/pick_single_ycb.py
+++ b/mani_skill/envs/tasks/tabletop/pick_single_ycb.py
@@ -24,7 +24,7 @@
 WARNED_ONCE = False
 
 
-@register_env("PickSingleYCB-v1", max_episode_steps=50)
+@register_env("PickSingleYCB-v1", max_episode_steps=50, asset_download_ids=["ycb"])
 class PickSingleYCBEnv(BaseEnv):
 
     SUPPORTED_ROBOTS = ["panda", "panda_wristcam", "xmate3_robotiq", "fetch"]
diff --git a/mani_skill/utils/assets/README.md b/mani_skill/utils/assets/README.md
new file mode 100644
index 000000000..e180788c6
--- /dev/null
+++ b/mani_skill/utils/assets/README.md
@@ -0,0 +1,4 @@
+# ManiSkill Asset Management
+
+Code for asset management in ManiSkill
+
diff --git a/mani_skill/utils/assets/__init__.py b/mani_skill/utils/assets/__init__.py
new file mode 100644
index 000000000..444061787
--- /dev/null
+++ b/mani_skill/utils/assets/__init__.py
@@ -0,0 +1,7 @@
+from .data import (
+    DATA_GROUPS,
+    DATA_SOURCES,
+    DataSource,
+    expand_data_group_into_individual_data_source_ids,
+    is_data_source_downloaded,
+)
diff --git a/mani_skill/utils/assets/data.py b/mani_skill/utils/assets/data.py
new file mode 100644
index 000000000..a9a07f5b2
--- /dev/null
+++ b/mani_skill/utils/assets/data.py
@@ -0,0 +1,193 @@
+"""
+Asset sources and tooling for managing the assets
+"""
+
+import os
+from dataclasses import dataclass
+from typing import Dict, List, Optional
+
+from mani_skill import ASSET_DIR, PACKAGE_ASSET_DIR
+from mani_skill.utils import io_utils
+
+
+@dataclass
+class DataSource:
+    source_type: str
+    """what kind of data is this"""
+    url: Optional[str] = None
+    hf_repo_id: Optional[str] = None
+    github_url: Optional[str] = None
+    target_path: Optional[str] = None
+    """the folder where the file will be downloaded to"""
+    checksum: Optional[str] = None
+    zip_dirname: Optional[str] = None
+    """what to rename a zip files generated directory to"""
+    filename: Optional[str] = None
+    """name to change the downloaded file to. If None, will not change the name"""
+    output_dir: str = ASSET_DIR
+
+
+DATA_SOURCES: Dict[str, DataSource] = {}
+"""Data sources map data source IDs to their respective DataSource objects which contain info on what the data is and where to download it"""
+DATA_GROUPS: Dict[str, List[str]] = {}
+"""Data groups map group ids (typically environment IDs) to a list of data source/group IDs for easy group management. data groups can be done hierarchicaly"""
+
+
+def is_data_source_downloaded(data_source_id: str):
+    data_source = DATA_SOURCES[data_source_id]
+    return os.path.exists(data_source.output_dir / data_source.target_path)
+
+
+def initialize_data_sources():
+    DATA_SOURCES["ycb"] = DataSource(
+        source_type="task_assets",
+        url="https://huggingface.co/datasets/haosulab/ManiSkill2/resolve/main/data/mani_skill2_ycb.zip",
+        target_path="assets/mani_skill2_ycb",
+        checksum="174001ba1003cc0c5adda6453f4433f55ec7e804f0f0da22d015d525d02262fb",
+    )
+    DATA_SOURCES["pick_clutter_ycb_configs"] = DataSource(
+        source_type="task_assets",
+        url="https://storage1.ucsd.edu/datasets/ManiSkill2022-assets/pick_clutter/ycb_train_5k.json.gz",
+        target_path="tasks/pick_clutter",
+        checksum="70ec176c7036f326ea7813b77f8c03bea9db5960198498957a49b2895a9ec338",
+    )
+    DATA_SOURCES["assembling_kits"] = DataSource(
+        source_type="task_assets",
+        url="https://storage1.ucsd.edu/datasets/ManiSkill2022-assets/assembling_kits_v1.zip",
+        target_path="tasks/assembling_kits",
+        checksum="e3371f17a07a012edaa3a0b3604fb1577f3fb921876c3d5ed59733dd75a6b4a0",
+    )
+    DATA_SOURCES["panda_avoid_obstacles"] = DataSource(
+        source_type="task_assets",
+        url="https://storage1.ucsd.edu/datasets/ManiSkill2022-assets/avoid_obstacles/panda_train_2k.json.gz",
+        target_path="tasks/avoid_obstacles",
+        checksum="44dae9a0804172515c290c1f49a1e7e72d76e40201a2c5c7d4a3ccd43b4d5be4",
+    )
+
+    # ---------------------------------------------------------------------------- #
+    # PartNet-mobility
+    # ---------------------------------------------------------------------------- #
+    category_uids = {}
+    for category in ["cabinet_drawer", "cabinet_door", "chair", "bucket", "faucet"]:
+        model_json = (
+            PACKAGE_ASSET_DIR / f"partnet_mobility/meta/info_{category}_train.json"
+        )
+        model_ids = set(io_utils.load_json(model_json).keys())
+        category_uids[category] = []
+        for model_id in model_ids:
+            uid = f"partnet_mobility/{model_id}"
+            DATA_SOURCES[uid] = DataSource(
+                source_type="objects",
+                url=f"https://storage1.ucsd.edu/datasets/ManiSkill2022-assets/partnet_mobility/dataset/{model_id}.zip",
+                target_path=ASSET_DIR / "partnet_mobility" / "dataset" / model_id,
+            )
+            category_uids[category].append(uid)
+
+    DATA_GROUPS["partnet_mobility_cabinet"] = set(
+        category_uids["cabinet_drawer"] + category_uids["cabinet_door"]
+    )
+    DATA_GROUPS["partnet_mobility_chair"] = category_uids["chair"]
+    DATA_GROUPS["partnet_mobility_bucket"] = category_uids["bucket"]
+    DATA_GROUPS["partnet_mobility_faucet"] = category_uids["faucet"]
+    DATA_GROUPS["partnet_mobility"] = set(
+        category_uids["cabinet_drawer"]
+        + category_uids["cabinet_door"]
+        + category_uids["chair"]
+        + category_uids["bucket"]
+        + category_uids["faucet"]
+    )
+
+    # DATA_GROUPS["OpenCabinetDrawer-v1"] = category_uids["cabinet_drawer"]
+    # DATA_GROUPS["OpenCabinetDoor-v1"] = category_uids["cabinet_door"]
+    # DATA_GROUPS["PushChair-v1"] = category_uids["chair"]
+    # DATA_GROUPS["MoveBucket-v1"] = category_uids["bucket"]
+    # DATA_GROUPS["TurnFaucet-v1"] = category_uids["faucet"]
+
+    # ---------------------------------------------------------------------------- #
+    # Interactable Scene Datasets
+    # ---------------------------------------------------------------------------- #
+    DATA_SOURCES["ReplicaCAD"] = DataSource(
+        source_type="scene",
+        hf_repo_id="haosulab/ReplicaCAD",
+        target_path="scene_datasets/replica_cad_dataset",
+    )
+
+    DATA_SOURCES["ReplicaCADRearrange"] = DataSource(
+        source_type="scene",
+        url="https://huggingface.co/datasets/haosulab/ReplicaCADRearrange/resolve/main/v1_extracted.zip",
+        target_path="scene_datasets/replica_cad_dataset/rearrange",
+    )
+
+    DATA_SOURCES["AI2THOR"] = DataSource(
+        source_type="scene",
+        url="https://huggingface.co/datasets/haosulab/AI2THOR/resolve/main/ai2thor.zip",
+        target_path="scene_datasets/ai2thor",
+    )
+
+    # Robots
+    DATA_SOURCES["xmate3_robotiq"] = DataSource(
+        source_type="robot",
+        url="https://storage1.ucsd.edu/datasets/ManiSkill2022-assets/xmate3_robotiq.zip",
+        target_path="robots/xmate3_robotiq",
+        checksum="ddda102a20eb41e28a0a501702e240e5d7f4084221a44f580e729f08b7c12d1a",
+    )
+    DATA_SOURCES["ur10e"] = DataSource(
+        source_type="robot",
+        url="https://github.com/haosulab/ManiSkill-UR10e/archive/refs/tags/v0.1.0.zip",
+        target_path="robots/ur10e",
+    )
+    DATA_SOURCES["anymal_c"] = DataSource(
+        source_type="robot",
+        url="https://github.com/haosulab/ManiSkill-ANYmalC/archive/refs/tags/v0.1.1.zip",
+        target_path="robots/anymal_c",
+    )
+    DATA_SOURCES["unitree_h1"] = DataSource(
+        source_type="robot",
+        url="https://github.com/haosulab/ManiSkill-UnitreeH1/archive/refs/tags/v0.1.0.zip",
+        target_path="robots/unitree_h1",
+    )
+    DATA_SOURCES["unitree_g1"] = DataSource(
+        source_type="robot",
+        url="https://github.com/haosulab/ManiSkill-UnitreeG1/archive/refs/tags/v0.1.0.zip",
+        target_path="robots/unitree_g1",
+    )
+    DATA_SOURCES["unitree_go2"] = DataSource(
+        source_type="robot",
+        url="https://github.com/haosulab/ManiSkill-UnitreeGo2/archive/refs/tags/v0.1.0.zip",
+        target_path="robots/unitree_go2",
+    )
+    DATA_SOURCES["stompy"] = DataSource(
+        source_type="robot",
+        url="https://github.com/haosulab/ManiSkill-Stompy/archive/refs/tags/v0.1.0.zip",
+        target_path="robots/stompy",
+    )
+    DATA_SOURCES["widowx250s"] = DataSource(
+        source_type="robot",
+        url="https://github.com/haosulab/ManiSkill-WidowX250S/archive/refs/tags/v0.1.0.zip",
+        target_path="robots/widowx",
+    )
+    DATA_SOURCES["googlerobot"] = DataSource(
+        source_type="robot",
+        url="https://github.com/haosulab/ManiSkill-GoogleRobot/archive/refs/tags/v0.1.0.zip",
+        target_path="robots/googlerobot",
+    )
+
+
+def expand_data_group_into_individual_data_source_ids(data_group_id: str):
+    """Expand a data group into a list of individual data source IDs"""
+    uids = []
+
+    def helper(uid):
+        nonlocal uids
+        if uid in DATA_GROUPS:
+            [helper(x) for x in DATA_GROUPS[uid]]
+        elif uid in DATA_SOURCES:
+            uids.append(uid)
+
+    for uid in DATA_GROUPS[data_group_id]:
+        helper(uid)
+    uids = list(set(uids))
+    return uids
+
+
+initialize_data_sources()
diff --git a/mani_skill/utils/download_asset.py b/mani_skill/utils/download_asset.py
index 5f5015aac..8c57c2668 100644
--- a/mani_skill/utils/download_asset.py
+++ b/mani_skill/utils/download_asset.py
@@ -5,221 +5,15 @@
 import shutil
 import urllib.request
 import zipfile
-from dataclasses import dataclass
 from pathlib import Path
-from typing import Dict, Optional
 from urllib.error import URLError
 
 from huggingface_hub import snapshot_download
 from tqdm.auto import tqdm
 
-from mani_skill import ASSET_DIR, PACKAGE_ASSET_DIR
-from mani_skill.utils.io_utils import load_json
-
-
-@dataclass
-class DataSource:
-    source_type: str
-    """what kind of data is this"""
-    url: Optional[str] = None
-    hf_repo_id: Optional[str] = None
-    github_url: Optional[str] = None
-    target_path: Optional[str] = None
-    """the folder where the file will be downloaded to"""
-    checksum: Optional[str] = None
-    zip_dirname: Optional[str] = None
-    """what to rename a zip files generated directory to"""
-    filename: Optional[str] = None
-    """name to change the downloaded file to. If None, will not change the name"""
-    output_dir: str = ASSET_DIR
-
-
-DATA_SOURCES: Dict[str, DataSource] = {}
-DATA_GROUPS = {}
-
-
-def initialize_sources():
-    """
-    Initialize the metadata for assets
-
-    Note that the current organization works as follows
-
-    - assets/* contain files for individual objects (.obj, .glb etc.) and articulations (.urdf etc.) that are generally reused.
-        E.g. Partnet Mobility and the YCB dataset.
-    - tasks/* contain files that are otherwise too big to upload to GitHub and are relevant for just that one task and will generally not be reused
-    - scene_datasets/* is a bit messier but contains a self-contained folder for each scene dataset each of which is likely organized differently.
-        These datasets often put their unique object and articulation files together with scene configuration data. In the future we will re-organize these
-        datasets so that all objects are put into assets and leave scene_datasets for scene configuration information.
-    - robots/* contains files for additional robots that are not included by default.
-
-    """
-
-    # TODO add google scanned objects
-    DATA_SOURCES["ycb"] = DataSource(
-        source_type="task_assets",
-        url="https://huggingface.co/datasets/haosulab/ManiSkill2/resolve/main/data/mani_skill2_ycb.zip",
-        target_path="assets/mani_skill2_ycb",
-        checksum="174001ba1003cc0c5adda6453f4433f55ec7e804f0f0da22d015d525d02262fb",
-    )
-    DATA_GROUPS["PickSingleYCB-v1"] = ["ycb"]
-
-    DATA_SOURCES["pick_clutter_ycb"] = DataSource(
-        source_type="task_assets",
-        url="https://storage1.ucsd.edu/datasets/ManiSkill2022-assets/pick_clutter/ycb_train_5k.json.gz",
-        target_path="tasks/pick_clutter",
-        checksum="70ec176c7036f326ea7813b77f8c03bea9db5960198498957a49b2895a9ec338",
-    )
-    DATA_GROUPS["PickClutterYCB-v1"] = ["ycb", "pick_clutter_ycb"]
-
-    DATA_SOURCES["assembling_kits"] = DataSource(
-        source_type="task_assets",
-        url="https://storage1.ucsd.edu/datasets/ManiSkill2022-assets/assembling_kits_v1.zip",
-        target_path="tasks/assembling_kits",
-        checksum="e3371f17a07a012edaa3a0b3604fb1577f3fb921876c3d5ed59733dd75a6b4a0",
-    )
-    DATA_GROUPS["AssemblingKits-v1"] = ["assembling_kits"]
-
-    DATA_SOURCES["panda_avoid_obstacles"] = DataSource(
-        source_type="task_assets",
-        url="https://storage1.ucsd.edu/datasets/ManiSkill2022-assets/avoid_obstacles/panda_train_2k.json.gz",
-        target_path="tasks/avoid_obstacles",
-        checksum="44dae9a0804172515c290c1f49a1e7e72d76e40201a2c5c7d4a3ccd43b4d5be4",
-    )
-    DATA_GROUPS["PandaAvoidObstacles-v1"] = ["panda_avoid_obstacles"]
-
-    DATA_SOURCES["pinch"] = DataSource(
-        source_type="task_assets",
-        url="https://storage1.ucsd.edu/datasets/ManiSkill2022-assets/pinch.zip",
-        target_path="tasks/pinch",
-        checksum="3281d2d777fad42e6d37371b2d3ee16fb1c39984907176718ca2e4f447326fe7",
-    )
-    DATA_GROUPS["Pinch-v1"] = ["pinch"]
-
-    DATA_SOURCES["write"] = DataSource(
-        source_type="task_assets",
-        url="https://storage1.ucsd.edu/datasets/ManiSkill2022-assets/write.zip",
-        target_path="tasks/write",
-        checksum="c5b49e581bfed9cfb2107a607faf52795f840e93f5a7ad389290314513b4b634",
-    )
-    DATA_GROUPS["Write-v1"] = ["write"]
-
-    # ---------------------------------------------------------------------------- #
-    # PartNet-mobility
-    # ---------------------------------------------------------------------------- #
-    category_uids = {}
-    for category in ["cabinet_drawer", "cabinet_door", "chair", "bucket", "faucet"]:
-        model_json = (
-            PACKAGE_ASSET_DIR / f"partnet_mobility/meta/info_{category}_train.json"
-        )
-        model_ids = set(load_json(model_json).keys())
-        category_uids[category] = []
-        for model_id in model_ids:
-            uid = f"partnet_mobility/{model_id}"
-            DATA_SOURCES[uid] = DataSource(
-                source_type="objects",
-                url=f"https://storage1.ucsd.edu/datasets/ManiSkill2022-assets/partnet_mobility/dataset/{model_id}.zip",
-                target_path=ASSET_DIR / "partnet_mobility" / "dataset" / model_id,
-            )
-            category_uids[category].append(uid)
-
-    DATA_GROUPS["partnet_mobility_cabinet"] = set(
-        category_uids["cabinet_drawer"] + category_uids["cabinet_door"]
-    )
-    DATA_GROUPS["partnet_mobility_chair"] = category_uids["chair"]
-    DATA_GROUPS["partnet_mobility_bucket"] = category_uids["bucket"]
-    DATA_GROUPS["partnet_mobility_faucet"] = category_uids["faucet"]
-    DATA_GROUPS["partnet_mobility"] = set(
-        category_uids["cabinet_drawer"]
-        + category_uids["cabinet_door"]
-        + category_uids["chair"]
-        + category_uids["bucket"]
-        + category_uids["faucet"]
-    )
-
-    DATA_GROUPS["OpenCabinetDrawer-v1"] = category_uids["cabinet_drawer"]
-    DATA_GROUPS["OpenCabinetDoor-v1"] = category_uids["cabinet_door"]
-    DATA_GROUPS["PushChair-v1"] = category_uids["chair"]
-    DATA_GROUPS["MoveBucket-v1"] = category_uids["bucket"]
-    DATA_GROUPS["TurnFaucet-v1"] = category_uids["faucet"]
-
-    # ---------------------------------------------------------------------------- #
-    # Interactable Scene Datasets
-    # ---------------------------------------------------------------------------- #
-    DATA_SOURCES["ReplicaCAD"] = DataSource(
-        source_type="scene",
-        hf_repo_id="haosulab/ReplicaCAD",
-        target_path="scene_datasets/replica_cad_dataset",
-    )
-
-    DATA_SOURCES["ReplicaCADRearrange"] = DataSource(
-        source_type="scene",
-        url="https://huggingface.co/datasets/haosulab/ReplicaCADRearrange/resolve/main/v1_extracted.zip",
-        target_path="scene_datasets/replica_cad_dataset/rearrange",
-    )
-
-    DATA_SOURCES["AI2THOR"] = DataSource(
-        source_type="scene",
-        url="https://huggingface.co/datasets/haosulab/AI2THOR/resolve/main/ai2thor.zip",
-        target_path="scene_datasets/ai2thor",
-    )
-
-
-def initialize_extra_sources():
-    DATA_SOURCES["xmate3_robotiq"] = DataSource(
-        source_type="robot",
-        url="https://storage1.ucsd.edu/datasets/ManiSkill2022-assets/xmate3_robotiq.zip",
-        target_path="robots/xmate3_robotiq",
-        checksum="ddda102a20eb41e28a0a501702e240e5d7f4084221a44f580e729f08b7c12d1a",
-    )
-    DATA_SOURCES["ur10e"] = DataSource(
-        source_type="robot",
-        url="https://github.com/haosulab/ManiSkill-UR10e/archive/refs/tags/v0.1.0.zip",
-        target_path="robots/ur10e",
-    )
-    DATA_SOURCES["anymal_c"] = DataSource(
-        source_type="robot",
-        url="https://github.com/haosulab/ManiSkill-ANYmalC/archive/refs/tags/v0.1.1.zip",
-        target_path="robots/anymal_c",
-    )
-    DATA_SOURCES["unitree_h1"] = DataSource(
-        source_type="robot",
-        url="https://github.com/haosulab/ManiSkill-UnitreeH1/archive/refs/tags/v0.1.0.zip",
-        target_path="robots/unitree_h1",
-    )
-    DATA_SOURCES["unitree_g1"] = DataSource(
-        source_type="robot",
-        url="https://github.com/haosulab/ManiSkill-UnitreeG1/archive/refs/tags/v0.1.0.zip",
-        target_path="robots/unitree_g1",
-    )
-    DATA_SOURCES["unitree_go2"] = DataSource(
-        source_type="robot",
-        url="https://github.com/haosulab/ManiSkill-UnitreeGo2/archive/refs/tags/v0.1.0.zip",
-        target_path="robots/unitree_go2",
-    )
-    DATA_SOURCES["stompy"] = DataSource(
-        source_type="robot",
-        url="https://github.com/haosulab/ManiSkill-Stompy/archive/refs/tags/v0.1.0.zip",
-        target_path="robots/stompy",
-    )
-    DATA_SOURCES["widowx250s"] = DataSource(
-        source_type="robot",
-        url="https://github.com/haosulab/ManiSkill-WidowX250S/archive/refs/tags/v0.1.0.zip",
-        target_path="robots/widowx",
-    )
-    DATA_SOURCES["googlerobot"] = DataSource(
-        source_type="robot",
-        url="https://github.com/haosulab/ManiSkill-GoogleRobot/archive/refs/tags/v0.1.0.zip",
-        target_path="robots/googlerobot",
-    )
-
-    # ---------------------------------------------------------------------------- #
-    # Visual backgrounds
-    # ---------------------------------------------------------------------------- #
-
-    # All backgrounds
-    # DATA_GROUPS["backgrounds"] = ["minimalistic_modern_bedroom"]
-
-    # TODO add Replica, MatterPort 3D?
+import mani_skill.envs  # import all environments to register them which auto registers data groups to allow asset download by environment ID.
+from mani_skill.utils import assets
+from mani_skill.utils.assets.data import DATA_GROUPS, DATA_SOURCES
 
 
 def prompt_yes_no(message):
@@ -251,7 +45,7 @@ def sha256sum(filename, chunk_size=4096):
 
 
 def download_from_hf_datasets(
-    data_source: DataSource,
+    data_source: assets.DataSource,
 ):
     output_dir = Path(data_source.output_dir)
     output_path = output_dir / data_source.target_path
@@ -265,10 +59,11 @@ def download_from_hf_datasets(
 
 
 def download(
-    data_source: DataSource,
+    data_source: assets.DataSource,
     verbose=True,
     non_interactive=True,
 ):
+    """download a given data source"""
     output_dir = Path(data_source.output_dir)
     # Create output directory
     if not output_dir.exists():
@@ -393,15 +188,11 @@ def parse_args(args=None):
 
 
 def main(args):
-    global DATA_SOURCES, DATA_GROUPS
     verbose = not args.quiet
 
-    initialize_sources()
-    initialize_extra_sources()
-
     if args.list:
         downloadable_ids = []
-        for k, v in DATA_SOURCES.items():
+        for k, v in assets.DATA_SOURCES.items():
             if v.source_type == args.list:
                 downloadable_ids.append(k)
         print(f"For category {args.list} the following asset UIDs are available")
@@ -409,17 +200,17 @@ def main(args):
         exit()
     if args.uid == "":
         print("Available asset (group) uids:")
-        print(list(DATA_GROUPS.keys()))
+        print(list(assets.DATA_GROUPS.keys()))
         return
     if args.uid == "all":
         if verbose:
             print("All assets will be downloaded. This may take a while.")
-        uids = list(DATA_SOURCES.keys())
+        uids = list(assets.DATA_SOURCES.keys())
         show_progress = True
-    elif args.uid in DATA_GROUPS:
-        uids = DATA_GROUPS[args.uid]
+    elif args.uid in assets.DATA_GROUPS:
+        uids = assets.expand_data_group_into_individual_data_source_ids(args.uid)
         show_progress = True
-    elif args.uid in DATA_SOURCES:
+    elif args.uid in assets.DATA_SOURCES:
         uids = [args.uid]
         show_progress = False
     else:
@@ -434,7 +225,7 @@ def main(args):
         kwargs["non_interactive"] = args.non_interactive
         if args.output_dir is not None:
             kwargs["output_dir"] = args.output_dir
-        output_path = download(DATA_SOURCES[uid], **kwargs)
+        output_path = download(assets.DATA_SOURCES[uid], **kwargs)
 
         if output_path is not None and verbose:
             print("=" * 80)
diff --git a/mani_skill/utils/registration.py b/mani_skill/utils/registration.py
index 1dcf0f67e..5e6c24dfb 100644
--- a/mani_skill/utils/registration.py
+++ b/mani_skill/utils/registration.py
@@ -4,13 +4,14 @@
 import sys
 from copy import deepcopy
 from functools import partial
-from typing import TYPE_CHECKING, Dict, Type
+from typing import TYPE_CHECKING, Dict, List, Optional, Type
 
 import gymnasium as gym
 from gymnasium.envs.registration import EnvSpec as GymEnvSpec
 from gymnasium.envs.registration import WrapperSpec
 
 from mani_skill import logger
+from mani_skill.utils import assets, download_asset
 from mani_skill.vector.wrappers.gymnasium import ManiSkillVectorEnv
 
 if TYPE_CHECKING:
@@ -23,17 +24,57 @@ def __init__(
         uid: str,
         cls: Type[BaseEnv],
         max_episode_steps=None,
+        asset_download_ids: Optional[List[str]] = [],
         default_kwargs: dict = None,
     ):
         """A specification for a ManiSkill environment."""
         self.uid = uid
         self.cls = cls
         self.max_episode_steps = max_episode_steps
+        self.asset_download_ids = asset_download_ids
         self.default_kwargs = {} if default_kwargs is None else default_kwargs
 
     def make(self, **kwargs):
         _kwargs = self.default_kwargs.copy()
         _kwargs.update(kwargs)
+
+        # check if all assets necessary are downloaded
+        assets_to_download = []
+        for asset_id in self.asset_download_ids or []:
+            is_data_group = asset_id in assets.DATA_GROUPS
+            if is_data_group:
+                found_data_group_assets = True
+                for (
+                    data_source_id
+                ) in assets.expand_data_group_into_individual_data_source_ids(asset_id):
+                    if not assets.is_data_source_downloaded(data_source_id):
+                        assets_to_download.append(data_source_id)
+                        found_data_group_assets = False
+                if not found_data_group_assets:
+                    print(
+                        f"Environment {self.uid} requires a set of assets in group {asset_id}. At least 1 of those assets could not be found"
+                    )
+            else:
+                if not assets.is_data_source_downloaded(asset_id):
+                    assets_to_download.append(asset_id)
+                    data_source = assets.DATA_SOURCES[asset_id]
+                    print(
+                        f"Could not find asset {asset_id} at {data_source.output_dir / data_source.target_path}"
+                    )
+        if len(assets_to_download) > 0:
+            if len(assets_to_download) <= 5:
+                asset_download_msg = ", ".join(assets_to_download)
+            else:
+                asset_download_msg = f"{assets_to_download[:5]} (and {len(assets_to_download) - 10} more)"
+            response = download_asset.prompt_yes_no(
+                f"Environment {self.uid} requires asset(s) {asset_download_msg} which could not be found. Would you like to download them now?"
+            )
+            if response:
+                for asset_id in assets_to_download:
+                    download_asset.download(assets.DATA_SOURCES[asset_id])
+            else:
+                print("Exiting as assets are not found or downloaded")
+                exit()
         return self.cls(**_kwargs)
 
     @property
@@ -52,7 +93,11 @@ def gym_spec(self):
 
 
 def register(
-    name: str, cls: Type[BaseEnv], max_episode_steps=None, default_kwargs: dict = None
+    name: str,
+    cls: Type[BaseEnv],
+    max_episode_steps=None,
+    asset_download_ids: List[str] = [],
+    default_kwargs: dict = None,
 ):
     """Register a ManiSkill environment."""
 
@@ -63,9 +108,19 @@ def register(
         logger.warn(f"Env {name} already registered")
     if not issubclass(cls, BaseEnv):
         raise TypeError(f"Env {name} must inherit from BaseEnv")
+
+    for asset_id in asset_download_ids:
+        if asset_id not in assets.DATA_SOURCES and asset_id not in assets.DATA_GROUPS:
+            raise KeyError(f"Asset {asset_id} not found in data sources or groups")
+
     REGISTERED_ENVS[name] = EnvSpec(
-        name, cls, max_episode_steps=max_episode_steps, default_kwargs=default_kwargs
+        name,
+        cls,
+        max_episode_steps=max_episode_steps,
+        asset_download_ids=asset_download_ids,
+        default_kwargs=default_kwargs,
     )
+    assets.DATA_GROUPS[name] = asset_download_ids
 
 
 class TimeLimitWrapper(gym.Wrapper):
@@ -119,12 +174,20 @@ def make_vec(env_id, **kwargs):
     return env
 
 
-def register_env(uid: str, max_episode_steps=None, override=False, **kwargs):
+def register_env(
+    uid: str,
+    max_episode_steps=None,
+    override=False,
+    asset_download_ids: List[str] = [],
+    **kwargs,
+):
     """A decorator to register ManiSkill environments.
 
     Args:
         uid (str): unique id of the environment.
         max_episode_steps (int): maximum number of steps in an episode.
+        asset_download_ids (List[str]): asset download ids the environment depends on. When environments are created
+            this list is checked to see if the user has all assets downloaded and if not, prompt the user if they wish to download them.
         override (bool): whether to override the environment if it is already registered.
 
     Notes:
@@ -156,6 +219,7 @@ def _register_env(cls):
             uid,
             cls,
             max_episode_steps=max_episode_steps,
+            asset_download_ids=asset_download_ids,
             default_kwargs=deepcopy(kwargs),
         )
 
diff --git a/mani_skill/utils/wrappers/record.py b/mani_skill/utils/wrappers/record.py
index 9c44ca9b9..14b81451e 100644
--- a/mani_skill/utils/wrappers/record.py
+++ b/mani_skill/utils/wrappers/record.py
@@ -14,6 +14,7 @@
 from mani_skill.envs.sapien_env import BaseEnv
 from mani_skill.utils import common, gym_utils
 from mani_skill.utils.io_utils import dump_json
+from mani_skill.utils.structs.types import Array
 from mani_skill.utils.visualization.misc import (
     images_to_video,
     put_info_on_image,
@@ -519,7 +520,9 @@ def flush_trajectory(
                 traj_id = "traj_{}".format(self._episode_id)
                 group = self._h5_file.create_group(traj_id, track_order=True)
 
-                def recursive_add_to_h5py(group: h5py.Group, data: dict, key):
+                def recursive_add_to_h5py(
+                    group: h5py.Group, data: Union[dict, Array], key
+                ):
                     """simple recursive data insertion for nested data structures into h5py, optimizing for visual data as well"""
                     if isinstance(data, dict):
                         subgrp = group.create_group(key, track_order=True)