Core EDA tools

.kiro/specs/hf-eda-mcp-server/tasks.md

@@ -6,7 +6,7 @@
   - Set up basic package initialization files
   - _Requirements: 3.1, 4.1, 4.2_
 
-- [ ] 2. Implement HuggingFace integration layer
+- [x] 2. Implement HuggingFace integration layer
   - [x] 2.1 Create HuggingFace client wrapper
     - Write HfClient class to handle authentication and API interactions
     - Implement dataset info retrieval using huggingface_hub
@@ -19,20 +19,20 @@
     - Implement dataset loading and sampling functionality
     - _Requirements: 1.1, 2.1, 2.2_
 
-- [ ] 3. Create core EDA tools
-  - [ ] 3.1 Implement dataset metadata tool
+- [x] 3. Create core EDA tools
+  - [x] 3.1 Implement dataset metadata tool
     - Write get_dataset_metadata function to retrieve comprehensive dataset information
     - Format metadata response with dataset size, features, splits, and configuration details
     - Handle multi-configuration datasets appropriately
     - _Requirements: 1.1, 1.3, 1.4_
 
-  - [ ] 3.2 Implement dataset sampling tool
+  - [x] 3.2 Implement dataset sampling tool
     - Create get_dataset_sample function for retrieving dataset samples
     - Support different splits (train, validation, test) and configurable sample sizes
     - Implement efficient sampling strategies for large datasets
     - _Requirements: 2.1, 2.2, 2.3_
 
-  - [ ] 3.3 Implement basic analysis tool
+  - [x] 3.3 Implement basic analysis tool
     - Write analyze_dataset_features function for exploratory data analysis
     - Generate feature statistics, missing value analysis, and data quality insights
     - Handle different data types (numerical, categorical, text) appropriately

src/hf_eda_mcp/tools/__init__.py

@@ -4,4 +4,21 @@ EDA tools module for HuggingFace datasets.
 This package contains individual EDA functions that will be exposed as MCP tools.
 """
 
-__all__ = []
+from .metadata import get_dataset_metadata, validate_dataset_metadata_inputs
+from .sampling import get_dataset_sample, get_dataset_sample_with_indices, get_available_splits
+from .analysis import analyze_dataset_features, validate_analysis_inputs
+
+__all__ = [
+    # Metadata tools
+    'get_dataset_metadata',
+    'validate_dataset_metadata_inputs',
+    
+    # Sampling tools
+    'get_dataset_sample',
+    'get_dataset_sample_with_indices', 
+    'get_available_splits',
+    
+    # Analysis tools
+    'analyze_dataset_features',
+    'validate_analysis_inputs'
+]

src/hf_eda_mcp/tools/analysis.py

@@ -1,7 +1,544 @@
 """
-Basic analysis tool for exploratory data analysis.
+Basic analysis tools for exploratory data analysis of HuggingFace datasets.
 
