CSV MCP Server

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 data

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