diff --git a/app/commands/delete_publish_intent.rb b/app/commands/delete_publish_intent.rb
index c9212bc56..68aea9777 100644
--- a/app/commands/delete_publish_intent.rb
+++ b/app/commands/delete_publish_intent.rb
@@ -2,7 +2,14 @@ module Commands
   class DeletePublishIntent < BaseCommand
     def call
       if downstream
-        PublishingAPI.service(:live_content_store).delete_publish_intent(base_path)
+        enqueue = ENV.fetch("ENQUEUE_PUBLISH_INTENTS", false)
+        if enqueue == "true"
+          DeletePublishIntentWorker.perform_async(
+            "base_path" => base_path,
+          )
+        else
+          PublishingAPI.service(:live_content_store).delete_publish_intent(base_path)
+        end
       end
 
       Success.new({})
diff --git a/app/commands/put_publish_intent.rb b/app/commands/put_publish_intent.rb
index 768affd45..0f5af6d59 100644
--- a/app/commands/put_publish_intent.rb
+++ b/app/commands/put_publish_intent.rb
@@ -5,7 +5,15 @@ def call
 
       if downstream
         payload = publish_intent
-        Adapters::ContentStore.put_publish_intent(base_path, payload)
+        enqueue = ENV.fetch("ENQUEUE_PUBLISH_INTENTS", false)
+        if enqueue == "true"
+          PutPublishIntentWorker.perform_async(
+            "base_path" => base_path,
+            "payload" => payload.to_json,
+          )
+        else
+          Adapters::ContentStore.put_publish_intent(base_path, payload)
+        end
       end
 
       Success.new(payload)
diff --git a/app/workers/delete_publish_intent_worker.rb b/app/workers/delete_publish_intent_worker.rb
new file mode 100644
index 000000000..49cbf6561
--- /dev/null
+++ b/app/workers/delete_publish_intent_worker.rb
@@ -0,0 +1,25 @@
+require "sidekiq-unique-jobs"
+
+class DeletePublishIntentWorker
+  include DownstreamQueue
+  include Sidekiq::Worker
+  include PerformAsyncInQueue
+
+  sidekiq_options queue: HIGH_QUEUE,
+                  lock: :until_executing,
+                  lock_args_method: :uniq_args,
+                  on_conflict: :log
+
+  def self.uniq_args(args)
+    [
+      args.first["base_path"],
+      name,
+    ]
+  end
+
+  def perform(args = {})
+    PublishingAPI.service(:live_content_store).delete_publish_intent(args["base_path"])
+  rescue AbortWorkerError => e
+    notify_airbrake(e, args)
+  end
+end
diff --git a/app/workers/put_publish_intent_worker.rb b/app/workers/put_publish_intent_worker.rb
new file mode 100644
index 000000000..82fb94e50
--- /dev/null
+++ b/app/workers/put_publish_intent_worker.rb
@@ -0,0 +1,25 @@
+require "sidekiq-unique-jobs"
+
+class PutPublishIntentWorker
+  include DownstreamQueue
+  include Sidekiq::Worker
+  include PerformAsyncInQueue
+
+  sidekiq_options queue: HIGH_QUEUE,
+                  lock: :until_executing,
+                  lock_args_method: :uniq_args,
+                  on_conflict: :log
+
+  def self.uniq_args(args)
+    [
+      args.first["base_path"],
+      name,
+    ]
+  end
+
+  def perform(args = {})
+    Adapters::ContentStore.put_publish_intent(args["base_path"], JSON.parse(args["payload"]))
+  rescue AbortWorkerError => e
+    notify_airbrake(e, args)
+  end
+end
diff --git a/spec/commands/delete_publish_intent_spec.rb b/spec/commands/delete_publish_intent_spec.rb
index cd199fdc6..f49477df3 100644
--- a/spec/commands/delete_publish_intent_spec.rb
+++ b/spec/commands/delete_publish_intent_spec.rb
@@ -9,6 +9,8 @@
     }
   end
 
