axom_mcp_transform

Instructions

Implementation Reference

• src/axom_mcp/handlers/transform.py:25-68 (handler)

  Main handler function 'handle_transform' that executes the axom_mcp_transform tool. It validates input using TransformInput schema, auto-detects input format, parses the input, applies transformation rules, and formats the output.

  async def handle_transform(arguments: Dict[str, Any]) -> str:
      """Handle axom_mcp_transform tool calls.
  
      Args:
          arguments: Tool arguments containing input data and format parameters
  
      Returns:
          JSON string with transformation result
      """
      # Validate input
      input_data = TransformInput(**arguments)
      input_str = input_data.input
      input_format = input_data.input_format
      output_format = input_data.output_format
      rules = input_data.rules or []
      template = input_data.template
  
      try:
          # Auto-detect input format if not specified
          if input_format is None:
              input_format = _detect_format(input_str)
  
          # Parse input to Python object
          parsed = _parse_input(input_str, input_format)
  
          # Apply transformation rules
          for rule in rules:
              parsed = _apply_rule(parsed, rule)
  
          # Convert to output format
          result = _format_output(parsed, output_format, template)
  
          return json.dumps(
              {
                  "success": True,
                  "input_format": input_format,
                  "output_format": output_format,
                  "result": result,
              }
          )
  
      except Exception as e:
          logger.error(f"Transform failed: {e}")
          return json.dumps({"error": str(e)})

• src/axom_mcp/schemas.py:288-312 (schema)

  Input schema definition 'TransformInput' for the axom_mcp_transform tool. Defines required fields (input, output_format) and optional fields (input_format, rules, template, chain) with validation constraints.

  class TransformInput(BaseModel):
      """Input schema for axom_mcp_transform tool."""
  
      model_config = {"extra": "forbid"}
  
      input: str = Field(
          ..., min_length=1, max_length=10_000_000, description="Input data to transform"
      )
      input_format: Optional[str] = Field(
          default=None,
          pattern="^(json|yaml|csv|markdown|code)$",
          description="Input format (auto-detected if not specified)",
      )
      output_format: str = Field(
          ..., pattern="^(json|yaml|csv|markdown|code)$", description="Output format"
      )
      rules: Optional[List[Dict[str, Any]]] = Field(
          default=None, description="Transformation rules"
      )
      template: Optional[str] = Field(
          default=None, max_length=10000, description="Template for transformation"
      )
      chain: Optional[List[Dict[str, Any]]] = Field(
          default=None, max_length=10, description="Chain operations after transformation"
      )

• src/axom_mcp/server.py:329-383 (registration)

  Tool registration for axom_mcp_transform in the TOOLS list. Defines the tool's name, description, input schema, and annotations. The schema specifies all supported formats (json, yaml, csv, markdown, code) and transformation options.

      Tool(
          name="axom_mcp_transform",
          description="""Transform data between formats and structures.
  
  Supported Formats:
  - json: JSON objects and arrays
  - yaml: YAML documents
  - csv: Comma-separated values
  - markdown: Markdown documents
  - code: Source code (with language detection)
  
  Transformation Rules:
  - field_mapping: Rename or restructure fields
  - filter: Include/exclude specific fields
  - sort: Sort arrays by field
  - aggregate: Group and aggregate data
  
  Template Support:
  Use Jinja2 templates for custom output formatting.
  
  Chain Support:
  Use chain parameter to continue processing transformed data.""",
          inputSchema={
              "type": "object",
              "properties": {
                  "input": {"type": "string", "description": "Input data to transform"},
                  "input_format": {
                      "type": "string",
                      "enum": ["json", "yaml", "csv", "markdown", "code"],
                      "description": "Input format (auto-detected if not specified)",
                  },
                  "output_format": {
                      "type": "string",
                      "enum": ["json", "yaml", "csv", "markdown", "code"],
                      "description": "Output format",
                  },
                  "rules": {
                      "type": "array",
                      "items": {"type": "object"},
                      "description": "Transformation rules",
                  },
                  "template": {
                      "type": "string",
                      "description": "Template for transformation",
                  },
                  "chain": {
                      "type": "array",
                      "items": {"type": "object"},
                      "description": "Chain operations after transformation",
                  },
              },
              "required": ["input", "output_format"],
          },
          annotations=TOOL_ANNOTATIONS["transform"],
      ),

• src/axom_mcp/server.py:498-499 (registration)

  Routing logic that calls handle_transform when the axom_mcp_transform tool is invoked. Maps the tool name to its handler function.

  elif name == "axom_mcp_transform":
      result = await handle_transform(arguments)

• src/axom_mcp/handlers/transform.py:71-115 (helper)

  Helper function _detect_format that auto-detects input format based on content patterns. Checks for JSON, YAML, CSV, markdown, and defaults to code format using regex heuristics.

  def _detect_format(input_str: str) -> str:
      """Auto-detect input format based on content."""
      stripped = input_str.strip()
  
      # Check for JSON
      if stripped.startswith("{") or stripped.startswith("["):
          try:
              json.loads(stripped)
              return "json"
          except json.JSONDecodeError:
              pass
  
      # Check for YAML (basic heuristics)
      yaml_indicators = [
          r"^\w+:\s*",  # key: value
          r"^-\s+",  # - list item (can be at start of line)
          r"^\s+-\s+",  # - list item (with indentation)
          r"^---\s*$",  # YAML document separator
      ]
      for pattern in yaml_indicators:
          if re.search(pattern, stripped, re.MULTILINE):
              return "yaml"
  
      # Check for CSV
      lines = stripped.split("\n")
      if len(lines) > 1:
          # Check if all lines have similar number of commas
          comma_counts = [line.count(",") for line in lines[:5]]
          if len(set(comma_counts)) == 1 and comma_counts[0] > 0:
              return "csv"
  
      # Check for Markdown
      md_indicators = [
          r"^#+\s+",  # Headers
          r"^\*\s+",  # Unordered lists
          r"^\d+\.\s+",  # Ordered lists
          r"\[.*\]\(.*\)",  # Links
          r"```",  # Code blocks
      ]
      for pattern in md_indicators:
          if re.search(pattern, stripped, re.MULTILINE):
              return "markdown"
  
      # Default to code
      return "code"