Spaces:
Running
Running
File size: 7,587 Bytes
8755993 a3bdcf1 8755993 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 | """
MCP Client for interacting with Refactor MCP Server.
Provides async methods to call MCP tools from other parts of the application.
"""
import logging
from typing import List, Dict, Optional
from code_chatbot.mcp.mcp_server import RefactorMCPServer, SearchResult, RefactorResult, RefactorSuggestion
logger = logging.getLogger(__name__)
class MCPClient:
"""
Client for Refactor MCP server.
Provides a simple interface to call MCP tools.
"""
def __init__(self, workspace_root: str):
"""
Initialize MCP client.
Args:
workspace_root: Root directory of the codebase
"""
self.server = RefactorMCPServer(workspace_root)
logger.info(f"MCP Client initialized for workspace: {workspace_root}")
def search_code(
self,
pattern: str,
file_pattern: str = "**/*.py",
context_lines: int = 2,
is_regex: bool = True
) -> List[SearchResult]:
"""
Search for patterns in codebase.
Args:
pattern: Search pattern (regex or literal)
file_pattern: Glob pattern for files to search
context_lines: Number of context lines before/after match
is_regex: Whether pattern is regex
Returns:
List of search results
"""
try:
results = self.server.code_search(
pattern=pattern,
file_pattern=file_pattern,
context_lines=context_lines,
is_regex=is_regex
)
logger.info(f"Code search completed: {len(results)} results")
return results
except Exception as e:
logger.error(f"Code search failed: {e}")
return []
def refactor_code(
self,
search_pattern: str,
replace_pattern: str,
file_pattern: str = "**/*.py",
dry_run: bool = True,
is_regex: bool = True
) -> RefactorResult:
"""
Perform regex-based code refactoring.
Args:
search_pattern: Pattern to search for
replace_pattern: Replacement string (supports capture groups)
file_pattern: Glob pattern for files to process
dry_run: If True, only show what would change
is_regex: Whether pattern is regex
Returns:
RefactorResult with changes made or to be made
"""
try:
result = self.server.code_refactor(
search_pattern=search_pattern,
replace_pattern=replace_pattern,
file_pattern=file_pattern,
dry_run=dry_run,
is_regex=is_regex
)
logger.info(f"Refactoring {'preview' if dry_run else 'complete'}: "
f"{result.files_changed} files, {result.total_replacements} replacements")
return result
except Exception as e:
logger.error(f"Refactoring failed: {e}")
return RefactorResult(
files_changed=0,
total_replacements=0,
changes=[],
dry_run=dry_run,
success=False,
error=str(e)
)
def suggest_refactorings(
self,
file_path: str,
max_suggestions: int = 5
) -> List[RefactorSuggestion]:
"""
Analyze code and suggest refactorings.
Args:
file_path: Path to file to analyze
max_suggestions: Maximum number of suggestions
Returns:
List of refactoring suggestions
"""
try:
suggestions = self.server.suggest_refactorings(
file_path=file_path,
max_suggestions=max_suggestions
)
logger.info(f"Generated {len(suggestions)} refactoring suggestions for {file_path}")
return suggestions
except Exception as e:
logger.error(f"Suggestion generation failed: {e}")
return []
def format_search_results(self, results: List[SearchResult], max_results: int = 10) -> str:
"""
Format search results for display.
Args:
results: List of search results
max_results: Maximum number of results to format
Returns:
Formatted string
"""
if not results:
return "No results found."
output = [f"Found {len(results)} matches:\n"]
for i, result in enumerate(results[:max_results], 1):
output.append(f"\n{i}. {result.file_path}:{result.line_number}")
output.append(f" {result.line_content}")
if result.context_before:
output.append(f" Context before:")
for line in result.context_before[-2:]:
output.append(f" {line}")
if len(results) > max_results:
output.append(f"\n... and {len(results) - max_results} more results")
return '\n'.join(output)
def format_refactor_result(self, result: RefactorResult) -> str:
"""
Format refactor result for display.
Args:
result: Refactor result
Returns:
Formatted string
"""
if not result.success:
return f"❌ Refactoring failed: {result.error}"
mode = "Preview" if result.dry_run else "Applied"
output = [
f"✅ Refactoring {mode}:",
f" Files changed: {result.files_changed}",
f" Total replacements: {result.total_replacements}\n"
]
for change in result.changes[:5]:
output.append(f"\n📄 {change['file_path']}")
output.append(f" Replacements: {change['replacements']}")
if change.get('preview'):
output.append(f" Preview:")
for line in change['preview'].split('\n')[:6]:
output.append(f" {line}")
if len(result.changes) > 5:
output.append(f"\n... and {len(result.changes) - 5} more files")
return '\n'.join(output)
def format_suggestions(self, suggestions: List[RefactorSuggestion]) -> str:
"""
Format refactoring suggestions for display.
Args:
suggestions: List of suggestions
Returns:
Formatted string
"""
if not suggestions:
return "No refactoring suggestions found."
output = [f"💡 Found {len(suggestions)} refactoring suggestions:\n"]
for i, suggestion in enumerate(suggestions, 1):
impact_emoji = {'low': '🟢', 'medium': '🟡', 'high': '🔴'}
emoji = impact_emoji.get(suggestion.estimated_impact, '⚪')
output.append(f"\n{i}. {emoji} {suggestion.type.replace('_', ' ').title()}")
output.append(f" Location: {suggestion.file_path}:L{suggestion.line_start}-L{suggestion.line_end}")
output.append(f" Issue: {suggestion.description}")
output.append(f" Suggestion: {suggestion.rationale}")
return '\n'.join(output)
# Convenience function
def get_mcp_client(workspace_root: str = ".") -> MCPClient:
"""Get an MCP client instance."""
return MCPClient(workspace_root)
|