validate_schema
Validates data against a schema definition to verify structure and constraints, ensuring data quality before processing.
Instructions
Validate data against a schema definition.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| session_id | Yes | ||
| schema | Yes |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- The core handler for validate_schema. Validates a DataFrame against a schema definition including type checks, nullable checks, min/max, pattern matching, allowed values, uniqueness, and string length constraints.
async def validate_schema( session_id: str, schema: dict[str, dict[str, Any]], ctx: Context = None ) -> dict[str, Any]: """ Validate data against a schema definition. Args: session_id: Session identifier schema: Schema definition with column rules Example: { "column_name": { "type": "int", # int, float, str, bool, datetime "nullable": False, "min": 0, "max": 100, "pattern": "^[A-Z]+$", "values": ["A", "B", "C"], # allowed values "unique": True } } ctx: FastMCP context Returns: Dict with validation results """ try: manager = get_session_manager() session = manager.get_session(session_id) if not session or session.df is None: return {"success": False, "error": "Invalid session or no data loaded"} df = session.df validation_errors = {} validation_summary = { "total_columns": len(schema), "valid_columns": 0, "invalid_columns": 0, "missing_columns": [], "extra_columns": [], } # Check for missing and extra columns schema_columns = set(schema.keys()) df_columns = set(df.columns) validation_summary["missing_columns"] = list(schema_columns - df_columns) validation_summary["extra_columns"] = list(df_columns - schema_columns) # Validate each column in schema for col_name, rules in schema.items(): if col_name not in df.columns: validation_errors[col_name] = [ {"error": "column_missing", "message": f"Column '{col_name}' not found in data"} ] validation_summary["invalid_columns"] += 1 continue col_errors = [] col_data = df[col_name] # Type validation expected_type = rules.get("type") if expected_type: type_valid = False if expected_type == "int": type_valid = pd.api.types.is_integer_dtype(col_data) elif expected_type == "float": type_valid = pd.api.types.is_float_dtype(col_data) elif expected_type == "str": type_valid = pd.api.types.is_string_dtype(col_data) or col_data.dtype == object elif expected_type == "bool": type_valid = pd.api.types.is_bool_dtype(col_data) elif expected_type == "datetime": type_valid = pd.api.types.is_datetime64_any_dtype(col_data) if not type_valid: col_errors.append( { "error": "type_mismatch", "message": f"Expected type '{expected_type}', got '{col_data.dtype}'", "actual_type": str(col_data.dtype), } ) # Nullable validation if not rules.get("nullable", True): null_count = col_data.isna().sum() if null_count > 0: col_errors.append( { "error": "null_values", "message": f"Column contains {null_count} null values", "null_count": int(null_count), "null_indices": df[col_data.isna()].index.tolist()[:100], } ) # Min/Max validation for numeric columns if pd.api.types.is_numeric_dtype(col_data): if "min" in rules: min_val = rules["min"] violations = col_data[col_data < min_val] if len(violations) > 0: col_errors.append( { "error": "min_violation", "message": f"{len(violations)} values below minimum {min_val}", "violation_count": len(violations), "min_found": float(violations.min()), } ) if "max" in rules: max_val = rules["max"] violations = col_data[col_data > max_val] if len(violations) > 0: col_errors.append( { "error": "max_violation", "message": f"{len(violations)} values above maximum {max_val}", "violation_count": len(violations), "max_found": float(violations.max()), } ) # Pattern validation for string columns if "pattern" in rules and ( col_data.dtype == object or pd.api.types.is_string_dtype(col_data) ): pattern = rules["pattern"] try: non_null = col_data.dropna() if len(non_null) > 0: matches = non_null.astype(str).str.match(pattern) violations = non_null[~matches] if len(violations) > 0: col_errors.append( { "error": "pattern_violation", "message": f"{len(violations)} values don't match pattern '{pattern}'", "violation_count": len(violations), "sample_violations": violations.head(10).tolist(), } ) except Exception as e: col_errors.append( {"error": "pattern_error", "message": f"Invalid regex pattern: {e!s}"} ) # Allowed values validation if "values" in rules: allowed = set(rules["values"]) actual = set(col_data.dropna().unique()) invalid = actual - allowed if invalid: col_errors.append( { "error": "invalid_values", "message": f"Found {len(invalid)} invalid values", "invalid_values": list(invalid)[:50], } ) # Uniqueness validation if rules.get("unique", False): duplicates = col_data.duplicated() if duplicates.any(): col_errors.append( { "error": "duplicate_values", "message": f"Column contains {duplicates.sum()} duplicate values", "duplicate_count": int(duplicates.sum()), } ) # Length validation for strings if col_data.dtype == object or pd.api.types.is_string_dtype(col_data): if "min_length" in rules: min_len = rules["min_length"] str_data = col_data.dropna().astype(str) short = str_data[str_data.str.len() < min_len] if len(short) > 0: col_errors.append( { "error": "min_length_violation", "message": f"{len(short)} values shorter than {min_len} characters", "violation_count": len(short), } ) if "max_length" in rules: max_len = rules["max_length"] str_data = col_data.dropna().astype(str) long = str_data[str_data.str.len() > max_len] if len(long) > 0: col_errors.append( { "error": "max_length_violation", "message": f"{len(long)} values longer than {max_len} characters", "violation_count": len(long), } ) if col_errors: validation_errors[col_name] = col_errors validation_summary["invalid_columns"] += 1 else: validation_summary["valid_columns"] += 1 is_valid = len(validation_errors) == 0 and len(validation_summary["missing_columns"]) == 0 session.record_operation( OperationType.VALIDATE, { "type": "schema_validation", "is_valid": is_valid, "errors_count": len(validation_errors), }, ) return { "success": True, "is_valid": is_valid, "summary": validation_summary, "validation_errors": validation_errors, } except Exception as e: logger.error(f"Error validating schema: {e!s}") return {"success": False, "error": str(e)} - src/csv_editor/server.py:380-388 (registration)Import and registration of validate_schema as a FastMCP tool. Imports the underlying function from tools.validation and wraps it with the @mcp.tool decorator.
from .tools.validation import validate_schema as _validate_schema @mcp.tool async def validate_schema( session_id: str, schema: dict[str, dict[str, Any]], ctx: Context = None ) -> dict[str, Any]: """Validate data against a schema definition.""" return await _validate_schema(session_id, schema, ctx) - The docstring and function signature defining the validate_schema input schema: session_id (str), schema (dict mapping column names to rule dicts with type, nullable, min, max, pattern, values, unique).
async def validate_schema( session_id: str, schema: dict[str, dict[str, Any]], ctx: Context = None ) -> dict[str, Any]: """ Validate data against a schema definition. Args: session_id: Session identifier schema: Schema definition with column rules Example: { "column_name": { "type": "int", # int, float, str, bool, datetime "nullable": False, "min": 0, "max": 100, "pattern": "^[A-Z]+$", "values": ["A", "B", "C"], # allowed values "unique": True } } ctx: FastMCP context - src/csv_editor/server.py:88-88 (registration)Capability listing - 'validate_schema' is advertised under data_validation capabilities.
"data_validation": ["validate_schema", "check_data_quality", "find_anomalies"], - Records the validation operation via session.record_operation with OperationType.VALIDATE.
session.record_operation( OperationType.VALIDATE, { "type": "schema_validation", "is_valid": is_valid, "errors_count": len(validation_errors), }, ) return { "success": True, "is_valid": is_valid, "summary": validation_summary, "validation_errors": validation_errors, } except Exception as e: logger.error(f"Error validating schema: {e!s}") return {"success": False, "error": str(e)}