feat(mcp): include job duration and end time in sync job results

airbyte/cloud/models.py

@@ -57,6 +57,8 @@ class _JobResponseLike(Protocol):
     bytes_synced: int | None
     rows_synced: int | None
     start_time: str
+    duration: str | None
+    last_updated_at: str | None
 
 
 class ConnectionStatus(StrEnum):
@@ -470,6 +472,12 @@ class CloudJobInfo(BaseModel):
     start_time: str
     """The job start time."""
 
+    duration: str | None = None
+    """The job's ISO 8601 duration (e.g. `PT1M9S`), if available."""
+
+    last_updated_at: str | None = None
+    """When the job was last updated, if available. For a finished job, this is when it ended."""
+
     @classmethod
     def from_api_response(cls, job: _JobResponseLike) -> CloudJobInfo:
         """Create a public model from an internal API job response."""
@@ -479,6 +487,8 @@ def from_api_response(cls, job: _JobResponseLike) -> CloudJobInfo:
             bytes_synced=job.bytes_synced,
             rows_synced=job.rows_synced,
             start_time=job.start_time,
+            duration=job.duration,
+            last_updated_at=job.last_updated_at,
         )
 
 

airbyte/cloud/sync_results.py

@@ -337,6 +337,19 @@ def start_time(self) -> datetime:
                     return ab_datetime_parse(raw_start_time)
             raise
 
+    @property
+    def duration(self) -> str | None:
+        """The job's ISO 8601 duration (e.g. `PT1M9S`), if available."""
+        return self._fetch_latest_job_info().duration
+
+    @property
+    def last_updated_at(self) -> str | None:
+        """When the job was last updated, if available.
+
+        For a finished job, this is when it ended.
+        """
+        return self._fetch_latest_job_info().last_updated_at
+
     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:

airbyte/mcp/cloud.py

@@ -535,6 +535,11 @@ class SyncJobResult(BaseModel):
     """Number of records synced in this job."""
     start_time: str
     """ISO 8601 timestamp of when the job started."""
+    duration: str | None = None
+    """ISO 8601 duration of the job (e.g. `PT1M9S` is 1 minute 9 seconds), or null if Cloud has
+    not reported one (e.g. the job is still running)."""
+    last_updated_at: str | None = None
+    """ISO 8601 timestamp of the job's last update. For a finished job, this is when it ended."""
     job_url: str
     """URL to view the job in Airbyte Cloud."""
 
@@ -1130,6 +1135,8 @@ def get_cloud_sync_status(
         "bytes_synced": sync_result.bytes_synced,
         "records_synced": sync_result.records_synced,
         "start_time": sync_result.start_time.isoformat(),
+        "duration": sync_result.duration,
+        "last_updated_at": sync_result.last_updated_at,
         "job_url": sync_result.job_url,
         "attempts": [],
     }
@@ -1232,6 +1239,8 @@ def list_cloud_sync_jobs(
             bytes_synced=sync_result.bytes_synced,
             records_synced=sync_result.records_synced,
             start_time=sync_result.start_time.isoformat(),
+            duration=sync_result.duration,
+            last_updated_at=sync_result.last_updated_at,
             job_url=sync_result.job_url,
         )
         for sync_result in sync_results

tests/unit_tests/test_cloud_api_util.py

@@ -1573,6 +1573,30 @@ def test_sdk_decodes_queued_job_status() -> None:
     assert CloudJobInfo.from_api_response(job).status is JobStatusEnum.QUEUED
 
 
+def test_cloud_job_info_preserves_duration_and_last_updated_at() -> None:
+    payload = {"data": [_raw_job_response(42, "succeeded")]}
+    payload["data"][0].update(
+        duration="PT1M9S",
+        lastUpdatedAt="2026-10-19T01:18:28Z",
+    )
+    job = utils.unmarshal_json(json.dumps(payload), models.JobsResponse).data[0]
+
+    job_info = CloudJobInfo.from_api_response(job)
+
+    assert job_info.duration == "PT1M9S"
+    assert job_info.last_updated_at == "2026-10-19T01:18:28Z"
+
+
+def test_cloud_job_info_defaults_missing_duration_and_last_updated_at_to_none() -> None:
+    payload = {"data": [_raw_job_response(42, "running")]}
+    job = utils.unmarshal_json(json.dumps(payload), models.JobsResponse).data[0]
+
+    job_info = CloudJobInfo.from_api_response(job)
+
+    assert job_info.duration is None
+    assert job_info.last_updated_at is None
+
+
 def test_get_job_logs_paginates_until_limit(monkeypatch: pytest.MonkeyPatch) -> None:
     """Verify job log pagination stops after collecting the requested limit."""
     captured_requests: list[api.ListJobsRequest] = []

