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

connector_builder_mcp/_connector_builder.py

@@ -16,7 +16,7 @@
     read_stream,
     resolve_manifest,
 )
-from airbyte_cdk.models.airbyte_protocol import (
+from airbyte_cdk.models import (
     AirbyteStream,
     ConfiguredAirbyteCatalog,
     DestinationSyncMode,
@@ -25,7 +25,7 @@
 from fastmcp import FastMCP
 from pydantic import BaseModel, Field
 
-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__)
 
@@ -46,6 +46,12 @@ 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="Raw slices data when include_raw_response_data=True"
+    )
 
 
 def _get_dummy_catalog(stream_name: str) -> ConfiguredAirbyteCatalog:
@@ -146,6 +152,16 @@ 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,
+        Field(
+            description="If True, include raw HTTP request and response data in slices structure"
+        ),
+    ] = False,
 ) -> StreamTestResult:
     """Execute reading from a connector stream.
 
@@ -154,9 +170,15 @@ def execute_stream_test_read(
         config: Connector configuration
         stream_name: Name of the stream to test
         max_records: Maximum number of records to read
+        include_records: If True, include actual record data in the response
+        include_raw_response_data: If True, include raw HTTP request and response data in slices structure
 
     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.
     """
     logger.info(f"Testing stream read for stream: {stream_name}")
 
@@ -185,22 +207,79 @@ 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
+
+                    if include_records:
+                        if isinstance(stream_data, dict) and "records" in stream_data:
+                            records_data = stream_data["records"]
+                        elif isinstance(stream_data, dict):
+                            records_data = [stream_data]
+
+                    if include_raw_response_data:
+                        slices = (
+                            stream_data.get("slices", []) if isinstance(stream_data, dict) else []
+                        )
+                        slices_data = filter_config_secrets(slices)  # 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:
+            if result.record and result.record.data:
+                try:
+                    stream_data = result.record.data
+                    slices = stream_data.get("slices", []) if isinstance(stream_data, dict) else []
+                    slices_data = filter_config_secrets(slices)  # 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:
+            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 []
+                    slices_data = filter_config_secrets(slices)  # 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,35 +14,42 @@ 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.
 
     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: