CSV MCP Server

Instructions

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]: