diff --git a/awx/api/serializers.py b/awx/api/serializers.py
index 08073a7bc0..c8a7b98cfc 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
@@ -2599,6 +2607,64 @@ class JobLaunchSerializer(BaseSerializer):
         obj.credential = JT_credential
         return attrs
 
+class WorkflowJobLaunchSerializer(BaseSerializer):
+
+    can_start_without_user_input = serializers.BooleanField(read_only=True)
+    variables_needed_to_start = serializers.ReadOnlyField()
+    survey_enabled = serializers.SerializerMethodField()
+    extra_vars = VerbatimField(required=False, write_only=True)
+    warnings = serializers.SerializerMethodField()
+    workflow_job_template_data = serializers.SerializerMethodField()
+
+    class Meta:
+        model = WorkflowJobTemplate
+        fields = ('can_start_without_user_input', 'extra_vars', 'warnings',
+                  'survey_enabled', 'variables_needed_to_start',
+                  'workflow_job_template_data')
+
+    def get_warnings(self, obj):
+        return obj.get_warnings()
+
+    def get_survey_enabled(self, obj):
+        if obj:
+            return obj.survey_enabled and 'spec' in obj.survey_spec
+        return False
+
+    def get_workflow_job_template_data(self, obj):
+        return dict(name=obj.name, id=obj.id, description=obj.description)
+
+    def validate(self, attrs):
+        errors = {}
+        obj = self.instance
+
+        extra_vars = attrs.get('extra_vars', {})
+
+        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):
+                    errors['extra_vars'] = 'Must be a valid JSON or YAML dictionary.'
+
+        if not isinstance(extra_vars, dict):
+            extra_vars = {}
+
+        if self.get_survey_enabled(obj):
+            validation_errors = obj.survey_variable_validation(extra_vars)
+            if validation_errors:
+                errors['variables_needed_to_start'] = validation_errors
+
+        if errors:
+            raise serializers.ValidationError(errors)
+
+        WFJT_extra_vars = obj.extra_vars
+        attrs = super(WorkflowJobLaunchSerializer, self).validate(attrs)
+        obj.extra_vars = WFJT_extra_vars
+        return attrs
+
 class NotificationTemplateSerializer(BaseSerializer):
     show_capabilities = ['edit', 'delete']
 
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..ba76e777b3 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
@@ -2781,27 +2786,42 @@ class WorkflowJobTemplateLabelList(JobTemplateLabelList):
     new_in_310 = True
 
 
-# TODO:
-class WorkflowJobTemplateLaunch(GenericAPIView):
+class WorkflowJobTemplateLaunch(RetrieveAPIView):
 
     model = WorkflowJobTemplate
-    serializer_class = EmptySerializer
+    serializer_class = WorkflowJobLaunchSerializer
     new_in_310 = True
+    is_job_start = True
+    always_allow_superuser = False
 
-    def get(self, request, *args, **kwargs):
-        data = {}
+    def update_raw_data(self, data):
         obj = self.get_object()
-        data['warnings'] = obj.get_warnings()
-        return Response(data)
+        extra_vars = data.pop('extra_vars', None) or {}
+        if obj:
+            for v in obj.variables_needed_to_start:
+                extra_vars.setdefault(v, u'')
+            if extra_vars:
+                data['extra_vars'] = extra_vars
+        return data
 
     def post(self, request, *args, **kwargs):
         obj = self.get_object()
         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)
-        data = dict(workflow_job=new_job.id)
+        serializer = self.serializer_class(instance=obj, data=request.data)
+        if not serializer.is_valid():
+            return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
+
+        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 = OrderedDict()
+        data['ignored_fields'] = ignored_fields
+        data.update(WorkflowJobSerializer(new_job, context=self.get_serializer_context()).to_representation(new_job))
+        data['workflow_job'] = new_job.id
         return Response(data, status=status.HTTP_201_CREATED)
 
 # TODO:
diff --git a/awx/main/access.py b/awx/main/access.py
index a33503920f..c3dc82a098 100644
--- a/awx/main/access.py
+++ b/awx/main/access.py
@@ -1520,8 +1520,8 @@ class WorkflowJobTemplateAccess(BaseAccess):
             return True
 
         # will check this if surveys are added to WFJT
-        # if 'survey_enabled' in data and data['survey_enabled']:
-        #     self.check_license(feature='surveys')
+        if 'survey_enabled' in data and data['survey_enabled']:
+            self.check_license(feature='surveys')
 
         return self.check_related('organization', Organization, data)
 
@@ -1530,8 +1530,8 @@ class WorkflowJobTemplateAccess(BaseAccess):
             # check basic license, node count
             self.check_license()
             # if surveys are added to WFJTs, check license here
-            # if obj.survey_enabled:
-            #     self.check_license(feature='surveys')
+            if obj.survey_enabled:
+                self.check_license(feature='surveys')
 
         # Super users can start any job
         if self.user.is_superuser:
@@ -1540,9 +1540,10 @@ class WorkflowJobTemplateAccess(BaseAccess):
         return self.user in obj.execute_role
 
     def can_change(self, obj, data):
