feat(mcp): add async source catalog discovery and comparison (discover_cloud_source_catalog)

airbyte/_util/api_util.py

@@ -3658,6 +3658,124 @@ def get_connection_catalog(
     )
 
 
+def run_source_discover_command(
+    source_id: str,
+    discover_request_id: str,
+    *,
+    api_root: str,
+    client_id: SecretString | None,
+    client_secret: SecretString | None,
+    bearer_token: SecretString | None,
+    config_api_root: str | None = None,
+) -> dict[str, Any]:
+    """Start a source discover command.
+
+    Uses the Config API endpoint: POST /v1/commands/run/discover.
+    """
+    return _make_config_api_request(
+        path="/commands/run/discover",
+        json={"id": discover_request_id, "actor_id": source_id},
+        api_root=api_root,
+        config_api_root=config_api_root,
+        client_id=client_id,
+        client_secret=client_secret,
+        bearer_token=bearer_token,
+    )
+
+
+def get_command_status(
+    command_id: str,
+    *,
+    api_root: str,
+    client_id: SecretString | None,
+    client_secret: SecretString | None,
+    bearer_token: SecretString | None,
+    config_api_root: str | None = None,
+) -> dict[str, Any]:
+    """Get the status of a Config API command."""
+    return _make_config_api_request(
+        path="/commands/status",
+        json={"id": command_id},
+        api_root=api_root,
+        config_api_root=config_api_root,
+        client_id=client_id,
+        client_secret=client_secret,
+        bearer_token=bearer_token,
+    )
+
+
+def get_discover_command_output(
+    command_id: str,
+    *,
+    api_root: str,
+    client_id: SecretString | None,
+    client_secret: SecretString | None,
+    bearer_token: SecretString | None,
+    config_api_root: str | None = None,
+) -> dict[str, Any]:
+    """Get the output of a completed source discover command."""
+    return _make_config_api_request(
+        path="/commands/output/discover",
+        json={"id": command_id},
+        api_root=api_root,
+        config_api_root=config_api_root,
+        client_id=client_id,
+        client_secret=client_secret,
+        bearer_token=bearer_token,
+    )
+
+
+def get_most_recent_source_catalog(
+    source_id: str,
+    *,
+    api_root: str,
+    client_id: SecretString | None,
+    client_secret: SecretString | None,
+    bearer_token: SecretString | None,
+    config_api_root: str | None = None,
+) -> dict[str, Any]:
+    """Get the most recent saved source catalog."""
+    return _make_config_api_request(
+        path="/sources/most_recent_source_actor_catalog",
+        json={"sourceId": source_id},
+        api_root=api_root,
+        config_api_root=config_api_root,
+        client_id=client_id,
+        client_secret=client_secret,
+        bearer_token=bearer_token,
+    )
+
+
+def get_catalog_diff(
+    current_catalog_id: str,
+    new_catalog_id: str,
+    *,
+    api_root: str,
+    client_id: SecretString | None,
+    client_secret: SecretString | None,
+    bearer_token: SecretString | None,
+    connection_id: str | None = None,
+    config_api_root: str | None = None,
+) -> dict[str, Any]:
+    """Compare two source catalogs without persisting changes to a connection."""
+    request_body: dict[str, Any] = {
+        "current_catalog_id": current_catalog_id,
+        "new_catalog_id": new_catalog_id,
+        "persist_changes": False,
+    }
+    if connection_id is not None:
+        request_body["connection_id"] = connection_id
+    return _make_config_api_request(
+        path="/catalogs/diff",
+        json=request_body,
+        api_root=api_root,
+        config_api_root=config_api_root,
+        client_id=client_id,
+        client_secret=client_secret,
+        bearer_token=bearer_token,
+    )
+
+
 def replace_connection_catalog(
     connection_id: str,
     configured_catalog_dict: dict[str, Any],

airbyte/cloud/__init__.py

@@ -94,6 +94,8 @@
 from airbyte.cloud.connections import CloudConnection
 from airbyte.cloud.models import (
     CloudDefaultContextInfo,
+    CloudSourceCatalogDiffStreamSummary,
+    CloudSourceCatalogDiscoverResult,
     CloudWorkspaceInfo,
     ConnectionStatus,
     JobStatusEnum,
@@ -133,6 +135,8 @@
     "CloudOrganization",
     "CloudWorkspace",
     "CloudConnection",
+    "CloudSourceCatalogDiffStreamSummary",
+    "CloudSourceCatalogDiscoverResult",
     "CloudClientConfig",
     "CloudDefaultContextInfo",
     "CloudWorkspaceInfo",

airbyte/cloud/connections.py

@@ -902,6 +902,7 @@ def import_raw_catalog(self, catalog: dict[str, Any]) -> None:
             bearer_token=self.workspace.bearer_token,
             config_api_root=self.workspace.config_api_root,
         )
+        self._connection_info = None
 
     def rename(self, name: str) -> CloudConnection:
         """Rename the connection.

airbyte/cloud/connectors.py

@@ -55,9 +55,11 @@
 
 from __future__ import annotations
 
+from datetime import UTC, datetime
 from http import HTTPStatus
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, ClassVar, Literal, Protocol
+from uuid import uuid4
 
 import jsonschema
 import requests
@@ -92,6 +94,8 @@
     CheckResult,
     CloudCustomSourceDefinitionInfo,
     CloudDestinationInfo,
+    CloudSourceCatalogDiffStreamSummary,
+    CloudSourceCatalogDiscoverResult,
     CloudSourceInfo,
     ConnectorFeature,
     ConnectorType,
@@ -129,6 +133,92 @@
 """409 messages that carry no Cloud detail: the platform default and PyAirbyte's fallback."""
 
 
+def _get_catalog_discover_result(
+    workspace: CloudWorkspace,
+    discover_request_id: str,
+) -> CloudSourceCatalogDiscoverResult:
+    """Get a source-discover result using a workspace's Config API credentials."""
+    if not isinstance(discover_request_id, str) or not discover_request_id.strip():
+        raise exc.AirbyteLibInputError(
+            message="`discover_request_id` must be a non-empty string.",
+            context={"discover_request_id": discover_request_id},
+        )
+
+    command_status = api_util.get_command_status(
+        discover_request_id,
+        api_root=workspace.api_root,
+        config_api_root=workspace.config_api_root,
+        client_id=workspace.client_id,
+        client_secret=workspace.client_secret,
+        bearer_token=workspace.bearer_token,
+    )
+    request_status = command_status["status"]
+    if request_status != "completed":
+        return CloudSourceCatalogDiscoverResult(
+            discover_request_id=discover_request_id,
+            request_status=request_status,
+        )
+
+    output = api_util.get_discover_command_output(
+        discover_request_id,
+        api_root=workspace.api_root,
+        config_api_root=workspace.config_api_root,
+        client_id=workspace.client_id,
+        client_secret=workspace.client_secret,
+        bearer_token=workspace.bearer_token,
+    )
+    failure = output.get("failureReason")
+    failure_reason = failure.get("externalMessage") if isinstance(failure, dict) else None
+    catalog = output.get("catalog")
+    stream_count = len(catalog["streams"]) if catalog is not None else None
+    return CloudSourceCatalogDiscoverResult(
+        discover_request_id=discover_request_id,
+        request_status=request_status,
+        outcome=output["status"],
+        source_catalog_id=output.get("catalogId"),
+        stream_count=stream_count,
+        failure_reason=failure_reason,
+    )
+
+
+def _build_catalog_diff_result(
+    source_catalog_id: str,
+    response: dict[str, Any],
+) -> CloudSourceCatalogDiscoverResult:
+    """Build a bounded summary of the Config API catalog-diff response."""
+    transforms = response["catalog_diff"]["transforms"]
+    summaries = []
+    for transform in transforms:
+        descriptor = transform["streamDescriptor"]
+        update_stream = transform.get("updateStream") or {}
+        field_transforms = update_stream.get("fieldTransforms", [])
+        attribute_transforms = update_stream.get("streamAttributeTransforms", [])
+        breaking_flags = [
+            change["breaking"]
+            for change in [*field_transforms, *attribute_transforms]
+            if "breaking" in change
+        ]
+        summaries.append(
+            CloudSourceCatalogDiffStreamSummary(
+                stream_name=descriptor["name"],
+                stream_namespace=descriptor.get("namespace"),
+                transform_type=transform["transformType"],
+                breaking_flags=breaking_flags,
+                field_additions_count=sum(
+                    change["transformType"] == "add_field" for change in field_transforms
+                ),
+                field_removals_count=sum(
+                    change["transformType"] == "remove_field" for change in field_transforms
+                ),
+            )
+        )
+    return CloudSourceCatalogDiscoverResult(
+        source_catalog_id=source_catalog_id,
+        catalog_diff_summary=summaries,
+        schema_change=response["schema_change"],
+    )
+
+
 class _ConnectorDefinitionLike(Protocol):
     """The connector definition fields PyAirbyte reads from the Cloud public API."""
 
@@ -1085,6 +1175,68 @@ def _fetch_connector_info(self) -> CloudSourceInfo:
             )
         )
 
