From b8ffab5f37a9a53c84c6a683a70dccbd43690943 Mon Sep 17 00:00:00 2001 From: Johnny Schmidt Date: Fri, 16 Oct 2026 12:08:29 -0700 Subject: [PATCH 1/2] feat(mcp): add a Fusion search backfill tool and correct the search tool descriptions --- airbyte/_direct_connectors/api_util.py | 27 +++ airbyte/_direct_connectors/models.py | 102 ++++++++- airbyte/cloud/connections.py | 38 ++++ airbyte/mcp/cloud.py | 106 +++++++++- tests/unit_tests/test_cloud_direct_execute.py | 197 ++++++++++++++++++ tests/unit_tests/test_mcp_cloud.py | 94 ++++++++- 6 files changed, 554 insertions(+), 10 deletions(-) diff --git a/airbyte/_direct_connectors/api_util.py b/airbyte/_direct_connectors/api_util.py index dfcaec795..34db7c517 100644 --- a/airbyte/_direct_connectors/api_util.py +++ b/airbyte/_direct_connectors/api_util.py @@ -19,6 +19,7 @@ from airbyte._direct_connectors.models import ( ConnectorEnablement, ExternalApiExecuteResult, + ExternalSearchBackfillRequestResult, ExternalSearchResult, ExternalSearchStatusResult, ) @@ -290,6 +291,32 @@ def get_cloud_connector_search_status( ) from ex +def request_cloud_connection_search_backfill( + *, + connection_id: str, + request_body: dict[str, Any], + credentials: _AirbyteCredentials, +) -> ExternalSearchBackfillRequestResult: + """Request a search backfill for a Cloud connection via the Cloud Config API. + + Posts to `/connections/{id}/backfill`. The request body is forwarded as-is. + """ + path = f"/connections/{connection_id}/backfill" + response = make_cloud_agent_request( + method="POST", + path=path, + credentials=credentials, + json=request_body, + ) + try: + return ExternalSearchBackfillRequestResult.model_validate(response) + except ValidationError as ex: + raise AirbyteCloudError( + message="Malformed Airbyte Cloud backfill response.", + context={"path": path, "errors": ex.errors(include_url=False)}, + ) from ex + + def get_cloud_connector_enablement( *, connector_id: str, diff --git a/airbyte/_direct_connectors/models.py b/airbyte/_direct_connectors/models.py index 4270a3806..8eb47d6eb 100644 --- a/airbyte/_direct_connectors/models.py +++ b/airbyte/_direct_connectors/models.py @@ -315,13 +315,20 @@ class ExternalSearchHit(BaseModel): """The matched entity's ID.""" entity_data: dict[str, Any] = Field(default_factory=dict) - """The matched entity's data, projected to the requested `fields` when given.""" + """The matched entity's data, projected to the requested `fields` when given. + + Null values are not stored, so an absent key means null or unset. A keyword-only hit + also omits any field longer than 4096 characters or split across keyword chunks; its + matched text is in `context`. + """ score: float """The match score. Higher is a better match.""" context: str | None = None - """The matched text, truncated to `max_context_chars` when requested.""" + """The matched text: the semantic match, or for a keyword-only hit the matched chunk + centered on the first query term, with `…` where text was left out. Truncated to + `max_context_chars` when requested.""" @field_validator("entity_id", mode="before") @classmethod @@ -406,15 +413,17 @@ def _collect_index_errors(self) -> ExternalSearchResult: class ExternalSearchBackfillStatus(BaseModel): - """Progress of the initial indexing backfill for one stream.""" + """Progress of the latest backfill run for one stream.""" model_config = ConfigDict(extra="allow") status: str - """The backfill status, for example `running` or `complete`.""" + """The backfill status: `starting` until a backfiller picks the run up, then `running` + and `finalizing`, and finally `complete`, `failed`, or `cancelled`.""" backfill_start_time: datetime | None = None - """The start of the backfilled time range.""" + """The run's start time: rows whose cursor field is earlier were skipped. `None` + backfills the whole history.""" expected_total_records: int | None = None """The number of records the backfill expects to collect.""" @@ -431,6 +440,12 @@ class ExternalSearchBackfillStatus(BaseModel): completed_at: datetime | None = None """When the backfill completed.""" + failed_at: datetime | None = None + """When the backfill failed.""" + + cancelled_at: datetime | None = None + """When the backfill was cancelled.""" + class ExternalSearchIndexStatus(BaseModel): """The status of one search index for a stream.""" @@ -519,6 +534,83 @@ def _none_to_empty(cls, value: object) -> object: return [] if value is None else value +class ExternalSearchBackfillRun(BaseModel): + """One stream's backfill run, as reported by a backfill request.""" + + model_config = ConfigDict(extra="allow") + + id: str + """The stream's backfill ID, kept across runs.""" + + stream_name: str + """The stream, or empty for the run that stands for every stream.""" + + source_id: str + """The connection's source.""" + + workspace_id: str + """The connection's workspace.""" + + connection_id: str | None = None + """The connection being backfilled.""" + + destination_id: str | None = None + """The destination the backfill reads from.""" + + requested_at: datetime | None = None + """When the run was requested.""" + + started_at: datetime | None = None + """When a backfiller started the run.""" + + completed_at: datetime | None = None + """When the run completed.""" + + failed_at: datetime | None = None + """When the run failed.""" + + cancel_requested_at: datetime | None = None + """When cancelling the run was requested.""" + + cancelled_at: datetime | None = None + """When the run was cancelled.""" + + expected_total_rows: int | None = None + """The number of rows the run expects to read, once counted.""" + + error: str | None = None + """The failure code of a failed run, for example `enqueue_failed`.""" + + +class ExternalSearchBackfillRequested(BaseModel): + """The outcome of a backfill request for one stream.""" + + model_config = ConfigDict(extra="allow") + + stream_name: str + """The stream. Empty when the request named no streams: that one run stands for every + stream selected in the connection, and is split into per-stream runs when it starts.""" + + outcome: str + """`requested` when a new run was queued, `running` when the stream's current run is + still queued or running and was kept, or `failed` when the run could not be queued.""" + + run_id: str | None = None + """The run's ID.""" + + backfill: ExternalSearchBackfillRun + """The stream's backfill run.""" + + +class ExternalSearchBackfillRequestResult(BaseModel): + """The result of requesting a search backfill for a connection.""" + + model_config = ConfigDict(extra="allow") + + streams: list[ExternalSearchBackfillRequested] + """Each requested stream's outcome.""" + + class ConnectorEnablement(BaseModel): """The Fusion features enabled for a deployed Cloud source or destination.""" diff --git a/airbyte/cloud/connections.py b/airbyte/cloud/connections.py index c296108cd..92469f6a0 100644 --- a/airbyte/cloud/connections.py +++ b/airbyte/cloud/connections.py @@ -7,6 +7,7 @@ from http import HTTPStatus from typing import TYPE_CHECKING, Any, Literal, overload +from airbyte._direct_connectors import api_util as agents_api_util from airbyte._util import api_util from airbyte.cloud._connection_catalog import ( _denormalize_catalog_to_api, @@ -72,6 +73,9 @@ def _validate_quartz_cron_expression(cron_expression: str) -> None: if TYPE_CHECKING: + from datetime import datetime + + from airbyte._direct_connectors.models import ExternalSearchBackfillRequestResult from airbyte.cloud.workspaces import CloudWorkspace @@ -457,6 +461,40 @@ def cancel_sync(self, job_id: int | None = None) -> SyncResult: _latest_job_info=CloudJobInfo.from_api_response(job_response), ) + def request_search_backfill( + self, + *, + streams: list[str] | None = None, + backfill_start_time: datetime | None = None, + ) -> ExternalSearchBackfillRequestResult: + """Request a search backfill that indexes data already in the connection's destination. + + Turning search on for a connection indexes only data from syncs that run afterwards; + existing destination data is indexed only by a backfill. `streams` names the streams + to backfill; omitted, every stream selected in the connection is backfilled. + `backfill_start_time` must be timezone-aware and keeps only rows whose cursor field (a + top-level timestamp or date) is at or after it; streams without such a cursor are + backfilled in full. Omitted, the whole history is backfilled. A stream whose current + run is still queued or running keeps it and is reported as `running`, so retrying is + safe. Requires search indexing to be enabled for the connection. + """ + if backfill_start_time is not None and backfill_start_time.utcoffset() is None: + raise AirbyteLibInputError( + message="`backfill_start_time` must include a timezone.", + guidance="For example, use `2026-01-01T00:00:00+00:00`.", + input_value=backfill_start_time.isoformat(), + ) + request_body: dict[str, Any] = {} + if streams is not None: + request_body["streams"] = list(streams) + if backfill_start_time is not None: + request_body["start_time"] = backfill_start_time.isoformat() + return agents_api_util.request_cloud_connection_search_backfill( + connection_id=self.connection_id, + request_body=request_body, + credentials=self.workspace._credentials, # noqa: SLF001 + ) + def __repr__(self) -> str: """String representation of the connection.""" return ( diff --git a/airbyte/mcp/cloud.py b/airbyte/mcp/cloud.py index 7e0b908a8..e09c35d14 100644 --- a/airbyte/mcp/cloud.py +++ b/airbyte/mcp/cloud.py @@ -11,6 +11,7 @@ __all__: list[str] = [] from collections.abc import Callable +from datetime import datetime from http import HTTPStatus from typing import TYPE_CHECKING, Annotated, Any, Final, Literal, TypeVar, cast @@ -26,6 +27,7 @@ ExternalApiExecuteResult, ExternalApiReadOnlyAction, ExternalApiWriteAction, + ExternalSearchBackfillRequestResult, ExternalSearchResult, ExternalSearchStatusResult, ExternalSearchStreamStatus, @@ -1998,8 +2000,10 @@ def _search_disabled_errors( f"Show the user this message as-is: Search is currently disabled for {label}. {steps}" ), "Once search is enabled, the connector's synced data must be indexed before " - "searches return results. Call `get_cloud_search_status` to check which streams " - "are indexed and whether their backfill is complete.", + "searches return results. New syncs are indexed as they run; data already in the " + "destination is indexed only after `request_cloud_search_backfill`. Call " + "`get_cloud_search_status` to check which streams are indexed and whether their " + "backfill is complete.", ] @@ -2067,7 +2071,10 @@ def execute_external_search_query( # noqa: PLR0913 # Explicit args mirror the max_context_chars: Annotated[ int | None, Field( - description="Truncate each hit's `context` to this many characters.", + description=( + "Truncate each hit's `context` to this many characters. A truncated keyword " + "context stays centered on the match." + ), default=None, ), ] = None, @@ -2116,6 +2123,11 @@ def execute_external_search_query( # noqa: PLR0913 # Explicit args mirror the `semantic` and `hybrid` searches. Pass `destination_id` for a source synced to several enabled destinations; the IDs are listed by `get_cloud_search_status`. Per-index failures are reported in `warnings`. + + A keyword-only hit leaves out of `entity_data` any field longer than 4096 characters + or split across keyword chunks; its `context` holds the matched text, centered on the + first query term. Read a dropped field's full value with `execute_external_api_query` + or `execute_external_sql_query`. """ connector = _get_cloud_connector( _get_cloud_workspace(ctx, workspace_id), connector_id, connector_type @@ -2190,6 +2202,13 @@ def get_cloud_search_status( with its destination and connection IDs, and each stream's backfill progress and search indexes. A stream is searchable once it has at least one index. + Turning search on indexes only data from new syncs; data already in the destination + is indexed only by a backfill requested with `request_cloud_search_backfill`. A + stream's `backfill` is null when none was requested. Otherwise `backfill.status` is + `starting`, `running`, `finalizing`, `complete`, `failed`, or `cancelled`, with + `failed_at` or `cancelled_at` set when the run ended that way. A stream whose table is + empty completes with 0 records. + `stream_name` and `namespace` filter the report client-side; a filter that matches no stream returns a `warnings` entry listing the indexed streams. """ @@ -2238,6 +2257,87 @@ def matches(stream: ExternalSearchStreamStatus) -> bool: return status.model_copy(update={"sources": sources, "warnings": warnings}) +@mcp_tool( + open_world=True, + pipeline_change=False, + external_access=True, +) +def request_cloud_search_backfill( + ctx: Context, + *, + connection_id: Annotated[ + str, + Field(description="The ID of the connection whose destination data to index."), + ], + streams: Annotated[ + list[str] | str | None, + Field( + description=( + "Distinct names of the connection's streams to backfill, as a list or a " + "JSON array string. Omit to backfill every stream selected in the connection." + ), + default=None, + ), + TraceArg.HASH, + ] = None, + backfill_start_time: Annotated[ + str | None, + Field( + description=( + "An ISO 8601 timestamp with a timezone, for example " + "`2026-01-01T00:00:00Z`. Only rows whose cursor field (a top-level " + "timestamp or date) is at or after it are backfilled; streams without such " + "a cursor are backfilled in full. Omit to backfill the whole history." + ), + default=None, + ), + TraceArg.VALUE, + ] = None, + workspace_id: Annotated[ + str | None, + Field( + description=WORKSPACE_ID_TIP_TEXT, + default=None, + ), + TraceArg.OMIT, + ] = None, +) -> ExternalSearchBackfillRequestResult: + """Request a search backfill that indexes data already in a connection's destination. + + Turning search on for a connection does not backfill: it indexes only data written by + syncs that run afterwards. Data already in the destination is searchable only after a + backfill, which reads it from the destination's tables. Only Snowflake (username and + password, or key pair auth) and BigQuery (service account key) destinations support + backfill. The connection must have search indexing enabled. + + Each stream gets one run. A stream whose current run is still queued or running keeps + it and comes back as `running`, so retrying is safe; a finished stream starts a new + run. Without `streams`, the result has one entry with an empty `stream_name` that + stands for every stream. + + Track progress with `get_cloud_search_status` on the connection's source or + destination: each stream's `backfill.status` moves from `starting` through `running` + and `finalizing` to `complete`, `failed`, or `cancelled`. A stream whose table exists + but is empty completes with 0 records. A run fails when, for example, the destination + is unsupported, its table does not exist, or its credentials cannot read it. + """ + try: + start_time = ( + None if backfill_start_time is None else datetime.fromisoformat(backfill_start_time) + ) + except ValueError: + raise AirbyteLibInputError( + message="`backfill_start_time` is not an ISO 8601 timestamp.", + guidance="Use a timestamp with a timezone, for example `2026-01-01T00:00:00Z`.", + input_value=backfill_start_time, + ) from None + connection = _get_cloud_workspace(ctx, workspace_id).get_connection(connection_id) + return connection.request_search_backfill( + streams=resolve_list_of_strings(streams), + backfill_start_time=start_time, + ) + + @mcp_tool( read_only=True, idempotent=True, diff --git a/tests/unit_tests/test_cloud_direct_execute.py b/tests/unit_tests/test_cloud_direct_execute.py index affc9922f..c7a8a4095 100644 --- a/tests/unit_tests/test_cloud_direct_execute.py +++ b/tests/unit_tests/test_cloud_direct_execute.py @@ -4,6 +4,7 @@ from __future__ import annotations import json +from datetime import datetime, timedelta, timezone from types import SimpleNamespace from typing import Any from unittest.mock import patch @@ -17,6 +18,7 @@ from airbyte._direct_connectors.models import ( ConnectorEnablement, ExternalApiExecuteResult, + ExternalSearchBackfillRequestResult, ExternalSearchHit, ExternalSearchResult, ExternalSearchStatusResult, @@ -24,6 +26,7 @@ ExternalSearchType, ) from airbyte.cloud import workspaces as cloud_workspaces +from airbyte.cloud.connections import CloudConnection from airbyte.cloud.connectors import ( CloudConnector, CloudDestination, @@ -42,6 +45,7 @@ from airbyte.cloud.workspaces import CloudWorkspace from airbyte.cloud._credentials import _AirbyteCredentials from airbyte.exceptions import ( + AirbyteAgentsUnavailableError, AirbyteCloudApiError, AirbyteCloudError, AirbyteExternalAccessNotEnabledError, @@ -1643,3 +1647,196 @@ def fail(**_: Any) -> Any: # noqa: ANN401 assert source._context_layer_inspect(warnings=warnings) is None # noqa: SLF001 assert len(warnings) == 1 assert "docs lookup failed" in warnings[0] + + +_BACKFILL_RESPONSE = { + "streams": [ + { + "stream_name": "issues", + "outcome": "requested", + "run_id": "run-1", + "backfill": { + "id": "backfill-1", + "stream_name": "issues", + "connection_id": "connection-1", + "source_id": "source-1", + "destination_id": "destination-1", + "workspace_id": "workspace-id", + "requested_at": "2026-10-16T12:00:00Z", + "started_at": None, + "completed_at": None, + "failed_at": None, + "cancel_requested_at": None, + "cancelled_at": None, + "expected_total_rows": None, + "error": None, + }, + }, + { + "stream_name": "users", + "outcome": "running", + "run_id": "run-0", + "backfill": { + "id": "backfill-0", + "stream_name": "users", + "source_id": "source-1", + "workspace_id": "workspace-id", + "started_at": "2026-10-16T11:00:00Z", + "expected_total_rows": 1200, + }, + }, + ] +} + + +@pytest.mark.parametrize( + ("kwargs", "expected_body"), + [ + pytest.param({}, {}, id="all_streams_whole_history"), + pytest.param( + { + "streams": ["issues", "users"], + "backfill_start_time": datetime(2026, 1, 1, tzinfo=timezone.utc), + }, + {"streams": ["issues", "users"], "start_time": "2026-01-01T00:00:00+00:00"}, + id="streams_and_start_time", + ), + pytest.param( + { + "backfill_start_time": datetime( + 2026, 1, 1, 9, tzinfo=timezone(timedelta(hours=2)) + ) + }, + {"start_time": "2026-01-01T09:00:00+02:00"}, + id="start_time_keeps_offset", + ), + ], +) +def test_request_search_backfill_request_shape( + monkeypatch: pytest.MonkeyPatch, + kwargs: dict[str, Any], + expected_body: dict[str, Any], +) -> None: + """The request posts to the Config API backfill route with Cloud's field names.""" + workspace = _make_workspace(monkeypatch) + _patch_context_layer(monkeypatch) + calls = _patch_search(monkeypatch, _BACKFILL_RESPONSE) + connection = CloudConnection(workspace=workspace, connection_id="connection-1") + + connection.request_search_backfill(**kwargs) + + assert len(calls) == 1 + assert calls[0]["method"] == "POST" + assert calls[0]["url"].endswith("/api/v1/connections/connection-1/backfill") + assert calls[0]["json"] == expected_body + + +def test_request_search_backfill_parses_response( + monkeypatch: pytest.MonkeyPatch, +) -> None: + workspace = _make_workspace(monkeypatch) + _patch_context_layer(monkeypatch) + _patch_search(monkeypatch, _BACKFILL_RESPONSE) + connection = CloudConnection(workspace=workspace, connection_id="connection-1") + + result = connection.request_search_backfill(streams=["issues", "users"]) + + assert isinstance(result, ExternalSearchBackfillRequestResult) + assert [(s.stream_name, s.outcome) for s in result.streams] == [ + ("issues", "requested"), + ("users", "running"), + ] + requested = result.streams[0].backfill + assert requested.connection_id == "connection-1" + assert requested.requested_at == datetime(2026, 10, 16, 12, tzinfo=timezone.utc) + assert requested.error is None + running = result.streams[1].backfill + assert running.connection_id is None + assert running.expected_total_rows == 1200 + + +def test_request_search_backfill_rejects_naive_start_time( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Cloud requires an RFC 3339 offset, so a naive time fails before any request.""" + workspace = _make_workspace(monkeypatch) + _patch_context_layer(monkeypatch) + calls = _patch_search(monkeypatch, _BACKFILL_RESPONSE) + connection = CloudConnection(workspace=workspace, connection_id="connection-1") + + with pytest.raises(AirbyteLibInputError, match="timezone"): + connection.request_search_backfill(backfill_start_time=datetime(2026, 1, 1)) + + assert calls == [] + + +@pytest.mark.parametrize( + "payload", + [ + pytest.param({}, id="missing_streams"), + pytest.param({"streams": [{"stream_name": "issues"}]}, id="missing_backfill"), + ], +) +def test_request_search_backfill_rejects_malformed_response( + monkeypatch: pytest.MonkeyPatch, + payload: dict[str, Any], +) -> None: + workspace = _make_workspace(monkeypatch) + _patch_context_layer(monkeypatch) + _patch_search(monkeypatch, payload) + connection = CloudConnection(workspace=workspace, connection_id="connection-1") + + with pytest.raises(AirbyteCloudError, match="Malformed Airbyte Cloud backfill"): + connection.request_search_backfill() + + +def test_request_search_backfill_requires_context_layer( + monkeypatch: pytest.MonkeyPatch, +) -> None: + workspace = _make_workspace(monkeypatch) + _patch_context_layer(monkeypatch, available=False) + calls = _patch_search(monkeypatch, _BACKFILL_RESPONSE) + connection = CloudConnection(workspace=workspace, connection_id="connection-1") + + with pytest.raises(AirbyteAgentsUnavailableError): + connection.request_search_backfill() + + assert calls == [] + + +def test_search_status_parses_failed_backfill() -> None: + """Cloud reports a failed run's status and `failed_at`.""" + result = ExternalSearchStatusResult.model_validate({ + "sources": [ + { + "source_id": "source-1", + "destination_id": "destination-1", + "connection_id": "connection-1", + "streams": [ + { + "name": "issues", + "namespace": None, + "backfill": { + "backfill_start_time": "2026-01-01T00:00:00Z", + "expected_total_records": 0, + "collected_records": 0, + "indexed_records": 0, + "started_at": "2026-10-16T11:00:00Z", + "completed_at": None, + "failed_at": "2026-10-16T11:05:00Z", + "cancelled_at": None, + "status": "failed", + }, + "indexes": [], + } + ], + } + ] + }) + + backfill = result.sources[0].streams[0].backfill + assert backfill is not None + assert backfill.status == "failed" + assert backfill.failed_at == datetime(2026, 10, 16, 11, 5, tzinfo=timezone.utc) + assert backfill.cancelled_at is None + assert backfill.completed_at is None diff --git a/tests/unit_tests/test_mcp_cloud.py b/tests/unit_tests/test_mcp_cloud.py index 53d58500a..a9239e62c 100644 --- a/tests/unit_tests/test_mcp_cloud.py +++ b/tests/unit_tests/test_mcp_cloud.py @@ -22,6 +22,7 @@ DirectAccessGuidanceSection, ExternalApiExecuteResult, ExternalApiWriteAction, + ExternalSearchBackfillRequestResult, ExternalSearchResult, ExternalSearchStatusResult, ExternalSearchType, @@ -40,6 +41,7 @@ from airbyte.mcp import cloud as cloud_mcp from airbyte.mcp import guidance as guidance_mcp from airbyte.mcp._arg_resolvers import resolve_list_of_dicts +from airbyte.mcp._tool_utils import _TOOL_POLICIES # noqa: PLC2701 from airbyte.mcp.cloud import ( CloudConnectionResult, CloudConnectorDetailsResult, @@ -55,6 +57,7 @@ AirbyteLibInputError, ) from fastmcp import Context, FastMCP +from fastmcp_extensions.decorators import _REGISTERED_TOOLS # noqa: PLC2701 @dataclass @@ -2502,8 +2505,10 @@ def test_search_tools_return_search_disabled_errors( expected_errors = [ expected_message, "Once search is enabled, the connector's synced data must be indexed before " - "searches return results. Call `get_cloud_search_status` to check which streams " - "are indexed and whether their backfill is complete.", + "searches return results. New syncs are indexed as they run; data already in the " + "destination is indexed only after `request_cloud_search_backfill`. Call " + "`get_cloud_search_status` to check which streams are indexed and whether their " + "backfill is complete.", ] assert search.errors == expected_errors assert search.hits == [] @@ -3477,3 +3482,88 @@ async def call_tool() -> None: asyncio.run(call_tool()) assert error.message == selected assert cloud_error_trace_message(error) == selected + + +def _backfill_workspace(monkeypatch: pytest.MonkeyPatch) -> SimpleNamespace: + """Patch `_get_cloud_workspace` to return a workspace serving a mock connection.""" + connection = MagicMock() + connection.request_search_backfill.return_value = ( + ExternalSearchBackfillRequestResult(streams=[]) + ) + workspace = SimpleNamespace(get_connection=MagicMock(return_value=connection)) + monkeypatch.setattr(cloud_mcp, "_get_cloud_workspace", lambda _ctx, _id: workspace) + return workspace + + +@pytest.mark.parametrize( + ("streams", "backfill_start_time", "expected_streams", "expected_start_time"), + [ + pytest.param(None, None, None, None, id="all_streams_whole_history"), + pytest.param( + '["issues", "users"]', + "2026-01-01T00:00:00Z", + ["issues", "users"], + datetime(2026, 1, 1, tzinfo=timezone.utc), + id="json_streams_and_utc_time", + ), + pytest.param( + ["issues"], + "2026-01-01T09:00:00+02:00", + ["issues"], + datetime(2026, 1, 1, 7, tzinfo=timezone.utc), + id="offset_time", + ), + ], +) +def test_request_cloud_search_backfill_forwards_args( + monkeypatch: pytest.MonkeyPatch, + streams: list[str] | str | None, + backfill_start_time: str | None, + expected_streams: list[str] | None, + expected_start_time: datetime | None, +) -> None: + workspace = _backfill_workspace(monkeypatch) + + cloud_mcp.request_cloud_search_backfill( + None, + connection_id="connection-1", + streams=streams, + backfill_start_time=backfill_start_time, + workspace_id=None, + ) + + workspace.get_connection.assert_called_once_with("connection-1") + connection = workspace.get_connection.return_value + connection.request_search_backfill.assert_called_once() + kwargs = connection.request_search_backfill.call_args.kwargs + assert kwargs["streams"] == expected_streams + assert kwargs["backfill_start_time"] == expected_start_time + + +def test_request_cloud_search_backfill_rejects_invalid_start_time( + monkeypatch: pytest.MonkeyPatch, +) -> None: + workspace = _backfill_workspace(monkeypatch) + + with pytest.raises(AirbyteLibInputError, match="ISO 8601"): + cloud_mcp.request_cloud_search_backfill( + None, + connection_id="connection-1", + backfill_start_time="last tuesday", + workspace_id=None, + ) + + workspace.get_connection.assert_not_called() + + +def test_request_cloud_search_backfill_is_a_write_tool() -> None: + """Requesting a backfill queues work, so the tool is neither read-only nor idempotent.""" + (annotations,) = [ + a for f, a in _REGISTERED_TOOLS if f is cloud_mcp.request_cloud_search_backfill + ] + assert annotations["readOnlyHint"] is False + assert annotations["destructiveHint"] is False + assert annotations["idempotentHint"] is False + policy = _TOOL_POLICIES["request_cloud_search_backfill"] + assert policy.external_access is True + assert policy.pipeline_change is False From 6a1196a4ccfe3a8eb7dbf71f628d2768ce6c0dc6 Mon Sep 17 00:00:00 2001 From: Johnny Schmidt Date: Fri, 16 Oct 2026 12:46:42 -0700 Subject: [PATCH 2/2] fix(mcp): name the backfill tool for the search index, mark it closed-world, and add it to the trace fixture --- airbyte/_direct_connectors/api_util.py | 2 +- airbyte/cloud/connections.py | 4 +-- airbyte/mcp/cloud.py | 9 +++---- tests/unit_tests/fixtures/arg_trace_plan.json | 6 +++++ tests/unit_tests/test_cloud_direct_execute.py | 22 ++++++++------- tests/unit_tests/test_mcp_cloud.py | 27 ++++++++++--------- 6 files changed, 40 insertions(+), 30 deletions(-) diff --git a/airbyte/_direct_connectors/api_util.py b/airbyte/_direct_connectors/api_util.py index 34db7c517..f81608f86 100644 --- a/airbyte/_direct_connectors/api_util.py +++ b/airbyte/_direct_connectors/api_util.py @@ -291,7 +291,7 @@ def get_cloud_connector_search_status( ) from ex -def request_cloud_connection_search_backfill( +def request_cloud_connection_search_index_backfill( *, connection_id: str, request_body: dict[str, Any], diff --git a/airbyte/cloud/connections.py b/airbyte/cloud/connections.py index 92469f6a0..15d29f401 100644 --- a/airbyte/cloud/connections.py +++ b/airbyte/cloud/connections.py @@ -461,7 +461,7 @@ def cancel_sync(self, job_id: int | None = None) -> SyncResult: _latest_job_info=CloudJobInfo.from_api_response(job_response), ) - def request_search_backfill( + def request_search_index_backfill( self, *, streams: list[str] | None = None, @@ -489,7 +489,7 @@ def request_search_backfill( request_body["streams"] = list(streams) if backfill_start_time is not None: request_body["start_time"] = backfill_start_time.isoformat() - return agents_api_util.request_cloud_connection_search_backfill( + return agents_api_util.request_cloud_connection_search_index_backfill( connection_id=self.connection_id, request_body=request_body, credentials=self.workspace._credentials, # noqa: SLF001 diff --git a/airbyte/mcp/cloud.py b/airbyte/mcp/cloud.py index e09c35d14..d1362d599 100644 --- a/airbyte/mcp/cloud.py +++ b/airbyte/mcp/cloud.py @@ -2001,7 +2001,7 @@ def _search_disabled_errors( ), "Once search is enabled, the connector's synced data must be indexed before " "searches return results. New syncs are indexed as they run; data already in the " - "destination is indexed only after `request_cloud_search_backfill`. Call " + "destination is indexed only after `request_cloud_search_index_backfill`. Call " "`get_cloud_search_status` to check which streams are indexed and whether their " "backfill is complete.", ] @@ -2203,7 +2203,7 @@ def get_cloud_search_status( search indexes. A stream is searchable once it has at least one index. Turning search on indexes only data from new syncs; data already in the destination - is indexed only by a backfill requested with `request_cloud_search_backfill`. A + is indexed only by a backfill requested with `request_cloud_search_index_backfill`. A stream's `backfill` is null when none was requested. Otherwise `backfill.status` is `starting`, `running`, `finalizing`, `complete`, `failed`, or `cancelled`, with `failed_at` or `cancelled_at` set when the run ended that way. A stream whose table is @@ -2258,11 +2258,10 @@ def matches(stream: ExternalSearchStreamStatus) -> bool: @mcp_tool( - open_world=True, pipeline_change=False, external_access=True, ) -def request_cloud_search_backfill( +def request_cloud_search_index_backfill( ctx: Context, *, connection_id: Annotated[ @@ -2332,7 +2331,7 @@ def request_cloud_search_backfill( input_value=backfill_start_time, ) from None connection = _get_cloud_workspace(ctx, workspace_id).get_connection(connection_id) - return connection.request_search_backfill( + return connection.request_search_index_backfill( streams=resolve_list_of_strings(streams), backfill_start_time=start_time, ) diff --git a/tests/unit_tests/fixtures/arg_trace_plan.json b/tests/unit_tests/fixtures/arg_trace_plan.json index 23b989e52..758a80928 100644 --- a/tests/unit_tests/fixtures/arg_trace_plan.json +++ b/tests/unit_tests/fixtures/arg_trace_plan.json @@ -282,6 +282,12 @@ "name": "fingerprint", "workspace_id": "omit" }, + "request_cloud_search_index_backfill": { + "backfill_start_time": "value str", + "connection_id": "hash", + "streams": "hash list", + "workspace_id": "omit" + }, "run_cloud_sync": { "connection_id": "hash", "wait": "value allowed=[false,true]", diff --git a/tests/unit_tests/test_cloud_direct_execute.py b/tests/unit_tests/test_cloud_direct_execute.py index c7a8a4095..9dc547f37 100644 --- a/tests/unit_tests/test_cloud_direct_execute.py +++ b/tests/unit_tests/test_cloud_direct_execute.py @@ -1712,7 +1712,7 @@ def fail(**_: Any) -> Any: # noqa: ANN401 ), ], ) -def test_request_search_backfill_request_shape( +def test_request_search_index_backfill_request_shape( monkeypatch: pytest.MonkeyPatch, kwargs: dict[str, Any], expected_body: dict[str, Any], @@ -1723,7 +1723,7 @@ def test_request_search_backfill_request_shape( calls = _patch_search(monkeypatch, _BACKFILL_RESPONSE) connection = CloudConnection(workspace=workspace, connection_id="connection-1") - connection.request_search_backfill(**kwargs) + connection.request_search_index_backfill(**kwargs) assert len(calls) == 1 assert calls[0]["method"] == "POST" @@ -1731,7 +1731,7 @@ def test_request_search_backfill_request_shape( assert calls[0]["json"] == expected_body -def test_request_search_backfill_parses_response( +def test_request_search_index_backfill_parses_response( monkeypatch: pytest.MonkeyPatch, ) -> None: workspace = _make_workspace(monkeypatch) @@ -1739,7 +1739,7 @@ def test_request_search_backfill_parses_response( _patch_search(monkeypatch, _BACKFILL_RESPONSE) connection = CloudConnection(workspace=workspace, connection_id="connection-1") - result = connection.request_search_backfill(streams=["issues", "users"]) + result = connection.request_search_index_backfill(streams=["issues", "users"]) assert isinstance(result, ExternalSearchBackfillRequestResult) assert [(s.stream_name, s.outcome) for s in result.streams] == [ @@ -1755,7 +1755,7 @@ def test_request_search_backfill_parses_response( assert running.expected_total_rows == 1200 -def test_request_search_backfill_rejects_naive_start_time( +def test_request_search_index_backfill_rejects_naive_start_time( monkeypatch: pytest.MonkeyPatch, ) -> None: """Cloud requires an RFC 3339 offset, so a naive time fails before any request.""" @@ -1765,7 +1765,9 @@ def test_request_search_backfill_rejects_naive_start_time( connection = CloudConnection(workspace=workspace, connection_id="connection-1") with pytest.raises(AirbyteLibInputError, match="timezone"): - connection.request_search_backfill(backfill_start_time=datetime(2026, 1, 1)) + connection.request_search_index_backfill( + backfill_start_time=datetime(2026, 1, 1) + ) assert calls == [] @@ -1777,7 +1779,7 @@ def test_request_search_backfill_rejects_naive_start_time( pytest.param({"streams": [{"stream_name": "issues"}]}, id="missing_backfill"), ], ) -def test_request_search_backfill_rejects_malformed_response( +def test_request_search_index_backfill_rejects_malformed_response( monkeypatch: pytest.MonkeyPatch, payload: dict[str, Any], ) -> None: @@ -1787,10 +1789,10 @@ def test_request_search_backfill_rejects_malformed_response( connection = CloudConnection(workspace=workspace, connection_id="connection-1") with pytest.raises(AirbyteCloudError, match="Malformed Airbyte Cloud backfill"): - connection.request_search_backfill() + connection.request_search_index_backfill() -def test_request_search_backfill_requires_context_layer( +def test_request_search_index_backfill_requires_context_layer( monkeypatch: pytest.MonkeyPatch, ) -> None: workspace = _make_workspace(monkeypatch) @@ -1799,7 +1801,7 @@ def test_request_search_backfill_requires_context_layer( connection = CloudConnection(workspace=workspace, connection_id="connection-1") with pytest.raises(AirbyteAgentsUnavailableError): - connection.request_search_backfill() + connection.request_search_index_backfill() assert calls == [] diff --git a/tests/unit_tests/test_mcp_cloud.py b/tests/unit_tests/test_mcp_cloud.py index a9239e62c..b84ec280c 100644 --- a/tests/unit_tests/test_mcp_cloud.py +++ b/tests/unit_tests/test_mcp_cloud.py @@ -2506,7 +2506,7 @@ def test_search_tools_return_search_disabled_errors( expected_message, "Once search is enabled, the connector's synced data must be indexed before " "searches return results. New syncs are indexed as they run; data already in the " - "destination is indexed only after `request_cloud_search_backfill`. Call " + "destination is indexed only after `request_cloud_search_index_backfill`. Call " "`get_cloud_search_status` to check which streams are indexed and whether their " "backfill is complete.", ] @@ -3487,7 +3487,7 @@ async def call_tool() -> None: def _backfill_workspace(monkeypatch: pytest.MonkeyPatch) -> SimpleNamespace: """Patch `_get_cloud_workspace` to return a workspace serving a mock connection.""" connection = MagicMock() - connection.request_search_backfill.return_value = ( + connection.request_search_index_backfill.return_value = ( ExternalSearchBackfillRequestResult(streams=[]) ) workspace = SimpleNamespace(get_connection=MagicMock(return_value=connection)) @@ -3515,7 +3515,7 @@ def _backfill_workspace(monkeypatch: pytest.MonkeyPatch) -> SimpleNamespace: ), ], ) -def test_request_cloud_search_backfill_forwards_args( +def test_request_cloud_search_index_backfill_forwards_args( monkeypatch: pytest.MonkeyPatch, streams: list[str] | str | None, backfill_start_time: str | None, @@ -3524,7 +3524,7 @@ def test_request_cloud_search_backfill_forwards_args( ) -> None: workspace = _backfill_workspace(monkeypatch) - cloud_mcp.request_cloud_search_backfill( + cloud_mcp.request_cloud_search_index_backfill( None, connection_id="connection-1", streams=streams, @@ -3534,19 +3534,19 @@ def test_request_cloud_search_backfill_forwards_args( workspace.get_connection.assert_called_once_with("connection-1") connection = workspace.get_connection.return_value - connection.request_search_backfill.assert_called_once() - kwargs = connection.request_search_backfill.call_args.kwargs + connection.request_search_index_backfill.assert_called_once() + kwargs = connection.request_search_index_backfill.call_args.kwargs assert kwargs["streams"] == expected_streams assert kwargs["backfill_start_time"] == expected_start_time -def test_request_cloud_search_backfill_rejects_invalid_start_time( +def test_request_cloud_search_index_backfill_rejects_invalid_start_time( monkeypatch: pytest.MonkeyPatch, ) -> None: workspace = _backfill_workspace(monkeypatch) with pytest.raises(AirbyteLibInputError, match="ISO 8601"): - cloud_mcp.request_cloud_search_backfill( + cloud_mcp.request_cloud_search_index_backfill( None, connection_id="connection-1", backfill_start_time="last tuesday", @@ -3556,14 +3556,17 @@ def test_request_cloud_search_backfill_rejects_invalid_start_time( workspace.get_connection.assert_not_called() -def test_request_cloud_search_backfill_is_a_write_tool() -> None: - """Requesting a backfill queues work, so the tool is neither read-only nor idempotent.""" +def test_request_cloud_search_index_backfill_is_a_write_tool() -> None: + """Requesting a backfill queues index work, so the tool is a closed-world write.""" (annotations,) = [ - a for f, a in _REGISTERED_TOOLS if f is cloud_mcp.request_cloud_search_backfill + a + for f, a in _REGISTERED_TOOLS + if f is cloud_mcp.request_cloud_search_index_backfill ] assert annotations["readOnlyHint"] is False assert annotations["destructiveHint"] is False assert annotations["idempotentHint"] is False - policy = _TOOL_POLICIES["request_cloud_search_backfill"] + assert annotations["openWorldHint"] is False + policy = _TOOL_POLICIES["request_cloud_search_index_backfill"] assert policy.external_access is True assert policy.pipeline_change is False