diff --git a/.github/workflows/build.yaml b/.github/workflows/build.yaml
index 7251f20b..2c6cc000 100644
--- a/.github/workflows/build.yaml
+++ b/.github/workflows/build.yaml
@@ -62,8 +62,22 @@ jobs:
             - '**/Cargo.toml'
             - 'Cargo.lock'
             - '**/*.rs'
+            - 'nix/meta.json'
+
+      # Operators that ship an agent (see `agent` in nix/meta.json) also build and publish its image.
+      - name: Determine Container Images
+        id: images
+        shell: bash
+        run: |
+          set -euo pipefail
+          IMAGES=$(jq -c --arg operator "$OPERATOR_NAME" '
+            [{component: "operator", name: $operator, "container-file": "docker/Dockerfile"}]
+            + if .agent then [{component: "agent", name: .agent.name, "container-file": "docker/Dockerfile.agent"}] else [] end
+          ' nix/meta.json)
+          echo "IMAGES=$IMAGES" | tee -a "$GITHUB_OUTPUT"
     outputs:
       detected: ${{ steps.check.outputs.detected }}
+      images: ${{ steps.images.outputs.IMAGES }}
 
   helm-lint:
     name: Lint Helm Chart
@@ -131,7 +145,7 @@ jobs:
         run: cargo udeps --workspace --all-targets
 
   build-container-image:
-    name: Build/Publish ${{ matrix.runner.arch }} Image
+    name: Build/Publish ${{ matrix.image.component }} ${{ matrix.runner.arch }} Image
     if: |
       github.repository_owner == 'stackabletech'
       && (github.event_name != 'merge_group')
@@ -146,6 +160,7 @@ jobs:
         runner:
           - { name: "ubuntu-latest", arch: "amd64" }
           - { name: "ubicloud-standard-8-arm", arch: "arm64" }
+        image: ${{ fromJSON(needs.detect-changes.outputs.images) }}
     runs-on: ${{ matrix.runner.name }}
     outputs:
       operator-version: ${{ steps.version.outputs.OPERATOR_VERSION }}
@@ -208,10 +223,10 @@ jobs:
         id: build
         uses: stackabletech/actions/build-container-image@34a64229959b15db6c5f307db8809805ba7edc55 # v0.18.3
         with:
-          image-name: ${{ env.OPERATOR_NAME }}
+          image-name: ${{ matrix.image.name }}
           image-index-manifest-tag: ${{ steps.version.outputs.OPERATOR_VERSION }}
           build-arguments: VERSION=${{ steps.version.outputs.OPERATOR_VERSION }}
-          container-file: docker/Dockerfile
+          container-file: ${{ matrix.image.container-file }}
 
       - name: Publish Container Image to oci.stackable.tech
         if: ${{ !github.event.pull_request.head.repo.fork }}
@@ -220,7 +235,7 @@ jobs:
           image-registry-uri: oci.stackable.tech
           image-registry-username: robot$sdp+github-action-build
           image-registry-password: ${{ secrets.HARBOR_ROBOT_SDP_GITHUB_ACTION_BUILD_SECRET }}
-          image-repository: sdp/${{ env.OPERATOR_NAME }}
+          image-repository: sdp/${{ matrix.image.name }}
           canonical-image-manifest-tag: ${{ steps.build.outputs.image-manifest-tag }}
           canonical-source-image-uri: ${{ steps.build.outputs.image-manifest-uri }}
 
@@ -231,12 +246,12 @@ jobs:
           image-registry-uri: quay.io
           image-registry-username: stackable+robot_sdp_github_action_build
           image-registry-password: ${{ secrets.QUAY_ROBOT_SDP_GITHUB_ACTION_BUILD_SECRET }}
-          image-repository: stackable/sdp/${{ env.OPERATOR_NAME }}
+          image-repository: stackable/sdp/${{ matrix.image.name }}
           canonical-image-manifest-tag: ${{ steps.build.outputs.image-manifest-tag }}
           canonical-source-image-uri: ${{ steps.build.outputs.image-manifest-uri }}
 
   publish-index-manifest:
-    name: Publish/Sign ${{ needs.build-container-image.outputs.operator-version }} Index
+    name: Publish/Sign ${{ matrix.image.component }} ${{ needs.build-container-image.outputs.operator-version }} Index
     if: |
       github.repository_owner == 'stackabletech'
       && (github.event_name != 'merge_group')
@@ -248,6 +263,10 @@ jobs:
     permissions:
       contents: read
       id-token: write
+    strategy:
+      fail-fast: false
+      matrix:
+        image: ${{ fromJSON(needs.detect-changes.outputs.images) }}
     runs-on: ubuntu-latest
     steps:
       - name: Checkout Repository
@@ -261,7 +280,7 @@ jobs:
           image-registry-uri: oci.stackable.tech
           image-registry-username: robot$sdp+github-action-build
           image-registry-password: ${{ secrets.HARBOR_ROBOT_SDP_GITHUB_ACTION_BUILD_SECRET }}
-          image-repository: sdp/${{ env.OPERATOR_NAME }}
+          image-repository: sdp/${{ matrix.image.name }}
           canonical-image-index-manifest-tag: ${{ needs.build-container-image.outputs.operator-version }}
 
       - name: Publish and Sign Image Index to quay.io
@@ -270,7 +289,7 @@ jobs:
           image-registry-uri: quay.io
           image-registry-username: stackable+robot_sdp_github_action_build
           image-registry-password: ${{ secrets.QUAY_ROBOT_SDP_GITHUB_ACTION_BUILD_SECRET }}
-          image-repository: stackable/sdp/${{ env.OPERATOR_NAME }}
+          image-repository: stackable/sdp/${{ matrix.image.name }}
           canonical-image-index-manifest-tag: ${{ needs.build-container-image.outputs.operator-version }}
 
   publish-helm-chart:
@@ -320,7 +339,7 @@ jobs:
           helm-version: v3.17.4 # This is currently the latest version which supports pushing to quay.io
 
   openshift-preflight-check:
-    name: Run OpenShift Preflight Check for ${{ needs.build-container-image.outputs.operator-version }}-${{ matrix.arch }}
+    name: Run OpenShift Preflight Check for ${{ matrix.image.component }} ${{ needs.build-container-image.outputs.operator-version }}-${{ matrix.arch }}
     if: |
       github.repository_owner == 'stackabletech'
       && (github.event_name != 'merge_group')
@@ -336,18 +355,19 @@ jobs:
         arch:
           - amd64
           - arm64
+        image: ${{ fromJSON(needs.detect-changes.outputs.images) }}
     runs-on: ubuntu-latest
     steps:
       - name: Run OpenShift Preflight Check for oci.stackable.tech
         uses: stackabletech/actions/run-openshift-preflight@34a64229959b15db6c5f307db8809805ba7edc55 # v0.18.3
         with:
-          image-index-uri: oci.stackable.tech/sdp/${{ env.OPERATOR_NAME }}:${{ needs.build-container-image.outputs.operator-version }}
+          image-index-uri: oci.stackable.tech/sdp/${{ matrix.image.name }}:${{ needs.build-container-image.outputs.operator-version }}
           image-architecture: ${{ matrix.arch }}
 
       - name: Run OpenShift Preflight Check for quay.io
         uses: stackabletech/actions/run-openshift-preflight@34a64229959b15db6c5f307db8809805ba7edc55 # v0.18.3
         with:
-          image-index-uri: quay.io/stackable/sdp/${{ env.OPERATOR_NAME }}:${{ needs.build-container-image.outputs.operator-version }}
+          image-index-uri: quay.io/stackable/sdp/${{ matrix.image.name }}:${{ needs.build-container-image.outputs.operator-version }}
           image-architecture: ${{ matrix.arch }}
 
   # This job is a required check in GitHub Settings for this repository.
diff --git a/.gitignore b/.gitignore
index 411cc85d..cc2e5c12 100644
--- a/.gitignore
+++ b/.gitignore
@@ -13,6 +13,7 @@ target/
 *.tgz
 
 result
+result-agent
 image.tar
 
 tilt_options.json
diff --git a/Cargo.lock b/Cargo.lock
index 75704686..9842b5ab 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -439,22 +439,13 @@ dependencies = [
 
 [[package]]
 name = "convert_case"
-version = "0.11.0"
+version = "0.12.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "affbf0190ed2caf063e3def54ff444b449371d55c58e513a95ab98eca50adb49"
+checksum = "1af709f1f33454bf52eadfc8c78b3b9ef9cb26fb54d16dc9cd9a7299f899fd1b"
 dependencies = [
  "unicode-segmentation",
 ]
 
-[[package]]
-name = "convert_case_extras"
-version = "0.2.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "589c70f0faf8aa9d17787557d5eae854d7755cac50f5c3d12c81d3d57661cebb"
-dependencies = [
- "convert_case",
-]
-
 [[package]]
 name = "core-foundation"
 version = "0.10.1"
@@ -773,9 +764,9 @@ dependencies = [
 
 [[package]]
 name = "educe"
-version = "0.7.6"
+version = "0.8.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e451fac8dd8dece16234604bf1efce6e90fddd8ab6ad4d66eec0eca5160959dd"
+checksum = "1ebdb1f8f8d2815bbe7b369a7da10343c82026a4503b68bd1279ba557611e96d"
 dependencies = [
  "enum-ordinalize",
  "proc-macro2",
@@ -1707,7 +1698,7 @@ dependencies = [
 [[package]]
 name = "k8s-version"
 version = "0.1.3"
-source = "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.118.0#bc6c84025c2dcc834b94bfb57ec72810ae5f5eb1"
+source = "git+https://github.com/stackabletech/operator-rs.git?branch=feat%2Fplatform-access#f3d2da737aa49086cbdaba806aaf2c8645251cfb"
 dependencies = [
  "darling 0.24.1",
  "regex",
@@ -1978,7 +1969,7 @@ dependencies = [
  "num-integer",
  "num-iter",
  "num-traits",
- "rand 0.8.7",
+ "rand 0.8.8",
  "smallvec",
  "zeroize",
 ]
@@ -2038,9 +2029,9 @@ checksum = "7c87def4c32ab89d880effc9e097653c8da5d6ef28e6b539d313baaacfbafcbe"
 
 [[package]]
 name = "opentelemetry"
-version = "0.32.0"
+version = "0.33.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b0142c63252a9e054e68a4c61a5778f7b14f576274d593f8ce883d191a099682"
+checksum = "6cdb0b1b267eb9db3331b434ed9ddab10d50e280a9adf9d13e5233e2002b61b5"
 dependencies = [
  "futures-core",
  "futures-sink",
@@ -2052,9 +2043,9 @@ dependencies = [
 
 [[package]]
 name = "opentelemetry-appender-tracing"
-version = "0.32.0"
+version = "0.33.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2c0080f0dc1d7c786f467cd85a4e395fcab11ee852004f39a29a18ab7c25d837"
+checksum = "e2b304e8d8f11326ba366a99ca5c7aaf7cd397b67268a332437175aeb42a40db"
 dependencies = [
  "opentelemetry",
  "tracing",
@@ -2064,9 +2055,9 @@ dependencies = [
 
 [[package]]
 name = "opentelemetry-http"
-version = "0.32.0"
+version = "0.33.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5683015d09e2df236ef005b17f6f196f0d5f6313c4fa43a7b6a53b52776e4331"
+checksum = "ee2c3625b8aa04209f7e01e513bc8044da9687fa38a9eacf59628ca5f3b87300"
 dependencies = [
  "async-trait",
  "bytes",
@@ -2077,11 +2068,12 @@ dependencies = [
 
 [[package]]
 name = "opentelemetry-otlp"
-version = "0.32.0"
+version = "0.33.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9966929966d17620d7c316c643ba62631826e10021409357772d5eea84f62c35"
+checksum = "699a67345e21962a231955b9059a157e428b219fa5df8efe11f08346fb23a344"
 dependencies = [
  "http",
+ "httpdate",
  "opentelemetry",
  "opentelemetry-http",
  "opentelemetry-proto",
@@ -2096,9 +2088,9 @@ dependencies = [
 
 [[package]]
 name = "opentelemetry-proto"
-version = "0.32.0"
+version = "0.33.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "56d658ba1faf63f7b9c492cfbe6e0ec365440a16132d3270c1065f7b33f1b638"
+checksum = "25da1ac11a0aeccf38d7f77ee0348715adaf8340f65ad46c94a02c6b20e2f65d"
 dependencies = [
  "opentelemetry",
  "opentelemetry_sdk",
@@ -2109,15 +2101,15 @@ dependencies = [
 
 [[package]]
 name = "opentelemetry-semantic-conventions"
-version = "0.32.1"
+version = "0.33.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c913ac17a6c451661ee255f4625d143e51647ae78ebd969b75e41c4442f4fe47"
+checksum = "d288b1e1bc3590052fbb8f0f635ef248dd9022cf1fbaeb6582b7ad6081a17b87"
 
 [[package]]
 name = "opentelemetry_sdk"
-version = "0.32.1"
+version = "0.33.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9b59f80e1ac4d5ff7a2db8fb6c80badb7f0f3f858211fba08dd9aaec750894f9"
+checksum = "cb39533d9d1c912123efd7d41d7e0c29d16917b60ce15b4c8d87cb1af7f67520"
 dependencies = [
  "futures-channel",
  "futures-executor",
@@ -2422,9 +2414,9 @@ checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf"
 
 [[package]]
 name = "rand"
-version = "0.8.7"
+version = "0.8.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "22f6172bdec972074665ed81ed53b71da00bfc44b65a753cfde883ec4c702a1a"
+checksum = "e058c7de0b26af77780c769414d6257830bb240f3c38477dbc2c16e5f54d6d4c"
 dependencies = [
  "rand_chacha 0.3.1",
  "rand_core 0.6.4",
@@ -3152,7 +3144,7 @@ checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596"
 [[package]]
 name = "stackable-certs"
 version = "0.4.1"
-source = "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.118.0#bc6c84025c2dcc834b94bfb57ec72810ae5f5eb1"
+source = "git+https://github.com/stackabletech/operator-rs.git?branch=feat%2Fplatform-access#f3d2da737aa49086cbdaba806aaf2c8645251cfb"
 dependencies = [
  "const-oid",
  "ecdsa",
@@ -3173,6 +3165,18 @@ dependencies = [
  "zeroize",
 ]
 
+[[package]]
+name = "stackable-kafka-agent"
+version = "0.0.0-dev"
+dependencies = [
+ "anyhow",
+ "built",
+ "clap",
+ "stackable-operator",
+ "tokio",
+ "tracing",
+]
+
 [[package]]
 name = "stackable-kafka-operator"
 version = "0.0.0-dev"
@@ -3197,15 +3201,15 @@ dependencies = [
 
 [[package]]
 name = "stackable-operator"
-version = "0.118.0"
-source = "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.118.0#bc6c84025c2dcc834b94bfb57ec72810ae5f5eb1"
+version = "0.119.0"
+source = "git+https://github.com/stackabletech/operator-rs.git?branch=feat%2Fplatform-access#f3d2da737aa49086cbdaba806aaf2c8645251cfb"
 dependencies = [
  "base64 0.23.1",
  "clap",
  "const_format",
  "delegate",
  "dockerfile-parser",
- "educe 0.7.6",
+ "educe 0.8.1",
  "either",
  "futures",
  "http",
@@ -3242,7 +3246,7 @@ dependencies = [
 [[package]]
 name = "stackable-operator-derive"
 version = "0.3.1"
-source = "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.118.0#bc6c84025c2dcc834b94bfb57ec72810ae5f5eb1"
+source = "git+https://github.com/stackabletech/operator-rs.git?branch=feat%2Fplatform-access#f3d2da737aa49086cbdaba806aaf2c8645251cfb"
 dependencies = [
  "darling 0.24.1",
  "proc-macro2",
@@ -3253,7 +3257,7 @@ dependencies = [
 [[package]]
 name = "stackable-shared"
 version = "0.1.2"
-source = "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.118.0#bc6c84025c2dcc834b94bfb57ec72810ae5f5eb1"
+source = "git+https://github.com/stackabletech/operator-rs.git?branch=feat%2Fplatform-access#f3d2da737aa49086cbdaba806aaf2c8645251cfb"
 dependencies = [
  "jiff",
  "k8s-openapi",
@@ -3270,7 +3274,7 @@ dependencies = [
 [[package]]
 name = "stackable-telemetry"
 version = "0.6.5"
-source = "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.118.0#bc6c84025c2dcc834b94bfb57ec72810ae5f5eb1"
+source = "git+https://github.com/stackabletech/operator-rs.git?branch=feat%2Fplatform-access#f3d2da737aa49086cbdaba806aaf2c8645251cfb"
 dependencies = [
  "axum",
  "clap",
@@ -3294,7 +3298,7 @@ dependencies = [
 [[package]]
 name = "stackable-versioned"
 version = "0.11.1"
-source = "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.118.0#bc6c84025c2dcc834b94bfb57ec72810ae5f5eb1"
+source = "git+https://github.com/stackabletech/operator-rs.git?branch=feat%2Fplatform-access#f3d2da737aa49086cbdaba806aaf2c8645251cfb"
 dependencies = [
  "kube",
  "schemars",
@@ -3308,10 +3312,9 @@ dependencies = [
 [[package]]
 name = "stackable-versioned-macros"
 version = "0.11.1"
-source = "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.118.0#bc6c84025c2dcc834b94bfb57ec72810ae5f5eb1"
+source = "git+https://github.com/stackabletech/operator-rs.git?branch=feat%2Fplatform-access#f3d2da737aa49086cbdaba806aaf2c8645251cfb"
 dependencies = [
  "convert_case",
- "convert_case_extras",
  "darling 0.24.1",
  "indoc",
  "itertools 0.15.0",
@@ -3325,8 +3328,8 @@ dependencies = [
 
 [[package]]
 name = "stackable-webhook"
-version = "0.9.2"
-source = "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.118.0#bc6c84025c2dcc834b94bfb57ec72810ae5f5eb1"
+version = "0.10.0"
+source = "git+https://github.com/stackabletech/operator-rs.git?branch=feat%2Fplatform-access#f3d2da737aa49086cbdaba806aaf2c8645251cfb"
 dependencies = [
  "arc-swap",
  "async-trait",
@@ -3349,7 +3352,7 @@ dependencies = [
  "tokio",
  "tokio-rustls",
  "tower",
- "tower-http 0.7.0",
+ "tower-http 0.7.1",
  "tracing",
  "tracing-opentelemetry",
  "x509-cert",
@@ -3742,9 +3745,9 @@ dependencies = [
 
 [[package]]
 name = "tower-http"
-version = "0.7.0"
+version = "0.7.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b11f75e912b0c2be01b63d8cf8057b8c3f97cf34abb3d431a3a4c8675498e233"
+checksum = "08a05a66a4fdd61cbbe0a1d755ffe0ca6aba159dd4820936a0ff8a8278245b9c"
 dependencies = [
  "bitflags 2.13.1",
  "bytes",
@@ -3828,9 +3831,9 @@ dependencies = [
 
 [[package]]
 name = "tracing-opentelemetry"
-version = "0.33.0"
+version = "0.34.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "adbc64cba7137545b8044cb1fe9814f7aacf3c6b5f9b45be8bb5db538befdb26"
+checksum = "0a904802a1b902f43638b677ff2a650847e3b4404101b6c586d648e8c1e3e8fe"
 dependencies = [
  "js-sys",
  "opentelemetry",
diff --git a/Cargo.nix b/Cargo.nix
index 1b1ba51b..b76c6a05 100644
--- a/Cargo.nix
+++ b/Cargo.nix
@@ -38,23 +38,21 @@ rec {
   # "public" attributes that we attempt to keep stable with new versions of crate2nix.
   #
 
-  rootCrate = rec {
-    packageId = "stackable-kafka-operator";
 
-    # Use this attribute to refer to the derivation building your root crate package.
-    # You can override the features with rootCrate.build.override { features = [ "default" "feature1" ... ]; }.
-    build = internal.buildRustCrateWithFeatures {
-      inherit packageId;
-    };
-
-    # Debug support which might change between releases.
-    # File a bug if you depend on any for non-debug work!
-    debug = internal.debugCrate { inherit packageId; };
-  };
   # Refer your crate build derivation by name here.
   # You can override the features with
   # workspaceMembers."${crateName}".build.override { features = [ "default" "feature1" ... ]; }.
   workspaceMembers = {
+    "stackable-kafka-agent" = rec {
+      packageId = "stackable-kafka-agent";
+      build = internal.buildRustCrateWithFeatures {
+        packageId = "stackable-kafka-agent";
+      };
+
+      # Debug support which might change between releases.
+      # File a bug if you depend on any for non-debug work!
+      debug = internal.debugCrate { inherit packageId; };
+    };
     "stackable-kafka-operator" = rec {
       packageId = "stackable-kafka-operator";
       build = internal.buildRustCrateWithFeatures {
@@ -1378,9 +1376,9 @@ rec {
       };
       "convert_case" = rec {
         crateName = "convert_case";
-        version = "0.11.0";
+        version = "0.12.0";
         edition = "2021";
-        sha256 = "0jfv1ajyr65bjlx533n5alfkfjdl8ks4zxfywdiz1jnj1qcz1yxg";
+        sha256 = "06zxk7w9jwlsrp4nvlalzckcpycy7f5wgj6zx99bym1lygqhkxqs";
         authors = [
           "rutrum <dave@rutrum.net>"
         ];
@@ -1392,25 +1390,6 @@ rec {
         ];
 
       };
-      "convert_case_extras" = rec {
-        crateName = "convert_case_extras";
-        version = "0.2.0";
-        edition = "2021";
-        sha256 = "1fyfc5vdblw15k8w7xahmif7bmslx3mdamvmg0brvapqzbq7172q";
-        authors = [
-          "rutrum <dave@rutrum.net>"
-        ];
-        dependencies = [
-          {
-            name = "convert_case";
-            packageId = "convert_case";
-          }
-        ];
-        features = {
-          "rand" = [ "dep:rand" ];
-          "random" = [ "rand" ];
-        };
-      };
       "core-foundation" = rec {
         crateName = "core-foundation";
         version = "0.10.1";
@@ -2380,11 +2359,11 @@ rec {
         };
         resolvedDefaultFeatures = [ "Clone" "Debug" "Hash" "PartialEq" ];
       };
-      "educe 0.7.6" = rec {
+      "educe 0.8.1" = rec {
         crateName = "educe";
-        version = "0.7.6";
+        version = "0.8.1";
         edition = "2024";
-        sha256 = "1par14babv60xrk4vbdnibfzv43frvpz2jv06iif3v4dvp4gllg4";
+        sha256 = "0vg925v5bfkr2aynhfshlhk21j230fhpv6ingfz5p0fjz3wb3g8y";
         procMacro = true;
         authors = [
           "Magic Len <len@magiclen.org>"
@@ -2407,13 +2386,7 @@ rec {
           {
             name = "syn";
             packageId = "syn 3.0.4";
-          }
-        ];
-        devDependencies = [
-          {
-            name = "syn";
-            packageId = "syn 3.0.4";
-            features = [ "full" ];
+            features = [ "visit" "visit-mut" ];
           }
         ];
         features = {
@@ -5412,7 +5385,7 @@ rec {
         workspace_member = null;
         src = pkgs.fetchgit {
           url = "https://github.com/stackabletech/operator-rs.git";
-          rev = "bc6c84025c2dcc834b94bfb57ec72810ae5f5eb1";
+          rev = "f3d2da737aa49086cbdaba806aaf2c8645251cfb";
           sha256 = "0cgziqra8097hp05ynib1qpw4c95n972f4w2rk9l3llyp8r1vmci";
         };
         libName = "k8s_version";
@@ -6461,7 +6434,7 @@ rec {
           }
           {
             name = "rand";
-            packageId = "rand 0.8.7";
+            packageId = "rand 0.8.8";
             optional = true;
             usesDefaultFeatures = false;
           }
@@ -6480,7 +6453,7 @@ rec {
         devDependencies = [
           {
             name = "rand";
-            packageId = "rand 0.8.7";
+            packageId = "rand 0.8.8";
             features = [ "small_rng" ];
           }
         ];
@@ -6627,9 +6600,9 @@ rec {
       };
       "opentelemetry" = rec {
         crateName = "opentelemetry";
-        version = "0.32.0";
+        version = "0.33.0";
         edition = "2021";
-        sha256 = "10ln14d1jgc8rvw97mblc9blzcgpg1bimim4d170b7ia4mijq55h";
+        sha256 = "1db15c0f4csj7v8zkbd9h3i503divafysd5l64rxpfby4qdhpnvc";
         dependencies = [
           {
             name = "futures-core";
@@ -6681,9 +6654,9 @@ rec {
       };
       "opentelemetry-appender-tracing" = rec {
         crateName = "opentelemetry-appender-tracing";
-        version = "0.32.0";
+        version = "0.33.0";
         edition = "2021";
-        sha256 = "0dyq4myan64sl8wly02jx0gb3jjz7575mn3w8rpphz0xvkq8001c";
+        sha256 = "1ns05asawxbi8cra6s3jnsbx6z5gg9fcm6ba6sx2c4ziv3l09cz2";
         libName = "opentelemetry_appender_tracing";
         dependencies = [
           {
@@ -6732,9 +6705,9 @@ rec {
       };
       "opentelemetry-http" = rec {
         crateName = "opentelemetry-http";
-        version = "0.32.0";
+        version = "0.33.0";
         edition = "2021";
-        sha256 = "0ca3drvm4fx5nskl7yn42dimy3bg35ppzc85y1p27pz215fh30sn";
+        sha256 = "003kp3rsb332b77yma9qza3rdnj4h2y17r81gsgj015ap0jkcb7f";
         libName = "opentelemetry_http";
         dependencies = [
           {
@@ -6770,16 +6743,15 @@ rec {
           "internal-logs" = [ "opentelemetry/internal-logs" ];
           "reqwest" = [ "dep:reqwest" ];
           "reqwest-blocking" = [ "dep:reqwest" "reqwest/blocking" ];
-          "reqwest-rustls" = [ "dep:reqwest" "reqwest/default-tls" ];
-          "reqwest-rustls-webpki-roots" = [ "dep:reqwest" "reqwest/default-tls" "reqwest/webpki-roots" ];
+          "reqwest-rustls" = [ "dep:reqwest" "reqwest/rustls" ];
         };
         resolvedDefaultFeatures = [ "reqwest" "reqwest-blocking" ];
       };
       "opentelemetry-otlp" = rec {
         crateName = "opentelemetry-otlp";
-        version = "0.32.0";
+        version = "0.33.0";
         edition = "2021";
-        sha256 = "0d9cys2flpidfxbr6h1103hjc633cax47ihnqgbj0xnicscr4rlr";
+        sha256 = "0i534gxld0zh27z8xpx5kwhqnhky2nd0bfam34ijm5i1bqs6g6k9";
         libName = "opentelemetry_otlp";
         dependencies = [
           {
@@ -6789,6 +6761,11 @@ rec {
             usesDefaultFeatures = false;
             features = [ "std" ];
           }
+          {
+            name = "httpdate";
+            packageId = "httpdate";
+            optional = true;
+          }
           {
             name = "opentelemetry";
             packageId = "opentelemetry";
@@ -6831,7 +6808,7 @@ rec {
             packageId = "tokio";
             optional = true;
             usesDefaultFeatures = false;
-            features = [ "sync" "rt" ];
+            features = [ "sync" "rt" "time" ];
           }
           {
             name = "tonic";
@@ -6867,17 +6844,15 @@ rec {
         ];
         features = {
           "default" = [ "http-proto" "reqwest-blocking-client" "trace" "metrics" "logs" "internal-logs" ];
-          "experimental-grpc-retry" = [ "grpc-tonic" "opentelemetry_sdk/experimental_async_runtime" "opentelemetry_sdk/rt-tokio" ];
-          "experimental-http-retry" = [ "opentelemetry_sdk/experimental_async_runtime" "opentelemetry_sdk/rt-tokio" "tokio" "httpdate" ];
           "flate2" = [ "dep:flate2" ];
           "grpc-tonic" = [ "tonic" "tonic-types" "prost" "http" "tokio" "opentelemetry-proto/gen-tonic" ];
           "gzip-http" = [ "flate2" ];
           "gzip-tonic" = [ "tonic/gzip" ];
           "http" = [ "dep:http" ];
-          "http-json" = [ "serde_json" "prost" "opentelemetry-http" "opentelemetry-proto/gen-tonic-messages" "opentelemetry-proto/with-serde" "http" "trace" "metrics" ];
-          "http-proto" = [ "prost" "opentelemetry-http" "opentelemetry-proto/gen-tonic-messages" "http" "trace" "metrics" ];
+          "http-json" = [ "serde_json" "prost" "opentelemetry-http" "opentelemetry-proto/gen-tonic-messages" "opentelemetry-proto/with-serde" "http" "httpdate" "trace" "metrics" ];
+          "http-proto" = [ "prost" "opentelemetry-http" "opentelemetry-proto/gen-tonic-messages" "http" "httpdate" "trace" "metrics" ];
           "httpdate" = [ "dep:httpdate" ];
-          "hyper-client" = [ "opentelemetry-http/hyper" ];
+          "hyper-client" = [ "opentelemetry-http/hyper" "tokio" ];
           "integration-testing" = [ "tonic" "prost" "tokio/full" "trace" "logs" ];
           "internal-logs" = [ "opentelemetry_sdk/internal-logs" "opentelemetry/internal-logs" ];
           "logs" = [ "opentelemetry/logs" "opentelemetry_sdk/logs" "opentelemetry-proto/logs" ];
@@ -6886,13 +6861,9 @@ rec {
           "prost" = [ "dep:prost" ];
           "reqwest" = [ "dep:reqwest" ];
           "reqwest-blocking-client" = [ "reqwest/blocking" "opentelemetry-http/reqwest-blocking" ];
-          "reqwest-client" = [ "reqwest" "opentelemetry-http/reqwest" ];
+          "reqwest-client" = [ "reqwest" "opentelemetry-http/reqwest" "tokio" ];
           "reqwest-rustls" = [ "reqwest" "opentelemetry-http/reqwest-rustls" ];
-          "reqwest-rustls-webpki-roots" = [ "reqwest" "opentelemetry-http/reqwest-rustls-webpki-roots" ];
-          "serde" = [ "dep:serde" ];
           "serde_json" = [ "dep:serde_json" ];
-          "serialize" = [ "serde" "serde_json" ];
-          "tls" = [ "tls-ring" ];
           "tls-aws-lc" = [ "tonic/tls-aws-lc" ];
           "tls-provider-agnostic" = [ "tonic/_tls-any" ];
           "tls-ring" = [ "tonic/tls-ring" ];
@@ -6906,13 +6877,13 @@ rec {
           "zstd-http" = [ "zstd" ];
           "zstd-tonic" = [ "tonic/zstd" ];
         };
-        resolvedDefaultFeatures = [ "default" "grpc-tonic" "gzip-tonic" "http" "http-proto" "internal-logs" "logs" "metrics" "opentelemetry-http" "prost" "reqwest" "reqwest-blocking-client" "tokio" "tonic" "tonic-types" "trace" ];
+        resolvedDefaultFeatures = [ "default" "grpc-tonic" "gzip-tonic" "http" "http-proto" "httpdate" "internal-logs" "logs" "metrics" "opentelemetry-http" "prost" "reqwest" "reqwest-blocking-client" "tokio" "tonic" "tonic-types" "trace" ];
       };
       "opentelemetry-proto" = rec {
         crateName = "opentelemetry-proto";
-        version = "0.32.0";
+        version = "0.33.0";
         edition = "2021";
-        sha256 = "0f5ny4rpnpq6q5q34b8k2q548rf31rpbxkwjqjwzfqxg3yx5imjn";
+        sha256 = "0pgnw8h6nb50jind8npn821szb8mhwsf0zppswwczv0a3b0imni5";
         libName = "opentelemetry_proto";
         dependencies = [
           {
@@ -6976,9 +6947,9 @@ rec {
       };
       "opentelemetry-semantic-conventions" = rec {
         crateName = "opentelemetry-semantic-conventions";
-        version = "0.32.1";
+        version = "0.33.0";
         edition = "2021";
-        sha256 = "0izyyi148774fndrdgcfwxx68l9y2ifn5x2mw8g6clf4lqbsq4y9";
+        sha256 = "11vvl60n1bdph9jypfhzrwi91pa8y9g663wgpcphb41mpkhv326j";
         libName = "opentelemetry_semantic_conventions";
         features = {
         };
@@ -6986,21 +6957,24 @@ rec {
       };
       "opentelemetry_sdk" = rec {
         crateName = "opentelemetry_sdk";
-        version = "0.32.1";
+        version = "0.33.0";
         edition = "2021";
-        sha256 = "1ycl11syranrinhgn4c2hlzhyzyvpa06ryxq5mxgzmf4387ghncv";
+        sha256 = "083myvvimjw7im65pq8cnqbnkl991iz1vm6pxwij348wklym6ffb";
         dependencies = [
           {
             name = "futures-channel";
             packageId = "futures-channel";
+            optional = true;
           }
           {
             name = "futures-executor";
             packageId = "futures-executor";
+            optional = true;
           }
           {
             name = "futures-util";
             packageId = "futures-util";
+            optional = true;
             usesDefaultFeatures = false;
             features = [ "std" "sink" "async-await-macro" ];
           }
@@ -7031,6 +7005,7 @@ rec {
           {
             name = "thiserror";
             packageId = "thiserror 2.0.20";
+            optional = true;
             usesDefaultFeatures = false;
           }
           {
@@ -7055,6 +7030,7 @@ rec {
         ];
         features = {
           "default" = [ "trace" "metrics" "logs" "internal-logs" ];
+          "experimental_async_runtime" = [ "dep:futures-channel" "dep:futures-executor" "dep:futures-util" "dep:thiserror" ];
           "experimental_logs_batch_log_processor_with_async_runtime" = [ "logs" "experimental_async_runtime" ];
           "experimental_metrics_bound_instruments" = [ "metrics" "opentelemetry/experimental_metrics_bound_instruments" ];
           "experimental_metrics_custom_reader" = [ "metrics" ];
@@ -7064,8 +7040,8 @@ rec {
           "http" = [ "dep:http" ];
           "internal-logs" = [ "opentelemetry/internal-logs" ];
           "jaeger_remote_sampler" = [ "trace" "opentelemetry-http" "http" "serde" "serde_json" "url" "experimental_async_runtime" ];
-          "logs" = [ "opentelemetry/logs" ];
-          "metrics" = [ "opentelemetry/metrics" ];
+          "logs" = [ "opentelemetry/logs" "dep:futures-channel" "dep:futures-executor" "dep:futures-util" ];
+          "metrics" = [ "opentelemetry/metrics" "dep:futures-channel" "dep:futures-executor" "dep:futures-util" "dep:thiserror" ];
           "opentelemetry-http" = [ "dep:opentelemetry-http" ];
           "percent-encoding" = [ "dep:percent-encoding" ];
           "rand" = [ "dep:rand" ];
@@ -7077,7 +7053,7 @@ rec {
           "testing" = [ "opentelemetry/testing" "trace" "metrics" "logs" "tokio/sync" ];
           "tokio" = [ "dep:tokio" ];
           "tokio-stream" = [ "dep:tokio-stream" ];
-          "trace" = [ "opentelemetry/trace" "rand" "percent-encoding" ];
+          "trace" = [ "opentelemetry/trace" "rand" "percent-encoding" "dep:futures-channel" "dep:futures-executor" "dep:futures-util" "dep:thiserror" ];
           "url" = [ "dep:url" ];
         };
         resolvedDefaultFeatures = [ "default" "experimental_async_runtime" "internal-logs" "logs" "metrics" "percent-encoding" "rand" "rt-tokio" "tokio" "tokio-stream" "trace" ];
@@ -7903,11 +7879,11 @@ rec {
         };
         resolvedDefaultFeatures = [ "alloc" "default" "std" "std_rng" "sys_rng" "thread_rng" ];
       };
-      "rand 0.8.7" = rec {
+      "rand 0.8.8" = rec {
         crateName = "rand";
-        version = "0.8.7";
+        version = "0.8.8";
         edition = "2018";
-        sha256 = "06iaf16fr0z8zly7anmn8ky0p80xnx9yv0gdcm30fwn9vqmigxi2";
+        sha256 = "0k3d9psya5icpiylff1w1wjbnc3q4pb1953n1iw7gbr61ggcfn70";
         authors = [
           "The Rand Project Developers"
           "The Rust Project Developers"
@@ -10327,7 +10303,7 @@ rec {
         workspace_member = null;
         src = pkgs.fetchgit {
           url = "https://github.com/stackabletech/operator-rs.git";
-          rev = "bc6c84025c2dcc834b94bfb57ec72810ae5f5eb1";
+          rev = "f3d2da737aa49086cbdaba806aaf2c8645251cfb";
           sha256 = "0cgziqra8097hp05ynib1qpw4c95n972f4w2rk9l3llyp8r1vmci";
         };
         libName = "stackable_certs";
@@ -10423,6 +10399,54 @@ rec {
         };
         resolvedDefaultFeatures = [ "default" "rustls" ];
       };
+      "stackable-kafka-agent" = rec {
+        crateName = "stackable-kafka-agent";
+        version = "0.0.0-dev";
+        edition = "2024";
+        crateBin = [
+          {
+            name = "stackable-kafka-agent";
+            path = "src/main.rs";
+            requiredFeatures = [ ];
+          }
+        ];
+        src = lib.cleanSourceWith { filter = sourceFilter;  src = ./rust/agent-binary; };
+        authors = [
+          "Stackable GmbH <info@stackable.tech>"
+        ];
+        dependencies = [
+          {
+            name = "anyhow";
+            packageId = "anyhow";
+          }
+          {
+            name = "clap";
+            packageId = "clap";
+          }
+          {
+            name = "stackable-operator";
+            packageId = "stackable-operator";
+            features = [ "crds" "webhook" ];
+          }
+          {
+            name = "tokio";
+            packageId = "tokio";
+            features = [ "full" ];
+          }
+          {
+            name = "tracing";
+            packageId = "tracing";
+          }
+        ];
+        buildDependencies = [
+          {
+            name = "built";
+            packageId = "built";
+            features = [ "chrono" "git2" ];
+          }
+        ];
+
+      };
       "stackable-kafka-operator" = rec {
         crateName = "stackable-kafka-operator";
         version = "0.0.0-dev";
@@ -10517,12 +10541,12 @@ rec {
       };
       "stackable-operator" = rec {
         crateName = "stackable-operator";
-        version = "0.118.0";
+        version = "0.119.0";
         edition = "2024";
         workspace_member = null;
         src = pkgs.fetchgit {
           url = "https://github.com/stackabletech/operator-rs.git";
-          rev = "bc6c84025c2dcc834b94bfb57ec72810ae5f5eb1";
+          rev = "f3d2da737aa49086cbdaba806aaf2c8645251cfb";
           sha256 = "0cgziqra8097hp05ynib1qpw4c95n972f4w2rk9l3llyp8r1vmci";
         };
         libName = "stackable_operator";
@@ -10553,7 +10577,7 @@ rec {
           }
           {
             name = "educe";
-            packageId = "educe 0.7.6";
+            packageId = "educe 0.8.1";
             usesDefaultFeatures = false;
             features = [ "Clone" "Debug" "Default" "PartialEq" "Eq" ];
           }
@@ -10721,7 +10745,7 @@ rec {
         workspace_member = null;
         src = pkgs.fetchgit {
           url = "https://github.com/stackabletech/operator-rs.git";
-          rev = "bc6c84025c2dcc834b94bfb57ec72810ae5f5eb1";
+          rev = "f3d2da737aa49086cbdaba806aaf2c8645251cfb";
           sha256 = "0cgziqra8097hp05ynib1qpw4c95n972f4w2rk9l3llyp8r1vmci";
         };
         procMacro = true;
@@ -10756,7 +10780,7 @@ rec {
         workspace_member = null;
         src = pkgs.fetchgit {
           url = "https://github.com/stackabletech/operator-rs.git";
-          rev = "bc6c84025c2dcc834b94bfb57ec72810ae5f5eb1";
+          rev = "f3d2da737aa49086cbdaba806aaf2c8645251cfb";
           sha256 = "0cgziqra8097hp05ynib1qpw4c95n972f4w2rk9l3llyp8r1vmci";
         };
         libName = "stackable_shared";
@@ -10837,7 +10861,7 @@ rec {
         workspace_member = null;
         src = pkgs.fetchgit {
           url = "https://github.com/stackabletech/operator-rs.git";
-          rev = "bc6c84025c2dcc834b94bfb57ec72810ae5f5eb1";
+          rev = "f3d2da737aa49086cbdaba806aaf2c8645251cfb";
           sha256 = "0cgziqra8097hp05ynib1qpw4c95n972f4w2rk9l3llyp8r1vmci";
         };
         libName = "stackable_telemetry";
@@ -10947,7 +10971,7 @@ rec {
         workspace_member = null;
         src = pkgs.fetchgit {
           url = "https://github.com/stackabletech/operator-rs.git";
-          rev = "bc6c84025c2dcc834b94bfb57ec72810ae5f5eb1";
+          rev = "f3d2da737aa49086cbdaba806aaf2c8645251cfb";
           sha256 = "0cgziqra8097hp05ynib1qpw4c95n972f4w2rk9l3llyp8r1vmci";
         };
         libName = "stackable_versioned";
@@ -10997,7 +11021,7 @@ rec {
         workspace_member = null;
         src = pkgs.fetchgit {
           url = "https://github.com/stackabletech/operator-rs.git";
-          rev = "bc6c84025c2dcc834b94bfb57ec72810ae5f5eb1";
+          rev = "f3d2da737aa49086cbdaba806aaf2c8645251cfb";
           sha256 = "0cgziqra8097hp05ynib1qpw4c95n972f4w2rk9l3llyp8r1vmci";
         };
         procMacro = true;
@@ -11010,10 +11034,6 @@ rec {
             name = "convert_case";
             packageId = "convert_case";
           }
-          {
-            name = "convert_case_extras";
-            packageId = "convert_case_extras";
-          }
           {
             name = "darling";
             packageId = "darling 0.24.1";
@@ -11060,12 +11080,12 @@ rec {
       };
       "stackable-webhook" = rec {
         crateName = "stackable-webhook";
-        version = "0.9.2";
+        version = "0.10.0";
         edition = "2024";
         workspace_member = null;
         src = pkgs.fetchgit {
           url = "https://github.com/stackabletech/operator-rs.git";
-          rev = "bc6c84025c2dcc834b94bfb57ec72810ae5f5eb1";
+          rev = "f3d2da737aa49086cbdaba806aaf2c8645251cfb";
           sha256 = "0cgziqra8097hp05ynib1qpw4c95n972f4w2rk9l3llyp8r1vmci";
         };
         libName = "stackable_webhook";
@@ -11171,7 +11191,7 @@ rec {
           }
           {
             name = "tower-http";
-            packageId = "tower-http 0.7.0";
+            packageId = "tower-http 0.7.1";
             features = [ "trace" ];
           }
           {
@@ -11376,7 +11396,7 @@ rec {
           "proc-macro" = [ "proc-macro2/proc-macro" "quote?/proc-macro" ];
           "test" = [ "syn-test-suite/all-features" ];
         };
-        resolvedDefaultFeatures = [ "clone-impls" "default" "derive" "extra-traits" "full" "parsing" "printing" "proc-macro" "visit-mut" ];
+        resolvedDefaultFeatures = [ "clone-impls" "default" "derive" "extra-traits" "full" "parsing" "printing" "proc-macro" "visit" "visit-mut" ];
       };
       "sync_wrapper" = rec {
         crateName = "sync_wrapper";
@@ -12542,11 +12562,11 @@ rec {
         };
         resolvedDefaultFeatures = [ "auth" "base64" "default" "follow-redirect" "futures-util" "map-response-body" "mime" "tower" "trace" "tracing" "util" "validate-request" ];
       };
-      "tower-http 0.7.0" = rec {
+      "tower-http 0.7.1" = rec {
         crateName = "tower-http";
-        version = "0.7.0";
+        version = "0.7.1";
         edition = "2018";
-        sha256 = "0cz2k1a6gj54lcqx9cxb6k7rfgwcgc2zi31xnq0vxhmh2blpa7xi";
+        sha256 = "172v4iw852pzl0v0k0nlklavlsnaw3zmbmx1w2xirmpxlik5m808";
         libName = "tower_http";
         authors = [
           "Tower Maintainers <team@tower-rs.com>"
@@ -12844,9 +12864,9 @@ rec {
       };
       "tracing-opentelemetry" = rec {
         crateName = "tracing-opentelemetry";
-        version = "0.33.0";
+        version = "0.34.0";
         edition = "2021";
-        sha256 = "09nvxy5m7nxmifz4b6szdcyczapp2jcgxcac0jw4ax8klz5n9g5d";
+        sha256 = "1zp8wg0yhj6nhv2vc0a182sf6iq8clmgyxxn70vg80mrl414i40a";
         libName = "tracing_opentelemetry";
         dependencies = [
           {
diff --git a/Cargo.toml b/Cargo.toml
index 2eac6bf5..8f9321a8 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -1,5 +1,5 @@
 [workspace]
-members = ["rust/operator-binary"]
+members = ["rust/operator-binary", "rust/agent-binary"]
 resolver = "2"
 
 [workspace.package]
@@ -10,7 +10,7 @@ edition = "2024"
 repository = "https://github.com/stackabletech/kafka-operator"
 
 [workspace.dependencies]
-stackable-operator = { git = "https://github.com/stackabletech/operator-rs.git", tag = "stackable-operator-0.118.0", features = ["crds", "webhook"] }
+stackable-operator = { git = "https://github.com/stackabletech/operator-rs.git", branch = "feat/platform-access", features = ["crds", "webhook"] }
 
 anyhow = "1.0"
 built = { version = "0.8", features = ["chrono", "git2"] }
diff --git a/Tiltfile b/Tiltfile
index 1769d19a..fba0c44f 100644
--- a/Tiltfile
+++ b/Tiltfile
@@ -34,7 +34,22 @@ custom_build(
 # We need to set the correct image annotation on the operator Deployment to use e.g.
 # oci.stackable.tech/sandbox/opa-operator:7y19m3d8clwxlv34v5q2x4p7v536s00g instead of
 # oci.stackable.tech/sandbox/opa-operator:0.0.0-dev (which does not exist)
-k8s_kind('Deployment', image_json_path='{.spec.template.metadata.annotations.internal\\.stackable\\.tech/image}')
+deployment_image_json_paths = ['{.spec.template.metadata.annotations.internal\\.stackable\\.tech/image}']
+
+# Operators that ship an agent (see `agent` in nix/meta.json) also build the agent image and pass
+# it to the operator via the `internal.stackable.tech/agent-image` annotation.
+if 'agent' in meta:
+    agent_image_name = operator_repository + '/' + meta['agent']['name']
+    custom_build(
+        agent_image_name,
+        'nix-build . -A dockerAgent --argstr dockerNameAgent "' + agent_image_name + '" -o result-agent && ./result-agent/load-image | docker load',
+        deps=['rust', 'Cargo.toml', 'Cargo.lock', 'default.nix', "nix", 'build.rs', 'vendor'],
+        ignore=['*.~undo-tree~'],
+        outputs_image_ref_to='result-agent/ref',
+    )
+    deployment_image_json_paths.append('{.spec.template.metadata.annotations.internal\\.stackable\\.tech/agent-image}')
+
+k8s_kind('Deployment', image_json_path=deployment_image_json_paths)
 k8s_kind('DaemonSet', image_json_path='{.spec.template.metadata.annotations.internal\\.stackable\\.tech/image}')
 
 # Optionally specify a custom Helm values file to be passed to the Helm deployment below.
diff --git a/crate-hashes.json b/crate-hashes.json
index 8357dcf2..e4f5f7b1 100644
--- a/crate-hashes.json
+++ b/crate-hashes.json
@@ -1,11 +1,11 @@
 {
-  "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.118.0#k8s-version@0.1.3": "0cgziqra8097hp05ynib1qpw4c95n972f4w2rk9l3llyp8r1vmci",
-  "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.118.0#stackable-certs@0.4.1": "0cgziqra8097hp05ynib1qpw4c95n972f4w2rk9l3llyp8r1vmci",
-  "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.118.0#stackable-operator-derive@0.3.1": "0cgziqra8097hp05ynib1qpw4c95n972f4w2rk9l3llyp8r1vmci",
-  "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.118.0#stackable-operator@0.118.0": "0cgziqra8097hp05ynib1qpw4c95n972f4w2rk9l3llyp8r1vmci",
-  "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.118.0#stackable-shared@0.1.2": "0cgziqra8097hp05ynib1qpw4c95n972f4w2rk9l3llyp8r1vmci",
-  "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.118.0#stackable-telemetry@0.6.5": "0cgziqra8097hp05ynib1qpw4c95n972f4w2rk9l3llyp8r1vmci",
-  "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.118.0#stackable-versioned-macros@0.11.1": "0cgziqra8097hp05ynib1qpw4c95n972f4w2rk9l3llyp8r1vmci",
-  "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.118.0#stackable-versioned@0.11.1": "0cgziqra8097hp05ynib1qpw4c95n972f4w2rk9l3llyp8r1vmci",
-  "git+https://github.com/stackabletech/operator-rs.git?tag=stackable-operator-0.118.0#stackable-webhook@0.9.2": "0cgziqra8097hp05ynib1qpw4c95n972f4w2rk9l3llyp8r1vmci"
+  "git+https://github.com/stackabletech/operator-rs.git?branch=feat%2Fplatform-access#k8s-version@0.1.3": "0cgziqra8097hp05ynib1qpw4c95n972f4w2rk9l3llyp8r1vmci",
+  "git+https://github.com/stackabletech/operator-rs.git?branch=feat%2Fplatform-access#stackable-certs@0.4.1": "0cgziqra8097hp05ynib1qpw4c95n972f4w2rk9l3llyp8r1vmci",
+  "git+https://github.com/stackabletech/operator-rs.git?branch=feat%2Fplatform-access#stackable-operator-derive@0.3.1": "0cgziqra8097hp05ynib1qpw4c95n972f4w2rk9l3llyp8r1vmci",
+  "git+https://github.com/stackabletech/operator-rs.git?branch=feat%2Fplatform-access#stackable-operator@0.119.0": "0cgziqra8097hp05ynib1qpw4c95n972f4w2rk9l3llyp8r1vmci",
+  "git+https://github.com/stackabletech/operator-rs.git?branch=feat%2Fplatform-access#stackable-shared@0.1.2": "0cgziqra8097hp05ynib1qpw4c95n972f4w2rk9l3llyp8r1vmci",
+  "git+https://github.com/stackabletech/operator-rs.git?branch=feat%2Fplatform-access#stackable-telemetry@0.6.5": "0cgziqra8097hp05ynib1qpw4c95n972f4w2rk9l3llyp8r1vmci",
+  "git+https://github.com/stackabletech/operator-rs.git?branch=feat%2Fplatform-access#stackable-versioned-macros@0.11.1": "0cgziqra8097hp05ynib1qpw4c95n972f4w2rk9l3llyp8r1vmci",
+  "git+https://github.com/stackabletech/operator-rs.git?branch=feat%2Fplatform-access#stackable-versioned@0.11.1": "0cgziqra8097hp05ynib1qpw4c95n972f4w2rk9l3llyp8r1vmci",
+  "git+https://github.com/stackabletech/operator-rs.git?branch=feat%2Fplatform-access#stackable-webhook@0.10.0": "0cgziqra8097hp05ynib1qpw4c95n972f4w2rk9l3llyp8r1vmci"
 }
\ No newline at end of file
diff --git a/default.nix b/default.nix
index 6feb190a..abaa026b 100644
--- a/default.nix
+++ b/default.nix
@@ -101,6 +101,8 @@
   }
 , meta ? pkgsLocal.lib.importJSON ./nix/meta.json
 , dockerName ? "oci.stackable.tech/sandbox/${meta.operator.name}"
+# Only used by operators that ship an agent (see `agent` in nix/meta.json)
+, dockerNameAgent ? "oci.stackable.tech/sandbox/${meta.agent.name}"
 , dockerTag ? null
 # Controls the amount of debug information included in the built operator binaries,
 # see https://doc.rust-lang.org/rustc/codegen-options/index.html#debuginfo
@@ -125,7 +127,11 @@ rec {
   inherit cargo sources pkgsLocal pkgsTarget meta;
   inherit (pkgsLocal) lib;
   pkgs = lib.warn "pkgs is not cross-compilation-aware, explicitly use either pkgsLocal or pkgsTarget" pkgsLocal;
-  build = cargo.allWorkspaceMembers;
+  # Operators that ship an agent (see `agent` in nix/meta.json) keep the agent binary out of the
+  # operator image, as it has its own image (`dockerAgent`).
+  build = if meta ? agent
+    then cargo.workspaceMembers."stackable-${meta.operator.name}".build
+    else cargo.allWorkspaceMembers;
   entrypoint = build+"/bin/stackable-${meta.operator.name}";
   # Run crds in the target environment, to avoid compiling everything twice
   crds = pkgsTarget.runCommand "${meta.operator.name}-crds.yaml" {}
@@ -143,13 +149,13 @@ rec {
   # build it in the local environment so that the generated load-image
   # can run locally.
   # That's still fine, as long as we only refer to pkgsTarget *inside* of the image.
-  dockerImage = pkgsLocal.dockerTools.streamLayeredImage {
+  dockerImageArgs = {
     name = dockerName;
     tag = dockerTag;
     contents = [
       # Kerberos 5 must be installed globally to load plugins correctly
       pkgsTarget.krb5
-      # Make the whole cargo workspace available on $PATH
+      # Make the operator binaries available on $PATH
       build
     ] ++ lib.optional includeShell [
       pkgsTarget.bashInteractive
@@ -189,6 +195,7 @@ rec {
       User = toString stackableUserUid;
     };
   };
+  dockerImage = pkgsLocal.dockerTools.streamLayeredImage dockerImageArgs;
   docker = pkgsLocal.linkFarm "${dockerImage.name}-docker" [
     {
       name = "load-image";
@@ -212,6 +219,29 @@ rec {
     }
   ];
 
+  # The image of the agent, only built for operators that ship one.
+  dockerAgent = if meta ? agent then
+    let
+      agentBuild = cargo.workspaceMembers."stackable-${meta.agent.name}".build;
+      agentImage = pkgsLocal.dockerTools.streamLayeredImage (dockerImageArgs // {
+        name = dockerNameAgent;
+        contents = [ agentBuild ] ++ lib.optional includeShell [
+          pkgsTarget.bashInteractive
+          pkgsTarget.coreutils
+          pkgsTarget.util-linuxMinimal
+        ];
+        config = dockerImageArgs.config // {
+          Entrypoint = [ "${agentBuild}/bin/stackable-${meta.agent.name}" ];
+        };
+      });
+    in pkgsLocal.linkFarm "${agentImage.name}-docker" [
+      { name = "load-image"; path = agentImage; }
+      { name = "ref"; path = pkgsLocal.writeText "${agentImage.name}-image-tag" "${agentImage.imageName}:${agentImage.imageTag}"; }
+      { name = "image-repo"; path = pkgsLocal.writeText "${agentImage.name}-repo" agentImage.imageName; }
+      { name = "image-tag"; path = pkgsLocal.writeText "${agentImage.name}-tag" agentImage.imageTag; }
+    ]
+  else null;
+
   # need to use vendored crate2nix because of https://github.com/kolloch/crate2nix/issues/264
   crate2nix = import sources.crate2nix { pkgs = pkgsLocal; };
   tilt = pkgsLocal.tilt;
diff --git a/deploy/helm/kafka-operator/templates/_helpers.tpl b/deploy/helm/kafka-operator/templates/_helpers.tpl
index 9c61cd51..281aa549 100644
--- a/deploy/helm/kafka-operator/templates/_helpers.tpl
+++ b/deploy/helm/kafka-operator/templates/_helpers.tpl
@@ -5,6 +5,14 @@ Expand the name of the chart.
 {{- default .Chart.Name .Values.nameOverride | trunc 63 | trimSuffix "-operator" }}
 {{- end }}
 
+{{/*
+Fixed product name. Unlike kafka-operator.name it ignores nameOverride, because it must match
+names hardcoded in the operator and the published image names.
+*/}}
+{{- define "kafka-operator.productName" -}}
+kafka
+{{- end }}
+
 {{/*
 Expand the name of the chart.
 */}}
@@ -84,3 +92,10 @@ Build the full operator container image reference.
 {{- define "kafka-operator.image" -}}
 {{- printf "%s/%s:%s" .Values.image.repository .Chart.Name (.Values.image.tag | default .Chart.AppVersion) -}}
 {{- end }}
+
+{{/*
+Build the full agent container image reference. The agent is released together with the operator.
+*/}}
+{{- define "kafka-operator.agentImage" -}}
+{{- printf "%s/%s-agent:%s" .Values.image.repository (include "kafka-operator.productName" .) (.Values.image.tag | default .Chart.AppVersion) -}}
+{{- end }}
diff --git a/deploy/helm/kafka-operator/templates/clusterrole-agent.yaml b/deploy/helm/kafka-operator/templates/clusterrole-agent.yaml
new file mode 100644
index 00000000..a1701506
--- /dev/null
+++ b/deploy/helm/kafka-operator/templates/clusterrole-agent.yaml
@@ -0,0 +1,21 @@
+---
+# Agent ClusterRole: bound (via per KafkaCluster RoleBinding) to the ServiceAccount that the
+# platform-access agent pod runs as.
+apiVersion: rbac.authorization.k8s.io/v1
+kind: ClusterRole
+metadata:
+  name: {{ include "kafka-operator.productName" . }}-agent-clusterrole
+  labels:
+  {{- include "kafka-operator.labels" . | nindent 4 }}
+rules:
+{{ if .Capabilities.APIVersions.Has "security.openshift.io/v1" }}
+  # On OpenShift, the agent pod must be allowed to use the nonroot-v2 SCC
+  - apiGroups:
+      - security.openshift.io
+    resources:
+      - securitycontextconstraints
+    resourceNames:
+      - nonroot-v2
+    verbs:
+      - use
+{{ end }}
diff --git a/deploy/helm/kafka-operator/templates/clusterrole-operator.yaml b/deploy/helm/kafka-operator/templates/clusterrole-operator.yaml
index 59a34e6c..cabc631e 100644
--- a/deploy/helm/kafka-operator/templates/clusterrole-operator.yaml
+++ b/deploy/helm/kafka-operator/templates/clusterrole-operator.yaml
@@ -41,7 +41,7 @@ rules:
       - list
       - patch
       - watch
-  # Required to bind the product ClusterRole to the per-cluster ServiceAccount.
+  # Required to bind the product and agent ClusterRoles to the per-cluster ServiceAccounts.
   - apiGroups:
       - rbac.authorization.k8s.io
     resources:
@@ -50,6 +50,7 @@ rules:
       - bind
     resourceNames:
       - {{ include "kafka-operator.name" . }}-clusterrole
+      - {{ include "kafka-operator.productName" . }}-agent-clusterrole
   # StatefulSet created per role group (broker, KRaft controller). Applied via
   # SSA, tracked for orphan cleanup, and owned by the controller.
   - apiGroups:
@@ -63,6 +64,18 @@ rules:
       - list
       - patch
       - watch
+  # Deployment of the platform-access agent.
+  - apiGroups:
+      - apps
+    resources:
+      - deployments
+    verbs:
+      - create
+      - delete
+      - get
+      - list
+      - patch
+      - watch
   # PodDisruptionBudget created per role group. Applied via SSA and tracked for orphan cleanup, and owned by the controller.
   - apiGroups:
       - policy
diff --git a/deploy/helm/kafka-operator/templates/deployment.yaml b/deploy/helm/kafka-operator/templates/deployment.yaml
index 6f7163dc..9fac4f33 100644
--- a/deploy/helm/kafka-operator/templates/deployment.yaml
+++ b/deploy/helm/kafka-operator/templates/deployment.yaml
@@ -16,6 +16,7 @@ spec:
     metadata:
       annotations:
         internal.stackable.tech/image: {{ include "kafka-operator.image" . }}
+        internal.stackable.tech/agent-image: {{ include "kafka-operator.agentImage" . }}
         {{- with .Values.podAnnotations }}
         {{- toYaml . | nindent 8 }}
         {{- end }}
@@ -70,6 +71,13 @@ spec:
             - name: IMAGE_REPOSITORY
               value: {{ .Values.image.productRepository | default .Values.image.repository }}
 
+            # The image of the agent the operator deploys alongside a cluster.
+            - name: AGENT_IMAGE
+              # Tilt can use annotations as image paths, but not env variables
+              valueFrom:
+                fieldRef:
+                  fieldPath: metadata.annotations['internal.stackable.tech/agent-image']
+
             # Operators need to know the node name they are running on, to e.g. discover the
             # Kubernetes domain name from the kubelet API.
             - name: KUBERNETES_NODE_NAME
diff --git a/docker/Dockerfile b/docker/Dockerfile
index 84458c3a..a96fc611 100644
--- a/docker/Dockerfile
+++ b/docker/Dockerfile
@@ -190,7 +190,7 @@ EOF
 
 COPY LICENSE /licenses/LICENSE
 
-COPY --from=builder --chown=${STACKABLE_USER_UID}:0 /app/* /usr/local/bin/
+COPY --from=builder --chown=${STACKABLE_USER_UID}:0 /app/stackable-kafka-operator* /app/stackable-src.tar.gz /usr/local/bin/
 
 USER ${STACKABLE_USER_UID}
 
diff --git a/docker/Dockerfile.agent b/docker/Dockerfile.agent
new file mode 100644
index 00000000..04f55b38
--- /dev/null
+++ b/docker/Dockerfile.agent
@@ -0,0 +1,158 @@
+# syntax=docker/dockerfile:1.16.0@sha256:e2dd261f92e4b763d789984f6eab84be66ab4f5f08052316d8eb8f173593acf7
+# NOTE: The syntax directive needs to be the first line in a Dockerfile
+# Find the latest versions here: https://hub.docker.com/r/docker/dockerfile/tags
+# And the changelogs: https://docs.docker.com/build/buildkit/dockerfile-release-notes/ or https://github.com/moby/buildkit/releases
+
+# https://docs.docker.com/build/checks/#fail-build-on-check-violations
+# check=error=true
+
+# We want to automatically use the latest. We also don't tag our images with a version.
+# hadolint ignore=DL3007
+FROM oci.stackable.tech/sdp/ubi10-rust-builder:latest AS builder
+
+
+# We want to automatically use the latest.
+# hadolint ignore=DL3007
+FROM registry.access.redhat.com/ubi10/ubi-minimal:latest AS agent
+
+ARG VERSION
+# NOTE (@Techassi): This is required for OpenShift/Red Hat certification
+# Keeping this as "1" seems to be fine since a couple of years /shrug
+ARG RELEASE="1"
+
+# These are chosen at random and are this high on purpose to have very little chance to clash with an existing user or group on the host system
+# NOTE: Please also update default.nix accordingly!
+ARG STACKABLE_USER_GID="574654813"
+ARG STACKABLE_USER_UID="782252253"
+ARG STACKABLE_USER_NAME="stackable"
+
+# Sets the default shell to Bash with strict error handling and robust pipeline processing.
+# "-e": Exits immediately if a command exits with a non-zero status
+# "-u": Treats unset variables as an error, preventing unexpected behavior from undefined variables.
+# "-o pipefail": Causes a pipeline to return the exit status of the last command in the pipe that failed, ensuring errors in any part of a pipeline are not ignored.
+# "-c": Allows the execution of commands passed as a string
+SHELL ["/bin/bash", "-euo", "pipefail", "-c"]
+
+# These labels have mostly been superseded by the OpenContainer spec annotations below but it doesn't hurt to include them
+# http://label-schema.org/rc1/
+LABEL name="Stackable Agent for Apache Kafka"
+LABEL maintainer="info@stackable.tech"
+LABEL vendor="Stackable GmbH"
+LABEL version="${VERSION}"
+LABEL release="${RELEASE}"
+LABEL summary="Manage resources inside Apache Kafka clusters."
+LABEL description="Manage resources inside Apache Kafka clusters."
+
+# Overwriting/Pinning UBI labels
+# https://github.com/projectatomic/ContainerApplicationGenericLabels
+LABEL vcs-ref=""
+LABEL distribution-scope="public"
+LABEL url="https://stackable.tech"
+ARG TARGETARCH
+LABEL architecture="${TARGETARCH}"
+LABEL com.redhat.component=""
+# It complains about it being an invalid label but RedHat uses it and we want to override it and it works....
+# hadolint ignore=DL3048
+LABEL com.redhat.license_terms=""
+LABEL io.buildah.version=""
+LABEL io.openshift.expose-services=""
+
+# https://github.com/opencontainers/image-spec/blob/64294bd7a2bf2537e1a6a34d687caae70300b0c4/annotations.md#annotations
+LABEL org.opencontainers.image.authors="info@stackable.tech"
+LABEL org.opencontainers.image.url="https://stackable.tech"
+LABEL org.opencontainers.image.vendor="Stackable GmbH"
+LABEL org.opencontainers.image.licenses="OSL-3.0"
+LABEL org.opencontainers.image.documentation="https://docs.stackable.tech/home/stable/kafka/"
+LABEL org.opencontainers.image.version="${VERSION}"
+LABEL org.opencontainers.image.revision="${RELEASE}"
+LABEL org.opencontainers.image.title="Stackable Agent for Apache Kafka"
+LABEL org.opencontainers.image.description="Manage resources inside Apache Kafka clusters."
+
+# https://docs.openshift.com/container-platform/4.16/openshift_images/create-images.html#defining-image-metadata
+# https://github.com/projectatomic/ContainerApplicationGenericLabels/blob/master/vendor/redhat/labels.md
+LABEL io.openshift.tags="ubi10,stackable,sdp,kafka"
+LABEL io.k8s.description="Manage resources inside Apache Kafka clusters."
+LABEL io.k8s.display-name="Stackable Agent for Apache Kafka"
+
+COPY <<EOF /etc/dnf/dnf.conf
+[main]
+install_weak_deps=0
+assumeyes=True
+tsflags=nodocs
+EOF
+
+# SC2028
+# echo won't expand escape sequences. Consider printf.
+# https://github.com/koalaman/shellcheck/wiki/SC2028
+# Reason: This is complaining about the complicated PS1 statement.
+# It seems to work as intended so I'm not going to touch it!
+#
+# SC3037
+# It complains about echo flags not being available in POSIX sh but we set the shell to bash
+# hadolint ignore=SC2028,SC3037
+RUN <<EOF
+microdnf update
+
+# The same debugging utilities as in the operator image, see docker/Dockerfile.
+# NOTE (@NickLarsenNZ): Maybe we should consider pinning package versions?
+# hadolint ignore=DL3041
+microdnf install \
+  findutils \
+  iputils \
+  less \
+  nano \
+  shadow-utils \
+  tar
+
+groupadd --gid ${STACKABLE_USER_GID} --system ${STACKABLE_USER_NAME}
+# The --no-log-init is required to work around a bug/problem in Go/Docker when very large UIDs are used
+# See https://github.com/moby/moby/issues/5419#issuecomment-41478290 for more context
+useradd \
+  --no-log-init \
+  --gid ${STACKABLE_USER_GID} \
+  --uid ${STACKABLE_USER_UID} \
+  --system \
+  --create-home \
+  --home-dir /stackable \
+   ${STACKABLE_USER_NAME}
+microdnf remove shadow-utils
+microdnf clean all
+rm -rf /var/cache/yum
+
+###
+### Make shell usage in our containers "nicer"
+###
+{
+  echo "alias ll='ls -alF --color=auto'"
+  echo "alias ls='ls --color=auto'"
+  echo "alias ..='cd ..'"
+  echo "export PS1='\u@\[\e[36m\]\H\[\e[m\] \[\e[32m\]\$(pwd)\[\e[m\] \\$ '"
+} >> /stackable/.bashrc
+
+echo -e "if [ -f ~/.bashrc ]; then\n\tsource ~/.bashrc\nfi" >> /stackable/.profile
+
+chown ${STACKABLE_USER_UID}:0 /stackable/.bashrc
+chown ${STACKABLE_USER_UID}:0 /stackable/.profile
+
+# All files and folders owned by root to support running as arbitrary users
+# This is best practice as all container users will belong to the root group (0)
+chown -R ${STACKABLE_USER_UID}:0 /stackable
+chmod -R g=u /stackable
+EOF
+
+COPY <<EOF /README.md
+# Stackable Agent for Apache Kafka
+
+* This image contains version ${VERSION} of the agent.
+* The agent binary can be found in `/usr/local/bin`
+* It is licensed under the OSL-3.0 - the full text can be found in `/licenses/LICENSE`
+EOF
+
+COPY LICENSE /licenses/LICENSE
+
+COPY --from=builder --chown=${STACKABLE_USER_UID}:0 /app/stackable-kafka-agent* /app/stackable-src.tar.gz /usr/local/bin/
+
+USER ${STACKABLE_USER_UID}
+
+ENTRYPOINT ["stackable-kafka-agent"]
+CMD ["run"]
diff --git a/extra/crds.yaml b/extra/crds.yaml
index 9283cc4f..eb762d07 100644
--- a/extra/crds.yaml
+++ b/extra/crds.yaml
@@ -2136,6 +2136,47 @@ spec:
                   type: object
                   x-kubernetes-preserve-unknown-fields: true
                 type: array
+              platformAccess:
+                description: Access of the Stackable Data Platform to this cluster.
+                nullable: true
+                properties:
+                  authentication:
+                    description: The credential the agent authenticates to Kafka with.
+                    oneOf:
+                    - required:
+                      - tls
+                    properties:
+                      tls:
+                        description: 'Source of a TLS client certificate: a secret-operator SecretClass or a static Secret.'
+                        oneOf:
+                        - required:
+                          - secretClass
+                        - required:
+                          - secret
+                        properties:
+                          secret:
+                            description: |-
+                              A static Secret holding the certificate in the keys `tls.crt` and `tls.key` (PEM), e.g. as a
+                              Secret of type `kubernetes.io/tls`.
+                            maxLength: 253
+                            minLength: 1
+                            pattern: ^[a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*$
+                            type: string
+                          secretClass:
+                            description: An AutoTLS SecretClass used to provision the certificate.
+                            maxLength: 253
+                            minLength: 1
+                            pattern: ^[a-z0-9]([-a-z0-9]*[a-z0-9])?(\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*$
+                            type: string
+                        type: object
+                    type: object
+                  enabled:
+                    default: false
+                    description: Whether the operator deploys the agent. False by default.
+                    type: boolean
+                required:
+                - authentication
+                type: object
             required:
             - brokers
             - image
diff --git a/nix/meta.json b/nix/meta.json
index a713d4ca..01626b0e 100644
--- a/nix/meta.json
+++ b/nix/meta.json
@@ -1 +1 @@
-{"operator": {"name": "kafka-operator", "pretty_string": "Apache Kafka", "product_string": "kafka", "url": "stackabletech/kafka-operator.git", "artifacthub_repository_id": "93d87788-9ae7-4f2f-870d-04134948785a", "chart_description": "Kubernetes operator for Apache Kafka. Deploy and run Kafka brokers with the Stackable Data Platform (SDP).", "keywords": ["apache-kafka", "kafka", "streaming", "messaging", "kubernetes", "operator", "big-data"]}}
+{"operator": {"name": "kafka-operator", "pretty_string": "Apache Kafka", "product_string": "kafka", "url": "stackabletech/kafka-operator.git", "artifacthub_repository_id": "93d87788-9ae7-4f2f-870d-04134948785a", "chart_description": "Kubernetes operator for Apache Kafka. Deploy and run Kafka brokers with the Stackable Data Platform (SDP).", "keywords": ["apache-kafka", "kafka", "streaming", "messaging", "kubernetes", "operator", "big-data"]}, "agent": {"name": "kafka-agent"}}
diff --git a/rust/agent-binary/Cargo.toml b/rust/agent-binary/Cargo.toml
new file mode 100644
index 00000000..c138e675
--- /dev/null
+++ b/rust/agent-binary/Cargo.toml
@@ -0,0 +1,24 @@
+[package]
+name = "stackable-kafka-agent"
+description = "Stackable Agent for Apache Kafka"
+version.workspace = true
+authors.workspace = true
+license.workspace = true
+edition.workspace = true
+repository.workspace = true
+publish = false
+
+[dependencies]
+stackable-operator.workspace = true
+
+anyhow.workspace = true
+clap.workspace = true
+tokio.workspace = true
+tracing.workspace = true
+
+[build-dependencies]
+built.workspace = true
+
+[[bin]]
+name = "stackable-kafka-agent"
+path = "src/main.rs"
diff --git a/rust/agent-binary/build.rs b/rust/agent-binary/build.rs
new file mode 100644
index 00000000..fa809bfd
--- /dev/null
+++ b/rust/agent-binary/build.rs
@@ -0,0 +1,3 @@
+fn main() {
+    built::write_built_file().unwrap();
+}
diff --git a/rust/agent-binary/src/main.rs b/rust/agent-binary/src/main.rs
new file mode 100644
index 00000000..0934f803
--- /dev/null
+++ b/rust/agent-binary/src/main.rs
@@ -0,0 +1,44 @@
+//! The Stackable Kafka agent: a per-cluster controller that reconciles in-cluster resources like `KafkaTopic`s.
+
+use clap::Parser;
+use stackable_operator::{
+    cli::{Command, CommonOptions},
+    telemetry::Tracing,
+    utils::signal::SignalWatcher,
+};
+
+mod built_info {
+    include!(concat!(env!("OUT_DIR"), "/built.rs"));
+}
+
+#[derive(clap::Parser)]
+#[clap(about, author)]
+struct Opts {
+    #[clap(subcommand)]
+    cmd: Command<CommonOptions>,
+}
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    match Opts::parse().cmd {
+        Command::Crd => anyhow::bail!("this agent has no CRDs, they are owned by the operator"),
+        Command::Run(CommonOptions { telemetry, .. }) => {
+            let _tracing_guard = Tracing::pre_configured(built_info::PKG_NAME, telemetry).init()?;
+
+            tracing::info!(
+                built_info.pkg_version = built_info::PKG_VERSION,
+                built_info.git_version = built_info::GIT_VERSION,
+                built_info.target = built_info::TARGET,
+                built_info.built_time_utc = built_info::BUILT_TIME_UTC,
+                built_info.rustc_version = built_info::RUSTC_VERSION,
+                "Starting {description}",
+                description = built_info::PKG_DESCRIPTION
+            );
+
+            let sigterm_watcher = SignalWatcher::sigterm()?;
+            sigterm_watcher.handle().await;
+        }
+    }
+
+    Ok(())
+}
diff --git a/rust/operator-binary/src/controller.rs b/rust/operator-binary/src/controller.rs
index 21e1d8de..1dc977ea 100644
--- a/rust/operator-binary/src/controller.rs
+++ b/rust/operator-binary/src/controller.rs
@@ -22,7 +22,7 @@ use stackable_operator::{
     constant,
     crd::listener,
     k8s_openapi::api::{
-        apps::v1::StatefulSet,
+        apps::v1::{Deployment, StatefulSet},
         core::v1::{ConfigMap, Service, ServiceAccount},
         policy::v1::PodDisruptionBudget,
         rbac::v1::RoleBinding,
@@ -65,7 +65,8 @@ use crate::{
         update_status::update_status,
     },
     crd::{
-        APP_NAME, KAFKA_OPERATOR_NAME, KafkaPodDescriptor, MetadataManager,
+        APP_NAME, KAFKA_OPERATOR_NAME, KafkaPlatformAccessAuthentication, KafkaPodDescriptor,
+        MetadataManager,
         authorization::KafkaAuthorizationConfig,
         role::{AnyConfig, AnyConfigOverrides, KafkaRole},
         v1alpha1,
@@ -112,6 +113,7 @@ pub struct Applied;
 /// bootstrap [`Listener`](listener)s.
 pub struct KubernetesResources<T> {
     pub stateful_sets: Vec<StatefulSet>,
+    pub deployments: Vec<Deployment>,
     pub services: Vec<Service>,
     pub listeners: Vec<listener::v1alpha1::Listener>,
     pub config_maps: Vec<ConfigMap>,
@@ -151,6 +153,7 @@ pub struct ValidatedCluster {
     /// address-less around the first reconcile runs; the listener-operator populates the ingress
     /// addresses and the `Listener` watch triggers a new run once it does.
     pub bootstrap_listeners: Vec<listener::v1alpha1::Listener>,
+    pub agent_config: Option<ValidatedAgentConfig>,
 }
 
 impl ValidatedCluster {
@@ -165,6 +168,7 @@ impl ValidatedCluster {
         role_configs: BTreeMap<KafkaRole, ValidatedRoleConfig>,
         role_group_configs: BTreeMap<KafkaRole, BTreeMap<RoleGroupName, ValidatedRoleGroupConfig>>,
         bootstrap_listeners: Vec<listener::v1alpha1::Listener>,
+        agent_config: Option<ValidatedAgentConfig>,
     ) -> Self {
         // `app_version_label_value` is constructed to be a valid label value, so it is also a
         // valid `ProductVersion`.
@@ -187,6 +191,7 @@ impl ValidatedCluster {
             role_configs,
             role_group_configs,
             bootstrap_listeners,
+            agent_config,
         }
     }
 
@@ -339,6 +344,12 @@ impl ValidatedClusterConfig {
     }
 }
 
+/// The configuration of the agent deployed with the cluster.
+pub struct ValidatedAgentConfig {
+    pub image: String,
+    pub authentication: KafkaPlatformAccessAuthentication,
+}
+
 /// Per-role configuration extracted during validation.
 ///
 /// Resolved from the raw [`v1alpha1::KafkaCluster`] spec during validation so the reconcile loop
@@ -397,6 +408,7 @@ pub type ValidatedRoleGroupConfig = stackable_operator::v2::role_utils::RoleGrou
 pub struct Ctx {
     pub client: stackable_operator::client::Client,
     pub operator_environment: OperatorEnvironmentOptions,
+    pub agent_image: String,
 }
 
 #[derive(Snafu, Debug, EnumDiscriminants)]
@@ -466,9 +478,13 @@ pub async fn reconcile_kafka(
         .context(DereferenceSnafu)?;
 
     // validate (no client required)
-    let validated_cluster =
-        validate::validate(kafka, dereferenced_objects, &ctx.operator_environment)
-            .context(ValidateClusterSnafu)?;
+    let validated_cluster = validate::validate(
+        kafka,
+        dereferenced_objects,
+        &ctx.operator_environment,
+        &ctx.agent_image,
+    )
+    .context(ValidateClusterSnafu)?;
 
     tracing::debug!(
         kerberos_enabled = validated_cluster.cluster_config.kafka_security.has_kerberos_enabled(),
@@ -628,6 +644,7 @@ pub(crate) mod test_support {
                 bootstrap_listeners: Vec::new(),
             },
             &operator_environment(),
+            &crate::crd::default_agent_image(&operator_environment().image_repository),
         )
     }
 }
@@ -780,6 +797,7 @@ spec: {}
                         operator_service_name: "kafka-operator".to_owned(),
                         image_repository: "oci.stackable.tech/sdp".to_owned(),
                     },
+                    agent_image: crate::crd::default_agent_image("oci.stackable.tech/sdp"),
                 });
 
                 reconcile_kafka(Arc::new(kafka), ctx).await
diff --git a/rust/operator-binary/src/controller/apply.rs b/rust/operator-binary/src/controller/apply.rs
index 1ce080f8..5f740ecc 100644
--- a/rust/operator-binary/src/controller/apply.rs
+++ b/rust/operator-binary/src/controller/apply.rs
@@ -78,6 +78,7 @@ impl<'a> Applier<'a> {
         // compile here instead of silently never being applied.
         let KubernetesResources {
             stateful_sets,
+            deployments,
             services,
             listeners,
             config_maps,
@@ -97,6 +98,7 @@ impl<'a> Applier<'a> {
         let config_maps = self.add_resources(config_maps).await?;
         let pod_disruption_budgets = self.add_resources(pod_disruption_budgets).await?;
         let stateful_sets = self.add_resources(stateful_sets).await?;
+        let deployments = self.add_resources(deployments).await?;
 
         self.cluster_resources
             .delete_orphaned_resources(self.client)
@@ -105,6 +107,7 @@ impl<'a> Applier<'a> {
 
         Ok(KubernetesResources {
             stateful_sets,
+            deployments,
             services,
             listeners,
             config_maps,
diff --git a/rust/operator-binary/src/controller/build/mod.rs b/rust/operator-binary/src/controller/build/mod.rs
index 1be5313e..18e220a5 100644
--- a/rust/operator-binary/src/controller/build/mod.rs
+++ b/rust/operator-binary/src/controller/build/mod.rs
@@ -14,6 +14,9 @@ use crate::{
                 listener::get_kafka_listener_config, product_logging::vector_config_file_content,
             },
             resource::{
+                agent::{
+                    build_agent_deployment, build_agent_role_binding, build_agent_service_account,
+                },
                 config_map::build_rolegroup_config_map,
                 discovery::build_discovery_configmap,
                 listener::build_broker_rolegroup_bootstrap_listener,
@@ -149,14 +152,24 @@ pub fn build(cluster: &ValidatedCluster) -> Result<KubernetesResources<Prepared>
 
     config_maps.push(build_discovery_configmap(cluster).context(DiscoveryConfigMapSnafu)?);
 
+    let mut deployments = vec![];
+    let mut service_accounts = vec![build_service_account(cluster)];
+    let mut role_bindings = vec![build_role_binding(cluster)];
+    if let Some(agent_config) = &cluster.agent_config {
+        deployments.push(build_agent_deployment(cluster, agent_config));
+        service_accounts.push(build_agent_service_account(cluster));
+        role_bindings.push(build_agent_role_binding(cluster));
+    }
+
     Ok(KubernetesResources {
         stateful_sets,
+        deployments,
         services,
         listeners,
         config_maps,
         pod_disruption_budgets,
-        service_accounts: vec![build_service_account(cluster)],
-        role_bindings: vec![build_role_binding(cluster)],
+        service_accounts,
+        role_bindings,
         status: PhantomData,
     })
 }
@@ -171,6 +184,16 @@ pub(crate) fn recommended_labels_for_cluster_resources(cluster: &ValidatedCluste
     )
 }
 
+pub(crate) fn recommended_labels_for_agent_resources(cluster: &ValidatedCluster) -> Labels {
+    label::recommended_labels_for_agent_resources(
+        &cluster.name,
+        &PRODUCT_NAME,
+        &cluster.product_version,
+        &OPERATOR_NAME,
+        &CONTROLLER_NAME,
+    )
+}
+
 pub(crate) fn recommended_labels_for_role_resources(
     cluster: &ValidatedCluster,
     role_name: &RoleName,
@@ -217,6 +240,10 @@ pub(crate) fn recommended_labels_for_unversioned_role_group_resources(
 }
 
 /// Selector labels matching the pods of a role group.
+pub(crate) fn agent_selector(cluster: &ValidatedCluster) -> Labels {
+    label::agent_selector(&cluster.name, &PRODUCT_NAME)
+}
+
 pub(crate) fn role_group_selector(
     cluster: &ValidatedCluster,
     role_name: &RoleName,
@@ -363,6 +390,8 @@ mod tests {
             sorted_names(&resources.role_bindings),
             ["simple-kafka-rolebinding"]
         );
+        // No agent without platform access.
+        assert!(resources.deployments.is_empty());
     }
 
     #[test]
diff --git a/rust/operator-binary/src/controller/build/resource/agent.rs b/rust/operator-binary/src/controller/build/resource/agent.rs
new file mode 100644
index 00000000..35db3480
--- /dev/null
+++ b/rust/operator-binary/src/controller/build/resource/agent.rs
@@ -0,0 +1,326 @@
+//! Builds the resources of the platform-access agent deployed alongside the cluster.
+
+use std::str::FromStr;
+
+use stackable_operator::{
+    builder::{
+        meta::ObjectMetaBuilder,
+        pod::{
+            PodBuilder, resources::ResourceRequirementsBuilder, security::PodSecurityContextBuilder,
+        },
+    },
+    constant,
+    k8s_openapi::{
+        api::{
+            apps::v1::{Deployment, DeploymentSpec},
+            core::v1::{EnvVarSource, ObjectFieldSelector, ServiceAccount},
+            rbac::v1::{RoleBinding, RoleRef, Subject},
+        },
+        apimachinery::pkg::apis::meta::v1::LabelSelector,
+    },
+    kube::api::ObjectMeta,
+    v2::{
+        builder::{meta::ownerreference_from_resource, pod::container::new_container_builder},
+        types::kubernetes::ContainerName,
+    },
+};
+
+use crate::{
+    controller::{
+        ValidatedAgentConfig, ValidatedCluster,
+        build::{agent_selector, recommended_labels_for_agent_resources},
+    },
+    crd::KafkaPlatformAccessAuthentication,
+};
+
+constant!(AGENT_CONTAINER_NAME: ContainerName = "agent");
+pub const AGENT_CLUSTER_ROLE_NAME: &str = "kafka-agent-clusterrole";
+
+pub fn build_agent_deployment(
+    cluster: &ValidatedCluster,
+    agent_config: &ValidatedAgentConfig,
+) -> Deployment {
+    let mut cb_agent = new_container_builder(&AGENT_CONTAINER_NAME);
+    let mut pod_builder = PodBuilder::new();
+
+    match &agent_config.authentication {
+        KafkaPlatformAccessAuthentication::Tls(credential) => {
+            credential.add_volumes_and_mounts(&mut pod_builder, vec![&mut cb_agent])
+        }
+    }
+
+    cb_agent
+        .image(&agent_config.image)
+        .args(vec!["run".to_owned()])
+        .add_env_var_from_source(
+            "KUBERNETES_NODE_NAME",
+            EnvVarSource {
+                field_ref: Some(ObjectFieldSelector {
+                    field_path: "spec.nodeName".to_owned(),
+                    ..ObjectFieldSelector::default()
+                }),
+                ..EnvVarSource::default()
+            },
+        )
+        // Saves the agent from looking the domain up via the kubelet, which needs extra RBAC.
+        .add_env_var(
+            "KUBERNETES_CLUSTER_DOMAIN",
+            cluster.cluster_domain.to_string(),
+        )
+        .resources(
+            ResourceRequirementsBuilder::new()
+                .with_cpu_request("100m")
+                .with_cpu_limit("500m")
+                .with_memory_request("128Mi")
+                .with_memory_limit("128Mi")
+                .build(),
+        );
+
+    pod_builder
+        .metadata(
+            ObjectMetaBuilder::new()
+                .with_labels(recommended_labels_for_agent_resources(cluster))
+                .build(),
+        )
+        .image_pull_secrets_from_product_image(&cluster.image)
+        .add_container(cb_agent.build())
+        .service_account_name(agent_name(cluster))
+        .security_context(PodSecurityContextBuilder::with_stackable_defaults().build());
+
+    Deployment {
+        metadata: agent_metadata(cluster),
+        spec: Some(DeploymentSpec {
+            replicas: Some(1),
+            selector: LabelSelector {
+                match_labels: Some(agent_selector(cluster).into()),
+                ..LabelSelector::default()
+            },
+            template: pod_builder.build_template(),
+            ..DeploymentSpec::default()
+        }),
+        status: None,
+    }
+}
+
+pub fn build_agent_service_account(cluster: &ValidatedCluster) -> ServiceAccount {
+    ServiceAccount {
+        metadata: agent_metadata(cluster),
+        ..ServiceAccount::default()
+    }
+}
+
+/// Binds the agent ServiceAccount to the agent ClusterRole deployed by the Helm chart.
+pub fn build_agent_role_binding(cluster: &ValidatedCluster) -> RoleBinding {
+    RoleBinding {
+        metadata: agent_metadata(cluster),
+        role_ref: RoleRef {
+            api_group: Some("rbac.authorization.k8s.io".to_owned()),
+            kind: "ClusterRole".to_owned(),
+            name: AGENT_CLUSTER_ROLE_NAME.to_owned(),
+        },
+        subjects: Some(vec![Subject {
+            kind: "ServiceAccount".to_owned(),
+            name: agent_name(cluster),
+            namespace: Some(cluster.namespace.to_string()),
+            ..Subject::default()
+        }]),
+    }
+}
+
+/// The name of all agent resources. Kinds don't share a namespace, so unlike a suffix per kind, this
+/// cannot collide with the resources of a cluster named `<cluster>-agent`.
+fn agent_name(cluster: &ValidatedCluster) -> String {
+    format!("{}-agent", cluster.name)
+}
+
+fn agent_metadata(cluster: &ValidatedCluster) -> ObjectMeta {
+    ObjectMetaBuilder::new()
+        .name_and_namespace(cluster)
+        .name(agent_name(cluster))
+        .ownerreference(ownerreference_from_resource(cluster, None, Some(true)))
+        .with_labels(recommended_labels_for_agent_resources(cluster))
+        .build()
+}
+
+#[cfg(test)]
+mod tests {
+    use serde_json::{Value, json};
+
+    use super::*;
+    use crate::{
+        controller::test_support::{app_version_label, minimal_kafka, validated_cluster},
+        crd::default_agent_image,
+    };
+
+    fn cluster(platform_access: &str) -> ValidatedCluster {
+        cluster_with_image("{productVersion: 3.9.2}", platform_access)
+    }
+
+    fn cluster_with_image(image: &str, platform_access: &str) -> ValidatedCluster {
+        let kafka = minimal_kafka(&format!(
+            r#"
+            apiVersion: kafka.stackable.tech/v1alpha1
+            kind: KafkaCluster
+            metadata:
+              name: simple-kafka
+              namespace: default
+              uid: 12345678-1234-1234-1234-123456789012
+            spec:
+              image: {image}
+              clusterConfig:
+                zookeeperConfigMapName: xyz
+              platformAccess: {platform_access}
+              brokers:
+                roleGroups:
+                  default:
+                    replicas: 1
+            "#
+        ));
+        validated_cluster(&kafka)
+    }
+
+    fn agent_deployment(platform_access: &str) -> Value {
+        let cluster = cluster(platform_access);
+        let agent_config = cluster
+            .agent_config
+            .as_ref()
+            .expect("platform access is enabled");
+        serde_json::to_value(build_agent_deployment(&cluster, agent_config))
+            .expect("must be serializable")
+    }
+
+    #[test]
+    fn agent_config_is_only_resolved_if_platform_access_is_enabled() {
+        assert!(
+            cluster("{enabled: false, authentication: {tls: {secretClass: tls}}}")
+                .agent_config
+                .is_none()
+        );
+
+        let cluster = cluster("{enabled: true, authentication: {tls: {secretClass: tls}}}");
+        let agent_config = cluster.agent_config.expect("platform access is enabled");
+        assert_eq!(agent_config.image, default_agent_image("oci.example.org"));
+    }
+
+    #[test]
+    fn test_deployment() {
+        let deployment =
+            agent_deployment("{enabled: true, authentication: {tls: {secretClass: tls}}}");
+
+        assert_eq!(deployment["metadata"]["name"], "simple-kafka-agent");
+        assert_eq!(deployment["spec"]["replicas"], 1);
+        // The selector must not match the broker or controller Pods.
+        assert_eq!(
+            deployment["spec"]["selector"]["matchLabels"],
+            json!({
+                "app.kubernetes.io/component": "agent",
+                "app.kubernetes.io/instance": "simple-kafka",
+                "app.kubernetes.io/name": "kafka"
+            })
+        );
+        assert_eq!(
+            deployment["spec"]["template"]["metadata"]["labels"],
+            json!({
+                "app.kubernetes.io/component": "agent",
+                "app.kubernetes.io/instance": "simple-kafka",
+                "app.kubernetes.io/managed-by": "kafka.stackable.tech_kafkacluster",
+                "app.kubernetes.io/name": "kafka",
+                "app.kubernetes.io/version": app_version_label("3.9.2"),
+                "stackable.tech/vendor": "Stackable"
+            })
+        );
+
+        let pod_spec = &deployment["spec"]["template"]["spec"];
+        assert_eq!(pod_spec["serviceAccountName"], "simple-kafka-agent");
+        assert_eq!(
+            pod_spec["containers"][0]["image"],
+            default_agent_image("oci.example.org")
+        );
+        assert_eq!(pod_spec["containers"][0]["args"], json!(["run"]));
+        assert_eq!(
+            pod_spec["containers"][0]["volumeMounts"],
+            json!([{"mountPath": "/stackable/secrets/tls-client-cert", "name": "tls-client-cert"}])
+        );
+        assert_eq!(pod_spec["volumes"][0]["name"], "tls-client-cert");
+        assert_eq!(
+            pod_spec["volumes"][0]["ephemeral"]["volumeClaimTemplate"]["metadata"]["annotations"]["secrets.stackable.tech/class"],
+            "tls"
+        );
+    }
+
+    #[test]
+    fn test_deployment_with_static_secret() {
+        let deployment =
+            agent_deployment("{enabled: true, authentication: {tls: {secret: my-cert}}}");
+
+        assert_eq!(
+            deployment["spec"]["template"]["spec"]["volumes"],
+            json!([{"name": "tls-client-cert", "secret": {"secretName": "my-cert"}}])
+        );
+    }
+
+    #[test]
+    fn test_deployment_uses_product_image_pull_secrets() {
+        let cluster = cluster_with_image(
+            "{productVersion: 3.9.2, pullSecrets: [{name: regcred}]}",
+            "{enabled: true, authentication: {tls: {secretClass: tls}}}",
+        );
+        let agent_config = cluster
+            .agent_config
+            .as_ref()
+            .expect("platform access is enabled");
+        let deployment = serde_json::to_value(build_agent_deployment(&cluster, agent_config))
+            .expect("must be serializable");
+
+        assert_eq!(
+            deployment["spec"]["template"]["spec"]["imagePullSecrets"],
+            json!([{"name": "regcred"}])
+        );
+    }
+
+    #[test]
+    fn test_role_binding() {
+        let cluster = cluster("{enabled: true, authentication: {tls: {secretClass: tls}}}");
+
+        assert_eq!(
+            json!({
+                "apiVersion": "rbac.authorization.k8s.io/v1",
+                "kind": "RoleBinding",
+                "metadata": {
+                    "labels": {
+                        "app.kubernetes.io/component": "agent",
+                        "app.kubernetes.io/instance": "simple-kafka",
+                        "app.kubernetes.io/managed-by": "kafka.stackable.tech_kafkacluster",
+                        "app.kubernetes.io/name": "kafka",
+                        "app.kubernetes.io/version": app_version_label("3.9.2"),
+                        "stackable.tech/vendor": "Stackable"
+                    },
+                    "name": "simple-kafka-agent",
+                    "namespace": "default",
+                    "ownerReferences": [
+                        {
+                            "apiVersion": "kafka.stackable.tech/v1alpha1",
+                            "controller": true,
+                            "kind": "KafkaCluster",
+                            "name": "simple-kafka",
+                            "uid": "12345678-1234-1234-1234-123456789012"
+                        }
+                    ]
+                },
+                "roleRef": {
+                    "apiGroup": "rbac.authorization.k8s.io",
+                    "kind": "ClusterRole",
+                    "name": "kafka-agent-clusterrole"
+                },
+                "subjects": [
+                    {
+                        "kind": "ServiceAccount",
+                        "name": "simple-kafka-agent",
+                        "namespace": "default"
+                    }
+                ]
+            }),
+            serde_json::to_value(build_agent_role_binding(&cluster)).expect("must be serializable")
+        );
+    }
+}
diff --git a/rust/operator-binary/src/controller/build/resource/mod.rs b/rust/operator-binary/src/controller/build/resource/mod.rs
index 7f458714..4acba626 100644
--- a/rust/operator-binary/src/controller/build/resource/mod.rs
+++ b/rust/operator-binary/src/controller/build/resource/mod.rs
@@ -1,5 +1,6 @@
 //! Builders that assemble Kubernetes resources for kafka rolegroups.
 
+pub mod agent;
 pub mod config_map;
 pub mod discovery;
 pub mod listener;
diff --git a/rust/operator-binary/src/controller/validate.rs b/rust/operator-binary/src/controller/validate.rs
index dd3bafb5..62fb2bff 100644
--- a/rust/operator-binary/src/controller/validate.rs
+++ b/rust/operator-binary/src/controller/validate.rs
@@ -30,8 +30,8 @@ use stackable_operator::{
 
 use crate::{
     controller::{
-        RoleGroupName, ValidatedCluster, ValidatedClusterConfig, ValidatedKafkaConfig,
-        ValidatedRoleConfig, ValidatedRoleGroupConfig,
+        RoleGroupName, ValidatedAgentConfig, ValidatedCluster, ValidatedClusterConfig,
+        ValidatedKafkaConfig, ValidatedRoleConfig, ValidatedRoleGroupConfig,
         dereference::DereferencedObjects,
         security::{self, ValidatedKafkaSecurity},
     },
@@ -199,6 +199,7 @@ pub fn validate(
     kafka: &v1alpha1::KafkaCluster,
     dereferenced_objects: DereferencedObjects,
     operator_environment: &OperatorEnvironmentOptions,
+    agent_image: &str,
 ) -> Result<ValidatedCluster> {
     let image = kafka
         .spec
@@ -333,6 +334,16 @@ pub fn validate(
         .cluster_domain
         .clone();
 
+    let agent_config = kafka
+        .spec
+        .platform_access
+        .as_ref()
+        .filter(|platform_access| platform_access.enabled)
+        .map(|platform_access| ValidatedAgentConfig {
+            image: agent_image.to_owned(),
+            authentication: platform_access.authentication.clone(),
+        });
+
     Ok(ValidatedCluster::new(
         name,
         namespace,
@@ -353,6 +364,7 @@ pub fn validate(
         role_configs,
         role_group_configs,
         dereferenced_objects.bootstrap_listeners,
+        agent_config,
     ))
 }
 
diff --git a/rust/operator-binary/src/crd/mod.rs b/rust/operator-binary/src/crd/mod.rs
index 900d0bb4..7f486d93 100644
--- a/rust/operator-binary/src/crd/mod.rs
+++ b/rust/operator-binary/src/crd/mod.rs
@@ -13,7 +13,7 @@ use snafu::Snafu;
 use stackable_operator::{
     commons::{
         cluster_operation::ClusterOperation, networking::DomainName,
-        product_image_selection::ProductImage,
+        platform_access::tls::TlsClientCredential, product_image_selection::ProductImage,
     },
     config::merge::Merge,
     constant,
@@ -45,6 +45,15 @@ use crate::crd::{
 };
 
 pub const CONTAINER_IMAGE_BASE_NAME: &str = "kafka";
+pub const AGENT_IMAGE_BASE_NAME: &str = "kafka-agent";
+
+/// The agent image of the same release as the operator.
+pub fn default_agent_image(image_repository: &str) -> String {
+    format!(
+        "{image_repository}/{AGENT_IMAGE_BASE_NAME}:{}",
+        crate::built_info::PKG_VERSION
+    )
+}
 pub const APP_NAME: &str = "kafka";
 pub const KAFKA_OPERATOR_NAME: &str = "kafka.stackable.tech";
 pub const FIELD_MANAGER: &str = "kafka-operator";
@@ -98,6 +107,26 @@ pub type ControllerRole = Role<
     JavaCommonConfig,
 >;
 
+/// Access of the Stackable Data Platform to this Kafka cluster, used by an agent to manage resources
+/// like `KafkaTopic`s. The credential must be authorized in Kafka by the user.
+#[derive(Clone, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct KafkaPlatformAccess {
+    /// Whether the operator deploys the agent. False by default.
+    #[serde(default)]
+    pub enabled: bool,
+
+    /// The credential the agent authenticates to Kafka with.
+    pub authentication: KafkaPlatformAccessAuthentication,
+}
+
+/// How the agent authenticates to Kafka.
+#[derive(Clone, Debug, Deserialize, Eq, JsonSchema, PartialEq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub enum KafkaPlatformAccessAuthentication {
+    Tls(TlsClientCredential),
+}
+
 #[versioned(
     version(name = "v1alpha1"),
     crates(
@@ -138,6 +167,10 @@ pub mod versioned {
         #[serde(default)]
         pub cluster_config: v1alpha1::KafkaClusterConfig,
 
+        /// Access of the Stackable Data Platform to this cluster.
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        pub platform_access: Option<KafkaPlatformAccess>,
+
         // no doc - docs in ClusterOperation struct.
         #[serde(default)]
         pub cluster_operation: ClusterOperation,
diff --git a/rust/operator-binary/src/main.rs b/rust/operator-binary/src/main.rs
index 075b48e1..809bd566 100644
--- a/rust/operator-binary/src/main.rs
+++ b/rust/operator-binary/src/main.rs
@@ -14,7 +14,7 @@ use stackable_operator::{
     crd::listener,
     eos::EndOfSupportChecker,
     k8s_openapi::api::{
-        apps::v1::StatefulSet,
+        apps::v1::{Deployment, StatefulSet},
         core::v1::{ConfigMap, Service, ServiceAccount},
         policy::v1::PodDisruptionBudget,
         rbac::v1::RoleBinding,
@@ -37,7 +37,7 @@ use stackable_operator::{
 
 use crate::{
     controller::KAFKA_FULL_CONTROLLER_NAME,
-    crd::{KAFKA_OPERATOR_NAME, KafkaCluster, KafkaClusterVersion, v1alpha1},
+    crd::{KAFKA_OPERATOR_NAME, KafkaCluster, KafkaClusterVersion, default_agent_image, v1alpha1},
     webhooks::conversion::create_webhook_server,
 };
 
@@ -67,6 +67,10 @@ struct Opts {
 struct KafkaRun {
     #[clap(flatten)]
     common: RunArguments,
+
+    /// The agent image. Defaults to the agent image belonging to this operator release.
+    #[arg(long, env)]
+    agent_image: Option<String>,
 }
 
 #[tokio::main]
@@ -83,7 +87,7 @@ async fn main() -> anyhow::Result<()> {
                     maintenance,
                     common,
                 },
-            ..
+            agent_image,
         }) => {
             // NOTE (@NickLarsenNZ): Before stackable-telemetry was used:
             // - The console log level was set by `KAFKA_OPERATOR_LOG`, and is now `CONSOLE_LOG` (when using Tracing::pre_configured).
@@ -171,6 +175,10 @@ async fn main() -> anyhow::Result<()> {
                     watch_namespace.get_api::<DeserializeGuard<StatefulSet>>(&client),
                     watcher::Config::default(),
                 )
+                .owns(
+                    watch_namespace.get_api::<DeserializeGuard<Deployment>>(&client),
+                    watcher::Config::default(),
+                )
                 .watches(
                     watch_namespace.get_api::<DeserializeGuard<ConfigMap>>(&client),
                     watcher::Config::default(),
@@ -188,6 +196,9 @@ async fn main() -> anyhow::Result<()> {
                     controller::error_policy,
                     Arc::new(controller::Ctx {
                         client: client.clone(),
+                        agent_image: agent_image.unwrap_or_else(|| {
+                            default_agent_image(&operator_environment.image_repository)
+                        }),
                         operator_environment,
                     }),
                 )
@@ -211,7 +222,7 @@ async fn main() -> anyhow::Result<()> {
                 .map(anyhow::Ok);
 
             let delayed_kafka_controller = async {
-                signal::crd_established(&client, v1alpha1::KafkaCluster::crd_name(), None).await?;
+                signal::crd_established(&client, v1alpha1::KafkaCluster::crd_name()).await?;
                 kafka_controller.await
             };
 
diff --git a/rust/operator-binary/src/webhooks/conversion.rs b/rust/operator-binary/src/webhooks/conversion.rs
index 912ea593..232e4dac 100644
--- a/rust/operator-binary/src/webhooks/conversion.rs
+++ b/rust/operator-binary/src/webhooks/conversion.rs
@@ -4,6 +4,7 @@ use stackable_operator::{
     kube::{Client, core::crd::MergeError},
     webhook::{
         WebhookServer, WebhookServerError, WebhookServerOptions,
+        health::HealthCheckRegistry,
         webhooks::{ConversionWebhook, ConversionWebhookOptions},
     },
 };
@@ -46,7 +47,11 @@ pub async fn create_webhook_server(
         webhook_service_name: operator_environment.operator_service_name.to_owned(),
     };
 
-    WebhookServer::new(vec![Box::new(conversion_webhook)], webhook_server_options)
-        .await
-        .context(CreateWebhookSnafu)
+    WebhookServer::new(
+        vec![Box::new(conversion_webhook)],
+        webhook_server_options,
+        HealthCheckRegistry::new(),
+    )
+    .await
+    .context(CreateWebhookSnafu)
 }