diff --git a/airbyte/_direct_connectors/models.py b/airbyte/_direct_connectors/models.py index 4270a3806..80201493d 100644 --- a/airbyte/_direct_connectors/models.py +++ b/airbyte/_direct_connectors/models.py @@ -291,7 +291,11 @@ class ExternalSearchStreamFilter(BaseModel): """The stream to search.""" source_id: str | None = None - """The source the stream belongs to. May be omitted on a source-level search.""" + """The source the stream belongs to. + + Required on a destination search. May be omitted on a source search, where it must + match the searched source if given. + """ namespace: str | None = None """The stream namespace, when the stream has one.""" diff --git a/airbyte/cloud/connectors.py b/airbyte/cloud/connectors.py index b66fdcfa1..7bdb15e73 100644 --- a/airbyte/cloud/connectors.py +++ b/airbyte/cloud/connectors.py @@ -526,7 +526,8 @@ def execute_search_query( # noqa: PLR0913 # Explicit args are the point of thi """Search this connector's indexed data. A source searches its own indexed streams; a destination searches every indexed - source synced to it. `min_similarity` and `max_similarity_diff` apply only to + source synced to it, so each `streams` entry on a destination search must name its + `source_id`. `min_similarity` and `max_similarity_diff` apply only to `semantic` and `hybrid` searches. `destination_id` selects the destination for a source synced to more than one enabled destination. Arguments left as `None` are omitted so the backend defaults apply. Requires search indexing to be enabled for @@ -581,6 +582,17 @@ def execute_search_query( # noqa: PLR0913 # Explicit args are the point of thi ), context={"errors": ex.errors(include_url=False)}, ) from None + if self.connector_type == ConnectorType.DESTINATION and any( + "source_id" not in stream for stream in resolved_streams or [] + ): + raise exc.AirbyteLibInputError( + message="Every `streams` entry needs a `source_id` on a destination search.", + guidance=( + "Add the `source_id` of the source each stream belongs to; the " + "destination's search status lists them." + ), + context={"connector_id": self.connector_id}, + ) optional_fields: dict[str, Any] = { "limit": limit, diff --git a/airbyte/mcp/cloud.py b/airbyte/mcp/cloud.py index 5684d767b..2ed5f25e8 100644 --- a/airbyte/mcp/cloud.py +++ b/airbyte/mcp/cloud.py @@ -1967,9 +1967,11 @@ def execute_external_search_query( # noqa: PLR0913 # Explicit args mirror the Field( description=( "Streams to search, as a list of objects or a JSON array string. Each " - "object takes `stream_name` plus optional `source_id`, `namespace`, and " + "object takes `stream_name`, `source_id`, and optional `namespace` and " "`fields` (the `entity_data` fields to return; an output projection, not " - "a match filter). Omit to search every indexed stream." + "a match filter). `source_id` is required on a destination search; on a " + "source search it may be omitted, and must match the searched source if " + "given. Omit `streams` to search every indexed stream." ), default=None, ), @@ -2025,8 +2027,10 @@ def execute_external_search_query( # noqa: PLR0913 # Explicit args mirror the """Search indexed data for a deployed Cloud source or destination. A source searches its own indexed data; a destination searches every indexed source - synced to it. When results are empty, call `get_cloud_search_status` to check which - streams are indexed and whether their backfill is complete. + synced to it. On a destination search, every `streams` entry needs the `source_id` + of the source that stream belongs to. When results are empty, call + `get_cloud_search_status` to check which streams are indexed and whether their + backfill is complete. Keep responses small: use `streams[].fields` to project each hit's `entity_data` (an output projection, not a match filter) and `max_context_chars` to truncate diff --git a/tests/unit_tests/test_cloud_direct_execute.py b/tests/unit_tests/test_cloud_direct_execute.py index b44f5787d..da5eed269 100644 --- a/tests/unit_tests/test_cloud_direct_execute.py +++ b/tests/unit_tests/test_cloud_direct_execute.py @@ -1155,6 +1155,12 @@ def test_search_rejects_malformed_response( pytest.param( "source", {"streams": [{"fields": ["id"]}]}, "`streams`", id="bad_stream" ), + pytest.param( + "destination", + {"streams": [{"stream_name": "issues"}]}, + "`source_id`", + id="destination_stream_without_source_id", + ), ], ) def test_search_input_validation(