diff --git a/app/models/manageiq/providers/awx/automation_manager/configuration_script.rb b/app/models/manageiq/providers/awx/automation_manager/configuration_script.rb
index ea93077..579b30d 100644
--- a/app/models/manageiq/providers/awx/automation_manager/configuration_script.rb
+++ b/app/models/manageiq/providers/awx/automation_manager/configuration_script.rb
@@ -5,6 +5,14 @@ class ManageIQ::Providers::Awx::AutomationManager::ConfigurationScript <
 
   supports :create
 
+  def self.manager_class
+    module_parent
+  end
+
+  def my_zone
+    manager&.my_zone
+  end
+
   def run_with_miq_job(options, userid = nil)
     options[:name] = "Job Template: #{name}"
     options[:ansible_template_id] = id
diff --git a/app/models/manageiq/providers/awx/automation_manager/configuration_workflow.rb b/app/models/manageiq/providers/awx/automation_manager/configuration_workflow.rb
index d13d974..0f62f33 100644
--- a/app/models/manageiq/providers/awx/automation_manager/configuration_workflow.rb
+++ b/app/models/manageiq/providers/awx/automation_manager/configuration_workflow.rb
@@ -1,6 +1,14 @@
 class ManageIQ::Providers::Awx::AutomationManager::ConfigurationWorkflow < ManageIQ::Providers::ExternalAutomationManager::ConfigurationWorkflow
   include ProviderObjectMixin
 
+  def self.manager_class
+    module_parent
+  end
+
+  def my_zone
+    manager&.my_zone
+  end
+
   def run_with_miq_job(options, userid = nil)
     options[:name] = "Workflow Template: #{name}"
     options[:ansible_template_id] = id
diff --git a/app/models/manageiq/providers/awx/automation_manager/job.rb b/app/models/manageiq/providers/awx/automation_manager/job.rb
index 0270b8c..7c7e9cc 100644
--- a/app/models/manageiq/providers/awx/automation_manager/job.rb
+++ b/app/models/manageiq/providers/awx/automation_manager/job.rb
@@ -1,6 +1,8 @@
 class ManageIQ::Providers::Awx::AutomationManager::Job <
   ManageIQ::Providers::ExternalAutomationManager::OrchestrationStack
 
+  include ProviderObjectMixin
+
   belongs_to :ext_management_system, :foreign_key => :ems_id, :class_name => "ManageIQ::Providers::AutomationManager"
   belongs_to :job_template, :foreign_key => :configuration_script_id, :class_name => "::ConfigurationScript"
   belongs_to :playbook, :foreign_key => :configuration_script_base_id
@@ -70,10 +72,16 @@ def job_plays
     resources.where(:resource_category => 'job_play').order(:start_time)
   end
 
+  def provider_object(connection = nil)
+    connection ||= ext_management_system.connect
+    connection.api.jobs.find(ems_ref)
+  end
+
   def refresh_ems
     require 'ansible_tower_client'
-    ext_management_system.with_provider_connection do |connection|
-      update_with_provider_object(connection.api.jobs.find(ems_ref))
+
+    with_provider_object do |raw_job|
+      update_with_provider_object(raw_job)
     end
   rescue AnsibleTowerClient::ResourceNotFoundError
     msg = "Awx Job #{name} with id(#{id}) does not exist on #{ext_management_system.name}"
@@ -145,8 +153,7 @@ def update_plays(raw_job)
 
   def raw_status
     require 'ansible_tower_client'
-    ext_management_system.with_provider_connection do |connection|
-      raw_job = connection.api.jobs.find(ems_ref)
+    with_provider_object do |raw_job|
       self.class.status_class.new(raw_job.status, nil)
     end
   rescue AnsibleTowerClient::ResourceNotFoundError
@@ -159,8 +166,8 @@ def raw_status
 
   def raw_stdout(format = 'txt')
     require 'ansible_tower_client'
-    ext_management_system.with_provider_connection do |connection|
-      connection.api.jobs.find(ems_ref).stdout(format)
+    with_provider_object do |raw_job|
+      raw_job.stdout(format)
     end
   rescue AnsibleTowerClient::ResourceNotFoundError
     msg = "Awx Job #{name} with id(#{id}) or its stdout does not exist on #{ext_management_system.name}"
diff --git a/app/models/manageiq/providers/awx/automation_manager/provision.rb b/app/models/manageiq/providers/awx/automation_manager/provision.rb
new file mode 100644
index 0000000..781e433
--- /dev/null
+++ b/app/models/manageiq/providers/awx/automation_manager/provision.rb
@@ -0,0 +1,5 @@
+class ManageIQ::Providers::Awx::AutomationManager::Provision < ManageIQ::Providers::AutomationManager::Provision
+  include StateMachine
+
+  TASK_DESCRIPTION = N_("AWX Job Template Provision")
+end
diff --git a/app/models/manageiq/providers/awx/automation_manager/provision/state_machine.rb b/app/models/manageiq/providers/awx/automation_manager/provision/state_machine.rb
new file mode 100644
index 0000000..77b19b7
--- /dev/null
+++ b/app/models/manageiq/providers/awx/automation_manager/provision/state_machine.rb
@@ -0,0 +1,56 @@
+module ManageIQ::Providers::Awx::AutomationManager::Provision::StateMachine
+  def run_provision
+    signal :provision
+  end
+
+  def provision
+    stack_class = "#{source.class.module_parent}::#{source.class.stack_type}".constantize
+    stack       = stack_class.create_stack(source)
+
+    phase_context[:stack_id] = stack.id
+    save!
+
+    signal :check_provisioned
+  end
+
+  def check_provisioned
+    if running?
+      requeue_phase
+    else
+      signal :post_provision
+    end
+  end
+
+  def post_provision
+    if succeeded?
+      signal :mark_as_completed
+    else
+      abort_job("Failed to provision stack", "error")
+    end
+  end
+
+  def running?
+    !stack.raw_status.completed?
+  end
+
+  def succeeded?
+    stack.raw_status.succeeded?
+  end
+
+  def mark_as_completed
+    update_and_notify_parent(:state => "finished", :message => "Stack provision is complete")
+    signal :finish
+  end
+
+  def finish
+    mark_execution_servers
+  end
+
+  def stack_klass
+    @stack_klass ||= "#{source.class.module_parent}::#{source.class.stack_type}".constantize
+  end
+
+  def stack
+    @stack ||= stack_klass.find(phase_context[:stack_id])
+  end
+end
diff --git a/app/models/manageiq/providers/awx/automation_manager/provision_workflow.rb b/app/models/manageiq/providers/awx/automation_manager/provision_workflow.rb
new file mode 100644
index 0000000..ce294b8
--- /dev/null
+++ b/app/models/manageiq/providers/awx/automation_manager/provision_workflow.rb
@@ -0,0 +1,18 @@
+class ManageIQ::Providers::Awx::AutomationManager::ProvisionWorkflow < ManageIQ::Providers::AutomationManager::ProvisionWorkflow
+  def dialog_name_from_automate(message = 'get_dialog_name', extra_attrs = {})
+    extra_attrs['platform'] ||= 'awx'
+    super
+  end
+
+  def allowed_configuration_scripts(*args)
+    job_templates = self.class.module_parent::ConfigurationScript.all.map do |cs|
+      build_ci_hash_struct(cs, %w[name description manager_name])
+    end
+
+    workflow_job_templates = self.class.module_parent::ConfigurationWorkflow.all.map do |cs|
+      build_ci_hash_struct(cs, %w[name description manager_name])
+    end
+
+    job_templates + workflow_job_templates
+  end
+end
diff --git a/spec/factories/configuration_script.rb b/spec/factories/configuration_script.rb
index d7ccb72..7d75c56 100644
--- a/spec/factories/configuration_script.rb
+++ b/spec/factories/configuration_script.rb
@@ -2,4 +2,7 @@
   factory :awx_configuration_script,
           :class  => "ManageIQ::Providers::Awx::AutomationManager::ConfigurationScript",
           :parent => :configuration_script
+  factory :awx_configuration_workflow,
+          :class  => "ManageIQ::Providers::Awx::AutomationManager::ConfigurationWorkflow",
+          :parent => :configuration_script
 end