tests/unit_tests/test_mcp_cloud.py

@@ -72,6 +72,8 @@ class _SyncResultLike:
     start_time: datetime
     bytes_synced: int = 0
     records_synced: int = 0
+    duration: str | None = None
+    last_updated_at: str | None = None
     job_url: str = "https://cloud.airbyte.com/jobs"
 
     def get_job_status(self) -> JobStatusEnum:
@@ -1061,6 +1063,71 @@ def _patch_sync_status_workspace(
     return workspace
 
 
+def test_list_cloud_sync_jobs_includes_duration_and_last_updated_at(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Finished jobs include Cloud timing metadata; running jobs may omit it."""
+    finished_job = _SyncResultLike(
+        job_id=42,
+        status=JobStatusEnum.SUCCEEDED,
+        start_time=datetime(2026, 10, 19, 1, 17, 19, tzinfo=timezone.utc),
+        duration="PT1M9S",
+        last_updated_at="2026-10-19T01:18:28Z",
+    )
+    running_job = _SyncResultLike(
+        job_id=43,
+        status=JobStatusEnum.RUNNING,
+        start_time=datetime(2026, 10, 19, 1, 17, 19, tzinfo=timezone.utc),
+    )
+    connection = SimpleNamespace(
+        get_previous_sync_logs=MagicMock(return_value=[finished_job, running_job])
+    )
+    _patch_sync_status_workspace(monkeypatch, connection)
+
+    result = cloud_mcp.list_cloud_sync_jobs(
+        ctx=cast(Context, object()),
+        connection_id="connection-1",
+        workspace_id=None,
+        max_jobs=20,
+        from_tail=None,
+        job_type=None,
+    )
+
+    assert result.jobs[0].duration == "PT1M9S"
+    assert result.jobs[0].last_updated_at == "2026-10-19T01:18:28Z"
+    assert result.jobs[1].duration is None
+    assert result.jobs[1].last_updated_at is None
+
+
+def test_get_cloud_sync_status_includes_duration_and_last_updated_at(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The status response includes the Cloud job duration and update timestamp."""
+    sync_result = SimpleNamespace(
+        job_id=42,
+        get_job_status=MagicMock(return_value=JobStatusEnum.SUCCEEDED),
+        bytes_synced=123,
+        records_synced=456,
+        start_time=datetime(2026, 10, 19, 1, 17, 19, tzinfo=timezone.utc),
+        duration="PT1M9S",
+        last_updated_at="2026-10-19T01:18:28Z",
+        job_url="https://cloud.airbyte.com/jobs/42",
+    )
+    connection = SimpleNamespace(get_sync_result=MagicMock(return_value=sync_result))
+    _patch_sync_status_workspace(monkeypatch, connection)
+
+    result = cloud_mcp.get_cloud_sync_status(
+        ctx=cast(Context, object()),
+        connection_id="connection-1",
+        job_id=42,
+        workspace_id=None,
+        include_attempts=False,
+    )
+
+    assert result["duration"] == "PT1M9S"
+    assert result["last_updated_at"] == "2026-10-19T01:18:28Z"
+
+
 @pytest.mark.parametrize(
     "failure_stage",
     [
@@ -1086,6 +1153,8 @@ def test_get_cloud_sync_status_surfaces_library_errors(
             bytes_synced=0,
             records_synced=0,
             start_time=datetime(2026, 1, 2, 3, 4, 5, tzinfo=timezone.utc),
+            duration=None,
+            last_updated_at=None,
             job_url="https://cloud.airbyte.com/jobs",
             get_attempts=get_attempts,
         )