diff --git a/project/Settings.scala b/project/Settings.scala
index e760da5..9b32bb9 100644
--- a/project/Settings.scala
+++ b/project/Settings.scala
@@ -15,7 +15,7 @@ object Settings extends Dependencies {
   val scala213 = "2.13.10"
   val scala3   = "3.2.2"
 
-  val nextVersion = "2.1.7"
+  val nextVersion = "2.2.1"
   val artifactVersion = {
     sys.env.get("LENSES_TAG_NAME") match {
       case Some(tag) => tag
diff --git a/secret-provider/src/main/scala/io/lenses/connect/secrets/config/AWSProviderConfig.scala b/secret-provider/src/main/scala/io/lenses/connect/secrets/config/AWSProviderConfig.scala
index 39ac45f..30d1924 100644
--- a/secret-provider/src/main/scala/io/lenses/connect/secrets/config/AWSProviderConfig.scala
+++ b/secret-provider/src/main/scala/io/lenses/connect/secrets/config/AWSProviderConfig.scala
@@ -21,6 +21,7 @@ object AWSProviderConfig {
   val AWS_SECRET_KEY:    String = "aws.secret.key"
   val AUTH_METHOD:       String = "aws.auth.method"
   val ENDPOINT_OVERRIDE: String = "aws.endpoint.override"
+  val AWS_CROSS_ACCOUNT_REGION:        String = "aws.cross.account.region"
 
   val config: ConfigDef = new ConfigDef()
     .define(
@@ -54,6 +55,13 @@ object AWSProviderConfig {
         | Default is 'credentials'
         |""".stripMargin,
     )
+    .define(
+      AWS_CROSS_ACCOUNT_REGION,
+      Type.STRING,
+      "",
+      Importance.MEDIUM,
+      "AWS region the Secrets manager is in when reading from alternate account",
+    )
     .define(
       WRITE_FILES,
       Type.BOOLEAN,
diff --git a/secret-provider/src/main/scala/io/lenses/connect/secrets/config/AWSProviderSettings.scala b/secret-provider/src/main/scala/io/lenses/connect/secrets/config/AWSProviderSettings.scala
index 63d4082..e984d3d 100644
--- a/secret-provider/src/main/scala/io/lenses/connect/secrets/config/AWSProviderSettings.scala
+++ b/secret-provider/src/main/scala/io/lenses/connect/secrets/config/AWSProviderSettings.scala
@@ -23,6 +23,7 @@ case class AWSProviderSettings(
   fileWriterOpts:   Option[FileWriterOptions],
   defaultTtl:       Option[Duration],
   endpointOverride: Option[String],
+  altRegion:        String
 )
 
 import io.lenses.connect.secrets.config.AbstractConfigExtensions._
@@ -39,6 +40,8 @@ object AWSProviderSettings {
     val authMode =
       getAuthenticationMethod(configs.getString(AWSProviderConfig.AUTH_METHOD))
 
+    val altRegion = configs.getString("aws.cross.account.region")
+
     if (authMode == AuthMode.CREDENTIALS) {
       if (accessKey.isEmpty)
         throw new ConnectException(
@@ -59,6 +62,7 @@ object AWSProviderSettings {
       defaultTtl =
         Option(configs.getLong(SECRET_DEFAULT_TTL).toLong).filterNot(_ == 0L).map(Duration.of(_, ChronoUnit.MILLIS)),
       endpointOverride,
+      altRegion      = altRegion
     )
   }
 }
diff --git a/secret-provider/src/main/scala/io/lenses/connect/secrets/providers/AWSHelper.scala b/secret-provider/src/main/scala/io/lenses/connect/secrets/providers/AWSHelper.scala
index fb1f64d..dc7edca 100644
--- a/secret-provider/src/main/scala/io/lenses/connect/secrets/providers/AWSHelper.scala
+++ b/secret-provider/src/main/scala/io/lenses/connect/secrets/providers/AWSHelper.scala
@@ -38,7 +38,9 @@ import scala.util.Try
 class AWSHelper(
   client:             SecretsManagerClient,
   defaultTtl:         Option[Duration],
-  fileWriterCreateFn: () => Option[FileWriter],
+  region:             String,
+  altRegion:          String,
+  fileWriterCreateFn: () => Option[FileWriter]
 )(
   implicit
   clock: Clock,
@@ -48,12 +50,26 @@ class AWSHelper(
   private val objectMapper = new ObjectMapper()
 
   // get the key value and ttl in the specified secret
-  override def lookup(secretId: String): Either[Throwable, ValueWithTtl[Map[String, String]]] =
+  override def lookup(secretId: String): Either[Throwable, ValueWithTtl[Map[String, String]]] = {
+    val secretName = getSecretName(secretId)
     for {
-      secretTtl         <- getTTL(secretId)
-      secretValue       <- getSecretValue(secretId)
+      secretTtl         <- getTTL(secretName)
+      secretValue       <- getSecretValue(secretName)
       parsedSecretValue <- parseSecretValue(secretValue)
     } yield ValueWithTtl(secretTtl, parsedSecretValue)
+  }
+
+  private def getSecretName(secretId: String): String = {
+    val hasAccount = secretId.indexOf("$")
+    if (hasAccount > -1) {
+      val secret_region = if (hasAccount > -1 && altRegion.length > 0) altRegion else region
+      val secret_array = secretId.split("\\$")
+      s"arn:aws:secretsmanager:${secret_region}:${secret_array(0)}:secret:${secret_array(1)}"
+    }
+    else {
+      secretId
+    }
+  }
 
   // determine the ttl for the secret
   def getTTL(
diff --git a/secret-provider/src/main/scala/io/lenses/connect/secrets/providers/AWSSecretProvider.scala b/secret-provider/src/main/scala/io/lenses/connect/secrets/providers/AWSSecretProvider.scala
index e6411fb..3059881 100644
--- a/secret-provider/src/main/scala/io/lenses/connect/secrets/providers/AWSSecretProvider.scala
+++ b/secret-provider/src/main/scala/io/lenses/connect/secrets/providers/AWSSecretProvider.scala
@@ -30,6 +30,8 @@ class AWSSecretProvider(testClient: Option[SecretsManagerClient]) extends Config
     val awsClient = testClient.getOrElse(createClient(settings))
     val helper = new AWSHelper(awsClient,
                                settings.defaultTtl,
+                               settings.region,
+                               settings.altRegion,
                                fileWriterCreateFn = () => settings.fileWriterOpts.map(_.createFileWriter()),
     )
     secretProvider = Some(