adv_scan_folder
Scan directories for security vulnerabilities, identify exploit examples, and analyze with Semgrep or LLM. Results are saved in JSON or Markdown formats for detailed review.
Instructions
Scan a directory for security vulnerabilities. Results are saved in the target directory.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| include_exploits | No | Whether to include exploit examples | |
| output_format | No | Output format for results (json or markdown) | json |
| path | No | Path to the directory to scan (must be a directory, not a file) | . |
| recursive | No | Whether to scan subdirectories | |
| severity_threshold | No | Minimum severity threshold | medium |
| use_llm | No | Whether to include LLM analysis prompts (for use with your client's LLM) | |
| use_semgrep | No | Whether to include Semgrep analysis | |
| use_validation | No | Whether to use LLM validation to filter false positives |
Implementation Reference
- Primary MCP handler for 'adv_scan_folder' tool. Validates input, orchestrates scan via ScanApplicationService.scan_directory, persists results, formats output as JSON, and returns MCP TextContent response.async def _handle_scan_folder( self, name: str, arguments: dict ) -> list[types.TextContent]: """Handle folder scanning requests.""" try: # Log MCP tool invocation at INFO level for visibility logger.info(f"MCP Tool Invoked: {name}") logger.info(f"Parameters: {arguments}") # Comprehensive input validation validated_args = self._input_validator.validate_mcp_arguments( arguments, tool_name="adv_scan_folder" ) path = validated_args.get("path", ".") use_semgrep = validated_args.get("use_semgrep", True) use_llm = validated_args.get("use_llm", False) use_validation = validated_args.get("use_validation", False) severity_threshold = validated_args.get("severity_threshold", "medium") timeout_seconds = validated_args.get("timeout_seconds") recursive = validated_args.get("recursive", True) output_format = validated_args.get("output_format", "json") # Use the same scan service as CLI for consistency and proper orchestration result = await self._scan_service.scan_directory( directory_path=path, requester="cli", enable_semgrep=use_semgrep, enable_llm=use_llm, enable_validation=use_validation, severity_threshold=severity_threshold, timeout_seconds=timeout_seconds, recursive=recursive, ) # Persist scan result automatically try: output_format_enum = OutputFormat.from_string(output_format) file_path = await self._persistence_service.persist_scan_result( result, output_format_enum ) logger.info(f"Scan result persisted to {file_path}") except Exception as e: logger.warning(f"Failed to persist scan result: {e}") # Don't fail the scan if persistence fails formatted_result = self._format_scan_result(result) # Add persistence info to the response formatted_result["persistence"] = { "output_format": output_format, "file_path": file_path if "file_path" in locals() else None, "persisted": "file_path" in locals(), } # Log successful completion with key metrics threat_count = ( len(result.threat_matches) if hasattr(result, "threat_matches") else 0 ) scan_duration = ( getattr(result.metadata, "scan_duration_seconds", 0) if hasattr(result, "metadata") else 0 ) files_scanned = ( getattr(result.metadata, "total_files_scanned", 0) if hasattr(result, "metadata") else 0 ) logger.info( f"[+] MCP Tool Completed: {name} | Threats: {threat_count} | Files: {files_scanned} | Duration: {scan_duration:.2f}s" ) return [ types.TextContent( type="text", text=json.dumps(formatted_result, indent=2, default=str), ) ] except (ValidationError, SecurityError, ConfigurationError) as e: logger.error(f"Directory scan failed: {e}") raise CleanAdversaryToolError(f"Scan failed: {str(e)}") except Exception as e: logger.error(f"Unexpected error in directory scan: {e}") logger.error(traceback.format_exc()) raise CleanAdversaryToolError(f"Internal error: {str(e)}")
- MCP Tool schema definition including inputSchema with properties, defaults, and descriptions for adv_scan_folder.Tool( name="adv_scan_folder", description="Scan a directory for security vulnerabilities using Clean Architecture. Automatically uses session-aware project analysis when LLM is configured.", inputSchema={ "type": "object", "properties": { "path": { "type": "string", "description": "Path to the directory to scan", "default": ".", }, "use_semgrep": { "type": "boolean", "description": "Enable Semgrep analysis", "default": True, }, "use_llm": { "type": "boolean", "description": "Enable LLM analysis", "default": False, }, "use_validation": { "type": "boolean", "description": "Enable LLM validation", "default": False, }, "severity_threshold": { "type": "string", "description": "Minimum severity level", "default": "medium", }, "timeout_seconds": { "type": "integer", "description": "Scan timeout in seconds", }, "recursive": { "type": "boolean", "description": "Scan subdirectories", "default": True, }, "output_format": { "type": "string", "description": "Output format for persisted scan results", "enum": ["json", "md", "markdown", "csv"], "default": "json", }, }, }, ), Tool(
- src/adversary_mcp_server/application/mcp_server.py:117-137 (registration)Registration of the call_tool dispatcher that routes 'adv_scan_folder' calls to the specific handler.@self.server.call_tool() async def tool_dispatcher( name: str, arguments: dict ) -> list[types.TextContent]: """Dispatch MCP tool calls to the appropriate handler.""" if name == "adv_scan_file": return await self._handle_scan_file(name, arguments) elif name == "adv_scan_folder": return await self._handle_scan_folder(name, arguments) elif name == "adv_scan_code": return await self._handle_scan_code(name, arguments) elif name == "adv_get_status": return await self._handle_get_status(name, arguments) elif name == "adv_get_version": return await self._handle_get_version(name, arguments) elif name == "adv_mark_false_positive": return await self._handle_mark_false_positive(name, arguments) elif name == "adv_unmark_false_positive": return await self._handle_unmark_false_positive(name, arguments) else: raise ValueError(f"Unknown tool: {name}")
- Core scanning logic invoked by the tool handler: creates domain ScanRequest for directory scan and executes via domain ScanOrchestrator.async def scan_directory( self, directory_path: str, *, requester: str = "application", enable_semgrep: bool = True, enable_llm: bool = False, enable_validation: bool = False, severity_threshold: str | None = None, timeout_seconds: int | None = None, recursive: bool = True, ) -> ScanResult: """ Scan a directory for security vulnerabilities. Args: directory_path: Path to the directory to scan requester: Who requested the scan enable_semgrep: Whether to enable Semgrep scanning enable_llm: Whether to enable LLM analysis enable_validation: Whether to enable LLM validation severity_threshold: Minimum severity level to include timeout_seconds: Scan timeout in seconds recursive: Whether to scan subdirectories Returns: ScanResult containing found threats and metadata """ # Create domain objects dir_path_obj = FilePath.from_string(directory_path) metadata = ScanMetadata( scan_id=str(uuid.uuid4()), scan_type="directory", timestamp=datetime.now(UTC), requester=requester, timeout_seconds=timeout_seconds or 600, # Default 10 minutes for directories enable_semgrep=enable_semgrep, enable_llm=enable_llm, enable_validation=enable_validation, ) context = ScanContext(target_path=dir_path_obj, metadata=metadata) severity_level = ( SeverityLevel.from_string(severity_threshold) if severity_threshold else None ) request = ScanRequest( context=context, enable_semgrep=enable_semgrep, enable_llm=enable_llm, enable_validation=enable_validation, severity_threshold=severity_level, ) # Validate and execute self._validation_service.validate_scan_request(request) self._validation_service.enforce_security_constraints(context) result = await self._scan_orchestrator.execute_scan(request) return result
- Input validation specific to adv_scan_folder: treats 'path' as directory path and applies security checks.if tool_name in ("adv_scan_folder", "adv_diff_scan"): # These tools expect directory paths validated[key] = str( InputValidator.validate_directory_path(str(value)) )