diff --git a/src/parenttext_pipeline/pipelines.py b/src/parenttext_pipeline/pipelines.py
index 76bdf4d..a1cb8ec 100644
--- a/src/parenttext_pipeline/pipelines.py
+++ b/src/parenttext_pipeline/pipelines.py
@@ -6,6 +6,8 @@
 from rpft.converters import create_flows
 from rapidpro_abtesting.main import apply_abtests
 
+from parenttext_pipeline.steps import update_expiration_time
+
 
 @dataclass(kw_only=True)
 class Config:
@@ -143,14 +145,12 @@ def run_pipeline(
         output_file_name_1_2 = source_file_name + "_1_2_modified_expiration_times"
         output_path_1_2 = os.path.join(outputpath, output_file_name_1_2 + ".json")
 
-        subprocess.run([
-            "node",
-            "./node_modules/@idems/idems-chatbot-tools/update_expiration_time.js",
+        update_expiration_time(
             input_path_1_2,
-            special_expiration,
             default_expiration,
-            output_path_1_2,
-        ])
+            special_expiration,
+            output_path_1_2
+        )
 
         print("Step 1 complete, created " + source_file_name + " and modified expiration times")
 
diff --git a/src/parenttext_pipeline/steps.py b/src/parenttext_pipeline/steps.py
new file mode 100644
index 0000000..199d695
--- /dev/null
+++ b/src/parenttext_pipeline/steps.py
@@ -0,0 +1,26 @@
+import json
+
+
+def update_expiration_time(in_fp, default, specifics_fp, out_fp):
+    with open(specifics_fp, 'r') as specifics_json:
+        specifics = json.load(specifics_json)
+
+    with open(in_fp, "r") as in_json:
+        org = json.load(in_json)
+
+    for flow in org.get("flows", []):
+        set_expiration(flow, default, specifics)
+
+    with open(out_fp, "w") as out_json:
+        json.dump(org, out_json)
+
+
+def set_expiration(flow, default, specifics={}):
+    expiration = specifics.get(flow["name"], default)
+
+    flow["expire_after_minutes"] = expiration
+
+    if "expires" in flow.get("metadata", {}):
+        flow["metadata"]["expires"] = expiration
+
+    return flow
diff --git a/tests/test_set_expiration.py b/tests/test_set_expiration.py
new file mode 100644
index 0000000..00c2f41
--- /dev/null
+++ b/tests/test_set_expiration.py
@@ -0,0 +1,56 @@
+from unittest import TestCase
+
+from parenttext_pipeline.steps import set_expiration
+
+
+class TestUpdateExpirationTime(TestCase):
+
+    def test_must_set_expiration_to_default(self):
+        default = 1440
+        flow = {
+            "name": "flow_1",
+            "expire_after_minutes": 0,
+        }
+        updated = set_expiration(flow, default)
+        self.assertEqual(updated["expire_after_minutes"], 1440)
+        self.assertFalse("metadata" in updated)
+
+    def test_must_set_metadata_expires_if_exists(self):
+        default = 360
+        flow = {
+            "name": "flow_1",
+            "expire_after_minutes": 0,
+            "metadata": {
+                "expires": 0
+            }
+        }
+        updated = set_expiration(flow, default)
+        self.assertEqual(updated["metadata"]["expires"], 360)
+        self.assertEqual(updated["expire_after_minutes"], 360)
+
+    def test_must_not_create_metadata_expires(self):
+        default = 480
+        flow = {
+            "name": "flow_1",
+            "expire_after_minutes": 0,
+            "metadata": {}
+        }
+        updated = set_expiration(flow, default)
+        self.assertFalse("expires" in updated["metadata"])
+        self.assertEqual(updated["expire_after_minutes"], 480)
+
+    def test_must_allow_per_flow_expiration_times(self):
+        default = 120
+        specifics = {
+            "flow_1": 240,
+        }
+        flow = {
+            "name": "flow_1",
+            "expire_after_minutes": 0,
+            "metadata": {
+                "expires": 0
+            },
+        }
+        updated = set_expiration(flow, default, specifics)
+        self.assertEqual(updated["expire_after_minutes"], 240)
+        self.assertEqual(updated["metadata"]["expires"], 240)