feat(mcp): Add get_connector_version_history tool to registry domain (#857)

Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

Author: aaronsteers

airbyte/__init__.py

@@ -125,18 +125,18 @@
 
 from typing import TYPE_CHECKING
 
+from airbyte import registry
 from airbyte.caches.bigquery import BigQueryCache
 from airbyte.caches.duckdb import DuckDBCache
 from airbyte.caches.util import get_colab_cache, get_default_cache, new_local_cache
 from airbyte.datasets import CachedDataset
 from airbyte.destinations.base import Destination
 from airbyte.destinations.util import get_destination
 from airbyte.records import StreamRecord
+from airbyte.registry import get_available_connectors
 from airbyte.results import ReadResult, WriteResult
 from airbyte.secrets import SecretSourceEnum, get_secret
-from airbyte.sources import registry
 from airbyte.sources.base import Source
-from airbyte.sources.registry import get_available_connectors
 from airbyte.sources.util import get_source
 
 

airbyte/_executors/base.py

@@ -14,7 +14,7 @@
 if TYPE_CHECKING:
     from collections.abc import Generator, Iterable, Iterator
 
-    from airbyte.sources.registry import ConnectorMetadata
+    from airbyte.registry import ConnectorMetadata
 
 
 _LATEST_VERSION = "latest"

airbyte/_executors/noop.py

@@ -17,7 +17,7 @@
     from typing import IO
 
     from airbyte._message_iterators import AirbyteMessageIterator
-    from airbyte.sources.registry import ConnectorMetadata
+    from airbyte.registry import ConnectorMetadata
 
 
 logger = logging.getLogger("airbyte")

airbyte/_executors/python.py

@@ -21,7 +21,7 @@
 
 
 if TYPE_CHECKING:
-    from airbyte.sources.registry import ConnectorMetadata
+    from airbyte.registry import ConnectorMetadata
 
 
 class VenvExecutor(Executor):

airbyte/_executors/util.py

@@ -20,7 +20,7 @@
 from airbyte._util.meta import which
 from airbyte._util.telemetry import EventState, log_install_state  # Non-public API
 from airbyte.constants import AIRBYTE_OFFLINE_MODE, DEFAULT_PROJECT_DIR, TEMP_DIR_OVERRIDE
-from airbyte.sources.registry import ConnectorMetadata, InstallType, get_connector_metadata
+from airbyte.registry import ConnectorMetadata, InstallType, get_connector_metadata
 from airbyte.version import get_version
 
 

airbyte/_registry_utils.py

@@ -0,0 +1,121 @@
+# Copyright (c) 2024 Airbyte, Inc., all rights reserved.
+"""@private Utility functions for working with the Airbyte connector registry."""
+
+import logging
+import re
+
+import requests
+
+
+logger = logging.getLogger("airbyte.registry")
+
+
+def parse_changelog_html(  # noqa: PLR0914
+    html_content: str, connector_name: str
+) -> list[dict[str, str | list[str] | None]]:
+    """Parse changelog HTML to extract version history.
+
+    Returns a list of dicts with keys: version, release_date, docker_image_url,
+    changelog_url, pr_url, pr_title, parsing_errors.
+    """
+    versions: list[dict[str, str | list[str] | None]] = []
+
+    connector_type = "sources" if connector_name.startswith("source-") else "destinations"
+    connector_short_name = connector_name.replace("source-", "").replace("destination-", "")
+
+    changelog_url = (
+        f"https://docs.airbyte.com/integrations/{connector_type}/{connector_short_name}#changelog"
+    )
+
+    row_pattern = re.compile(
+        r"<tr><td[^>]*>([^<]+)<td[^>]*>([^<]+)<td[^>]*>(.*?)<td[^>]*>(.*?)<tr>", re.DOTALL
+    )
+
+    pr_pattern = re.compile(
+        r"<a href=https://github\.com/airbytehq/airbyte/pull/(\d+)[^>]*>(\d+)</a>"
+    )
+
+    for match in row_pattern.finditer(html_content):
+        version = match.group(1).strip()
+        date = match.group(2).strip()
+        pr_cell = match.group(3)
+        subject = match.group(4).strip()
+
+        if not re.match(r"\d{4}-\d{2}-\d{2}", date):
+            continue
+
+        pr_matches = list(pr_pattern.finditer(pr_cell))
+        pr_url = None
+        pr_title = None
+        parsing_errors = []
+
+        if pr_matches:
+            first_pr = pr_matches[0]
+            pr_number = first_pr.group(1)
+            pr_url = f"https://github.com/airbytehq/airbyte/pull/{pr_number}"
+
+            pr_title = re.sub(r"<[^>]+>", "", subject)
+            pr_title = pr_title.replace("&quot;", '"').replace("&amp;", "&")
+            pr_title = pr_title.replace("&lt;", "<").replace("&gt;", ">")
+            pr_title = pr_title.strip()
+
+            if len(pr_matches) > 1:
+                parsing_errors.append(
+                    f"Multiple PRs found for version {version}, using first PR: {pr_number}"
+                )
+        else:
+            parsing_errors.append(f"No PR link found in changelog for version {version}")
+
+        docker_image_url = f"https://hub.docker.com/r/airbyte/{connector_name}/tags?name={version}"
+
+        versions.append(
+            {
+                "version": version,
+                "release_date": date or None,
+                "docker_image_url": docker_image_url,
+                "changelog_url": changelog_url,
+                "pr_url": pr_url,
+                "pr_title": pr_title,
+                "parsing_errors": parsing_errors,
+            }
+        )
+
+    return versions
+
+
+def fetch_registry_version_date(connector_name: str, version: str) -> str | None:
+    """Fetch the release date for a specific version from the registry.
+
+    Returns the release date string (YYYY-MM-DD) if found, None otherwise.
+    """
+    try:  # noqa: PLR1702
+        registry_url = "https://connectors.airbyte.com/files/registries/v0/oss_registry.json"
+        response = requests.get(registry_url, timeout=10)
+        response.raise_for_status()
+        registry_data = response.json()
+
+        connector_list = registry_data.get("sources", []) + registry_data.get("destinations", [])
+
+        for connector in connector_list:
+            docker_repo = connector.get("dockerRepository", "")
+            if docker_repo == f"airbyte/{connector_name}":
+                releases = connector.get("releases", {})
+                release_candidates = releases.get("releaseCandidates", {})
+
+                if version in release_candidates:
+                    version_data = release_candidates[version]
+                    generated = version_data.get("generated", {})
+                    git_info = generated.get("git", {})
+                    commit_timestamp = git_info.get("commit_timestamp")
+
+                    if commit_timestamp:
+                        date_match = re.match(r"(\d{4}-\d{2}-\d{2})", commit_timestamp)
+                        if date_match:
+                            return date_match.group(1)
+
+                break
+        else:
+            return None
+    except Exception as e:
+        logger.debug(f"Failed to fetch registry date for {connector_name} v{version}: {e}")
+    return None

airbyte/_util/registry_spec.py

@@ -10,7 +10,7 @@
 import requests
 
 from airbyte import exceptions as exc
-from airbyte.sources.registry import get_connector_metadata
+from airbyte.registry import get_connector_metadata
 from airbyte.version import get_version
 
 

airbyte/mcp/connector_registry.py

@@ -4,20 +4,32 @@
 # Note: Deferred type evaluation must be avoided due to FastMCP/Pydantic needing
 # types to be available at import time for tool registration.
 import contextlib
+import logging
 from typing import Annotated, Any, Literal
 
+import requests
 from fastmcp import FastMCP
 from pydantic import BaseModel, Field
 
+from airbyte import exceptions as exc
 from airbyte._executors.util import DEFAULT_MANIFEST_URL
 from airbyte._util.meta import is_docker_installed
 from airbyte.mcp._tool_utils import mcp_tool, register_tools
 from airbyte.mcp._util import resolve_list_of_strings
-from airbyte.sources import get_available_connectors
-from airbyte.sources.registry import ConnectorMetadata, InstallType, get_connector_metadata
+from airbyte.registry import (
+    ConnectorMetadata,
+    ConnectorVersionInfo,
+    InstallType,
+    get_available_connectors,
+    get_connector_metadata,
+)
+from airbyte.registry import get_connector_version_history as _get_connector_version_history
 from airbyte.sources.util import get_source
 
 
+logger = logging.getLogger("airbyte.mcp")
+
+
 @mcp_tool(
     domain="registry",
     read_only=True,
@@ -161,6 +173,71 @@ def get_connector_info(
     )
 
 
+@mcp_tool(
+    domain="registry",
+    read_only=True,
+    idempotent=True,
+)
+def get_connector_version_history(
+    connector_name: Annotated[
+        str,
+        Field(
+            description="The name of the connector (e.g., 'source-faker', 'destination-postgres')"
+        ),
+    ],
+    num_versions_to_validate: Annotated[
+        int,
+        Field(
+            description=(
+                "Number of most recent versions to validate with registry data for accurate "
+                "release dates. Defaults to 5."
+            ),
+            default=5,
+        ),
+    ] = 5,
+    limit: Annotated[
+        int | None,
+        Field(
+            description=(
+                "DEPRECATED: Use num_versions_to_validate instead. "
+                "Maximum number of versions to return (most recent first). "
+                "If specified, only the first N versions will be returned."
+            ),
+            default=None,
+        ),
+    ] = None,
+) -> list[ConnectorVersionInfo] | Literal["Connector not found.", "Failed to fetch changelog."]:
+    """Get version history for a connector.
+
+    This tool retrieves the version history for a connector, including:
+    - Version number
+    - Release date (from changelog, with registry override for recent versions)
+    - DockerHub URL for the version
+    - Changelog URL
+    - PR URL and title (scraped from changelog)
+
+    For the most recent N versions (default 5), release dates are fetched from the
+    registry for accuracy. For older versions, changelog dates are used.
+
+    Returns:
+        List of version information, sorted by most recent first.
+    """
+    try:
+        versions = _get_connector_version_history(
+            connector_name=connector_name,
+            num_versions_to_validate=num_versions_to_validate,
+        )
+    except exc.AirbyteConnectorNotRegisteredError:
+        return "Connector not found."
+    except requests.exceptions.RequestException:
+        logger.exception(f"Failed to fetch changelog for {connector_name}")
+        return "Failed to fetch changelog."
+    else:
+        if limit is not None and limit > 0:
+            return versions[:limit]
+        return versions
+
+
 def register_connector_registry_tools(app: FastMCP) -> None:
     """@private Register tools with the FastMCP app.
 

airbyte/mcp/local_ops.py

@@ -15,11 +15,11 @@
 from airbyte.caches.util import get_default_cache
 from airbyte.mcp._tool_utils import mcp_tool, register_tools
 from airbyte.mcp._util import resolve_config, resolve_list_of_strings
+from airbyte.registry import get_connector_metadata
 from airbyte.secrets.config import _get_secret_sources
 from airbyte.secrets.env_vars import DotenvSecretManager
 from airbyte.secrets.google_gsm import GoogleGSMSecretManager
 from airbyte.sources.base import Source
-from airbyte.sources.registry import get_connector_metadata
 
 
 if TYPE_CHECKING: