MCP Data Wrangler

Instructions

Implementation Reference

• src/mcp_server_data_wrangler/tools/data_var.py:62-80 (handler)

  The async handler function for the 'data_var' tool. It parses input arguments into DataVarInputSchema, computes variance for each column of the dataframe using pandas .var(ddof=...), formats results as a dictionary, and returns it as JSON text content.

  async def handle_data_var(
      arguments: dict[str, Any],
  ) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
      data_var_input = DataVarInputSchema.from_args(arguments)
      var_df = data_var_input.df.var(ddof=data_var_input.ddof)
  
      # Convert the DataFrame to a dictionary format
      var_dict = {
          "description": f"Variance values for each column with ddof={data_var_input.ddof}",
          "var_values": {col: str(val) if val is not None else None for col, val in zip(var_df.columns, var_df.row(0))},
      }
  
      return [
          types.TextContent(
              type="text",
              text=json.dumps(var_dict),
          )
      ]

• src/mcp_server_data_wrangler/tools/data_var.py:13-60 (schema)

  Pydantic schema class DataVarInputSchema extending Data, including input_schema() for MCP tool schema, and factory methods from_schema() and from_args() to load dataframe from file and validate inputs.

  class DataVarInputSchema(Data):
      model_config = ConfigDict(
          validate_assignment=True,
          frozen=True,
          extra="forbid",
          arbitrary_types_allowed=True,
      )
  
      ddof: int = Field(
          default=1, description="Delta Degrees of Freedom: the divisor used in the calculation is N - ddof", ge=0
      )
  
      @staticmethod
      def input_schema() -> dict:
          return {
              "type": "object",
              "properties": {
                  "input_data_file_path": {
                      "type": "string",
                      "description": "Path to the input data file",
                  },
                  "ddof": {
                      "type": "integer",
                      "description": "Delta Degrees of Freedom: the divisor used in the calculation is N - ddof",
                      "minimum": 0,
                      "default": 1,
                  },
              },
              "required": ["input_data_file_path"],
          }
  
      @staticmethod
      def from_schema(input_data_file_path: str, ddof: int = 1) -> "DataVarInputSchema":
          data = Data.from_file(input_data_file_path)
          return DataVarInputSchema(
              df=data.df,
              ddof=ddof,
          )
  
      @staticmethod
      def from_args(arguments: dict[str, Any]) -> "DataVarInputSchema":
          input_data_file_path = arguments["input_data_file_path"]
          ddof = arguments.get("ddof", 1)
          return DataVarInputSchema.from_schema(
              input_data_file_path=input_data_file_path,
              ddof=ddof,
          )

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

  MCP Tool object registration for 'data_var', using enum value for name and description, and DataVarInputSchema.input_schema() for input validation.

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

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

  Handler function mapping for 'data_var' tool to handle_data_var in the tool_to_handler() dictionary.

  MCPServerDataWrangler.data_var.value[0]: handle_data_var,