diff --git a/api/v1beta1/types_class.go b/api/v1beta1/types_class.go
index 452eab517f8..b9690694a91 100644
--- a/api/v1beta1/types_class.go
+++ b/api/v1beta1/types_class.go
@@ -48,6 +48,7 @@ type AzureClusterClassSpec struct {
 	// - GermanCloud: "AzureGermanCloud"
 	// - PublicCloud: "AzurePublicCloud"
 	// - USGovernmentCloud: "AzureUSGovernmentCloud"
+	// - StackCloud: "HybridEnvironment"
 	//
 	// Note that values other than the default must also be accompanied by corresponding changes to the
 	// aso-controller-settings Secret to configure ASO to refer to the non-Public cloud. ASO currently does
@@ -77,6 +78,12 @@ type AzureClusterClassSpec struct {
 	// See: https://learn.microsoft.com/azure/reliability/availability-zones-overview
 	// +optional
 	FailureDomains clusterv1.FailureDomains `json:"failureDomains,omitempty"`
+
+	// ARMEndpoint specifies a URL for the ARM Resource Manager endpoint.
+	// It may only be specified when the AzureEnvironment is set to AzureStackCloud,
+	// in which case it is required.
+	// +optional
+	ARMEndpoint string `json:"armEndpoint,omitempty"`
 }
 
 // AzureManagedControlPlaneClassSpec defines the AzureManagedControlPlane properties that may be shared across several azure managed control planes.
@@ -186,6 +193,7 @@ type AzureManagedControlPlaneClassSpec struct {
 	// - PublicCloud: "AzurePublicCloud"
 	// - USGovernmentCloud: "AzureUSGovernmentCloud"
 	//
+	//
 	// Note that values other than the default must also be accompanied by corresponding changes to the
 	// aso-controller-settings Secret to configure ASO to refer to the non-Public cloud. ASO currently does
 	// not support referring to multiple different clouds in a single installation. The following fields must
diff --git a/azure/defaults.go b/azure/defaults.go
index 0f86619c7d1..28f7aca1d34 100644
--- a/azure/defaults.go
+++ b/azure/defaults.go
@@ -27,6 +27,7 @@ import (
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/policy"
 	"github.com/Azure/azure-sdk-for-go/sdk/resourcemanager/compute/armcompute/v5"
 	"github.com/Azure/azure-sdk-for-go/sdk/tracing/azotel"
+	"github.com/Azure/go-autorest/autorest/azure"
 	"go.opentelemetry.io/otel"
 
 	"sigs.k8s.io/cluster-api-provider-azure/util/tele"
@@ -44,6 +45,8 @@ const (
 	ChinaCloudName = "AzureChinaCloud"
 	// USGovernmentCloudName is the name of the Azure US Government cloud.
 	USGovernmentCloudName = "AzureUSGovernmentCloud"
+	// StackCloudName is the name for Azure Stack hybrid cloud environments.
+	StackCloudName = "HybridEnvironment"
 )
 
 const (
@@ -109,6 +112,16 @@ const (
 	CustomHeaderPrefix = "infrastructure.cluster.x-k8s.io/custom-header-"
 )
 
+const (
+	// StackAPIVersion is the API version profile to set for ARM clients. See:
+	// https://learn.microsoft.com/en-us/azure-stack/user/azure-stack-profiles-azure-resource-manager-versions?view=azs-2408#overview-of-the-2020-09-01-hybrid-profile
+	StackAPIVersionProfile = "2020-06-01"
+
+	// StackDiskAPIVersionProfile is the API Version to set for the disk client.
+	// API Version Profile "2020-06-01" is not supported for disks.
+	StackDiskAPIVersionProfile = "2018-06-01"
+)
+
 var (
 	// LinuxBootstrapExtensionCommand is the command the VM bootstrap extension will execute to verify Linux nodes bootstrap completes successfully.
 	LinuxBootstrapExtensionCommand = fmt.Sprintf("for i in $(seq 1 %d); do test -f %s && break; if [ $i -eq %d ]; then exit 1; else sleep %d; fi; done", bootstrapExtensionRetries, bootstrapSentinelFile, bootstrapExtensionRetries, bootstrapExtensionSleep)
@@ -357,7 +370,7 @@ func UserAgent() string {
 }
 
 // ARMClientOptions returns default ARM client options for CAPZ SDK v2 requests.
-func ARMClientOptions(azureEnvironment string, extraPolicies ...policy.Policy) (*arm.ClientOptions, error) {
+func ARMClientOptions(azureEnvironment, armEndpoint string, extraPolicies ...policy.Policy) (*arm.ClientOptions, error) {
 	opts := &arm.ClientOptions{}
 
 	switch azureEnvironment {
@@ -367,6 +380,21 @@ func ARMClientOptions(azureEnvironment string, extraPolicies ...policy.Policy) (
 		opts.Cloud = cloud.AzureChina
 	case USGovernmentCloudName:
 		opts.Cloud = cloud.AzureGovernment
+	case StackCloudName:
+		cloudEnv, err := azure.EnvironmentFromURL(armEndpoint)
+		if err != nil {
+			return nil, fmt.Errorf("unable to get Azure Stack cloud environment: %w", err)
+		}
+		opts.APIVersion = StackAPIVersionProfile
+		opts.Cloud = cloud.Configuration{
+			ActiveDirectoryAuthorityHost: cloudEnv.ActiveDirectoryEndpoint,
+			Services: map[cloud.ServiceName]cloud.ServiceConfiguration{
+				cloud.ResourceManager: {
+					Audience: cloudEnv.TokenAudience,
+					Endpoint: cloudEnv.ResourceManagerEndpoint,
+				},
+			},
+		}
 	case "":
 		// No cloud name provided, so leave at defaults.
 	default:
diff --git a/azure/defaults_test.go b/azure/defaults_test.go
index 88708df72cf..2a627dcdc80 100644
--- a/azure/defaults_test.go
+++ b/azure/defaults_test.go
@@ -38,6 +38,7 @@ func TestARMClientOptions(t *testing.T) {
 	tests := []struct {
 		name          string
 		cloudName     string
+		armEndpoint   string
 		expectedCloud cloud.Configuration
 		expectError   bool
 	}{
@@ -72,7 +73,7 @@ func TestARMClientOptions(t *testing.T) {
 			t.Parallel()
 			g := NewWithT(t)
 
-			opts, err := ARMClientOptions(tc.cloudName)
+			opts, err := ARMClientOptions(tc.cloudName, tc.armEndpoint)
 			if tc.expectError {
 				g.Expect(err).To(HaveOccurred())
 				return
@@ -99,7 +100,7 @@ func TestPerCallPolicies(t *testing.T) {
 	defer server.Close()
 
 	// Call the factory function and ensure it has both PerCallPolicies.
-	opts, err := ARMClientOptions("")
+	opts, err := ARMClientOptions("", "")
 	g.Expect(err).NotTo(HaveOccurred())
 	g.Expect(opts.PerCallPolicies).To(HaveLen(2))
 	g.Expect(opts.PerCallPolicies).To(ContainElement(BeAssignableToTypeOf(correlationIDPolicy{})))
@@ -184,7 +185,7 @@ func TestCustomPutPatchHeaderPolicy(t *testing.T) {
 			// Create options with a custom PUT/PATCH header per-call policy
 			getterMock := mock_azure.NewMockResourceSpecGetterWithHeaders(mockCtrl)
 			getterMock.EXPECT().CustomHeaders().Return(tc.headers).AnyTimes()
-			opts, err := ARMClientOptions("", CustomPutPatchHeaderPolicy{Headers: tc.headers})
+			opts, err := ARMClientOptions("", "", CustomPutPatchHeaderPolicy{Headers: tc.headers})
 			g.Expect(err).NotTo(HaveOccurred())
 
 			// Create a request
diff --git a/azure/errors.go b/azure/errors.go
index 0d719e80037..8e77269acd4 100644
--- a/azure/errors.go
+++ b/azure/errors.go
@@ -34,6 +34,12 @@ func ResourceNotFound(err error) bool {
 	return errors.As(err, &rerr) && rerr.StatusCode == http.StatusNotFound
 }
 
+// BadRequest parses an error to check if it its status code is Bad Request (400).
+func BadRequest(err error) bool {
+	var rerr *azcore.ResponseError
+	return errors.As(err, &rerr) && rerr.StatusCode == http.StatusBadRequest
+}
+
 // VMDeletedError is returned when a virtual machine is deleted outside of capz.
 type VMDeletedError struct {
 	ProviderID string
diff --git a/azure/scope/clients.go b/azure/scope/clients.go
index b766b066747..e472cc8925c 100644
--- a/azure/scope/clients.go
+++ b/azure/scope/clients.go
@@ -81,12 +81,12 @@ func (c *AzureClients) HashKey() string {
 	return base64.URLEncoding.EncodeToString(hasher.Sum(nil))
 }
 
-func (c *AzureClients) setCredentialsWithProvider(ctx context.Context, subscriptionID, environmentName string, credentialsProvider CredentialsProvider) error {
+func (c *AzureClients) setCredentialsWithProvider(ctx context.Context, subscriptionID, environmentName, armEndpoint string, credentialsProvider CredentialsProvider) error {
 	if credentialsProvider == nil {
 		return fmt.Errorf("credentials provider cannot have an empty value")
 	}
 
-	settings, err := c.getSettingsFromEnvironment(environmentName)
+	settings, err := c.getSettingsFromEnvironment(environmentName, armEndpoint)
 	if err != nil {
 		return err
 	}
@@ -121,7 +121,7 @@ func (c *AzureClients) setCredentialsWithProvider(ctx context.Context, subscript
 	return err
 }
 
-func (c *AzureClients) getSettingsFromEnvironment(environmentName string) (s auth.EnvironmentSettings, err error) {
+func (c *AzureClients) getSettingsFromEnvironment(environmentName, armEndpoint string) (s auth.EnvironmentSettings, err error) {
 	s = auth.EnvironmentSettings{
 		Values: map[string]string{},
 	}
@@ -138,6 +138,8 @@ func (c *AzureClients) getSettingsFromEnvironment(environmentName string) (s aut
 	setValue(s, "AZURE_AD_RESOURCE")
 	if v := s.Values["AZURE_ENVIRONMENT"]; v == "" {
 		s.Environment = azureautorest.PublicCloud
+	} else if len(armEndpoint) > 0 {
+		s.Environment, err = azureautorest.EnvironmentFromURL(armEndpoint)
 	} else {
 		s.Environment, err = azureautorest.EnvironmentFromName(v)
 	}
diff --git a/azure/scope/cluster.go b/azure/scope/cluster.go
index bff9416387c..0c2d55f45ae 100644
--- a/azure/scope/cluster.go
+++ b/azure/scope/cluster.go
@@ -84,7 +84,8 @@ func NewClusterScope(ctx context.Context, params ClusterScopeParams) (*ClusterSc
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to init credentials provider")
 	}
-	err = params.AzureClients.setCredentialsWithProvider(ctx, params.AzureCluster.Spec.SubscriptionID, params.AzureCluster.Spec.AzureEnvironment, credentialsProvider)
+	spec := params.AzureCluster.Spec
+	err = params.AzureClients.setCredentialsWithProvider(ctx, spec.SubscriptionID, spec.AzureEnvironment, spec.ARMEndpoint, credentialsProvider)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to configure azure settings and credentials for Identity")
 	}
@@ -557,7 +558,7 @@ func (s *ClusterScope) VNetSpec() azure.ASOResourceSpecGetter[*asonetworkv1api20
 
 // PrivateDNSSpec returns the private dns zone spec.
 func (s *ClusterScope) PrivateDNSSpec() (zoneSpec azure.ResourceSpecGetter, linkSpec, recordSpec []azure.ResourceSpecGetter) {
-	if s.IsAPIServerPrivate() {
+	if s.IsAPIServerPrivate() && !s.IsHybridEnvironment() {
 		resourceGroup := s.ResourceGroup()
 		if s.AzureCluster.Spec.NetworkSpec.PrivateDNSZoneResourceGroup != "" {
 			resourceGroup = s.AzureCluster.Spec.NetworkSpec.PrivateDNSZoneResourceGroup
@@ -1233,3 +1234,8 @@ func (s *ClusterScope) getLastAppliedSecurityRules(nsgName string) map[string]in
 	}
 	return lastAppliedSecurityRules
 }
+
+// IsHybridEnvironment returns true if the cluster is running on Azure Stack.
+func (s *ClusterScope) IsHybridEnvironment() bool {
+	return strings.EqualFold(s.Environment.Name, azure.StackCloudName)
+}
diff --git a/azure/scope/machine.go b/azure/scope/machine.go
index d6a96964f36..1f91de2eb59 100644
--- a/azure/scope/machine.go
+++ b/azure/scope/machine.go
@@ -150,7 +150,8 @@ func (m *MachineScope) InitMachineCache(ctx context.Context) error {
 		}
 
 		m.cache.availabilitySetSKU, err = skuCache.Get(ctx, string(armcompute.AvailabilitySetSKUTypesAligned), resourceskus.AvailabilitySets)
-		if err != nil {
+		// Resource SKU API for availability sets may not be available in Azure Stack environments.
+		if err != nil && !strings.EqualFold(m.CloudEnvironment(), "HybridEnvironment") {
 			return errors.Wrapf(err, "failed to get availability set SKU %s in compute api", string(armcompute.AvailabilitySetSKUTypesAligned))
 		}
 	}
@@ -494,12 +495,13 @@ func (m *MachineScope) AvailabilitySetSpec() azure.ResourceSpecGetter {
 	}
 
 	spec := &availabilitysets.AvailabilitySetSpec{
-		Name:           availabilitySetName,
-		ResourceGroup:  m.NodeResourceGroup(),
-		ClusterName:    m.ClusterName(),
-		Location:       m.Location(),
-		SKU:            nil,
-		AdditionalTags: m.AdditionalTags(),
+		Name:             availabilitySetName,
+		ResourceGroup:    m.NodeResourceGroup(),
+		ClusterName:      m.ClusterName(),
+		Location:         m.Location(),
+		CloudEnvironment: m.CloudEnvironment(),
+		SKU:              nil,
+		AdditionalTags:   m.AdditionalTags(),
 	}
 
 	if m.cache != nil {
diff --git a/azure/scope/managedcontrolplane.go b/azure/scope/managedcontrolplane.go
index 0956cf6478a..53e4e4c934f 100644
--- a/azure/scope/managedcontrolplane.go
+++ b/azure/scope/managedcontrolplane.go
@@ -97,7 +97,7 @@ func NewManagedControlPlaneScope(ctx context.Context, params ManagedControlPlane
 		return nil, errors.Wrap(err, "failed to init credentials provider")
 	}
 
-	if err := params.AzureClients.setCredentialsWithProvider(ctx, params.ControlPlane.Spec.SubscriptionID, params.ControlPlane.Spec.AzureEnvironment, credentialsProvider); err != nil {
+	if err := params.AzureClients.setCredentialsWithProvider(ctx, params.ControlPlane.Spec.SubscriptionID, params.ControlPlane.Spec.AzureEnvironment, "", credentialsProvider); err != nil {
 		return nil, errors.Wrap(err, "failed to configure azure settings and credentials for Identity")
 	}
 
diff --git a/azure/services/availabilitysets/client.go b/azure/services/availabilitysets/client.go
index 859789d3f37..800d5a4560e 100644
--- a/azure/services/availabilitysets/client.go
+++ b/azure/services/availabilitysets/client.go
@@ -34,7 +34,7 @@ type AzureClient struct {
 
 // NewClient creates a new availability sets client from an authorizer.
 func NewClient(auth azure.Authorizer) (*AzureClient, error) {
-	opts, err := azure.ARMClientOptions(auth.CloudEnvironment())
+	opts, err := azure.ARMClientOptions(auth.CloudEnvironment(), auth.BaseURI())
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to create availabilitysets client options")
 	}
diff --git a/azure/services/availabilitysets/spec.go b/azure/services/availabilitysets/spec.go
index ea522da07ee..809e9fa0ef9 100644
--- a/azure/services/availabilitysets/spec.go
+++ b/azure/services/availabilitysets/spec.go
@@ -19,24 +19,27 @@ package availabilitysets
 import (
 	"context"
 	"strconv"
+	"strings"
 
 	"github.com/Azure/azure-sdk-for-go/sdk/resourcemanager/compute/armcompute/v5"
 	"github.com/pkg/errors"
 	"k8s.io/utils/ptr"
 
 	infrav1 "sigs.k8s.io/cluster-api-provider-azure/api/v1beta1"
+	"sigs.k8s.io/cluster-api-provider-azure/azure"
 	"sigs.k8s.io/cluster-api-provider-azure/azure/converters"
 	"sigs.k8s.io/cluster-api-provider-azure/azure/services/resourceskus"
 )
 
 // AvailabilitySetSpec defines the specification for an availability set.
 type AvailabilitySetSpec struct {
-	Name           string
-	ResourceGroup  string
-	ClusterName    string
-	Location       string
-	SKU            *resourceskus.SKU
-	AdditionalTags infrav1.Tags
+	Name             string
+	ResourceGroup    string
+	ClusterName      string
+	Location         string
+	CloudEnvironment string
+	SKU              *resourceskus.SKU
+	AdditionalTags   infrav1.Tags
 }
 
 // ResourceName returns the name of the availability set.
@@ -64,20 +67,10 @@ func (s *AvailabilitySetSpec) Parameters(_ context.Context, existing interface{}
 		return nil, nil
 	}
 
-	if s.SKU == nil {
-		return nil, errors.New("unable to get required availability set SKU from machine cache")
-	}
-
-	var faultDomainCount *int32
-	faultDomainCountStr, ok := s.SKU.GetCapability(resourceskus.MaximumPlatformFaultDomainCount)
-	if !ok {
-		return nil, errors.Errorf("unable to get required availability set SKU capability %s", resourceskus.MaximumPlatformFaultDomainCount)
-	}
-	count, err := strconv.ParseInt(faultDomainCountStr, 10, 32)
+	faultDomainCount, err := getFaultDomainCount(s.SKU, s.CloudEnvironment)
 	if err != nil {
-		return nil, errors.Wrapf(err, "unable to parse availability set fault domain count")
+		return nil, err
 	}
-	faultDomainCount = ptr.To[int32](int32(count))
 
 	asParams := armcompute.AvailabilitySet{
 		SKU: &armcompute.SKU{
@@ -98,3 +91,27 @@ func (s *AvailabilitySetSpec) Parameters(_ context.Context, existing interface{}
 
 	return asParams, nil
 }
+
+func getFaultDomainCount(SKU *resourceskus.SKU, cloudEnvironment string) (*int32, error) {
+	// Azure Stack environments may not implement the resource SKU API
+	// for availability sets. Use a default value instead.
+	if strings.EqualFold(cloudEnvironment, azure.StackCloudName) {
+		return ptr.To(int32(2)), nil
+	}
+
+	if SKU == nil {
+		return nil, errors.New("unable to get required availability set SKU from machine cache")
+	}
+
+	var faultDomainCount *int32
+	faultDomainCountStr, ok := SKU.GetCapability(resourceskus.MaximumPlatformFaultDomainCount)
+	if !ok {
+		return nil, errors.Errorf("unable to get required availability set SKU capability %s", resourceskus.MaximumPlatformFaultDomainCount)
+	}
+	count, err := strconv.ParseInt(faultDomainCountStr, 10, 32)
+	if err != nil {
+		return nil, errors.Wrapf(err, "unable to parse availability set fault domain count")
+	}
+	faultDomainCount = ptr.To[int32](int32(count))
+	return faultDomainCount, nil
+}
diff --git a/azure/services/disks/client.go b/azure/services/disks/client.go
index 58cdb4345fc..af4ebe27aab 100644
--- a/azure/services/disks/client.go
+++ b/azure/services/disks/client.go
@@ -18,6 +18,7 @@ package disks
 
 import (
 	"context"
+	"strings"
 	"time"
 
 	"github.com/Azure/azure-sdk-for-go/sdk/azcore/runtime"
@@ -37,7 +38,10 @@ type azureClient struct {
 
 // newClient creates a new disks client from an authorizer.
 func newClient(auth azure.Authorizer, apiCallTimeout time.Duration) (*azureClient, error) {
-	opts, err := azure.ARMClientOptions(auth.CloudEnvironment())
+	opts, err := azure.ARMClientOptions(auth.CloudEnvironment(), auth.BaseURI())
+	if strings.EqualFold(auth.CloudEnvironment(), azure.StackCloudName) {
+		opts.APIVersion = azure.StackDiskAPIVersionProfile
+	}
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to create disks client options")
 	}
diff --git a/azure/services/identities/client.go b/azure/services/identities/client.go
index e12a25a10ab..c1cca1bd747 100644
--- a/azure/services/identities/client.go
+++ b/azure/services/identities/client.go
@@ -41,7 +41,7 @@ type AzureClient struct {
 
 // NewClient creates a new MSI client from an authorizer.
 func NewClient(auth azure.Authorizer) (Client, error) {
-	opts, err := azure.ARMClientOptions(auth.CloudEnvironment())
+	opts, err := azure.ARMClientOptions(auth.CloudEnvironment(), auth.BaseURI())
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to create identities client options")
 	}
@@ -54,7 +54,7 @@ func NewClient(auth azure.Authorizer) (Client, error) {
 
 // NewClientBySub creates a new MSI client with a given subscriptionID.
 func NewClientBySub(auth azure.Authorizer, subscriptionID string) (Client, error) {
-	opts, err := azure.ARMClientOptions(auth.CloudEnvironment())
+	opts, err := azure.ARMClientOptions(auth.CloudEnvironment(), auth.BaseURI())
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to create identities client options")
 	}
diff --git a/azure/services/inboundnatrules/client.go b/azure/services/inboundnatrules/client.go
index 0b8f5f6e3a9..8500cd5d3aa 100644
--- a/azure/services/inboundnatrules/client.go
+++ b/azure/services/inboundnatrules/client.go
@@ -44,7 +44,7 @@ var _ client = (*azureClient)(nil)
 
 // newClient creates a new inbound NAT rules client from an authorizer.
 func newClient(auth azure.Authorizer, apiCallTimeout time.Duration) (*azureClient, error) {
-	opts, err := azure.ARMClientOptions(auth.CloudEnvironment())
+	opts, err := azure.ARMClientOptions(auth.CloudEnvironment(), auth.BaseURI())
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to create inboundnatrules client options")
 	}
diff --git a/azure/services/loadbalancers/client.go b/azure/services/loadbalancers/client.go
index 0981f4855a4..972875c398c 100644
--- a/azure/services/loadbalancers/client.go
+++ b/azure/services/loadbalancers/client.go
@@ -39,7 +39,7 @@ type azureClient struct {
 
 // newClient creates a new load balancer client from an authorizer.
 func newClient(auth azure.Authorizer, apiCallTimeout time.Duration) (*azureClient, error) {
-	opts, err := azure.ARMClientOptions(auth.CloudEnvironment())
+	opts, err := azure.ARMClientOptions(auth.CloudEnvironment(), auth.BaseURI())
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to get load balancer client options")
 	}
@@ -86,7 +86,7 @@ func (ac *azureClient) CreateOrUpdateAsync(ctx context.Context, spec azure.Resou
 	}
 
 	// Create a new client that knows how to add etag headers to the request.
-	clientOpts, err := azure.ARMClientOptions(ac.auth.CloudEnvironment(), extraPolicies...)
+	clientOpts, err := azure.ARMClientOptions(ac.auth.CloudEnvironment(), ac.auth.BaseURI(), extraPolicies...)
 	if err != nil {
 		return nil, nil, errors.Wrap(err, "failed to create loadbalancer client options")
 	}
diff --git a/azure/services/networkinterfaces/client.go b/azure/services/networkinterfaces/client.go
index 62fefd8d304..114f827cd05 100644
--- a/azure/services/networkinterfaces/client.go
+++ b/azure/services/networkinterfaces/client.go
@@ -37,7 +37,7 @@ type azureClient struct {
 
 // NewClient creates a new network interfaces client from an authorizer.
 func NewClient(auth azure.Authorizer, apiCallTimeout time.Duration) (*azureClient, error) { //nolint:revive // leave it as is
-	opts, err := azure.ARMClientOptions(auth.CloudEnvironment())
+	opts, err := azure.ARMClientOptions(auth.CloudEnvironment(), auth.BaseURI())
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to create networkinterfaces client options")
 	}
diff --git a/azure/services/privatedns/link_client.go b/azure/services/privatedns/link_client.go
index 0c2104d32ae..7c913633be1 100644
--- a/azure/services/privatedns/link_client.go
+++ b/azure/services/privatedns/link_client.go
@@ -37,7 +37,7 @@ type azureVirtualNetworkLinksClient struct {
 
 // newVirtualNetworkLinksClient creates a virtual network links client from an authorizer.
 func newVirtualNetworkLinksClient(auth azure.Authorizer, apiCallTimeout time.Duration) (*azureVirtualNetworkLinksClient, error) {
-	opts, err := azure.ARMClientOptions(auth.CloudEnvironment())
+	opts, err := azure.ARMClientOptions(auth.CloudEnvironment(), auth.BaseURI())
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to create virtualnetworkslink client options")
 	}
diff --git a/azure/services/privatedns/record_client.go b/azure/services/privatedns/record_client.go
index 01e6ce7ff8c..1483bd77cf2 100644
--- a/azure/services/privatedns/record_client.go
+++ b/azure/services/privatedns/record_client.go
@@ -34,7 +34,7 @@ type azureRecordsClient struct {
 
 // newRecordSetsClient creates a record sets client from an authorizer.
 func newRecordSetsClient(auth azure.Authorizer) (*azureRecordsClient, error) {
-	opts, err := azure.ARMClientOptions(auth.CloudEnvironment())
+	opts, err := azure.ARMClientOptions(auth.CloudEnvironment(), auth.BaseURI())
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to create recordsets client options")
 	}
diff --git a/azure/services/privatedns/zone_client.go b/azure/services/privatedns/zone_client.go
index 86e137251c5..193dbd5f102 100644
--- a/azure/services/privatedns/zone_client.go
+++ b/azure/services/privatedns/zone_client.go
@@ -37,7 +37,7 @@ type azureZonesClient struct {
 
 // newPrivateZonesClient creates a private zones client from an authorizer.
 func newPrivateZonesClient(auth azure.Authorizer, apiCallTimeout time.Duration) (*azureZonesClient, error) {
-	opts, err := azure.ARMClientOptions(auth.CloudEnvironment())
+	opts, err := azure.ARMClientOptions(auth.CloudEnvironment(), auth.BaseURI())
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to create privatezones client options")
 	}
diff --git a/azure/services/publicips/client.go b/azure/services/publicips/client.go
index b37d05529f2..396c66f92b3 100644
--- a/azure/services/publicips/client.go
+++ b/azure/services/publicips/client.go
@@ -37,7 +37,7 @@ type AzureClient struct {
 
 // NewClient creates a new public IP client from an authorizer.
 func NewClient(auth azure.Authorizer, apiCallTimeout time.Duration) (*AzureClient, error) {
-	opts, err := azure.ARMClientOptions(auth.CloudEnvironment())
+	opts, err := azure.ARMClientOptions(auth.CloudEnvironment(), auth.BaseURI())
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to create publicips client options")
 	}
diff --git a/azure/services/resourcehealth/client.go b/azure/services/resourcehealth/client.go
index a597dbfae97..189b507873c 100644
--- a/azure/services/resourcehealth/client.go
+++ b/azure/services/resourcehealth/client.go
@@ -38,7 +38,7 @@ type azureClient struct {
 
 // newClient creates a new resource health client from an authorizer.
 func newClient(auth azure.Authorizer) (*azureClient, error) {
-	opts, err := azure.ARMClientOptions(auth.CloudEnvironment())
+	opts, err := azure.ARMClientOptions(auth.CloudEnvironment(), auth.BaseURI())
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to create resourcehealth client options")
 	}
diff --git a/azure/services/resourceskus/client.go b/azure/services/resourceskus/client.go
index ed84116192f..c442f8c7ed5 100644
--- a/azure/services/resourceskus/client.go
+++ b/azure/services/resourceskus/client.go
@@ -40,7 +40,7 @@ var _ Client = &AzureClient{}
 
 // NewClient creates a new Resource SKUs client from an authorizer.
 func NewClient(auth azure.Authorizer) (*AzureClient, error) {
-	opts, err := azure.ARMClientOptions(auth.CloudEnvironment())
+	opts, err := azure.ARMClientOptions(auth.CloudEnvironment(), auth.BaseURI())
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to create resourceskus client options")
 	}
diff --git a/azure/services/roleassignments/client.go b/azure/services/roleassignments/client.go
index dd7643ec8d3..a35f3897966 100644
--- a/azure/services/roleassignments/client.go
+++ b/azure/services/roleassignments/client.go
@@ -34,7 +34,7 @@ type azureClient struct {
 
 // newClient creates a new role assignments client from an authorizer.
 func newClient(auth azure.Authorizer) (*azureClient, error) {
-	opts, err := azure.ARMClientOptions(auth.CloudEnvironment())
+	opts, err := azure.ARMClientOptions(auth.CloudEnvironment(), auth.BaseURI())
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to create roleassignments client options")
 	}
diff --git a/azure/services/routetables/client.go b/azure/services/routetables/client.go
index b535d6d0c66..b3bac74da59 100644
--- a/azure/services/routetables/client.go
+++ b/azure/services/routetables/client.go
@@ -37,7 +37,7 @@ type azureClient struct {
 
 // newClient creates a new route tables client from an authorizer.
 func newClient(auth azure.Authorizer, apiCallTimeout time.Duration) (*azureClient, error) {
-	opts, err := azure.ARMClientOptions(auth.CloudEnvironment())
+	opts, err := azure.ARMClientOptions(auth.CloudEnvironment(), auth.BaseURI())
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to create routetables client options")
 	}
diff --git a/azure/services/scalesets/client.go b/azure/services/scalesets/client.go
index d498a37240d..2b906f8d791 100644
--- a/azure/services/scalesets/client.go
+++ b/azure/services/scalesets/client.go
@@ -67,7 +67,7 @@ func NewClient(auth azure.Authorizer, apiCallTimeout time.Duration) (*AzureClien
 
 // newVirtualMachineScaleSetVMsClient creates a vmss VM client from an authorizer.
 func newVirtualMachineScaleSetVMsClient(auth azure.Authorizer) (*armcompute.VirtualMachineScaleSetVMsClient, error) {
-	opts, err := azure.ARMClientOptions(auth.CloudEnvironment())
+	opts, err := azure.ARMClientOptions(auth.CloudEnvironment(), auth.BaseURI())
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to create scalesetvms client options")
 	}
@@ -80,7 +80,7 @@ func newVirtualMachineScaleSetVMsClient(auth azure.Authorizer) (*armcompute.Virt
 
 // newVirtualMachineScaleSetsClient creates a vmss client from an authorizer.
 func newVirtualMachineScaleSetsClient(auth azure.Authorizer) (*armcompute.VirtualMachineScaleSetsClient, error) {
-	opts, err := azure.ARMClientOptions(auth.CloudEnvironment())
+	opts, err := azure.ARMClientOptions(auth.CloudEnvironment(), auth.BaseURI())
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to create scalesets client options")
 	}
diff --git a/azure/services/scalesetvms/client.go b/azure/services/scalesetvms/client.go
index b4aab4471e5..11ce72b63b7 100644
--- a/azure/services/scalesetvms/client.go
+++ b/azure/services/scalesetvms/client.go
@@ -46,7 +46,7 @@ var _ client = &azureClient{}
 
 // newClient creates a VMSS client from an authorizer.
 func newClient(auth azure.Authorizer, apiCallTimeout time.Duration) (*azureClient, error) {
-	opts, err := azure.ARMClientOptions(auth.CloudEnvironment())
+	opts, err := azure.ARMClientOptions(auth.CloudEnvironment(), auth.BaseURI())
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to create scalesetvms client options")
 	}
diff --git a/azure/services/securitygroups/client.go b/azure/services/securitygroups/client.go
index 607048860d0..c09d00e05be 100644
--- a/azure/services/securitygroups/client.go
+++ b/azure/services/securitygroups/client.go
@@ -39,7 +39,7 @@ type azureClient struct {
 
 // newClient creates a new security groups client from an authorizer.
 func newClient(auth azure.Authorizer, apiCallTimeout time.Duration) (*azureClient, error) {
-	opts, err := azure.ARMClientOptions(auth.CloudEnvironment())
+	opts, err := azure.ARMClientOptions(auth.CloudEnvironment(), auth.BaseURI())
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to create securitygroups client options")
 	}
@@ -84,7 +84,7 @@ func (ac *azureClient) CreateOrUpdateAsync(ctx context.Context, spec azure.Resou
 	}
 
 	// Create a new client that knows how to add the etag header.
-	clientOpts, err := azure.ARMClientOptions(ac.auth.CloudEnvironment(), extraPolicies...)
+	clientOpts, err := azure.ARMClientOptions(ac.auth.CloudEnvironment(), ac.auth.BaseURI(), extraPolicies...)
 	if err != nil {
 		return nil, nil, errors.Wrap(err, "failed to create securitygroups client options")
 	}
diff --git a/azure/services/tags/client.go b/azure/services/tags/client.go
index ebde41d8e79..c63b7937f13 100644
--- a/azure/services/tags/client.go
+++ b/azure/services/tags/client.go
@@ -41,7 +41,7 @@ var _ client = (*AzureClient)(nil)
 
 // NewClient creates a tags client from an authorizer.
 func NewClient(auth azure.Authorizer) (*AzureClient, error) {
-	opts, err := azure.ARMClientOptions(auth.CloudEnvironment())
+	opts, err := azure.ARMClientOptions(auth.CloudEnvironment(), auth.BaseURI())
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to create tags client options")
 	}
diff --git a/azure/services/virtualmachines/client.go b/azure/services/virtualmachines/client.go
index 1e2bbea08d4..e7b63c12404 100644
--- a/azure/services/virtualmachines/client.go
+++ b/azure/services/virtualmachines/client.go
@@ -49,7 +49,7 @@ var _ Client = &AzureClient{}
 
 // NewClient creates a VMs client from an authorizer.
 func NewClient(auth azure.Authorizer, apiCallTimeout time.Duration) (*AzureClient, error) {
-	opts, err := azure.ARMClientOptions(auth.CloudEnvironment())
+	opts, err := azure.ARMClientOptions(auth.CloudEnvironment(), auth.BaseURI())
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to create virtualmachines client options")
 	}
@@ -109,14 +109,21 @@ func (ac *AzureClient) CreateOrUpdateAsync(ctx context.Context, spec azure.Resou
 // request to Azure and if accepted without error, the func will return a Poller which can be used to track the ongoing
 // progress of the operation.
 func (ac *AzureClient) DeleteAsync(ctx context.Context, spec azure.ResourceSpecGetter, resumeToken string) (poller *runtime.Poller[armcompute.VirtualMachinesClientDeleteResponse], err error) {
-	ctx, _, done := tele.StartSpanWithLogger(ctx, "virtualmachines.AzureClient.Delete")
+	ctx, log, done := tele.StartSpanWithLogger(ctx, "virtualmachines.AzureClient.Delete")
 	defer done()
 
 	forceDelete := ptr.To(true)
 	opts := &armcompute.VirtualMachinesClientBeginDeleteOptions{ResumeToken: resumeToken, ForceDeletion: forceDelete}
 	poller, err = ac.virtualmachines.BeginDelete(ctx, spec.ResourceGroupName(), spec.ResourceName(), opts)
 	if err != nil {
-		return nil, err
+		if azure.BadRequest(err) {
+			log.Info("Failed to Begin VM Delete with Force Deletion, retrying without the force flag")
+			opts.ForceDeletion = ptr.To(false)
+			poller, err = ac.virtualmachines.BeginDelete(ctx, spec.ResourceGroupName(), spec.ResourceName(), opts)
+		}
+		if err != nil {
+			return nil, err
+		}
 	}
 
 	ctx, cancel := context.WithTimeout(ctx, ac.apiCallTimeout)
diff --git a/azure/services/vmextensions/client.go b/azure/services/vmextensions/client.go
index 187007ff784..09a58540811 100644
--- a/azure/services/vmextensions/client.go
+++ b/azure/services/vmextensions/client.go
@@ -37,7 +37,7 @@ type azureClient struct {
 
 // newClient creates a new vm extensions client from an authorizer.
 func newClient(auth azure.Authorizer, apiCallTimeout time.Duration) (*azureClient, error) {
-	opts, err := azure.ARMClientOptions(auth.CloudEnvironment())
+	opts, err := azure.ARMClientOptions(auth.CloudEnvironment(), auth.BaseURI())
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to create virtualmachineextensions client options")
 	}
diff --git a/azure/services/vnetpeerings/client.go b/azure/services/vnetpeerings/client.go
index 893c545d487..c556dbb4f23 100644
--- a/azure/services/vnetpeerings/client.go
+++ b/azure/services/vnetpeerings/client.go
@@ -37,7 +37,7 @@ type AzureClient struct {
 
 // NewClient creates a new virtual network peerings client from an authorizer.
 func NewClient(auth azure.Authorizer, apiCallTimeout time.Duration) (*AzureClient, error) {
-	opts, err := azure.ARMClientOptions(auth.CloudEnvironment())
+	opts, err := azure.ARMClientOptions(auth.CloudEnvironment(), auth.BaseURI())
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to create vnetpeerings client options")
 	}
diff --git a/config/crd/bases/infrastructure.cluster.x-k8s.io_azureclusters.yaml b/config/crd/bases/infrastructure.cluster.x-k8s.io_azureclusters.yaml
index 5692ad1ea9c..1293762205f 100644
--- a/config/crd/bases/infrastructure.cluster.x-k8s.io_azureclusters.yaml
+++ b/config/crd/bases/infrastructure.cluster.x-k8s.io_azureclusters.yaml
@@ -84,6 +84,12 @@ spec:
                   AdditionalTags is an optional set of tags to add to Azure resources managed by the Azure provider, in addition to the
                   ones added by default.
                 type: object
+              armEndpoint:
+                description: |-
+                  ARMEndpoint specifies a URL for the ARM Resource Manager endpoint.
+                  It may only be specified when the AzureEnvironment is set to AzureStackCloud,
+                  in which case it is required.
+                type: string
               azureEnvironment:
                 description: |-
                   AzureEnvironment is the name of the AzureCloud to be used.
@@ -92,6 +98,7 @@ spec:
                   - GermanCloud: "AzureGermanCloud"
                   - PublicCloud: "AzurePublicCloud"
                   - USGovernmentCloud: "AzureUSGovernmentCloud"
+                  - StackCloud: "HybridEnvironment"
 
                   Note that values other than the default must also be accompanied by corresponding changes to the
                   aso-controller-settings Secret to configure ASO to refer to the non-Public cloud. ASO currently does
diff --git a/config/crd/bases/infrastructure.cluster.x-k8s.io_azureclustertemplates.yaml b/config/crd/bases/infrastructure.cluster.x-k8s.io_azureclustertemplates.yaml
index b1cffc10c83..b0caaada440 100644
--- a/config/crd/bases/infrastructure.cluster.x-k8s.io_azureclustertemplates.yaml
+++ b/config/crd/bases/infrastructure.cluster.x-k8s.io_azureclustertemplates.yaml
@@ -57,6 +57,12 @@ spec:
                           AdditionalTags is an optional set of tags to add to Azure resources managed by the Azure provider, in addition to the
                           ones added by default.
                         type: object
+                      armEndpoint:
+                        description: |-
+                          ARMEndpoint specifies a URL for the ARM Resource Manager endpoint.
+                          It may only be specified when the AzureEnvironment is set to AzureStackCloud,
+                          in which case it is required.
+                        type: string
                       azureEnvironment:
                         description: |-
                           AzureEnvironment is the name of the AzureCloud to be used.
@@ -65,6 +71,7 @@ spec:
                           - GermanCloud: "AzureGermanCloud"
                           - PublicCloud: "AzurePublicCloud"
                           - USGovernmentCloud: "AzureUSGovernmentCloud"
+                          - StackCloud: "HybridEnvironment"
 
                           Note that values other than the default must also be accompanied by corresponding changes to the
                           aso-controller-settings Secret to configure ASO to refer to the non-Public cloud. ASO currently does
diff --git a/controllers/azuremachine_reconciler.go b/controllers/azuremachine_reconciler.go
index 544ccc02694..d0bb9ef8d7d 100644
--- a/controllers/azuremachine_reconciler.go
+++ b/controllers/azuremachine_reconciler.go
@@ -18,6 +18,7 @@ package controllers
 
 import (
 	"context"
+	"strings"
 
 	"github.com/pkg/errors"
 
@@ -101,10 +102,13 @@ func newAzureMachineService(machineScope *scope.MachineScope) (*azureMachineServ
 			virtualmachinesSvc,
 			roleAssignmentsSvc,
 			vmextensionsSvc,
-			tagsSvc,
 		},
 		skuCache: cache,
 	}
+	if !strings.EqualFold(machineScope.CloudEnvironment(), azure.StackCloudName) {
+		ams.services = append(ams.services, tagsSvc)
+	}
+
 	ams.Reconcile = ams.reconcile
 	ams.Pause = ams.pause
 	ams.Delete = ams.delete