+    def _validate_source_id(self) -> None:
+        if not self.source_id.strip():
+            raise exc.AirbyteLibInputError(
+                message="`source_id` must be a non-empty string.",
+                context={"source_id": self.source_id},
+            )
+
+    def start_catalog_discover(self) -> CloudSourceCatalogDiscoverResult:
+        """Start an asynchronous source catalog discover command."""
+        self._validate_source_id()
+        discover_request_id = str(uuid4())
+        api_util.run_source_discover_command(
+            source_id=self.source_id,
+            discover_request_id=discover_request_id,
+            api_root=self.workspace.api_root,
+            config_api_root=self.workspace.config_api_root,
+            client_id=self.workspace.client_id,
+            client_secret=self.workspace.client_secret,
+            bearer_token=self.workspace.bearer_token,
+        )
+        return CloudSourceCatalogDiscoverResult(
+            source_id=self.source_id,
+            discover_request_id=discover_request_id,
+            request_status="pending",
+        )
+
+    def get_catalog_discover_result(
+        self,
+        discover_request_id: str,
+    ) -> CloudSourceCatalogDiscoverResult:
+        """Get the status and completed output of a source catalog discover."""
+        result = _get_catalog_discover_result(
+            workspace=self.workspace,
+            discover_request_id=discover_request_id,
+        )
+        return result.model_copy(update={"source_id": self.source_id})
+
+    def get_latest_saved_catalog(self) -> CloudSourceCatalogDiscoverResult:
+        """Get the timestamp and stream count of the latest saved catalog."""
+        self._validate_source_id()
+        response = api_util.get_most_recent_source_catalog(
+            source_id=self.source_id,
+            api_root=self.workspace.api_root,
+            config_api_root=self.workspace.config_api_root,
+            client_id=self.workspace.client_id,
+            client_secret=self.workspace.client_secret,
+            bearer_token=self.workspace.bearer_token,
+        )
+        catalog = response.get("catalog")
+        streams = catalog.get("streams", []) if isinstance(catalog, dict) else []
+        if not streams:
+            return CloudSourceCatalogDiscoverResult(source_id=self.source_id)
+
+        updated_at = response.get("updatedAt")
+        return CloudSourceCatalogDiscoverResult(
+            source_id=self.source_id,
+            stream_count=len(streams),
+            updated_at=(
+                datetime.fromtimestamp(updated_at, tz=UTC) if updated_at is not None else None
+            ),
+        )
+
     def rename(self, name: str) -> CloudSource:
         """Rename the source.
 

airbyte/cloud/models.py

@@ -5,6 +5,7 @@
 
 from collections.abc import Mapping
 from dataclasses import asdict, dataclass, is_dataclass
+from datetime import datetime  # noqa: TC003  # Needed at runtime for Pydantic field types.
 from enum import Enum
 from typing import Any, Literal, Protocol
 
@@ -509,6 +510,32 @@ def from_api_response(cls, source: _SourceResponseLike) -> CloudSourceInfo:
         )
 
 
+class CloudSourceCatalogDiffStreamSummary(BaseModel):
+    """Summary of one stream transformation in a catalog diff."""
+
+    stream_name: str
+    stream_namespace: str | None = None
+    transform_type: str
+    breaking_flags: list[bool] = Field(default_factory=list)
+    field_additions_count: int = 0
+    field_removals_count: int = 0
+
+
+class CloudSourceCatalogDiscoverResult(BaseModel):
+    """Status or outcome of discovering a Cloud source catalog."""
+
+    source_id: str | None = None
+    discover_request_id: str | None = None
+    request_status: Literal["pending", "running", "completed", "cancelled"] | None = None
+    outcome: Literal["succeeded", "failed"] | None = None
+    source_catalog_id: str | None = None
+    stream_count: int | None = None
+    updated_at: datetime | None = None
+    failure_reason: str | None = None
+    catalog_diff_summary: list[CloudSourceCatalogDiffStreamSummary] | None = None
+    schema_change: Literal["no_change", "non_breaking", "breaking"] | None = None
+
+
 class CloudDestinationInfo(BaseModel):
     """Information about an Airbyte Cloud destination."""