diff --git a/airbyte/_executors/python.py b/airbyte/_executors/python.py
index 57a5e0094..0b898d4e3 100644
--- a/airbyte/_executors/python.py
+++ b/airbyte/_executors/python.py
@@ -69,6 +69,7 @@ def __init__(
         with suppress(Exception):
             self.install_root.mkdir(parents=True, exist_ok=True)
         self.use_python = use_python
+        self._console_script_name: str | None = None
 
     def _get_venv_name(self) -> str:
         return f".venv-{self.name}"
@@ -76,9 +77,85 @@ def _get_venv_name(self) -> str:
     def _get_venv_path(self) -> Path:
         return self.install_root / self._get_venv_name()
 
+    def _get_pypi_package_name(self) -> str:
+        if self.metadata and self.metadata.pypi_package_name:
+            return self.metadata.pypi_package_name
+        return f"airbyte-{self.name}"
+
+    def _discover_console_script_names(self) -> list[str]:
+        """Return the installed package's console script names, if discoverable."""
+        if not self.interpreter_path.exists():
+            return []
+
+        package_name = self._get_pypi_package_name()
+        connector_name = self.name
+        discovery_script = "\n".join(
+            [
+                "import importlib.metadata as metadata",
+                "import re",
+                "",
+                f"package_name = {package_name!r}",
+                f"connector_name = {connector_name!r}",
+                "",
+                "def canonicalize(name):",
+                '    return re.sub(r"[-_.]+", "-", name).lower()',
+                "",
+                "canonical_package_name = canonicalize(package_name)",
+                "entry_points = [",
+                "    ep",
+                '    for ep in metadata.entry_points(group="console_scripts")',
+                "    if ep.dist is not None",
+                "    and canonicalize(ep.dist.name) == canonical_package_name",
+                "]",
+                "for entry_point_name in sorted(ep.name for ep in entry_points):",
+                "    print(entry_point_name)",
+            ]
+        )
+        try:
+            result = subprocess.check_output(
+                [str(self.interpreter_path), "-c", discovery_script],
+                universal_newlines=True,
+                stderr=subprocess.PIPE,
+            ).strip()
+        except (FileNotFoundError, subprocess.CalledProcessError):
+            return []
+
+        return result.splitlines()
+
+    def _resolve_console_script_name(self) -> str | None:
+        """Resolve the connector CLI executable name within the virtual environment."""
+        suffix: Literal[".exe", ""] = ".exe" if is_windows() else ""
+        if self._console_script_name:
+            cached_path = get_bin_dir(self._get_venv_path()) / (self._console_script_name + suffix)
+            if cached_path.exists():
+                return self._console_script_name
+            self._console_script_name = None
+
+        default_name = self.name + suffix
+        default_path = get_bin_dir(self._get_venv_path()) / default_name
+        if default_path.exists():
+            self._console_script_name = self.name
+            return self._console_script_name
+
+        discovered_names = self._discover_console_script_names()
+        if self.name in discovered_names:
+            discovered_name = self.name
+        elif len(discovered_names) == 1:
+            discovered_name = discovered_names[0]
+        else:
+            return None
+
+        discovered_path = get_bin_dir(self._get_venv_path()) / (discovered_name + suffix)
+        if discovered_path.exists():
+            self._console_script_name = discovered_name
+            return self._console_script_name
+
+        return None
+
     def _get_connector_path(self) -> Path:
         suffix: Literal[".exe", ""] = ".exe" if is_windows() else ""
-        return get_bin_dir(self._get_venv_path()) / (self.name + suffix)
+        script_name = self._resolve_console_script_name() or self.name
+        return get_bin_dir(self._get_venv_path()) / (script_name + suffix)
 
     @property
     def interpreter_path(self) -> Path:
@@ -103,6 +180,7 @@ def uninstall(self) -> None:
             rmtree(str(self._get_venv_path()))
 
         self.reported_version = None  # Reset the reported version from the previous installation
+        self._console_script_name = None
 
     @property
     def docs_url(self) -> str:
@@ -182,6 +260,7 @@ def install(self) -> None:
             raise exc.AirbyteConnectorInstallationError from ex
 
         # Assuming the installation succeeded, store the installed version
+        self._console_script_name = None
         self.reported_version = self.get_installed_version(raise_on_error=False, recheck=True)
         log_install_state(self.name, state=EventState.SUCCEEDED)
         print(
@@ -212,7 +291,6 @@ def get_installed_version(
         if not recheck and self.reported_version:
             return self.reported_version
 
-        connector_name = self.name
         if not self.interpreter_path.exists():
             # No point in trying to detect the version if the interpreter does not exist
             if raise_on_error:
@@ -225,16 +303,12 @@ def get_installed_version(
             return None
 
         try:
-            package_name = (
-                self.metadata.pypi_package_name
-                if self.metadata and self.metadata.pypi_package_name
-                else f"airbyte-{connector_name}"
-            )
+            package_name = self._get_pypi_package_name()
             return subprocess.check_output(
                 [
                     self.interpreter_path,
                     "-c",
-                    f"from importlib.metadata import version; print(version('{package_name}'))",
+                    f"from importlib.metadata import version; print(version({package_name!r}))",
                 ],
                 universal_newlines=True,
                 stderr=subprocess.PIPE,  # Don't print to stderr
@@ -281,21 +355,22 @@ def ensure_installation(
             self.install()
             reinstalled = True
 
-        elif not self._get_connector_path().exists():
+        elif not self._resolve_console_script_name():
             if not auto_fix:
                 raise exc.AirbyteConnectorInstallationError(
                     message="Could not locate connector executable within the virtual environment.",
                     connector_name=self.name,
                     context={
                         "connector_path": self._get_connector_path(),
+                        "discovered_console_scripts": self._discover_console_script_names(),
                     },
                 )
 
             # If the connector path does not exist, uninstall and re-install.
             # This is sometimes caused by a failed or partial installation.
             print(
-                "Connector executable not found within the virtual environment "
-                f"at {self._get_connector_path()!s}.\nReinstalling...",
+                "Connector executable not found in virtual environment bin directory "
+                f"{get_bin_dir(self._get_venv_path())!s}.\nReinstalling...",
                 file=sys.stderr,
             )
             self.uninstall()
@@ -304,15 +379,15 @@ def ensure_installation(
 
         # By now, everything should be installed. Raise an exception if not.
 
-        connector_path = self._get_connector_path()
-        if not connector_path.exists():
+        if not self._resolve_console_script_name():
             raise exc.AirbyteConnectorInstallationError(
                 message="Connector's executable could not be found within the virtual environment.",
                 connector_name=self.name,
                 context={
                     "connector_path": self._get_connector_path(),
+                    "discovered_console_scripts": self._discover_console_script_names(),
                 },
-            ) from FileNotFoundError(connector_path)
+            ) from FileNotFoundError(self._get_connector_path())
 
         if self.enforce_version:
             version_after_reinstall: str | None = None
diff --git a/tests/integration_tests/fixtures/source-wrong-exe-ambiguous/setup.py b/tests/integration_tests/fixtures/source-wrong-exe-ambiguous/setup.py
new file mode 100644
index 000000000..024e7e4aa
--- /dev/null
+++ b/tests/integration_tests/fixtures/source-wrong-exe-ambiguous/setup.py
@@ -0,0 +1,21 @@
+#
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+#
+from __future__ import annotations
+
+from setuptools import setup
+
+setup(
+    name="airbyte-source-wrong-exe",
+    version="0.0.1",
+    description="Test Source with ambiguous executable names",
+    author="Airbyte",
+    author_email="contact@airbyte.io",
+    packages=["source_wrong_exe_ambiguous"],
+    entry_points={
+        "console_scripts": [
+            "helper-script-a=source_wrong_exe_ambiguous.run:run",
+            "helper-script-b=source_wrong_exe_ambiguous.run:run",
+        ],
+    },
+)
diff --git a/tests/integration_tests/fixtures/source-wrong-exe-ambiguous/source_wrong_exe_ambiguous/__init__.py b/tests/integration_tests/fixtures/source-wrong-exe-ambiguous/source_wrong_exe_ambiguous/__init__.py
new file mode 100644
index 000000000..f70ecfc3a
--- /dev/null
+++ b/tests/integration_tests/fixtures/source-wrong-exe-ambiguous/source_wrong_exe_ambiguous/__init__.py
@@ -0,0 +1 @@
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
diff --git a/tests/integration_tests/fixtures/source-wrong-exe-ambiguous/source_wrong_exe_ambiguous/run.py b/tests/integration_tests/fixtures/source-wrong-exe-ambiguous/source_wrong_exe_ambiguous/run.py
new file mode 100644
index 000000000..4c757d9f8
--- /dev/null
+++ b/tests/integration_tests/fixtures/source-wrong-exe-ambiguous/source_wrong_exe_ambiguous/run.py
@@ -0,0 +1,24 @@
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+from __future__ import annotations
+
+import json
+import sys
+
+sample_spec = {
+    "type": "SPEC",
+    "spec": {
+        "documentationUrl": "https://example.com",
+        "connectionSpecification": {
+            "$schema": "http://json-schema.org/draft-07/schema#",
+            "type": "object",
+            "properties": {
+                "apiKey": {"type": "string"},
+            },
+        },
+    },
+}
+
+
+def run() -> None:
+    if len(sys.argv) > 1 and sys.argv[1] == "spec":
+        print(json.dumps(sample_spec))
diff --git a/tests/integration_tests/fixtures/source-wrong-exe-normalized/setup.py b/tests/integration_tests/fixtures/source-wrong-exe-normalized/setup.py
new file mode 100644
index 000000000..0a38f218d
--- /dev/null
+++ b/tests/integration_tests/fixtures/source-wrong-exe-normalized/setup.py
@@ -0,0 +1,20 @@
+#
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+#
+from __future__ import annotations
+
+from setuptools import setup
+
+setup(
+    name="Airbyte_Source_Wrong_Exe",
+    version="0.0.1",
+    description="Test Source with normalized distribution name",
+    author="Airbyte",
+    author_email="contact@airbyte.io",
+    packages=["source_wrong_exe_normalized"],
+    entry_points={
+        "console_scripts": [
+            "normalized-script-name=source_wrong_exe_normalized.run:run",
+        ],
+    },
+)
diff --git a/tests/integration_tests/fixtures/source-wrong-exe-normalized/source_wrong_exe_normalized/__init__.py b/tests/integration_tests/fixtures/source-wrong-exe-normalized/source_wrong_exe_normalized/__init__.py
new file mode 100644
index 000000000..f70ecfc3a
--- /dev/null
+++ b/tests/integration_tests/fixtures/source-wrong-exe-normalized/source_wrong_exe_normalized/__init__.py
@@ -0,0 +1 @@
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
diff --git a/tests/integration_tests/fixtures/source-wrong-exe-normalized/source_wrong_exe_normalized/run.py b/tests/integration_tests/fixtures/source-wrong-exe-normalized/source_wrong_exe_normalized/run.py
new file mode 100644
index 000000000..4c757d9f8
--- /dev/null
+++ b/tests/integration_tests/fixtures/source-wrong-exe-normalized/source_wrong_exe_normalized/run.py
@@ -0,0 +1,24 @@
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+from __future__ import annotations
+
+import json
+import sys
+
+sample_spec = {
+    "type": "SPEC",
+    "spec": {
+        "documentationUrl": "https://example.com",
+        "connectionSpecification": {
+            "$schema": "http://json-schema.org/draft-07/schema#",
+            "type": "object",
+            "properties": {
+                "apiKey": {"type": "string"},
+            },
+        },
+    },
+}
+
+
+def run() -> None:
+    if len(sys.argv) > 1 and sys.argv[1] == "spec":
+        print(json.dumps(sample_spec))
diff --git a/tests/integration_tests/fixtures/source-wrong-exe/setup.py b/tests/integration_tests/fixtures/source-wrong-exe/setup.py
new file mode 100644
index 000000000..87a370024
--- /dev/null
+++ b/tests/integration_tests/fixtures/source-wrong-exe/setup.py
@@ -0,0 +1,21 @@
+#
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+#
+from __future__ import annotations
+
+from setuptools import setup
+
+# Intentionally mismatched console script name — regression fixture for issue #290.
+setup(
+    name="airbyte-source-wrong-exe",
+    version="0.0.1",
+    description="Test Source with mismatched executable name",
+    author="Airbyte",
+    author_email="contact@airbyte.io",
+    packages=["source_wrong_exe"],
+    entry_points={
+        "console_scripts": [
+            "wrong-script-name=source_wrong_exe.run:run",
+        ],
+    },
+)
diff --git a/tests/integration_tests/fixtures/source-wrong-exe/source_wrong_exe/__init__.py b/tests/integration_tests/fixtures/source-wrong-exe/source_wrong_exe/__init__.py
new file mode 100644
index 000000000..f70ecfc3a
--- /dev/null
+++ b/tests/integration_tests/fixtures/source-wrong-exe/source_wrong_exe/__init__.py
@@ -0,0 +1 @@
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
diff --git a/tests/integration_tests/fixtures/source-wrong-exe/source_wrong_exe/run.py b/tests/integration_tests/fixtures/source-wrong-exe/source_wrong_exe/run.py
new file mode 100644
index 000000000..4c757d9f8
--- /dev/null
+++ b/tests/integration_tests/fixtures/source-wrong-exe/source_wrong_exe/run.py
@@ -0,0 +1,24 @@
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+from __future__ import annotations
+
+import json
+import sys
+
+sample_spec = {
+    "type": "SPEC",
+    "spec": {
+        "documentationUrl": "https://example.com",
+        "connectionSpecification": {
+            "$schema": "http://json-schema.org/draft-07/schema#",
+            "type": "object",
+            "properties": {
+                "apiKey": {"type": "string"},
+            },
+        },
+    },
+}
+
+
+def run() -> None:
+    if len(sys.argv) > 1 and sys.argv[1] == "spec":
+        print(json.dumps(sample_spec))
diff --git a/tests/integration_tests/test_console_script_discovery.py b/tests/integration_tests/test_console_script_discovery.py
new file mode 100644
index 000000000..6cb2da957
--- /dev/null
+++ b/tests/integration_tests/test_console_script_discovery.py
@@ -0,0 +1,72 @@
+# Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+"""Regression tests for console script discovery.
+
+See https://github.com/airbytehq/PyAirbyte/issues/290.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from airbyte._executors.python import VenvExecutor
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+FIXTURE_DIR = REPO_ROOT / "tests/integration_tests/fixtures"
+
+
+@pytest.fixture(autouse=True)
+def _use_uv_for_install(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Local connector installs require uv in this environment."""
+    monkeypatch.setattr("airbyte._executors.python.NO_UV", False)
+
+
+@pytest.mark.parametrize(
+    "fixture_name, expected_script_name",
+    [
+        pytest.param(
+            "source-wrong-exe", "wrong-script-name", id="exact-distribution-name"
+        ),
+        pytest.param(
+            "source-wrong-exe-normalized",
+            "normalized-script-name",
+            id="normalized-distribution-name",
+        ),
+    ],
+)
+def test_discovers_installed_console_script_with_different_name(
+    tmp_path: Path,
+    fixture_name: str,
+    expected_script_name: str,
+) -> None:
+    executor = VenvExecutor(
+        name="source-wrong-exe",
+        pip_url=str(FIXTURE_DIR / fixture_name),
+        install_root=tmp_path,
+    )
+
+    executor.install()
+    executor.ensure_installation()
+
+    assert executor._resolve_console_script_name() == expected_script_name  # noqa: SLF001
+    assert executor._get_connector_path().name == expected_script_name  # noqa: SLF001
+
+    executor._get_connector_path().unlink()  # noqa: SLF001
+    assert executor._resolve_console_script_name() is None  # noqa: SLF001
+
+
+def test_declines_ambiguous_console_script_discovery(tmp_path: Path) -> None:
+    executor = VenvExecutor(
+        name="source-wrong-exe",
+        pip_url=str(FIXTURE_DIR / "source-wrong-exe-ambiguous"),
+        install_root=tmp_path,
+    )
+
+    executor.install()
+
+    assert executor._discover_console_script_names() == [  # noqa: SLF001
+        "helper-script-a",
+        "helper-script-b",
+    ]
+    assert executor._resolve_console_script_name() is None  # noqa: SLF001