diff --git a/awx/api/serializers.py b/awx/api/serializers.py
index 08073a7bc0..2409dc1e74 100644
--- a/awx/api/serializers.py
+++ b/awx/api/serializers.py
@@ -2191,7 +2191,7 @@ class WorkflowJobTemplateSerializer(LabelsListMixin, UnifiedJobTemplateSerialize
 
     class Meta:
         model = WorkflowJobTemplate
-        fields = ('*', 'extra_vars', 'organization')
+        fields = ('*', 'extra_vars', 'organization', 'survey_enabled',)
 
     def get_related(self, obj):
         res = super(WorkflowJobTemplateSerializer, self).get_related(obj)
@@ -2205,7 +2205,7 @@ class WorkflowJobTemplateSerializer(LabelsListMixin, UnifiedJobTemplateSerialize
             #notification_templates_any = reverse('api:system_job_template_notification_templates_any_list', args=(obj.pk,)),
             #notification_templates_success = reverse('api:system_job_template_notification_templates_success_list', args=(obj.pk,)),
             #notification_templates_error = reverse('api:system_job_template_notification_templates_error_list', args=(obj.pk,)),
-
+            survey_spec = reverse('api:workflow_job_template_survey_spec', args=(obj.pk,)),
         ))
         return res
 
@@ -2236,6 +2236,14 @@ class WorkflowJobSerializer(LabelsListMixin, UnifiedJobSerializer):
             res['cancel'] = reverse('api:workflow_job_cancel', args=(obj.pk,))
         return res
 
+    def to_representation(self, obj):
+        ret = super(WorkflowJobSerializer, self).to_representation(obj)
+        if obj is None:
+            return ret
+        if 'extra_vars' in ret:
+            ret['extra_vars'] = obj.display_extra_vars()
+        return ret
+
 # TODO:
 class WorkflowJobListSerializer(WorkflowJobSerializer, UnifiedJobListSerializer):
     pass
diff --git a/awx/api/urls.py b/awx/api/urls.py
index bdb9063c39..4699d9f2bf 100644
--- a/awx/api/urls.py
+++ b/awx/api/urls.py
@@ -263,6 +263,7 @@ workflow_job_template_urls = patterns('awx.api.views',
     url(r'^(?P<pk>[0-9]+)/jobs/$',                      'workflow_job_template_jobs_list'),
     url(r'^(?P<pk>[0-9]+)/launch/$',                    'workflow_job_template_launch'),
     url(r'^(?P<pk>[0-9]+)/schedules/$',                 'workflow_job_template_schedules_list'),
+    url(r'^(?P<pk>[0-9]+)/survey_spec/$',               'workflow_job_template_survey_spec'),
     url(r'^(?P<pk>[0-9]+)/workflow_nodes/$',            'workflow_job_template_workflow_nodes_list'),
     url(r'^(?P<pk>[0-9]+)/labels/$',                    'workflow_job_template_label_list'),
 #    url(r'^(?P<pk>[0-9]+)/cancel/$',                    'workflow_job_template_cancel'),
diff --git a/awx/api/views.py b/awx/api/views.py
index cc0cb8e100..0ed269bc08 100644
--- a/awx/api/views.py
+++ b/awx/api/views.py
@@ -2376,6 +2376,11 @@ class JobTemplateSurveySpec(GenericAPIView):
         obj.save()
         return Response()
 
+class WorkflowJobTemplateSurveySpec(JobTemplateSurveySpec):
+
+    model = WorkflowJobTemplate
+    parent_model = WorkflowJobTemplate
+
 class JobTemplateActivityStreamList(SubListAPIView):
 
     model = ActivityStream
@@ -2792,6 +2797,7 @@ class WorkflowJobTemplateLaunch(GenericAPIView):
         data = {}
         obj = self.get_object()
         data['warnings'] = obj.get_warnings()
+        data['passwords_needed_to_start'] = obj.passwords_needed_to_start
         return Response(data)
 
     def post(self, request, *args, **kwargs):
@@ -2799,9 +2805,12 @@ class WorkflowJobTemplateLaunch(GenericAPIView):
         if not request.user.can_access(self.model, 'start', obj):
             raise PermissionDenied()
 
-        new_job = obj.create_unified_job(**request.data)
-        new_job.signal_start(**request.data)
+        prompted_fields, ignored_fields = obj._accept_or_ignore_job_kwargs(**request.data)
+
+        new_job = obj.create_unified_job(**prompted_fields)
+        new_job.signal_start(**prompted_fields)
         data = dict(workflow_job=new_job.id)
+        data['ignored_fields'] = ignored_fields
         return Response(data, status=status.HTTP_201_CREATED)
 
 # TODO:
diff --git a/awx/main/migrations/0041_v310_workflow_surveys.py b/awx/main/migrations/0041_v310_workflow_surveys.py
new file mode 100644
index 0000000000..1df93bfe35
--- /dev/null
+++ b/awx/main/migrations/0041_v310_workflow_surveys.py
@@ -0,0 +1,30 @@
+# -*- coding: utf-8 -*-
+from __future__ import unicode_literals
+
+from django.db import migrations, models
+import jsonfield.fields
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('main', '0040_v310_artifacts'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='workflowjob',
+            name='survey_passwords',
+            field=jsonfield.fields.JSONField(default={}, editable=False, blank=True),
+        ),
+        migrations.AddField(
+            model_name='workflowjobtemplate',
+            name='survey_enabled',
+            field=models.BooleanField(default=False),
+        ),
+        migrations.AddField(
+            model_name='workflowjobtemplate',
+            name='survey_spec',
+            field=jsonfield.fields.JSONField(default={}, blank=True),
+        ),
+    ]
diff --git a/awx/main/models/jobs.py b/awx/main/models/jobs.py
index 5c8599052b..a5bc0d0a27 100644
--- a/awx/main/models/jobs.py
+++ b/awx/main/models/jobs.py
@@ -5,7 +5,6 @@
 import datetime
 import hmac
 import json
