diff --git a/awx_collection/plugins/modules/tower_workflow_launch.py b/awx_collection/plugins/modules/tower_workflow_launch.py
index 31535fd52c..0e7d8da75f 100644
--- a/awx_collection/plugins/modules/tower_workflow_launch.py
+++ b/awx_collection/plugins/modules/tower_workflow_launch.py
@@ -36,7 +36,7 @@ options:
       type: str
     inventory:
       description:
-        - Inventory to use for the job, only used if prompt for inventory is set.
+        - Inventory to use for the job ran with this workflow, only used if prompt for inventory is set.
       type: str
     limit:
       description:
@@ -44,14 +44,14 @@ options:
       type: str
     scm_branch:
       description:
-        - A specific of the SCM project to run the template on.
+        - A specific branch of the SCM project to run the template on.
         - This is only applicable if your project allows for branch override.
       type: str
       version_added: "3.7"
     extra_vars:
       description:
         - Any extra vars required to launch the job.
-      type: str
+      type: dict
     wait:
       description:
         - Wait for the workflow to complete.
@@ -86,24 +86,22 @@ job_info:
 
 
 EXAMPLES = '''
-- name: Launch a workflow
+- name: Launch a workflow with a timeout of 10 seconds
   tower_workflow_launch:
-    name: "Test Workflow"
-  delegate_to: localhost
-  run_once: true
-  register: workflow_results
+    workflow_template: "Test Workflow"
+    timeout: 10
 
-- name: Launch a Workflow with parameters without waiting
+- name: Launch a Workflow with extra_vars without waiting
   tower_workflow_launch:
-    name: "Test workflow"
-    extra_vars: "---\nmy: var"
+    workflow_template: "Test workflow"
+    extra_vars:
+      var1: My First Variable
+      var2: My Second Variable
     wait: False
-  delegate_to: localhost
-  run_once: true
-  register: workflow_task_info
 '''
 
 from ..module_utils.tower_api import TowerModule
+import json
 import time
 
 
@@ -115,7 +113,7 @@ def main():
         inventory=dict(),
         limit=dict(),
         scm_branch=dict(),
-        extra_vars=dict(),
+        extra_vars=dict(type='dict'),
         wait=dict(required=False, default=True, type='bool'),
         interval=dict(required=False, default=1.0, type='float'),
         timeout=dict(required=False, default=None, type='int'),
@@ -135,6 +133,11 @@ def main():
     interval = module.params.get('interval')
     timeout = module.params.get('timeout')
 
+    # Special treatment of extra_vars parameter
+    extra_vars = module.params.get('extra_vars')
+    if extra_vars is not None:
+        optional_args['extra_vars'] = json.dumps(extra_vars)
+
     # Create a datastructure to pass into our job launch
     post_data = {}
     for key in optional_args.keys():
@@ -149,7 +152,7 @@ def main():
     lookup_data = {'name': name}
     if organization:
         lookup_data['organization'] = module.resolve_name_to_id('organizations', organization)
-    workflow_job_template = module.get_one('workflow_job_templates', **{'data': lookup_data})
+    workflow_job_template = module.get_one('workflow_job_templates', data=lookup_data)
 
     if workflow_job_template is None:
         module.fail_json(msg="Unable to find workflow job template")
@@ -168,13 +171,13 @@ def main():
         if variable_name in post_data and not workflow_job_template[check_vars_to_prompts[variable_name]]:
             param_errors.append("The field {0} was specified but the workflow job template does not allow for it to be overridden".format(variable_name))
     if len(param_errors) > 0:
-        module.fail_json(msg="Parameters specified which can not be passed into wotkflow job template, see errors for details", **{'errors': param_errors})
+        module.fail_json(msg="Parameters specified which can not be passed into wotkflow job template, see errors for details", errors=param_errors)
 
     # Launch the job
-    result = module.post_endpoint(workflow_job_template['related']['launch'], **{'data': post_data})
+    result = module.post_endpoint(workflow_job_template['related']['launch'], data=post_data)
 
     if result['status_code'] != 201:
-        module.fail_json(msg="Failed to launch workflow, see response for details", **{'response': result})
+        module.fail_json(msg="Failed to launch workflow, see response for details", response=result)
 
     module.json_output['changed'] = True
     module.json_output['id'] = result['json']['id']
@@ -201,7 +204,7 @@ def main():
         result = module.get_endpoint(job_url)
         module.json_output['status'] = result['json']['status']
 
-    # If the job has failed, we want to raise an Exception for that so we get a non-zero response.
+    # If the job has failed, we want to raise a task failure for that so we get a non-zero response.
     if result['json']['failed']:
         module.json_output['msg'] = 'The workflow "{0}" failed'.format(name)
         module.fail_json(**module.json_output)
diff --git a/awx_collection/tests/integration/targets/tower_workflow_launch/tasks/main.yml b/awx_collection/tests/integration/targets/tower_workflow_launch/tasks/main.yml
index 3d3c41feb2..9f9c76e2a8 100644
--- a/awx_collection/tests/integration/targets/tower_workflow_launch/tasks/main.yml
+++ b/awx_collection/tests/integration/targets/tower_workflow_launch/tasks/main.yml
@@ -70,6 +70,26 @@
       - result is not failed
       - "'id' in result['job_info']"
 
+- name: Prompt the workflow's extra_vars on launch
+  tower_workflow_job_template:
+    name: "{{ wfjt_name1 }}"
+    state: present
+    ask_variables_on_launch: true
+
+- name: Kick off a workflow with extra_vars
+  tower_workflow_launch:
+    validate_certs: false
+    workflow_template: "{{ wfjt_name1 }}"
+    extra_vars:
+      var1: My First Variable
+      var2: My Second Variable
+  ignore_errors: true
+  register: result
+
+- assert:
+    that:
+      - result is not failed
+
 - name: Clean up test workflow
   tower_workflow_job_template:
     name: "{{ wfjt_name1 }}"