CSV Editor

Instructions

Implementation Reference

• src/csv_editor/tools/analytics.py:383-460 (handler)

  Core implementation of get_value_counts tool: retrieves DataFrame from session, computes value counts with options for normalization, sorting, and top_n, converts to dictionary format, adds statistics, and handles errors.

  async def get_value_counts(
      session_id: str,
      column: str,
      normalize: bool = False,
      sort: bool = True,
      ascending: bool = False,
      top_n: Optional[int] = None,
      ctx: Context = None
  ) -> Dict[str, Any]:
      """
      Get value counts for a column.
      
      Args:
          session_id: Session identifier
          column: Column name to count values
          normalize: Return proportions instead of counts
          sort: Sort by frequency
          ascending: Sort order
          top_n: Return only top N values
          ctx: FastMCP context
          
      Returns:
          Dict with value counts
      """
      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
          
          if column not in df.columns:
              return {"success": False, "error": f"Column '{column}' not found"}
          
          # Get value counts
          value_counts = df[column].value_counts(
              normalize=normalize,
              sort=sort,
              ascending=ascending,
              dropna=False
          )
          
          # Apply top_n if specified
          if top_n:
              value_counts = value_counts.head(top_n)
          
          # Convert to dict
          counts_dict = {}
          for value, count in value_counts.items():
              key = str(value) if not pd.isna(value) else "NaN"
              counts_dict[key] = float(count) if normalize else int(count)
          
          # Calculate additional statistics
          unique_count = df[column].nunique(dropna=False)
          null_count = df[column].isna().sum()
          
          session.record_operation(OperationType.ANALYZE, {
              "type": "value_counts",
              "column": column,
              "normalize": normalize,
              "top_n": top_n
          })
          
          return {
              "success": True,
              "column": column,
              "value_counts": counts_dict,
              "unique_values": int(unique_count),
              "null_count": int(null_count),
              "total_count": len(df),
              "normalized": normalize
          }
          
      except Exception as e:
          logger.error(f"Error getting value counts: {str(e)}")
          return {"success": False, "error": str(e)}

• src/csv_editor/server.py:358-368 (registration)

  Registers the get_value_counts tool using @mcp.tool decorator, providing the tool interface and delegating execution to the imported implementation.

  async def get_value_counts(
      session_id: str,
      column: str,
      normalize: bool = False,
      sort: bool = True,
      ascending: bool = False,
      top_n: Optional[int] = None,
      ctx: Context = None
  ) -> Dict[str, Any]:
      """Get value counts for a column."""
      return await _get_value_counts(session_id, column, normalize, sort, ascending, top_n, ctx)