data_max_horizontal
Calculate maximum values across columns for each row in your dataset using this preprocessing tool. Ideal for data aggregation and transformation tasks, ensuring efficient row-wise analysis.
Instructions
Maximum values across columns for each row
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| input_data_file_path | No | Path to the input data file |
Implementation Reference
- The main handler function that executes the 'data_max_horizontal' tool logic: loads data, computes max across columns per row, returns JSON.
async def handle_data_max_horizontal( arguments: dict[str, Any], ) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]: data_max_input = DataMaxInputSchema.from_args(arguments) try: max_horizontal_df = data_max_input.df.max_horizontal() # Convert the DataFrame to a dictionary format max_horizontal_dict = { "description": "Maximum values across columns for each row", "max_values": {str(i): str(val) if val is not None else None for i, val in enumerate(max_horizontal_df)}, } return [ types.TextContent( type="text", text=json.dumps(max_horizontal_dict), ) ] except Exception as e: logger.error(f"Error calculating max: {e}") return [ types.TextContent( type="text", text=json.dumps( { "error": "Failed to calculate max values.", "message": str(e), } ), ) ] - Pydantic schema for input validation: requires input_data_file_path, loads Dataframe.
class DataMaxInputSchema(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) -> "DataMaxInputSchema": data = Data.from_file(input_data_file_path) return DataMaxInputSchema(df=data.df) @staticmethod def from_args(arguments: dict[str, Any]) -> "DataMaxInputSchema": input_data_file_path = arguments["input_data_file_path"] return DataMaxInputSchema.from_schema(input_data_file_path=input_data_file_path) - src/mcp_server_data_wrangler/tools/tools.py:97-101 (registration)Tool registration in MCPServerDataWrangler.tools(): defines name, description, and inputSchema.
types.Tool( name=MCPServerDataWrangler.data_max_horizontal.value[0], description=MCPServerDataWrangler.data_max_horizontal.value[1], inputSchema=DataMaxInputSchema.input_schema(), ), - src/mcp_server_data_wrangler/tools/tools.py:157-159 (registration)Handler mapping in MCPServerDataWrangler.tool_to_handler(): maps tool name to handle_data_max_horizontal.
MCPServerDataWrangler.data_max.value[0]: handle_data_max, MCPServerDataWrangler.data_max_horizontal.value[0]: handle_data_max_horizontal, MCPServerDataWrangler.data_min.value[0]: handle_data_min, - Enum definition in MCPServerDataWrangler providing the tool name and description.
data_max_horizontal = ("data_max_horizontal", "Maximum values across columns for each row")