-import yaml
 import logging
 import time
 from urlparse import urljoin
@@ -33,7 +32,11 @@ from awx.main.models.notifications import (
     NotificationTemplate,
     JobNotificationMixin,
 )
-from awx.main.utils import ignore_inventory_computed_fields
+from awx.main.utils import (
+    decrypt_field,
+    ignore_inventory_computed_fields,
+    parse_yaml_or_json,
+)
 from awx.main.redact import PlainTextCleaner
 from awx.main.fields import ImplicitRoleField
 from awx.main.models.mixins import ResourceMixin
@@ -188,7 +191,7 @@ class JobOptions(BaseModel):
         else:
             return []
 
-class JobTemplate(UnifiedJobTemplate, JobOptions, ResourceMixin):
+class JobTemplate(UnifiedJobTemplate, SurveyJobTemplate, JobOptions, ResourceMixin):
     '''
     A job template is a reusable job definition for applying a project (with
     playbook) to an inventory source with a given credential.
@@ -232,15 +235,6 @@ class JobTemplate(UnifiedJobTemplate, JobOptions, ResourceMixin):
         blank=True,
         default=False,
     )
-
-    survey_enabled = models.BooleanField(
-        default=False,
-    )
-
-    survey_spec = JSONField(
-        blank=True,
-        default={},
-    )
     admin_role = ImplicitRoleField(
         parent_role=['project.organization.admin_role', 'inventory.organization.admin_role']
     )
@@ -318,125 +312,6 @@ class JobTemplate(UnifiedJobTemplate, JobOptions, ResourceMixin):
                 not self.passwords_needed_to_start and
                 not self.variables_needed_to_start)
 
-    @property
-    def variables_needed_to_start(self):
-        vars = []
-        if self.survey_enabled and 'spec' in self.survey_spec:
-            for survey_element in self.survey_spec['spec']:
-                if survey_element['required']:
-                    vars.append(survey_element['variable'])
-        return vars
-
-    def survey_password_variables(self):
-        vars = []
-        if self.survey_enabled and 'spec' in self.survey_spec:
-            # Get variables that are type password
-            for survey_element in self.survey_spec['spec']:
-                if survey_element['type'] == 'password':
-                    vars.append(survey_element['variable'])
-        return vars
-
-    def survey_variable_validation(self, data):
-        errors = []
-        if not self.survey_enabled:
-            return errors
-        if 'name' not in self.survey_spec:
-            errors.append("'name' missing from survey spec.")
-        if 'description' not in self.survey_spec:
-            errors.append("'description' missing from survey spec.")
-        for survey_element in self.survey_spec.get("spec", []):
-            if survey_element['variable'] not in data and \
-               survey_element['required']:
-                errors.append("'%s' value missing" % survey_element['variable'])
-            elif survey_element['type'] in ["textarea", "text", "password"]:
-                if survey_element['variable'] in data:
-                    if type(data[survey_element['variable']]) not in (str, unicode):
-                        errors.append("Value %s for '%s' expected to be a string." % (data[survey_element['variable']],
-                                                                                      survey_element['variable']))
-                        continue
-                    if 'min' in survey_element and survey_element['min'] not in ["", None] and len(data[survey_element['variable']]) < int(survey_element['min']):
-                        errors.append("'%s' value %s is too small (length is %s must be at least %s)." %
-                                      (survey_element['variable'], data[survey_element['variable']], len(data[survey_element['variable']]), survey_element['min']))
-                    if 'max' in survey_element and survey_element['max'] not in ["", None] and len(data[survey_element['variable']]) > int(survey_element['max']):
-                        errors.append("'%s' value %s is too large (must be no more than %s)." %
-                                      (survey_element['variable'], data[survey_element['variable']], survey_element['max']))
-            elif survey_element['type'] == 'integer':
-                if survey_element['variable'] in data:
-                    if type(data[survey_element['variable']]) != int:
-                        errors.append("Value %s for '%s' expected to be an integer." % (data[survey_element['variable']],
-                                                                                        survey_element['variable']))
-                        continue
-                    if 'min' in survey_element and survey_element['min'] not in ["", None] and survey_element['variable'] in data and \
-                       data[survey_element['variable']] < int(survey_element['min']):
-                        errors.append("'%s' value %s is too small (must be at least %s)." %
-                                      (survey_element['variable'], data[survey_element['variable']], survey_element['min']))
-                    if 'max' in survey_element and survey_element['max'] not in ["", None] and survey_element['variable'] in data and \
-                       data[survey_element['variable']] > int(survey_element['max']):
-                        errors.append("'%s' value %s is too large (must be no more than %s)." %
-                                      (survey_element['variable'], data[survey_element['variable']], survey_element['max']))
-            elif survey_element['type'] == 'float':
-                if survey_element['variable'] in data:
-                    if type(data[survey_element['variable']]) not in (float, int):
-                        errors.append("Value %s for '%s' expected to be a numeric type." % (data[survey_element['variable']],
-                                                                                            survey_element['variable']))
-                        continue
-                    if 'min' in survey_element and survey_element['min'] not in ["", None] and data[survey_element['variable']] < float(survey_element['min']):
-                        errors.append("'%s' value %s is too small (must be at least %s)." %
-                                      (survey_element['variable'], data[survey_element['variable']], survey_element['min']))
-                    if 'max' in survey_element and survey_element['max'] not in ["", None] and data[survey_element['variable']] > float(survey_element['max']):
-                        errors.append("'%s' value %s is too large (must be no more than %s)." %
-                                      (survey_element['variable'], data[survey_element['variable']], survey_element['max']))
-            elif survey_element['type'] == 'multiselect':
-                if survey_element['variable'] in data:
-                    if type(data[survey_element['variable']]) != list:
-                        errors.append("'%s' value is expected to be a list." % survey_element['variable'])
-                    else:
-                        for val in data[survey_element['variable']]:
-                            if val not in survey_element['choices']:
-                                errors.append("Value %s for '%s' expected to be one of %s." % (val, survey_element['variable'],
-                                                                                               survey_element['choices']))
-            elif survey_element['type'] == 'multiplechoice':
-                if survey_element['variable'] in data:
-                    if data[survey_element['variable']] not in survey_element['choices']:
-                        errors.append("Value %s for '%s' expected to be one of %s." % (data[survey_element['variable']],
-                                                                                       survey_element['variable'],
-                                                                                       survey_element['choices']))
-        return errors
-
-    def _update_unified_job_kwargs(self, **kwargs):
-        if 'launch_type' in kwargs and kwargs['launch_type'] == 'relaunch':
-            return kwargs
-
-        # Job Template extra_vars
-        extra_vars = self.extra_vars_dict
-
-        # Overwrite with job template extra vars with survey default vars
-        if self.survey_enabled and 'spec' in self.survey_spec:
-            for survey_element in self.survey_spec.get("spec", []):
-                if 'default' in survey_element and survey_element['default']:
-                    extra_vars[survey_element['variable']] = survey_element['default']
-
-        # transform to dict
-        if 'extra_vars' in kwargs:
-            kwargs_extra_vars = kwargs['extra_vars']
-            if not isinstance(kwargs_extra_vars, dict):
-                try:
-                    kwargs_extra_vars = json.loads(kwargs_extra_vars)
-                except Exception:
-                    try:
-                        kwargs_extra_vars = yaml.safe_load(kwargs_extra_vars)
-                        assert isinstance(kwargs_extra_vars, dict)
-                    except:
-                        kwargs_extra_vars = {}
-        else:
-            kwargs_extra_vars = {}
-
-        # Overwrite job template extra vars with explicit job extra vars
-        # and add on job extra vars
-        extra_vars.update(kwargs_extra_vars)
-        kwargs['extra_vars'] = json.dumps(extra_vars)
-        return kwargs
-
     def _ask_for_vars_dict(self):
         return dict(
             extra_vars=self.ask_variables_on_launch,
@@ -466,16 +341,7 @@ class JobTemplate(UnifiedJobTemplate, JobOptions, ResourceMixin):
                     if field == 'extra_vars' and self.survey_enabled and self.survey_spec:
                         # Accept vars defined in the survey and no others
                         survey_vars = [question['variable'] for question in self.survey_spec.get('spec', [])]
-                        extra_vars = kwargs[field]
-                        if isinstance(extra_vars, basestring):
-                            try:
-                                extra_vars = json.loads(extra_vars)
-                            except (ValueError, TypeError):
-                                try:
-                                    extra_vars = yaml.safe_load(extra_vars)
-                                    assert isinstance(extra_vars, dict)
-                                except (yaml.YAMLError, TypeError, AttributeError, AssertionError):
-                                    extra_vars = {}
+                        extra_vars = parse_yaml_or_json(kwargs[field])
                         for key in extra_vars:
                             if key in survey_vars:
                                 prompted_fields[field][key] = extra_vars[key]
@@ -529,7 +395,7 @@ class JobTemplate(UnifiedJobTemplate, JobOptions, ResourceMixin):
             any_notification_templates = set(any_notification_templates + list(base_notification_templates.filter(organization_notification_templates_for_any=self.project.organization)))
         return dict(error=list(error_notification_templates), success=list(success_notification_templates), any=list(any_notification_templates))
 
-class Job(UnifiedJob, JobOptions, JobNotificationMixin):
+class Job(UnifiedJob, SurveyJob, JobOptions, JobNotificationMixin):
     '''
     A job applies a project (with playbook) to an inventory source with a given
     credential.  It represents a single invocation of ansible-playbook with the
