fix(mcp): require source_id on destination search streams

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

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,

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

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(