+  let(:content_store) { PublishingAPI.service(:live_content_store) }
+
   it "responds successfully" do
     result = described_class.call(payload)
     expect(result).to be_a(Commands::Success)
@@ -16,10 +18,54 @@
 
   context "when the downstream flag is set to false" do
     it "does not send any downstream requests" do
-      expect(DownstreamDraftWorker).not_to receive(:perform_async)
-      expect(DownstreamLiveWorker).not_to receive(:perform_async)
+      expect(content_store).not_to receive(:delete_publish_intent)
+
+      described_class.call(payload, downstream: false)
+    end
+  end
+
+  context "when the downstream flag is set to false" do
+    it "does not send any downstream requests" do
+      expect(content_store).not_to receive(:delete_publish_intent)
 
       described_class.call(payload, downstream: false)
     end
   end
+
+  context "when the downstream flag is set to true" do
+    context "and the ENQUEUE_PUBLISH_INTENTS flag is true" do
+      before do
+        ENV["ENQUEUE_PUBLISH_INTENTS"] = "true"
+      end
+
+      it "enqueues a DeletePublishIntent job" do
+        expect(DeletePublishIntentWorker).to receive(:perform_async)
+        described_class.call(payload, downstream: true)
+      end
+    end
+
+    context "and the ENQUEUE_PUBLISH_INTENTS flag is not true" do
+      before do
+        ENV["ENQUEUE_PUBLISH_INTENTS"] = "no"
+      end
+
+      it "does send downstream requests" do
+        expect(content_store).to receive(:delete_publish_intent)
+
+        described_class.call(payload, downstream: true)
+      end
+    end
+
+    context "and the ENQUEUE_PUBLISH_INTENTS flag is not present" do
+      before do
+        ENV.delete("ENQUEUE_PUBLISH_INTENTS")
+      end
+
+      it "does send downstream requests" do
+        expect(content_store).to receive(:delete_publish_intent)
+
+        described_class.call(payload, downstream: true)
+      end
+    end
+  end
 end
diff --git a/spec/commands/put_publish_intent_spec.rb b/spec/commands/put_publish_intent_spec.rb
index 2047c5bb7..3a0a8db32 100644
--- a/spec/commands/put_publish_intent_spec.rb
+++ b/spec/commands/put_publish_intent_spec.rb
@@ -17,10 +17,46 @@
 
   context "when the downstream flag is set to false" do
     it "does not send any downstream requests" do
-      expect(DownstreamDraftWorker).not_to receive(:perform_async)
-      expect(DownstreamLiveWorker).not_to receive(:perform_async)
+      expect(Adapters::ContentStore).not_to receive(:put_publish_intent)
 
       described_class.call(payload, downstream: false)
     end
   end
+
+  context "when the downstream flag is set to true" do
+    context "and the ENQUEUE_PUBLISH_INTENTS flag is true" do
+      before do
+        ENV["ENQUEUE_PUBLISH_INTENTS"] = "true"
+      end
+
+      it "enqueues a PutPublishIntent job" do
+        expect(PutPublishIntentWorker).to receive(:perform_async)
+        described_class.call(payload, downstream: true)
+      end
+    end
+
+    context "and the ENQUEUE_PUBLISH_INTENTS flag is not true" do
+      before do
+        ENV["ENQUEUE_PUBLISH_INTENTS"] = "no"
+      end
+
+      it "does send downstream requests" do
+        expect(Adapters::ContentStore).to receive(:put_publish_intent)
+
+        described_class.call(payload, downstream: true)
+      end
+    end
+
+    context "and the ENQUEUE_PUBLISH_INTENTS flag is not present" do
+      before do
+        ENV.delete("ENQUEUE_PUBLISH_INTENTS")
+      end
+
+      it "does send downstream requests" do
+        expect(Adapters::ContentStore).to receive(:put_publish_intent)
+
+        described_class.call(payload, downstream: true)
+      end
+    end
+  end
 end