feat(cloud): Add ability to view Cloud sync logs, including SyncAttempt abstraction for fetching job attempts

airbyte/_util/api_util.py

@@ -14,6 +14,7 @@
 from __future__ import annotations
 
 import json
+from http import HTTPStatus
 from typing import TYPE_CHECKING, Any, Literal
 
 import airbyte_api
@@ -860,18 +861,27 @@ def _make_config_api_request(
         "Authorization": f"Bearer {bearer_token}",
         "User-Agent": "PyAirbyte Client",
     }
+    full_url = config_api_root + path
     response = requests.request(
         method="POST",
-        url=config_api_root + path,
+        url=full_url,
         headers=headers,
         json=json,
     )
     if not status_ok(response.status_code):
         try:
             response.raise_for_status()
         except requests.HTTPError as ex:
+            error_message = f"API request failed with status {response.status_code}"
+            if response.status_code == HTTPStatus.FORBIDDEN:  # 403 error
+                error_message += f" (Forbidden) when accessing: {full_url}"
             raise AirbyteError(
+                message=error_message,
                 context={
+                    "full_url": full_url,
+                    "config_api_root": config_api_root,
+                    "path": path,
+                    "status_code": response.status_code,
                     "url": response.request.url,
                     "body": response.request.body,
                     "response": response.__dict__,

airbyte/cloud/sync_results.py

@@ -103,8 +103,11 @@
 import time
 from collections.abc import Iterator, Mapping
 from dataclasses import asdict, dataclass
-from datetime import datetime
-from typing import TYPE_CHECKING, Any, final
+from typing import TYPE_CHECKING, Any
+
+from typing_extensions import final
+
+from airbyte_cdk.utils.datetime_helpers import ab_datetime_parse
 
 from airbyte._util import api_util
 from airbyte.cloud.constants import FAILED_STATUSES, FINAL_STATUSES
@@ -117,6 +120,8 @@
 """The default timeout for waiting for a sync job to complete, in seconds."""
 
 if TYPE_CHECKING:
+    from datetime import datetime
+
     import sqlalchemy
 
     from airbyte._util.api_imports import ConnectionResponse, JobResponse, JobStatusEnum
@@ -125,6 +130,88 @@
     from airbyte.cloud.workspaces import CloudWorkspace
 
 
+@dataclass
+class SyncAttempt:
+    """Represents a single attempt of a sync job.
+
+    **This class is not meant to be instantiated directly.** Instead, obtain a `SyncAttempt` by
+    calling `.SyncResult.get_attempts()`.
+    """
+
+    workspace: CloudWorkspace
+    connection: CloudConnection
+    job_id: int
+    attempt_number: int
+    _attempt_data: dict[str, Any] | None = None
+
+    @property
+    def attempt_id(self) -> int:
+        """Return the attempt ID."""
+        return self._get_attempt_data()["id"]
+
+    @property
+    def status(self) -> str:
+        """Return the attempt status."""
+        return self._get_attempt_data()["status"]
+
+    @property
+    def bytes_synced(self) -> int:
+        """Return the number of bytes synced in this attempt."""
+        return self._get_attempt_data().get("bytesSynced", 0)
+
+    @property
+    def records_synced(self) -> int:
+        """Return the number of records synced in this attempt."""
+        return self._get_attempt_data().get("recordsSynced", 0)
+
+    @property
+    def created_at(self) -> datetime:
+        """Return the creation time of the attempt."""
+        timestamp = self._get_attempt_data()["createdAt"]
+        return ab_datetime_parse(timestamp)
+
+    def _get_attempt_data(self) -> dict[str, Any]:
+        """Get attempt data from the provided attempt data."""
+        if self._attempt_data is None:
+            raise ValueError(
+                "Attempt data not provided. SyncAttempt should be created via "
+                "SyncResult.get_attempts()."
+            )
+        return self._attempt_data["attempt"]
+
+    def get_full_log_text(self) -> str:
+        """Return the complete log text for this attempt.
+
+        Returns:
+            String containing all log text for this attempt, with lines separated by newlines.
+        """
+        if self._attempt_data is None:
+            return ""
+
+        logs_data = self._attempt_data.get("logs")
+        if not logs_data:
+            return ""
+
+        result = ""
+
+        if "events" in logs_data:
+            log_events = logs_data["events"]
+            if log_events:
+                log_lines = []
+                for event in log_events:
+                    timestamp = event.get("timestamp", "")
+                    level = event.get("level", "INFO")
+                    message = event.get("message", "")
+                    log_lines.append(f"[{timestamp}] {level}: {message}")
+                result = "\n".join(log_lines)
+        elif "logLines" in logs_data:
+            log_lines = logs_data["logLines"]
+            if log_lines:
+                result = "\n".join(log_lines)
+
+        return result
+
+
 @dataclass
 class SyncResult:
     """The result of a sync operation.
@@ -141,6 +228,7 @@ class SyncResult:
     _latest_job_info: JobResponse | None = None
     _connection_response: ConnectionResponse | None = None
     _cache: CacheBase | None = None
+    _job_with_attempts_info: dict[str, Any] | None = None
 
     @property
     def job_url(self) -> str:
@@ -213,8 +301,53 @@ def records_synced(self) -> int:
     @property
     def start_time(self) -> datetime:
         """Return the start time of the sync job in UTC."""
-        # Parse from ISO 8601 format:
-        return datetime.fromisoformat(self._fetch_latest_job_info().start_time)
+        try:
+            return ab_datetime_parse(self._fetch_latest_job_info().start_time)
+        except (ValueError, TypeError) as e:
+            if "Invalid isoformat string" in str(e):
+                job_info_raw = api_util._make_config_api_request(  # noqa: SLF001
+                    api_root=self.workspace.api_root,
+                    path="/jobs/get",
+                    json={"id": self.job_id},
+                    client_id=self.workspace.client_id,
+                    client_secret=self.workspace.client_secret,
+                )
+                raw_start_time = job_info_raw.get("startTime")
+                if raw_start_time:
+                    return ab_datetime_parse(raw_start_time)
+            raise
+
+    def _fetch_job_with_attempts(self) -> dict[str, Any]:
+        """Fetch job info with attempts from Config API using lazy loading pattern."""
+        if self._job_with_attempts_info is not None:
+            return self._job_with_attempts_info
+
+        self._job_with_attempts_info = api_util._make_config_api_request(  # noqa: SLF001  # Config API helper
+            api_root=self.workspace.api_root,
+            path="/jobs/get",
+            json={
+                "id": self.job_id,
+            },
+            client_id=self.workspace.client_id,
+            client_secret=self.workspace.client_secret,
+        )
+        return self._job_with_attempts_info
+
+    def get_attempts(self) -> list[SyncAttempt]:
+        """Return a list of attempts for this sync job."""
+        job_with_attempts = self._fetch_job_with_attempts()
+        attempts_data = job_with_attempts.get("attempts", [])
+
+        return [
+            SyncAttempt(
+                workspace=self.workspace,
+                connection=self.connection,
+                job_id=self.job_id,
+                attempt_number=i,
+                _attempt_data=attempt_data,
+            )
+            for i, attempt_data in enumerate(attempts_data, start=0)
+        ]
 
     def raise_failure_status(
         self,
@@ -362,4 +495,5 @@ def __len__(self) -> int:
 
 __all__ = [
     "SyncResult",
+    "SyncAttempt",
 ]

airbyte/mcp/_cloud_ops.py

@@ -1,13 +1,12 @@
 # Copyright (c) 2024 Airbyte, Inc., all rights reserved.
 """Airbyte Cloud MCP operations."""
 
-from typing import Annotated
+from typing import Annotated, Any
 
 from fastmcp import FastMCP
 from pydantic import Field
 
 from airbyte import cloud, get_destination, get_source
-from airbyte._util.api_imports import JobStatusEnum
 from airbyte.cloud.auth import (
     resolve_cloud_api_url,
     resolve_cloud_client_id,
@@ -343,19 +342,64 @@ def get_cloud_sync_status(
             default=None,
         ),
     ],
-) -> JobStatusEnum | None:
+    *,
+    include_attempts: Annotated[
+        bool,
+        Field(
+            description="Whether to include detailed attempts information.",
+            default=False,
+        ),
+    ],
+) -> dict[str, Any]:
     """Get the status of a sync job from the Airbyte Cloud.
 
     By default, the `AIRBYTE_CLIENT_ID`, `AIRBYTE_CLIENT_SECRET`, `AIRBYTE_WORKSPACE_ID`,
     and `AIRBYTE_API_ROOT` environment variables will be used to authenticate with the
     Airbyte Cloud API.
     """
-    workspace: CloudWorkspace = _get_cloud_workspace()
-    connection = workspace.get_connection(connection_id=connection_id)
+    try:
+        workspace: CloudWorkspace = _get_cloud_workspace()
+        connection = workspace.get_connection(connection_id=connection_id)
 
-    # If a job ID is provided, get the job by ID.
-    sync_result: cloud.SyncResult | None = connection.get_sync_result(job_id=job_id)
-    return sync_result.get_job_status() if sync_result else None
+        # If a job ID is provided, get the job by ID.
+        sync_result: cloud.SyncResult | None = connection.get_sync_result(job_id=job_id)
+
+        if not sync_result:
+            return {"status": None, "job_id": None, "attempts": []}
+
+        result = {
+            "status": sync_result.get_job_status(),
+            "job_id": sync_result.job_id,
+            "bytes_synced": sync_result.bytes_synced,
+            "records_synced": sync_result.records_synced,
+            "start_time": sync_result.start_time.isoformat(),
+            "job_url": sync_result.job_url,
+            "attempts": [],
+        }
+
+        if include_attempts:
+            attempts = sync_result.get_attempts()
+            result["attempts"] = [
+                {
+                    "attempt_number": attempt.attempt_number,
+                    "attempt_id": attempt.attempt_id,
+                    "status": attempt.status,
+                    "bytes_synced": attempt.bytes_synced,
+                    "records_synced": attempt.records_synced,
+                    "created_at": attempt.created_at.isoformat(),
+                }
+                for attempt in attempts
+            ]
+
+        return result  # noqa: TRY300
+
+    except Exception as ex:
+        return {
+            "status": None,
+            "job_id": job_id,
+            "error": f"Failed to get sync status for connection '{connection_id}': {ex}",
+            "attempts": [],
+        }
 
 
 # @app.tool()  # << deferred
@@ -382,6 +426,69 @@ def list_deployed_cloud_destination_connectors() -> list[CloudDestination]:
     return workspace.list_destinations()
 
 
+# @app.tool()  # << deferred
+def get_cloud_sync_logs(
+    connection_id: Annotated[
+        str,
+        Field(description="The ID of the Airbyte Cloud connection."),
+    ],
+    job_id: Annotated[
+        int | None,
+        Field(description="Optional job ID. If not provided, the latest job will be used."),
+    ] = None,
+    attempt_number: Annotated[
+        int | None,
+        Field(
+            description="Optional attempt number. If not provided, the latest attempt will be used."
+        ),
+    ] = None,
+) -> str:
+    """Get the logs from a sync job attempt on Airbyte Cloud.
+
+    By default, the `AIRBYTE_CLIENT_ID`, `AIRBYTE_CLIENT_SECRET`, `AIRBYTE_WORKSPACE_ID`,
+    and `AIRBYTE_API_ROOT` environment variables will be used to authenticate with the
+    Airbyte Cloud API.
+    """
+    try:
+        workspace: CloudWorkspace = _get_cloud_workspace()
+        connection = workspace.get_connection(connection_id=connection_id)
+
+        sync_result: cloud.SyncResult | None = connection.get_sync_result(job_id=job_id)
+
+        if not sync_result:
+            return f"No sync job found for connection '{connection_id}'"
+
+        attempts = sync_result.get_attempts()
+
+        if not attempts:
+            return f"No attempts found for job '{sync_result.job_id}'"
+
+        if attempt_number is not None:
+            target_attempt = None
+            for attempt in attempts:
+                if attempt.attempt_number == attempt_number:
+                    target_attempt = attempt
+                    break
+
+            if target_attempt is None:
+                return f"Attempt number {attempt_number} not found for job '{sync_result.job_id}'"
+        else:
+            target_attempt = max(attempts, key=lambda a: a.attempt_number)
+
+        logs = target_attempt.get_full_log_text()
+
+        if not logs:
+            return (
+                f"No logs available for job '{sync_result.job_id}', "
+                f"attempt {target_attempt.attempt_number}"
+            )
+
+        return logs  # noqa: TRY300
+
+    except Exception as ex:
+        return f"Failed to get logs for connection '{connection_id}': {ex}"
+
+
 # @app.tool()  # << deferred
 def list_deployed_cloud_connections() -> list[CloudConnection]:
     """List all deployed connections in the Airbyte Cloud workspace.
@@ -403,6 +510,7 @@ def register_cloud_ops_tools(app: FastMCP) -> None:
     app.tool(create_connection_on_cloud)
     app.tool(run_cloud_sync)
     app.tool(get_cloud_sync_status)
+    app.tool(get_cloud_sync_logs)
     app.tool(list_deployed_cloud_source_connectors)
     app.tool(list_deployed_cloud_destination_connectors)
     app.tool(list_deployed_cloud_connections)

bin/test_mcp_tool.py

@@ -15,6 +15,10 @@
 
     poe mcp-tool-test check_airbyte_cloud_workspace '{}'
     poe mcp-tool-test list_deployed_cloud_connections '{}'
+    poe mcp-tool-test get_cloud_sync_status \
+        '{"connection_id": "0791e193-811b-4fcf-91c3-f8c5963e74a0", "include_attempts": true}'
+    poe mcp-tool-test get_cloud_sync_logs \
+        '{"connection_id": "0791e193-811b-4fcf-91c3-f8c5963e74a0"}'
 """
 
 import asyncio