feat(mcp): add a Fusion search backfill tool and correct the search tool descriptions

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_index_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,

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."""
 

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_index_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_index_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 (

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_index_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_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
+    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,86 @@ def matches(stream: ExternalSearchStreamStatus) -> bool:
     return status.model_copy(update={"sources": sources, "warnings": warnings})
 
 
+@mcp_tool(
+    pipeline_change=False,
+    external_access=True,
+)
+def request_cloud_search_index_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_index_backfill(
+        streams=resolve_list_of_strings(streams),
+        backfill_start_time=start_time,
+    )
+
+
 @mcp_tool(
     read_only=True,
     idempotent=True,