hanzoai/python-sdk
2
0
Fork 0
mirror of https://github.com/hanzoai/python-sdk.git synced 2026-10-23 20:23:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: add refactoring tools and cleanup search

Browse Source 
- Add comprehensive refactor tool with rename, extract_function,
  extract_variable, inline, move, and organize_imports actions
- Rename unified_search.py to search_tool.py for cleaner naming
- Make vector search opt-in (not required by default)
- Fix KeyError in ripgrep JSON output parsing
- Update LSP tool to use pyright for Python (faster than pylsp)
- Add 26 tests for refactor tool, all passing
- Bump version to 0.9.8

The refactor tool provides LSP/AST-powered code transformations:
- Rename symbols across entire codebase
- Extract code blocks to functions
- Extract expressions to variables
- Inline variables at all usage sites
- Move symbols between files
- Organize import statements
This commit is contained in:
zeekay
2025-12-21 03:27:47 -08:00
parent df02cec405
commit 70aa0c1875
15 changed files with 2333 additions and 427 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/hanzo-mcp/examples/unified_search_demo.py
-162
View File
@@ -1,162 +0,0 @@
#!/usr/bin/env python3
"""Demo of the unified search tool - THE primary search interface.
This demonstrates how the unified search tool intelligently combines multiple
search modalities to find anything in your codebase.
"""
import sys
import asyncio
from pathlib import Path
# Add parent directory to path
sys.path.insert(0, str(Path(__file__).parent.parent))
from hanzo_mcp.tools.search import create_unified_search_tool
async def demo_unified_search():
"""Demonstrate unified search capabilities."""
print("=== Hanzo Unified Search Demo ===\n")
# Create the search tool
search = create_unified_search_tool()
# Demo 1: Simple text search
print("1. Simple text search for 'UnifiedSearch':")
result = await search.run(pattern="UnifiedSearch", max_results_per_type=3)
if result.data and "results" in result.data:
print(f" Found {len(result.data['results'])} results")
print(f" Search types used: {', '.join(result.data['statistics']['search_types_used'])}")
for r in result.data["results"][:3]:
print(f" - {r['file']}:{r['line']} ({r['type']})")
print()
# Demo 2: Natural language query
print("2. Natural language query:")
result = await search.run(pattern="how does pagination work in search results", max_results_per_type=3)
if result.data:
stats = result.data["statistics"]
print(f" Search types: {', '.join(stats['search_types_used'])}")
print(f" Total matches: {stats['total_matches']}")
print(" Note: Vector search would be used if embedding model is available")
print()
# Demo 3: Code pattern search
print("3. Code pattern search for class definitions:")
result = await search.run(
pattern="class.*Tool", # Regex for Tool classes
max_results_per_type=5,
)
if result.data and "results" in result.data:
print(f" Found {len(result.data['results'])} Tool classes")
unique_files = set(r["file"] for r in result.data["results"])
print(f" Across {len(unique_files)} files")
print()
# Demo 4: File search
print("4. Find Python test files:")
result = await search.run(pattern="test_*.py", search_files=True, max_results_per_type=5)
if result.data:
stats = result.data["statistics"]
if "files" in stats["search_types_used"]:
file_results = [r for r in result.data["results"] if r["type"] == "file"]
print(f" Found {len(file_results)} test files")
for r in file_results[:3]:
print(f" - {r['file']} ({r['context']['semantic']})")
print()
# Demo 5: Symbol search
print("5. Symbol/identifier search:")
result = await search.run(
pattern="BaseTool", # Looking for the BaseTool class
max_results_per_type=3,
)
if result.data:
# Should find symbol definitions
symbol_results = [r for r in result.data["results"] if r.get("context", {}).get("node_type")]
if symbol_results:
print(f" Found {len(symbol_results)} symbol matches")
else:
print(f" Found {len(result.data['results'])} text matches")
print()
# Demo 6: Combined search with pagination
print("6. Paginated results demo:")
result = await search.run(pattern="async def", page_size=10, page=1)
if result.data and "pagination" in result.data:
pag = result.data["pagination"]
print(f" Page {pag['page']} of {pag['total_pages']}")
print(f" Showing {len(result.data['results'])} of {pag['total_results']} total results")
print(f" Has next page: {pag['has_next']}")
print()
# Demo 7: Search with filters
print("7. Filtered search (Python files only):")
result = await search.run(pattern="import", include="*.py", max_results_per_type=5)
if result.data:
print(f" Found {len(result.data['results'])} import statements in Python files")
stats = result.data["statistics"]
print(f" Search completed in {stats['time_ms'].get('text', 0)}ms")
print()
# Summary
print("\n=== Search Intelligence ===")
print("The unified search tool automatically:")
print("- Detects query intent (code vs natural language)")
print("- Runs appropriate search types in parallel")
print("- Combines and ranks results by relevance")
print("- Deduplicates across search types")
print("- Provides context and semantic information")
print("- Handles pagination for large result sets")
print("\nIt's THE primary search interface for finding anything!")
async def demo_find_tool():
"""Demonstrate the find tool for fast file discovery."""
print("\n\n=== Find Tool Demo ===\n")
from hanzo_mcp.tools.search import create_find_tool
find = create_find_tool()
# Demo 1: Find by pattern
print("1. Find all Python files:")
result = await find.run(pattern="*.py", max_results=5)
if result.data and "results" in result.data:
print(f" Found {result.data['statistics']['total_found']} Python files")
print(f" Search time: {result.data['statistics']['search_time_ms']}ms")
print()
# Demo 2: Find with size filter
print("2. Find large files (>10KB):")
result = await find.run(pattern="*", min_size="10KB", max_results=5)
if result.data and "results" in result.data:
for r in result.data["results"][:3]:
size_kb = r["size"] / 1024
print(f" - {r['name']} ({size_kb:.1f} KB)")
print()
# Demo 3: Find recently modified
print("3. Find recently modified files:")
result = await find.run(pattern="*", modified_after="1 hour ago", max_results=5)
if result.data and "results" in result.data:
print(f" Found {len(result.data['results'])} recently modified files")
print("\nFind tool is optimized for fast file discovery by name and attributes!")
if __name__ == "__main__":
# Run demos
asyncio.run(demo_unified_search())
asyncio.run(demo_find_tool())
pkg/hanzo-mcp/hanzo_mcp/config/default_tools.json
+1 -2
View File
@@ -51,7 +51,6 @@
"grep_ast": true,
"git_search": true,
"batch_search": true,
"unified_search": true,
"symbols": true,
"mode": true,
"vector": true,
@@ -137,7 +136,7 @@
"aider", "cline", "openhands", "oh", "hanzo_dev", "think", "critic"
],
"search": [
"grep", "grep_ast", "ast", "search", "unified_search", "symbols", "mode",
"grep", "grep_ast", "ast", "search", "symbols", "mode",
"git_search", "batch_search", "vector", "node", "vector_search", "vector_index", "find_files"
],
"process": [
pkg/hanzo-mcp/hanzo_mcp/prompts/tool_explorer.py
+1 -1
View File
@@ -152,7 +152,7 @@ batch(
1. **Use batch for parallel operations**: When you need to run multiple independent operations, use batch to run them concurrently.
2. **Combine search tools**: Use unified_search for comprehensive results, grep for simple text matching, and grep_ast for code structure understanding.
2. **Combine search tools**: Use search for comprehensive results, grep for simple text matching, and grep_ast for code structure understanding.
3. **Leverage agents for complex tasks**: Use dispatch_agent for tasks requiring deep analysis or multiple steps.
pkg/hanzo-mcp/hanzo_mcp/tools/__init__.py
+22 -1
View File
@@ -54,6 +54,7 @@ try: # pragma: no cover
from hanzo_mcp.tools.common.permissions import PermissionManager
from hanzo_mcp.tools.common.tool_enable import ToolEnableTool
from hanzo_mcp.tools.common.tool_disable import ToolDisableTool
from hanzo_mcp.tools.common.plugin_loader import load_user_plugins
# Try memory tools
try:
@@ -78,6 +79,9 @@ except Exception:
def apply_environment_from_mode():
pass
def load_user_plugins():
return {}
# Try to import LSP tool
try:
@@ -87,6 +91,14 @@ try:
except ImportError:
LSP_TOOL_AVAILABLE = False
# Try to import refactor tool
try:
from hanzo_mcp.tools.refactor import RefactorTool, create_refactor_tool
REFACTOR_TOOL_AVAILABLE = True
except ImportError:
REFACTOR_TOOL_AVAILABLE = False
def register_all_tools(
mcp_server: FastMCP,
@@ -183,7 +195,7 @@ def register_all_tools(
"vector_search": is_tool_enabled("vector_search", False),
}
# Create project manager if vector tools, batch_search, or unified_search are enabled
# Create project manager if vector tools, batch_search, or search are enabled
if (
any(vector_enabled.values())
or filesystem_enabled.get("batch_search", False)
@@ -475,6 +487,15 @@ def register_all_tools(
except Exception as e:
logger.warning(f"Failed to register LSP tool: {e}")
# Register refactor tool if enabled
if is_tool_enabled("refactor", True) and REFACTOR_TOOL_AVAILABLE:
try:
tool = create_refactor_tool()
tool.register(mcp_server)
all_tools[tool.name] = tool
except Exception as e:
logger.warning(f"Failed to register refactor tool: {e}")
# Register user plugins last (so they can override built-in tools)
for plugin_name, plugin in plugins.items():
if is_tool_enabled(plugin_name, True):
pkg/hanzo-mcp/hanzo_mcp/tools/filesystem/__init__.py
+11 -11
View File
@@ -28,14 +28,14 @@ from hanzo_mcp.tools.filesystem.content_replace import ContentReplaceTool
try:
from hanzo_mcp.tools.search import (
FindTool,
UnifiedSearch,
SearchTool,
create_find_tool,
create_unified_search_tool,
create_search_tool,
)
UNIFIED_SEARCH_AVAILABLE = True
SEARCH_AVAILABLE = True
except ImportError:
UNIFIED_SEARCH_AVAILABLE = False
SEARCH_AVAILABLE = False
# Export all tool classes
__all__ = [
@@ -87,8 +87,8 @@ def get_read_only_filesystem_tools(
tools.append(SearchTool(permission_manager, project_manager))
# Add new search tools if available
if UNIFIED_SEARCH_AVAILABLE:
tools.extend([create_unified_search_tool(), create_find_tool()])
if SEARCH_AVAILABLE:
tools.extend([create_search_tool(), create_find_tool()])
return tools
@@ -124,8 +124,8 @@ def get_filesystem_tools(permission_manager: PermissionManager, project_manager=
tools.append(SearchTool(permission_manager, project_manager))
# Add new search tools if available
if UNIFIED_SEARCH_AVAILABLE:
tools.extend([create_unified_search_tool(), create_find_tool()])
if SEARCH_AVAILABLE:
tools.extend([create_search_tool(), create_find_tool()])
return tools
@@ -171,8 +171,8 @@ def register_filesystem_tools(
}
# Add new search tools if available
if UNIFIED_SEARCH_AVAILABLE:
tool_classes["unified_search"] = lambda pm: create_unified_search_tool()
if SEARCH_AVAILABLE:
tool_classes["search"] = lambda pm: create_search_tool()
tool_classes["find"] = lambda pm: create_find_tool()
tools = []
@@ -188,7 +188,7 @@ def register_filesystem_tools(
elif tool_name == "watch":
# Watch tool is a singleton
tools.append(tool_class(permission_manager))
elif tool_name in ["unified_search", "find"]:
elif tool_name in ["search", "find"]:
# New search tools are factory functions
tools.append(tool_class(permission_manager))
else:
pkg/hanzo-mcp/hanzo_mcp/tools/lsp/lsp_tool.py
+533 -123
View File
@@ -1,5 +1,3 @@
from hanzo_mcp.tools.common.auto_timeout import auto_timeout
"""Language Server Protocol (LSP) tool for code intelligence.
This tool provides on-demand LSP configuration and installation for various
@@ -9,13 +7,16 @@ rename symbol, and diagnostics.
"""
import os
import re
import json
import shutil
import asyncio
import logging
from typing import Any, Dict, Optional
import subprocess
import atexit
from typing import Any, Dict, List, Optional, Tuple
from pathlib import Path
from dataclasses import dataclass
from dataclasses import dataclass, field
from hanzo_mcp.types import MCPResourceDocument
from hanzo_mcp.tools.common.base import BaseTool
@@ -39,12 +40,12 @@ LSP_SERVERS = {
],
},
"python": {
"name": "pylsp",
"install_cmd": ["pip", "install", "python-lsp-server[all]"],
"check_cmd": ["pylsp", "--version"],
"start_cmd": ["pylsp"],
"root_markers": ["pyproject.toml", "setup.py", "requirements.txt"],
"file_extensions": [".py"],
"name": "pyright",
"install_cmd": ["npm", "install", "-g", "pyright"],
"check_cmd": ["pyright-langserver", "--version"],
"start_cmd": ["pyright-langserver", "--stdio"],
"root_markers": ["pyproject.toml", "setup.py", "requirements.txt", "pyrightconfig.json"],
"file_extensions": [".py", ".pyi"],
"capabilities": [
"definition",
"references",
@@ -52,6 +53,7 @@ LSP_SERVERS = {
"diagnostics",
"hover",
"completion",
"typeDefinition",
],
},
"typescript": {
@@ -156,9 +158,43 @@ LSP_SERVERS = {
"completion",
],
},
"solidity": {
"name": "solc",
"install_cmd": ["npm", "install", "-g", "solc"],
"check_cmd": ["solc", "--version"],
"start_cmd": ["solc", "--lsp"],
"root_markers": ["foundry.toml", "hardhat.config.js", "hardhat.config.ts", "truffle-config.js"],
"file_extensions": [".sol"],
"capabilities": [
"definition",
"references",
"diagnostics",
"hover",
"completion",
],
},
}
# Global LSP server registry - singleton per language:root_uri
_GLOBAL_SERVERS: Dict[str, "LSPServer"] = {}
_GLOBAL_LOCK = asyncio.Lock()
_CLEANUP_REGISTERED = False
logger = logging.getLogger(__name__)
def _cleanup_all_servers():
"""Cleanup all LSP servers on process exit."""
for server in _GLOBAL_SERVERS.values():
if server.process and server.process.returncode is None:
try:
server.process.terminate()
except Exception:
pass
_GLOBAL_SERVERS.clear()
@dataclass
class LSPServer:
"""Represents an LSP server instance."""
@@ -168,6 +204,14 @@ class LSPServer:
config: Dict[str, Any]
root_uri: str
initialized: bool = False
request_id: int = field(default=0)
pending_responses: Dict[int, asyncio.Future] = field(default_factory=dict)
lock: asyncio.Lock = field(default_factory=asyncio.Lock)
def next_id(self) -> int:
"""Get next request ID."""
self.request_id += 1
return self.request_id
class LSPTool(BaseTool):
@@ -222,8 +266,11 @@ class LSPTool(BaseTool):
def __init__(self):
super().__init__()
self.servers: Dict[str, LSPServer] = {}
self.logger = logging.getLogger(__name__)
# Use global servers registry for persistence across tool instances
global _CLEANUP_REGISTERED
if not _CLEANUP_REGISTERED:
atexit.register(_cleanup_all_servers)
_CLEANUP_REGISTERED = True
def _get_language_from_file(self, file_path: str) -> Optional[str]:
"""Detect language from file extension."""
@@ -270,13 +317,13 @@ class LSPTool(BaseTool):
if not config:
return False
self.logger.info(f"Installing {config['name']} for {language}")
logger.info(f"Installing {config['name']} for {language}")
try:
# Check if installer is available
installer = config["install_cmd"][0]
if not shutil.which(installer):
self.logger.error(f"Installer {installer} not found")
logger.error(f"Installer {installer} not found")
return False
# Run installation command
@@ -289,108 +336,254 @@ class LSPTool(BaseTool):
stdout, stderr = await result.communicate()
if result.returncode != 0:
self.logger.error(f"Installation failed: {stderr.decode()}")
logger.error(f"Installation failed: {stderr.decode()}")
return False
self.logger.info(f"Successfully installed {config['name']}")
logger.info(f"Successfully installed {config['name']}")
return True
except Exception as e:
self.logger.error(f"Installation error: {e}")
logger.error(f"Installation error: {e}")
return False
async def _ensure_lsp_running(self, language: str, root_uri: str) -> Optional[LSPServer]:
"""Ensure LSP server is running for language."""
# Check if already running
"""Ensure LSP server is running for language (uses global singleton)."""
server_key = f"{language}:{root_uri}"
if server_key in self.servers:
server = self.servers[server_key]
# Fast path - check without lock first
if server_key in _GLOBAL_SERVERS:
server = _GLOBAL_SERVERS[server_key]
if server.process and server.process.returncode is None:
return server
# Check if installed
if not await self._check_lsp_installed(language):
# Try to install
if not await self._install_lsp(language):
# Need to start server - acquire global lock
async with _GLOBAL_LOCK:
# Double-check after acquiring lock
if server_key in _GLOBAL_SERVERS:
server = _GLOBAL_SERVERS[server_key]
if server.process and server.process.returncode is None:
return server
# Server died, remove it
del _GLOBAL_SERVERS[server_key]
# Check if installed
if not await self._check_lsp_installed(language):
if not await self._install_lsp(language):
return None
# Start LSP server
config = LSP_SERVERS[language]
try:
process = await asyncio.create_subprocess_exec(
*config["start_cmd"],
stdin=asyncio.subprocess.PIPE,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
cwd=root_uri,
)
server = LSPServer(language=language, process=process, config=config, root_uri=root_uri)
# Initialize LSP
if not await self._initialize_lsp(server):
if server.process:
server.process.terminate()
return None
_GLOBAL_SERVERS[server_key] = server
logger.info(f"Started global LSP server: {config['name']} for {root_uri}")
return server
except Exception as e:
logger.error(f"Failed to start LSP: {e}")
return None
# Start LSP server
config = LSP_SERVERS[language]
try:
process = await asyncio.create_subprocess_exec(
*config["start_cmd"],
stdin=asyncio.subprocess.PIPE,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
cwd=root_uri,
)
server = LSPServer(language=language, process=process, config=config, root_uri=root_uri)
# Initialize LSP
await self._initialize_lsp(server)
self.servers[server_key] = server
return server
except Exception as e:
self.logger.error(f"Failed to start LSP: {e}")
return None
async def _initialize_lsp(self, server: LSPServer):
async def _initialize_lsp(self, server: LSPServer) -> bool:
"""Send initialize request to LSP server."""
# This is a simplified initialization
# In a real implementation, you'd use the full LSP protocol
init_params = {
"processId": os.getpid(),
"rootUri": f"file://{server.root_uri}",
"rootPath": server.root_uri,
"capabilities": {
"workspace": {
"workspaceFolders": True,
"applyEdit": True,
"symbol": {"dynamicRegistration": True},
},
"textDocument": {
"definition": {"dynamicRegistration": True},
"synchronization": {
"dynamicRegistration": True,
"willSave": True,
"willSaveWaitUntil": True,
"didSave": True,
},
"completion": {
"dynamicRegistration": True,
"completionItem": {
"snippetSupport": True,
"documentationFormat": ["markdown", "plaintext"],
},
},
"hover": {
"dynamicRegistration": True,
"contentFormat": ["markdown", "plaintext"],
},
"signatureHelp": {"dynamicRegistration": True},
"definition": {"dynamicRegistration": True, "linkSupport": True},
"references": {"dynamicRegistration": True},
"rename": {"dynamicRegistration": True},
}
"documentHighlight": {"dynamicRegistration": True},
"documentSymbol": {"dynamicRegistration": True},
"formatting": {"dynamicRegistration": True},
"rename": {"dynamicRegistration": True, "prepareSupport": True},
"publishDiagnostics": {"relatedInformation": True},
"implementation": {"dynamicRegistration": True, "linkSupport": True},
"typeDefinition": {"dynamicRegistration": True, "linkSupport": True},
},
},
"workspaceFolders": [
{"uri": f"file://{server.root_uri}", "name": Path(server.root_uri).name}
],
}
# Send initialize request
request = {
"jsonrpc": "2.0",
"id": 1,
"id": server.next_id(),
"method": "initialize",
"params": init_params,
}
await self._send_request(server, request)
server.initialized = True
response = await self._send_request(server, request, timeout=60.0)
if not response or "error" in response:
logger.error(f"Failed to initialize LSP: {response}")
return False
async def _send_request(self, server: LSPServer, request: Dict[str, Any]) -> Optional[Dict[str, Any]]:
"""Send JSON-RPC request to LSP server."""
if not server.process or server.process.returncode is not None:
# Send initialized notification
await self._send_notification(server, "initialized", {})
server.initialized = True
logger.info(f"LSP server {server.config['name']} initialized for {server.root_uri}")
return True
async def _read_lsp_message(self, server: LSPServer, timeout: float = 30.0) -> Optional[Dict[str, Any]]:
"""Read a single LSP message from server stdout."""
if not server.process or not server.process.stdout:
return None
try:
# Serialize request
request_str = json.dumps(request)
content_length = len(request_str.encode("utf-8"))
# Read headers until empty line
headers = {}
while True:
line = await asyncio.wait_for(
server.process.stdout.readline(),
timeout=timeout
)
if not line:
return None
line_str = line.decode("utf-8").strip()
if not line_str:
break
if ":" in line_str:
key, value = line_str.split(":", 1)
headers[key.strip().lower()] = value.strip()
# Send LSP message
message = f"Content-Length: {content_length}\r\n\r\n{request_str}"
server.process.stdin.write(message.encode("utf-8"))
# Get content length
content_length = int(headers.get("content-length", 0))
if content_length == 0:
return None
# Read content
content = await asyncio.wait_for(
server.process.stdout.read(content_length),
timeout=timeout
)
return json.loads(content.decode("utf-8"))
except asyncio.TimeoutError:
logger.debug("Timeout reading LSP message")
return None
except Exception as e:
logger.error(f"Error reading LSP message: {e}")
return None
async def _send_request(self, server: LSPServer, request: Dict[str, Any], timeout: float = 30.0) -> Optional[Dict[str, Any]]:
"""Send JSON-RPC request to LSP server and wait for response."""
if not server.process or server.process.returncode is not None:
return None
if not server.process.stdin:
return None
import time
async with server.lock:
try:
# Serialize request
request_str = json.dumps(request)
content = request_str.encode("utf-8")
content_length = len(content)
# Build and send LSP message
header = f"Content-Length: {content_length}\r\n\r\n"
message = header.encode("utf-8") + content
server.process.stdin.write(message)
await server.process.stdin.drain()
# Read response(s) until we get one with matching id
request_id = request.get("id")
start_time = time.monotonic()
deadline = start_time + timeout
while time.monotonic() < deadline:
remaining = max(0.1, deadline - time.monotonic())
response = await self._read_lsp_message(server, timeout=remaining)
if response is None:
# Check if server died
if server.process.returncode is not None:
logger.error(f"LSP server died with code {server.process.returncode}")
return None
continue
# Check if this is our response
if "id" in response and response["id"] == request_id:
return response
# Handle notifications (no id) - log and continue
if "id" not in response and "method" in response:
logger.debug(f"LSP notification: {response.get('method')}")
continue
logger.warning(f"Timeout waiting for response to request {request_id}")
return None
except Exception as e:
logger.error(f"LSP communication error: {e}")
return None
async def _send_notification(self, server: LSPServer, method: str, params: Dict[str, Any]) -> bool:
"""Send JSON-RPC notification (no response expected)."""
if not server.process or server.process.returncode is not None:
return False
if not server.process.stdin:
return False
try:
notification = {
"jsonrpc": "2.0",
"method": method,
"params": params,
}
notification_str = json.dumps(notification)
content = notification_str.encode("utf-8")
content_length = len(content)
header = f"Content-Length: {content_length}\r\n\r\n"
message = header.encode("utf-8") + content
server.process.stdin.write(message)
await server.process.stdin.drain()
# Read response (simplified - real implementation needs proper parsing)
# This is a placeholder - actual LSP requires parsing Content-Length headers
response_data = await server.process.stdout.readline()
if response_data:
return json.loads(response_data.decode("utf-8"))
return True
except Exception as e:
self.logger.error(f"LSP communication error: {e}")
return None
logger.error(f"Failed to send notification: {e}")
return False
async def run(
self,
@@ -474,7 +667,6 @@ class LSPTool(BaseTool):
return MCPResourceDocument(data=result)
@auto_timeout("lsp")
async def call(self, **kwargs) -> str:
"""Tool interface for MCP - converts result to JSON string."""
result = await self.run(**kwargs)
@@ -500,6 +692,53 @@ class LSPTool(BaseTool):
new_name=new_name,
)
async def _open_document(self, server: LSPServer, file_path: str) -> bool:
"""Notify LSP server that a document is opened."""
abs_path = str(Path(file_path).resolve())
uri = f"file://{abs_path}"
# Read file content
try:
with open(abs_path, "r", encoding="utf-8") as f:
content = f.read()
except Exception as e:
logger.error(f"Failed to read file {file_path}: {e}")
return False
# Determine language ID
language_id = server.language
if server.language == "typescript":
ext = Path(file_path).suffix.lower()
if ext in [".tsx", ".jsx"]:
language_id = "typescriptreact" if "ts" in ext else "javascriptreact"
elif ext in [".js", ".mjs", ".cjs"]:
language_id = "javascript"
params = {
"textDocument": {
"uri": uri,
"languageId": language_id,
"version": 1,
"text": content,
}
}
return await self._send_notification(server, "textDocument/didOpen", params)
def _parse_location(self, location: Dict[str, Any]) -> Dict[str, Any]:
"""Parse LSP Location into a readable format."""
uri = location.get("uri", "")
file_path = uri.replace("file://", "") if uri.startswith("file://") else uri
range_info = location.get("range", {})
start = range_info.get("start", {})
end = range_info.get("end", {})
return {
"file": file_path,
"start": {"line": start.get("line", 0) + 1, "character": start.get("character", 0)},
"end": {"line": end.get("line", 0) + 1, "character": end.get("character", 0)},
}
async def _execute_lsp_action(
self,
server: LSPServer,
@@ -510,76 +749,247 @@ class LSPTool(BaseTool):
new_name: Optional[str],
) -> Dict[str, Any]:
"""Execute specific LSP action."""
abs_path = str(Path(file).resolve())
uri = f"file://{abs_path}"
# This is a simplified implementation
# Real implementation would use proper LSP protocol
# Open document first
await self._open_document(server, file)
# Build position (LSP uses 0-based line and character)
position = {
"line": (line - 1) if line else 0,
"character": character if character else 0,
}
if action == "definition":
# textDocument/definition request
return {
"action": "definition",
"file": file,
"position": {"line": line, "character": character},
"note": "LSP integration pending full implementation",
"fallback": "Use mcp__lsp__find_definition tool for now",
request = {
"jsonrpc": "2.0",
"id": server.next_id(),
"method": "textDocument/definition",
"params": {
"textDocument": {"uri": uri},
"position": position,
},
}
response = await self._send_request(server, request)
if response and "result" in response:
result = response["result"]
if result is None:
return {"action": "definition", "file": file, "result": None, "message": "No definition found"}
if isinstance(result, list):
return {
"action": "definition",
"file": file,
"definitions": [self._parse_location(loc) for loc in result],
}
elif isinstance(result, dict):
return {
"action": "definition",
"file": file,
"definition": self._parse_location(result),
}
return {"action": "definition", "file": file, "error": response.get("error") if response else "No response"}
elif action == "references":
# textDocument/references request
return {
"action": "references",
"file": file,
"position": {"line": line, "character": character},
"note": "LSP integration pending full implementation",
"fallback": "Use mcp__lsp__find_references tool for now",
request = {
"jsonrpc": "2.0",
"id": server.next_id(),
"method": "textDocument/references",
"params": {
"textDocument": {"uri": uri},
"position": position,
"context": {"includeDeclaration": True},
},
}
response = await self._send_request(server, request)
if response and "result" in response:
result = response["result"]
if result is None:
return {"action": "references", "file": file, "references": [], "count": 0}
refs = [self._parse_location(loc) for loc in result] if isinstance(result, list) else []
return {"action": "references", "file": file, "references": refs, "count": len(refs)}
return {"action": "references", "file": file, "error": response.get("error") if response else "No response"}
elif action == "rename":
# textDocument/rename request
return {
"action": "rename",
"file": file,
"position": {"line": line, "character": character},
"new_name": new_name,
"note": "LSP integration pending full implementation",
"fallback": "Use mcp__lsp__rename_symbol tool for now",
if not new_name:
return {"action": "rename", "error": "new_name is required for rename action"}
request = {
"jsonrpc": "2.0",
"id": server.next_id(),
"method": "textDocument/rename",
"params": {
"textDocument": {"uri": uri},
"position": position,
"newName": new_name,
},
}
response = await self._send_request(server, request)
if response and "result" in response:
result = response["result"]
if result is None:
return {"action": "rename", "file": file, "error": "Rename not possible at this location"}
# Parse workspace edit
changes = {}
if "changes" in result:
for file_uri, edits in result["changes"].items():
file_path = file_uri.replace("file://", "")
changes[file_path] = [
{
"range": self._parse_location({"uri": file_uri, "range": edit["range"]}),
"newText": edit["newText"],
}
for edit in edits
]
elif "documentChanges" in result:
for change in result["documentChanges"]:
if "textDocument" in change:
file_path = change["textDocument"]["uri"].replace("file://", "")
changes[file_path] = [
{
"range": self._parse_location({"uri": change["textDocument"]["uri"], "range": edit["range"]}),
"newText": edit["newText"],
}
for edit in change.get("edits", [])
]
return {
"action": "rename",
"file": file,
"new_name": new_name,
"changes": changes,
"files_affected": len(changes),
}
return {"action": "rename", "file": file, "error": response.get("error") if response else "No response"}
elif action == "diagnostics":
# textDocument/diagnostic request
# For diagnostics, we read file to trigger analysis and then wait for publishDiagnostics
# However, diagnostics are typically push-based, so we'll use a workaround
# Some LSP servers support textDocument/diagnostic (LSP 3.17+)
request = {
"jsonrpc": "2.0",
"id": server.next_id(),
"method": "textDocument/diagnostic",
"params": {
"textDocument": {"uri": uri},
},
}
response = await self._send_request(server, request, timeout=10.0)
if response and "result" in response:
result = response["result"]
items = result.get("items", []) if isinstance(result, dict) else []
return {
"action": "diagnostics",
"file": file,
"diagnostics": [
{
"severity": item.get("severity", 1),
"message": item.get("message", ""),
"range": self._parse_location({"uri": uri, "range": item.get("range", {})}),
"source": item.get("source", ""),
}
for item in items
],
"count": len(items),
}
# Fallback: Return note about push-based diagnostics
return {
"action": "diagnostics",
"file": file,
"note": "LSP integration pending full implementation",
"fallback": "Use mcp__lsp__get_diagnostics tool for now",
"note": "Diagnostics are push-based; use language-specific tools (go vet, pylint, etc.) for on-demand checking",
}
elif action == "hover":
# textDocument/hover request
return {
"action": "hover",
"file": file,
"position": {"line": line, "character": character},
"note": "LSP integration pending full implementation",
request = {
"jsonrpc": "2.0",
"id": server.next_id(),
"method": "textDocument/hover",
"params": {
"textDocument": {"uri": uri},
"position": position,
},
}
response = await self._send_request(server, request)
if response and "result" in response:
result = response["result"]
if result is None:
return {"action": "hover", "file": file, "result": None, "message": "No hover info"}
contents = result.get("contents", "")
# Handle different content formats
if isinstance(contents, dict):
hover_text = contents.get("value", str(contents))
elif isinstance(contents, list):
hover_text = "\n".join(
c.get("value", str(c)) if isinstance(c, dict) else str(c)
for c in contents
)
else:
hover_text = str(contents)
return {
"action": "hover",
"file": file,
"position": {"line": line, "character": character},
"contents": hover_text,
}
return {"action": "hover", "file": file, "error": response.get("error") if response else "No response"}
elif action == "completion":
# textDocument/completion request
return {
"action": "completion",
"file": file,
"position": {"line": line, "character": character},
"note": "LSP integration pending full implementation",
request = {
"jsonrpc": "2.0",
"id": server.next_id(),
"method": "textDocument/completion",
"params": {
"textDocument": {"uri": uri},
"position": position,
},
}
response = await self._send_request(server, request, timeout=10.0)
if response and "result" in response:
result = response["result"]
if result is None:
return {"action": "completion", "file": file, "completions": [], "count": 0}
return {"error": "Unknown action"}
items = result if isinstance(result, list) else result.get("items", [])
completions = [
{
"label": item.get("label", ""),
"kind": item.get("kind", 0),
"detail": item.get("detail", ""),
"documentation": (
item.get("documentation", {}).get("value", "")
if isinstance(item.get("documentation"), dict)
else str(item.get("documentation", ""))
),
}
for item in items[:50] # Limit to 50 items
]
return {
"action": "completion",
"file": file,
"position": {"line": line, "character": character},
"completions": completions,
"count": len(completions),
}
return {"action": "completion", "file": file, "error": response.get("error") if response else "No response"}
return {"error": f"Unknown action: {action}"}
async def cleanup(self):
"""Clean up LSP servers."""
for server in self.servers.values():
if server.process and server.process.returncode is None:
server.process.terminate()
await server.process.wait()
"""Clean up LSP servers (uses global registry)."""
async with _GLOBAL_LOCK:
for server_key, server in list(_GLOBAL_SERVERS.items()):
if server.process and server.process.returncode is None:
try:
server.process.terminate()
await asyncio.wait_for(server.process.wait(), timeout=5.0)
except asyncio.TimeoutError:
server.process.kill()
except Exception:
pass
_GLOBAL_SERVERS.clear()
# Tool registration
pkg/hanzo-mcp/hanzo_mcp/tools/refactor/__init__.py
+13
View File
@@ -0,0 +1,13 @@
"""Refactoring tools for code transformation.
This module provides advanced refactoring capabilities:
- rename: Rename symbols across codebase using LSP
- extract_function: Extract code to new function
- inline: Inline functions or variables
- move: Move code between files
- change_signature: Modify function signatures
"""
from hanzo_mcp.tools.refactor.refactor_tool import RefactorTool, create_refactor_tool
__all__ = ["RefactorTool", "create_refactor_tool"]
pkg/hanzo-mcp/hanzo_mcp/tools/refactor/refactor_tool.py
+1123
View File
File diff suppressed because it is too large Load Diff
pkg/hanzo-mcp/hanzo_mcp/tools/search/__init__.py
+3 -3
View File
@@ -1,11 +1,11 @@
"""Search tools for finding code, files, and information."""
from .find_tool import FindTool, create_find_tool
from .unified_search import UnifiedSearch, create_unified_search_tool
from .search_tool import SearchTool, create_search_tool
__all__ = [
"UnifiedSearch",
"create_unified_search_tool",
"SearchTool",
"create_search_tool",
"FindTool",
"create_find_tool",
]
pkg/hanzo-mcp/hanzo_mcp/tools/search/find_tool.py
+44 -78
View File
@@ -1,9 +1,8 @@
"""Fast file finding tool using ffind and intelligent caching."""
"""Fast file finding tool using ffind library and intelligent caching."""
import os
import time
import fnmatch
import subprocess
from typing import Any, Set, Dict, List, Optional
from pathlib import Path
from datetime import datetime
@@ -13,11 +12,11 @@ from hanzo_mcp.types import MCPResourceDocument
from hanzo_mcp.tools.common.base import BaseTool
from hanzo_mcp.tools.common.auto_timeout import auto_timeout
# Check if ffind command is available
# Check if ffind library is available
try:
subprocess.run(["ffind", "--version"], capture_output=True, check=True)
from ffind.ffind import search as ffind_search
FFIND_AVAILABLE = True
except (subprocess.CalledProcessError, FileNotFoundError):
except ImportError:
FFIND_AVAILABLE = False
@@ -188,21 +187,32 @@ class FindTool(BaseTool):
self._gitignore_cache[root] = patterns
return patterns
def _should_ignore(self, path: str, ignore_patterns: Set[str]) -> bool:
"""Check if path should be ignored."""
def _should_ignore(self, path: str, ignore_patterns: Set[str], root: Optional[Path] = None) -> bool:
"""Check if path should be ignored.
Args:
path: The absolute path to check
ignore_patterns: Set of gitignore-style patterns
root: Optional search root - only check parents BELOW this root
"""
path_obj = Path(path)
for pattern in ignore_patterns:
# Check against full path and basename
# Check against basename only
if fnmatch.fnmatch(path_obj.name, pattern):
return True
if fnmatch.fnmatch(str(path_obj), pattern):
return True
# Check if any parent directory matches
for parent in path_obj.parents:
if fnmatch.fnmatch(parent.name, pattern):
return True
# Check relative path if we have a root
if root:
try:
rel_path = path_obj.relative_to(root)
# Check relative path parts (parents below root)
for part in rel_path.parts[:-1]: # Exclude the filename itself
if fnmatch.fnmatch(part, pattern):
return True
except ValueError:
# Path is not relative to root, skip parent checks
pass
return False
@@ -441,75 +451,31 @@ class FindTool(BaseTool):
respect_gitignore: bool,
ignore_patterns: Set[str],
) -> List[FileMatch]:
"""Use ffind for fast file discovery."""
"""Use ffind library for fast file discovery."""
matches = []
# Configure ffind
ffind_args = {
"path": str(root),
"pattern": pattern,
"regex": regex,
"case_sensitive": case_sensitive,
"follow_symlinks": follow_symlinks,
}
if fuzzy:
ffind_args["fuzzy"] = True
if max_depth:
ffind_args["max_depth"] = max_depth
try:
# Build ffind command
cmd = ["ffind"]
if not case_sensitive:
cmd.append("-i")
if regex:
cmd.append("-E")
if fuzzy:
cmd.append("-f")
if follow_symlinks:
cmd.append("-L")
if max_depth:
cmd.extend(["-D", str(max_depth)])
# Add path and pattern (ffind expects directory first)
cmd.append(str(root))
cmd.append(pattern)
# Run ffind command
result = subprocess.run(cmd, capture_output=True, text=True, check=False)
if result.returncode != 0:
# Fall back to Python implementation on error
return await self._find_with_python(
pattern,
root,
file_type,
case_sensitive,
regex,
fuzzy,
False,
max_depth,
follow_symlinks,
respect_gitignore,
ignore_patterns,
)
# Parse results
results = result.stdout.strip().split("\n") if result.stdout else []
# Use ffind as Python library directly
results = ffind_search(
directory=str(root),
file_pattern=pattern,
path_match=False, # Match filename only, not full path
follow_symlinks=follow_symlinks,
output=False, # Don't print
colored=False,
ignore_hidden=True,
ignore_case=not case_sensitive,
ignore_vcs=respect_gitignore,
return_results=True,
fuzzy=fuzzy,
)
for path in results:
if not path: # Skip empty lines
continue
# Check ignore patterns
if self._should_ignore(path, ignore_patterns):
# Check ignore patterns (pass root to only check relative parents)
if self._should_ignore(path, ignore_patterns, root):
continue
# Get file info
@@ -609,7 +575,7 @@ class FindTool(BaseTool):
# Filter directories to skip
if respect_gitignore:
dirnames[:] = [
d for d in dirnames if not self._should_ignore(os.path.join(dirpath, d), ignore_patterns)
d for d in dirnames if not self._should_ignore(os.path.join(dirpath, d), ignore_patterns, root)
]
# Check directories
@@ -617,7 +583,7 @@ class FindTool(BaseTool):
for dirname in dirnames:
if matcher(dirname):
full_path = os.path.join(dirpath, dirname)
if not self._should_ignore(full_path, ignore_patterns):
if not self._should_ignore(full_path, ignore_patterns, root):
try:
stat = os.stat(full_path)
match = FileMatch(
@@ -638,7 +604,7 @@ class FindTool(BaseTool):
for filename in filenames:
full_path = os.path.join(dirpath, filename)
if self._should_ignore(full_path, ignore_patterns):
if self._should_ignore(full_path, ignore_patterns, root):
continue
# Match against filename
pkg/hanzo-mcp/hanzo_mcp/tools/search/unified_search.py → pkg/hanzo-mcp/hanzo_mcp/tools/search/search_tool.py
+203 -17
View File
@@ -39,6 +39,14 @@ try:
except ImportError:
VECTOR_SEARCH_AVAILABLE = False
# LSP availability check
try:
from hanzo_mcp.tools.lsp.lsp_tool import LSPTool, LSP_SERVERS
LSP_AVAILABLE = True
except ImportError:
LSP_AVAILABLE = False
@dataclass
class SearchResult:
@@ -76,7 +84,7 @@ class SearchResult:
return hash((self.file_path, self.line_number, self.column, self.match_text))
class UnifiedSearch(BaseTool):
class SearchTool(BaseTool):
"""THE primary search tool - your universal interface for finding anything.
This is the main search tool you should use for finding:
@@ -137,14 +145,20 @@ class UnifiedSearch(BaseTool):
Automatically detects query intent and runs appropriate searches in parallel.
"""
def __init__(self):
def __init__(self, enable_vector_index: bool = False):
"""Initialize search tool.
Args:
enable_vector_index: Whether to enable vector search indexing (default: False).
When False, vector search is disabled for faster startup.
Set to True to enable semantic search capabilities.
"""
super().__init__()
self.ripgrep_available = self._check_ripgrep()
self.vector_db = None
self.embedder = None
if VECTOR_SEARCH_AVAILABLE:
self._init_vector_search()
self._vector_initialized = False
self._enable_vector_index = enable_vector_index
def _check_ripgrep(self) -> bool:
"""Check if ripgrep is available."""
@@ -154,8 +168,23 @@ class UnifiedSearch(BaseTool):
except Exception:
return False
def _init_vector_search(self):
"""Initialize vector search components."""
def _init_vector_search(self) -> bool:
"""Initialize vector search components lazily.
Returns:
True if initialization succeeded, False otherwise.
"""
if self._vector_initialized:
return self.vector_db is not None
self._vector_initialized = True
if not VECTOR_SEARCH_AVAILABLE:
return False
if not self._enable_vector_index:
return False
try:
self.embedder = SentenceTransformer("all-MiniLM-L6-v2")
self.vector_db = chromadb.Client()
@@ -163,9 +192,11 @@ class UnifiedSearch(BaseTool):
self.collection = self.vector_db.get_or_create_collection(
name="code_search", metadata={"description": "Code semantic search"}
)
return True
except Exception as e:
print(f"Failed to initialize vector search: {e}")
self.vector_db = None
return False
def _should_use_vector_search(self, query: str) -> bool:
"""Determine if vector search would be helpful."""
@@ -212,6 +243,7 @@ class UnifiedSearch(BaseTool):
enable_ast: bool = None,
enable_vector: bool = None,
enable_symbol: bool = None,
enable_lsp: bool = None,
page_size: int = 50,
page: int = 1,
**kwargs,
@@ -228,6 +260,7 @@ class UnifiedSearch(BaseTool):
search_files: Also search for matching filenames
search_memory: Search in memory/knowledge base (auto-detected if None)
enable_*: Force enable/disable specific search types (auto if None)
enable_lsp: Enable LSP-based reference search (auto if None)
page_size: Results per page (default: 50)
page: Page number to retrieve (default: 1)
"""
@@ -244,7 +277,12 @@ class UnifiedSearch(BaseTool):
enable_text = True # Always use text search as baseline
if enable_vector is None:
enable_vector = self._should_use_vector_search(pattern) and VECTOR_SEARCH_AVAILABLE
# Only enable vector search if explicitly enabled via enable_vector_index
enable_vector = (
self._enable_vector_index
and self._should_use_vector_search(pattern)
and VECTOR_SEARCH_AVAILABLE
)
if enable_ast is None:
enable_ast = self._should_use_ast_search(pattern) and TREESITTER_AVAILABLE
@@ -252,6 +290,10 @@ class UnifiedSearch(BaseTool):
if enable_symbol is None:
enable_symbol = self._should_use_symbol_search(pattern)
if enable_lsp is None:
# Use LSP for symbol-like queries when LSP is available
enable_lsp = LSP_AVAILABLE and self._should_use_symbol_search(pattern)
# Collect results from all enabled search types
all_results = []
search_stats = {
@@ -287,8 +329,8 @@ class UnifiedSearch(BaseTool):
search_stats["search_types_used"].append("symbol")
all_results.extend(symbol_results)
# 4. Vector search - for semantic similarity
if enable_vector and self.vector_db:
# 4. Vector search - for semantic similarity (lazy init)
if enable_vector and self._init_vector_search():
start = time.time()
vector_results = await self._vector_search(
pattern, path, include, exclude, max_results_per_type, context_lines
@@ -313,6 +355,14 @@ class UnifiedSearch(BaseTool):
search_stats["search_types_used"].append("memory")
all_results.extend(memory_results)
# 7. LSP search - for precise symbol references and definitions
if enable_lsp and LSP_AVAILABLE:
start = time.time()
lsp_results = await self._lsp_search(pattern, path, include, max_results_per_type)
search_stats["time_ms"]["lsp"] = int((time.time() - start) * 1000)
search_stats["search_types_used"].append("lsp")
all_results.extend(lsp_results)
# Deduplicate and rank results
unique_results = self._deduplicate_results(all_results)
ranked_results = self._rank_results(unique_results, pattern)
@@ -433,11 +483,23 @@ class UnifiedSearch(BaseTool):
if data.get("type") == "match":
match_data = data["data"]
# Handle different ripgrep output formats
lines_data = match_data.get("lines", {})
match_text = lines_data.get("text", "") if isinstance(lines_data, dict) else str(lines_data)
# Get file path - handle both dict and str formats
path_data = match_data.get("path", {})
file_path = path_data.get("text", str(path_data)) if isinstance(path_data, dict) else str(path_data)
# Get column safely
submatches = match_data.get("submatches", [{}])
column = submatches[0].get("start", 0) if submatches else 0
result = SearchResult(
file_path=match_data["path"]["text"],
line_number=match_data["line_number"],
column=match_data["submatches"][0]["start"],
match_text=match_data["lines"]["text"].strip(),
file_path=file_path,
line_number=match_data.get("line_number", 0),
column=column,
match_text=match_text.strip() if isinstance(match_text, str) else str(match_text),
context_before=[],
context_after=[],
match_type="text",
@@ -727,6 +789,125 @@ class UnifiedSearch(BaseTool):
return results
async def _lsp_search(
self,
pattern: str,
path: str,
include: Optional[str],
max_results: int,
) -> List[SearchResult]:
"""Search using LSP for precise symbol references.
Uses AST search to find initial symbol locations, then LSP to find
all references across the codebase.
"""
results = []
if not LSP_AVAILABLE:
return results
try:
# First, find files that might contain the symbol
root_path = Path(path).resolve()
# Get file extensions we can search with LSP
supported_extensions = set()
for lang_info in LSP_SERVERS.values():
supported_extensions.update(lang_info.get("extensions", []))
# Find matching files using grep to locate the symbol
grep_cmd = ["rg", "--files-with-matches", "-l", pattern]
if include:
grep_cmd.extend(["--glob", include])
grep_cmd.append(str(root_path))
try:
result = subprocess.run(
grep_cmd,
capture_output=True,
text=True,
timeout=10,
)
matching_files = [f.strip() for f in result.stdout.strip().split("\n") if f.strip()]
except (subprocess.SubprocessError, FileNotFoundError):
matching_files = []
if not matching_files:
return results
# Create LSP tool
lsp_tool = LSPTool()
# For each matching file, try to find references using LSP
files_checked = 0
for file_path in matching_files[:5]: # Limit to first 5 files to avoid slowdown
if files_checked >= 3: # Also limit files we actually query LSP on
break
# Check if file extension is supported
ext = Path(file_path).suffix
if ext not in supported_extensions:
continue
# Find the symbol's line and column in this file
try:
with open(file_path, "r", encoding="utf-8", errors="ignore") as f:
for line_num, line in enumerate(f, 1):
col = line.find(pattern)
if col >= 0:
# Found the pattern, try to get LSP references
try:
lsp_result = await lsp_tool.run(
action="references",
file=file_path,
line=line_num,
character=col,
)
if lsp_result.data and "references" in lsp_result.data:
for ref in lsp_result.data["references"][:max_results]:
ref_path = ref.get("uri", "").replace("file://", "")
ref_line = ref.get("range", {}).get("start", {}).get("line", 0) + 1
# Read the line content
try:
with open(ref_path, "r", encoding="utf-8", errors="ignore") as rf:
lines = rf.readlines()
match_text = lines[ref_line - 1].strip() if ref_line <= len(lines) else pattern
except (IOError, IndexError):
match_text = pattern
result = SearchResult(
file_path=ref_path,
line_number=ref_line,
column=ref.get("range", {}).get("start", {}).get("character", 0),
match_text=match_text,
context_before=[],
context_after=[],
match_type="lsp",
score=0.95, # High confidence from LSP
semantic_context=f"LSP reference to '{pattern}'",
)
results.append(result)
if len(results) >= max_results:
return results
except Exception:
# LSP query failed, continue to next occurrence
pass
files_checked += 1
break # Only check first occurrence in each file
except (IOError, UnicodeDecodeError):
continue
except Exception as e:
print(f"LSP search error: {e}")
return results
def _deduplicate_results(self, results: List[SearchResult]) -> List[SearchResult]:
"""Remove duplicate results across search types."""
seen = set()
@@ -948,6 +1129,11 @@ class CodeIndexer:
# Tool registration
def create_unified_search_tool():
"""Factory function to create unified search tool."""
return UnifiedSearch()
def create_search_tool(enable_vector_index: bool = False):
"""Factory function to create search tool.
Args:
enable_vector_index: Enable vector/semantic search (requires chromadb + sentence-transformers).
Default False for fast startup without database dependencies.
"""
return SearchTool(enable_vector_index=enable_vector_index)
pkg/hanzo-mcp/pyproject.toml
+4 -4
View File
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "hanzo-mcp"
version = "0.9.3"
version = "0.9.8"
description = "The Zen of Hanzo MCP: One server to rule them all. The ultimate MCP that orchestrates all others."
readme = "README.md"
requires-python = ">=3.12"
@@ -88,7 +88,9 @@ include = ["hanzo_mcp"]
[tool.pytest.ini_options]
# Configuration for pytest
addopts = "--no-header --no-summary"
addopts = "--no-header --no-summary -p asyncio"
asyncio_mode = "auto"
asyncio_default_fixture_loop_scope = "function"
# Note: --cov-fail-under=96.3 will be added after achieving target coverage
markers = [
"asyncio: mark test as using asyncio",
@@ -101,8 +103,6 @@ testpaths = ["tests"]
python_files = ["test_*.py", "*_test.py"]
python_classes = ["Test*"]
python_functions = ["test_*"]
asyncio_mode = "auto"
asyncio_default_fixture_loop_scope = "function"
[tool.mypy]
python_version = "3.12"
pkg/hanzo-mcp/tests/conftest.py
+3 -6
View File
@@ -10,6 +10,9 @@ import tempfile
import pytest
# Register pytest-asyncio plugin explicitly
pytest_plugins = ["pytest_asyncio"]
# Add tests directory to path so we can import test_utils
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
@@ -33,12 +36,6 @@ os.environ["PYTEST_CURRENT_TEST"] = "1" # Mark as pytest run
# Configure pytest
def pytest_configure(config):
"""Configure pytest."""
# Register asyncio marker
config.addinivalue_line("markers", "asyncio: mark test as using asyncio")
# Configure pytest-asyncio
config._inicache["asyncio_default_fixture_loop_scope"] = "function"
# Register custom markers
config.addinivalue_line("markers", "requires_hanzo_agents: mark test as requiring hanzo-agents SDK")
config.addinivalue_line("markers", "requires_memory_tools: mark test as requiring hanzo-memory package")
pkg/hanzo-mcp/tests/test_refactor_tool.py
+340
View File
@@ -0,0 +1,340 @@
"""Tests for the refactor tool."""
import os
import tempfile
import pytest
from hanzo_mcp.tools.refactor import RefactorTool, create_refactor_tool
class TestRefactorTool:
"""Tests for RefactorTool class."""
@pytest.fixture
def tool(self):
"""Create a refactor tool instance."""
return create_refactor_tool()
@pytest.fixture
def temp_python_file(self):
"""Create a temporary Python file for testing."""
with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f:
f.write('''def old_function(x, y):
result = x + y
return result
def caller():
value = old_function(1, 2)
return value
class MyClass:
def method(self):
return old_function(3, 4)
''')
temp_path = f.name
yield temp_path
# Cleanup
if os.path.exists(temp_path):
os.unlink(temp_path)
@pytest.fixture
def temp_js_file(self):
"""Create a temporary JavaScript file for testing."""
with tempfile.NamedTemporaryFile(mode="w", suffix=".js", delete=False) as f:
f.write('''function oldFunction(x, y) {
const result = x + y;
return result;
}
function caller() {
const value = oldFunction(1, 2);
return value;
}
''')
temp_path = f.name
yield temp_path
if os.path.exists(temp_path):
os.unlink(temp_path)
def test_tool_creation(self, tool):
"""Test that the tool can be created."""
assert tool is not None
assert tool.name == "refactor"
def test_tool_has_description(self, tool):
"""Test that the tool has a description."""
assert tool.description is not None
assert "refactor" in tool.description.lower()
@pytest.mark.asyncio
async def test_invalid_action(self, tool):
"""Test error handling for invalid action."""
result = await tool.run(action="invalid_action", file="test.py")
assert result.data["error"] is not None
assert "Invalid action" in result.data["error"]
@pytest.mark.asyncio
async def test_file_not_found(self, tool):
"""Test error handling for non-existent file."""
result = await tool.run(action="rename", file="/nonexistent/file.py", line=1, column=0, new_name="new")
assert "error" in result.data
assert "not found" in result.data["error"].lower()
@pytest.mark.asyncio
async def test_find_references_missing_args(self, tool, temp_python_file):
"""Test find_references without required arguments."""
result = await tool.run(action="find_references", file=temp_python_file)
assert result.data["success"] is False
assert "required" in result.data["errors"][0].lower()
@pytest.mark.asyncio
async def test_find_references(self, tool, temp_python_file):
"""Test finding references to a symbol."""
result = await tool.run(
action="find_references",
file=temp_python_file,
line=1,
column=4, # 'old_function'
)
assert result.data["success"] is True
assert result.data["action"] == "find_references"
assert len(result.data["changes"]) > 0
@pytest.mark.asyncio
async def test_rename_preview(self, tool, temp_python_file):
"""Test rename in preview mode."""
result = await tool.run(
action="rename",
file=temp_python_file,
line=1,
column=4, # 'old_function'
new_name="new_function",
preview=True,
)
assert result.data["success"] is True
assert result.data["action"] == "rename"
assert "preview" in result.data
assert len(result.data["preview"]) > 0
# Verify file wasn't modified
with open(temp_python_file, "r") as f:
content = f.read()
assert "old_function" in content
assert "new_function" not in content
@pytest.mark.asyncio
async def test_rename_apply(self, tool, temp_python_file):
"""Test rename with actual application."""
result = await tool.run(
action="rename",
file=temp_python_file,
line=1,
column=4, # 'old_function'
new_name="new_function",
preview=False,
)
assert result.data["success"] is True
assert result.data["action"] == "rename"
assert result.data["changes_applied"] > 0
# Verify file was modified
with open(temp_python_file, "r") as f:
content = f.read()
assert "new_function" in content
# All occurrences should be renamed
assert "old_function" not in content
@pytest.mark.asyncio
async def test_extract_function_missing_name(self, tool, temp_python_file):
"""Test extract_function without new_name."""
result = await tool.run(
action="extract_function",
file=temp_python_file,
start_line=2,
end_line=3,
)
assert result.data["success"] is False
assert "new_name" in result.data["errors"][0].lower()
@pytest.mark.asyncio
async def test_extract_function_preview(self, tool, temp_python_file):
"""Test extract_function in preview mode."""
result = await tool.run(
action="extract_function",
file=temp_python_file,
start_line=2,
end_line=3,
new_name="compute_result",
preview=True,
)
assert result.data["success"] is True
assert result.data["action"] == "extract_function"
assert "preview" in result.data
assert len(result.data["preview"]) > 0
@pytest.mark.asyncio
async def test_extract_variable_missing_args(self, tool, temp_python_file):
"""Test extract_variable without required arguments."""
result = await tool.run(
action="extract_variable",
file=temp_python_file,
)
assert result.data["success"] is False
@pytest.mark.asyncio
async def test_inline_missing_args(self, tool, temp_python_file):
"""Test inline without required arguments."""
result = await tool.run(
action="inline",
file=temp_python_file,
)
assert result.data["success"] is False
@pytest.mark.asyncio
async def test_move_missing_target(self, tool, temp_python_file):
"""Test move without target_file."""
result = await tool.run(
action="move",
file=temp_python_file,
line=1,
)
assert result.data["success"] is False
assert "target_file" in result.data["errors"][0].lower()
@pytest.mark.asyncio
async def test_organize_imports_python(self, tool):
"""Test organize imports for Python file."""
with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f:
f.write('''import os
from typing import List
import sys
from collections import defaultdict
import json
def main():
pass
''')
temp_path = f.name
try:
result = await tool.run(
action="organize_imports",
file=temp_path,
preview=False,
)
assert result.data["success"] is True
with open(temp_path, "r") as f:
content = f.read()
# Imports should be sorted
lines = content.split("\n")
import_lines = [l for l in lines if l.startswith("import ")]
from_lines = [l for l in lines if l.startswith("from ")]
# Check imports are sorted
assert import_lines == sorted(import_lines, key=str.lower)
assert from_lines == sorted(from_lines, key=str.lower)
finally:
if os.path.exists(temp_path):
os.unlink(temp_path)
class TestRefactorToolHelpers:
"""Tests for RefactorTool helper methods."""
@pytest.fixture
def tool(self):
"""Create a refactor tool instance."""
return RefactorTool()
def test_get_identifier_at_basic(self, tool):
"""Test getting identifier at position."""
line = "def my_function(arg1, arg2):"
assert tool._get_identifier_at(line, 4) == "my_function"
assert tool._get_identifier_at(line, 16) == "arg1"
assert tool._get_identifier_at(line, 23) == "arg2"
def test_get_identifier_at_edge_cases(self, tool):
"""Test edge cases for identifier detection."""
# Empty line
assert tool._get_identifier_at("", 0) is None
# Column out of range
assert tool._get_identifier_at("hello", 100) == "hello"
# No identifier at position
assert tool._get_identifier_at(" ", 2) is None
def test_get_indentation(self, tool):
"""Test indentation detection."""
assert tool._get_indentation("def foo():") == ""
assert tool._get_indentation(" return x") == " "
assert tool._get_indentation("\t\treturn y") == "\t\t"
assert tool._get_indentation("") == ""
def test_get_language(self, tool):
"""Test language detection from file extension."""
assert tool._get_language("test.py") == "python"
assert tool._get_language("test.js") == "javascript"
assert tool._get_language("test.ts") == "typescript"
assert tool._get_language("test.go") == "go"
assert tool._get_language("test.rs") == "rust"
assert tool._get_language("test.unknown") == "unknown"
def test_find_used_variables(self, tool):
"""Test finding used variables in code."""
code = "result = x + y"
vars = tool._find_used_variables(code, "python")
assert "x" in vars
assert "y" in vars
assert "result" in vars
def test_find_defined_variables_python(self, tool):
"""Test finding defined variables in Python."""
code = "x = 10\ny = 20"
vars = tool._find_defined_variables(code, "python")
assert "x" in vars
assert "y" in vars
def test_find_defined_variables_javascript(self, tool):
"""Test finding defined variables in JavaScript."""
code = "const x = 10;\nlet y = 20;"
vars = tool._find_defined_variables(code, "javascript")
assert "x" in vars
assert "y" in vars
def test_build_function_python(self, tool):
"""Test building Python function."""
func = tool._build_function("my_func", ["x", "y"], "return x + y", "python", "")
assert "def my_func(x, y):" in func
assert "return x + y" in func
def test_build_function_javascript(self, tool):
"""Test building JavaScript function."""
func = tool._build_function("myFunc", ["x", "y"], "return x + y;", "javascript", "")
assert "function myFunc(x, y)" in func
assert "return x + y;" in func
def test_build_variable_declaration_python(self, tool):
"""Test building Python variable declaration."""
decl = tool._build_variable_declaration("x", "10 + 20", "python", " ")
assert decl == " x = 10 + 20"
def test_build_variable_declaration_javascript(self, tool):
"""Test building JavaScript variable declaration."""
decl = tool._build_variable_declaration("x", "10 + 20", "javascript", " ")
assert decl == " const x = 10 + 20;"
# Test factory function
def test_create_refactor_tool():
"""Test the factory function."""
tool = create_refactor_tool()
assert isinstance(tool, RefactorTool)
assert tool.name == "refactor"
pkg/hanzo-mcp/tests/test_unified_search.py → pkg/hanzo-mcp/tests/test_search_tool.py
+32 -19
View File
@@ -1,4 +1,4 @@
"""Test unified search tool functionality."""
"""Test search tool functionality."""
import os
import asyncio
@@ -6,17 +6,17 @@ import tempfile
from pathlib import Path
import pytest
from hanzo_mcp.tools.search import create_find_tool, create_unified_search_tool
from hanzo_mcp.tools.search import create_find_tool, create_search_tool
@pytest.mark.asyncio
async def test_unified_search_basic():
"""Test basic unified search functionality."""
async def test_search_basic():
"""Test basic search functionality."""
# Create search tool
search_tool = create_unified_search_tool()
search_tool = create_search_tool()
# Test with current directory
result = await search_tool.run(pattern="UnifiedSearch", path=".", max_results_per_type=5)
result = await search_tool.run(pattern="SearchTool", path=".", max_results_per_type=5)
assert result.data is not None
assert "results" in result.data
@@ -29,15 +29,15 @@ async def test_unified_search_basic():
# Check statistics
stats = result.data["statistics"]
assert "query" in stats
assert stats["query"] == "UnifiedSearch"
assert stats["query"] == "SearchTool"
assert "search_types_used" in stats
assert "text" in stats["search_types_used"] # Should use text search
@pytest.mark.asyncio
async def test_unified_search_auto_detection():
async def test_search_auto_detection():
"""Test automatic search type detection."""
search_tool = create_unified_search_tool()
search_tool = create_search_tool()
# Test natural language query (should trigger vector search if available)
result = await search_tool.run(pattern="how does search work in this codebase", max_results_per_type=5)
@@ -55,9 +55,9 @@ async def test_unified_search_auto_detection():
@pytest.mark.asyncio
async def test_unified_search_code_patterns():
async def test_search_code_patterns():
"""Test code pattern search."""
search_tool = create_unified_search_tool()
search_tool = create_search_tool()
# Test AST pattern detection
result = await search_tool.run(pattern="class SearchResult", max_results_per_type=5)
@@ -74,9 +74,9 @@ async def test_unified_search_code_patterns():
@pytest.mark.asyncio
async def test_unified_search_with_files():
async def test_search_with_files():
"""Test file search integration."""
search_tool = create_unified_search_tool()
search_tool = create_search_tool()
# Test file search
result = await search_tool.run(pattern="*.py", search_files=True, max_results_per_type=10)
@@ -151,20 +151,20 @@ async def test_find_tool_fuzzy_search():
# Test fuzzy matching
result = await find_tool.run(
pattern="unifd", # Misspelled "unified"
pattern="srchtl", # Misspelled "search_tool"
fuzzy=True,
max_results=5,
)
assert result.data is not None
# Fuzzy search might find "unified_search.py"
# Fuzzy search might find "search_tool.py"
# But results depend on actual files in directory
@pytest.mark.asyncio
async def test_unified_search_pagination():
"""Test pagination in unified search."""
search_tool = create_unified_search_tool()
async def test_search_pagination():
"""Test pagination in search."""
search_tool = create_search_tool()
# First page
result1 = await search_tool.run(pattern="def", page_size=5, page=1)
@@ -195,8 +195,21 @@ async def test_unified_search_pagination():
assert files1 != files2 or len(files1) == 0 or len(files2) == 0
@pytest.mark.asyncio
async def test_search_vector_opt_in():
"""Test that vector search is opt-in and disabled by default."""
# Default: vector search disabled
search_tool = create_search_tool()
assert search_tool._enable_vector_index is False
assert search_tool._vector_initialized is False
# Explicitly enable vector search
search_tool_with_vector = create_search_tool(enable_vector_index=True)
assert search_tool_with_vector._enable_vector_index is True
if __name__ == "__main__":
# Run tests
asyncio.run(test_unified_search_basic())
asyncio.run(test_search_basic())
asyncio.run(test_find_tool_basic())
print("Basic tests passed!")