-        # # Check survey license if surveys are added to WFJTs
-        # if 'survey_enabled' in data and obj.survey_enabled != data['survey_enabled'] and data['survey_enabled']:
-        #     self.check_license(feature='surveys')
+        # Check survey license if surveys are added to WFJTs
+        if (data and 'survey_enabled' in data and
+                obj.survey_enabled != data['survey_enabled'] and data['survey_enabled']):
+            self.check_license(feature='surveys')
 
         if self.user.is_superuser:
             return True
diff --git a/awx/main/migrations/0049_v310_workflow_surveys.py b/awx/main/migrations/0049_v310_workflow_surveys.py
new file mode 100644
index 0000000000..bc3865d33c
--- /dev/null
+++ b/awx/main/migrations/0049_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', '0048_v310_instance_capacity'),
+    ]
+
+    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..7621649889 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,10 +32,13 @@ from awx.main.models.notifications import (
     NotificationTemplate,
     JobNotificationMixin,
 )
-from awx.main.utils import ignore_inventory_computed_fields
+from awx.main.utils import (
+    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
+from awx.main.models.mixins import ResourceMixin, SurveyJobTemplateMixin, SurveyJobMixin
 from awx.main.models.base import PERM_INVENTORY_SCAN
 
 from awx.main.consumers import emit_channel_notification
@@ -188,7 +190,7 @@ class JobOptions(BaseModel):
         else:
             return []
 
-class JobTemplate(UnifiedJobTemplate, JobOptions, ResourceMixin):
+class JobTemplate(UnifiedJobTemplate, JobOptions, SurveyJobTemplateMixin, 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 +234,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 +311,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 +340,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 +394,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, JobOptions, SurveyJobMixin, 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 +419,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 +592,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..efe36bcefc 100644
--- a/awx/main/models/mixins.py
+++ b/awx/main/models/mixins.py
@@ -1,15 +1,20 @@
+# Python
+import json
+
 # Django
 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 (
     Role, RoleAncestorEntry, get_roles_on_resource
 )
+from awx.main.utils import parse_yaml_or_json
 
 
-__all__ = ['ResourceMixin']
+__all__ = ['ResourceMixin', 'SurveyJobTemplateMixin', 'SurveyJobMixin']
 
 class ResourceMixin(models.Model):
 
@@ -60,3 +65,155 @@ class ResourceMixin(models.Model):
 
         return get_roles_on_resource(self, accessor)
 
+
+class SurveyJobTemplateMixin(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 SurveyJobMixin(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/unified_jobs.py b/awx/main/models/unified_jobs.py
index e9925a6b17..2c46157007 100644
--- a/awx/main/models/unified_jobs.py
+++ b/awx/main/models/unified_jobs.py
@@ -292,12 +292,6 @@ class UnifiedJobTemplate(PolymorphicModel, CommonModelNameNotUnique, Notificatio
         '''
         raise NotImplementedError # Implement in subclass.
 
-    def _update_unified_job_kwargs(self, **kwargs):
-        '''
-        Hook for subclasses to update kwargs.
-        '''
-        return kwargs   # Override if needed in subclass.
-
     @property
     def notification_templates(self):
         '''
@@ -346,7 +340,10 @@ class UnifiedJobTemplate(PolymorphicModel, CommonModelNameNotUnique, Notificatio
                     m2m_fields[field_name] = getattr(self, field_name)
                 else:
                     create_kwargs[field_name] = getattr(self, field_name)
-        new_kwargs = self._update_unified_job_kwargs(**create_kwargs)
+        if hasattr(self, '_update_unified_job_kwargs'):
+            new_kwargs = self._update_unified_job_kwargs(**create_kwargs)
+        else:
+            new_kwargs = create_kwargs
         unified_job = unified_job_class(**new_kwargs)
         # For JobTemplate-based jobs with surveys, add passwords to list for perma-redaction
         if hasattr(self, 'survey_spec') and getattr(self, 'survey_enabled', False):
diff --git a/awx/main/models/workflow.py b/awx/main/models/workflow.py
index ff457fd207..0a9c081680 100644
--- a/awx/main/models/workflow.py
+++ b/awx/main/models/workflow.py
@@ -20,8 +20,9 @@ from awx.main.models.rbac import (
     ROLE_SINGLETON_SYSTEM_AUDITOR
 )
 from awx.main.fields import ImplicitRoleField
-from awx.main.models.mixins import ResourceMixin
+from awx.main.models.mixins import ResourceMixin, SurveyJobTemplateMixin, SurveyJobMixin
 from awx.main.redact import REPLACE_STR
+from awx.main.utils import parse_yaml_or_json
 
 from copy import copy
 
@@ -231,12 +232,15 @@ class WorkflowJobNode(WorkflowNodeBase):
         if aa_dict:
             self.ancestor_artifacts = aa_dict
             self.save(update_fields=['ancestor_artifacts'])
+        password_dict = {}
         if '_ansible_no_log' in aa_dict:
-            # TODO: merge Workflow Job survey passwords into this
-            password_dict = {}
             for key in aa_dict:
                 if key != '_ansible_no_log':
                     password_dict[key] = REPLACE_STR
+        workflow_job_survey_passwords = self.workflow_job.survey_passwords
+        if workflow_job_survey_passwords:
+            password_dict.update(workflow_job_survey_passwords)
+        if password_dict:
             data['survey_passwords'] = password_dict
         # process extra_vars
         # TODO: still lack consensus about variable precedence
@@ -260,7 +264,9 @@ class WorkflowJobOptions(BaseModel):
         default='',
     )
 
-class WorkflowJobTemplate(UnifiedJobTemplate, WorkflowJobOptions, ResourceMixin):
+    extra_vars_dict = VarsDictProperty('extra_vars', True)
+
+class WorkflowJobTemplate(UnifiedJobTemplate, WorkflowJobOptions, SurveyJobTemplateMixin, ResourceMixin):
 
     class Meta:
         app_label = 'main'
@@ -290,7 +296,7 @@ class WorkflowJobTemplate(UnifiedJobTemplate, WorkflowJobOptions, ResourceMixin)
 
     @classmethod
     def _get_unified_job_field_names(cls):
-        return ['name', 'description', 'extra_vars', 'labels', 'schedule', 'launch_type']
+        return ['name', 'description', 'extra_vars', 'labels', 'survey_passwords', 'schedule', 'launch_type']
 
     def get_absolute_url(self):
         return reverse('api:workflow_job_template_detail', args=(self.pk,))
@@ -318,6 +324,29 @@ 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['extra_vars'][key] = extra_vars[key]
+                else:
+                    ignored_fields['extra_vars'][key] = extra_vars[key]
+        else:
+            prompted_fields['extra_vars'] = extra_vars
+
+        return prompted_fields, ignored_fields
+
+    def can_start_without_user_input(self):
+        '''Return whether WFJT can be launched without survey passwords.'''
+        return not bool(self.variables_needed_to_start)
+
     def get_warnings(self):
         warning_data = {}
         for node in self.workflow_job_template_nodes.all():
@@ -372,7 +401,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, WorkflowJobOptions, SurveyJobMixin, JobNotificationMixin, WorkflowJobInheritNodesMixin):
 
     class Meta:
         app_label = 'main'
@@ -387,8 +416,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/tests/unit/models/test_job_unit.py b/awx/main/tests/unit/models/test_survey_models.py
similarity index 62%
rename from awx/main/tests/unit/models/test_job_unit.py
rename to awx/main/tests/unit/models/test_survey_models.py
index ff55d7103b..4f9653755e 100644
--- a/awx/main/tests/unit/models/test_job_unit.py
+++ b/awx/main/tests/unit/models/test_survey_models.py
@@ -2,7 +2,10 @@ import pytest
 import json
 
 from awx.main.tasks import RunJob
-from awx.main.models import Job
+from awx.main.models import (
+    Job,
+    WorkflowJobTemplate
+)
 
 
 @pytest.fixture
@@ -65,3 +68,35 @@ def test_job_args_unredacted_passwords(job):
     ev_index = args.index('-e') + 1
     extra_vars = json.loads(args[ev_index])
     assert extra_vars['secret_key'] == 'my_password'
+
+class TestWorkflowSurveys:
+    def test_update_kwargs_survey_defaults(self, survey_spec_factory):
+        "Assure that the survey default over-rides a JT variable"
+        spec = survey_spec_factory('var1')
+        spec['spec'][0]['default'] = 3
+        spec['spec'][0]['required'] = False
+        wfjt = WorkflowJobTemplate(
+            name="test-wfjt",
+            survey_spec=spec,
+            survey_enabled=True,
+            extra_vars="var1: 5"
+        )
+        updated_extra_vars = wfjt._update_unified_job_kwargs()
+        assert 'extra_vars' in updated_extra_vars
+        assert json.loads(updated_extra_vars['extra_vars'])['var1'] == 3
+        assert wfjt.can_start_without_user_input()
+
+    def test_variables_needed_to_start(self, survey_spec_factory):
+        "Assure that variables_needed_to_start output contains mandatory vars"
+        spec = survey_spec_factory(['question1', 'question2', 'question3'])
+        spec['spec'][0]['required'] = False
+        spec['spec'][1]['required'] = True
+        spec['spec'][2]['required'] = False
+        wfjt = WorkflowJobTemplate(
+            name="test-wfjt",
+            survey_spec=spec,
+            survey_enabled=True,
+            extra_vars="question2: hiworld"
+        )
+        assert wfjt.variables_needed_to_start == ['question2']
+        assert not wfjt.can_start_without_user_input()
diff --git a/awx/main/utils.py b/awx/main/utils.py
index 58123972b7..3614a99039 100644
--- a/awx/main/utils.py
+++ b/awx/main/utils.py
@@ -5,6 +5,7 @@
 import base64
 import hashlib
 import json
+import yaml
 import logging
 import os
 import re
@@ -478,6 +479,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(vars_dict, dict)
+        except (yaml.YAMLError, TypeError, AttributeError, AssertionError):
+            vars_dict = {}
+    return vars_dict
 
 @memoize()
 def get_system_task_capacity():