diff --git a/taskflow/storage.py b/taskflow/storage.py
index 449aafb6..9bd3affd 100644
--- a/taskflow/storage.py
+++ b/taskflow/storage.py
@@ -760,6 +760,13 @@ class Storage(object):
         self._flowdetail.state = state
         self._with_connection(self._save_flow_detail)
 
+    @lock_utils.write_locked
+    def update_flow_metadata(self, update_with):
+        """Update flowdetails metadata and save it."""
+        if update_with:
+            self._flowdetail.meta.update(update_with)
+            self._with_connection(self._save_flow_detail)
+
     @lock_utils.read_locked
     def get_flow_state(self):
         """Get state from flow details."""
diff --git a/taskflow/tests/unit/test_storage.py b/taskflow/tests/unit/test_storage.py
index a27a811c..826c0467 100644
--- a/taskflow/tests/unit/test_storage.py
+++ b/taskflow/tests/unit/test_storage.py
@@ -200,6 +200,12 @@ class StorageTestMixin(object):
                                 "^Name 'xxx' is not mapped",
                                 s.fetch, 'xxx')
 
+    def test_flow_metadata_update(self):
+        s = self._get_storage()
+        update_with = {'test_data': True}
+        s.update_flow_metadata(update_with)
+        self.assertTrue(s._flowdetail.meta['test_data'])
+
     def test_task_metadata_update_with_none(self):
         s = self._get_storage()
         s.ensure_atom(test_utils.NoopTask('my task'))