filter_data
Filter CSV data by applying conditions to columns, returning matching rows with optional row limits for data analysis.
Instructions
Filter CSV data based on conditions.
Args:
filename: Name of the CSV file
conditions: Dictionary of column conditions.
Simple: {"column": "value"}
Complex: {"column": {"gt": 5, "lt": 10, "contains": "text"}}
limit: Optional limit on number of rows to return
Returns:
Dictionary with filtered dataInput Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| filename | Yes | ||
| conditions | Yes | ||
| limit | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- csv_mcp_server/server.py:180-202 (registration)Registration and handler for the 'filter_data' MCP tool. This function is decorated with @mcp.tool() making it available as an MCP tool, and it delegates the logic to CSVManager.filter_data.
@mcp.tool() def filter_data( filename: str, conditions: Dict[str, Any], limit: Optional[int] = None ) -> Dict[str, Any]: """ Filter CSV data based on conditions. Args: filename: Name of the CSV file conditions: Dictionary of column conditions. Simple: {"column": "value"} Complex: {"column": {"gt": 5, "lt": 10, "contains": "text"}} limit: Optional limit on number of rows to return Returns: Dictionary with filtered data """ try: return csv_manager.filter_data(filename, conditions, limit) except Exception as e: return {"success": False, "error": str(e)} - csv_mcp_server/csv_manager.py:360-409 (handler)Core implementation of data filtering logic in CSVManager class. Applies conditions to pandas DataFrame and returns filtered results.
def filter_data(self, filename: str, conditions: Dict[str, Any], limit: Optional[int] = None) -> Dict[str, Any]: """Filter CSV data based on conditions.""" filepath = self._get_file_path(filename) if not filepath.exists(): raise FileNotFoundError(f"CSV file '{filename}' not found") try: df = pd.read_csv(filepath) # Apply filters for column, condition in conditions.items(): if column not in df.columns: raise ValueError(f"Column '{column}' not found in CSV") if isinstance(condition, dict): # Handle complex conditions like {"gt": 5, "lt": 10} if "eq" in condition: df = df[df[column] == condition["eq"]] if "ne" in condition: df = df[df[column] != condition["ne"]] if "gt" in condition: df = df[df[column] > condition["gt"]] if "gte" in condition: df = df[df[column] >= condition["gte"]] if "lt" in condition: df = df[df[column] < condition["lt"]] if "lte" in condition: df = df[df[column] <= condition["lte"]] if "contains" in condition: df = df[df[column].astype(str).str.contains(condition["contains"], na=False)] else: # Simple equality filter df = df[df[column] == condition] # Apply limit if specified if limit and limit > 0: df = df.head(limit) return { "success": True, "filename": filename, "conditions": conditions, "filtered_data": df.to_dict('records'), "filtered_rows": len(df), "original_rows": len(pd.read_csv(filepath)) } except Exception as e: logger.error(f"Failed to filter data: {e}") raise - csv_mcp_server/server.py:181-185 (schema)Function signature defines the input schema (parameters) and output type for the MCP tool.
def filter_data( filename: str, conditions: Dict[str, Any], limit: Optional[int] = None ) -> Dict[str, Any]: