diff --git a/airbyte-cdk/bulk/core/base/src/main/kotlin/io/airbyte/cdk/spec/SpecOperation.kt b/airbyte-cdk/bulk/core/base/src/main/kotlin/io/airbyte/cdk/spec/SpecOperation.kt
index 1d7bd1eae92a..a848571e850f 100644
--- a/airbyte-cdk/bulk/core/base/src/main/kotlin/io/airbyte/cdk/spec/SpecOperation.kt
+++ b/airbyte-cdk/bulk/core/base/src/main/kotlin/io/airbyte/cdk/spec/SpecOperation.kt
@@ -5,6 +5,7 @@ import io.airbyte.cdk.Operation
 import io.airbyte.cdk.command.ConfigurationJsonObjectSupplier
 import io.airbyte.cdk.output.OutputConsumer
 import io.airbyte.protocol.models.v0.ConnectorSpecification
+import io.micronaut.context.annotation.DefaultImplementation
 import io.micronaut.context.annotation.Requires
 import io.micronaut.context.annotation.Value
 import jakarta.inject.Singleton
@@ -15,13 +16,24 @@ import java.net.URI
 class SpecOperation(
     @Value("\${airbyte.connector.metadata.documentation-url}") val documentationUrl: String,
     val configJsonObjectSupplier: ConfigurationJsonObjectSupplier<*>,
+    val extendSpecification: SpecificationExtender,
     val outputConsumer: OutputConsumer,
 ) : Operation {
     override fun execute() {
-        outputConsumer.accept(
+        val spec =
             ConnectorSpecification()
                 .withDocumentationUrl(URI.create(documentationUrl))
-                .withConnectionSpecification(configJsonObjectSupplier.jsonSchema),
-        )
+                .withConnectionSpecification(configJsonObjectSupplier.jsonSchema)
+        outputConsumer.accept(extendSpecification(spec))
+    }
+}
+
+interface SpecificationExtender : (ConnectorSpecification) -> ConnectorSpecification
+
+@Singleton
+@DefaultImplementation
+class IdentitySpecificationExtender : SpecificationExtender {
+    override fun invoke(specification: ConnectorSpecification): ConnectorSpecification {
+        return specification
     }
 }
diff --git a/airbyte-cdk/bulk/core/load/src/main/kotlin/io/airbyte/cdk/spec/DestinationSpecification.kt b/airbyte-cdk/bulk/core/load/src/main/kotlin/io/airbyte/cdk/spec/DestinationSpecification.kt
new file mode 100644
index 000000000000..f513470dc829
--- /dev/null
+++ b/airbyte-cdk/bulk/core/load/src/main/kotlin/io/airbyte/cdk/spec/DestinationSpecification.kt
@@ -0,0 +1,28 @@
+/*
+ * Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.cdk.spec
+
+import io.airbyte.protocol.models.v0.ConnectorSpecification
+import io.airbyte.protocol.models.v0.DestinationSyncMode
+import io.micronaut.context.annotation.Replaces
+import io.micronaut.context.annotation.Requires
+import jakarta.inject.Singleton
+
+@Singleton
+@Replaces(IdentitySpecificationExtender::class)
+@Requires(env = ["destination"])
+class DestinationSpecificationExtender(private val spec: DestinationSpecification) :
+    SpecificationExtender {
+    override fun invoke(specification: ConnectorSpecification): ConnectorSpecification {
+        return specification
+            .withSupportedDestinationSyncModes(spec.supportedSyncModes)
+            .withSupportsIncremental(spec.supportsIncremental)
+    }
+}
+
+interface DestinationSpecification {
+    val supportedSyncModes: List<DestinationSyncMode>
+    val supportsIncremental: Boolean
+}