Blind-Auditor-MCP

Instructions

Implementation Reference

• src/main.py:225-326 (handler)

  The primary handler function for the 'update_rules' tool, decorated with @mcp.tool() for registration. Implements logic for listing, adding, removing, and updating audit rules by delegating to RulesLoader instance.
  @mcp.tool() def update_rules( action: str, rule_id: str = "", severity: str = "", description: str = "", weight: int = 0 ) -> str: """ Update audit rules configuration. Args: action: Operation to perform - "add", "remove", "update", or "list" rule_id: Rule identifier (required for add/remove/update) severity: Rule severity level - "CRITICAL", "WARNING", or "PREFERENCE" (for add/update) description: Rule description (for add/update) weight: Point deduction weight 0-100 (for add/update) Returns: Status message with operation result Examples: # List all rules update_rules(action="list") # Add a new rule update_rules( action="add", rule_id="SEC-001", severity="CRITICAL", description="No hardcoded API keys", weight=50 ) # Remove a rule update_rules(action="remove", rule_id="SEC-001") # Update a rule update_rules( action="update", rule_id="SEC-001", description="Updated description" ) """ print(f"DEBUG: update_rules called with action={action}, rule_id={rule_id}", file=sys.stderr) # Reload rules to get latest state rules_loader.load() # Handle list action if action == "list": return rules_loader.list_rules() # Validate rule_id for non-list actions if not rule_id: return "❌ Error: rule_id is required for add/remove/update actions." # Handle add action if action == "add": if not severity or not description: return "❌ Error: severity and description are required for adding a rule." result = rules_loader.add_rule(rule_id, severity, description, weight) if result["status"] == "success": return f"✅ {result['message']}\n\n{rules_loader.list_rules()}" else: return f"❌ {result['message']}" # Handle remove action elif action == "remove": result = rules_loader.remove_rule(rule_id) if result["status"] == "success": return f"✅ {result['message']}\n\n{rules_loader.list_rules()}" else: return f"❌ {result['message']}" # Handle update action elif action == "update": # Only pass non-empty values update_kwargs = {} if severity: update_kwargs["severity"] = severity if description: update_kwargs["description"] = description if weight > 0: update_kwargs["weight"] = weight if not update_kwargs: return "❌ Error: At least one field (severity, description, or weight) must be provided for update." result = rules_loader.update_rule(rule_id, **update_kwargs) if result["status"] == "success": return f"✅ {result['message']}\n\n{rules_loader.list_rules()}" else: return f"❌ {result['message']}" else: return f"❌ Error: Invalid action '{action}'. Must be one of: add, remove, update, list"
• src/rules.py:10-226 (helper)

  The RulesLoader class containing all helper methods (load, add_rule, remove_rule, update_rule, list_rules) called by the update_rules handler to manage the rules.json file.
  class RulesLoader: """Loads and manages audit rules from rules.json.""" def __init__(self, rules_path: str = "rules.json"): self.rules_path = Path(rules_path) self.rules_data: Dict[str, Any] = {} def load(self) -> Dict[str, Any]: """Load rules from the JSON file.""" if not self.rules_path.exists(): # Return default empty structure return { "project_name": "Unknown", "strict_mode": True, "max_retries": 3, "rules": [] } with open(self.rules_path, 'r', encoding='utf-8') as f: self.rules_data = json.load(f) return self.rules_data def save(self) -> None: """Save current rules data back to JSON file.""" with open(self.rules_path, 'w', encoding='utf-8') as f: json.dump(self.rules_data, f, indent=2, ensure_ascii=False) def get_rules(self) -> List[Dict[str, Any]]: """Get the list of rules.""" return self.rules_data.get("rules", []) def get_max_retries(self) -> int: """Get the maximum retry count.""" return self.rules_data.get("max_retries", 3) def add_rule(self, rule_id: str, severity: str, description: str, weight: int) -> Dict[str, str]: """ Add a new rule to the rules list. Args: rule_id: Unique identifier for the rule (e.g., "SEC-001") severity: Rule severity level ("CRITICAL", "WARNING", "PREFERENCE") description: Description of the rule weight: Point deduction weight for violations Returns: Dict with status and message """ # Validate severity valid_severities = ["CRITICAL", "WARNING", "PREFERENCE"] if severity not in valid_severities: return { "status": "error", "message": f"Invalid severity '{severity}'. Must be one of: {', '.join(valid_severities)}" } # Check for duplicate rule_id existing_rules = self.get_rules() if any(rule.get("id") == rule_id for rule in existing_rules): return { "status": "error", "message": f"Rule with ID '{rule_id}' already exists. Use update_rule to modify it." } # Validate weight if not isinstance(weight, int) or weight < 0 or weight > 100: return { "status": "error", "message": f"Weight must be an integer between 0 and 100, got {weight}" } # Add the new rule new_rule = { "id": rule_id, "severity": severity, "description": description, "weight": weight } self.rules_data["rules"].append(new_rule) self.save() return { "status": "success", "message": f"Rule '{rule_id}' added successfully." } def remove_rule(self, rule_id: str) -> Dict[str, str]: """ Remove a rule by its ID. Args: rule_id: The ID of the rule to remove Returns: Dict with status and message """ existing_rules = self.get_rules() original_count = len(existing_rules) # Filter out the rule with matching ID self.rules_data["rules"] = [ rule for rule in existing_rules if rule.get("id") != rule_id ] if len(self.rules_data["rules"]) == original_count: return { "status": "error", "message": f"Rule with ID '{rule_id}' not found." } self.save() return { "status": "success", "message": f"Rule '{rule_id}' removed successfully." } def update_rule( self, rule_id: str, severity: Optional[str] = None, description: Optional[str] = None, weight: Optional[int] = None ) -> Dict[str, str]: """ Update an existing rule. Args: rule_id: The ID of the rule to update severity: New severity level (optional) description: New description (optional) weight: New weight (optional) Returns: Dict with status and message """ existing_rules = self.get_rules() rule_found = False for rule in existing_rules: if rule.get("id") == rule_id: rule_found = True # Validate and update severity if severity is not None: valid_severities = ["CRITICAL", "WARNING", "PREFERENCE"] if severity not in valid_severities: return { "status": "error", "message": f"Invalid severity '{severity}'. Must be one of: {', '.join(valid_severities)}" } rule["severity"] = severity # Update description if description is not None: rule["description"] = description # Validate and update weight if weight is not None: if not isinstance(weight, int) or weight < 0 or weight > 100: return { "status": "error", "message": f"Weight must be an integer between 0 and 100, got {weight}" } rule["weight"] = weight break if not rule_found: return { "status": "error", "message": f"Rule with ID '{rule_id}' not found." } self.save() return { "status": "success", "message": f"Rule '{rule_id}' updated successfully." } def list_rules(self) -> str: """ Get a formatted string of all current rules. Returns: Formatted string listing all rules """ rules = self.get_rules() if not rules: return "No rules configured." lines = ["Current Audit Rules:", ""] for i, rule in enumerate(rules, 1): lines.append(f"{i}. [{rule.get('severity', 'UNKNOWN')}] {rule.get('id', 'NO_ID')}") lines.append(f" Description: {rule.get('description', 'No description')}") lines.append(f" Weight: {rule.get('weight', 0)} points") lines.append("") return "\n".join(lines) def format_rules_for_prompt(self) -> str: """Format rules as a readable string for prompt injection.""" rules = self.get_rules() if not rules: return "No rules configured." formatted = [] for rule in rules: severity = rule.get("severity", "UNKNOWN") description = rule.get("description", "") rule_id = rule.get("id", "") formatted.append(f"[{severity}] {rule_id}: {description}") return "\n".join(formatted)
• src/main.py:14-22 (registration)

  MCP server initialization and run, where tools including update_rules are registered via decorators and exposed via mcp.run().
  from mcp.server.fastmcp import FastMCP print("DEBUG: Importing state and rules", file=sys.stderr) from state import SessionState from rules import RulesLoader # Initialize the MCP server print("DEBUG: Creating FastMCP instance", file=sys.stderr) mcp = FastMCP("Blind Auditor")