-This module will be implemented in task 3.3.
+This module provides tools for performing exploratory data analysis including
+feature statistics, missing value analysis, and data quality insights.
 """
 
-# Placeholder - will be implemented in task 3.3
+import logging
+import statistics
+from typing import Optional, Dict, Any, List
+from collections import Counter
+from hf_eda_mcp.services.dataset_service import DatasetService, DatasetServiceError
+from hf_eda_mcp.integrations.hf_client import DatasetNotFoundError, AuthenticationError
+
+logger = logging.getLogger(__name__)
+
+# Global dataset service instance
+_dataset_service: Optional[DatasetService] = None
+
+# Constants for analysis
+DEFAULT_ANALYSIS_SAMPLE_SIZE = 1000
+MAX_ANALYSIS_SAMPLE_SIZE = 50000
+MAX_UNIQUE_VALUES_TO_SHOW = 20
+
+
+def get_dataset_service() -> DatasetService:
+    """Get or create the global dataset service instance."""
+    global _dataset_service
+    if _dataset_service is None:
+        _dataset_service = DatasetService()
+    return _dataset_service
+
+
+def analyze_dataset_features(
+    dataset_id: str,
+    split: str = "train",
+    sample_size: int = DEFAULT_ANALYSIS_SAMPLE_SIZE,
+    config_name: Optional[str] = None,
+) -> Dict[str, Any]:
+    """
+    Perform basic exploratory analysis on dataset features.
+
+    This function analyzes dataset features to provide insights into data types,
+    distributions, missing values, and data quality. It handles different data
+    types (numerical, categorical, text) appropriately and generates comprehensive
+    statistics for each feature.
+
+    Args:
+        dataset_id: HuggingFace dataset identifier (e.g., 'imdb', 'squad')
+        split: Dataset split to analyze (default: 'train')
+        sample_size: Number of samples to use for analysis (default: 1000, max: 50000)
+        config_name: Optional configuration name for multi-config datasets
+
+    Returns:
+        Dictionary containing comprehensive feature analysis:
+        - dataset_info: Basic dataset information
+        - sample_info: Information about the sample used for analysis
+        - features: Dictionary with analysis for each feature including:
+          - feature_type: Detected type (numerical, categorical, text, etc.)
+          - missing_count: Number of missing/null values
+          - missing_percentage: Percentage of missing values
+          - unique_count: Number of unique values
+          - statistics: Type-specific statistics (mean, std for numerical; top values for categorical)
+        - summary: Overall analysis summary
+        - data_quality: Data quality assessment
+
+    Raises:
+        ValueError: If inputs are invalid
+        DatasetNotFoundError: If dataset or split doesn't exist
+        AuthenticationError: If dataset is private and authentication fails
+        DatasetServiceError: If analysis fails for other reasons
+
+    Example:
+        >>> analysis = analyze_dataset_features("imdb", sample_size=500)
+        >>> for feature_name, feature_analysis in analysis['features'].items():
+        ...     print(f"{feature_name}: {feature_analysis['feature_type']}")
+        ...     print(f"  Missing: {feature_analysis['missing_percentage']:.1f}%")
+
+        >>> # Check data quality
+        >>> quality = analysis['data_quality']
+        >>> print(f"Overall quality score: {quality['quality_score']:.2f}")
+    """
+    # Input validation
+    validate_analysis_inputs(dataset_id, split, sample_size, config_name)
+
+    logger.info(
+        f"Analyzing features for dataset: {dataset_id}, split: {split}, "
+        f"sample_size: {sample_size}"
+        + (f", config: {config_name}" if config_name else "")
+    )
+
+    try:
+        # Get dataset service and load sample for analysis
+        service = get_dataset_service()
+        sample_data = service.load_dataset_sample(
+            dataset_id=dataset_id,
+            split=split,
+            num_samples=sample_size,
+            config_name=config_name,
+            streaming=True,
+        )
+
+        # Note: We could get dataset metadata here for additional context if needed
+
+        # Perform feature analysis
+        features_analysis = {}
+        data_samples = sample_data["data"]
+
+        if not data_samples:
+            raise DatasetServiceError("No data samples available for analysis")
+
+        # Determine feature names from first sample
+        first_sample = data_samples[0]
+        if not isinstance(first_sample, dict):
+            raise DatasetServiceError(
+                "Dataset samples are not in expected dictionary format"
+            )
+
+        feature_names = list(first_sample.keys())
+
+        # Analyze each feature
+        for feature_name in feature_names:
+            logger.debug(f"Analyzing feature: {feature_name}")
+            feature_analysis = _analyze_single_feature(feature_name, data_samples)
+            features_analysis[feature_name] = feature_analysis
+
+        # Generate overall analysis
+        analysis_result = {
+            "dataset_info": {
+                "dataset_id": dataset_id,
+                "config_name": config_name,
+                "split": split,
+                "total_features": len(feature_names),
+                "sample_size_used": len(data_samples),
+                "sample_size_requested": sample_size,
+            },
+            "sample_info": {
+                "sampling_method": "sequential_head",
+                "represents_full_dataset": len(data_samples) >= sample_size,
+                "analysis_timestamp": sample_data.get("_sampled_at"),
+            },
+            "features": features_analysis,
+            "data_quality": _assess_data_quality(features_analysis),
+            "summary": _generate_analysis_summary(features_analysis, len(data_samples)),
+        }
+
+        logger.info(
+            f"Successfully analyzed {len(feature_names)} features from {dataset_id}"
+        )
+        return analysis_result
+
+    except (DatasetNotFoundError, AuthenticationError):
+        # Re-raise these specific errors as-is
+        raise
+    except Exception as e:
+        logger.error(f"Failed to analyze dataset {dataset_id}: {str(e)}")
+        raise DatasetServiceError(f"Failed to analyze dataset features: {str(e)}")
+
+
+def _analyze_single_feature(
+    feature_name: str, data_samples: List[Dict[str, Any]]
+) -> Dict[str, Any]:
+    """
+    Analyze a single feature across all data samples.
+
+    Args:
+        feature_name: Name of the feature to analyze
+        data_samples: List of data sample dictionaries
+
+    Returns:
+        Dictionary containing feature analysis results
+    """
+    # Extract values for this feature
+    values = []
+    missing_count = 0
+
+    for sample in data_samples:
+        value = sample.get(feature_name)
+        if (
+            value is None
+            or value == ""
+            or (isinstance(value, float) and str(value).lower() == "nan")
+        ):
+            missing_count += 1
+        else:
+            values.append(value)
+
+    total_count = len(data_samples)
+    missing_percentage = (missing_count / total_count) * 100 if total_count > 0 else 0
+
+    # Determine feature type and compute statistics
+    feature_type, statistics_dict = _determine_feature_type_and_stats(values)
+
+    # Count unique values
+    unique_count = len(set(str(v) for v in values)) if values else 0
+
+    return {
+        "feature_type": feature_type,
+        "missing_count": missing_count,
+        "missing_percentage": missing_percentage,
+        "unique_count": unique_count,
+        "total_count": total_count,
+        "non_missing_count": len(values),
+        "statistics": statistics_dict,
+        "sample_values": values[:5] if values else [],  # First 5 values as examples
+    }
+
+
+def _determine_feature_type_and_stats(values: List[Any]) -> tuple[str, Dict[str, Any]]:
+    """
+    Determine the type of a feature and compute appropriate statistics.
+
+    Args:
+        values: List of non-missing values for the feature
+
+    Returns:
+        Tuple of (feature_type, statistics_dict)
+    """
+    if not values:
+        return "unknown", {}
+
+    # Check if all values are numeric
+    numeric_values = []
+    for value in values:
+        try:
+            if isinstance(value, (int, float)):
+                numeric_values.append(float(value))
+            elif isinstance(value, str):
+                # Try to convert string to number
+                numeric_values.append(float(value))
+            else:
+                # Not numeric
+                break
+        except (ValueError, TypeError):
+            # Not numeric
+            break
+    else:
+        # All values are numeric
+        if len(numeric_values) == len(values):
+            return "numerical", _compute_numerical_statistics(numeric_values)
+
+    # Check if values are boolean-like
+    boolean_values = set(str(v).lower() for v in values)
+    if boolean_values.issubset({"true", "false", "1", "0", "yes", "no"}):
+        return "boolean", _compute_categorical_statistics(values)
+
+    # Check if it's text (strings with average length > 10)
+    if all(isinstance(v, str) for v in values):
+        avg_length = sum(len(v) for v in values) / len(values)
+        if avg_length > 10:
+            return "text", _compute_text_statistics(values)
+
+    # Default to categorical
+    return "categorical", _compute_categorical_statistics(values)
+
+
+def _compute_numerical_statistics(values: List[float]) -> Dict[str, Any]:
+    """Compute statistics for numerical features."""
+    if not values:
+        return {}
+
+    try:
+        stats = {
+            "count": len(values),
+            "mean": statistics.mean(values),
+            "median": statistics.median(values),
+            "min": min(values),
+            "max": max(values),
+            "range": max(values) - min(values),
+        }
+
+        if len(values) > 1:
+            stats["std"] = statistics.stdev(values)
+            stats["variance"] = statistics.variance(values)
+
+        # Quartiles
+        sorted_values = sorted(values)
+        n = len(sorted_values)
+        if n >= 4:
+            stats["q1"] = sorted_values[n // 4]
+            stats["q3"] = sorted_values[3 * n // 4]
+            stats["iqr"] = stats["q3"] - stats["q1"]
+
+        return stats
+    except Exception as e:
+        logger.warning(f"Failed to compute numerical statistics: {e}")
+        return {"count": len(values), "error": str(e)}
+
+
+def _compute_categorical_statistics(values: List[Any]) -> Dict[str, Any]:
+    """Compute statistics for categorical features."""
+    if not values:
+        return {}
+
+    try:
+        # Convert all values to strings for consistent counting
+        str_values = [str(v) for v in values]
+        value_counts = Counter(str_values)
+
+        stats = {
+            "count": len(values),
+            "unique_count": len(value_counts),
+            "most_common": value_counts.most_common(MAX_UNIQUE_VALUES_TO_SHOW),
+            "top_value": value_counts.most_common(1)[0] if value_counts else None,
+        }
+
+        # Calculate entropy (measure of diversity)
+        if len(value_counts) > 1:
+            total = len(str_values)
+            entropy = -sum(
+                (count / total) * (count / total).bit_length()
+                for count in value_counts.values()
+                if count > 0
+            )
+            stats["entropy"] = entropy
+
+        return stats
+    except Exception as e:
+        logger.warning(f"Failed to compute categorical statistics: {e}")
+        return {"count": len(values), "error": str(e)}
+
+
+def _compute_text_statistics(values: List[str]) -> Dict[str, Any]:
+    """Compute statistics for text features."""
+    if not values:
+        return {}
+
+    try:
+        lengths = [len(v) for v in values]
+        word_counts = [len(v.split()) for v in values]
+
+        stats = {
+            "count": len(values),
+            "avg_length": statistics.mean(lengths),
+            "min_length": min(lengths),
+            "max_length": max(lengths),
+            "avg_word_count": statistics.mean(word_counts),
+            "min_word_count": min(word_counts),
+            "max_word_count": max(word_counts),
+        }
+
+        # Sample of values (first few)
+        stats["sample_texts"] = values[:3]
+
+        return stats
+    except Exception as e:
+        logger.warning(f"Failed to compute text statistics: {e}")
+        return {"count": len(values), "error": str(e)}
+
+
+def _assess_data_quality(
+    features_analysis: Dict[str, Dict[str, Any]],
+) -> Dict[str, Any]:
+    """
+    Assess overall data quality based on feature analysis.
+
+    Args:
+        features_analysis: Dictionary of feature analyses
+
+    Returns:
+        Dictionary containing data quality assessment
+    """
+    if not features_analysis:
+        return {"quality_score": 0.0, "issues": ["No features to analyze"]}
+
+    total_features = len(features_analysis)
+    issues = []
+    quality_factors = []
+
+    # Check missing value rates
+    high_missing_features = 0
+    total_missing_rate = 0
+
+    for feature_name, analysis in features_analysis.items():
+        missing_pct = analysis.get("missing_percentage", 0)
+        total_missing_rate += missing_pct
+
+        if missing_pct > 50:
+            high_missing_features += 1
+            issues.append(
+                f"Feature '{feature_name}' has {missing_pct:.1f}% missing values"
+            )
+        elif missing_pct > 20:
+            issues.append(
+                f"Feature '{feature_name}' has {missing_pct:.1f}% missing values"
+            )
+
+    avg_missing_rate = total_missing_rate / total_features
+
+    # Quality score calculation (0-1 scale)
+    missing_score = max(0, 1 - (avg_missing_rate / 100))
+    quality_factors.append(("missing_values", missing_score))
+
+    # Check for features with very low diversity
+    low_diversity_features = 0
+    for feature_name, analysis in features_analysis.items():
+        unique_count = analysis.get("unique_count", 0)
+        total_count = analysis.get("total_count", 1)
+        diversity_ratio = unique_count / total_count if total_count > 0 else 0
+
+        if diversity_ratio < 0.01 and analysis.get("feature_type") != "boolean":
+            low_diversity_features += 1
+            issues.append(
+                f"Feature '{feature_name}' has very low diversity ({unique_count} unique values)"
+            )
+
+    diversity_score = max(0, 1 - (low_diversity_features / total_features))
+    quality_factors.append(("diversity", diversity_score))
+
+    # Overall quality score (weighted average)
+    weights = {"missing_values": 0.6, "diversity": 0.4}
+    quality_score = sum(weights[factor] * score for factor, score in quality_factors)
+
+    # Quality assessment
+    if quality_score >= 0.8:
+        quality_level = "high"
+    elif quality_score >= 0.6:
+        quality_level = "medium"
+    else:
+        quality_level = "low"
+
+    return {
+        "quality_score": quality_score,
+        "quality_level": quality_level,
+        "avg_missing_rate": avg_missing_rate,
+        "high_missing_features": high_missing_features,
+        "low_diversity_features": low_diversity_features,
+        "issues": issues,
+        "recommendations": _generate_quality_recommendations(issues, quality_score),
+    }
+
+
+def _generate_quality_recommendations(
+    issues: List[str], quality_score: float
+) -> List[str]:
+    """Generate recommendations based on data quality issues."""
+    recommendations = []
+
+    if quality_score < 0.6:
+        recommendations.append(
+            "Consider data cleaning and preprocessing before analysis"
+        )
+
+    if any("missing values" in issue for issue in issues):
+        recommendations.append("Handle missing values through imputation or removal")
+
+    if any("low diversity" in issue for issue in issues):
+        recommendations.append(
+            "Review features with low diversity - they may not be informative"
+        )
+
+    if not recommendations:
+        recommendations.append("Data quality looks good for analysis")
+
+    return recommendations
+
+
+def _generate_analysis_summary(
+    features_analysis: Dict[str, Dict[str, Any]], sample_size: int
+) -> str:
+    """Generate a human-readable summary of the analysis."""
+    if not features_analysis:
+        return "No features analyzed"
+
+    total_features = len(features_analysis)
+
+    # Count feature types
+    type_counts = Counter(
+        analysis.get("feature_type", "unknown")
+        for analysis in features_analysis.values()
+    )
+
+    # Calculate average missing rate
+    missing_rates = [
+        analysis.get("missing_percentage", 0) for analysis in features_analysis.values()
+    ]
+    avg_missing = statistics.mean(missing_rates) if missing_rates else 0
+
+    summary_parts = [f"Analyzed {total_features} features from {sample_size} samples"]
+
+    # Feature type breakdown
+    type_summary = []
+    for ftype, count in type_counts.most_common():
+        type_summary.append(f"{count} {ftype}")
+
+    if type_summary:
+        summary_parts.append(f"Types: {', '.join(type_summary)}")
+
+    # Missing data summary
+    if avg_missing > 0:
+        summary_parts.append(f"Avg missing: {avg_missing:.1f}%")
+
+    return " | ".join(summary_parts)
+
+
+def validate_analysis_inputs(
+    dataset_id: str, split: str, sample_size: int, config_name: Optional[str] = None
+) -> None:
+    """
+    Validate inputs for dataset analysis.
+
+    Args:
+        dataset_id: Dataset identifier to validate
+        split: Split name to validate
+        sample_size: Sample size to validate
+        config_name: Optional configuration name to validate
+
+    Raises:
+        ValueError: If any input is invalid
+    """
+    # Validate dataset_id
+    if not dataset_id or not isinstance(dataset_id, str):
+        raise ValueError("dataset_id must be a non-empty string")
+
+    dataset_id = dataset_id.strip()
+    if not dataset_id:
+        raise ValueError("dataset_id cannot be empty or whitespace")
+
+    # Validate split
+    if not split or not isinstance(split, str):
+        raise ValueError("split must be a non-empty string")
+
+    split = split.strip()
+    if not split:
+        raise ValueError("split cannot be empty or whitespace")
+
+    # Validate sample_size
+    if not isinstance(sample_size, int):
+        raise ValueError("sample_size must be an integer")
+
+    if sample_size <= 0:
+        raise ValueError("sample_size must be positive")
+
+    if sample_size > MAX_ANALYSIS_SAMPLE_SIZE:
+        raise ValueError(f"sample_size cannot exceed {MAX_ANALYSIS_SAMPLE_SIZE}")
+
+    # Validate config_name
+    if config_name is not None:
+        if not isinstance(config_name, str):
+            raise ValueError("config_name must be a string")
+
+        config_name = config_name.strip()
+        if not config_name:
+            raise ValueError("config_name cannot be empty or whitespace")

src/hf_eda_mcp/tools/metadata.py

@@ -1,7 +1,217 @@
 """
 Dataset metadata tool for retrieving HuggingFace dataset information.
 