diff --git a/spec/factories/miq_request_task.rb b/spec/factories/miq_request_task.rb
new file mode 100644
index 0000000..4983ace
--- /dev/null
+++ b/spec/factories/miq_request_task.rb
@@ -0,0 +1,3 @@
+FactoryBot.define do
+  factory :miq_provision_awx, :parent => :miq_provision, :class => "ManageIQ::Providers::Awx::AutomationManager::Provision"
+end
diff --git a/spec/factories/orchestration_stack.rb b/spec/factories/orchestration_stack.rb
new file mode 100644
index 0000000..ed3d7f3
--- /dev/null
+++ b/spec/factories/orchestration_stack.rb
@@ -0,0 +1,3 @@
+FactoryBot.define do
+  factory :orchestration_stack_awx, :class => ManageIQ::Providers::Awx::AutomationManager::Job
+end
diff --git a/spec/models/manageiq/providers/awx/automation_manager/provision_spec.rb b/spec/models/manageiq/providers/awx/automation_manager/provision_spec.rb
new file mode 100644
index 0000000..5d7736b
--- /dev/null
+++ b/spec/models/manageiq/providers/awx/automation_manager/provision_spec.rb
@@ -0,0 +1,126 @@
+describe ManageIQ::Providers::Awx::AutomationManager::Provision do
+  require "ansible_tower_client"
+
+  let(:admin)        { FactoryBot.create(:user_admin) }
+  let(:zone)         { EvmSpecHelper.local_miq_server.zone }
+  let(:ems)          { FactoryBot.create(:automation_manager_awx, :zone => zone).tap { |ems| ems.authentications << FactoryBot.create(:authentication, :status => "Valid", :auth_key => "abcd") } }
+  let(:job_template) { FactoryBot.create(:awx_configuration_script, :manager => ems) }
+  let(:miq_request)  { FactoryBot.create(:miq_provision_request, :requester => admin, :source => job_template)}
+  let(:options)      { {:source => [job_template.id, job_template.name]} }
+  let(:new_stack)    { FactoryBot.create(:orchestration_stack_awx, :ext_management_system => ems, :status => stack_status) }
+  let(:stack_status) { "pending" }
+  let(:phase)        { nil }
+  let(:subject)     do
+    FactoryBot.create(
+      :miq_provision_awx,
+      :userid       => admin.userid,
+      :miq_request  => miq_request,
+      :source       => job_template,
+      :request_type => 'template',
+      :state        => 'pending',
+      :status       => 'Ok',
+      :options      => options,
+      :phase        => phase
+    )
+  end
+
+  it ".my_role" do
+    expect(subject.my_role).to eq("ems_operations")
+  end
+
+  it ".my_queue_name" do
+    expect(subject.my_queue_name).to eq(ems.queue_name_for_ems_operations)
+  end
+
+  describe ".run_provision" do
+    before do
+      allow(described_class.module_parent::Job).to receive(:create_stack).with(job_template).and_return(new_stack)
+    end
+
+    it "calls create_stack" do
+      expect(described_class.module_parent::Job).to receive(:create_stack)
+
+      subject.run_provision
+    end
+
+    it "sets stack_id" do
+      subject.run_provision
+
+      expect(subject.reload.phase_context).to include(:stack_id => new_stack.id)
+    end
+
+    it "queues check_provisioned" do
+      subject.instance_variable_set(:@stack, new_stack)
+      allow(new_stack).to receive(:raw_status).and_return(new_stack.class.status_class.new(stack_status, nil))
+
+      subject.run_provision
+
+      expect(subject.reload.phase).to eq("check_provisioned")
+    end
+
+    context "when create_stack fails" do
+      before do
+        expect(described_class.module_parent::Job).to receive(:create_stack).and_raise
+      end
+
+      it "marks the job as failed" do
+        subject.run_provision
+
+        expect(subject.reload).to have_attributes(:state => "finished", :status => "Error")
+      end
+    end
+  end
+
+  describe "check_provisioned" do
+    let(:phase) { "check_provisioned" }
+
+    before do
+      allow(new_stack).to receive(:raw_status).and_return(new_stack.class.status_class.new(stack_status, nil))
+      subject.instance_variable_set(:@stack, new_stack)
+      subject.phase_context[:stack_id] = new_stack.id
+    end
+
+    context "when the plan is still running" do
+      let(:stack_status) { "running" }
+
+      it "requeues check_provisioned" do
+        subject.check_provisioned
+
+        expect(subject.reload).to have_attributes(
+          :phase  => "check_provisioned",
+          :state  => "pending",
+          :status => "Ok"
+        )
+      end
+    end
+
+    context "when the plan is finished" do
+      let(:stack_status) { "successful" }
+
+      it "finishes the job" do
+        subject.check_provisioned
+
+        expect(subject.reload).to have_attributes(
+          :phase  => "finish",
+          :state  => "finished",
+          :status => "Ok"
+        )
+      end
+    end
+
+    context "when the plan is errored" do
+      let(:stack_status) { "failed" }
+
+      it "finishes the job" do
+        subject.phase_context[:stack_id] = new_stack.id
+        subject.check_provisioned
+
+        expect(subject.reload).to have_attributes(
+          :phase  => "finish",
+          :state  => "finished",
+          :status => "Error"
+        )
+      end
+    end
+  end
+end
diff --git a/spec/models/manageiq/providers/awx/automation_manager/provision_workflow_spec.rb b/spec/models/manageiq/providers/awx/automation_manager/provision_workflow_spec.rb
new file mode 100644
index 0000000..eb013c4
--- /dev/null
+++ b/spec/models/manageiq/providers/awx/automation_manager/provision_workflow_spec.rb
@@ -0,0 +1,54 @@
+describe ManageIQ::Providers::Awx::AutomationManager::ProvisionWorkflow do
+  include Spec::Support::WorkflowHelper
+
+  let(:admin)        { FactoryBot.create(:user_with_group) }
+  let(:manager)      { FactoryBot.create(:automation_manager_awx) }
+  let(:dialog)       { FactoryBot.create(:miq_provision_configuration_script_dialog) }
+
+  describe "#allowed_configuration_scripts" do
+    context "with no configuration_scripts or configuration_workflows" do
+      it "returns an empty set" do
+        workflow = described_class.new({:provision_dialog_name => dialog.name}, admin.userid)
+        expect(workflow.allowed_configuration_scripts).to be_empty
+      end
+    end
+
+    context "with a configuration_script" do
+      let!(:configuration_script) { FactoryBot.create(:awx_configuration_script,   :manager => manager) }
+
+      it "returns the configuration script" do
+        workflow = described_class.new({:provision_dialog_name => dialog.name}, admin.userid)
+
+        allowed = workflow.allowed_configuration_scripts
+        expect(allowed.count).to eq(1)
+        expect(allowed.first).to have_attributes(:id => configuration_script.id, :name => configuration_script.name)
+      end
+    end
+
+    context "with a configuration_workflow" do
+      let!(:configuration_workflow) { FactoryBot.create(:awx_configuration_workflow, :manager => manager) }
+
+      it "returns the configuration workflow" do
+        workflow = described_class.new({:provision_dialog_name => dialog.name}, admin.userid)
+
+        allowed = workflow.allowed_configuration_scripts
+        expect(allowed.count).to eq(1)
+        expect(allowed.first).to have_attributes(:id => configuration_workflow.id, :name => configuration_workflow.name)
+      end
+    end
+
+    context "with a configuration_script and a configuration_workflow" do
+      let!(:configuration_script)   { FactoryBot.create(:awx_configuration_script,   :manager => manager) }
+      let!(:configuration_workflow) { FactoryBot.create(:awx_configuration_workflow, :manager => manager) }
+
+      it "includes both job_templates and workflow_job_templates" do
+        workflow = described_class.new({:provision_dialog_name => dialog.name}, admin.userid)
+
+        allowed = workflow.allowed_configuration_scripts
+        expect(allowed.count).to eq(2)
+        expect(allowed.first).to have_attributes(:id => configuration_script.id, :name => configuration_script.name)
+        expect(allowed.last).to  have_attributes(:id => configuration_workflow.id, :name => configuration_workflow.name)
+      end
+    end
+  end
+end