diff --git a/pom.xml b/pom.xml
index f33beb9ef..f504c79ca 100644
--- a/pom.xml
+++ b/pom.xml
@@ -41,6 +41,9 @@
     <maven.compiler.target>8</maven.compiler.target>
     <maven.compiler.release>8</maven.compiler.release>
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+    <aws.java.sdk.version>2.21.46</aws.java.sdk.version>
+    <aws.java.crt.version>0.29.24</aws.java.crt.version>
+    <com.amazonaws.version>1.12.441</com.amazonaws.version>
   </properties>
 
   <dependencyManagement>
@@ -48,7 +51,7 @@
       <dependency>
         <groupId>com.amazonaws</groupId>
         <artifactId>aws-java-sdk-bom</artifactId>
-        <version>1.12.441</version>
+        <version>${com.amazonaws.version}</version>
         <type>pom</type>
         <scope>import</scope>
       </dependency>
@@ -56,7 +59,7 @@
       <dependency>
         <groupId>software.amazon.awssdk</groupId>
         <artifactId>bom</artifactId>
-        <version>2.20.38</version>
+        <version>${aws.java.sdk.version}</version>
         <optional>true</optional>
         <type>pom</type>
         <scope>import</scope>
@@ -68,13 +71,11 @@
     <dependency>
       <groupId>software.amazon.awssdk</groupId>
       <artifactId>s3</artifactId>
-      <version>2.20.38</version>
     </dependency>
 
     <dependency>
       <groupId>software.amazon.awssdk</groupId>
       <artifactId>kms</artifactId>
-      <version>2.20.38</version>
       <optional>true</optional>
     </dependency>
 
@@ -82,7 +83,7 @@
     <dependency>
       <groupId>software.amazon.awssdk.crt</groupId>
       <artifactId>aws-crt</artifactId>
-      <version>0.29.24</version>
+      <version>${aws.java.crt.version}</version>
       <optional>true</optional>
     </dependency>
 
@@ -164,7 +165,6 @@
     <dependency>
       <groupId>software.amazon.awssdk</groupId>
       <artifactId>sts</artifactId>
-      <version>2.20.38</version>
       <optional>true</optional>
       <scope>test</scope>
     </dependency>
diff --git a/src/main/java/software/amazon/encryption/s3/S3AsyncEncryptionClient.java b/src/main/java/software/amazon/encryption/s3/S3AsyncEncryptionClient.java
index 2ed1ba3f5..f44c57545 100644
--- a/src/main/java/software/amazon/encryption/s3/S3AsyncEncryptionClient.java
+++ b/src/main/java/software/amazon/encryption/s3/S3AsyncEncryptionClient.java
@@ -30,6 +30,7 @@
 import software.amazon.awssdk.services.s3.model.PutObjectRequest;
 import software.amazon.awssdk.services.s3.model.PutObjectResponse;
 import software.amazon.awssdk.services.s3.model.S3Request;
+import software.amazon.awssdk.services.s3.multipart.MultipartConfiguration;
 import software.amazon.encryption.s3.internal.GetEncryptedObjectPipeline;
 import software.amazon.encryption.s3.internal.NoRetriesAsyncRequestBody;
 import software.amazon.encryption.s3.internal.PutEncryptedObjectPipeline;
