Axom MCP Server

Instructions

Implementation Reference

• src/axom_mcp/handlers/exec.py:57-82 (handler)

  Main handler function handle_exec that routes axom_mcp_exec tool calls to appropriate operation handlers (read, write, shell) with input validation using ExecInput schema

  async def handle_exec(arguments: Dict[str, Any]) -> str:
      """Handle axom_mcp_exec tool calls.
      
      Args:
          arguments: Tool arguments containing operation and parameters
          
      Returns:
          JSON string with operation result
      """
      # Validate input
      input_data = ExecInput(**arguments)
      operation = input_data.operation
      target = input_data.target
      
      try:
          if operation == "read":
              return await _handle_read(target)
          elif operation == "write":
              return await _handle_write(target, input_data.data)
          elif operation == "shell":
              return await _handle_shell(target)
          else:
              return json.dumps({"error": f"Unknown operation: {operation}"})
      except Exception as e:
          logger.error(f"Exec operation failed: {e}")
          return json.dumps({"error": str(e)})

• src/axom_mcp/handlers/exec.py:85-195 (handler)

  Operation-specific handler functions: _handle_read (lines 85-116), _handle_write (lines 119-150), and _handle_shell (lines 153-195) that implement file operations and shell command execution with security validations

  async def _handle_read(target: str) -> str:
      """Read file contents."""
      try:
          path = _validate_path(target)
          
          if not path.exists():
              return json.dumps({"error": f"File not found: {target}"})
          
          if not path.is_file():
              return json.dumps({"error": f"Not a file: {target}"})
          
          # Check file size
          if path.stat().st_size > MAX_FILE_SIZE:
              return json.dumps({
                  "error": f"File too large: {target} (max {MAX_FILE_SIZE} bytes)"
              })
          
          # Read file content
          content = path.read_text(encoding="utf-8", errors="replace")
          
          return json.dumps({
              "success": True,
              "operation": "read",
              "target": str(path),
              "content": content,
              "size": len(content),
          })
      except ValueError as e:
          return json.dumps({"error": str(e)})
      except Exception as e:
          logger.error(f"Failed to read file: {e}")
          return json.dumps({"error": str(e)})
  
  
  async def _handle_write(target: str, data: Optional[str]) -> str:
      """Write data to file."""
      # Check if write operations are allowed (AXOM_READ_ONLY defaults to False)
      if _env_flag_enabled("AXOM_READ_ONLY", default=False):
          return json.dumps({
              "error": "Write operations disabled. AXOM_READ_ONLY is enabled."
          })
      
      if data is None:
          return json.dumps({"error": "data is required for write operation"})
      
      try:
          path = _validate_path(target)
          
          # Create parent directories if needed
          path.parent.mkdir(parents=True, exist_ok=True)
          
          # Write file
          path.write_text(data, encoding="utf-8")
          
          return json.dumps({
              "success": True,
              "operation": "write",
              "target": str(path),
              "size": len(data),
              "message": f"Successfully wrote {len(data)} bytes to {path}",
          })
      except ValueError as e:
          return json.dumps({"error": str(e)})
      except Exception as e:
          logger.error(f"Failed to write file: {e}")
          return json.dumps({"error": str(e)})
  
  
  async def _handle_shell(command: str) -> str:
      """Execute shell command."""
      # Check if shell operations are allowed (AXOM_READ_ONLY defaults to False)
      if _env_flag_enabled("AXOM_READ_ONLY", default=False):
          return json.dumps({
              "error": "Shell operations disabled. AXOM_READ_ONLY is enabled."
          })
      
      try:
          # Execute command with timeout
          process = await asyncio.create_subprocess_shell(
              command,
              stdout=asyncio.subprocess.PIPE,
              stderr=asyncio.subprocess.PIPE,
          )
          
          try:
              stdout, stderr = await asyncio.wait_for(
                  process.communicate(),
                  timeout=60.0  # 60 second timeout
              )
          except asyncio.TimeoutError:
              process.kill()
              return json.dumps({
                  "error": "Command timed out after 60 seconds",
                  "command": command,
              })
          
          # Decode output
          stdout_str = stdout.decode("utf-8", errors="replace") if stdout else ""
          stderr_str = stderr.decode("utf-8", errors="replace") if stderr else ""
          
          return json.dumps({
              "success": process.returncode == 0,
              "operation": "shell",
              "command": command,
              "exit_code": process.returncode,
              "stdout": stdout_str,
              "stderr": stderr_str,
          })
      except Exception as e:
          logger.error(f"Failed to execute shell command: {e}")
          return json.dumps({"error": str(e)})

