diff --git a/airbyte/_executor.py b/airbyte/_executor.py
index 8214306c..96611854 100644
--- a/airbyte/_executor.py
+++ b/airbyte/_executor.py
@@ -10,6 +10,7 @@
 from shutil import rmtree
 from typing import IO, TYPE_CHECKING, Any, NoReturn, cast
 
+from overrides import overrides
 from rich import print
 from typing_extensions import Literal
 
@@ -78,6 +79,18 @@ def install(self) -> None:
     def uninstall(self) -> None:
         pass
 
+    def get_installed_version(
+        self,
+        *,
+        raise_on_error: bool = False,
+        recheck: bool = False,
+    ) -> str | None:
+        """Detect the version of the connector installed."""
+        _ = raise_on_error, recheck  # Unused
+        raise NotImplementedError(
+            f"'{type(self).__name__}' class cannot yet detect connector versions."
+        )
+
 
 @contextmanager
 def _stream_from_subprocess(args: list[str]) -> Generator[Iterable[str], None, None]:
@@ -238,7 +251,7 @@ def install(self) -> None:
             raise exc.AirbyteConnectorInstallationError from ex
 
         # Assuming the installation succeeded, store the installed version
-        self.reported_version = self._get_installed_version(raise_on_error=False, recheck=True)
+        self.reported_version = self.get_installed_version(raise_on_error=False, recheck=True)
         log_install_state(self.name, state=EventState.SUCCEEDED)
         print(
             f"Connector '{self.name}' installed successfully!\n"
@@ -246,7 +259,8 @@ def install(self) -> None:
             f"{self.docs_url}#reference\n"
         )
 
-    def _get_installed_version(
+    @overrides
+    def get_installed_version(
         self,
         *,
         raise_on_error: bool = False,
@@ -315,7 +329,7 @@ def ensure_installation(
         """
         # Store the installed version (or None if not installed)
         if not self.reported_version:
-            self.reported_version = self._get_installed_version()
+            self.reported_version = self.get_installed_version()
 
         original_installed_version = self.reported_version
 
diff --git a/airbyte/sources/base.py b/airbyte/sources/base.py
index a40cf9c5..a19d1908 100644
--- a/airbyte/sources/base.py
+++ b/airbyte/sources/base.py
@@ -435,6 +435,14 @@ def _with_logging(records: Iterable[dict[str, Any]]) -> Iterator[dict[str, Any]]
             stream_metadata=configured_stream,
         )
 
+    @property
+    def connector_version(self) -> str | None:
+        """Return the version of the connector as reported by the executor.
+
+        Returns None if the version cannot be determined.
+        """
+        return self.executor.get_installed_version()
+
     def get_documents(
         self,
         stream: str,
diff --git a/tests/integration_tests/test_source_test_fixture.py b/tests/integration_tests/test_source_test_fixture.py
index 9007725f..f0c0ef8d 100644
--- a/tests/integration_tests/test_source_test_fixture.py
+++ b/tests/integration_tests/test_source_test_fixture.py
@@ -239,7 +239,7 @@ def test_version_enforcement(
                 install_if_missing=False,
             )
             if requested_version:  # Don't raise if a version is not requested
-                assert source.executor._get_installed_version(raise_on_error=True) == (
+                assert source.executor.get_installed_version(raise_on_error=True) == (
                     requested_version or latest_available_version
                 ).replace("latest", latest_available_version)
             source.executor.ensure_installation(auto_fix=False)