feat: add options for returning record data and raw responses/requests in execute_stream_test_read

connector_builder_mcp/_connector_builder.py

@@ -31,7 +31,7 @@
 from pydantic import BaseModel, Field
 
 from connector_builder_mcp._secrets import hydrate_config, register_secrets_tools
-from connector_builder_mcp._util import validate_manifest_structure
+from connector_builder_mcp._util import filter_config_secrets, validate_manifest_structure
 
 logger = logging.getLogger(__name__)
 
@@ -52,6 +52,75 @@ class StreamTestResult(BaseModel):
     message: str
     records_read: int = 0
     errors: list[str] = []
+    records: list[dict[str, Any]] | None = Field(
+        default=None, description="Actual record data when include_records=True"
+    )
+    slices: list[dict[str, Any]] | None = Field(
+        default=None,
+        description="By default, raw request and response data will be returned only upon error. "
+        "Set to True or False, to always or never return the raw response data.",
+    )
+
+
+def _without_records(slices: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """Remove records from slices structure while preserving other data.
+
+    Args:
+        slices: List of slice objects from CDK output
+
+    Returns:
+        Slices with records removed from pages
+    """
+    result = []
+    for slice_obj in slices:
+        if not isinstance(slice_obj, dict):
+            result.append(slice_obj)
+            continue
+
+        slice_copy = slice_obj.copy()
+        if "pages" in slice_copy and isinstance(slice_copy["pages"], list):
+            pages_copy = []
+            for page in slice_copy["pages"]:
+                if isinstance(page, dict):
+                    page_copy = page.copy()
+                    page_copy.pop("records", None)
+                    pages_copy.append(page_copy)
+                else:
+                    pages_copy.append(page)
+            slice_copy["pages"] = pages_copy
+        result.append(slice_copy)
+    return result
+
+
+def _without_raw_requests_and_responses(slices: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """Remove raw request and response data from slices structure.
+
+    Args:
+        slices: List of slice objects from CDK output
+
+    Returns:
+        Slices with request and response data removed from pages
+    """
+    result = []
+    for slice_obj in slices:
+        if not isinstance(slice_obj, dict):
+            result.append(slice_obj)
+            continue
+
+        slice_copy = slice_obj.copy()
+        if "pages" in slice_copy and isinstance(slice_copy["pages"], list):
+            pages_copy = []
+            for page in slice_copy["pages"]:
+                if isinstance(page, dict):
+                    page_copy = page.copy()
+                    page_copy.pop("request", None)
+                    page_copy.pop("response", None)
+                    pages_copy.append(page_copy)
+                else:
+                    pages_copy.append(page)
+            slice_copy["pages"] = pages_copy
+        result.append(slice_copy)
+    return result
 
 
 def _get_dummy_catalog(stream_name: str) -> ConfiguredAirbyteCatalog:
@@ -234,22 +303,33 @@ def execute_stream_test_read(
         int,
         Field(description="Maximum number of records to read", ge=1, le=1000),
     ] = 10,
+    include_records: Annotated[
+        bool,
+        Field(description="If True, include actual record data in the response"),
+    ] = False,
+    include_raw_response_data: Annotated[
+        bool | None,
+        Field(
+            description="By default, raw request and response data will be returned only upon error. "
+            "Set to True or False, to always or never return the raw response data."
+        ),
     dotenv_path: Annotated[
         Path | None,
         Field(description="Optional path to .env file for secret hydration"),
     ] = None,
 ) -> StreamTestResult:
     """Execute reading from a connector stream.
 
-    Args:
-        manifest: The connector manifest
-        config: Connector configuration
-        stream_name: Name of the stream to test
-        max_records: Maximum number of records to read
-        dotenv_path: Optional path to .env file for secret hydration
-
     Returns:
-        Test result with success status and details
+        Test result with success status and details, optionally including record and HTTP data
+
+    Note:
+        Raw request and response data in slices is automatically sanitized using filter_config_secrets()
+        to prevent accidental exposure of API keys and other sensitive information.
+
+        To keep a manageable context window, it is recommended to always include a 'max_records'
+        argument when running with include_records=True or include_raw_response_data=True. This
+        prevents overflow which can happen if the stream has millions of records.
     """
     logger.info(f"Testing stream read for stream: {stream_name}")
 
@@ -279,22 +359,105 @@ def execute_stream_test_read(
         )
 
         if result.type.value == "RECORD":
+            records_data = None
+            slices_data: list[dict[str, Any]] | None = None
+
+            if result.record and result.record.data:
+                try:
+                    stream_data = result.record.data
+                    slices = stream_data.get("slices", []) if isinstance(stream_data, dict) else []
+
+                    if include_records:
+                        records_data = []
+                        for slice_obj in slices:
+                            if isinstance(slice_obj, dict) and "pages" in slice_obj:
+                                for page in slice_obj["pages"]:
+                                    if isinstance(page, dict) and "records" in page:
+                                        records_data.extend(page["records"])
+
+                    if slices:
+                        slices_data = slices.copy()
+
+                        if not include_records:
+                            slices_data = _without_records(slices_data)
+
+                        if include_raw_response_data is False:
+                            slices_data = _without_raw_requests_and_responses(slices_data)
+
+                        slices_data = filter_config_secrets(slices_data)  # type: ignore[assignment]
+
+                except Exception as e:
+                    logger.warning(f"Failed to extract data: {e}")
+
             return StreamTestResult(
                 success=True,
                 message=f"Successfully read from stream {stream_name}",
                 records_read=max_records,
+                records=records_data,
+                slices=slices_data,
             )
 
         error_msg = "Failed to read from stream"
         if hasattr(result, "trace") and result.trace:
             error_msg = result.trace.error.message
 
-        return StreamTestResult(success=False, message=error_msg, errors=[error_msg])
+        slices_data = None
+
+        if include_raw_response_data is not False:
+            if result.record and result.record.data:
+                try:
+                    stream_data = result.record.data
+                    slices = stream_data.get("slices", []) if isinstance(stream_data, dict) else []
+                    if slices:
+                        slices_data = slices.copy()
+
+                        if not include_records:
+                            slices_data = _without_records(slices_data)
+
+                        if include_raw_response_data is False:
+                            slices_data = _without_raw_requests_and_responses(slices_data)
+
+                        slices_data = filter_config_secrets(slices_data)  # type: ignore[assignment]
+                except Exception as e:
+                    logger.warning(f"Failed to extract debug data: {e}")
+
+        return StreamTestResult(
+            success=False,
+            message=error_msg,
+            errors=[error_msg],
+            slices=slices_data,
+        )
 
     except Exception as e:
         logger.error(f"Error testing stream read: {e}")
         error_msg = f"Stream test error: {str(e)}"
-        return StreamTestResult(success=False, message=error_msg, errors=[error_msg])
+
+        slices_data = None
+
+        if include_raw_response_data is not False:
+            try:
+                if "result" in locals() and result.record and result.record.data:
+                    stream_data = result.record.data
+                    slices = stream_data.get("slices", []) if isinstance(stream_data, dict) else []
+                    if slices:
+                        slices_data = slices.copy()
+
+                        if not include_records:
+                            slices_data = _without_records(slices_data)
+
+                        if include_raw_response_data is False:
+                            slices_data = _without_raw_requests_and_responses(slices_data)
+
+                        slices_data = filter_config_secrets(slices_data)  # type: ignore[assignment]
+            except Exception:
+                pass
+
+        return StreamTestResult(
+            success=False,
+            message=error_msg,
+            errors=[error_msg],
+            slices=slices_data,
+        )
 
 
 def get_resolved_manifest(

connector_builder_mcp/_util.py

@@ -14,37 +14,44 @@ def initialize_logging() -> None:
     )
 
 
-def filter_config_secrets(config: dict[str, Any]) -> dict[str, Any]:
+def filter_config_secrets(
+    config: dict[str, Any] | list[Any] | Any,
+) -> dict[str, Any] | list[Any] | Any:
     """Filter sensitive information from configuration for logging.
 
     Note: For config hydration with secrets, see _secrets.hydrate_config()
 
     Args:
-        config: Configuration dictionary that may contain secrets
+        config: Configuration dictionary, list, or other value that may contain secrets
 
     Returns:
-        Configuration dictionary with sensitive values masked
+        Configuration with sensitive values masked
     """
-    filtered = config.copy()
-    sensitive_keys = {
-        "password",
-        "token",
-        "key",
-        "secret",
-        "credential",
-        "api_key",
-        "access_token",
-        "refresh_token",
-        "client_secret",
-    }
-
-    for key, value in filtered.items():
-        if isinstance(value, dict):
-            filtered[key] = filter_config_secrets(value)
-        elif any(sensitive in key.lower() for sensitive in sensitive_keys):
-            filtered[key] = "***REDACTED***"
-
-    return filtered
+    if isinstance(config, dict):
+        filtered = config.copy()
+        sensitive_keys = {
+            "password",
+            "token",
+            "key",
+            "secret",
+            "credential",
+            "api_key",
+            "access_token",
+            "refresh_token",
+            "client_secret",
+        }
+
+        for key, value in filtered.items():
+            if isinstance(value, dict | list):
+                filtered[key] = filter_config_secrets(value)
+            elif any(sensitive in key.lower() for sensitive in sensitive_keys):
+                filtered[key] = "***REDACTED***"
+
+        return filtered
+    elif isinstance(config, list):
+        return [filter_config_secrets(item) for item in config]
+    else:
+        return config
 
 
 def validate_manifest_structure(manifest: dict[str, Any]) -> bool: