diff --git a/.github/workflows/publish-command.yml b/.github/workflows/publish-command.yml
index a665f07a6645..1b8b882c4cb3 100644
--- a/.github/workflows/publish-command.yml
+++ b/.github/workflows/publish-command.yml
@@ -44,6 +44,11 @@ jobs:
     runs-on: ${{ needs.start-publish-image-runner.outputs.label }}
     environment: more-secrets
     steps:
+      - name: Set up Cloud SDK
+        uses: google-github-actions/setup-gcloud@master
+        with:
+          service_account_key: ${{ secrets.SPEC_CACHE_SERVICE_ACCOUNT_KEY }}
+          export_default_credentials: true
       - name: Search for valid connector name format
         id: regex
         uses: AsasInnab/regex-action@v1
@@ -174,8 +179,8 @@ jobs:
           SOURCE_BAMBOO_HR_CREDS: ${{ secrets.SOURCE_BAMBOO_HR_CREDS }}
           SOURCE_BIGCOMMERCE_CREDS: ${{ secrets.SOURCE_BIGCOMMERCE_CREDS }}
       - run: |
-          docker login -u airbytebot -p ${DOCKER_PASSWORD}
-          ./tools/integrations/manage.sh publish airbyte-integrations/${{ github.event.inputs.connector }} ${{ github.event.inputs.run-tests }}
+          echo "$SPEC_CACHE_SERVICE_ACCOUNT_KEY" > spec_cache_key_file.json && docker login -u airbytebot -p ${DOCKER_PASSWORD}
+          ./tools/integrations/manage.sh publish airbyte-integrations/${{ github.event.inputs.connector }} ${{ github.event.inputs.run-tests }} --publish_spec_to_cache
         name: publish ${{ github.event.inputs.connector }}
         id: publish
         env:
