feat(mcp): Add job_type filtering support to list_cloud_sync_jobs (#974)

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

Author: aaronsteers

airbyte/_util/api_util.py

@@ -587,6 +587,7 @@ def get_job_logs(  # noqa: PLR0913  # Too many arguments - needed for auth flexi
     bearer_token: SecretString | None,
     offset: int | None = None,
     order_by: str | None = None,
+    job_type: models.JobTypeEnum | None = None,
 ) -> list[models.JobResponse]:
     """Get a list of jobs for a connection.
 
@@ -600,6 +601,8 @@ def get_job_logs(  # noqa: PLR0913  # Too many arguments - needed for auth flexi
         bearer_token: Bearer token for authentication (alternative to client credentials).
         offset: Number of jobs to skip from the beginning. Defaults to None (0).
         order_by: Field and direction to order by (e.g., "createdAt|DESC"). Defaults to None.
+        job_type: Filter by job type (e.g., JobTypeEnum.SYNC, JobTypeEnum.REFRESH).
+            If not specified, defaults to sync and reset jobs only (API default behavior).
 
     Returns:
         A list of JobResponse objects.
@@ -617,6 +620,7 @@ def get_job_logs(  # noqa: PLR0913  # Too many arguments - needed for auth flexi
             limit=limit,
             offset=offset,
             order_by=order_by,
+            job_type=job_type,
         ),
     )
     if status_ok(response.status_code) and response.jobs_response:

airbyte/cloud/connections.py

@@ -12,7 +12,7 @@
 
 
 if TYPE_CHECKING:
-    from airbyte_api.models import ConnectionResponse, JobResponse
+    from airbyte_api.models import ConnectionResponse, JobResponse, JobTypeEnum
 
     from airbyte.cloud.workspaces import CloudWorkspace
 
@@ -291,6 +291,7 @@ def get_previous_sync_logs(
         limit: int = 20,
         offset: int | None = None,
         from_tail: bool = True,
+        job_type: JobTypeEnum | None = None,
     ) -> list[SyncResult]:
         """Get previous sync jobs for a connection with pagination support.
 
@@ -304,6 +305,8 @@ def get_previous_sync_logs(
             from_tail: If True, returns jobs ordered newest-first (createdAt DESC).
                 If False, returns jobs ordered oldest-first (createdAt ASC).
                 Defaults to True.
+            job_type: Filter by job type (e.g., JobTypeEnum.SYNC, JobTypeEnum.REFRESH).
+                If not specified, defaults to sync and reset jobs only (API default behavior).
 
         Returns:
             A list of SyncResult objects representing the sync jobs.
@@ -320,6 +323,7 @@ def get_previous_sync_logs(
             limit=limit,
             offset=offset,
             order_by=order_by,
+            job_type=job_type,
             client_id=self.workspace.client_id,
             client_secret=self.workspace.client_secret,
             bearer_token=self.workspace.bearer_token,

airbyte/mcp/cloud.py

@@ -4,6 +4,7 @@
 from pathlib import Path
 from typing import Annotated, Any, Literal, cast
 
+from airbyte_api.models import JobTypeEnum
 from fastmcp import Context, FastMCP
 from fastmcp_extensions import get_mcp_config, mcp_tool, register_mcp_tools
 from pydantic import BaseModel, Field
@@ -737,6 +738,17 @@ def list_cloud_sync_jobs(
             default=None,
         ),
     ],
+    job_type: Annotated[
+        JobTypeEnum | None,
+        Field(
+            description=(
+                "Filter by job type. Options: 'sync', 'reset', 'refresh', 'clear'. "
+                "If not specified, defaults to sync and reset jobs only (API default). "
+                "Use 'refresh' to find refresh jobs or 'clear' to find clear jobs."
+            ),
+            default=None,
+        ),
+    ],
 ) -> SyncJobListResult:
     """List sync jobs for a connection with pagination support.
 
@@ -767,6 +779,7 @@ def list_cloud_sync_jobs(
         limit=effective_limit,
         offset=jobs_offset,
         from_tail=from_tail,
+        job_type=job_type,
     )
 
     jobs = [