diff --git a/airbyte/secrets/google_gsm.py b/airbyte/secrets/google_gsm.py
index b361dbdb..a1d9caea 100644
--- a/airbyte/secrets/google_gsm.py
+++ b/airbyte/secrets/google_gsm.py
@@ -130,10 +130,16 @@ def __init__(
 
     def get_secret(self, secret_name: str) -> SecretString | None:
         """Get a named secret from Google Colab user secrets."""
+        full_name = secret_name
+        if "projects/" not in full_name:
+            # This is not yet fully qualified
+            full_name = f"projects/{self.project}/secrets/{secret_name}/versions/latest"
+
+        if "/versions/" not in full_name:
+            full_name += "/versions/latest"
+
         return SecretString(
-            self.secret_client.access_secret_version(
-                name=f"projects/{self.project}/secrets/{secret_name}/versions/latest"
-            ).payload.data.decode("UTF-8")
+            self.secret_client.access_secret_version(name=full_name).payload.data.decode("UTF-8")
         )
 
     def fetch_secrets(
@@ -155,11 +161,10 @@ def fetch_secrets(
             Iterable[SecretHandle]: An iterable of `SecretHandle` objects for the matching secrets.
         """
         gsm_secrets: ListSecretsPager = self.secret_client.list_secrets(
-            secretmanager.ListSecretsRequest(
-                request={
-                    "filter": filter_string,
-                }
-            )
+            request=secretmanager.ListSecretsRequest(
+                filter=filter_string,
+                parent=f"projects/{self.project}",
+            ),
         )
 
         return [
@@ -205,3 +210,39 @@ def fetch_connector_secrets(
             label_key=self.CONNECTOR_LABEL,
             label_value=connector_name,
         )
+
+    def fetch_connector_secret(
+        self,
+        connector_name: str,
+    ) -> SecretHandle:
+        """Fetch secret in the secret manager, using the connector name as a filter for the label.
+
+        This method is a convenience method that returns the first secret found for the connector.
+
+        The label key used to filter the secrets is defined by the `CONNECTOR_LABEL` attribute,
+        which defaults to 'connector'.
+
+        Args:
+            connector_name (str): The name of the connector to filter by.
+
+        Returns:
+            SecretHandle: The matching secret.
+        """
+        results: Iterable[SecretHandle] = self.fetch_connector_secrets(connector_name)
+        try:
+            result = next(iter(results))
+        except StopIteration:
+            raise exc.PyAirbyteError(
+                message="No secrets found for connector.",
+                guidance=(
+                    "Please check that the connector name is correct "
+                    "and that the secret is correctly labeled."
+                ),
+                context={
+                    "project": self.project,
+                    "connector_name": connector_name,
+                    "label_key": self.CONNECTOR_LABEL,
+                },
+            ) from None
+
+        return result
diff --git a/tests/integration_tests/secrets/__init__.py b/tests/integration_tests/secrets/__init__.py
new file mode 100644
index 00000000..e69de29b
diff --git a/tests/integration_tests/secrets/test_gsm_secrets.py b/tests/integration_tests/secrets/test_gsm_secrets.py
new file mode 100644
index 00000000..c8dfec98
--- /dev/null
+++ b/tests/integration_tests/secrets/test_gsm_secrets.py
@@ -0,0 +1,56 @@
+# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+"""Tests for the GSM secrets manager."""
+from __future__ import annotations
+
+from airbyte.secrets.base import SecretHandle
+from airbyte.secrets.google_gsm import GoogleGSMSecretManager
+
+
+def test_get_gsm_secret(ci_secret_manager: GoogleGSMSecretManager) -> dict:
+    assert ci_secret_manager.get_secret(
+        "SECRET_DESTINATION_DUCKDB__MOTHERDUCK__CREDS",
+    ).parse_json()
+
+
+def test_get_gsm_secrets_with_filter(ci_secret_manager: GoogleGSMSecretManager) -> None:
+    """Test fetching connector secrets."""
+    secrets = ci_secret_manager.fetch_secrets(
+        filter_string="labels.connector=source-bigquery",
+    )
+    assert secrets is not None
+    secrets_list = list(secrets)
+    assert len(secrets_list) > 0
+    assert secrets_list[0].get_value().is_json()
+
+
+def test_get_gsm_secrets_by_label(ci_secret_manager: GoogleGSMSecretManager) -> None:
+    """Test fetching connector secrets."""
+    secrets = ci_secret_manager.fetch_secrets_by_label(
+        label_key="connector",
+        label_value="source-salesforce",
+    )
+    assert secrets is not None
+    secrets_list = list(secrets)
+    assert len(secrets_list) > 0
+    assert secrets_list[0].get_value().is_json()
+
+
+def test_get_connector_secrets(ci_secret_manager: GoogleGSMSecretManager) -> None:
+    """Test fetching connector secrets."""
+    secrets = ci_secret_manager.fetch_connector_secrets(
+        "source-salesforce"
+    )
+    assert secrets is not None
+    secrets_list = list(secrets)
+    assert len(secrets_list) > 0
+    assert secrets_list[0].get_value().is_json()
+
+
+def test_first_connector_secret(ci_secret_manager: GoogleGSMSecretManager) -> None:
+    """Test fetching connector secrets."""
+    secret = ci_secret_manager.fetch_connector_secret(
+        "source-salesforce"
+    )
+    assert secret is not None
+    assert isinstance(secret, SecretHandle)
+    assert secret.get_value().is_json()