diff --git a/airbyte-config/specs/src/main/java/io/airbyte/config/specs/SeedConnectorSpecGenerator.java b/airbyte-config/specs/src/main/java/io/airbyte/config/specs/SeedConnectorSpecGenerator.java
index 723464f2f308..c68ea8d1f99c 100644
--- a/airbyte-config/specs/src/main/java/io/airbyte/config/specs/SeedConnectorSpecGenerator.java
+++ b/airbyte-config/specs/src/main/java/io/airbyte/config/specs/SeedConnectorSpecGenerator.java
@@ -14,9 +14,11 @@
 import io.airbyte.commons.yaml.Yamls;
 import io.airbyte.config.DockerImageSpec;
 import io.airbyte.config.EnvConfigs;
+import io.airbyte.config.Spec;
 import io.airbyte.protocol.models.ConnectorSpecification;
 import java.io.IOException;
 import java.nio.file.Path;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
@@ -27,21 +29,17 @@
 import org.slf4j.LoggerFactory;
 
 /**
- * This script is responsible for ensuring that up-to-date {@link ConnectorSpecification}s for every
- * connector definition in the seed are stored in a corresponding resource file, for the purpose of
- * seeding the specs into the config database on server startup. See
- * ./airbyte-config/specs/readme.md for more details on how this class is run and how it fits into
- * the project.
+ * This script is responsible for ensuring that up-to-date {@link ConnectorSpecification}s for every connector definition in the seed are stored in a
+ * corresponding resource file, for the purpose of seeding the specs into the config database on server startup. See ./airbyte-config/specs/readme.md
+ * for more details on how this class is run and how it fits into the project.
  * <p>
- * Specs are stored in a separate file from the definitions in an effort to keep the definitions
- * yaml files human-readable and easily-editable, as specs can be rather large.
+ * Specs are stored in a separate file from the definitions in an effort to keep the definitions yaml files human-readable and easily-editable, as
+ * specs can be rather large.
  * <p>
- * Specs are fetched from the GCS spec cache bucket, so if any specs are missing from the bucket
- * then this will fail. Note that this script only pulls specs from the bucket cache; it never
- * pushes specs to the bucket. Since this script runs at build time, the decision was to depend on
- * the bucket cache rather than running a docker container to fetch the spec during the build which
- * could be slow and unwieldy. If there is a failure, check the bucket cache and figure out how to
- * get the correct spec in there.
+ * Specs are fetched from the GCS spec cache bucket, so if any specs are missing from the bucket then this will fail. Note that this script only pulls
+ * specs from the bucket cache; it never pushes specs to the bucket. Since this script runs at build time, the decision was to depend on the bucket
+ * cache rather than running a docker container to fetch the spec during the build which could be slow and unwieldy. If there is a failure, check the
+ * bucket cache and figure out how to get the correct spec in there.
  */
 @SuppressWarnings("PMD.SignatureDeclareThrowsException")
 public class SeedConnectorSpecGenerator {
@@ -50,6 +48,8 @@ public class SeedConnectorSpecGenerator {
   private static final String DOCKER_IMAGE_TAG_FIELD = "dockerImageTag";
   private static final String DOCKER_IMAGE_FIELD = "dockerImage";
   private static final String SPEC_FIELD = "spec";
+  private static final String SPECS_FIELD = "specs";
+  private static final String SPEC_TYPE_FIELD = "spec_type";
   private static final String SPEC_BUCKET_NAME = new EnvConfigs().getSpecCacheBucket();
 
   private static final Logger LOGGER = LoggerFactory.getLogger(SeedConnectorSpecGenerator.class);
@@ -95,19 +95,32 @@ private JsonNode yamlToJson(final Path root, final String fileName) {
     return Yamls.deserialize(yamlString);
   }
 
+  final List<Spec> getSpecVariants(final JsonNode specs) {
+    if (specs == null) {
+      return Collections.emptyList();
+    }
+    return MoreIterators.toList(specs.elements())
+        .stream()
+        .map(json -> new Spec()
+            .withSpecType(Jsons.object(json.get(SPEC_TYPE_FIELD), Spec.SpecType.class))
+            .withSpec(Jsons.object(json.get(SPEC_FIELD), ConnectorSpecification.class)))
+        .toList();
+  }
+
   @VisibleForTesting
   final List<DockerImageSpec> fetchUpdatedSeedSpecs(final JsonNode seedDefinitions, final JsonNode currentSeedSpecs) {
     final List<String> seedDefinitionsDockerImages = MoreIterators.toList(seedDefinitions.elements())
         .stream()
         .map(json -> String.format("%s:%s", json.get(DOCKER_REPOSITORY_FIELD).asText(), json.get(DOCKER_IMAGE_TAG_FIELD).asText()))
-        .collect(Collectors.toList());
+        .toList();
 
     final Map<String, DockerImageSpec> currentSeedImageToSpec = MoreIterators.toList(currentSeedSpecs.elements())
         .stream()
         .collect(Collectors.toMap(
             json -> json.get(DOCKER_IMAGE_FIELD).asText(),
             json -> new DockerImageSpec().withDockerImage(json.get(DOCKER_IMAGE_FIELD).asText())
-                .withSpec(Jsons.object(json.get(SPEC_FIELD), ConnectorSpecification.class))));
+                .withSpec(Jsons.object(json.get(SPEC_FIELD), ConnectorSpecification.class))
+                .withSpecs(getSpecVariants(json.get(SPECS_FIELD)))));
 
     return seedDefinitionsDockerImages
         .stream()
diff --git a/airbyte-config/specs/src/test/java/io/airbyte/config/specs/SeedConnectorSpecGeneratorTest.java b/airbyte-config/specs/src/test/java/io/airbyte/config/specs/SeedConnectorSpecGeneratorTest.java
index 5750a152487c..6b3facba568e 100644
--- a/airbyte-config/specs/src/test/java/io/airbyte/config/specs/SeedConnectorSpecGeneratorTest.java
+++ b/airbyte-config/specs/src/test/java/io/airbyte/config/specs/SeedConnectorSpecGeneratorTest.java
@@ -15,6 +15,8 @@
 import com.google.common.collect.ImmutableMap;
 import io.airbyte.commons.json.Jsons;
 import io.airbyte.config.DockerImageSpec;
+import io.airbyte.config.Spec;
+import io.airbyte.config.Spec.SpecType;
 import io.airbyte.config.StandardDestinationDefinition;
 import io.airbyte.protocol.models.ConnectorSpecification;
 import java.util.Arrays;
@@ -138,8 +140,20 @@ void testNoFetchIsPerformedIfAllSpecsUpToDate() {
         .withDockerImageTag(DOCKER_TAG1)
         .withName(CONNECTOR_NAME1)
         .withDocumentationUrl(DOCUMENTATION_URL);
-    final ConnectorSpecification spec = new ConnectorSpecification().withConnectionSpecification(Jsons.jsonNode(ImmutableMap.of("foo", "bar")));
-    final DockerImageSpec dockerImageSpec = new DockerImageSpec().withDockerImage(DOCKER_REPOSITORY1 + ":" + DOCKER_TAG1).withSpec(spec);
+    final ConnectorSpecification defaultSpec = new ConnectorSpecification()
+        .withConnectionSpecification(Jsons.jsonNode(ImmutableMap.of("foo", "bar")));
+    final Spec cloudSpec = new Spec()
+        .withSpecType(SpecType.CLOUD)
+        .withSpec(new ConnectorSpecification()
+            .withConnectionSpecification(Jsons.jsonNode(ImmutableMap.of("foo", "bar", "mode", "cloud"))));
+    final Spec ossSpec = new Spec()
+        .withSpecType(SpecType.CLOUD)
+        .withSpec(new ConnectorSpecification()
+            .withConnectionSpecification(Jsons.jsonNode(ImmutableMap.of("foo", "bar", "mode", "oss"))));
+
+    final DockerImageSpec dockerImageSpec = new DockerImageSpec()
+        .withDockerImage(DOCKER_REPOSITORY1 + ":" + DOCKER_TAG1).withSpec(defaultSpec)
+        .withSpecs(List.of(cloudSpec, ossSpec));
 
     final JsonNode seedDefinitions = Jsons.jsonNode(List.of(sourceDefinition));
     final JsonNode seedSpecs = Jsons.jsonNode(List.of(dockerImageSpec));
@@ -151,4 +165,19 @@ void testNoFetchIsPerformedIfAllSpecsUpToDate() {
     verify(bucketSpecFetcherMock, never()).attemptFetch(any());
   }
 
+  @Test
+  void testGetSpecVariants() {
+    final Spec cloudSpec = new Spec()
+        .withSpecType(SpecType.CLOUD)
+        .withSpec(new ConnectorSpecification()
+            .withConnectionSpecification(Jsons.jsonNode(ImmutableMap.of("foo", "bar", "mode", "cloud"))));
+    final Spec ossSpec = new Spec()
+        .withSpecType(SpecType.CLOUD)
+        .withSpec(new ConnectorSpecification()
+            .withConnectionSpecification(Jsons.jsonNode(ImmutableMap.of("foo", "bar", "mode", "oss"))));
+
+    assertEquals(List.of(cloudSpec, ossSpec),
+        seedConnectorSpecGenerator.getSpecVariants(Jsons.jsonNode(List.of(cloudSpec, ossSpec))));
+  }
+
 }