MCP Data Wrangler

Instructions

Implementation Reference

• src/mcp_server_data_wrangler/tools/data_count.py:41-59 (handler)

  The asynchronous handler function for the 'data_count' tool that loads the input data, computes the count of non-null values per column, formats it as a dictionary, and returns it as JSON text content.

  async def handle_data_count(
      arguments: dict[str, Any],
  ) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
      data_count_input = DataCountInputSchema.from_args(arguments)
      count_df = data_count_input.df.count()
  
      # Convert the DataFrame to a dictionary format
      count_dict = {
          "description": "Number of non-null elements for each column",
          "counts": {col: int(val) for col, val in zip(count_df.columns, count_df.row(0))},
      }
  
      return [
          types.TextContent(
              type="text",
              text=json.dumps(count_dict),
          )
      ]

• src/mcp_server_data_wrangler/tools/data_count.py:10-39 (schema)

  Pydantic input schema class for the 'data_count' tool, including methods to define the JSON schema, load data from file path in arguments, and validate/parse input.

  class DataCountInputSchema(Data):
      model_config = ConfigDict(
          validate_assignment=True,
          frozen=True,
          extra="forbid",
          arbitrary_types_allowed=True,
      )
  
      @staticmethod
      def input_schema() -> dict:
          return {
              "type": "object",
              "properties": {
                  "input_data_file_path": {
                      "type": "string",
                      "description": "Path to the input data file",
                  },
              },
          }
  
      @staticmethod
      def from_schema(input_data_file_path: str) -> "DataCountInputSchema":
          data = Data.from_file(input_data_file_path)
          return DataCountInputSchema(df=data.df)
  
      @staticmethod
      def from_args(arguments: dict[str, Any]) -> "DataCountInputSchema":
          input_data_file_path = arguments["input_data_file_path"]
          return DataCountInputSchema.from_schema(input_data_file_path=input_data_file_path)

• src/mcp_server_data_wrangler/tools/tools.py:87-91 (registration)

  Tool object registration for 'data_count' in MCPServerDataWrangler.tools() method, defining the tool's name, description, and input schema for MCP server listing.

  types.Tool(
      name=MCPServerDataWrangler.data_count.value[0],
      description=MCPServerDataWrangler.data_count.value[1],
      inputSchema=DataCountInputSchema.input_schema(),
  ),

• src/mcp_server_data_wrangler/tools/tools.py:156-156 (registration)

  Dispatch mapping in MCPServerDataWrangler.tool_to_handler() dictionary that associates the 'data_count' tool name with its handler function.

  MCPServerDataWrangler.data_count.value[0]: handle_data_count,