diff --git a/.gitignore b/.gitignore
index a06a3faaaba0c..5b3731f4eba0a 100644
--- a/.gitignore
+++ b/.gitignore
@@ -26,3 +26,4 @@ docs/.hugo_build.lock
 *.log
 pkg/executables/TestDeployTemplate*
 pkg/files/config
+*.yaml
\ No newline at end of file
diff --git a/cmd/eksctl-anywhere/cmd/createcluster.go b/cmd/eksctl-anywhere/cmd/createcluster.go
index f67779f68dacc..6526be9513731 100644
--- a/cmd/eksctl-anywhere/cmd/createcluster.go
+++ b/cmd/eksctl-anywhere/cmd/createcluster.go
@@ -23,6 +23,7 @@ import (
 	"github.com/aws/eks-anywhere/pkg/validations/createvalidations"
 	"github.com/aws/eks-anywhere/pkg/workflow/management"
 	"github.com/aws/eks-anywhere/pkg/workflows"
+	"github.com/aws/eks-anywhere/pkg/workflows/workload"
 )
 
 type createClusterOptions struct {
@@ -186,7 +187,8 @@ func (cc *createClusterOptions) createCluster(cmd *cobra.Command, _ []string) er
 		WithEksdInstaller().
 		WithPackageInstaller(clusterSpec, cc.installPackages, cc.managementKubeconfig).
 		WithValidatorClients().
-		WithCreateClusterDefaulter(createCLIConfig)
+		WithCreateClusterDefaulter(createCLIConfig).
+		WithClusterApplier()
 
 	if cc.timeoutOptions.noTimeouts {
 		factory.WithNoTimeouts()
@@ -252,6 +254,22 @@ func (cc *createClusterOptions) createCluster(cmd *cobra.Command, _ []string) er
 		}
 
 		err = wflw.Run(ctx)
+	} else if features.UseControllerViaCLIWorkflow().IsActive() && clusterConfig.IsManaged() {
+
+		logger.Info("-----------------------------------------------------")
+		logger.Info("POC Inside controller via CLI workflow")
+
+		createWorkloadCluster := workload.NewCreateWorkload(
+			deps.Provider,
+			deps.ClusterManager,
+			deps.GitOpsFlux,
+			deps.Writer,
+			deps.ClusterApplier,
+			deps.EksdInstaller,
+			deps.PackageInstaller,
+		)
+		err = createWorkloadCluster.Run(ctx, clusterSpec, createValidations)
+
 	} else {
 		err = createCluster.Run(ctx, clusterSpec, createValidations, cc.forceClean)
 	}
diff --git a/pkg/cluster/builder.go b/pkg/cluster/builder.go
index 9f35550499b09..fe1baff6eaf0a 100644
--- a/pkg/cluster/builder.go
+++ b/pkg/cluster/builder.go
@@ -125,7 +125,9 @@ func (b FileSpecBuilder) Build(clusterConfigURL string) (*Spec, error) {
 	}
 
 	releaseVersion := v1alpha1.EksaVersion(release.Version)
-	config.Cluster.Spec.EksaVersion = &releaseVersion
+	if config.Cluster.Spec.EksaVersion == nil {
+		config.Cluster.Spec.EksaVersion = &releaseVersion
+	}
 	eksaRelease := buildEKSARelease(release, bundlesManifest)
 
 	return NewSpec(config, bundlesManifest, eksdReleases, eksaRelease)
diff --git a/pkg/clustermanager/applier.go b/pkg/clustermanager/applier.go
index 7127a0cf027c6..ac463ae734089 100644
--- a/pkg/clustermanager/applier.go
+++ b/pkg/clustermanager/applier.go
@@ -167,13 +167,13 @@ func (a Applier) Run(ctx context.Context, spec *cluster.Spec, managementCluster
 		}
 	}
 
-	a.log.V(3).Info("Waiting for worker nodes to be ready after upgrade")
+	a.log.V(3).Info("Waiting for worker nodes to be ready")
 	retry = a.retrierForWait(waitStartTime)
 	if err := cluster.WaitForCondition(ctx, a.log, client, spec.Cluster, a.conditionCheckoutTotalCount, retry, anywherev1.WorkersReadyCondition); err != nil {
 		return errors.Wrapf(err, "waiting for cluster's workers to be ready")
 	}
 