• src/axom_mcp/schemas.py:209-232 (schema)

  ExecInput Pydantic schema defining the input validation for axom_mcp_exec tool with operation (read/write/shell), target (file path or command), data (for writes), and optional chain parameter

  class ExecInput(BaseModel):
      """Input schema for axom_mcp_exec tool."""
  
      model_config = {"extra": "forbid"}
  
      operation: str = Field(
          ...,
          pattern="^(read|write|shell)$",
          description="Operation type: read, write, or shell",
      )
      target: str = Field(
          ...,
          min_length=1,
          max_length=4096,
          description="File path for read/write, or command for shell",
      )
      data: Optional[str] = Field(
          default=None,
          max_length=10_000_000,
          description="Data to write (for write operation)",
      )
      chain: Optional[List[Dict[str, Any]]] = Field(
          default=None, max_length=10, description="Chain of subsequent operations"
      )

• src/axom_mcp/server.py:171-221 (registration)

  Tool registration in TOOLS list defining axom_mcp_exec with name, description, inputSchema, and annotations for MCP protocol

          name="axom_mcp_exec",
          description="""Execute file operations and shell commands with chain-reaction support.
  
  Operations:
  - read: Read file contents from allowed directories
  - write: Write data to files (unless AXOM_READ_ONLY=true)
  - shell: Execute shell commands (unless AXOM_READ_ONLY=true)
  
  Chain Reactions:
  Chain multiple operations together using the chain parameter. Each step can reference
  the previous result using ${_result} variable substitution.
  
  Example:
  {
    "operation": "read",
    "target": "/file.txt",
    "chain": [
      {
        "tool": "axom_mcp_transform",
        "args": {"input": "${_result.content}", "output_format": "json"}
      }
    ]
  }
  
  Security:
  - File operations restricted to allowed directories (cwd, ~/)
  - Shell/write operations enabled by default (set AXOM_READ_ONLY=true to disable)
  - Input size limits: 10MB max for files""",
          inputSchema={
              "type": "object",
              "properties": {
                  "operation": {
                      "type": "string",
                      "enum": ["read", "write", "shell"],
                      "description": "Operation type",
                  },
                  "target": {"type": "string", "description": "File path or command"},
                  "data": {
                      "type": "string",
                      "description": "Data to write (for write operation)",
                  },
                  "chain": {
                      "type": "array",
                      "items": {"type": "object"},
                      "description": "Chain of subsequent operations",
                  },
              },
              "required": ["operation", "target"],
          },
          annotations=TOOL_ANNOTATIONS["exec"],
      ),

• src/axom_mcp/server.py:486-506 (registration)

  call_tool function that routes incoming tool requests, specifically line 492-493 where axom_mcp_exec calls are forwarded to handle_exec handler

  @server.call_tool()
  async def call_tool(name: str, arguments: Dict[str, Any]) -> List[TextContent]:
      """Handle tool calls."""
      try:
          if name == "axom_mcp_memory":
              result = await handle_memory(arguments)
          elif name == "axom_mcp_exec":
              result = await handle_exec(arguments)
          elif name == "axom_mcp_analyze":
              result = await handle_analyze(arguments)
          elif name == "axom_mcp_discover":
              result = await handle_discover(arguments)
          elif name == "axom_mcp_transform":
              result = await handle_transform(arguments)
          else:
              return [TextContent(type="text", text=f"Unknown tool: {name}")]
  
          return [TextContent(type="text", text=result)]
      except Exception as e:
          logger.error(f"Tool call failed: {name} - {e}")
          return [TextContent(type="text", text=f"Error: {str(e)}")]