diff --git a/app/celery/tasks.py b/app/celery/tasks.py
index 993bdea0f..46468e9fd 100644
--- a/app/celery/tasks.py
+++ b/app/celery/tasks.py
@@ -172,14 +172,18 @@ def process_job(job_id):
             )
 
         if template.template_type == 'email':
-            send_email.apply_async((
-                str(job.service_id),
-                create_uuid(),
-                '"{}" <{}@{}>'.format(
+            if service.reply_to_email_address:
+                from_email = service.reply_to_email_address
+            else:
+                from_email = '"{}" <{}@{}>'.format(
                     service.name,
                     service.email_from,
                     current_app.config['NOTIFY_EMAIL_DOMAIN']
-                ).encode('ascii', 'ignore').decode('ascii'),
+                )
+            send_email.apply_async((
+                str(job.service_id),
+                create_uuid(),
+                from_email.encode('ascii', 'ignore').decode('ascii'),
                 encrypted,
                 datetime.utcnow().strftime(DATETIME_FORMAT)),
                 queue='bulk-email')
diff --git a/tests/app/celery/test_tasks.py b/tests/app/celery/test_tasks.py
index 0d92dff29..ec537ceba 100644
--- a/tests/app/celery/test_tasks.py
+++ b/tests/app/celery/test_tasks.py
@@ -945,9 +945,58 @@ def test_email_reset_password_should_send_email(notify_db, notify_db_session, no
         aws_ses_client.send_email(current_app.config['VERIFY_CODE_FROM_EMAIL_ADDRESS'],
                                   reset_password_message['to'],
                                   "Reset password for GOV.UK Notify",
+
                                   message)
 
 
+def test_process_email_job_should_use_email_from_if_no_reply_to(sample_email_job, mocker, mock_celery_remove_job):
+    mocker.patch('app.celery.tasks.s3.get_job_from_s3', return_value=load_example_csv('email'))
+    mocker.patch('app.celery.tasks.send_email.apply_async')
+    mocker.patch('app.encryption.encrypt', return_value='something_encrypted')
+    mocker.patch('app.celery.tasks.create_uuid', return_value='uuid')
+
+    sample_email_job.service.reply_to_email_address = None
+
+    process_job(sample_email_job.id)
+
+    tasks.send_email.apply_async.assert_called_once_with(
+        (
+            str(sample_email_job.service_id),
+            'uuid',
+            "\"{}\" <{}@{}>".format(
+                sample_email_job.service.name,
+                sample_email_job.service.email_from,
+                current_app.config['NOTIFY_EMAIL_DOMAIN']
+            ),
+            'something_encrypted',
+            ANY
+        ),
+        queue="bulk-email"
+    )
+
+
+def test_process_email_job_should_use_reply_to_email(sample_email_job, mocker, mock_celery_remove_job):
+    mocker.patch('app.celery.tasks.s3.get_job_from_s3', return_value=load_example_csv('email'))
+    mocker.patch('app.celery.tasks.send_email.apply_async')
+    mocker.patch('app.encryption.encrypt', return_value='something_encrypted')
+    mocker.patch('app.celery.tasks.create_uuid', return_value='uuid')
+
+    sample_email_job.service.reply_to_email_address = 'somereply@testservice.gov.uk'
+
+    process_job(sample_email_job.id)
+
+    tasks.send_email.apply_async.assert_called_once_with(
+        (
+            str(sample_email_job.service_id),
+            'uuid',
+            'somereply@testservice.gov.uk',
+            'something_encrypted',
+            ANY
+        ),
+        queue="bulk-email"
+    )
+
+
 def _notification_json(template, to, personalisation=None, job_id=None):
     notification = {
         "template": template.id,