diff --git a/awx/api/serializers.py b/awx/api/serializers.py
index a19c9bd406..db4888b31f 100644
--- a/awx/api/serializers.py
+++ b/awx/api/serializers.py
@@ -1955,7 +1955,7 @@ class BulkHostSerializer(HostSerializer):
     instance_id = serializers.CharField(required=False, max_length=1024)
     description = serializers.CharField(required=False)
     enabled = serializers.BooleanField(default=True, required=False)
-    variables = serializers.CharField(allow_blank=True, required=False)
+    variables = serializers.CharField(write_only=True, required=False)
 
     class Meta:
         fields = (
@@ -4565,6 +4565,7 @@ class BulkJobNodeSerializer(serializers.Serializer):
     survey_passwords = serializers.CharField(required=False, write_only=True, allow_blank=False)
     job_slice_count = serializers.IntegerField(required=False, min_value=1)
     timeout = serializers.IntegerField(required=False, min_value=1)
+    extra_data = serializers.JSONField(write_only=True, required=False)
 
     class Meta:
         fields = (
@@ -4689,11 +4690,16 @@ class BulkJobLaunchSerializer(BaseSerializer):
         job_node_data = validated_data.pop('jobs')
         # FIXME: Need to set organization on the WorkflowJob in order for users to be able to see it --
         # normally their permission is sourced from the underlying WorkflowJobTemplate
-        # maybe we need to add Organization to WorkflowJobd
-        wfj_limit = validated_data.pop('limit', None)
+        # maybe we need to add Organization to WorkflowJob
+        wfj_deferred_attr_names = ('skip_tags', 'limit', 'job_tags')
+        wfj_deferred_vals = {}
+        for item in wfj_deferred_attr_names:
+            wfj_deferred_vals[item] = validated_data.pop(item, None)
+
         wfj = WorkflowJob.objects.create(**validated_data, is_bulk_job=True)
-        if wfj_limit:
-            wfj.limit = wfj_limit
+        for key, val in wfj_deferred_vals.items():
+            if val:
+                setattr(wfj, key, val)
         nodes = []
         node_m2m_objects = {}
         node_m2m_object_types_to_through_model = {
@@ -4717,7 +4723,6 @@ class BulkJobLaunchSerializer(BaseSerializer):
         )
         node_deferred_attrs = {}
         for node_attrs in job_node_data:
-
             # we need to add any m2m objects after creation via the through model
             node_m2m_objects[node_attrs['identifier']] = {}
             node_deferred_attrs[node_attrs['identifier']] = {}
diff --git a/awx/main/migrations/0175_workflowjob_is_bulk_job.py b/awx/main/migrations/0175_workflowjob_is_bulk_job.py
index 8e6c835e8a..c1e71afc73 100644
--- a/awx/main/migrations/0175_workflowjob_is_bulk_job.py
+++ b/awx/main/migrations/0175_workflowjob_is_bulk_job.py
@@ -4,7 +4,6 @@ from django.db import migrations, models
 
 
 class Migration(migrations.Migration):
-
     dependencies = [
         ('main', '0174_ensure_org_ee_admin_roles'),
     ]
diff --git a/awx_collection/meta/runtime.yml b/awx_collection/meta/runtime.yml
index e59becd4ea..3e50a52e69 100644
--- a/awx_collection/meta/runtime.yml
+++ b/awx_collection/meta/runtime.yml
@@ -6,6 +6,8 @@ action_groups:
     - ad_hoc_command_cancel
     - ad_hoc_command_wait
     - application
+    - bulk_job_launch
+    - bulk_host_create
     - controller_meta
     - credential_input_source
     - credential
diff --git a/awx_collection/plugins/modules/bulk_host_create.py b/awx_collection/plugins/modules/bulk_host_create.py
index f87045b487..3e6ea91cc5 100644
--- a/awx_collection/plugins/modules/bulk_host_create.py
+++ b/awx_collection/plugins/modules/bulk_host_create.py
@@ -16,22 +16,43 @@ author: "Seth Foster (@fosterseth)"
 short_description: Bulk host create in Automation Platform Controller
 description:
     - Single-request bulk host creation in Automation Platform Controller.
-    - Designed to efficiently add many hosts to an inventory.
+    - Provides a way to add many hosts at once to an inventory in Controller.
 options:
     hosts:
       description:
         - List of hosts to add to inventory.
       required: True
-      type: str
+      type: list
+      elements: dict
+      suboptions:
+        name:
+          description:
+            - The name to use for the host.
+          type: str
+          require: True
+        description:
+          - The description to use for the host.
+        type: str
+        enabled:
+          description:
+            - If the host should be enabled.
+          type: bool
+        variables:
+          description:
+            - Variables to use for the host.
+          type: dict
+        instance_id:
+          description:
+            - instance_id to use for the host.
+          type: str
     inventory:
       description:
-        - Inventory the hosts should be made a member of.
+        - Inventory ID the hosts should be made a member of.
       required: True
-      type: str
+      type: int
 extends_documentation_fragment: awx.awx.auth
 '''
 
-
 EXAMPLES = '''
 - name: Bulk host create
   bulk_host_create:
@@ -44,11 +65,12 @@ EXAMPLES = '''
 from ..module_utils.controller_api import ControllerAPIModule
 import json
 
+
 def main():
     # Any additional arguments that are not fields of the item can be added here
     argument_spec = dict(
         hosts=dict(required=True, type='list'),
-        inventory=dict(),
+        inventory=dict(required=True, type='int'),
     )
 
     # Create a module for ourselves
@@ -58,6 +80,9 @@ def main():
     inventory = module.params.get('inventory')
     hosts = module.params.get('hosts')
 
+    for h in hosts:
+      if 'variables' in h:
+        h['variables'] = json.dumps(h['variables'])
     # Launch the jobs
     result = module.post_endpoint("bulk/host_create", data={"inventory": inventory, "hosts": hosts})
 
@@ -70,4 +95,4 @@ def main():
 
 
 if __name__ == '__main__':
-    main()
\ No newline at end of file
+    main()
diff --git a/awx_collection/plugins/modules/bulk_job_launch.py b/awx_collection/plugins/modules/bulk_job_launch.py
index e6e8c7b103..7a6e800823 100644
--- a/awx_collection/plugins/modules/bulk_job_launch.py
+++ b/awx_collection/plugins/modules/bulk_job_launch.py
@@ -16,44 +16,134 @@ author: "Seth Foster (@fosterseth)"
 short_description: Bulk job launch in Automation Platform Controller
 description:
     - Single-request bulk job launch in Automation Platform Controller.
-    - The result is flat workflow, each job specified in the parameter jobs results in a workflow job node.
+    - Creates a workflow where each node corresponds to an item specified in the jobs option.
     - Any options specified at the top level will inherited by the launched jobs (if prompt on launch is enabled for those fields).
-    - Designed to efficiently start many jobs at once.
+    - Provides a way to submit many jobs at once to Controller.
 options:
     jobs:
       description:
         - List of jobs to create.
-        - Any promptable field on unified_job_template can be provided as a field on the list item (e.g. limit).
       required: True
       type: list
+      elements: dict
+      suboptions:
+        unified_job_template:
+          description:
+            - Job template ID to use when launching.
+          type: int
+          required: True
+        inventory:
+          description:
+            - Inventory ID applied as a prompt, if job template prompts for inventory
+          type: int
+        execution_environment:
+          description:
+            - Execution environment ID applied as a prompt, if job template prompts for execution environments
+          type: int
+        instance_groups:
+          description:
+            - Instance group IDs applied as a prompt, if job template prompts for instance groups
+          type: list
+          elements: int
+        credentials:
+          description:
+            - Credential IDs applied as a prompt, if job template prompts for credentials
+          type: list
+          elements: int
+        labels:
+          description:
+            - Label IDs to use for the job, if job template prompts for labels
+          type: list
+          elements: int
+        extra_data:
+          description:
+            - Extra variables to apply at launch time, if job template prompts for extra variables
+          type: dict
+          default: {}
+        diff_mode:
+          description:
+            - Show the changes made by Ansible tasks where supported
+          type: bool
+        verbosity:
+          description:
+            - Verbosity level for this ad hoc command run
+          type: int
+          choices: [ 0, 1, 2, 3, 4, 5 ]
+        scm_branch:
+          description:
+            - SCM branch applied as a prompt, if job template prompts for SCM branch
+            - This is only applicable if the project allows for branch override
+          type: str
+        job_type:
+          description:
+            - Job type applied as a prompt, if job template prompts for job type
+          type: str
+          choices:
+            - 'run'
+            - 'check'
+        job_tags:
+          description:
+            - Job tags applied as a prompt, if job template prompts for job tags
+          type: str
+        skip_tags:
+          description:
+            - Tags to skip, applied as a prompt, if job template prompts for job tags
+          type: str
+        limit:
+          description:
+            - Limit to act on, applied as a prompt, if job template prompts for limit
+          type: str
+        forks:
+          description:
+            - The number of parallel or simultaneous processes to use while executing the playbook, if job template prompts for forks
+          type: int
+        job_slice_count:
+          description:
+            - The number of jobs to slice into at runtime, if job template prompts for job slices.
+            - Will cause the Job Template to launch a workflow if value is greater than 1.
+          type: int
+          default: '1'
+        identifier:
+          description:
+            - Identifier for the resulting workflow node that represents this job
+          type: str
+        timeout:
+          description:
+            - Maximum time in seconds to wait for a job to finish (server-side), if job template prompts for timeout.
+          type: int
     name:
       description:
         - The name of the bulk job that is created
       required: False
       type: str
+    description:
+      description:
+        - Optional description of this bulk job.
+      type: str
     organization:
       description:
         - If not provided, will use the organization the user is in.
         - Required if the user belongs to more than one organization.
         - Affects who can see the resulting bulk job.
-      type: str
+      type: int
     inventory:
       description:
-        - Inventory to use for the jobs ran within the bulk job, only used if prompt for inventory is set.
-      type: str
-    limit:
-      description:
-        - Limit to use for the I(job_template).
-      type: str
+        - Inventory ID to use for the jobs ran within the bulk job, only used if prompt for inventory is set.
+      type: int
     scm_branch:
       description:
         - A specific branch of the SCM project to run the template on.
-        - This is only applicable if your project allows for branch override.
+        - This is only applicable if the project allows for branch override.
       type: str
     extra_vars:
       description:
         - Any extra vars required to launch the job.
+        - Extends the extra_data field at the individual job level.
       type: dict
+    limit:
+      description:
+        - Limit to use for the bulk job.
+      type: str
     job_tags:
       description:
         - A comma-separated list of playbook tags to specify what parts of the playbooks should be executed.
@@ -64,7 +154,7 @@ options:
       type: str
     wait:
       description:
-        - Wait for the workflow to complete.
+        - Wait for the bulk job to complete.
       default: True
       type: bool
     interval:
@@ -73,32 +163,33 @@ options:
       required: False
       default: 2
       type: float
-    timeout:
-      description:
-        - If waiting for the workflow to complete this will abort after this
-          amount of seconds
-      type: int
 extends_documentation_fragment: awx.awx.auth
 '''
 
 RETURN = '''
 job_info:
-    description: dictionary containing information about the workflow executed
-    returned: If workflow launched
+    description: dictionary containing information about the bulk job executed
+    returned: If bulk job launched
     type: dict
 '''
 
-
 EXAMPLES = '''
 - name: Launch bulk jobs
   bulk_job_launch:
     name: My Bulk Job Launch
     jobs:
       - unified_job_template: 7
+        skip_tags: foo
       - unified_job_template: 10
         limit: foo
+        extra_data:
+          food: carrot
+          color: orange
     limit: bar
-    inventory: 1 # only affects job templates with prompt on launch enabled for inventory
+    extra_vars: # these override / extend extra_data at the job level
+      food: grape
+      animal: owl
+    inventory: 1
 
 - name: Launch bulk jobs with lookup plugin
   bulk_job_launch:
@@ -111,13 +202,14 @@ EXAMPLES = '''
 from ..module_utils.controller_api import ControllerAPIModule
 import json
 
+
 def main():
     # Any additional arguments that are not fields of the item can be added here
     argument_spec = dict(
         jobs=dict(required=True, type='list'),
         name=dict(),
-        organization=dict(),
-        inventory=dict(),
+        organization=dict(type='int'),
+        inventory=dict(type='int'),
         limit=dict(),
         scm_branch=dict(),
         extra_vars=dict(type='dict'),
@@ -131,15 +223,31 @@ def main():
     # Create a module for ourselves
     module = ControllerAPIModule(argument_spec=argument_spec)
 
+    post_data_names = (
+        'jobs',
+        'name',
+        'organization',
+        'inventory',
+        'limit',
+        'scm_branch',
+        'extra_vars',
+        'job_tags',
+        'skip_tags',
+    )
+    post_data = {}
+    for p in post_data_names:
+        val = module.params.get(p)
+        if val:
+            post_data[p] = val
+
     # Extract our parameters
-    name = module.params.get('name')
     wait = module.params.get('wait')
     timeout = module.params.get('timeout')
     interval = module.params.get('interval')
-    jobs = module.params.get('jobs')
+    name = module.params.get('name')
 
     # Launch the jobs
-    result = module.post_endpoint("bulk/job_launch", data={"jobs": jobs})
+    result = module.post_endpoint("bulk/job_launch", data=post_data)
 
     if result['status_code'] != 201:
         module.fail_json(msg="Failed to launch bulk jobs, see response for details", response=result)
@@ -148,7 +256,7 @@ def main():
     module.json_output['id'] = result['json']['id']
     module.json_output['status'] = result['json']['status']
     # This is for backwards compatability
-    module.json_output['job_info'] = {'id': result['json']['id']}
+    module.json_output['job_info'] = result['json']
 
     if not wait:
         module.exit_json(**module.json_output)
@@ -160,4 +268,4 @@ def main():
 
 
 if __name__ == '__main__':
-    main()
\ No newline at end of file
+    main()
diff --git a/awx_collection/test/awx/test_bulk.py b/awx_collection/test/awx/test_bulk.py
new file mode 100644
index 0000000000..fbccf3d1d8
--- /dev/null
+++ b/awx_collection/test/awx/test_bulk.py
@@ -0,0 +1,42 @@
+from __future__ import absolute_import, division, print_function
+
+__metaclass__ = type
+
+import pytest
+
+from awx.main.models import WorkflowJob
+
+@pytest.mark.django_db
+def test_bulk_job_launch(run_module, admin_user, job_template):
+    jobs = [dict(unified_job_template=job_template.id)]
+    result = run_module(
+        'bulk_job_launch',
+        {
+            'name': "foo-bulk-job",
+            'jobs': jobs,
+            'extra_vars': {'animal': 'owl'},
+            'limit': 'foo',
+            'wait': False,
+        },
+        admin_user,
+    )
+
+    bulk_job = WorkflowJob.objects.get(name="foo-bulk-job")
+    assert bulk_job.extra_vars == '{"animal": "owl"}'
+    assert bulk_job.limit == "foo"
+
+
+@pytest.mark.django_db
+def test_bulk_host_create(run_module, admin_user, inventory):
+    hosts = [dict(name="127.0.0.1"), dict(name="foo.dns.org")]
+    result = run_module(
+        'bulk_host_create',
+        {
+            'inventory': inventory.id,
+            'hosts': hosts,
+        },
+        admin_user,
+    )
+    resp_hosts = inventory.hosts.all().values_list('name', flat=True)
+    for h in hosts:
+        assert h['name'] in resp_hosts
\ No newline at end of file
diff --git a/awx_collection/test/awx/test_completeness.py b/awx_collection/test/awx/test_completeness.py
index 451c1a61d3..ef3d70727a 100644
--- a/awx_collection/test/awx/test_completeness.py
+++ b/awx_collection/test/awx/test_completeness.py
@@ -44,6 +44,12 @@ no_endpoint_for_module = [
     'subscriptions',  # Subscription deals with config/subscriptions
 ]
 
+# Add modules with endpoints that are not at /api/v2
+extra_endpoints = {
+    'bulk_job_launch': '/api/v2/bulk/job_launch/',
+    'bulk_host_create': '/api/v2/bulk/host_create/',
+}
+
 # Global module parameters we can ignore
 ignore_parameters = ['state', 'new_name', 'update_secrets', 'copy_from']
 
@@ -73,6 +79,8 @@ no_api_parameter_ok = {
     'user': ['new_username', 'organization'],
     # workflow_approval parameters that do not apply when approving an approval node.
     'workflow_approval': ['action', 'interval', 'timeout', 'workflow_job_id'],
+    # bulk
+    'bulk_job_launch': ['interval', 'wait'],
 }
 
 # When this tool was created we were not feature complete. Adding something in here indicates a module
@@ -228,6 +236,10 @@ def test_completeness(collection_import, request, admin_user, job_template, exec
         user=admin_user,
         expect=None,
     )
+
+    for key, val in extra_endpoints.items():
+        endpoint_response.data[key] = val
+
     for endpoint in endpoint_response.data.keys():
         # Module names are singular and endpoints are plural so we need to convert to singular
         singular_endpoint = '{0}'.format(endpoint)
diff --git a/awx_collection/tests/integration/targets/bulk_host_create/main.yml b/awx_collection/tests/integration/targets/bulk_host_create/main.yml
new file mode 100644
index 0000000000..c5b0437f37
--- /dev/null
+++ b/awx_collection/tests/integration/targets/bulk_host_create/main.yml
@@ -0,0 +1,51 @@
+---
+- name: Generate a random string for test
+  set_fact:
+    test_id: "{{ lookup('password', '/dev/null chars=ascii_letters length=16') }}"
+  when: test_id is not defined
+
+- name: Generate a unique name
+  set_fact:
+    bulk_host_name: "AWX-Collection-tests-bulk_host_create-{{ test_id }}"
+  
+- name: Get our collection package
+  controller_meta:
+  register: controller_meta
+
+- name: Generate the name of our plugin
+  set_fact:
+    plugin_name: "{{ controller_meta.prefix }}.controller_api"
+
+
+- name: Create an inventory
+  inventory:
+    name: "{{ bulk_host_name }}"
+    organization: Default
+    state: present
+  register: inventory_result
+  
+
+- name: Bulk Host Create
+  bulk_host_create:
+    hosts:
+      - name: "123.456.789.123"
+        description: "myhost1"
+        variables:
+          food: carrot
+          color: orange
+      - name: example.dns.gg
+        description: "myhost2"
+        enabled: false
+    inventory: "{{ inventory_result.id }}"
+  register: result
+
+- assert:
+    that:
+      - result is not failed
+
+# cleanup
+- name: Delete inventory
+  inventory:
+    name: "{{ bulk_host_name }}"
+    organization: Default
+    state: absent
\ No newline at end of file
diff --git a/awx_collection/tests/integration/targets/bulk_job_launch/main.yml b/awx_collection/tests/integration/targets/bulk_job_launch/main.yml
new file mode 100644
index 0000000000..6c014e3732
--- /dev/null
+++ b/awx_collection/tests/integration/targets/bulk_job_launch/main.yml
@@ -0,0 +1,69 @@
+---
+- name: Generate a random string for test
+  set_fact:
+    test_id: "{{ lookup('password', '/dev/null chars=ascii_letters length=16') }}"
+  when: test_id is not defined
+
+- name: Generate a unique name
+  set_fact:
+    bulk_job_name: "AWX-Collection-tests-bulk_job_launch-{{ test_id }}"
+  
+- name: Get our collection package
+  controller_meta:
+  register: controller_meta
+
+- name: Generate the name of our plugin
+  set_fact:
+    plugin_name: "{{ controller_meta.prefix }}.controller_api"
+
+- name: Get Inventory
+  set_fact:
+    inventory_id: "{{ lookup(plugin_name, 'inventories', query_params={'name': 'Demo Inventory'}, return_ids=True ) }}"
+
+- name: Create a Job Template
+  job_template:
+    name: "{{ bulk_job_name }}"
+    copy_from: "Demo Job Template"
+    ask_variables_on_launch: true
+    ask_inventory_on_launch: true
+    ask_skip_tags_on_launch: true
+    allow_simultaneous: true
+    state: present
+  register: jt_result
+
+- name: Create Bulk Job
+  bulk_job_launch:
+    name: "{{ bulk_job_name }}"
+    jobs:
+      - unified_job_template: "{{ jt_result.id }}"
+        inventory: "{{ inventory_id }}"
+        skip_tags: "skipfoo,skipbar"
+        extra_data:
+          animal: fish
+          color: orange
+      - unified_job_template: "{{ jt_result.id }}"
+    extra_vars:
+      animal: bear
+      food: carrot
+    skip_tags: "skipbaz"
+    job_tags: "Hello World"
+    limit: "localhost"
+    wait: False
+    inventory: "{{ inventory_id }}"
+  register: result
+
+- assert:
+    that:
+      - result is not failed
+      - "'id' in result"
+      - result['job_info']['skip_tags'] == "skipbaz"
+      - result['job_info']['limit'] == "localhost"
+      - result['job_info']['job_tags'] == "Hello World"
+      - result['job_info']['inventory'] == {{ inventory_id }}
+      - "result['job_info']['extra_vars'] == '{\"animal\": \"bear\", \"food\": \"carrot\"}'"
+
+# cleanup
+- name: Delete Job Template
+  job_template:
+    name: "{{ bulk_job_name }}"
+    state: absent
\ No newline at end of file
diff --git a/awxkit/awxkit/api/pages/bulk.py b/awxkit/awxkit/api/pages/bulk.py
index 197b5c2585..b7342cab45 100644
--- a/awxkit/awxkit/api/pages/bulk.py
+++ b/awxkit/awxkit/api/pages/bulk.py
@@ -10,3 +10,12 @@ class Bulk(base.Base):
 
 
 page.register_page([resources.bulk, (resources.bulk, 'get')], Bulk)
+
+
+class BulkJobLaunch(base.Base):
+    def post(self, payload={}):
+        result = self.connection.post(self.endpoint, payload)
+        return self.walk(result.json()['url'])
+
+
+page.register_page(resources.bulk_job_launch, BulkJobLaunch)
diff --git a/awxkit/awxkit/api/resources.py b/awxkit/awxkit/api/resources.py
index 31bf6c5829..448a0bb582 100644
--- a/awxkit/awxkit/api/resources.py
+++ b/awxkit/awxkit/api/resources.py
@@ -14,6 +14,7 @@ class Resources(object):
     _auth = 'auth/'
     _authtoken = 'authtoken/'
     _bulk = 'bulk/'
+    _bulk_job_launch = 'bulk/job_launch/'
     _config = 'config/'
     _config_attach = 'config/attach/'
     _credential = r'credentials/\d+/'