@@ -554,11 +420,6 @@ class Job(UnifiedJob, JobOptions, JobNotificationMixin):
         editable=False,
         through='JobHostSummary',
     )
-    survey_passwords = JSONField(
-        blank=True,
-        default={},
-        editable=False,
-    )
     artifacts = JSONField(
         blank=True,
         default={},
@@ -732,19 +593,6 @@ class Job(UnifiedJob, JobOptions, JobNotificationMixin):
         evars.update(extra_vars)
         self.update_fields(extra_vars=json.dumps(evars))
 
-    def display_extra_vars(self):
-        '''
-        Hides fields marked as passwords in survey.
-        '''
-        if self.survey_passwords:
-            extra_vars = json.loads(self.extra_vars)
-            for key, value in self.survey_passwords.items():
-                if key in extra_vars:
-                    extra_vars[key] = value
-            return json.dumps(extra_vars)
-        else:
-            return self.extra_vars
-
     def display_artifacts(self):
         '''
         Hides artifacts if they are marked as no_log type artifacts.
diff --git a/awx/main/models/mixins.py b/awx/main/models/mixins.py
index 3f9f2043d5..9cb6b18b77 100644
--- a/awx/main/models/mixins.py
+++ b/awx/main/models/mixins.py
@@ -2,6 +2,7 @@
 from django.db import models
 from django.contrib.contenttypes.models import ContentType
 from django.contrib.auth.models import User # noqa
+from jsonfield import JSONField
 
 # AWX
 from awx.main.models.rbac import (
diff --git a/awx/main/models/unified_jobs.py b/awx/main/models/unified_jobs.py
index e9925a6b17..da37caeebd 100644
--- a/awx/main/models/unified_jobs.py
+++ b/awx/main/models/unified_jobs.py
@@ -36,7 +36,7 @@ from awx.main.utils import decrypt_field, _inventory_updates
 from awx.main.redact import UriCleaner, REPLACE_STR
 from awx.main.consumers import emit_channel_notification
 
-__all__ = ['UnifiedJobTemplate', 'UnifiedJob']
+__all__ = ['UnifiedJobTemplate', 'UnifiedJob', 'SurveyJobTemplate', 'SurveyJob']
 
 logger = logging.getLogger('awx.main.models.unified_jobs')
 
@@ -937,3 +937,154 @@ class UnifiedJob(PolymorphicModel, PasswordFieldsModel, CommonModelNameNotUnique
             if settings.BROKER_URL.startswith('amqp://'):
                 self._force_cancel()
         return self.cancel_flag
+
+class SurveyJobTemplate(models.Model):
+    class Meta:
+        abstract = True
+
+    survey_enabled = models.BooleanField(
+        default=False,
+    )
+    survey_spec = JSONField(
+        blank=True,
+        default={},
+    )
+
+    def survey_password_variables(self):
+        vars = []
+        if self.survey_enabled and 'spec' in self.survey_spec:
+            # Get variables that are type password
+            for survey_element in self.survey_spec['spec']:
+                if survey_element['type'] == 'password':
+                    vars.append(survey_element['variable'])
+        return vars
+
+    @property
+    def variables_needed_to_start(self):
+        vars = []
+        if self.survey_enabled and 'spec' in self.survey_spec:
+            for survey_element in self.survey_spec['spec']:
+                if survey_element['required']:
+                    vars.append(survey_element['variable'])
+        return vars
+
+    def _update_unified_job_kwargs(self, **kwargs):
+        '''
+        Combine extra_vars with variable precedence order:
+          JT extra_vars -> JT survey defaults -> runtime extra_vars
+        '''
+        if 'launch_type' in kwargs and kwargs['launch_type'] == 'relaunch':
+            return kwargs
+
+        # Job Template extra_vars
+        extra_vars = self.extra_vars_dict
+
+        # Overwrite with job template extra vars with survey default vars
+        if self.survey_enabled and 'spec' in self.survey_spec:
+            for survey_element in self.survey_spec.get("spec", []):
+                if 'default' in survey_element and survey_element['default']:
+                    extra_vars[survey_element['variable']] = survey_element['default']
+
+        # transform to dict
+        if 'extra_vars' in kwargs:
+            kwargs_extra_vars = kwargs['extra_vars']
+            kwargs_extra_vars = parse_yaml_or_json(kwargs_extra_vars)
+        else:
+            kwargs_extra_vars = {}
+
+        # Overwrite job template extra vars with explicit job extra vars
+        # and add on job extra vars
+        extra_vars.update(kwargs_extra_vars)
+        kwargs['extra_vars'] = json.dumps(extra_vars)
+        return kwargs
+
+    def survey_variable_validation(self, data):
+        errors = []
+        if not self.survey_enabled:
+            return errors
+        if 'name' not in self.survey_spec:
+            errors.append("'name' missing from survey spec.")
+        if 'description' not in self.survey_spec:
+            errors.append("'description' missing from survey spec.")
+        for survey_element in self.survey_spec.get("spec", []):
+            if survey_element['variable'] not in data and \
+               survey_element['required']:
+                errors.append("'%s' value missing" % survey_element['variable'])
+            elif survey_element['type'] in ["textarea", "text", "password"]:
+                if survey_element['variable'] in data:
+                    if type(data[survey_element['variable']]) not in (str, unicode):
+                        errors.append("Value %s for '%s' expected to be a string." % (data[survey_element['variable']],
+                                                                                      survey_element['variable']))
+                        continue
+                    if 'min' in survey_element and survey_element['min'] not in ["", None] and len(data[survey_element['variable']]) < int(survey_element['min']):
+                        errors.append("'%s' value %s is too small (length is %s must be at least %s)." %
+                                      (survey_element['variable'], data[survey_element['variable']], len(data[survey_element['variable']]), survey_element['min']))
+                    if 'max' in survey_element and survey_element['max'] not in ["", None] and len(data[survey_element['variable']]) > int(survey_element['max']):
+                        errors.append("'%s' value %s is too large (must be no more than %s)." %
+                                      (survey_element['variable'], data[survey_element['variable']], survey_element['max']))
+            elif survey_element['type'] == 'integer':
+                if survey_element['variable'] in data:
+                    if type(data[survey_element['variable']]) != int:
+                        errors.append("Value %s for '%s' expected to be an integer." % (data[survey_element['variable']],
+                                                                                        survey_element['variable']))
+                        continue
+                    if 'min' in survey_element and survey_element['min'] not in ["", None] and survey_element['variable'] in data and \
+                       data[survey_element['variable']] < int(survey_element['min']):
+                        errors.append("'%s' value %s is too small (must be at least %s)." %
+                                      (survey_element['variable'], data[survey_element['variable']], survey_element['min']))
+                    if 'max' in survey_element and survey_element['max'] not in ["", None] and survey_element['variable'] in data and \
+                       data[survey_element['variable']] > int(survey_element['max']):
+                        errors.append("'%s' value %s is too large (must be no more than %s)." %
+                                      (survey_element['variable'], data[survey_element['variable']], survey_element['max']))
+            elif survey_element['type'] == 'float':
+                if survey_element['variable'] in data:
+                    if type(data[survey_element['variable']]) not in (float, int):
+                        errors.append("Value %s for '%s' expected to be a numeric type." % (data[survey_element['variable']],
+                                                                                            survey_element['variable']))
+                        continue
+                    if 'min' in survey_element and survey_element['min'] not in ["", None] and data[survey_element['variable']] < float(survey_element['min']):
+                        errors.append("'%s' value %s is too small (must be at least %s)." %
+                                      (survey_element['variable'], data[survey_element['variable']], survey_element['min']))
+                    if 'max' in survey_element and survey_element['max'] not in ["", None] and data[survey_element['variable']] > float(survey_element['max']):
+                        errors.append("'%s' value %s is too large (must be no more than %s)." %
+                                      (survey_element['variable'], data[survey_element['variable']], survey_element['max']))
+            elif survey_element['type'] == 'multiselect':
+                if survey_element['variable'] in data:
+                    if type(data[survey_element['variable']]) != list:
+                        errors.append("'%s' value is expected to be a list." % survey_element['variable'])
+                    else:
+                        for val in data[survey_element['variable']]:
+                            if val not in survey_element['choices']:
+                                errors.append("Value %s for '%s' expected to be one of %s." % (val, survey_element['variable'],
+                                                                                               survey_element['choices']))
+            elif survey_element['type'] == 'multiplechoice':
+                if survey_element['variable'] in data:
+                    if data[survey_element['variable']] not in survey_element['choices']:
+                        errors.append("Value %s for '%s' expected to be one of %s." % (data[survey_element['variable']],
+                                                                                       survey_element['variable'],
+                                                                                       survey_element['choices']))
+        return errors
+
+
+class SurveyJob(models.Model):
+    class Meta:
+        abstract = True
+
+    survey_passwords = JSONField(
+        blank=True,
+        default={},
+        editable=False,
+    )
+
+    def display_extra_vars(self):
+        '''
+        Hides fields marked as passwords in survey.
+        '''
+        if self.survey_passwords:
+            extra_vars = json.loads(self.extra_vars)
+            for key, value in self.survey_passwords.items():
+                if key in extra_vars:
+                    extra_vars[key] = value
+            return json.dumps(extra_vars)
+        else:
+            return self.extra_vars
diff --git a/awx/main/models/workflow.py b/awx/main/models/workflow.py
index ff457fd207..a25bfebb3c 100644
--- a/awx/main/models/workflow.py
+++ b/awx/main/models/workflow.py
@@ -13,6 +13,7 @@ from jsonfield import JSONField
 
 # AWX
 from awx.main.models import UnifiedJobTemplate, UnifiedJob
+from awx.main.models.unified_jobs import SurveyJobTemplate, SurveyJob
 from awx.main.models.notifications import JobNotificationMixin
 from awx.main.models.base import BaseModel, CreatedModifiedModel, VarsDictProperty
 from awx.main.models.rbac import (
@@ -22,6 +23,7 @@ from awx.main.models.rbac import (
 from awx.main.fields import ImplicitRoleField
 from awx.main.models.mixins import ResourceMixin
 from awx.main.redact import REPLACE_STR
+from awx.main.utils import parse_yaml_or_json
 
 from copy import copy
 
@@ -260,7 +262,9 @@ class WorkflowJobOptions(BaseModel):
         default='',
     )
 
-class WorkflowJobTemplate(UnifiedJobTemplate, WorkflowJobOptions, ResourceMixin):
+    extra_vars_dict = VarsDictProperty('extra_vars', True)
+
+class WorkflowJobTemplate(UnifiedJobTemplate, SurveyJobTemplate, WorkflowJobOptions, ResourceMixin):
 
     class Meta:
         app_label = 'main'
@@ -318,6 +322,25 @@ class WorkflowJobTemplate(UnifiedJobTemplate, WorkflowJobOptions, ResourceMixin)
         workflow_job.inherit_job_template_workflow_nodes()
         return workflow_job
 
+    def _accept_or_ignore_job_kwargs(self, extra_vars=None, **kwargs):
+        # Only accept allowed survey variables
+        ignored_fields = {}
+        prompted_fields = {}
+        prompted_fields['extra_vars'] = {}
+        ignored_fields['extra_vars'] = {}
+        extra_vars = parse_yaml_or_json(extra_vars)
+        if self.survey_enabled and self.survey_spec:
+            survey_vars = [question['variable'] for question in self.survey_spec.get('spec', [])]
+            for key in extra_vars:
+                if key in survey_vars:
+                    prompted_fields[field][key] = extra_vars[key]
+                else:
+                    ignored_fields[field][key] = extra_vars[key]
+        else:
+            prompted_fields['extra_vars'] = extra_vars
+
+        return prompted_fields, ignored_fields
+
     def get_warnings(self):
         warning_data = {}
         for node in self.workflow_job_template_nodes.all():
@@ -372,7 +395,7 @@ class WorkflowJobInheritNodesMixin(object):
                 self._inherit_relationship(old_node, new_node, node_ids_map, node_type)
                 
 
-class WorkflowJob(UnifiedJob, WorkflowJobOptions, JobNotificationMixin, WorkflowJobInheritNodesMixin):
+class WorkflowJob(UnifiedJob, SurveyJob, WorkflowJobOptions, JobNotificationMixin, WorkflowJobInheritNodesMixin):
 
     class Meta:
         app_label = 'main'
@@ -387,8 +410,6 @@ class WorkflowJob(UnifiedJob, WorkflowJobOptions, JobNotificationMixin, Workflow
         on_delete=models.SET_NULL,
     )
 
-    extra_vars_dict = VarsDictProperty('extra_vars', True)
-
     @classmethod
     def _get_parent_field_name(cls):
         return 'workflow_job_template'
diff --git a/awx/main/utils.py b/awx/main/utils.py
index 58123972b7..c513a9bf5d 100644
--- a/awx/main/utils.py
+++ b/awx/main/utils.py
@@ -478,6 +478,22 @@ def cache_list_capabilities(page, prefetch_list, model, user):
             if obj.pk in ids_with_role:
                 obj.capabilities_cache[display_method] = True
 
+def parse_yaml_or_json(vars_str):
+    '''
+    Attempt to parse a string with variables, and if attempt fails,
+    return an empty dictionary.
+    '''
+    if isinstance(vars_str, dict):
+        return vars_str
+    try:
+        vars_dict = json.loads(vars_str)
+    except (ValueError, TypeError):
+        try:
+            vars_dict = yaml.safe_load(vars_str)
+            assert isinstance(extra_vars, dict)
+        except (yaml.YAMLError, TypeError, AttributeError, AssertionError):
+            vars_dict = {}
+    return vars_dict
 
 @memoize()
 def get_system_task_capacity():