@@ -71,6 +72,7 @@ public class S3AsyncEncryptionClient extends DelegatingS3AsyncClient {
     private final boolean _enableDelayedAuthenticationMode;
     private final boolean _enableMultipartPutObject;
     private final long _bufferSize;
+    private final boolean _clientMultipartEnabled;
 
     private S3AsyncEncryptionClient(Builder builder) {
         super(builder._wrappedClient);
@@ -81,6 +83,7 @@ private S3AsyncEncryptionClient(Builder builder) {
         _enableDelayedAuthenticationMode = builder._enableDelayedAuthenticationMode;
         _enableMultipartPutObject = builder._enableMultipartPutObject;
         _bufferSize = builder._bufferSize;
+        _clientMultipartEnabled = builder._multipartEnabled != null && builder._multipartEnabled;
     }
 
     /**
@@ -147,16 +150,19 @@ public CompletableFuture<PutObjectResponse> putObject(PutObjectRequest putObject
     }
 
     private CompletableFuture<PutObjectResponse> multipartPutObject(PutObjectRequest putObjectRequest, AsyncRequestBody requestBody) {
-        S3AsyncClient crtClient;
-        if (_wrappedClient instanceof S3CrtAsyncClient) {
+        S3AsyncClient mpuClient;
+        if (_wrappedClient instanceof S3CrtAsyncClient && !_clientMultipartEnabled) {
             // if the wrappedClient is a CRT, use it
-            crtClient = _wrappedClient;
-        } else {
-            // else create a default one
-            crtClient = S3AsyncClient.crtCreate();
+            mpuClient = _wrappedClient;
+        } else if (_clientMultipartEnabled) {
+            mpuClient = _wrappedClient;
+        }
+        else {
+            // else create a default CRT client
+            mpuClient = S3AsyncClient.crtCreate();
         }
         PutEncryptedObjectPipeline pipeline = PutEncryptedObjectPipeline.builder()
-                .s3AsyncClient(crtClient)
+                .s3AsyncClient(mpuClient)
                 .cryptoMaterialsManager(_cryptoMaterialsManager)
                 .secureRandom(_secureRandom)
                 .build();
@@ -291,8 +297,12 @@ public static class Builder implements S3AsyncClientBuilder {
         private S3Configuration _serviceConfiguration = null;
         private Boolean _accelerate = null;
         private Boolean _disableMultiRegionAccessPoints = null;
+        private Boolean _disableS3ExpressSessionAuth = null;
         private Boolean _forcePathStyle = null;
         private Boolean _useArnRegion = null;
+        private Boolean _crossRegionAccessEnabled = null;
+        private Boolean _multipartEnabled = null;
+        private MultipartConfiguration _multipartConfiguration = null;
 
         private Builder() {
         }
@@ -696,6 +706,12 @@ public Builder disableMultiRegionAccessPoints(Boolean disableMultiRegionAccessPo
             return this;
         }
 
+        @Override
+        public S3AsyncClientBuilder disableS3ExpressSessionAuth(Boolean disableS3ExpressSessionAuth) {
+            _disableS3ExpressSessionAuth = disableS3ExpressSessionAuth;
+            return this;
+        }
+
         /**
          * Forces this client to use path-style addressing for buckets.
          *
@@ -719,6 +735,24 @@ public Builder useArnRegion(Boolean useArnRegion) {
             return this;
         }
 
+        @Override
+        public Builder multipartEnabled(Boolean enabled) {
+            _multipartEnabled = enabled;
+            return this;
+        }
+
+        @Override
+        public S3AsyncClientBuilder multipartConfiguration(MultipartConfiguration multipartConfiguration) {
+            _multipartConfiguration = multipartConfiguration;
+            return this;
+        }
+
+        @Override
+        public Builder crossRegionAccessEnabled(Boolean crossRegionAccessEnabled) {
+            _crossRegionAccessEnabled = crossRegionAccessEnabled;
+            return this;
+        }
+
         /**
          * Validates and builds the S3AsyncEncryptionClient according
          * to the configuration options passed to the Builder object.
@@ -737,6 +771,12 @@ public S3AsyncEncryptionClient build() {
                 _bufferSize = DEFAULT_BUFFER_SIZE_BYTES;
             }
 
+            // The S3 Async Client has its own multipart setting,
+            // we enforce that the S3EC multipart PutObject setting is enabled as well.
+            if (_multipartEnabled != null && _multipartEnabled && !_enableMultipartPutObject) {
+                throw new S3EncryptionClientException("EnableMultipartPutObject MUST be enabled when the MultipartEnabled option is set to true.");
+            }
+
             if (_wrappedClient == null) {
                 _wrappedClient = S3AsyncClient.builder()
                         .credentialsProvider(_awsCredentialsProvider)
@@ -751,8 +791,12 @@ public S3AsyncEncryptionClient build() {
                         .serviceConfiguration(_serviceConfiguration)
                         .accelerate(_accelerate)
                         .disableMultiRegionAccessPoints(_disableMultiRegionAccessPoints)
+                        .disableS3ExpressSessionAuth(_disableS3ExpressSessionAuth)
                         .forcePathStyle(_forcePathStyle)
                         .useArnRegion(_useArnRegion)
+                        .crossRegionAccessEnabled(_crossRegionAccessEnabled)
+                        .multipartEnabled(_multipartEnabled)
+                        .multipartConfiguration(_multipartConfiguration)
                         .build();
             }
 
diff --git a/src/main/java/software/amazon/encryption/s3/S3EncryptionClient.java b/src/main/java/software/amazon/encryption/s3/S3EncryptionClient.java
index 1c3f6c27a..7f3f1f6be 100644
--- a/src/main/java/software/amazon/encryption/s3/S3EncryptionClient.java
+++ b/src/main/java/software/amazon/encryption/s3/S3EncryptionClient.java
@@ -544,8 +544,10 @@ public static class Builder implements S3BaseClientBuilder<Builder, S3Encryption
         private S3Configuration _serviceConfiguration = null;
         private Boolean _accelerate = null;
         private Boolean _disableMultiRegionAccessPoints = null;
+        private Boolean _disableS3ExpressSessionAuth = null;
         private Boolean _forcePathStyle = null;
         private Boolean _useArnRegion = null;
+        private Boolean _crossRegionAccessEnabled = null;
         private SdkHttpClient _httpClient = null;
         private SdkHttpClient.Builder _httpClientBuilder = null;
         private SdkAsyncHttpClient _asyncHttpClient = null;
@@ -918,6 +920,18 @@ public Builder disableMultiRegionAccessPoints(Boolean disableMultiRegionAccessPo
             return this;
         }
 
+        /**
+         * Disables this client's usage of Session Auth for S3Express buckets and reverts to using conventional SigV4 for
+         * those.
+         *
+         * @param disableS3ExpressSessionAuth
+         */
+        @Override
+        public Builder disableS3ExpressSessionAuth(Boolean disableS3ExpressSessionAuth) {
+            _disableS3ExpressSessionAuth = disableS3ExpressSessionAuth;
+            return this;
+        }
+
         /**
          * Forces this client to use path-style addressing for buckets.
          *
@@ -941,6 +955,17 @@ public Builder useArnRegion(Boolean useArnRegion) {
             return this;
         }
 
+        /**
+         * Enables cross-region bucket access for this client
+         *
+         * @param crossRegionAccessEnabled
+         */
+        @Override
+        public Builder crossRegionAccessEnabled(Boolean crossRegionAccessEnabled) {
+            _crossRegionAccessEnabled = crossRegionAccessEnabled;
+            return this;
+        }
+
         /**
          * Sets the {@link SdkHttpClient} that the SDK service client will use to make HTTP calls. This HTTP client may be
          * shared between multiple SDK service clients to share a common connection pool. To create a client you must use an
@@ -1052,6 +1077,8 @@ public S3EncryptionClient build() {
                         .useArnRegion(_useArnRegion)
                         .httpClient(_httpClient)
                         .httpClientBuilder(_httpClientBuilder)
+                        .disableS3ExpressSessionAuth(_disableS3ExpressSessionAuth)
+                        .crossRegionAccessEnabled(_crossRegionAccessEnabled)
                         .build();
             }
 
@@ -1070,6 +1097,9 @@ public S3EncryptionClient build() {
                         .useArnRegion(_useArnRegion)
                         .httpClient(_asyncHttpClient)
                         .httpClientBuilder(_asyncHttpClientBuilder)
+                        .disableS3ExpressSessionAuth(_disableS3ExpressSessionAuth)
+                        .crossRegionAccessEnabled(_crossRegionAccessEnabled)
+                        // TODO: Add MPU stuff here too
                         .build();
             }
 
diff --git a/src/test/java/software/amazon/encryption/s3/S3AsyncEncryptionClientTest.java b/src/test/java/software/amazon/encryption/s3/S3AsyncEncryptionClientTest.java
index f1abb7f64..e98b75c8e 100644
--- a/src/test/java/software/amazon/encryption/s3/S3AsyncEncryptionClientTest.java
+++ b/src/test/java/software/amazon/encryption/s3/S3AsyncEncryptionClientTest.java
@@ -20,6 +20,7 @@
 import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
 import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
 import software.amazon.awssdk.core.ResponseBytes;
+import software.amazon.awssdk.core.ResponseInputStream;
 import software.amazon.awssdk.core.async.AsyncRequestBody;
 import software.amazon.awssdk.core.async.AsyncResponseTransformer;
 import software.amazon.awssdk.core.client.config.ClientOverrideConfiguration;
@@ -39,6 +40,7 @@
 import software.amazon.awssdk.services.s3.model.ObjectIdentifier;
 import software.amazon.awssdk.services.s3.model.PutObjectResponse;
 import software.amazon.awssdk.services.s3.model.S3Exception;
+import software.amazon.awssdk.services.s3.multipart.MultipartConfiguration;
 import software.amazon.encryption.s3.materials.KmsKeyring;
 import software.amazon.encryption.s3.utils.BoundedInputStream;
 import software.amazon.encryption.s3.utils.S3EncryptionClientTestResources;
@@ -52,7 +54,9 @@
 import java.security.Provider;
 import java.security.Security;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.ExecutorService;
@@ -63,6 +67,7 @@
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
+import static software.amazon.encryption.s3.S3EncryptionClient.withAdditionalConfiguration;
 import static software.amazon.encryption.s3.utils.S3EncryptionClientTestResources.ALTERNATE_KMS_KEY;
 import static software.amazon.encryption.s3.utils.S3EncryptionClientTestResources.BUCKET;
 import static software.amazon.encryption.s3.utils.S3EncryptionClientTestResources.KMS_KEY_ID;
@@ -74,12 +79,15 @@
 public class S3AsyncEncryptionClientTest {
 
     private static SecretKey AES_KEY;
+    private static Provider PROVIDER;
 
     @BeforeAll
     public static void setUp() throws NoSuchAlgorithmException {
         KeyGenerator keyGen = KeyGenerator.getInstance("AES");
         keyGen.init(256);
         AES_KEY = keyGen.generateKey();
+        Security.addProvider(new BouncyCastleProvider());
+        PROVIDER = Security.getProvider("BC");
     }
 
     @Test
@@ -755,4 +763,50 @@ public void tinyBufferTest() throws IOException {
         exec.shutdown();
     }
 
+    @Test
+    public void wrappedClientMultipartUpload() throws IOException {
+        final String objectKey = appendTestSuffix("multipart-put-object-async-wrapped-client");
+
+        final long fileSizeLimit = 1024 * 1024 * 100;
+        final InputStream inputStream = new BoundedInputStream(fileSizeLimit);
+        final InputStream objectStreamForResult = new BoundedInputStream(fileSizeLimit);
+
+        S3AsyncClient v3Client = S3AsyncEncryptionClient.builder()
+            .kmsKeyId(KMS_KEY_ID)
+            .enableDelayedAuthenticationMode(true)
+            .cryptoProvider(PROVIDER)
+            // .enableMultipartPutObject(true)
+            // .multipartEnabled(true)
+            .multipartConfiguration(
+                MultipartConfiguration.builder()
+                    .thresholdInBytes((long)100000000)
+                    .minimumPartSizeInBytes((long)500000)
+                    .build()
+            )
+            .build();
+
+        Map<String, String> encryptionContext = new HashMap<>();
+        encryptionContext.put("user-metadata-key", "user-metadata-value-v3-to-v3");
+
+        ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
+
+        CompletableFuture<PutObjectResponse> futurePut = v3Client.putObject(builder -> builder
+                .bucket(BUCKET)
+                .overrideConfiguration(withAdditionalConfiguration(encryptionContext))
+                .key(objectKey), AsyncRequestBody.fromInputStream(inputStream, fileSizeLimit, singleThreadExecutor));
+        futurePut.join();
+        singleThreadExecutor.shutdown();
+
+        // Asserts
+        CompletableFuture<ResponseInputStream<GetObjectResponse>> getFuture = v3Client.getObject(builder -> builder
+                .bucket(BUCKET)
+                .overrideConfiguration(S3EncryptionClient.withAdditionalConfiguration(encryptionContext))
+                .key(objectKey), AsyncResponseTransformer.toBlockingInputStream());
+        ResponseInputStream<GetObjectResponse> output = getFuture.join();
+
+        assertTrue(IOUtils.contentEquals(objectStreamForResult, output));
+
+        deleteObject(BUCKET, objectKey, v3Client);
+        v3Client.close();
+    }
 }