diff --git a/airbyte-commons-worker/src/main/kotlin/io/airbyte/workers/sync/OrchestratorConstants.kt b/airbyte-commons-worker/src/main/kotlin/io/airbyte/workers/sync/OrchestratorConstants.kt
index a27a9f3c9f0..eb90a4c3536 100644
--- a/airbyte-commons-worker/src/main/kotlin/io/airbyte/workers/sync/OrchestratorConstants.kt
+++ b/airbyte-commons-worker/src/main/kotlin/io/airbyte/workers/sync/OrchestratorConstants.kt
@@ -44,6 +44,7 @@ object OrchestratorConstants {
       // add variables defined in this file
       addAll(
         setOf(
+          EnvVar.FEATURE_FLAG_BASEURL.toString(),
           EnvVar.FEATURE_FLAG_CLIENT.toString(),
           EnvVar.FEATURE_FLAG_PATH.toString(),
           LOG_LEVEL,
diff --git a/airbyte-commons/src/main/kotlin/io/airbyte/commons/envvar/EnvVar.kt b/airbyte-commons/src/main/kotlin/io/airbyte/commons/envvar/EnvVar.kt
index e50246f9576..936992252cf 100644
--- a/airbyte-commons/src/main/kotlin/io/airbyte/commons/envvar/EnvVar.kt
+++ b/airbyte-commons/src/main/kotlin/io/airbyte/commons/envvar/EnvVar.kt
@@ -35,6 +35,7 @@ enum class EnvVar {
   DOCKER_HOST,
   DOCKER_NETWORK,
 
+  FEATURE_FLAG_BASEURL,
   FEATURE_FLAG_CLIENT,
   FEATURE_FLAG_PATH,
 
diff --git a/airbyte-container-orchestrator/src/main/resources/application.yml b/airbyte-container-orchestrator/src/main/resources/application.yml
index 0ba4a04c0b4..ba2d531e2e3 100644
--- a/airbyte-container-orchestrator/src/main/resources/application.yml
+++ b/airbyte-container-orchestrator/src/main/resources/application.yml
@@ -86,6 +86,7 @@ airbyte:
     client: ${FEATURE_FLAG_CLIENT:config}
     path: ${FEATURE_FLAG_PATH:/flags}
     api-key: ${LAUNCHDARKLY_KEY:}
+    base-url: ${FEATURE_FLAG_BASEURL:}
   internal-api:
     auth-header:
       name: ${AIRBYTE_API_AUTH_HEADER_NAME:}
diff --git a/airbyte-featureflag-server/Dockerfile b/airbyte-featureflag-server/Dockerfile
new file mode 100644
index 00000000000..44e9c7ce78e
--- /dev/null
+++ b/airbyte-featureflag-server/Dockerfile
@@ -0,0 +1,16 @@
+ARG JDK_IMAGE=airbyte/airbyte-base-java-image:3.2.3
+
+FROM scratch as builder
+WORKDIR /app
+ADD airbyte-app.tar /app
+
+FROM ${JDK_IMAGE}
+EXPOSE 8007 5005
+ENV APPLICATION airbyte-featureflag-server
+ENV VERSION ${VERSION}
+
+WORKDIR /app
+COPY --chown=airbyte:airbyte --from=builder /app /app
+USER airbyte:airbyte
+
+ENTRYPOINT ["/bin/bash", "-c", "airbyte-app/bin/${APPLICATION}"]
diff --git a/airbyte-featureflag-server/build.gradle.kts b/airbyte-featureflag-server/build.gradle.kts
new file mode 100644
index 00000000000..e0a4366e28a
--- /dev/null
+++ b/airbyte-featureflag-server/build.gradle.kts
@@ -0,0 +1,61 @@
+import java.util.Properties
+
+plugins {
+  id("io.airbyte.gradle.jvm.app")
+  id("io.airbyte.gradle.docker")
+}
+
+dependencies {
+  ksp(libs.bundles.micronaut.annotation.processor)
+  ksp(libs.v3.swagger.annotations)
+  ksp(platform(libs.micronaut.platform))
+
+  annotationProcessor(libs.bundles.micronaut.annotation.processor)
+  annotationProcessor(libs.micronaut.jaxrs.processor)
+  annotationProcessor(platform(libs.micronaut.platform))
+
+  compileOnly(libs.v3.swagger.annotations)
+  compileOnly(libs.micronaut.openapi.annotations)
+
+  implementation(platform(libs.micronaut.platform))
+  implementation(libs.bundles.micronaut)
+  implementation(libs.bundles.micronaut.kotlin)
+  implementation(libs.log4j.impl)
+  implementation(libs.micronaut.http)
+  implementation(libs.micronaut.jaxrs.server)
+  implementation(libs.micronaut.security)
+  implementation(libs.v3.swagger.annotations)
+  implementation(libs.jackson.databind)
+  implementation(libs.jackson.dataformat)
+  implementation(libs.jackson.kotlin)
+
+  implementation(project(":oss:airbyte-commons"))
+
+  testImplementation(libs.bundles.micronaut.test)
+  testImplementation(libs.mockk)
+}
+
+val env = Properties().apply {
+  load(rootProject.file(".env.dev").inputStream())
+}
+
+airbyte {
+  application {
+    mainClass = "io.airbyte.featureflag.server.ApplicationKt"
+    defaultJvmArgs = listOf("-XX:+ExitOnOutOfMemoryError", "-XX:MaxRAMPercentage=75.0")
+    @Suppress("UNCHECKED_CAST")
+    localEnvVars.putAll(env.toMap() as Map<String, String>)
+    localEnvVars.putAll(
+      mapOf(
+        "AIRBYTE_ROLE" to (System.getenv("AIRBYTE_ROLE") ?: "undefined"),
+        "AIRBYTE_VERSION" to env["VERSION"].toString(),
+        "MICRONAUT_ENVIRONMENTS" to "control-plane",
+        "SERVICE_NAME" to project.name,
+        "TRACKING_STRATEGY" to env["TRACKING_STRATEGY"].toString(),
+      )
+    )
+  }
+  docker {
+    imageName = "featureflag-server"
+  }
+}
diff --git a/airbyte-featureflag-server/src/main/kotlin/io/airbyte/featureflag/server/Application.kt b/airbyte-featureflag-server/src/main/kotlin/io/airbyte/featureflag/server/Application.kt
new file mode 100644
index 00000000000..18c9bc853ae
--- /dev/null
+++ b/airbyte-featureflag-server/src/main/kotlin/io/airbyte/featureflag/server/Application.kt
@@ -0,0 +1,7 @@
+package io.airbyte.featureflag.server
+
+import io.micronaut.runtime.Micronaut.run
+
+fun main(args: Array<String>) {
+  run(*args)
+}
diff --git a/airbyte-featureflag-server/src/main/kotlin/io/airbyte/featureflag/server/FeatureFlagApi.kt b/airbyte-featureflag-server/src/main/kotlin/io/airbyte/featureflag/server/FeatureFlagApi.kt
new file mode 100644
index 00000000000..e0728861580
--- /dev/null
+++ b/airbyte-featureflag-server/src/main/kotlin/io/airbyte/featureflag/server/FeatureFlagApi.kt
@@ -0,0 +1,150 @@
+package io.airbyte.featureflag.server
+
+import io.airbyte.featureflag.server.model.Context
+import io.airbyte.featureflag.server.model.FeatureFlag
+import io.airbyte.featureflag.server.model.Rule
+import io.micronaut.http.HttpStatus
+import io.micronaut.http.annotation.Body
+import io.micronaut.http.annotation.Consumes
+import io.micronaut.http.annotation.Controller
+import io.micronaut.http.annotation.PathVariable
+import io.micronaut.http.annotation.QueryValue
+import io.micronaut.scheduling.TaskExecutors
+import io.micronaut.scheduling.annotation.ExecuteOn
+import io.swagger.v3.oas.annotations.media.Content
+import io.swagger.v3.oas.annotations.media.Schema
+import io.swagger.v3.oas.annotations.parameters.RequestBody
+import io.swagger.v3.oas.annotations.responses.ApiResponse
+import io.swagger.v3.oas.annotations.responses.ApiResponses
+import jakarta.ws.rs.DELETE
+import jakarta.ws.rs.GET
+import jakarta.ws.rs.POST
+import jakarta.ws.rs.PUT
+import jakarta.ws.rs.Path
+
+@Controller("/api/v1/feature-flags")
+@ExecuteOn(TaskExecutors.IO)
+class FeatureFlagApi(private val ffs: FeatureFlagService) {
+  @DELETE
+  @Path("/{key}")
+  @ApiResponses(
+    value = [
+      ApiResponse(
+        responseCode = "200",
+        description = "The flag evaluation",
+      ),
+    ],
+  )
+  fun delete(
+    @PathVariable key: String,
+  ) {
+    ffs.delete(key)
+  }
+
+  @GET
+  @Path("/{key}")
+  @ApiResponses(
+    value = [
+      ApiResponse(
+        responseCode = "200",
+        description = "The FeatureFlag",
+        content = [Content(schema = Schema(implementation = FeatureFlag::class))],
+      ),
+    ],
+  )
+  fun get(
+    @PathVariable key: String,
+  ): FeatureFlag {
+    return ffs.get(key) ?: throw KnownException(HttpStatus.NOT_FOUND, "$key not found")
+  }
+
+  @PUT
+  @Path("/")
+  @Consumes("application/json")
+  @ApiResponses(
+    value = [
+      ApiResponse(
+        responseCode = "200",
+        description = "The FeatureFlag",
+        content = [Content(schema = Schema(implementation = FeatureFlag::class))],
+      ),
+    ],
+  )
+  fun put(
+    @RequestBody(content = [Content(schema = Schema(implementation = FeatureFlag::class))]) @Body request: FeatureFlag,
+  ): FeatureFlag {
+    return ffs.put(request)
+  }
+
+  @GET
+  @Path("/{key}/evaluate")
+  @ApiResponses(
+    value = [
+      ApiResponse(
+        responseCode = "200",
+        description = "The flag evaluation",
+      ),
+    ],
+  )
+  fun evaluate(
+    @PathVariable key: String,
+    @QueryValue("kind") kind: List<String> = emptyList(),
+    @QueryValue("value") value: List<String> = emptyList(),
+  ): String {
+    val context = kind.zip(value).toMap()
+    return ffs.eval(key, context) ?: throw KnownException(HttpStatus.NOT_FOUND, "$key not found")
+  }
+
+  @DELETE
+  @Path("/{key}/rules")
+  @ApiResponses(
+    value = [
+      ApiResponse(
+        responseCode = "200",
+        description = "The flag evaluation",
+      ),
+    ],
+  )
+  fun deleteRule(
+    @PathVariable key: String,
+    @RequestBody(content = [Content(schema = Schema(implementation = Context::class))]) @Body context: Context,
+  ): FeatureFlag {
+    return ffs.removeRule(key, context)
+  }
+
+  @POST
+  @Path("/{key}/rules")
+  @ApiResponses(
+    value = [
+      ApiResponse(
+        responseCode = "200",
+        description = "The flag evaluation",
+      ),
+    ],
+  )
+  fun postRule(
+    @PathVariable key: String,
+    @RequestBody(content = [Content(schema = Schema(implementation = Rule::class))]) @Body rule: Rule,
+  ): FeatureFlag {
+    return ffs.addRule(key, rule)
+  }
+
+  @PUT
+  @Path("/{key}/rules")
+  @ApiResponses(
+    value = [
+      ApiResponse(
+        responseCode = "200",
+        description = "The flag evaluation",
+      ),
+    ],
+  )
+  fun putRule(
+    @PathVariable key: String,
+    @RequestBody(content = [Content(schema = Schema(implementation = Rule::class))]) @Body rule: Rule,
+  ): FeatureFlag {
+    return ffs.updateRule(key, rule)
+  }
+}
+
+class KnownException(val httpCode: HttpStatus, message: String) : Throwable(message)
diff --git a/airbyte-featureflag-server/src/main/kotlin/io/airbyte/featureflag/server/FeatureFlagService.kt b/airbyte-featureflag-server/src/main/kotlin/io/airbyte/featureflag/server/FeatureFlagService.kt
new file mode 100644
index 00000000000..e627892f5ab
--- /dev/null
+++ b/airbyte-featureflag-server/src/main/kotlin/io/airbyte/featureflag/server/FeatureFlagService.kt
@@ -0,0 +1,136 @@
+package io.airbyte.featureflag.server
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.fasterxml.jackson.dataformat.yaml.YAMLFactory
+import com.fasterxml.jackson.module.kotlin.registerKotlinModule
+import io.airbyte.featureflag.server.model.Context
+import io.airbyte.featureflag.server.model.FeatureFlag
+import io.airbyte.featureflag.server.model.Rule
+import io.micronaut.context.annotation.Property
+import jakarta.inject.Singleton
+import java.nio.file.Path
+import kotlin.io.path.exists
+import kotlin.io.path.isRegularFile
+
+// This is open for testing, creating an interface might be the way to go
+@Singleton
+open class FeatureFlagService(
+  @Property(name = "airbyte.feature-flag.path") configPath: Path?,
+) {
+  private val flags = mutableMapOf<String, MutableFeatureFlag>()
+
+  init {
+    configPath?.also { path ->
+      if (path.exists() && path.isRegularFile()) {
+        val yamlMapper = ObjectMapper(YAMLFactory()).registerKotlinModule()
+        val config = yamlMapper.readValue(path.toFile(), ConfigFileFlags::class.java)
+        config.flags.forEach {
+          put(it.toFeatureFlag())
+        }
+      }
+    }
+  }
+
+  open fun delete(key: String) {
+    flags.remove(key)
+  }
+
+  open fun eval(
+    key: String,
+    context: Map<String, String>,
+  ): String? {
+    val flag = flags[key] ?: return null
+    for (rule in flag.rules) {
+      if (rule.context.matches(context)) {
+        return rule.value
+      }
+    }
+    return flag.default
+  }
+
+  open fun get(key: String): FeatureFlag? {
+    return flags[key]?.toFeatureFlag()
+  }
+
+  open fun addRule(
+    key: String,
+    rule: Rule,
+  ): FeatureFlag {
+    val flag = flags[key] ?: throw Exception("$key not found")
+
+    if (flag.rules.any { it.context == rule.context }) {
+      throw Exception("$key already has a rule for context ${rule.context}")
+    }
+    flag.rules.add(rule.toMutableRule())
+    return flag.toFeatureFlag()
+  }
+
+  open fun updateRule(
+    key: String,
+    rule: Rule,
+  ): FeatureFlag {
+    val flag = flags[key] ?: throw Exception("$key not found")
+    flag.rules
+      .find { it.context == rule.context }
+      ?.apply { value = rule.value }
+      ?: throw Exception("$key does not have a rule for context ${rule.context}")
+    return flag.toFeatureFlag()
+  }
+
+  open fun removeRule(
+    key: String,
+    context: Context,
+  ): FeatureFlag {
+    val flag = flags[key] ?: throw Exception("$key not found")
+    flag.rules.removeIf { it.context == context }
+    return flag.toFeatureFlag()
+  }
+
+  open fun put(flag: FeatureFlag): FeatureFlag {
+    flags[flag.key] = flag.toMutableFeatureFlag()
+    return get(flag.key) ?: throw Exception("Failed to put flag $flag")
+  }
+
+  open fun put(
+    key: String,
+    default: String,
+    rules: List<Rule> = emptyList(),
+  ): FeatureFlag {
+    val flag = FeatureFlag(key = key, default = default, rules = rules)
+    return put(flag)
+  }
+
+  private fun Context.matches(env: Map<String, String>): Boolean = env[kind] == value
+
+  private fun MutableFeatureFlag.toFeatureFlag(): FeatureFlag = FeatureFlag(key = key, default = default, rules = rules.map { it.toRule() }.toList())
+
+  private fun FeatureFlag.toMutableFeatureFlag(): MutableFeatureFlag =
+    MutableFeatureFlag(key = key, default = default, rules = rules.map { it.toMutableRule() }.toMutableList())
+
+  private fun MutableRule.toRule(): Rule = Rule(context = context, value = value)
+
+  private fun Rule.toMutableRule(): MutableRule = MutableRule(context = context, value = value)
+
+  private data class MutableFeatureFlag(val key: String, val default: String, val rules: MutableList<MutableRule>)
+
+  private data class MutableRule(val context: Context, var value: String)
+
+  private data class ConfigFileFlags(val flags: List<ConfigFileFlag>)
+
+  private data class ConfigFileFlag(
+    val name: String,
+    val serve: String,
+    val context: List<ConfigFileFlagContext>? = null,
+  ) {
+    fun toFeatureFlag(): FeatureFlag {
+      val rules = context?.flatMap { context -> context.include.map { Rule(Context(kind = context.type, value = it), value = context.serve) } }
+      return FeatureFlag(key = name, default = serve, rules = rules ?: emptyList())
+    }
+  }
+
+  private data class ConfigFileFlagContext(
+    val type: String,
+    val serve: String,
+    val include: List<String> = listOf(),
+  )
+}
diff --git a/airbyte-featureflag-server/src/main/kotlin/io/airbyte/featureflag/server/KnownExceptionHandler.kt b/airbyte-featureflag-server/src/main/kotlin/io/airbyte/featureflag/server/KnownExceptionHandler.kt
new file mode 100644
index 00000000000..c27a3a352c6
--- /dev/null
+++ b/airbyte-featureflag-server/src/main/kotlin/io/airbyte/featureflag/server/KnownExceptionHandler.kt
@@ -0,0 +1,18 @@
+package io.airbyte.featureflag.server
+
+import io.micronaut.http.HttpRequest
+import io.micronaut.http.HttpResponse
+import io.micronaut.http.annotation.Produces
+import io.micronaut.http.server.exceptions.ExceptionHandler
+import jakarta.inject.Singleton
+
+@Produces
+@Singleton
+class KnownExceptionHandler : ExceptionHandler<KnownException, HttpResponse<String>> {
+  override fun handle(
+    request: HttpRequest<Any>,
+    exception: KnownException,
+  ): HttpResponse<String> {
+    return HttpResponse.status<Any>(exception.httpCode).body(exception.message)
+  }
+}
diff --git a/airbyte-featureflag-server/src/main/kotlin/io/airbyte/featureflag/server/model/Model.kt b/airbyte-featureflag-server/src/main/kotlin/io/airbyte/featureflag/server/model/Model.kt
new file mode 100644
index 00000000000..79f31c7c8e6
--- /dev/null
+++ b/airbyte-featureflag-server/src/main/kotlin/io/airbyte/featureflag/server/model/Model.kt
@@ -0,0 +1,11 @@
+package io.airbyte.featureflag.server.model
+
+data class Context(val kind: String, val value: String)
+
+data class Rule(val context: Context, val value: String)
+
+data class FeatureFlag(
+  val key: String,
+  val default: String,
+  val rules: List<Rule> = listOf(),
+)
diff --git a/airbyte-featureflag-server/src/main/resources/application.yml b/airbyte-featureflag-server/src/main/resources/application.yml
new file mode 100644
index 00000000000..d275e9ab59a
--- /dev/null
+++ b/airbyte-featureflag-server/src/main/resources/application.yml
@@ -0,0 +1,57 @@
+micronaut:
+  application:
+    name: airbyte-featureflag-server
+  executors:
+    io:
+      type: fixed
+      n-threads: 10
+    metrics:
+      enabled: false
+  security:
+    enabled: false
+  server:
+    port: 8007
+    idle-timeout: ${HTTP_IDLE_TIMEOUT:5m}
+    netty:
+      access-logger:
+        enabled: ${HTTP_ACCESS_LOG_ENABLED:true}
+
+airbyte:
+  feature-flag:
+    path: ${FEATURE_FLAG_PATH:/flags}
+
+endpoints:
+  beans:
+    enabled: true
+    sensitive: false
+  env:
+    enabled: true
+    sensitive: false
+  health:
+    enabled: true
+    jdbc:
+      enabled: false
+    sensitive: false
+  info:
+    enabled: true
+    sensitive: true
+  loggers:
+    enabled: true
+    sensitive: false
+  metrics:
+    enabled: ${MICROMETER_METRICS_ENABLED:false}
+    sensitive: false
+  refresh:
+    enabled: false
+    sensitive: true
+  routes:
+    enabled: true
+    sensitive: false
+  threaddump:
+    enabled: true
+    sensitive: true
+
+jackson:
+  mapper:
+    ACCEPT_CASE_INSENSITIVE_ENUMS: true
+  serialization-inclusion: always
diff --git a/airbyte-featureflag-server/src/test/kotlin/io/airbyte/featureflag/server/FeatureFlagApiTest.kt b/airbyte-featureflag-server/src/test/kotlin/io/airbyte/featureflag/server/FeatureFlagApiTest.kt
new file mode 100644
index 00000000000..98ae35c8b4b
--- /dev/null
+++ b/airbyte-featureflag-server/src/test/kotlin/io/airbyte/featureflag/server/FeatureFlagApiTest.kt
@@ -0,0 +1,130 @@
+package io.airbyte.featureflag.server
+
+import io.airbyte.commons.json.Jsons
+import io.airbyte.featureflag.server.model.Context
+import io.airbyte.featureflag.server.model.FeatureFlag
+import io.airbyte.featureflag.server.model.Rule
+import io.micronaut.context.annotation.Replaces
+import io.micronaut.context.env.Environment
+import io.micronaut.http.HttpRequest
+import io.micronaut.http.HttpResponse
+import io.micronaut.http.client.HttpClient
+import io.micronaut.http.client.annotation.Client
+import io.micronaut.http.client.exceptions.HttpClientResponseException
+import io.micronaut.test.annotation.MockBean
+import io.micronaut.test.extensions.junit5.annotation.MicronautTest
+import io.mockk.every
+import io.mockk.mockk
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.assertThrows
+
+@MicronautTest(environments = [Environment.TEST])
+class FeatureFlagApiTest(
+  @Client("/") val client: HttpClient,
+) {
+  private val ffs = mockk<FeatureFlagService>()
+
+  @MockBean(FeatureFlagService::class)
+  @Replaces(FeatureFlagService::class)
+  fun featureFlagService(): FeatureFlagService = ffs
+
+  @Test
+  fun `test evaluation with a context`() {
+    val key = "test-eval"
+    val evalResult = "eval-result"
+    every { ffs.eval(key, mapOf("c" to "c1", "w" to "w1")) } returns evalResult
+
+    val response = call<String>(HttpRequest.GET("/api/v1/feature-flags/$key/evaluate?kind=c&value=c1&kind=w&value=w1"))
+    assertEquals(200, response.status.code)
+    assertEquals(evalResult, response.body.get())
+  }
+
+  @Test
+  fun `test delete flag`() {
+    val key = "test-delete"
+    every { ffs.delete(key) } returns Unit
+
+    val response = call<String>(HttpRequest.DELETE("/api/v1/feature-flags/$key"))
+    assertEquals(200, response.status.code)
+  }
+
+  @Test
+  fun `test evaluation with an empty context`() {
+    val key = "test-eval"
+    val evalResult = "eval-result"
+    every { ffs.eval(key, mapOf()) } returns evalResult
+
+    val response = call<String>(HttpRequest.GET("/api/v1/feature-flags/$key/evaluate"))
+    assertEquals(200, response.status.code)
+    assertEquals(evalResult, response.body.get())
+  }
+
+  @Test
+  fun `test get returns the response`() {
+    val flag = FeatureFlag(key = "my-flag", default = "default")
+    every { ffs.get("my-flag") } returns flag
+
+    val response = call<FeatureFlag>(HttpRequest.GET("/api/v1/feature-flags/my-flag"))
+    assertEquals(200, response.status.code)
+    assertEquals(flag, response.body.get())
+  }
+
+  @Test
+  fun `test get not found`() {
+    every { ffs.get(any()) } returns null
+
+    val response = callError(HttpRequest.GET("/api/v1/feature-flags/not-found-flag"))
+    assertEquals(404, response.status.code)
+  }
+
+  @Test
+  fun `test put`() {
+    val flag = FeatureFlag(key = "flag", default = "default", rules = listOf(Rule(context = Context(kind = "c", value = "c1"), value = "c1v")))
+    every { ffs.put(flag) }.returns(flag)
+
+    val response = call<FeatureFlag>(HttpRequest.PUT("/api/v1/feature-flags/", Jsons.serialize(flag)))
+    assertEquals(200, response.status.code)
+    assertEquals(flag, response.body.get())
+  }
+
+  @Test
+  fun `test rule delete`() {
+    val context = Context(kind = "r", value = "r1")
+    val flag = FeatureFlag(key = "rule-test", default = "some default")
+    every { ffs.removeRule(flag.key, context) }.returns(flag)
+
+    val response = call<FeatureFlag>(HttpRequest.DELETE("/api/v1/feature-flags/${flag.key}/rules", Jsons.serialize(context)))
+    assertEquals(200, response.status.code)
+    assertEquals(flag, response.body.get())
+  }
+
+  @Test
+  fun `test rule post`() {
+    val rule = Rule(context = Context(kind = "r", value = "r1"), value = "r1-value")
+    val flag = FeatureFlag(key = "rule-test", default = "some default")
+    every { ffs.addRule(flag.key, rule) }.returns(flag)
+
+    val response = call<FeatureFlag>(HttpRequest.POST("/api/v1/feature-flags/${flag.key}/rules", Jsons.serialize(rule)))
+    assertEquals(200, response.status.code)
+    assertEquals(flag, response.body.get())
+  }
+
+  @Test
+  fun `test rule put`() {
+    val rule = Rule(context = Context(kind = "r", value = "r1"), value = "r1-value")
+    val flag = FeatureFlag(key = "rule-test", default = "some default")
+    every { ffs.updateRule(flag.key, rule) }.returns(flag)
+
+    val response = call<FeatureFlag>(HttpRequest.PUT("/api/v1/feature-flags/${flag.key}/rules", Jsons.serialize(rule)))
+    assertEquals(200, response.status.code)
+    assertEquals(flag, response.body.get())
+  }
+
+  private inline fun <reified T> call(request: HttpRequest<Any>): HttpResponse<T> = client.toBlocking().exchange(request, T::class.java)
+
+  private fun callError(request: HttpRequest<Any>): HttpResponse<*> {
+    val exception = assertThrows<HttpClientResponseException> { call<String>(request) }
+    return exception.response
+  }
+}
diff --git a/airbyte-featureflag-server/src/test/kotlin/io/airbyte/featureflag/server/FeatureFlagServiceTest.kt b/airbyte-featureflag-server/src/test/kotlin/io/airbyte/featureflag/server/FeatureFlagServiceTest.kt
new file mode 100644
index 00000000000..40e2c3876f3
--- /dev/null
+++ b/airbyte-featureflag-server/src/test/kotlin/io/airbyte/featureflag/server/FeatureFlagServiceTest.kt
@@ -0,0 +1,180 @@
+package io.airbyte.featureflag.server
+
+import io.airbyte.featureflag.server.model.Context
+import io.airbyte.featureflag.server.model.FeatureFlag
+import io.airbyte.featureflag.server.model.Rule
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.BeforeEach
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.assertThrows
+import java.nio.file.Path
+
+class FeatureFlagServiceTest {
+  private lateinit var ffs: FeatureFlagService
+
+  @BeforeEach
+  fun setup() {
+    ffs = FeatureFlagService(null)
+  }
+
+  @Test
+  fun `create and read a feature flag`() {
+    val key = "sample-flag"
+    val defaultValue = "something"
+
+    val expected = FeatureFlag(key = key, default = defaultValue)
+    val putResult = ffs.put(key = key, default = defaultValue)
+    assertEquals(expected, putResult)
+
+    val getResult = ffs.get(key)
+    assertEquals(expected, getResult)
+  }
+
+  @Test
+  fun `deleting a flag`() {
+    val flag1 = FeatureFlag(key = "flag1", default = "flag1 default")
+    val flag2 = FeatureFlag(key = "flag2", default = "flag2 default")
+
+    ffs.put(flag1)
+    assertEquals(flag1, ffs.get(flag1.key))
+    ffs.put(flag2)
+    ffs.delete(flag1.key)
+    assertEquals(null, ffs.get(flag1.key))
+    ffs.delete(flag1.key)
+    assertEquals(null, ffs.get(flag1.key))
+    assertEquals(flag2, ffs.get(flag2.key))
+  }
+
+  @Test
+  fun `eval is a find first evaluation of the list of rules`() {
+    val defaultValue = "default-value"
+    val w1Value = "value-w1"
+    val c1Value = "value-c1"
+    val c2Value = "value-c2"
+
+    val evalFlag =
+      FeatureFlag(
+        key = "eval-test",
+        default = defaultValue,
+        rules =
+          listOf(
+            Rule(context = Context(kind = "w", value = "w1"), value = w1Value),
+            Rule(context = Context(kind = "c", value = "c1"), value = c1Value),
+            Rule(context = Context(kind = "c", value = "c2"), value = c2Value),
+          ),
+      )
+    ffs.put(evalFlag)
+
+    assertEquals(defaultValue, ffs.eval(evalFlag.key, mapOf()))
+    assertEquals(defaultValue, ffs.eval(evalFlag.key, mapOf("other" to "val")))
+    assertEquals(w1Value, ffs.eval(evalFlag.key, mapOf("w" to "w1", "c" to "c1")))
+    assertEquals(c1Value, ffs.eval(evalFlag.key, mapOf("w" to "w2", "c" to "c1")))
+    assertEquals(c2Value, ffs.eval(evalFlag.key, mapOf("w" to "w2", "c" to "c2")))
+    assertEquals(defaultValue, ffs.eval(evalFlag.key, mapOf("w" to "w2", "c" to "c3")))
+    assertEquals(null, ffs.eval("no such flag", mapOf()))
+  }
+
+  @Test
+  fun `key not found returns null`() {
+    assertEquals(null, ffs.get("not found"))
+  }
+
+  @Test
+  fun `modifying a flag`() {
+    val controlFlag =
+      FeatureFlag(key = "control", default = "false", rules = listOf(Rule(Context(kind = "workspace", value = "admin"), value = "true")))
+    ffs.put(controlFlag)
+
+    val initialFlag = FeatureFlag(key = "iterating", default = "default")
+    ffs.put(initialFlag)
+
+    val rule1 = Rule(context = Context(kind = "s", value = "s1"), value = "s1value")
+    val expectedAdd1 = initialFlag.copy(rules = listOf(*initialFlag.rules.toTypedArray(), rule1))
+    val add1Result = ffs.addRule(key = initialFlag.key, rule = rule1)
+    assertEquals(expectedAdd1, add1Result)
+    val getAdd1Result = ffs.get(initialFlag.key)
+    assertEquals(expectedAdd1, getAdd1Result)
+
+    val rule2 = Rule(context = Context(kind = "s", value = "s2"), value = "s2value")
+    val expectedAdd2 = expectedAdd1.copy(rules = listOf(*expectedAdd1.rules.toTypedArray(), rule2))
+    ffs.addRule(key = initialFlag.key, rule = rule2)
+    val getAdd2Result = ffs.get(initialFlag.key)
+    assertEquals(expectedAdd2, getAdd2Result)
+
+    assertThrows<Exception> { ffs.addRule(key = initialFlag.key, rule = rule2) }
+    val getAfterFailedAdd = ffs.get(initialFlag.key)
+    assertEquals(expectedAdd2, getAfterFailedAdd)
+
+    val rule3 = rule2.copy(value = "new s2value")
+    val expectedAdd3 = expectedAdd1.copy(rules = listOf(*expectedAdd1.rules.toTypedArray(), rule3))
+    val updateRule3Result = ffs.updateRule(key = initialFlag.key, rule = rule3)
+    assertEquals(expectedAdd3, updateRule3Result)
+    val getAdd3Result = ffs.get(initialFlag.key)
+    assertEquals(expectedAdd3, getAdd3Result)
+
+    assertThrows<Exception> { ffs.updateRule(key = initialFlag.key, rule = rule1.copy(context = Context(kind = "s", value = "s0"))) }
+    val getAfterFailedUpdate = ffs.get(initialFlag.key)
+    assertEquals(expectedAdd3, getAfterFailedUpdate)
+
+    val removeResult = ffs.removeRule(key = initialFlag.key, context = rule1.context)
+    val expectedRemove = initialFlag.copy(rules = listOf(rule3))
+    assertEquals(expectedRemove, removeResult)
+    val getAfterRemove = ffs.get(initialFlag.key)
+    assertEquals(expectedRemove, getAfterRemove)
+
+    val removeResult2 = ffs.removeRule(key = initialFlag.key, context = rule1.context)
+    assertEquals(expectedRemove, removeResult2)
+
+    val controlCheck = ffs.get(key = controlFlag.key)
+    assertEquals(controlFlag, controlCheck)
+  }
+
+  @Test
+  fun `put overrides the key`() {
+    val controlFlag =
+      FeatureFlag(key = "control", default = "false", rules = listOf(Rule(Context(kind = "workspace", value = "admin"), value = "true")))
+
+    val key = "override-test"
+    val initialDefault = "my first default"
+    val rules =
+      listOf(
+        Rule(context = Context(kind = "workspace", value = "1"), value = "will get overridden"),
+        Rule(context = Context(kind = "connection", value = "1"), value = "will get overridden too"),
+      )
+
+    ffs.put(controlFlag)
+
+    val firstPut = ffs.put(key = key, default = initialDefault, rules = rules)
+    assertEquals(FeatureFlag(key = key, default = initialDefault, rules = rules), firstPut)
+
+    val overrideDefault = "new default"
+    val secondPut = ffs.put(key = key, default = overrideDefault)
+    assertEquals(FeatureFlag(key = key, default = overrideDefault), secondPut)
+
+    val getResult = ffs.get(key = key)
+    assertEquals(FeatureFlag(key = key, default = overrideDefault), getResult)
+
+    val controlCheck = ffs.get(key = controlFlag.key)
+    assertEquals(controlFlag, controlCheck)
+  }
+
+  @Test
+  fun `verify file loading`() {
+    val workspace = "workspace"
+    val ffs = FeatureFlagService(Path.of("src", "test", "resources", "flags.yml"))
+
+    assertEquals("true", ffs.get("test-true")?.default)
+    assertEquals("false", ffs.get("test-false")?.default)
+    assertEquals("example", ffs.get("test-string")?.default)
+    assertEquals("1234", ffs.get("test-int")?.default)
+    assertEquals("example", ffs.eval("test-string", mapOf(workspace to "any")))
+    assertEquals("context", ffs.eval("test-string", mapOf(workspace to "00000000-aaaa-0000-aaaa-000000000000")))
+    assertEquals("true", ffs.eval("test-context-boolean", mapOf()))
+    assertEquals("false", ffs.eval("test-context-boolean", mapOf(workspace to "00000000-aaaa-0000-aaaa-000000000000")))
+    assertEquals("cccc", ffs.eval("test-context-string", mapOf("connection" to "00000000-dddd-0000-dddd-000000000000")))
+    assertEquals("aaaa", ffs.eval("test-context-string", mapOf(workspace to "00000000-aaaa-0000-aaaa-000000000000")))
+    assertEquals("aaaa", ffs.eval("test-context-string", mapOf(workspace to "00000000-dddd-0000-dddd-000000000000")))
+    // this is returning bbbb instead of aaaa because we return the first match
+    assertEquals("bbbb", ffs.eval("test-context-string", mapOf(workspace to "00000000-bbbb-0000-bbbb-000000000000")))
+  }
+}
diff --git a/airbyte-featureflag-server/src/test/resources/flags.yml b/airbyte-featureflag-server/src/test/resources/flags.yml
new file mode 100644
index 00000000000..2f70097176b
--- /dev/null
+++ b/airbyte-featureflag-server/src/test/resources/flags.yml
@@ -0,0 +1,41 @@
+flags:
+  - name: test-true
+    serve: true
+  - name: test-false
+    serve: false
+  - name: test-string
+    serve: "example"
+    context:
+      - type: "workspace"
+        include:
+          - "00000000-aaaa-0000-aaaa-000000000000"
+        serve: "context"
+
+  - name: test-context-boolean
+    serve: true
+    context:
+      - type: "workspace"
+        include:
+          - "00000000-aaaa-0000-aaaa-000000000000"
+        serve: false
+
+  - name: test-context-string
+    serve: "all"
+    context:
+      - type: "workspace"
+        include:
+          - "00000000-aaaa-0000-aaaa-000000000000"
+          - "00000000-dddd-0000-dddd-000000000000"
+        serve: "aaaa"
+      - type: "workspace"
+        include:
+          - "00000000-bbbb-0000-bbbb-000000000000"
+          - "00000000-dddd-0000-dddd-000000000000"
+        serve: "bbbb"
+      - type: "connection"
+        include:
+          - "00000000-dddd-0000-dddd-000000000000"
+        serve: "cccc"
+
+  - name: test-int
+    serve: 1234
diff --git a/airbyte-featureflag/build.gradle.kts b/airbyte-featureflag/build.gradle.kts
index 297b424d8fc..c0431a1ee36 100644
--- a/airbyte-featureflag/build.gradle.kts
+++ b/airbyte-featureflag/build.gradle.kts
@@ -13,6 +13,8 @@ dependencies {
   implementation(libs.jackson.databind)
   implementation(libs.jackson.dataformat)
   implementation(libs.jackson.kotlin)
+  implementation(libs.okhttp)
+  implementation(project(":oss:airbyte-commons"))
 
   kspTest(platform(libs.micronaut.platform))
   kspTest(libs.bundles.micronaut.test.annotation.processor)
diff --git a/airbyte-featureflag/src/main/kotlin/Client.kt b/airbyte-featureflag/src/main/kotlin/Client.kt
index db6215a195f..a296e94471a 100644
--- a/airbyte-featureflag/src/main/kotlin/Client.kt
+++ b/airbyte-featureflag/src/main/kotlin/Client.kt
@@ -15,7 +15,10 @@ import io.micronaut.context.annotation.Property
 import io.micronaut.context.annotation.Requires
 import io.micronaut.context.annotation.Secondary
 import jakarta.inject.Inject
+import jakarta.inject.Named
 import jakarta.inject.Singleton
+import okhttp3.OkHttpClient
+import okhttp3.Request
 import org.slf4j.LoggerFactory
 import java.lang.Thread.MIN_PRIORITY
 import java.nio.file.Path
@@ -72,6 +75,7 @@ internal const val CONFIG_FF_CLIENT = "airbyte.feature-flag.client"
 
 /** If [CONFIG_FF_CLIENT] equals this value, return the [LaunchDarklyClient], otherwise the [ConfigFileClient]. */
 internal const val CONFIG_FF_CLIENT_VAL_LAUNCHDARKLY = "launchdarkly"
+internal const val CONFIG_FF_CLIENT_VAL_FFS = "ffs"
 
 /** Config key to provide the api-key as required by the [LaunchDarklyClient]. */
 internal const val CONFIG_FF_APIKEY = "airbyte.feature-flag.api-key"
@@ -79,6 +83,9 @@ internal const val CONFIG_FF_APIKEY = "airbyte.feature-flag.api-key"
 /** Config key to provide the location of the flags config file used by the [ConfigFileClient]. */
 internal const val CONFIG_FF_PATH = "airbyte.feature-flag.path"
 
+/** Config key to provide the base URL used by the [FeatureFlagServiceClient] */
+internal const val CONFIG_FF_BASEURL = "airbyte.feature-flag.base-url"
+
 /**
  * Config file based feature-flag client.
  *
@@ -89,6 +96,7 @@ internal const val CONFIG_FF_PATH = "airbyte.feature-flag.path"
  * If the [config] is provided, it will be watched for changes and the internal representation of the [config] will be updated to match.
  */
 @Singleton
+@Requires(property = CONFIG_FF_CLIENT, notEquals = CONFIG_FF_CLIENT_VAL_FFS)
 @Requires(property = CONFIG_FF_CLIENT, notEquals = CONFIG_FF_CLIENT_VAL_LAUNCHDARKLY)
 class ConfigFileClient(
   @Property(name = CONFIG_FF_PATH) config: Path?,
@@ -146,8 +154,7 @@ class ConfigFileClient(
 }
 
 /**
- * LaunchDarkly based feature-flag client. Feature-flags are derived from an external source (the LDClient).
- * Also supports flags defined via environment-variables via the [EnvVar] class.
+ * FeatureFlagService based feature-flag client.
  *
  * @param [client] the Launch-Darkly client for interfacing with Launch-Darkly.
  */
@@ -179,6 +186,55 @@ class LaunchDarklyClient(private val client: LDClient) : FeatureFlagClient {
   }
 }
 
+@Singleton
+@Requires(property = CONFIG_FF_CLIENT, value = CONFIG_FF_CLIENT_VAL_FFS)
+class FeatureFlagServiceClient(
+  @Named("ffsHttpClient") private val httpClient: OkHttpClient,
+  @Property(name = CONFIG_FF_BASEURL) private val baseUrl: String,
+) : FeatureFlagClient {
+  private val basePath = "/api/v1/feature-flags"
+
+  override fun boolVariation(
+    flag: Flag<Boolean>,
+    context: Context,
+  ): Boolean {
+    return callFeatureFlagService(flag.key, context)?.toBoolean() ?: flag.default
+  }
+
+  override fun stringVariation(
+    flag: Flag<String>,
+    context: Context,
+  ): String {
+    return callFeatureFlagService(flag.key, context) ?: flag.default
+  }
+
+  override fun intVariation(
+    flag: Flag<Int>,
+    context: Context,
+  ): Int {
+    return callFeatureFlagService(flag.key, context)?.toInt() ?: flag.default
+  }
+
+  private fun callFeatureFlagService(
+    key: String,
+    context: Context,
+  ): String? {
+    val request =
+      Request.Builder()
+        .url("$baseUrl$basePath/$key/evaluate?${context.toQueryParams()}")
+        .build()
+    return httpClient.newCall(request).execute().use {
+      if (it.code == 200) it.body?.string() else null
+    }
+  }
+
+  private fun Context.toQueryParams(): String =
+    when (this) {
+      is Multi -> contexts.joinToString("&") { it.toQueryParams() }
+      else -> "kind=$kind&value=$key"
+    }
+}
+
 /**
  * Test feature-flag client. Only to be used in test scenarios.
  *
diff --git a/airbyte-featureflag/src/main/kotlin/config/Factory.kt b/airbyte-featureflag/src/main/kotlin/config/Factory.kt
index aa07503ba58..b4d055e14da 100644
--- a/airbyte-featureflag/src/main/kotlin/config/Factory.kt
+++ b/airbyte-featureflag/src/main/kotlin/config/Factory.kt
@@ -7,11 +7,14 @@ package io.airbyte.featureflag.config
 import com.launchdarkly.sdk.server.LDClient
 import io.airbyte.featureflag.CONFIG_FF_APIKEY
 import io.airbyte.featureflag.CONFIG_FF_CLIENT
+import io.airbyte.featureflag.CONFIG_FF_CLIENT_VAL_FFS
 import io.airbyte.featureflag.CONFIG_FF_CLIENT_VAL_LAUNCHDARKLY
 import io.micronaut.context.annotation.Factory
 import io.micronaut.context.annotation.Property
 import io.micronaut.context.annotation.Requires
+import jakarta.inject.Named
 import jakarta.inject.Singleton
+import okhttp3.OkHttpClient
 
 @Factory
 class Factory {
@@ -20,4 +23,9 @@ class Factory {
   fun ldClient(
     @Property(name = CONFIG_FF_APIKEY) apiKey: String,
   ): LDClient = LDClient(apiKey)
+
+  @Singleton
+  @Requires(property = CONFIG_FF_CLIENT, value = CONFIG_FF_CLIENT_VAL_FFS)
+  @Named("ffsHttpClient")
+  fun ffsHttpClient(): OkHttpClient = OkHttpClient.Builder().build()
 }
diff --git a/airbyte-featureflag/src/main/kotlin/tests/TestFlagsSetter.kt b/airbyte-featureflag/src/main/kotlin/tests/TestFlagsSetter.kt
new file mode 100644
index 00000000000..4e533eb3878
--- /dev/null
+++ b/airbyte-featureflag/src/main/kotlin/tests/TestFlagsSetter.kt
@@ -0,0 +1,98 @@
+package io.airbyte.featureflag.tests
+
+import io.airbyte.commons.json.Jsons
+import io.airbyte.featureflag.Context
+import io.airbyte.featureflag.Flag
+import okhttp3.MediaType.Companion.toMediaType
+import okhttp3.OkHttpClient
+import okhttp3.Request
+import okhttp3.RequestBody.Companion.toRequestBody
+
+class TestFlagsSetter {
+  private val baseurl = "http://local.airbyte.dev/api/v1/feature-flags"
+  private val httpClient = OkHttpClient().newBuilder().build()
+
+  class FlagOverride<T>(
+    private val flag: Flag<T>,
+    context: Context,
+    value: T,
+    private val testFlags: TestFlagsSetter,
+  ) : AutoCloseable {
+    init {
+      testFlags.setFlag(flag, context, value)
+    }
+
+    override fun close() {
+      testFlags.deleteFlag(flag)
+    }
+  }
+
+  fun <T> withFlag(
+    flag: Flag<T>,
+    context: Context,
+    value: T,
+  ) = FlagOverride(flag, context, value, this)
+
+  fun <T> deleteFlag(flag: Flag<T>) {
+    httpClient.newCall(
+      Request.Builder()
+        .url("$baseurl/${flag.key}")
+        .delete()
+        .build(),
+    ).execute()
+  }
+
+  fun <T> setFlag(
+    flag: Flag<T>,
+    context: Context,
+    value: T,
+  ) {
+    val requestFlag =
+      ApiFeatureFlag(
+        key = flag.key,
+        default = flag.default.toString(),
+        rules =
+          listOf(
+            ApiRule(
+              context = ApiContext(kind = context.kind, value = context.key),
+              value = value.toString(),
+            ),
+          ),
+      )
+    httpClient.newCall(
+      Request.Builder()
+        .url(baseurl)
+        .put(Jsons.serialize(requestFlag).toRequestBody("application/json".toMediaType()))
+        .build(),
+    ).execute()
+  }
+
+  fun <T> getFlag(flag: Flag<T>) {
+    httpClient.newCall(
+      Request.Builder()
+        .url("$baseurl/${flag.key}")
+        .build(),
+    ).execute()
+  }
+
+  fun <T> evalFlag(
+    flag: Flag<T>,
+    context: Context,
+  ) {
+    httpClient.newCall(
+      Request.Builder()
+        .url("$baseurl/${flag.key}/evaluate?kind=${context.kind}&value=${context.key}")
+        .build(),
+    ).execute()
+  }
+
+  private data class ApiContext(val kind: String, val value: String)
+
+  private data class ApiRule(val context: ApiContext, val value: String)
+
+  private data class ApiFeatureFlag(
+    val key: String,
+    val default: String,
+    val rules: List<ApiRule> = listOf(),
+  )
+}
diff --git a/airbyte-featureflag/src/test/kotlin/ClientTest.kt b/airbyte-featureflag/src/test/kotlin/ClientTest.kt
index 87d036473ef..0f16279f632 100644
--- a/airbyte-featureflag/src/test/kotlin/ClientTest.kt
+++ b/airbyte-featureflag/src/test/kotlin/ClientTest.kt
@@ -18,6 +18,10 @@ import io.mockk.slot
 import io.mockk.verify
 import jakarta.inject.Inject
 import jakarta.inject.Singleton
+import okhttp3.Call
+import okhttp3.OkHttpClient
+import okhttp3.Response
+import okhttp3.ResponseBody
 import org.junit.jupiter.api.BeforeEach
 import org.junit.jupiter.api.Test
 import java.nio.file.Path
@@ -242,6 +246,102 @@ class ConfigFileClientTest {
   }
 }
 
+class FeatureFlagServiceClientTest {
+  val baseUrl = "http://test.featureflag.ab.com"
+
+  @Test
+  fun `verify api call`() {
+    val httpClient = mockk<OkHttpClient>(relaxed = true)
+    val client = FeatureFlagServiceClient(httpClient, baseUrl)
+    client.boolVariation(Temporary(key = "test-url-gen", default = true), Connection(UUID.randomUUID()))
+    verify {
+      httpClient.newCall(
+        match {
+          it.url.host == "test.featureflag.ab.com" &&
+            it.url.scheme == "http" &&
+            it.url.encodedPath.startsWith("/api/v1/feature-flags/test-url-gen/evaluate")
+        },
+      )
+    }
+  }
+
+  @Test
+  fun `verify payload generation`() {
+    val testFlag = Temporary(key = "test-feature-flag-client", default = "default")
+    val connectionId = UUID.randomUUID()
+    val workspaceId = UUID.randomUUID()
+
+    val httpClient =
+      mockk<OkHttpClient> {
+        every { newCall(match { it.url.query == "kind=connection&value=$connectionId" }) } returns mockResponse("connection-eval")
+        every { newCall(match { it.url.query == "kind=workspace&value=$workspaceId" }) } returns mockResponse("workspace-eval")
+        every {
+          newCall(match { it.url.query == "kind=connection&value=$connectionId&kind=workspace&value=$workspaceId" })
+        } returns mockResponse("multi-eval")
+      }
+    val client = FeatureFlagServiceClient(httpClient, baseUrl)
+
+    with(client) {
+      assertEquals("connection-eval", stringVariation(testFlag, Connection(connectionId)))
+      assertEquals("workspace-eval", stringVariation(testFlag, Workspace(workspaceId)))
+      assertEquals("multi-eval", stringVariation(testFlag, Multi(listOf(Connection(connectionId), Workspace(workspaceId)))))
+    }
+  }
+
+  @Test
+  fun `verify response conversion`() {
+    val testBooleanFlag = Temporary(key = "test-feature-flag-client-bool", default = false)
+    val testIntFlag = Temporary(key = "test-feature-flag-client-int", default = 7)
+    val testStringFlag = Temporary(key = "test-feature-flag-client-string", default = "fancy")
+    val connectionId = UUID.randomUUID()
+
+    val httpClient =
+      mockk<OkHttpClient> {
+        every { newCall(match { it.url.encodedPath.endsWith("/${testBooleanFlag.key}/evaluate") }) } returns mockResponse("true")
+        every { newCall(match { it.url.encodedPath.endsWith("/${testIntFlag.key}/evaluate") }) } returns mockResponse("777")
+        every { newCall(match { it.url.encodedPath.endsWith("/${testStringFlag.key}/evaluate") }) } returns mockResponse("airbyte")
+      }
+    val client = FeatureFlagServiceClient(httpClient, baseUrl)
+
+    with(client) {
+      assertEquals(true, boolVariation(testBooleanFlag, Connection(connectionId)))
+      assertEquals(777, intVariation(testIntFlag, Connection(connectionId)))
+      assertEquals("airbyte", stringVariation(testStringFlag, Connection(connectionId)))
+    }
+  }
+
+  @Test
+  fun `verify error handling`() {
+    val flag = Temporary(key = "error-flag", default = 42)
+    val httpClient =
+      mockk<OkHttpClient> {
+        every { newCall(match { it.url.encodedPath.endsWith("/${flag.key}/evaluate") }) } returns mockResponse("not found", statusCode = 404)
+      }
+    val client = FeatureFlagServiceClient(httpClient, baseUrl)
+
+    with(client) {
+      assertEquals(42, intVariation(flag, Connection(UUID.randomUUID())))
+    }
+  }
+
+  private fun mockResponse(
+    bodyString: String,
+    statusCode: Int = 200,
+  ): Call {
+    return mockk<Call> {
+      every { execute() } returns
+        mockk<Response> {
+          every { code } returns statusCode
+          every { body } returns
+            mockk<ResponseBody> {
+              every { string() } returns bodyString
+            }
+          every { close() } returns Unit
+        }
+    }
+  }
+}
+
 class LaunchDarklyClientTest {
   @Test
   fun `verify cloud functionality`() {
diff --git a/airbyte-server/src/main/resources/application.yml b/airbyte-server/src/main/resources/application.yml
index 3438dcd8eb9..24f4dc4e173 100644
--- a/airbyte-server/src/main/resources/application.yml
+++ b/airbyte-server/src/main/resources/application.yml
@@ -130,6 +130,7 @@ airbyte:
     client: ${FEATURE_FLAG_CLIENT:}
     path: ${FEATURE_FLAG_PATH:/flags}
     api-key: ${LAUNCHDARKLY_KEY:}
+    base-url: ${FEATURE_FLAG_BASEURL:}
   flyway:
     configs:
       initialization-timeout-ms: ${CONFIGS_DATABASE_INITIALIZATION_TIMEOUT_MS:60000}
diff --git a/airbyte-tests/build.gradle.kts b/airbyte-tests/build.gradle.kts
index 792ada4b48b..5c3193d358f 100644
--- a/airbyte-tests/build.gradle.kts
+++ b/airbyte-tests/build.gradle.kts
@@ -78,6 +78,7 @@ dependencies {
   implementation(project(":oss:airbyte-test-utils"))
   implementation(project(":oss:airbyte-commons-worker"))
   implementation(project(":oss:airbyte-container-orchestrator"))
+  implementation(project(":oss:airbyte-featureflag"))
 
   implementation(libs.bundles.kubernetes.client)
   implementation(libs.platform.testcontainers)
diff --git a/airbyte-workers/src/main/resources/application.yml b/airbyte-workers/src/main/resources/application.yml
index a9e717d09e6..53db0a14d5e 100644
--- a/airbyte-workers/src/main/resources/application.yml
+++ b/airbyte-workers/src/main/resources/application.yml
@@ -100,6 +100,7 @@ airbyte:
     client: ${FEATURE_FLAG_CLIENT:}
     path: ${FEATURE_FLAG_PATH:/flags}
     api-key: ${LAUNCHDARKLY_KEY:}
+    base-url: ${FEATURE_FLAG_BASEURL:}
   internal-api:
     auth-header:
       name: ${AIRBYTE_API_AUTH_HEADER_NAME:}
diff --git a/airbyte-workload-api-server/src/main/resources/application.yml b/airbyte-workload-api-server/src/main/resources/application.yml
index 258f6f00724..8294024134c 100644
--- a/airbyte-workload-api-server/src/main/resources/application.yml
+++ b/airbyte-workload-api-server/src/main/resources/application.yml
@@ -55,6 +55,7 @@ airbyte:
     client: ${FEATURE_FLAG_CLIENT:}
     path: ${FEATURE_FLAG_PATH:/flags}
     api-key: ${LAUNCHDARKLY_KEY:}
+    base-url: ${FEATURE_FLAG_BASEURL:}
 
 endpoints:
   beans:
diff --git a/airbyte-workload-launcher/src/main/resources/application.yml b/airbyte-workload-launcher/src/main/resources/application.yml
index 8813d35ba32..539dac027aa 100644
--- a/airbyte-workload-launcher/src/main/resources/application.yml
+++ b/airbyte-workload-launcher/src/main/resources/application.yml
@@ -37,6 +37,7 @@ airbyte:
     client: ${FEATURE_FLAG_CLIENT:}
     path: ${FEATURE_FLAG_PATH:/flags}
     api-key: ${LAUNCHDARKLY_KEY:}
+    base-url: ${FEATURE_FLAG_BASEURL:}
   kubernetes:
     client:
       call-timeout-sec: ${KUBERNETES_CLIENT_CALL_TIMEOUT_SECONDS:30}
diff --git a/charts/airbyte-featureflag-server/.gitignore b/charts/airbyte-featureflag-server/.gitignore
new file mode 100644
index 00000000000..88e91e8a8f3
--- /dev/null
+++ b/charts/airbyte-featureflag-server/.gitignore
@@ -0,0 +1,2 @@
+# Charts are downloaded at install time with `helm dep build`.
+charts
diff --git a/charts/airbyte-featureflag-server/Chart.yaml b/charts/airbyte-featureflag-server/Chart.yaml
new file mode 100644
index 00000000000..9e4171bb15a
--- /dev/null
+++ b/charts/airbyte-featureflag-server/Chart.yaml
@@ -0,0 +1,31 @@
+apiVersion: v2
+name: featureflag-server
+description: Helm chart to deploy the workload-api service
+
+# A chart can be either an 'application' or a 'library' chart.
+#
+# Application charts are a collection of templates that can be packaged into versioned archives
+# to be deployed.
+#
+# Library charts provide useful utilities or functions for the chart developer. They're included as
+# a dependency of application charts to inject those utilities and functions into the rendering
+# pipeline. Library charts do not define any templates and therefore cannot be deployed.
+type: application
+
+# This is the chart version. This version number should be incremented each time you make changes
+# to the chart and its templates, including the app version.
+# Versions are expected to follow Semantic Versioning (https://semver.org/)
+version: 0.399.0
+
+# This is the version number of the application being deployed. This version number should be
+# incremented each time you make changes to the application. Versions are not expected to
+# follow Semantic Versioning. They should reflect the version the application is using.
+# It is recommended to use it with quotes.
+appVersion: dev
+
+dependencies:
+  - name: common
+    repository: https://charts.bitnami.com/bitnami
+    tags:
+      - bitnami-common
+    version: 1.x.x
diff --git a/charts/airbyte-featureflag-server/README.md b/charts/airbyte-featureflag-server/README.md
new file mode 100644
index 00000000000..6fa0705921a
--- /dev/null
+++ b/charts/airbyte-featureflag-server/README.md
@@ -0,0 +1,85 @@
+# featureflag-server
+
+![Version: 0.67.17](https://img.shields.io/badge/Version-0.67.17-informational?style=flat-square) ![Type: application](https://img.shields.io/badge/Type-application-informational?style=flat-square) ![AppVersion: dev](https://img.shields.io/badge/AppVersion-dev-informational?style=flat-square)
+
+Helm chart to deploy the featureflag-server
+
+## Requirements
+
+| Repository | Name | Version |
+|------------|------|---------|
+| https://charts.bitnami.com/bitnami | common | 1.x.x |
+
+## Values
+
+| Key | Type | Default | Description |
+|-----|------|---------|-------------|
+| affinity | object | `{}` |  |
+| containerSecurityContext | object | `{}` |  |
+| debug.enabled | bool | `false` |  |
+| debug.remoteDebugPort | int | `5005` |  |
+| enabled | bool | `false` |  |
+| env_vars | object | `{}` |  |
+| extraContainers | list | `[]` |  |
+| extraEnv | list | `[]` |  |
+| extraInitContainers | list | `[]` |  |
+| extraLabels | object | `{}` |  |
+| extraSelectorLabels | object | `{}` |  |
+| extraVolumeMounts | list | `[]` |  |
+| extraVolumes | list | `[]` |  |
+| global.configMapName | string | `""` |  |
+| global.credVolumeOverride | string | `""` |  |
+| global.database.secretName | string | `""` |  |
+| global.database.secretValue | string | `""` |  |
+| global.deploymentMode | string | `"oss"` |  |
+| global.extraContainers | list | `[]` |  |
+| global.extraLabels | object | `{}` |  |
+| global.extraSelectorLabels | object | `{}` |  |
+| global.logs.accessKey.existingSecret | string | `""` |  |
+| global.logs.accessKey.existingSecretKey | string | `""` |  |
+| global.logs.accessKey.password | string | `"minio"` |  |
+| global.logs.gcs.bucket | string | `""` |  |
+| global.logs.gcs.credentials | string | `""` |  |
+| global.logs.gcs.credentialsJson | string | `""` |  |
+| global.logs.minio.enabled | bool | `true` |  |
+| global.logs.s3.bucket | string | `"airbyte-dev-logs"` |  |
+| global.logs.s3.bucketRegion | string | `""` |  |
+| global.logs.s3.enabled | bool | `false` |  |
+| global.logs.secretKey.existingSecret | string | `""` |  |
+| global.logs.secretKey.existingSecretKey | string | `""` |  |
+| global.logs.secretKey.password | string | `"minio123"` |  |
+| global.secretName | string | `""` |  |
+| global.serviceAccountName | string | `"placeholderServiceAccount"` |  |
+| image.pullPolicy | string | `"IfNotPresent"` |  |
+| image.repository | string | `"airbyte/workload-api-server"` |  |
+| ingress.annotations | object | `{}` |  |
+| ingress.className | string | `""` |  |
+| ingress.enabled | bool | `false` |  |
+| ingress.hosts | list | `[]` |  |
+| ingress.tls | list | `[]` |  |
+| livenessProbe.enabled | bool | `true` |  |
+| livenessProbe.failureThreshold | int | `3` |  |
+| livenessProbe.initialDelaySeconds | int | `60` |  |
+| livenessProbe.periodSeconds | int | `10` |  |
+| livenessProbe.successThreshold | int | `1` |  |
+| livenessProbe.timeoutSeconds | int | `1` |  |
+| log.level | string | `"INFO"` |  |
+| nodeSelector | object | `{}` |  |
+| podAnnotations | object | `{}` |  |
+| podLabels | object | `{}` |  |
+| readinessProbe.enabled | bool | `true` |  |
+| readinessProbe.failureThreshold | int | `3` |  |
+| readinessProbe.initialDelaySeconds | int | `30` |  |
+| readinessProbe.periodSeconds | int | `10` |  |
+| readinessProbe.successThreshold | int | `1` |  |
+| readinessProbe.timeoutSeconds | int | `1` |  |
+| replicaCount | int | `1` |  |
+| resources.limits | object | `{}` |  |
+| resources.requests | object | `{}` |  |
+| secrets | object | `{}` |  |
+| service.annotations | object | `{}` |  |
+| service.port | int | `8007` |  |
+| service.type | string | `"ClusterIP"` |  |
+| tolerations | list | `[]` |  |
+| workloadApi | object | `{}` |  |
+
diff --git a/charts/airbyte-featureflag-server/templates/_helpers.tpl b/charts/airbyte-featureflag-server/templates/_helpers.tpl
new file mode 100644
index 00000000000..e26d0f224a8
--- /dev/null
+++ b/charts/airbyte-featureflag-server/templates/_helpers.tpl
@@ -0,0 +1,72 @@
+{{/*
+Expand the name of the chart.
+*/}}
+{{- define "airbyte.name" -}}
+{{- default .Chart.Name .Values.nameOverride | trunc 63 | trimSuffix "-" }}
+{{- end }}
+
+{{/*
+Create a default fully qualified app name.
+We truncate at 63 chars because some Kubernetes name fields are limited to this (by the DNS naming spec).
+If release name contains chart name it will be used as a full name.
+*/}}
+{{- define "airbyte.fullname" -}}
+{{- if .Values.fullnameOverride }}
+{{- .Values.fullnameOverride | trunc 63 | trimSuffix "-" }}
+{{- else }}
+{{- $name := default .Chart.Name .Values.nameOverride }}
+{{- if contains $name .Release.Name }}
+{{- .Release.Name | trunc 63 | trimSuffix "-" }}
+{{- else }}
+{{- printf "%s-%s" .Release.Name $name | trunc 63 | trimSuffix "-" }}
+{{- end }}
+{{- end }}
+{{- end }}
+
+{{/*
+Create chart name and version as used by the chart label.
+*/}}
+{{- define "airbyte.chart" -}}
+{{- printf "%s-%s" .Chart.Name .Chart.Version | replace "+" "_" | trunc 63 | trimSuffix "-" }}
+{{- end }}
+
+{{/*
+Common labels
+*/}}
+{{- define "airbyte.labels" -}}
+helm.sh/chart: {{ include "airbyte.chart" . }}
+{{ include "airbyte.selectorLabels" . }}
+{{- if .Chart.AppVersion }}
+app.kubernetes.io/version: {{ .Chart.AppVersion | quote }}
+{{- end }}
+app.kubernetes.io/managed-by: {{ .Release.Service }}
+{{- end }}
+
+{{/*
+Selector labels
+*/}}
+{{- define "airbyte.selectorLabels" -}}
+app.kubernetes.io/name: {{ include "airbyte.name" . }}
+app.kubernetes.io/instance: {{ .Release.Name }}
+{{- end }}
+
+{{/*
+Define db secret
+*/}}
+
+{{- define "database.secret.name" -}}
+{{- printf "%s-postgresql" .Release.Name }}
+{{- end }}
+
+{{/*
+Define imageTag
+*/}}
+{{- define "workload-api.imageTag" -}}
+{{- if .Values.image.tag }}
+    {{- printf "%s" .Values.image.tag }}
+{{- else if ((.Values.global.image).tag) }}
+    {{- printf "%s" .Values.global.image.tag }}
+{{- else }}
+    {{- printf "%s" .Chart.AppVersion }}
+{{- end }}
+{{- end }}
diff --git a/charts/airbyte-featureflag-server/templates/deployment.yaml b/charts/airbyte-featureflag-server/templates/deployment.yaml
new file mode 100644
index 00000000000..89567b336db
--- /dev/null
+++ b/charts/airbyte-featureflag-server/templates/deployment.yaml
@@ -0,0 +1,177 @@
+{{- if eq .Values.global.edition "community"}}
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: {{ include "common.names.fullname" . }}
+  labels:
+    {{- include "airbyte.labels" . | nindent 4 }}
+spec:
+  minReadySeconds: 30
+  replicas: {{ .Values.replicaCount }}
+  selector:
+    matchLabels:
+      {{- include "airbyte.selectorLabels" . | nindent 6 }}
+      {{- if .Values.extraSelectorLabels }}
+      {{ toYaml (mergeOverwrite .Values.extraSelectorLabels .Values.global.extraSelectorLabels) | nindent 6 }}
+      {{- end }}
+  strategy:
+    type: RollingUpdate
+    rollingUpdate:
+      maxUnavailable: 0
+      maxSurge: 100%
+  template:
+    metadata:
+      labels:
+        {{- include "airbyte.selectorLabels" . | nindent 8 }}
+        {{- if .Values.extraSelectorLabels }}
+        {{ toYaml (mergeOverwrite .Values.extraSelectorLabels .Values.global.extraSelectorLabels) | nindent 8 }}
+        {{- end }}
+        {{- if .Values.podLabels }}
+          {{- include "common.tplvalues.render" (dict "value" .Values.podLabels "context" $) | nindent 8 }}
+        {{- end }}
+      {{- if .Values.podAnnotations }}
+      annotations:
+        {{- include "common.tplvalues.render" (dict "value" .Values.podAnnotations "context" $) | nindent 8 }}
+      {{- end }}
+    spec:
+      serviceAccountName: {{ .Values.global.serviceAccountName }}
+      {{- if .Values.global.imagePullSecrets }}
+      imagePullSecrets:
+        {{- range .Values.global.imagePullSecrets }}
+          {{- printf "- name: %s" .name | nindent 8 }}
+        {{- end }}
+      {{- end }}
+      {{- if .Values.nodeSelector }}
+      nodeSelector: {{- include "common.tplvalues.render" (dict "value" .Values.nodeSelector "context" $) | nindent 8 }}
+      {{- end }}
+      {{- if .Values.tolerations }}
+      tolerations: {{- include "common.tplvalues.render" (dict "value" .Values.tolerations "context" $) | nindent 8 }}
+      {{- end }}
+      {{- if .Values.affinity }}
+      affinity: {{- include "common.tplvalues.render" (dict "value" .Values.affinity "context" $) | nindent 8 }}
+      {{- end }}
+      {{- if .Values.extraInitContainers }}
+      initContainers:
+      {{- toYaml .Values.extraInitContainers | nindent 6 }}
+      {{- end }}
+      containers:
+      - name: airbyte-featureflag-server-container
+        image: {{ printf "%s:%s" .Values.image.repository (include "featureflag-server.imageTag" .) }}
+        imagePullPolicy: "{{ .Values.image.pullPolicy }}"
+        env:
+        {{- if .Values.debug.enabled }}
+        - name: JAVA_TOOL_OPTIONS
+          value: "-Xdebug -agentlib:jdwp=transport=dt_socket,address=0.0.0.0:{{ .Values.debug.remoteDebugPort }},server=y,suspend=n"
+        {{- end }}
+        {{- if eq .Values.global.deploymentMode "oss" }}
+        - name: AIRBYTE_VERSION
+          valueFrom:
+            configMapKeyRef:
+              name: {{ .Release.Name }}-airbyte-env
+              key: AIRBYTE_VERSION
+        - name: MICROMETER_METRICS_ENABLED
+          valueFrom:
+            configMapKeyRef:
+              name: {{ .Release.Name }}-airbyte-env
+              key: MICROMETER_METRICS_ENABLED
+        - name: MICROMETER_METRICS_STATSD_FLAVOR
+          valueFrom:
+            configMapKeyRef:
+              name: {{ .Release.Name }}-airbyte-env
+              key: MICROMETER_METRICS_STATSD_FLAVOR
+        - name: STATSD_HOST
+          valueFrom:
+            configMapKeyRef:
+              name: {{ .Release.Name }}-airbyte-env
+              key: STATSD_HOST
+        - name: STATSD_PORT
+          valueFrom:
+            configMapKeyRef:
+              name: {{ .Release.Name }}-airbyte-env
+              key: STATSD_PORT
+
+        {{- end }}
+
+        # Values from env
+        {{- if or .Values.env_vars .Values.global.env_vars }}
+        {{- range $k, $v := mergeOverwrite .Values.env_vars .Values.global.env_vars }}
+        - name: {{ $k }}
+          value: {{ $v | quote }}
+        {{- end }}
+        {{- end }}
+
+        # Values from extraEnv for more compability(if you want to use external secret source or other stuff)
+        {{- if .Values.extraEnv }}
+        {{- toYaml .Values.extraEnv | nindent 8 }}
+        {{- end }}
+
+        {{- if .Values.livenessProbe.enabled }}
+        livenessProbe:
+          httpGet:
+            path: /health/liveness
+            port: http
+          initialDelaySeconds: {{ .Values.livenessProbe.initialDelaySeconds }}
+          periodSeconds: {{ .Values.livenessProbe.periodSeconds }}
+          timeoutSeconds: {{ .Values.livenessProbe.timeoutSeconds }}
+          successThreshold: {{ .Values.livenessProbe.successThreshold }}
+          failureThreshold: {{ .Values.livenessProbe.failureThreshold }}
+        {{- end }}
+        {{- if .Values.readinessProbe.enabled }}
+        readinessProbe:
+          httpGet:
+            path: /health/liveness
+            port: http
+          initialDelaySeconds: {{ .Values.readinessProbe.initialDelaySeconds }}
+          periodSeconds: {{ .Values.readinessProbe.periodSeconds }}
+          timeoutSeconds: {{ .Values.readinessProbe.timeoutSeconds }}
+          successThreshold: {{ .Values.readinessProbe.successThreshold }}
+          failureThreshold: {{ .Values.readinessProbe.failureThreshold }}
+        {{- end }}
+
+        ports:
+        - name: http
+          containerPort: 8007
+          protocol: TCP
+        {{- if .Values.debug.enabled }}
+        - name: debug 
+          containerPort: {{ .Values.debug.remoteDebugPort }}
+          protocol: TCP
+        {{- end }}
+        {{- if .Values.resources }}
+        resources: {{- toYaml .Values.resources | nindent 10 }}
+        {{- end }}
+        {{- if .Values.containerSecurityContext }}
+        securityContext: {{- toYaml .Values.containerSecurityContext | nindent 10 }}
+        {{- end }}
+        volumeMounts:
+        {{- if and (eq .Values.global.deploymentMode "oss") (eq (lower (default "" .Values.global.storage.type)) "gcs") }}
+        - name: gcs-log-creds-volume
+          mountPath: /secrets/gcs-log-creds
+          readOnly: true
+        {{- end }}
+        {{- if .Values.extraVolumeMounts }}
+{{- toYaml .Values.extraVolumeMounts | nindent 8 }}
+        {{- end }}
+        {{- if .Values.global.extraVolumeMounts }}
+{{- toYaml .Values.global.extraVolumeMounts | nindent 8 }}
+        {{- end }}
+      {{- if .Values.extraContainers }}
+      {{ toYaml .Values.extraContainers | nindent 6 }}
+      {{- end }}
+      {{- if .Values.global.extraContainers }}
+      {{ toYaml .Values.global.extraContainers | nindent 6 }}
+      {{- end }}
+      securityContext: {{- toYaml .Values.podSecurityContext | nindent 8 }}
+      volumes:
+      {{- if and (eq .Values.global.deploymentMode "oss") (eq (lower (default "" .Values.global.storage.type)) "gcs") }}
+      - name: gcs-log-creds-volume
+        secret:
+          secretName: {{ ternary (printf "%s-gcs-log-creds" ( .Release.Name )) (.Values.global.credVolumeOverride) (eq .Values.global.deploymentMode "oss") }}
+      {{- end }}
+      {{- if .Values.extraVolumes }}
+      {{- toYaml .Values.extraVolumes | nindent 6 }}
+      {{- end }}
+      {{- if .Values.global.extraVolumes }}
+      {{- toYaml .Values.global.extraVolumes | nindent 6 }}
+      {{- end }}
+{{- end }}
diff --git a/charts/airbyte-featureflag-server/templates/secrets.yaml b/charts/airbyte-featureflag-server/templates/secrets.yaml
new file mode 100644
index 00000000000..73889bf5d78
--- /dev/null
+++ b/charts/airbyte-featureflag-server/templates/secrets.yaml
@@ -0,0 +1,17 @@
+# Create secrets only for the local deployment
+  {{- if .Values.secrets }}
+apiVersion: v1
+kind: Secret
+metadata:
+  name: featureflag-secrets
+  labels:
+    {{- include "airbyte.labels" . | nindent 4 }}
+  annotations:
+    helm.sh/hook: pre-install,pre-upgrade
+    helm.sh/hook-weight: "-1"
+type: Opaque
+data:
+  {{- range $k, $v := mergeOverwrite .Values.secrets .Values.global.secrets }}
+    {{ $k }}: {{ if $v }}{{ $v | b64enc  }} {{else}}""{{end}}
+    {{- end }}
+  {{- end }}
diff --git a/charts/airbyte-featureflag-server/templates/service.yaml b/charts/airbyte-featureflag-server/templates/service.yaml
new file mode 100644
index 00000000000..35bee968fee
--- /dev/null
+++ b/charts/airbyte-featureflag-server/templates/service.yaml
@@ -0,0 +1,22 @@
+apiVersion: v1
+kind: Service
+metadata:
+  name: {{.Release.Name }}-featureflag-server-svc
+  {{- with .Values.service.annotations }}
+  annotations:
+    {{- toYaml . | nindent 4 }}
+  {{- end }}
+  labels:
+    {{- include "airbyte.labels" . | nindent 4 }}
+spec:
+  type: {{ .Values.service.type }}
+  ports:
+    - port: {{ .Values.service.port }}
+      targetPort: http
+      protocol: TCP
+      name: http
+  selector:
+    {{- include "airbyte.selectorLabels" . | nindent 4 }}
+    {{- if .Values.extraSelectorLabels }}
+    {{ toYaml (mergeOverwrite .Values.extraSelectorLabels .Values.global.extraSelectorLabels) | nindent 4 }}
+    {{- end }}
diff --git a/charts/airbyte-featureflag-server/values.yaml b/charts/airbyte-featureflag-server/values.yaml
new file mode 100644
index 00000000000..c8346de56b9
--- /dev/null
+++ b/charts/airbyte-featureflag-server/values.yaml
@@ -0,0 +1,270 @@
+global:
+  serviceAccountName: placeholderServiceAccount
+  deploymentMode: oss
+  edition: community
+  configMapName: ""
+  secretName: ""
+  credVolumeOverride: ""
+  extraContainers: []
+  ## extraSelectorLabels [object] - use to specify own additional selector labels for deployment
+  extraSelectorLabels: {}
+  ## extraLabels [object] - use to specify own additional labels for deployment
+  extraLabels: {}
+  database:
+    secretName: ""
+    secretValue: ""
+
+  enterprise:
+    workloadsOptIn: false
+
+  logs:
+    ##  logs.accessKey.password Logs Access Key
+    ##  logs.accessKey.existingSecret
+    ##  logs.accessKey.existingSecretKey
+    accessKey:
+      password: minio
+      existingSecret: ""
+      existingSecretKey: ""
+    ##  logs.secretKey.password Logs Secret Key
+    ##  logs.secretKey.existingSecret
+    ##  logs.secretKey.existingSecretKey
+    secretKey:
+      password: minio123
+      existingSecret: ""
+      existingSecretKey: ""
+
+    ##  logs.minio.enabled Switch to enable or disable the Minio helm chart
+    minio:
+      enabled: true
+
+    ##  logs.s3.enabled Switch to enable or disable custom S3 Log location
+    ##  logs.s3.bucket Bucket name where logs should be stored
+    ##  logs.s3.bucketRegion Region of the bucket (must be empty if using minio)
+    s3:
+      enabled: false
+      bucket: airbyte-dev-logs
+      bucketRegion: ""
+
+    ## Google Cloud Storage (GCS) Log Location Configuration
+    ##  logs.gcs.bucket GCS bucket name
+    ##  logs.gcs.credentials The path the GCS creds are written to
+    ##  logs.gcs.credentialsJson Base64 encoded json GCP credentials file contents
+    gcs:
+      bucket: ""
+      # If you are mounting an existing secret to extraVolumes on scheduler, server and worker
+      # deployments, then set credentials to the path of the mounted JSON file
+      credentials: ""
+      # If credentialsJson is set then credentials auto resolves (to /secrets/gcs-log-creds/gcp.json)
+      credentialsJson: ""
+
+enabled: false
+##  replicaCount Number of server replicas
+replicaCount: 1
+
+##  image.repository The repository to use for the airbyte workload api server image.
+##  image.pullPolicy the pull policy to use for the airbyte workload api server image
+##  image.tag The airbyte featureflag server image tag. Defaults to the chart's AppVersion
+image:
+  repository: airbyte/featureflag-server
+  pullPolicy: IfNotPresent
+
+##  podAnnotations [object] Add extra annotations to the server pod
+##
+podAnnotations: {}
+
+##  podLabels [object] Add extra labels to the server pod
+##
+podLabels: {}
+
+##  containerSecurityContext Security context for the container
+## Examples:
+## containerSecurityContext:
+##    runAsNonRoot: true
+##    runAsUser: 1000
+##    readOnlyRootFilesystem: true
+containerSecurityContext: {}
+
+## Configure extra options for the server containers' liveness and readiness probes
+## ref: https://kubernetes.io/docs/tasks/configure-pod-container/configure-liveness-readiness-probes/#configure-probes
+##  livenessProbe.enabled Enable livenessProbe on the server
+##  livenessProbe.initialDelaySeconds Initial delay seconds for livenessProbe
+##  livenessProbe.periodSeconds Period seconds for livenessProbe
+##  livenessProbe.timeoutSeconds Timeout seconds for livenessProbe
+##  livenessProbe.failureThreshold Failure threshold for livenessProbe
+##  livenessProbe.successThreshold Success threshold for livenessProbe
+##
+livenessProbe:
+  enabled: true
+  initialDelaySeconds: 60
+  periodSeconds: 10
+  timeoutSeconds: 1
+  failureThreshold: 3
+  successThreshold: 1
+
+##  readinessProbe.enabled Enable readinessProbe on the server
+##  readinessProbe.initialDelaySeconds Initial delay seconds for readinessProbe
+##  readinessProbe.periodSeconds Period seconds for readinessProbe
+##  readinessProbe.timeoutSeconds Timeout seconds for readinessProbe
+##  readinessProbe.failureThreshold Failure threshold for readinessProbe
+##  readinessProbe.successThreshold Success threshold for readinessProbe
+##
+readinessProbe:
+  enabled: true
+  initialDelaySeconds: 30
+  periodSeconds: 10
+  timeoutSeconds: 1
+  failureThreshold: 3
+  successThreshold: 1
+
+## Server app resource requests and limits
+## ref: http://kubernetes.io/docs/user-guide/compute-resources/
+## We usually recommend not to specify default resources and to leave this as a conscious
+## choice for the user. This also increases chances charts run on environments with little
+## resources, such as Minikube. If you do want to specify resources, uncomment the following
+## lines, adjust them as necessary, and remove the curly braces after 'resources:'.
+##  resources.limits [object] The resources limits for the server container
+##  resources.requests [object] The requested resources for the server container
+resources:
+  ## Example:
+  ## limits:
+  ##    cpu: 200m
+  ##    memory: 1Gi
+  limits: {}
+  ## Examples:
+  ## requests:
+  ##    memory: 256Mi
+  ##    cpu: 250m
+  requests: {}
+
+##  service.type The service type to use for the API server
+##  service.port The service port to expose the API server on
+service:
+  type: ClusterIP
+  port: 8007
+  annotations: {}
+
+##  nodeSelector [object] Node labels for pod assignment
+## Ref: https://kubernetes.io/docs/user-guide/node-selection/
+##
+nodeSelector: {}
+
+##  tolerations [array] Tolerations for server pod assignment.
+## ref: https://kubernetes.io/docs/concepts/configuration/taint-and-toleration/
+##
+tolerations: []
+
+##  affinity [object] Affinity and anti-affinity for server pod assignment.
+## ref: https://kubernetes.io/docs/concepts/scheduling-eviction/assign-pod-node/#affinity-and-anti-affinity
+##
+affinity: {}
+
+## Configure the ingress resource that allows you to access the Airbyte API.
+## ref: http://kubernetes.io/docs/user-guide/ingress/
+##  webapp.ingress.enabled Set to true to enable ingress record generation
+##  webapp.ingress.className Specifies ingressClassName for clusters >= 1.18+
+##  webapp.ingress.annotations [object] Ingress annotations done as key:value pairs
+##  webapp.ingress.hosts The list of hostnames to be covered with this ingress record.
+##  webapp.ingress.tls [array] Custom ingress TLS configuration
+ingress:
+  enabled: false
+  className: ""
+  annotations: {}
+  # kubernetes.io/ingress.class: nginx
+  # kubernetes.io/tls-acme: "true"
+  hosts: []
+  # - host: chart-example.local
+  #   paths:
+  #   - path: /
+  #     pathType: ImplementationSpecific
+
+  tls: []
+  # - secretName: chart-example-tls
+  #   hosts:
+  #   - chart-example.local
+
+##  log.level The log level to log at
+log:
+  level: "INFO"
+
+##  extraVolumeMounts [array] Additional volumeMounts for server container(s).
+## Examples (when using `containerSecurityContext.readOnlyRootFilesystem=true`):
+## extraVolumeMounts:
+##   - name: tmpdir
+##     mountPath: /tmp
+##
+extraVolumeMounts: []
+
+##  extraVolumes [array] Additional volumes for server pod(s).
+## Examples (when using `containerSecurityContext.readOnlyRootFilesystem=true`):
+## extraVolumes:
+##   - name: tmpdir
+##     emptyDir: {}
+##
+extraVolumes: []
+
+##  extraContainer [array] Additional container for server pod(s)
+## Example:
+# extraContainers:
+#   - name: otel_collector
+#     image: somerepo/someimage:sometag
+#     args: [
+#         "--important-args"
+#     ]
+#     ports:
+#       - containerPort: 443
+#     volumeMounts:
+#       - name: volumeMountCool
+#         mountPath: /some/path
+#         readOnly: true
+extraContainers: []
+
+##  extraInitContainers [array] Additional init containers for server pod(s)
+## Example:
+# extraInitContainers:
+#   - name: sleepy
+#     image: alpine
+#     command: ['sleep', '60']
+
+extraInitContainers: []
+
+##  extraEnv [array] Supply extra env variables to main container using full notation
+## Example: (With default env vars and values taken from generated config map)
+# extraEnv:
+#   - name: AIRBYTE_VERSION
+#     valueFrom:
+#       configMapKeyRef:
+#         name: airbyte-env
+#         key: AIRBYTE_VERSION
+##
+extraEnv: []
+##  secrets [object] Supply additional secrets to container
+## Example:
+## secrets:
+##   DATABASE_PASSWORD: strong-password
+##   DATABASE_USER: my-db-user
+secrets: {}
+
+##  env_vars [object] Supply extra env variables to main container using simplified notation
+## Example:
+# env_vars:
+#   AIRBYTE_VERSION: 0.40.4
+
+#   # Airbyte Internal Database, see https://docs.airbyte.io/operator-guides/configuring-airbyte-db
+#   DATABASE_HOST: airbyte-db-svc
+#   DATABASE_PORT: 5432
+#   DATABASE_DB: airbyte
+#   # translate manually DATABASE_URL:jdbc:postgresql://${DATABASE_HOST}:${DATABASE_PORT/${DATABASE_DB}
+#   DATABASE_URL: jdbc:postgresql://airbyte-db-svc:5432/airbyte
+#   JOBS_DATABASE_MINIMUM_FLYWAY_MIGRATION_VERSION: 0.29.15.001
+#   CONFIGS_DATABASE_MINIMUM_FLYWAY_MIGRATION_VERSION: 0.35.15.001
+env_vars: {}
+
+
+## extraSelectorLabels [object] - use to specify own additional selector labels for deployment
+extraSelectorLabels: {}
+## extraLabels [object] - use to specify own additional labels for deployment
+extraLabels: {}
+
+debug:
+  enabled: false
+  remoteDebugPort: 5005
diff --git a/charts/airbyte-workload-api-server/templates/_helpers.tpl b/charts/airbyte-workload-api-server/templates/_helpers.tpl
index e26d0f224a8..99db75284c2 100644
--- a/charts/airbyte-workload-api-server/templates/_helpers.tpl
+++ b/charts/airbyte-workload-api-server/templates/_helpers.tpl
@@ -70,3 +70,12 @@ Define imageTag
     {{- printf "%s" .Chart.AppVersion }}
 {{- end }}
 {{- end }}
+{{- define "featureflag-server.imageTag" -}}
+{{- if .Values.image.tag }}
+    {{- printf "%s" .Values.image.tag }}
+{{- else if ((.Values.global.image).tag) }}
+    {{- printf "%s" .Values.global.image.tag }}
+{{- else }}
+    {{- printf "%s" .Chart.AppVersion }}
+{{- end }}
+{{- end }}
diff --git a/charts/airbyte/Chart.yaml b/charts/airbyte/Chart.yaml
index 169f5c79eb9..1b7150a5851 100644
--- a/charts/airbyte/Chart.yaml
+++ b/charts/airbyte/Chart.yaml
@@ -1,7 +1,11 @@
+### This Chart.yaml file is used for local testing purposes
+### If you want to test the chart before creating PR, rename this file into Chart.yaml and test the deployment
+
 apiVersion: v2
 name: airbyte
 description: Helm chart to deploy airbyte
 
+
 # A chart can be either an 'application' or a 'library' chart.
 #
 # Application charts are a collection of templates that can be packaged into versioned archives
diff --git a/charts/airbyte/Chart.yaml.local b/charts/airbyte/Chart.yaml.local
index 37497b1a739..2295cb9480d 100644
--- a/charts/airbyte/Chart.yaml.local
+++ b/charts/airbyte/Chart.yaml.local
@@ -95,4 +95,7 @@ dependencies:
     name: keycloak-setup
     repository: "file://../airbyte-keycloak-setup"
     version: "*"
-
+  - condition: featureflag-server.enabled
+    name: featureflag-server
+    repository: "file://../airbyte-featureflag-server"
+    version: "*"
diff --git a/charts/airbyte/Chart.yaml.test b/charts/airbyte/Chart.yaml.test
index a0618f2bf9c..1f1b178061d 100644
--- a/charts/airbyte/Chart.yaml.test
+++ b/charts/airbyte/Chart.yaml.test
@@ -85,3 +85,7 @@ dependencies:
     name: keycloak-setup
     repository: "file://../airbyte-keycloak-setup"
     version: "*"
+  - condition: featureflag-server.enabled
+    name: featureflag-server
+    repository: "file://../airbyte-featureflag-server"
+    version: "*"
diff --git a/charts/airbyte/values.yaml b/charts/airbyte/values.yaml
index a26cee26c8e..9ba20295f42 100644
--- a/charts/airbyte/values.yaml
+++ b/charts/airbyte/values.yaml
@@ -1959,3 +1959,122 @@ workload-api-server:
     # - secretName: chart-example-tls
     #   hosts:
     #   - chart-example.local
+
+featureflag-server:
+  enabled: false
+
+
+  # -- workload-api-server replicas
+  replicaCount: 1
+
+  image:
+    # -- The repository to use for the airbyte-workload-api-server image.
+    repository: airbyte/featureflag-server
+    # -- The pull policy to use for the airbyte-workload-api-server image
+    pullPolicy: IfNotPresent
+
+  # -- Security context for the container
+  podSecurityContext:
+    # gid=1000(airbyte)
+    fsGroup: 1000
+
+  containerSecurityContext:
+    allowPrivilegeEscalation: false
+    runAsNonRoot: true
+    # uid=1000(airbyte)
+    runAsUser: 1000
+    # gid=1000(airbyte)
+    runAsGroup: 1000
+    readOnlyRootFilesystem: false
+    capabilities:
+      drop: ["ALL"]
+    seccompProfile:
+      type: RuntimeDefault
+
+  livenessProbe:
+    # -- Enable livenessProbe on the server
+    enabled: true
+    # -- Initial delay seconds for livenessProbe
+    initialDelaySeconds: 30
+    # -- Period seconds for livenessProbe
+    periodSeconds: 10
+    # -- Timeout seconds for livenessProbe
+    timeoutSeconds: 10
+    # -- Failure threshold for livenessProbe
+    failureThreshold: 3
+    # -- Success threshold for livenessProbe
+    successThreshold: 1
+
+  readinessProbe:
+    # -- Enable readinessProbe on the server
+    enabled: true
+    # -- Initial delay seconds for readinessProbe
+    initialDelaySeconds: 10
+    # -- Period seconds for readinessProbe
+    periodSeconds: 10
+    # -- Timeout seconds for readinessProbe
+    timeoutSeconds: 10
+    # -- Failure threshold for readinessProbe
+    failureThreshold: 3
+    # -- Success threshold for readinessProbe
+    successThreshold: 1
+
+  ## airbyte-workload-api-server resource requests and limits
+  ## ref: http://kubernetes.io/docs/user-guide/compute-resources/
+  ## We usually recommend not to specify default resources and to leave this as a conscious
+  ## choice for the user. This also increases chances charts run on environments with little
+  ## resources, such as Minikube. If you do want to specify resources, uncomment the following
+  ## lines, adjust them as necessary, and remove the curly braces after 'resources:'.
+  resources:
+    ## Example:
+    ## limits:
+    ##    cpu: 200m
+    ##    memory: 1Gi
+    # -- The resources limits for the airbyte-workload-api-server container
+    limits: {}
+    ## Examples:
+    ## requests:
+    ##    memory: 256Mi
+    ##    cpu: 250m
+    # -- The requested resources for the airbyte-workload-api-server container
+    requests: {}
+
+  log:
+    # -- The log level at which to log
+    level: "INFO"
+
+  # -- Node labels for pod assignment, see https://kubernetes.io/docs/user-guide/node-selection/
+  nodeSelector: {}
+
+  # -- Tolerations for webapp pod assignment, see https://kubernetes.io/docs/concepts/configuration/taint-and-toleration/
+  tolerations: []
+
+  # -- Affinity and anti-affinity for webapp pod assignment, see
+  # https://kubernetes.io/docs/concepts/scheduling-eviction/assign-pod-node/#affinity-and-anti-affinity
+  affinity: {}
+
+  env_vars: {}
+  service:
+    port: 8007
+
+  # Configure the ingress resource that allows you to access the Airbyte Workload API, see http://kubernetes.io/docs/user-guide/ingress/
+  ingress:
+    # -- Set to true to enable ingress record generation
+    enabled: false
+    # -- Specifies ingressClassName for clusters >= 1.18+
+    className: ""
+    # -- Ingress annotations done as key:value pairs
+    annotations: {}
+    # kubernetes.io/ingress.class: nginx
+    # kubernetes.io/tls-acme: "true"
+    # -- The list of hostnames to be covered with this ingress record
+    hosts: []
+    # - host: chart-example.local
+    #   paths:
+    #   - path: /
+    #     pathType: ImplementationSpecific
+    # -- Custom ingress TLS configuration
+    tls: []
+    # - secretName: chart-example-tls
+    #   hosts:
+    #   - chart-example.local
diff --git a/charts/airbyte/values.yaml.test b/charts/airbyte/values.yaml.test
index 4ba00e8bfed..8df23e2427d 100644
--- a/charts/airbyte/values.yaml.test
+++ b/charts/airbyte/values.yaml.test
@@ -1728,6 +1728,102 @@ workload-api-server:
     requests: {}
 
 
+  ##  workload-api-server.log.level The log level to log at.
+  log:
+    level: "INFO"
+
+  env_vars: {}
+  service:
+    port: 8007
+
+  ## Configure the ingress resource that allows you to access the Airbyte API.
+  ## ref: http://kubernetes.io/docs/user-guide/ingress/
+  ##  webapp.ingress.enabled Set to true to enable ingress record generation
+  ##  webapp.ingress.className Specifies ingressClassName for clusters >= 1.18+
+  ##  webapp.ingress.annotations [object] Ingress annotations done as key:value pairs
+  ##  webapp.ingress.hosts The list of hostnames to be covered with this ingress record.
+  ##  webapp.ingress.tls [array] Custom ingress TLS configuration
+  ingress:
+    enabled: false
+    className: ""
+    annotations: {}
+    # kubernetes.io/ingress.class: nginx
+    # kubernetes.io/tls-acme: "true"
+    hosts: []
+    # - host: chart-example.local
+    #   paths:
+    #   - path: /
+    #     pathType: ImplementationSpecific
+
+    tls: []
+    # - secretName: chart-example-tls
+    #   hosts:
+    #   - chart-example.local
+
+featureflag-server:
+  enabled: false
+  replicaCount: 1
+
+  ##  workload-api-server.image.repository The repository to use for the airbyte workload-api-server image.
+  ##  workload-api-server.image.pullPolicy the pull policy to use for the airbyte workload-api-server image
+  ##  workload-api-server.image.tag The airbyte workload-api-server image tag. Defaults to the chart's AppVersion
+  image:
+    repository: airbyte/airbyte-featureflag-server
+    pullPolicy: IfNotPresent
+
+  ## Configure extra options for the server containers' liveness and readiness probes
+  ## ref: https://kubernetes.io/docs/tasks/configure-pod-container/configure-liveness-readiness-probes/#configure-probes
+  ##  server.livenessProbe.enabled Enable livenessProbe on the server
+  ##  server.livenessProbe.initialDelaySeconds Initial delay seconds for livenessProbe
+  ##  server.livenessProbe.periodSeconds Period seconds for livenessProbe
+  ##  server.livenessProbe.timeoutSeconds Timeout seconds for livenessProbe
+  ##  server.livenessProbe.failureThreshold Failure threshold for livenessProbe
+  ##  server.livenessProbe.successThreshold Success threshold for livenessProbe
+  ##
+  livenessProbe:
+    enabled: true
+    initialDelaySeconds: 30
+    periodSeconds: 10
+    timeoutSeconds: 10
+    failureThreshold: 3
+    successThreshold: 1
+
+  ##  server.readinessProbe.enabled Enable readinessProbe on the server
+  ##  server.readinessProbe.initialDelaySeconds Initial delay seconds for readinessProbe
+  ##  server.readinessProbe.periodSeconds Period seconds for readinessProbe
+  ##  server.readinessProbe.timeoutSeconds Timeout seconds for readinessProbe
+  ##  server.readinessProbe.failureThreshold Failure threshold for readinessProbe
+  ##  server.readinessProbe.successThreshold Success threshold for readinessProbe
+  ##
+  readinessProbe:
+    enabled: true
+    initialDelaySeconds: 10
+    periodSeconds: 10
+    timeoutSeconds: 10
+    failureThreshold: 3
+    successThreshold: 1
+
+  ## workload-api-server resource requests and limits
+  ## ref: http://kubernetes.io/docs/user-guide/compute-resources/
+  ## We usually recommend not to specify default resources and to leave this as a conscious
+  ## choice for the user. This also increases chances charts run on environments with little
+  ## resources, such as Minikube. If you do want to specify resources, uncomment the following
+  ## lines, adjust them as necessary, and remove the curly braces after 'resources:'.
+  ##  workload-api-server.resources.limits [object] The resources limits for the workload-api-server container
+  ##  workload-api-server.resources.requests [object] The requested resources for the workload-api-server container
+  resources:
+    ## Example:
+    ## limits:
+    ##    cpu: 200m
+    ##    memory: 1Gi
+    limits: {}
+    ## Examples:
+    ## requests:
+    ##    memory: 256Mi
+    ##    cpu: 250m
+    requests: {}
+
+
   ##  workload-api-server.log.level The log level to log at.
   log:
     level: "INFO"
diff --git a/settings.gradle.kts b/settings.gradle.kts
index 7f904b4e6fe..35f30ce8a82 100644
--- a/settings.gradle.kts
+++ b/settings.gradle.kts
@@ -100,6 +100,7 @@ include(":oss:airbyte-commons-worker")
 include(":oss:airbyte-config:config-persistence")
 include(":oss:airbyte-config:config-secrets")
 include(":oss:airbyte-featureflag")
+include(":oss:airbyte-featureflag-server")
 include(":oss:airbyte-db:jooq")
 include(":oss:airbyte-micronaut-temporal")
 include(":oss:airbyte-notification")
@@ -156,6 +157,7 @@ project(":oss:airbyte-commons-worker").projectDir = file("airbyte-commons-worker
 project(":oss:airbyte-config:config-persistence").projectDir = file("airbyte-config/config-persistence")
 project(":oss:airbyte-config:config-secrets").projectDir = file("airbyte-config/config-secrets")
 project(":oss:airbyte-featureflag").projectDir = file("airbyte-featureflag")
+project(":oss:airbyte-featureflag-server").projectDir = file("airbyte-featureflag-server")
 project(":oss:airbyte-db:jooq").projectDir = file("airbyte-db/jooq")
 project(":oss:airbyte-micronaut-temporal").projectDir = file("airbyte-micronaut-temporal")
 project(":oss:airbyte-notification").projectDir = file("airbyte-notification")