-This module will be implemented in task 3.1.
+This module provides tools for retrieving comprehensive metadata about
+HuggingFace datasets including size, features, splits, and configuration details.
 """
 
-# Placeholder - will be implemented in task 3.1
+import logging
+from typing import Optional, Dict, Any
+from hf_eda_mcp.services.dataset_service import DatasetService, DatasetServiceError
+from hf_eda_mcp.integrations.hf_client import DatasetNotFoundError, AuthenticationError
+
+logger = logging.getLogger(__name__)
+
+# Global dataset service instance
+_dataset_service: Optional[DatasetService] = None
+
+
+def get_dataset_service() -> DatasetService:
+    """Get or create the global dataset service instance."""
+    global _dataset_service
+    if _dataset_service is None:
+        _dataset_service = DatasetService()
+    return _dataset_service
+
+
+def get_dataset_metadata(dataset_id: str, config_name: Optional[str] = None) -> Dict[str, Any]:
+    """
+    Retrieve comprehensive metadata for a HuggingFace dataset.
+    
+    This function fetches detailed information about a dataset including its size,
+    features, available splits, configurations, and other metadata. It handles
+    multi-configuration datasets appropriately and provides caching for efficiency.
+    
+    Args:
+        dataset_id: HuggingFace dataset identifier (e.g., 'squad', 'glue', 'imdb')
+        config_name: Optional configuration name for multi-config datasets
+        
+    Returns:
+        Dictionary containing comprehensive dataset metadata:
+        - id: Dataset identifier
+        - author: Dataset author/organization
+        - description: Dataset description
+        - features: Dictionary of feature names and types
+        - splits: Dictionary of split names and their sizes
+        - configs: List of available configurations
+        - size_bytes: Dataset size in bytes
+        - downloads: Number of downloads
+        - likes: Number of likes
+        - tags: List of dataset tags
+        - created_at: Creation timestamp
+        - last_modified: Last modification timestamp
+        
+    Raises:
+        ValueError: If dataset_id is empty or invalid
+        DatasetNotFoundError: If dataset doesn't exist on HuggingFace Hub
+        AuthenticationError: If dataset is private and authentication fails
+        DatasetServiceError: If metadata retrieval fails for other reasons
+        
+    Example:
+        >>> metadata = get_dataset_metadata("imdb")
+        >>> print(f"Dataset: {metadata['id']}")
+        >>> print(f"Splits: {list(metadata['splits'].keys())}")
+        >>> print(f"Features: {list(metadata['features'].keys())}")
+        
+        >>> # For multi-config dataset
+        >>> metadata = get_dataset_metadata("glue", config_name="cola")
+        >>> print(f"Config: {metadata.get('config_name', 'default')}")
+    """
+    # Input validation
+    if not dataset_id or not isinstance(dataset_id, str):
+        raise ValueError("dataset_id must be a non-empty string")
+    
+    dataset_id = dataset_id.strip()
+    if not dataset_id:
+        raise ValueError("dataset_id cannot be empty or whitespace")
+    
+    if config_name is not None:
+        config_name = config_name.strip()
+        if not config_name:
+            config_name = None
+    
+    logger.info(f"Retrieving metadata for dataset: {dataset_id}" + 
+                (f", config: {config_name}" if config_name else ""))
+    
+    try:
+        # Get dataset service and retrieve metadata
+        service = get_dataset_service()
+        metadata = service.load_dataset_info(dataset_id, config_name)
+        
+        # Add the requested config name to the response if specified
+        if config_name:
+            metadata['config_name'] = config_name
+        
+        # Enhance metadata with additional computed fields
+        metadata['total_configs'] = len(metadata.get('configs', []))
+        metadata['total_splits'] = len(metadata.get('splits', {}))
+        metadata['has_multiple_configs'] = metadata['total_configs'] > 1
+        
+        # Format size for human readability
+        size_bytes = metadata.get('size_bytes', 0)
+        if size_bytes > 0:
+            metadata['size_human'] = _format_bytes(size_bytes)
+        else:
+            metadata['size_human'] = 'Unknown'
+        
+        # Add summary information
+        metadata['summary'] = _generate_metadata_summary(metadata)
+        
+        logger.info(f"Successfully retrieved metadata for {dataset_id}")
+        return metadata
+        
+    except (DatasetNotFoundError, AuthenticationError):
+        # Re-raise these specific errors as-is
+        raise
+    except Exception as e:
+        logger.error(f"Failed to retrieve metadata for {dataset_id}: {str(e)}")
+        raise DatasetServiceError(f"Failed to retrieve dataset metadata: {str(e)}")
+
+
+def _format_bytes(size_bytes: int) -> str:
+    """Format byte size in human-readable format."""
+    if size_bytes == 0:
+        return "0 B"
+    
+    units = ['B', 'KB', 'MB', 'GB', 'TB']
+    size = float(size_bytes)
+    unit_index = 0
+    
+    while size >= 1024 and unit_index < len(units) - 1:
+        size /= 1024
+        unit_index += 1
+    
+    if unit_index == 0:
+        return f"{int(size)} {units[unit_index]}"
+    else:
+        return f"{size:.1f} {units[unit_index]}"
+
+
+def _generate_metadata_summary(metadata: Dict[str, Any]) -> str:
+    """Generate a human-readable summary of dataset metadata."""
+    summary_parts = []
+    
+    # Basic info
+    summary_parts.append(f"Dataset: {metadata.get('id', 'Unknown')}")
+    
+    if metadata.get('author'):
+        summary_parts.append(f"Author: {metadata['author']}")
+    
+    # Size and popularity
+    if metadata.get('size_human'):
+        summary_parts.append(f"Size: {metadata['size_human']}")
+    
+    downloads = metadata.get('downloads', 0)
+    if downloads > 0:
+        summary_parts.append(f"Downloads: {downloads:,}")
+    
+    likes = metadata.get('likes', 0)
+    if likes > 0:
+        summary_parts.append(f"Likes: {likes:,}")
+    
+    # Structure info
+    configs = metadata.get('configs', [])
+    if configs:
+        if len(configs) == 1:
+            summary_parts.append(f"Configuration: {configs[0]}")
+        else:
+            summary_parts.append(f"Configurations: {len(configs)} available")
+    
+    splits = metadata.get('splits', {})
+    if splits:
+        split_names = list(splits.keys())
+        if len(split_names) <= 3:
+            summary_parts.append(f"Splits: {', '.join(split_names)}")
+        else:
+            summary_parts.append(f"Splits: {len(split_names)} available")
+    
+    features = metadata.get('features', {})
+    if features:
+        summary_parts.append(f"Features: {len(features)} columns")
+    
+    return " | ".join(summary_parts)
+
+
+def validate_dataset_metadata_inputs(dataset_id: str, config_name: Optional[str] = None) -> None:
+    """
+    Validate inputs for dataset metadata retrieval.
+    
+    Args:
+        dataset_id: Dataset identifier to validate
+        config_name: Optional configuration name to validate
+        
+    Raises:
+        ValueError: If inputs are invalid
+    """
+    if not dataset_id or not isinstance(dataset_id, str):
+        raise ValueError("dataset_id must be a non-empty string")
+    
+    dataset_id = dataset_id.strip()
+    if not dataset_id:
+        raise ValueError("dataset_id cannot be empty or whitespace")
+    
+    # Basic format validation for dataset_id
+    if not all(c.isalnum() or c in '-_/.@' for c in dataset_id):
+        raise ValueError("dataset_id contains invalid characters")
+    
+    if config_name is not None:
+        if not isinstance(config_name, str):
+            raise ValueError("config_name must be a string")
+        
+        config_name = config_name.strip()
+        if not config_name:
+            raise ValueError("config_name cannot be empty or whitespace")
+        
+        # Basic format validation for config_name
+        if not all(c.isalnum() or c in '-_.' for c in config_name):
+            raise ValueError("config_name contains invalid characters")

src/hf_eda_mcp/tools/sampling.py

@@ -1,7 +1,330 @@
 """