-	a.log.V(3).Info("Waiting for cluster upgrade to be completed")
+	a.log.V(3).Info("Waiting for cluster to be ready")
 	retry = a.retrierForWait(waitStartTime)
 	if err := cluster.WaitForCondition(ctx, a.log, client, spec.Cluster, a.conditionCheckoutTotalCount, retry, anywherev1.ReadyCondition); err != nil {
 		return errors.Wrapf(err, "waiting for cluster to be ready")
diff --git a/pkg/features/features.go b/pkg/features/features.go
index 479ad7cf8b8f1..baeb2865ad149 100644
--- a/pkg/features/features.go
+++ b/pkg/features/features.go
@@ -5,6 +5,7 @@ const (
 	CloudStackKubeVipDisabledEnvVar = "CLOUDSTACK_KUBE_VIP_DISABLED"
 	CheckpointEnabledEnvVar         = "CHECKPOINT_ENABLED"
 	UseNewWorkflowsEnvVar           = "USE_NEW_WORKFLOWS"
+	UseControllerForWorkloadCli     = "USE_CONTROLLER_FOR_WORKLOAD_CLI"
 )
 
 func FeedGates(featureGates []string) {
@@ -45,3 +46,11 @@ func UseNewWorkflows() Feature {
 		IsActive: globalFeatures.isActiveForEnvVar(UseNewWorkflowsEnvVar),
 	}
 }
+
+// UseControllerViaCLIWorkflow is used for the controller behind the CLI workflow.
+func UseControllerViaCLIWorkflow() Feature {
+	return Feature{
+		Name:     "Use new workflow logic for workload cluster creation leveraging controller via CLI",
+		IsActive: globalFeatures.isActiveForEnvVar(UseControllerForWorkloadCli),
+	}
+}
diff --git a/pkg/task/task.go b/pkg/task/task.go
index 6f64e1f19eb2b..9e25a34b78945 100644
--- a/pkg/task/task.go
+++ b/pkg/task/task.go
@@ -36,7 +36,7 @@ type CommandContext struct {
 	EksdInstaller             interfaces.EksdInstaller
 	PackageInstaller          interfaces.PackageInstaller
 	EksdUpgrader              interfaces.EksdUpgrader
-	ClusterUpgrader           interfaces.ClusterUpgrader
+	ClusterApplier            interfaces.ClusterApplier
 	CAPIManager               interfaces.CAPIManager
 	ClusterSpec               *cluster.Spec
 	CurrentClusterSpec        *cluster.Spec
diff --git a/pkg/workflows/interfaces/interfaces.go b/pkg/workflows/interfaces/interfaces.go
index fe0343fef74c5..4474a48be6ad6 100644
--- a/pkg/workflows/interfaces/interfaces.go
+++ b/pkg/workflows/interfaces/interfaces.go
@@ -81,7 +81,7 @@ type PackageInstaller interface {
 	InstallCuratedPackages(ctx context.Context)
 }
 
-// ClusterUpgrader upgrades the cluster and waits until it's ready.
-type ClusterUpgrader interface {
+// ClusterApplier upgrades the cluster and waits until it's ready.
+type ClusterApplier interface {
 	Run(ctx context.Context, spec *cluster.Spec, managementCluster types.Cluster) error
 }
diff --git a/pkg/workflows/management/upgrade.go b/pkg/workflows/management/upgrade.go
index be9bc38a940de..804ac6a5998af 100644
--- a/pkg/workflows/management/upgrade.go
+++ b/pkg/workflows/management/upgrade.go
@@ -22,7 +22,7 @@ type Upgrade struct {
 	eksdInstaller     interfaces.EksdInstaller
 	eksdUpgrader      interfaces.EksdUpgrader
 	upgradeChangeDiff *types.ChangeDiff
-	clusterUpgrader   interfaces.ClusterUpgrader
+	clusterUpgrader   interfaces.ClusterApplier
 }
 
 // NewUpgrade builds a new upgrade construct.
@@ -33,7 +33,7 @@ func NewUpgrade(provider providers.Provider,
 	writer filewriter.FileWriter,
 	eksdUpgrader interfaces.EksdUpgrader,
 	eksdInstaller interfaces.EksdInstaller,
-	clusterUpgrade interfaces.ClusterUpgrader,
+	clusterUpgrade interfaces.ClusterApplier,
 ) *Upgrade {
 	upgradeChangeDiff := types.NewChangeDiff()
 	return &Upgrade{
@@ -63,7 +63,7 @@ func (c *Upgrade) Run(ctx context.Context, clusterSpec *cluster.Spec, management
 		EksdInstaller:     c.eksdInstaller,
 		EksdUpgrader:      c.eksdUpgrader,
 		UpgradeChangeDiff: c.upgradeChangeDiff,
-		ClusterUpgrader:   c.clusterUpgrader,
+		ClusterApplier:    c.clusterUpgrader,
 	}
 	if features.IsActive(features.CheckpointEnabled()) {
 		return task.NewTaskRunner(&setupAndValidate{}, c.writer, task.WithCheckpointFile()).RunTask(ctx, commandContext)
diff --git a/pkg/workflows/management/upgrade_cluster.go b/pkg/workflows/management/upgrade_cluster.go
index fc68d0c143b26..cb9272c7ec88a 100644
--- a/pkg/workflows/management/upgrade_cluster.go
+++ b/pkg/workflows/management/upgrade_cluster.go
@@ -13,7 +13,7 @@ type upgradeCluster struct{}
 // Run upgradeCluster performs actions needed to upgrade the management cluster.
 func (s *upgradeCluster) Run(ctx context.Context, commandContext *task.CommandContext) task.Task {
 	logger.Info("Upgrading management cluster")
-	if err := commandContext.ClusterUpgrader.Run(ctx, commandContext.ClusterSpec, *commandContext.ManagementCluster); err != nil {
+	if err := commandContext.ClusterApplier.Run(ctx, commandContext.ClusterSpec, *commandContext.ManagementCluster); err != nil {
 		commandContext.SetError(err)
 		return &workflows.CollectMgmtClusterDiagnosticsTask{}
 	}
diff --git a/pkg/workflows/workload/create.go b/pkg/workflows/workload/create.go
new file mode 100644
index 0000000000000..24af73f45bc2e
--- /dev/null
+++ b/pkg/workflows/workload/create.go
@@ -0,0 +1,61 @@
+package workload
+
+import (
+	"context"
+
+	"github.com/aws/eks-anywhere/pkg/cluster"
+	"github.com/aws/eks-anywhere/pkg/filewriter"
+	"github.com/aws/eks-anywhere/pkg/logger"
+	"github.com/aws/eks-anywhere/pkg/providers"
+	"github.com/aws/eks-anywhere/pkg/task"
+	"github.com/aws/eks-anywhere/pkg/workflows/interfaces"
+)
+
+// CreateWorkload is a schema for create cluster.
+type CreateWorkload struct {
+	provider         providers.Provider
+	clusterManager   interfaces.ClusterManager
+	gitOpsManager    interfaces.GitOpsManager
+	writer           filewriter.FileWriter
+	eksdInstaller    interfaces.EksdInstaller
+	clusterApplier   interfaces.ClusterApplier
+	packageInstaller interfaces.PackageInstaller
+}
+
+// NewCreateWorkload builds a new create construct.
+func NewCreateWorkload(provider providers.Provider,
+	clusterManager interfaces.ClusterManager, gitOpsManager interfaces.GitOpsManager,
+	writer filewriter.FileWriter,
+	clusterCreate interfaces.ClusterApplier,
+	eksdInstaller interfaces.EksdInstaller,
+	packageInstaller interfaces.PackageInstaller,
+) *CreateWorkload {
+	return &CreateWorkload{
+		provider:         provider,
+		clusterManager:   clusterManager,
+		gitOpsManager:    gitOpsManager,
+		writer:           writer,
+		eksdInstaller:    eksdInstaller,
+		clusterApplier:   clusterCreate,
+		packageInstaller: packageInstaller,
+	}
+}
+
+// Run Create implements create functionality for workload cluster's create operation.
+func (c *CreateWorkload) Run(ctx context.Context, clusterSpec *cluster.Spec, validator interfaces.Validator) error {
+	logger.Info("POC New workflow creating workload cluster using the controller")
+	commandContext := &task.CommandContext{
+		Provider:          c.provider,
+		ClusterManager:    c.clusterManager,
+		GitOpsManager:     c.gitOpsManager,
+		ClusterSpec:       clusterSpec,
+		Writer:            c.writer,
+		Validations:       validator,
+		ManagementCluster: clusterSpec.ManagementCluster,
+		ClusterApplier:    c.clusterApplier,
+	}
+
+	err := task.NewTaskRunner(&setAndValidateWorkloadTask{}, c.writer).RunTask(ctx, commandContext)
+
+	return err
+}
diff --git a/pkg/workflows/workload/createcluster.go b/pkg/workflows/workload/createcluster.go
new file mode 100644
index 0000000000000..6b3e03bae0f23
--- /dev/null
+++ b/pkg/workflows/workload/createcluster.go
@@ -0,0 +1,83 @@
+package workload
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/aws/eks-anywhere/pkg/logger"
+	"github.com/aws/eks-anywhere/pkg/task"
+	"github.com/aws/eks-anywhere/pkg/validations"
+	"github.com/aws/eks-anywhere/pkg/workflows"
+)
+
+// task related entities
+
+type setAndValidateWorkloadTask struct{}
+
+// SetAndValidateTask implementation
+
+// Run createCluster performs actions needed to create the workload cluster.
+func (s *setAndValidateWorkloadTask) Run(ctx context.Context, commandContext *task.CommandContext) task.Task {
+	logger.Info("POC Performing setup and validations")
+	runner := validations.NewRunner()
+	runner.Register(s.providerValidation(ctx, commandContext)...)
+	runner.Register(commandContext.GitOpsManager.Validations(ctx, commandContext.ClusterSpec)...)
+	runner.Register(commandContext.Validations.PreflightValidations(ctx)...)
+
+	err := runner.Run()
+	if err != nil {
+		commandContext.SetError(err)
+		return nil
+	}
+	return &createCluster{}
+}
+
+func (s *setAndValidateWorkloadTask) providerValidation(ctx context.Context, commandContext *task.CommandContext) []validations.Validation {
+	return []validations.Validation{
+		func() *validations.ValidationResult {
+			return &validations.ValidationResult{
+				Name: fmt.Sprintf("POC workload cluster's %s Provider setup is valid", commandContext.Provider.Name()),
+				Err:  commandContext.Provider.SetupAndValidateCreateCluster(ctx, commandContext.ClusterSpec),
+			}
+		},
+	}
+}
+
+func (s *setAndValidateWorkloadTask) Name() string {
+	return "setup-validate"
+}
+
+func (s *setAndValidateWorkloadTask) Restore(ctx context.Context, commandContext *task.CommandContext, completedTask *task.CompletedTask) (task.Task, error) {
+	return nil, nil
+}
+
+func (s *setAndValidateWorkloadTask) Checkpoint() *task.CompletedTask {
+	return nil
+}
+
+type createCluster struct{}
+
+// Run createCluster performs actions needed to create the management cluster.
+func (s *createCluster) Run(ctx context.Context, commandContext *task.CommandContext) task.Task {
+	logger.Info("Creating workload cluster")
+	if err := commandContext.ClusterApplier.Run(ctx, commandContext.ClusterSpec, *commandContext.ManagementCluster); err != nil {
+		commandContext.SetError(err)
+		return &workflows.CollectMgmtClusterDiagnosticsTask{}
+	}
+
+	return &writeClusterConfig{}
+}
+
+func (s *createCluster) Name() string {
+	return "create-workload-cluster"
+}
+
+func (s *createCluster) Checkpoint() *task.CompletedTask {
+	return &task.CompletedTask{
+		Checkpoint: nil,
+	}
+}
+
+func (s *createCluster) Restore(ctx context.Context, commandContext *task.CommandContext, completedTask *task.CompletedTask) (task.Task, error) {
+	return &writeClusterConfig{}, nil
+}
diff --git a/pkg/workflows/workload/writeclusterconfig.go b/pkg/workflows/workload/writeclusterconfig.go
new file mode 100644
index 0000000000000..d5e1bcbdd8a93
--- /dev/null
+++ b/pkg/workflows/workload/writeclusterconfig.go
@@ -0,0 +1,39 @@
+package workload
+
+import (
+	"context"
+
+	"github.com/aws/eks-anywhere/pkg/clustermarshaller"
+	"github.com/aws/eks-anywhere/pkg/logger"
+	"github.com/aws/eks-anywhere/pkg/task"
+)
+
+type writeClusterConfig struct{}
+
+// Run writeClusterConfig writes new management cluster's cluster config file to the destination after the create process.
+func (s *writeClusterConfig) Run(ctx context.Context, commandContext *task.CommandContext) task.Task {
+	logger.Info("Writing cluster config file")
+	err := clustermarshaller.WriteClusterConfig(commandContext.ClusterSpec, commandContext.Provider.DatacenterConfig(commandContext.ClusterSpec), commandContext.Provider.MachineConfigs(commandContext.ClusterSpec), commandContext.Writer)
+	if err != nil {
+		commandContext.SetError(err)
+	}
+
+	if commandContext.OriginalError == nil {
+		logger.MarkSuccess("Cluster created!")
+	}
+	return nil
+}
+
+func (s *writeClusterConfig) Name() string {
+	return "write-cluster-config"
+}
+
+func (s *writeClusterConfig) Checkpoint() *task.CompletedTask {
+	return &task.CompletedTask{
+		Checkpoint: nil,
+	}
+}
+
+func (s *writeClusterConfig) Restore(ctx context.Context, commandContext *task.CommandContext, completedTask *task.CompletedTask) (task.Task, error) {
+	return nil, nil
+}