mcp-server-tree-sitter

"""Code analysis tools using tree-sitter.""" import os from collections import Counter, defaultdict from typing import Any, Dict, List, Optional, Set, Tuple from ..exceptions import SecurityError from ..language.query_templates import get_query_template from ..utils.context import MCPContext from ..utils.file_io import get_comment_prefix, read_text_file from ..utils.security import validate_file_access from ..utils.tree_sitter_helpers import ( ensure_language, ensure_node, get_node_text, parse_with_cached_tree, ) def extract_symbols( project: Any, file_path: str, language_registry: Any, symbol_types: Optional[List[str]] = None, exclude_class_methods: bool = False, ) -> Dict[str, List[Dict[str, Any]]]: """ Extract symbols (functions, classes, etc) from a file. Args: project: Project object file_path: Path to the file relative to project root language_registry: Language registry object symbol_types: Types of symbols to extract (functions, classes, imports, etc.) exclude_class_methods: Whether to exclude methods from function count Returns: Dictionary of symbols by type """ abs_path = project.get_file_path(file_path) try: validate_file_access(abs_path, project.root_path) except SecurityError as e: raise SecurityError(f"Access denied: {e}") from e language = language_registry.language_for_file(file_path) if not language: raise ValueError(f"Could not detect language for {file_path}") # Default symbol types if not specified if symbol_types is None: # Language-specific defaults based on their structural elements if language == "rust": symbol_types = ["functions", "structs", "imports"] elif language == "go": symbol_types = ["functions", "structs", "imports"] elif language == "c": symbol_types = ["functions", "structs", "imports"] elif language == "cpp": symbol_types = ["functions", "classes", "structs", "imports"] elif language == "typescript": symbol_types = ["functions", "classes", "interfaces", "imports"] elif language == "swift": symbol_types = ["functions", "classes", "structs", "imports"] elif language == "java": symbol_types = ["functions", "classes", "interfaces", "imports"] elif language == "kotlin": symbol_types = ["functions", "classes", "interfaces", "imports"] elif language == "julia": symbol_types = ["functions", "modules", "structs", "imports"] elif language == "apl": symbol_types = ["functions", "namespaces", "variables", "imports"] else: symbol_types = ["functions", "classes", "imports"] # Get query templates for each symbol type queries = {} for symbol_type in symbol_types: template = get_query_template(language, symbol_type) if template: queries[symbol_type] = template if not queries: raise ValueError(f"No query templates available for {language} and {symbol_types}") # Parse file and extract symbols try: # Get language object language_obj = language_registry.get_language(language) safe_lang = ensure_language(language_obj) # Parse with cached tree tree, source_bytes = parse_with_cached_tree(abs_path, language, safe_lang) # Execute queries symbols: Dict[str, List[Dict[str, Any]]] = {} # Track class ranges to identify methods class_ranges = [] # Process classes first if we need to filter out class methods if exclude_class_methods and "classes" in queries: if "classes" not in symbols: symbols["classes"] = [] class_query = safe_lang.query(queries["classes"]) class_matches = class_query.captures(tree.root_node) # Process class locations to identify their boundaries process_symbol_matches(class_matches, "classes", symbols, source_bytes, tree) # Extract class body ranges to check if functions are inside classes # Use a more generous range to ensure we catch all methods for class_symbol in symbols["classes"]: start_row = class_symbol["location"]["start"]["row"] # For class end, we need to estimate where the class body might end # by scanning the file for likely class boundaries source_lines = source_bytes.decode("utf-8", errors="replace").splitlines() # Find a reasonable estimate for where the class ends end_row = min(start_row + 30, len(source_lines) - 1) class_ranges.append((start_row, end_row)) # Now process all symbol types for symbol_type, query_string in queries.items(): # Skip classes if we already processed them if symbol_type == "classes" and exclude_class_methods and class_ranges: continue if symbol_type not in symbols: symbols[symbol_type] = [] query = safe_lang.query(query_string) matches = query.captures(tree.root_node) process_symbol_matches( matches, symbol_type, symbols, source_bytes, tree, (class_ranges if exclude_class_methods and symbol_type == "functions" else None), ) # Handle aliased imports specifically for Python if symbol_type == "imports" and language == "python": # Look for aliased imports that might have been missed aliased_query_string = """ (import_from_statement module_name: (dotted_name) @import.from name: (aliased_import)) @import """ aliased_query = safe_lang.query(aliased_query_string) aliased_matches = aliased_query.captures(tree.root_node) for match in aliased_matches: node = None capture_name = "" # Handle different return types if isinstance(match, tuple) and len(match) == 2: node, capture_name = match elif hasattr(match, "node") and hasattr(match, "capture_name"): node, capture_name = match.node, match.capture_name elif isinstance(match, dict) and "node" in match and "capture" in match: node, capture_name = match["node"], match["capture"] else: continue if capture_name == "import.from": module_name = get_node_text(node, source_bytes) # Add this module to the import list symbols["imports"].append( { "name": module_name, "type": "imports", "location": { "start": { "row": node.start_point[0], "column": node.start_point[1], }, "end": { "row": node.end_point[0], "column": node.end_point[1], }, }, } ) # Additionally, run a query to get all aliased imports directly alias_query_string = "(aliased_import) @alias" alias_query = safe_lang.query(alias_query_string) alias_matches = alias_query.captures(tree.root_node) for match in alias_matches: node = None capture_name = "" # Handle different return types if isinstance(match, tuple) and len(match) == 2: node, capture_name = match elif hasattr(match, "node") and hasattr(match, "capture_name"): node, capture_name = match.node, match.capture_name elif isinstance(match, dict) and "node" in match and "capture" in match: node, capture_name = match["node"], match["capture"] else: continue if capture_name == "alias": alias_text = get_node_text(node, source_bytes) module_name = "" # Try to get the module name from parent if node.parent and node.parent.parent: for child in node.parent.parent.children: if hasattr(child, "type") and child.type == "dotted_name": module_name = get_node_text(child, source_bytes) break # Add this aliased import to the import list symbols["imports"].append( { "name": alias_text, "type": "imports", "location": { "start": { "row": node.start_point[0], "column": node.start_point[1], }, "end": { "row": node.end_point[0], "column": node.end_point[1], }, }, } ) # Also add the module if we found it if module_name: symbols["imports"].append( { "name": module_name, "type": "imports", "location": { "start": { "row": node.start_point[0], "column": 0, # Set to beginning of line }, "end": { "row": node.end_point[0], "column": node.end_point[1], }, }, } ) return symbols except Exception as e: raise ValueError(f"Error extracting symbols from {file_path}: {e}") from e def process_symbol_matches( matches: Any, symbol_type: str, symbols_dict: Dict[str, List[Dict[str, Any]]], source_bytes: bytes, tree: Any, class_ranges: Optional[List[Tuple[int, int]]] = None, ) -> None: """ Process matches from a query and extract symbols. Args: matches: Query matches result symbol_type: Type of symbol being processed symbols_dict: Dictionary to store extracted symbols source_bytes: Source file bytes tree: Parsed syntax tree class_ranges: Optional list of class ranges to filter out class methods """ # Helper function to check if a node is inside a class def is_inside_class(node_row: int) -> bool: if not class_ranges: return False for start_row, end_row in class_ranges: if start_row <= node_row <= end_row: return True return False # Track functions that should be filtered out (methods inside classes) filtered_methods: List[int] = [] # Helper function to process a single node into a symbol def process_node(node: Any, capture_name: str) -> None: try: safe_node = ensure_node(node) # Skip methods inside classes if processing functions with class ranges if class_ranges is not None and is_inside_class(safe_node.start_point[0]): filtered_methods.append(safe_node.start_point[0]) return # Special handling for imports if symbol_type == "imports": # For imports, accept more capture types (.module, .from, .item, .alias, etc.) if not (capture_name.startswith("import.") or capture_name == "import"): return # For aliased imports, we want to include both the original name and the alias if capture_name == "import.alias": # This is an alias in an import statement like "from datetime import datetime as dt" # Get the module and item information module_name = None item_name = None # Get the parent import_from_statement node if safe_node.parent and safe_node.parent.parent: import_node = safe_node.parent.parent for child in import_node.children: if child.type == "dotted_name": # First dotted_name is usually the module if module_name is None: module_name = get_node_text(child, source_bytes, decode=True) # Look for the imported item elif item_name is None and safe_node.parent and safe_node.parent.children: for item_child in safe_node.parent.children: if item_child.type == "dotted_name": item_name = get_node_text(item_child, source_bytes, decode=True) break # Create a descriptive name for the aliased import text = get_node_text(safe_node, source_bytes, decode=True) alias_text = text if module_name and item_name: # Handle both str and bytes cases if ( isinstance(module_name, bytes) or isinstance(item_name, bytes) or isinstance(alias_text, bytes) ): module_name_str = ( module_name.decode("utf-8") if isinstance(module_name, bytes) else module_name ) item_name_str = item_name.decode("utf-8") if isinstance(item_name, bytes) else item_name alias_text_str = alias_text.decode("utf-8") if isinstance(alias_text, bytes) else alias_text text = f"{module_name_str}.{item_name_str} as {alias_text_str}" else: text = f"{module_name}.{item_name} as {alias_text}" elif module_name: # Handle both str and bytes cases if isinstance(module_name, bytes) or isinstance(alias_text, bytes): module_name_str = ( module_name.decode("utf-8") if isinstance(module_name, bytes) else module_name ) alias_text_str = alias_text.decode("utf-8") if isinstance(alias_text, bytes) else alias_text text = f"{module_name_str} as {alias_text_str}" else: text = f"{module_name} as {alias_text}" # For other symbol types elif not capture_name.endswith(".name") and not capture_name == symbol_type: return text = get_node_text(safe_node, source_bytes, decode=True) symbol = { "name": text, "type": symbol_type, "location": { "start": { "row": safe_node.start_point[0], "column": safe_node.start_point[1], }, "end": { "row": safe_node.end_point[0], "column": safe_node.end_point[1], }, }, } # Add to symbols list symbols_dict[symbol_type].append(symbol) except Exception: # Skip problematic nodes pass # Process nodes based on return format if isinstance(matches, dict): # Dictionary format: {capture_name: [node1, node2, ...], ...} for capture_name, nodes in matches.items(): for node in nodes: process_node(node, capture_name) else: # List format: [(node1, capture_name1), (node2, capture_name2), ...] for match in matches: # Handle different return types from query.captures() if isinstance(match, tuple) and len(match) == 2: # Direct tuple unpacking node, capture_name = match elif hasattr(match, "node") and hasattr(match, "capture_name"): # Object with node and capture_name attributes node, capture_name = match.node, match.capture_name elif isinstance(match, dict) and "node" in match and "capture" in match: # Dictionary with node and capture keys node, capture_name = match["node"], match["capture"] else: # Skip if format is unknown continue process_node(node, capture_name) def analyze_project_structure( project: Any, language_registry: Any, scan_depth: int = 3, mcp_ctx: Optional[Any] = None ) -> Dict[str, Any]: """ Analyze the overall structure of a project. Args: project: Project object language_registry: Language registry object scan_depth: Depth to scan for detailed analysis (higher is slower) mcp_ctx: Optional MCP context for progress reporting Returns: Project structure analysis """ root = project.root_path # Create context for progress reporting ctx = MCPContext(mcp_ctx) with ctx.progress_scope(100, "Analyzing project structure") as progress: # Update language information (5%) project.scan_files(language_registry) progress.update(5) # Count files by language languages = project.languages # Find potential entry points based on common patterns entry_points = [] entry_patterns = { "python": ["__main__.py", "main.py", "app.py", "run.py", "manage.py"], "javascript": ["index.js", "app.js", "main.js", "server.js"], "typescript": ["index.ts", "app.ts", "main.ts", "server.ts"], "go": ["main.go"], "rust": ["main.rs"], "java": ["Main.java", "App.java"], } for language, patterns in entry_patterns.items(): if language in languages: for pattern in patterns: # Look for pattern in root and src directories for entry_path in ["", "src/", "lib/"]: candidate = root / entry_path / pattern if candidate.is_file(): rel_path = str(candidate.relative_to(root)) entry_points.append( { "path": rel_path, "language": language, } ) # Look for build configuration files build_files = [] build_patterns = { "python": [ "setup.py", "pyproject.toml", "requirements.txt", "Pipfile", "environment.yml", ], "javascript": ["package.json", "yarn.lock", "npm-shrinkwrap.json"], "typescript": ["tsconfig.json"], "go": ["go.mod", "go.sum"], "rust": ["Cargo.toml", "Cargo.lock"], "java": ["pom.xml", "build.gradle", "build.gradle.kts"], "generic": ["Makefile", "CMakeLists.txt", "Dockerfile", "docker-compose.yml"], } for category, patterns in build_patterns.items(): for pattern in patterns: candidate = root / pattern if candidate.is_file(): rel_path = str(candidate.relative_to(root)) build_files.append( { "path": rel_path, "type": category, } ) # Analyze directory structure dir_counts: Counter = Counter() file_counts: Counter = Counter() for current_dir, dirs, files in os.walk(root): rel_dir = os.path.relpath(current_dir, root) if rel_dir == ".": rel_dir = "" # Skip hidden directories and common excludes # Get config from dependency injection from ..api import get_config config = get_config() dirs[:] = [d for d in dirs if not d.startswith(".") and d not in config.security.excluded_dirs] # Count directories dir_counts[rel_dir] = len(dirs) # Count files by extension for file in files: if file.startswith("."): continue ext = os.path.splitext(file)[1].lower()[1:] if ext: key = f"{rel_dir}/.{ext}" if rel_dir else f".{ext}" file_counts[key] += 1 # Detailed analysis of key files if scan_depth > 0 key_files_analysis = {} if scan_depth > 0: # Analyze a sample of files from each language for language, _ in languages.items(): extensions = [ext for ext, lang in language_registry._language_map.items() if lang == language] if not extensions: continue # Find sample files sample_files = [] for ext in extensions: # Look for files with this extension pattern = f"**/*.{ext}" for path in root.glob(pattern): if path.is_file(): rel_path = str(path.relative_to(root)) sample_files.append(rel_path) if len(sample_files) >= scan_depth: break if len(sample_files) >= scan_depth: break # Analyze sample files if sample_files: language_analysis = [] for file_path in sample_files: try: symbols = extract_symbols(project, file_path, language_registry) # Summarize symbols symbol_counts = { symbol_type: len(symbols_list) for symbol_type, symbols_list in symbols.items() } language_analysis.append( { "file": file_path, "symbols": symbol_counts, } ) except Exception: # Skip problematic files continue if language_analysis: key_files_analysis[language] = language_analysis return { "name": project.name, "path": str(project.root_path), "languages": languages, "entry_points": entry_points, "build_files": build_files, "dir_counts": dict(dir_counts), "file_counts": dict(file_counts), "total_files": sum(languages.values()), "key_files_analysis": key_files_analysis, } def find_dependencies( project: Any, file_path: str, language_registry: Any, ) -> Dict[str, List[str]]: """ Find dependencies of a file. Args: project: Project object file_path: Path to the file relative to project root language_registry: Language registry object Returns: Dictionary of dependencies (imports, includes, etc.) """ abs_path = project.get_file_path(file_path) try: validate_file_access(abs_path, project.root_path) except SecurityError as e: raise SecurityError(f"Access denied: {e}") from e language = language_registry.language_for_file(file_path) if not language: raise ValueError(f"Could not detect language for {file_path}") # Get the appropriate query for imports query_string = get_query_template(language, "imports") if not query_string: raise ValueError(f"Import query not available for {language}") # Parse file and extract imports try: # Get language object language_obj = language_registry.get_language(language) safe_lang = ensure_language(language_obj) # Parse with cached tree tree, source_bytes = parse_with_cached_tree(abs_path, language, safe_lang) # Execute query query = safe_lang.query(query_string) matches = query.captures(tree.root_node) # Organize imports by type imports: Dict[str, List[str]] = defaultdict(list) # Track additional import information to handle aliased imports module_imports: Set[str] = set() # Helper function to process an import node def process_import_node(node: Any, capture_name: str) -> None: try: safe_node = ensure_node(node) text = get_node_text(safe_node, source_bytes) # Determine the import category if capture_name.startswith("import."): category = capture_name.split(".", 1)[1] else: category = "import" # Ensure we're adding a string to the list text_str = text.decode("utf-8") if isinstance(text, bytes) else text imports[category].append(text_str) # Add to module_imports for tracking all imported modules if category == "from": # Handle 'from X import Y' cases parts = text_str.split() if parts: module_part = parts[0].strip() module_imports.add(module_part) elif category == "module": # Handle 'import X' cases text_str = text_str.strip() module_imports.add(text_str) elif category == "alias": # Handle explicitly captured aliases from 'from X import Y as Z' cases # The module itself will be captured separately via the 'from' capture pass elif category == "item" and text: # For individual imported items, make sure to add the module name if it exists if hasattr(safe_node, "parent") and safe_node.parent: parent_node = safe_node.parent # The import_from_statement node # Find the module_name node for child in parent_node.children: if ( hasattr(child, "type") and child.type == "dotted_name" and child != safe_node and hasattr(child, "text") ): module_name_text = get_node_text(child, source_bytes) module_name_str = ( module_name_text.decode("utf-8") if isinstance(module_name_text, bytes) else module_name_text ) module_imports.add(module_name_str) break elif "import" in text_str: # Fallback for raw import statements parts = text_str.split() if len(parts) > 1 and parts[0] == "from": # Handle 'from datetime import datetime as dt' case part = parts[1].strip() module_imports.add(str(part)) elif "from" in text_str and "import" in text_str: # Another way to handle 'from X import Y' patterns # text_str is already properly decoded from_parts = text_str.split("from", 1)[1].split("import", 1) if len(from_parts) > 0: module_name = from_parts[0].strip() module_imports.add(module_name) elif parts[0] == "import": for module in " ".join(parts[1:]).split(","): module = module.strip().split(" as ")[0].strip() module_imports.add(module) except Exception: # Skip problematic nodes pass # Handle different return formats from query.captures() if isinstance(matches, dict): # Dictionary format: {capture_name: [node1, node2, ...], ...} for capture_name, nodes in matches.items(): for node in nodes: process_import_node(node, capture_name) else: # List format: [(node1, capture_name1), (node2, capture_name2), ...] for match in matches: # Handle different return types from query.captures() if isinstance(match, tuple) and len(match) == 2: # Direct tuple unpacking node, capture_name = match elif hasattr(match, "node") and hasattr(match, "capture_name"): # Object with node and capture_name attributes node, capture_name = match.node, match.capture_name elif isinstance(match, dict) and "node" in match and "capture" in match: # Dictionary with node and capture keys node, capture_name = match["node"], match["capture"] else: # Skip if format is unknown continue process_import_node(node, capture_name) # Add all detected modules to the result if module_imports: # Convert module_imports Set[str] to List[str] module_list = list(module_imports) imports["module"] = list(set(imports.get("module", []) + module_list)) # For Python, specifically check for aliased imports if language == "python": # Look for aliased imports directly aliased_query_string = "(aliased_import) @alias" aliased_query = safe_lang.query(aliased_query_string) aliased_matches = aliased_query.captures(tree.root_node) # Process aliased imports for match in aliased_matches: # Initialize variables aliased_node: Optional[Any] = None # We're not using aliased_capture_name but need to unpack it _: str = "" # Handle different return types if isinstance(match, tuple) and len(match) == 2: aliased_node, _ = match elif hasattr(match, "node") and hasattr(match, "capture_name"): aliased_node, _ = match.node, match.capture_name elif isinstance(match, dict) and "node" in match and "capture" in match: aliased_node, _ = match["node"], match["capture"] else: continue # Extract module name from parent if aliased_node is not None and aliased_node.parent and aliased_node.parent.parent: for child in aliased_node.parent.parent.children: if hasattr(child, "type") and child.type == "dotted_name": module_name_text = get_node_text(child, source_bytes) if module_name_text: module_name_str = ( module_name_text.decode("utf-8") if isinstance(module_name_text, bytes) else module_name_text ) module_imports.add(module_name_str) break # Update the module list with any new module imports if module_imports: module_list = list(module_imports) imports["module"] = list(set(imports.get("module", []) + module_list)) return dict(imports) except Exception as e: raise ValueError(f"Error finding dependencies in {file_path}: {e}") from e def analyze_code_complexity( project: Any, file_path: str, language_registry: Any, ) -> Dict[str, Any]: """ Analyze code complexity. Args: project: Project object file_path: Path to the file relative to project root language_registry: Language registry object Returns: Complexity metrics """ abs_path = project.get_file_path(file_path) try: validate_file_access(abs_path, project.root_path) except SecurityError as e: raise SecurityError(f"Access denied: {e}") from e language = language_registry.language_for_file(file_path) if not language: raise ValueError(f"Could not detect language for {file_path}") # Parse file try: # Get language object language_obj = language_registry.get_language(language) safe_lang = ensure_language(language_obj) # Parse with cached tree tree, source_bytes = parse_with_cached_tree(abs_path, language, safe_lang) # Calculate basic metrics # Read lines from file using utility lines = read_text_file(abs_path) line_count = len(lines) empty_lines = sum(1 for line in lines if line.strip() == "") comment_lines = 0 # Language-specific comment detection using utility comment_prefix = get_comment_prefix(language) if comment_prefix: # Count comments for text lines comment_lines = sum(1 for line in lines if line.strip().startswith(comment_prefix)) # Get function and class definitions, excluding methods from count symbols = extract_symbols( project, file_path, language_registry, ["functions", "classes"], exclude_class_methods=True, ) function_count = len(symbols.get("functions", [])) class_count = len(symbols.get("classes", [])) # Calculate cyclomatic complexity using AST complexity_nodes = { "python": [ "if_statement", "for_statement", "while_statement", "try_statement", ], "javascript": [ "if_statement", "for_statement", "while_statement", "try_statement", ], "typescript": [ "if_statement", "for_statement", "while_statement", "try_statement", ], # Add more languages... } cyclomatic_complexity = 1 # Base complexity if language in complexity_nodes: # Count decision points decision_types = complexity_nodes[language] def count_nodes(node: Any, types: List[str]) -> int: safe_node = ensure_node(node) count = 0 if safe_node.type in types: count += 1 for child in safe_node.children: count += count_nodes(child, types) return count cyclomatic_complexity += count_nodes(tree.root_node, decision_types) # Calculate maintainability metrics code_lines = line_count - empty_lines - comment_lines comment_ratio = comment_lines / line_count if line_count > 0 else 0 # Estimate average function length avg_func_lines = float(code_lines / function_count if function_count > 0 else code_lines) return { "line_count": line_count, "code_lines": code_lines, "empty_lines": empty_lines, "comment_lines": comment_lines, "comment_ratio": comment_ratio, "function_count": function_count, "class_count": class_count, "avg_function_lines": round(avg_func_lines, 2), "cyclomatic_complexity": cyclomatic_complexity, "language": language, } except Exception as e: raise ValueError(f"Error analyzing complexity in {file_path}: {e}") from e