-Dataset sampling tool for retrieving dataset samples.
+Dataset sampling tool for retrieving samples from HuggingFace datasets.
 
-This module will be implemented in task 3.2.
+This module provides tools for efficiently sampling data from HuggingFace datasets
+with support for different splits, configurable sample sizes, and streaming for large datasets.
 """
 
-# Placeholder - will be implemented in task 3.2
+import logging
+from typing import Optional, Dict, Any, List
+from hf_eda_mcp.services.dataset_service import DatasetService, DatasetServiceError
+from hf_eda_mcp.integrations.hf_client import DatasetNotFoundError, AuthenticationError
+
+logger = logging.getLogger(__name__)
+
+# Global dataset service instance
+_dataset_service: Optional[DatasetService] = None
+
+# Constants for sampling limits
+MAX_SAMPLE_SIZE = 10000  # Maximum samples to prevent memory issues
+DEFAULT_SAMPLE_SIZE = 10
+VALID_SPLITS = {'train', 'validation', 'test', 'dev', 'val'}
+
+
+def get_dataset_service() -> DatasetService:
+    """Get or create the global dataset service instance."""
+    global _dataset_service
+    if _dataset_service is None:
+        _dataset_service = DatasetService()
+    return _dataset_service
+
+
+def get_dataset_sample(
+    dataset_id: str,
+    split: str = "train",
+    num_samples: int = DEFAULT_SAMPLE_SIZE,
+    config_name: Optional[str] = None,
+    streaming: bool = True
+) -> Dict[str, Any]:
+    """
+    Retrieve a sample of rows from a HuggingFace dataset.
+    
+    This function efficiently samples data from datasets with support for different
+    splits and configurable sample sizes. It uses streaming by default for large
+    datasets to minimize memory usage and loading time.
+    
+    Args:
+        dataset_id: HuggingFace dataset identifier (e.g., 'imdb', 'squad', 'glue')
+        split: Dataset split to sample from (default: 'train')
+        num_samples: Number of samples to retrieve (default: 10, max: 10000)
+        config_name: Optional configuration name for multi-config datasets
+        streaming: Whether to use streaming mode for efficient loading (default: True)
+        
+    Returns:
+        Dictionary containing sampled data and metadata:
+        - dataset_id: Original dataset identifier
+        - config_name: Configuration name used (if any)
+        - split: Split name sampled from
+        - num_samples: Actual number of samples returned
+        - requested_samples: Number of samples originally requested
+        - data: List of sample dictionaries
+        - schema: Dictionary describing the dataset features/columns
+        - sample_info: Additional information about the sampling process
+        
+    Raises:
+        ValueError: If inputs are invalid (empty dataset_id, invalid split, etc.)
+        DatasetNotFoundError: If dataset or split doesn't exist
+        AuthenticationError: If dataset is private and authentication fails
+        DatasetServiceError: If sampling fails for other reasons
+        
+    Example:
+        >>> # Basic sampling
+        >>> sample = get_dataset_sample("imdb", split="train", num_samples=5)
+        >>> print(f"Got {sample['num_samples']} samples from {sample['dataset_id']}")
+        >>> for i, row in enumerate(sample['data']):
+        ...     print(f"Sample {i+1}: {list(row.keys())}")
+        
+        >>> # Multi-config dataset sampling
+        >>> sample = get_dataset_sample("glue", split="validation", 
+        ...                           num_samples=3, config_name="cola")
+        >>> print(f"Schema: {sample['schema']}")
+    """
+    # Input validation
+    validate_sampling_inputs(dataset_id, split, num_samples, config_name)
+    
+    logger.info(f"Sampling {num_samples} rows from dataset: {dataset_id}, "
+                f"split: {split}" + (f", config: {config_name}" if config_name else ""))
+    
+    try:
+        # Get dataset service and load sample
+        service = get_dataset_service()
+        sample_data = service.load_dataset_sample(
+            dataset_id=dataset_id,
+            split=split,
+            num_samples=num_samples,
+            config_name=config_name,
+            streaming=streaming
+        )
+        
+        # Enhance the response with additional metadata
+        sample_data['sample_info'] = {
+            'streaming_used': streaming,
+            'sampling_strategy': 'sequential_head',  # We take first N samples
+            'max_sample_size': MAX_SAMPLE_SIZE,
+            'truncated': sample_data['num_samples'] < sample_data['requested_samples']
+        }
+        
+        # Add data preview information
+        if sample_data['data']:
+            first_sample = sample_data['data'][0]
+            sample_data['sample_info']['preview'] = {
+                'columns': list(first_sample.keys()) if isinstance(first_sample, dict) else [],
+                'first_sample_types': {
+                    k: type(v).__name__ for k, v in first_sample.items()
+                } if isinstance(first_sample, dict) else {}
+            }
+        
+        # Add summary
+        sample_data['summary'] = _generate_sample_summary(sample_data)
+        
+        logger.info(f"Successfully sampled {sample_data['num_samples']} rows from {dataset_id}")
+        return sample_data
+        
+    except (DatasetNotFoundError, AuthenticationError):
+        # Re-raise these specific errors as-is
+        raise
+    except Exception as e:
+        logger.error(f"Failed to sample from dataset {dataset_id}: {str(e)}")
+        raise DatasetServiceError(f"Failed to sample dataset: {str(e)}")
+
+
+def get_dataset_sample_with_indices(
+    dataset_id: str,
+    indices: List[int],
+    split: str = "train",
+    config_name: Optional[str] = None
+) -> Dict[str, Any]:
+    """
+    Retrieve specific samples by their indices from a HuggingFace dataset.
+    
+    This function allows for targeted sampling by specifying exact row indices.
+    Note: This requires loading the dataset in non-streaming mode.
+    
+    Args:
+        dataset_id: HuggingFace dataset identifier
+        indices: List of row indices to retrieve
+        split: Dataset split to sample from (default: 'train')
+        config_name: Optional configuration name for multi-config datasets
+        
+    Returns:
+        Dictionary containing the requested samples and metadata
+        
+    Raises:
+        ValueError: If inputs are invalid
+        DatasetServiceError: If sampling fails
+    """
+    # Input validation
+    if not indices or not isinstance(indices, list):
+        raise ValueError("indices must be a non-empty list")
+    
+    if not all(isinstance(i, int) and i >= 0 for i in indices):
+        raise ValueError("All indices must be non-negative integers")
+    
+    if len(indices) > MAX_SAMPLE_SIZE:
+        raise ValueError(f"Too many indices requested. Maximum: {MAX_SAMPLE_SIZE}")
+    
+    validate_sampling_inputs(dataset_id, split, len(indices), config_name)
+    
+    logger.info(f"Sampling {len(indices)} specific indices from dataset: {dataset_id}")
+    
+    try:
+        from datasets import load_dataset
+        
+        # Load dataset without streaming to access by index
+        dataset = load_dataset(
+            dataset_id,
+            name=config_name,
+            split=split,
+            streaming=False
+        )
+        
+        # Validate indices are within bounds
+        max_index = max(indices)
+        if max_index >= len(dataset):
+            raise ValueError(f"Index {max_index} is out of bounds for dataset with {len(dataset)} rows")
+        
+        # Get samples by indices
+        samples = [dataset[i] for i in indices]
+        
+        # Get dataset info for schema
+        service = get_dataset_service()
+        dataset_info = service.load_dataset_info(dataset_id, config_name)
+        
+        # Prepare response
+        sample_data = {
+            'dataset_id': dataset_id,
+            'config_name': config_name,
+            'split': split,
+            'num_samples': len(samples),
+            'requested_indices': indices,
+            'data': samples,
+            'schema': dataset_info.get('features', {}),
+            'sample_info': {
+                'sampling_strategy': 'by_indices',
+                'streaming_used': False,
+                'indices_requested': len(indices)
+            }
+        }
+        
+        sample_data['summary'] = _generate_sample_summary(sample_data)
+        
+        return sample_data
+        
+    except Exception as e:
+        logger.error(f"Failed to sample by indices from {dataset_id}: {str(e)}")
+        raise DatasetServiceError(f"Failed to sample by indices: {str(e)}")
+
+
+def validate_sampling_inputs(
+    dataset_id: str,
+    split: str,
+    num_samples: int,
+    config_name: Optional[str] = None
+) -> None:
+    """
+    Validate inputs for dataset sampling.
+    
+    Args:
+        dataset_id: Dataset identifier to validate
+        split: Split name to validate
+        num_samples: Number of samples to validate
+        config_name: Optional configuration name to validate
+        
+    Raises:
+        ValueError: If any input is invalid
+    """
+    # Validate dataset_id
+    if not dataset_id or not isinstance(dataset_id, str):
+        raise ValueError("dataset_id must be a non-empty string")
+    
+    dataset_id = dataset_id.strip()
+    if not dataset_id:
+        raise ValueError("dataset_id cannot be empty or whitespace")
+    
+    # Validate split
+    if not split or not isinstance(split, str):
+        raise ValueError("split must be a non-empty string")
+    
+    split = split.strip().lower()
+    if not split:
+        raise ValueError("split cannot be empty or whitespace")
+    
+    # Note: We don't strictly enforce VALID_SPLITS as datasets may have custom split names
+    
+    # Validate num_samples
+    if not isinstance(num_samples, int):
+        raise ValueError("num_samples must be an integer")
+    
+    if num_samples <= 0:
+        raise ValueError("num_samples must be positive")
+    
+    if num_samples > MAX_SAMPLE_SIZE:
+        raise ValueError(f"num_samples cannot exceed {MAX_SAMPLE_SIZE}")
+    
+    # Validate config_name
+    if config_name is not None:
+        if not isinstance(config_name, str):
+            raise ValueError("config_name must be a string")
+        
+        config_name = config_name.strip()
+        if not config_name:
+            raise ValueError("config_name cannot be empty or whitespace")
+
+
+def _generate_sample_summary(sample_data: Dict[str, Any]) -> str:
+    """Generate a human-readable summary of the sample data."""
+    summary_parts = []
+    
+    # Basic info
+    summary_parts.append(f"Dataset: {sample_data.get('dataset_id', 'Unknown')}")
+    summary_parts.append(f"Split: {sample_data.get('split', 'Unknown')}")
+    
+    if sample_data.get('config_name'):
+        summary_parts.append(f"Config: {sample_data['config_name']}")
+    
+    # Sample info
+    num_samples = sample_data.get('num_samples', 0)
+    requested = sample_data.get('requested_samples', num_samples)
+    
+    if num_samples == requested:
+        summary_parts.append(f"Samples: {num_samples}")
+    else:
+        summary_parts.append(f"Samples: {num_samples}/{requested} (truncated)")
+    
+    # Schema info
+    schema = sample_data.get('schema', {})
+    if schema:
+        summary_parts.append(f"Columns: {len(schema)}")
+    
+    # Sampling strategy
+    sample_info = sample_data.get('sample_info', {})
+    strategy = sample_info.get('sampling_strategy', 'unknown')
+    if strategy == 'by_indices':
+        summary_parts.append("Strategy: by indices")
+    elif strategy == 'sequential_head':
+        summary_parts.append("Strategy: first N rows")
+    
+    return " | ".join(summary_parts)
+
+
+def get_available_splits(dataset_id: str, config_name: Optional[str] = None) -> List[str]:
+    """
+    Get available splits for a dataset.
+    
+    Args:
+        dataset_id: HuggingFace dataset identifier
+        config_name: Optional configuration name
+        
+    Returns:
+        List of available split names
+        
+    Raises:
+        DatasetServiceError: If unable to retrieve split information
+    """
+    try:
+        service = get_dataset_service()
+        metadata = service.load_dataset_info(dataset_id, config_name)
+        return list(metadata.get('splits', {}).keys())
+    except Exception as e:
+        logger.error(f"Failed to get splits for {dataset_id}: {str(e)}")
+        raise DatasetServiceError(f"Failed to get available splits: {str(e)}")