diff --git a/airbyte-scheduler/client/src/main/java/io/airbyte/scheduler/client/BucketSpecCacheSchedulerClient.java b/airbyte-scheduler/client/src/main/java/io/airbyte/scheduler/client/BucketSpecCacheSchedulerClient.java
index 785271f8bc11..ccbbf7492a9d 100644
--- a/airbyte-scheduler/client/src/main/java/io/airbyte/scheduler/client/BucketSpecCacheSchedulerClient.java
+++ b/airbyte-scheduler/client/src/main/java/io/airbyte/scheduler/client/BucketSpecCacheSchedulerClient.java
@@ -90,15 +90,19 @@ public SynchronousResponse<AirbyteCatalog> createDiscoverSchemaJob(final SourceC
 
   @Override
   public SynchronousResponse<ConnectorSpecification> createGetSpecJob(final String dockerImage) throws IOException {
+    LOGGER.debug("getting spec!");
     Optional<ConnectorSpecification> cachedSpecOptional;
     // never want to fail because we could not fetch from off board storage.
     try {
       cachedSpecOptional = bucketSpecFetcher.apply(dockerImage);
+      LOGGER.debug("Spec bucket cache: Call to cache did not fail.");
     } catch (final RuntimeException e) {
       cachedSpecOptional = Optional.empty();
+      LOGGER.debug("Spec bucket cache: Call to cache failed.");
     }
 
     if (cachedSpecOptional.isPresent()) {
+      LOGGER.debug("Spec bucket cache: Cache hit.");
       final long now = Instant.now().toEpochMilli();
       final SynchronousJobMetadata mockMetadata = new SynchronousJobMetadata(
           UUID.randomUUID(),
@@ -110,6 +114,7 @@ public SynchronousResponse<ConnectorSpecification> createGetSpecJob(final String
           Path.of(""));
       return new SynchronousResponse<>(cachedSpecOptional.get(), mockMetadata);
     } else {
+      LOGGER.debug("Spec bucket cache: Cache miss.");
       return client.createGetSpecJob(dockerImage);
     }
   }
@@ -127,12 +132,14 @@ private static Optional<ConnectorSpecification> attemptToFetchSpecFromBucket(fin
     Preconditions.checkArgument(dockerImageComponents.length == 2, "Invalidate docker image: " + dockerImage);
     final String dockerImageName = dockerImageComponents[0];
     final String dockerImageTag = dockerImageComponents[1];
-    final Blob specAsBlob =
-        storage.get(bucketName, Path.of("specs").resolve(dockerImageName).resolve(dockerImageTag).resolve("spec.json").toString());
+
+    final Path specPath = Path.of("specs").resolve(dockerImageName).resolve(dockerImageTag).resolve("spec.json");
+    LOGGER.debug("Checking path for cached spec: {} {}", bucketName, specPath);
+    final Blob specAsBlob = storage.get(bucketName, specPath.toString());
 
     // if null it means the object was not found.
     if (specAsBlob == null) {
-      LOGGER.warn("Spec not found in bucket storage");
+      LOGGER.debug("Spec not found in bucket storage");
       return Optional.empty();
     }
 
diff --git a/tools/integrations/manage.sh b/tools/integrations/manage.sh
index ab1ea539200a..c26f4cfae0e0 100755
--- a/tools/integrations/manage.sh
+++ b/tools/integrations/manage.sh
@@ -7,10 +7,11 @@ set -x
 
 USAGE="
 Usage: $(basename "$0") <cmd>
+For publish, if you want to push the spec to the spec cache, provide a path to a service account key file that can write to the cache.
 Available commands:
   scaffold
   build  <integration_root_path> [<run_tests>]
-  publish  <integration_root_path> [<run_tests>]
+  publish  <integration_root_path> [<run_tests>] [--publish_spec_to_cache] [--publish_spec_to_cache_with_key_file <path to keyfile>]
 "
 
 _check_tag_exists() {
@@ -47,6 +48,25 @@ cmd_publish() {
   [ -d "$path" ] || error "Path must be the root path of the integration"
 
   local run_tests=$1; shift || run_tests=true
+  local publish_spec_to_cache
+  local spec_cache_writer_sa_key_file
+
+  while [ $# -ne 0 ]; do
+    case "$1" in
+    --publish_spec_to_cache)
+      publish_spec_to_cache=true
+      shift 1
+      ;;
+    --publish_spec_to_cache_with_key_file)
+      publish_spec_to_cache=true
+      spec_cache_writer_sa_key_file="$2"
+      shift 2
+      ;;
+    *)
+      error "Unknown option: $1"
+      ;;
+    esac
+  done
 
   cmd_build "$path" "$run_tests"
 
@@ -69,6 +89,34 @@ cmd_publish() {
   echo "Publishing new version ($versioned_image)"
   docker push "$versioned_image"
   docker push "$latest_image"
+
+  if [[ "true" == "${publish_spec_to_cache}" ]]; then
+    echo "Publishing and writing to spec cache."
+
+    # publish spec to cache. do so, by running get spec locally and then pushing it to gcs.
+    local tmp_spec_file; tmp_spec_file=$(mktemp)
+    docker run --rm "$versioned_image" spec | \
+      # 1. filter out any lines that are not valid json.
+      jq -R "fromjson? | ." | \
+      # 2. grab any json that has a spec in it.
+      # 3. if there are more than one, take the first one.
+      # 4. if there are none, throw an error.
+      jq -s "map(select(.spec != null)) | map(.spec) | first | if . != null then . else error(\"no spec found\") end" \
+      > "$tmp_spec_file"
+    docker run --rm "$versioned_image" spec | jq .spec > "$tmp_spec_file"
+
+    # use service account key file is provided.
+    if [[ -n "${spec_cache_writer_sa_key_file}" ]]; then
+      echo "Using provided service account key"
+      gcloud auth activate-service-account --key-file "$spec_cache_writer_sa_key_file"
+    else
+      echo "Using environment gcloud"
+    fi
+
+    gsutil cp "$tmp_spec_file" gs://io-airbyte-cloud-spec-cache/specs/"$image_name"/"$image_version"/spec.json
+  else
+    echo "Publishing without writing to spec cache."
+  fi
 }
 
 main() {