feat: Add scanning capability to get_record() and new fetch_source_stream_record MCP tool

- Add allow_scanning and scan_timeout_seconds parameters to get_record()
- When allow_scanning=True, iterate through get_records() to find matching records
- Allow searching by non-primary-key fields when scanning is enabled
- Implement time-based timeout with configurable scan_timeout_seconds
- Update _normalize_and_validate_pk() to support strict_pk_field parameter
- Add fetch_source_stream_record MCP tool for single record fetching
- Add comprehensive unit tests for scanning feature (8 new tests)
- All tests passing (23/23)

Co-Authored-By: AJ Steers <aj@airbyte.io>

Author: devin-ai-integration[bot]

airbyte/mcp/local_ops.py

@@ -461,6 +461,124 @@ def read_source_stream_records(
         return records
 
 
+@mcp_tool(
+    domain="local",
+    read_only=True,
+    extra_help_text=_CONFIG_HELP,
+)
+def fetch_source_stream_record(  # noqa: PLR0913
+    source_connector_name: Annotated[
+        str,
+        Field(description="The name of the source connector."),
+    ],
+    config: Annotated[
+        dict | str | None,
+        Field(
+            description="The configuration for the source connector as a dict or JSON string.",
+            default=None,
+        ),
+    ],
+    config_file: Annotated[
+        str | Path | None,
+        Field(
+            description="Path to a YAML or JSON file containing the source connector config.",
+            default=None,
+        ),
+    ],
+    config_secret_name: Annotated[
+        str | None,
+        Field(
+            description="The name of the secret containing the configuration.",
+            default=None,
+        ),
+    ],
+    *,
+    stream_name: Annotated[
+        str,
+        Field(description="The name of the stream to fetch the record from."),
+    ],
+    pk_value: Annotated[
+        str | dict[str, Any],
+        Field(
+            description="Either a primary key value as a string (e.g., '123') or a dict "
+            "with a single entry where the key is the field name and the value is the field value "
+            "(e.g., {'id': '123'} or {'email': 'user@example.com'}). "
+            "When allow_scanning=False, dict keys must match the primary key. "
+            "When allow_scanning=True, dict keys can be any field name.",
+        ),
+    ],
+    allow_scanning: Annotated[
+        bool,
+        Field(
+            description="When True, enables scanning through records to find a match. "
+            "This allows searching by non-primary-key fields but is slower and may not find "
+            "the record if it's not in the first scan_timeout_seconds of data. Default: False.",
+            default=False,
+        ),
+    ],
+    scan_timeout_seconds: Annotated[
+        int,
+        Field(
+            description="Maximum time in seconds to scan for a record when allow_scanning=True. "
+            "Default: 5 seconds.",
+            default=5,
+        ),
+    ],
+    override_execution_mode: Annotated[
+        Literal["docker", "python", "yaml", "auto"],
+        Field(
+            description="Optionally override the execution method to use for the connector. "
+            "This parameter is ignored if manifest_path is provided (yaml mode will be used).",
+            default="auto",
+        ),
+    ],
+    manifest_path: Annotated[
+        str | Path | None,
+        Field(
+            description="Path to a local YAML manifest file for declarative connectors.",
+            default=None,
+        ),
+    ],
+) -> dict[str, Any] | str:
+    """Fetch a single record from a source connector by primary key or field value.
+
+    This tool is only supported for declarative (YAML-based) sources.
+    """
+    try:
+        source: Source = _get_mcp_source(
+            connector_name=source_connector_name,
+            override_execution_mode=override_execution_mode,
+            manifest_path=manifest_path,
+        )
+        config_dict = resolve_config(
+            config=config,
+            config_file=config_file,
+            config_secret_name=config_secret_name,
+            config_spec_jsonschema=source.config_spec,
+        )
+        source.set_config(config_dict)
+
+        record = source.get_record(
+            stream_name=stream_name,
+            pk_value=pk_value,
+            allow_scanning=allow_scanning,
+            scan_timeout_seconds=scan_timeout_seconds,
+        )
+
+        print(
+            f"Retrieved record with {pk_value} from stream '{stream_name}'",
+            file=sys.stderr,
+        )
+
+    except Exception as ex:
+        tb_str = traceback.format_exc()
+        return (
+            f"Error fetching record from source '{source_connector_name}': {ex!r}, {ex!s}\n{tb_str}"
+        )
+    else:
+        return record
+
+
 @mcp_tool(
     domain="local",
     read_only=True,

airbyte/sources/base.py

@@ -614,38 +614,42 @@ def _normalize_and_validate_pk(
         self,
         stream_name: str,
         pk_value: LookupValue | dict[str, LookupValue],
-    ) -> str:
+        *,
+        strict_pk_field: bool = True,
+    ) -> tuple[str, str]:
         """Normalize and validate primary key input.
 
         Accepts either a direct primary key value or a dict with a single entry
-        where the key matches the stream's primary key field name.
+        where the key matches the stream's primary key field name (when strict_pk_field=True).
 
         Args:
             stream_name: Name of the stream
             pk_value: Either a direct PK value or a dict with single entry
+            strict_pk_field: When True, dict keys must match the stream's primary key.
+                When False (for scanning), dict keys can be any field name.
 
         Returns:
-            The primary key value as a string
+            A tuple of (field_name, field_value) as strings
 
         Raises:
             PyAirbyteInputError: If validation fails
             NotImplementedError: If the stream has composite or no primary key
+                (when strict_pk_field=True)
         """
         primary_key = self._get_stream_primary_key(stream_name)
 
-        if len(primary_key) == 0:
-            raise NotImplementedError(
-                f"Stream '{stream_name}' does not have a primary key defined. "
-                "Cannot fetch individual records without a primary key."
-            )
-
-        if len(primary_key) > 1:
-            raise NotImplementedError(
-                f"Stream '{stream_name}' has a composite primary key {primary_key}. "
-                "Fetching by composite primary key is not yet supported."
-            )
+        if strict_pk_field:
+            if len(primary_key) == 0:
+                raise NotImplementedError(
+                    f"Stream '{stream_name}' does not have a primary key defined. "
+                    "Cannot fetch individual records without a primary key."
+                )
 
-        pk_field = primary_key[0]
+            if len(primary_key) > 1:
+                raise NotImplementedError(
+                    f"Stream '{stream_name}' has a composite primary key {primary_key}. "
+                    "Fetching by composite primary key is not yet supported."
+                )
 
         if isinstance(pk_value, dict):
             if len(pk_value) != 1:
@@ -663,55 +667,88 @@ def _normalize_and_validate_pk(
             dict_key = next(iter(pk_value.keys()))
             dict_value = pk_value[dict_key]
 
-            if dict_key != pk_field:
-                raise exc.PyAirbyteInputError(
-                    message="The key in the pk_value dict does not match the stream's primary key.",
-                    input_value=dict_key,
-                    context={
-                        "stream_name": stream_name,
-                        "expected_key": pk_field,
-                        "actual_key": dict_key,
-                    },
-                )
+            if strict_pk_field:
+                pk_field = primary_key[0]
+                if dict_key != pk_field:
+                    raise exc.PyAirbyteInputError(
+                        message=(
+                            "The key in the pk_value dict does not match "
+                            "the stream's primary key."
+                        ),
+                        input_value=dict_key,
+                        context={
+                            "stream_name": stream_name,
+                            "expected_key": pk_field,
+                            "actual_key": dict_key,
+                        },
+                    )
 
-            return str(dict_value)
+            return dict_key, str(dict_value)
 
-        return str(pk_value)
+        if len(primary_key) == 0:
+            raise exc.PyAirbyteInputError(
+                message=(
+                    "When passing a scalar pk_value, the stream must have a primary key defined. "
+                    "Use a dict to specify the field name explicitly."
+                ),
+                input_value=str(pk_value),
+                context={
+                    "stream_name": stream_name,
+                },
+            )
+
+        return primary_key[0], str(pk_value)
 
     def get_record(
         self,
         stream_name: str,
         *,
         pk_value: LookupValue | dict[str, LookupValue],
+        allow_scanning: bool = False,
+        scan_timeout_seconds: int = 5,
     ) -> dict[str, Any]:
-        """Fetch a single record from a stream by primary key.
+        """Fetch a single record from a stream by primary key or field value.
 
         This method is only supported for declarative (YAML-based) sources.
 
         Args:
             stream_name: Name of the stream to fetch from
             pk_value: Either a direct primary key value (e.g., "123") or a dict
-                with a single entry where the key is the primary key field name
-                and the value is the primary key value (e.g., {"id": "123"}).
-                The dict form provides explicit validation that you're using
-                the correct primary key field.
+                with a single entry where the key is the field name and the value
+                is the field value (e.g., {"id": "123"} or {"email": "user@example.com"}).
+                When allow_scanning=False, dict keys must match the primary key.
+                When allow_scanning=True, dict keys can be any field name.
+            allow_scanning: When True, enables scanning through records to find a match.
+                This allows searching by non-primary-key fields but is slower and may
+                not find the record if it's not in the first scan_timeout_seconds of data.
+                Default: False (use fast primary key lookup only).
+            scan_timeout_seconds: Maximum time in seconds to scan for a record when
+                allow_scanning=True. Default: 5 seconds.
 
         Returns:
             The fetched record as a dict
 
         Raises:
             NotImplementedError: If the source is not a declarative source, or if
                 the stream has a composite primary key or no primary key
+                (when allow_scanning=False)
             PyAirbyteInputError: If the stream doesn't exist or pk_value validation fails
-            RecordNotFoundException: If the record is not found (from CDK)
+            ValueError: If the record is not found within the timeout period
+                (when allow_scanning=True)
 
         Example:
             ```python
             source = get_source("source-rest-api-tutorial", config=config)
 
             record = source.get_record("users", pk_value="123")
-
             record = source.get_record("users", pk_value={"id": "123"})
+
+            record = source.get_record(
+                "users",
+                pk_value={"email": "user@example.com"},
+                allow_scanning=True,
+                scan_timeout_seconds=10,
+            )
             ```
         """
         if not isinstance(self.executor, DeclarativeExecutor):
@@ -720,12 +757,49 @@ def get_record(
                 f"This source uses {type(self.executor).__name__}."
             )
 
-        validated_pk_str = self._normalize_and_validate_pk(stream_name, pk_value)
+        field_name, field_value = self._normalize_and_validate_pk(
+            stream_name, pk_value, strict_pk_field=not allow_scanning
+        )
+
+        # Try fast path first if the field is the primary key
+        primary_key = self._get_stream_primary_key(stream_name)
+        if len(primary_key) == 1 and field_name == primary_key[0]:
+            try:
+                return self.executor.fetch_record(
+                    stream_name=stream_name,
+                    pk_value=field_value,
+                    config=self._config_dict,
+                )
+            except Exception:
+                if not allow_scanning:
+                    raise
+
+        if not allow_scanning:
+            raise ValueError(
+                f"Record with {field_name}={field_value} not found in stream '{stream_name}'. "
+                "Consider using allow_scanning=True to search by non-primary-key fields."
+            )
+
+        start_time = time.time()
+        records_checked = 0
+
+        for record in self.get_records(stream_name):
+            elapsed = time.time() - start_time
+            if elapsed > scan_timeout_seconds:
+                raise ValueError(
+                    f"Record with {field_name}={field_value} not found in stream '{stream_name}' "
+                    f"within {scan_timeout_seconds} seconds (checked {records_checked} records). "
+                    "Consider increasing scan_timeout_seconds or using a primary key lookup."
+                )
+
+            records_checked += 1
+
+            if field_name in record and str(record[field_name]) == field_value:
+                return record
 
-        return self.executor.fetch_record(
-            stream_name=stream_name,
-            pk_value=validated_pk_str,
-            config=self._config_dict,
+        raise ValueError(
+            f"Record with {field_name}={field_value} not found in stream '{stream_name}' "
+            f"(checked {records_checked} records in {time.time() - start_time:.1f} seconds)."
         )
 
     def get_documents(