MCP Code Indexer

""" MCP响应格式化模块 负责将检索结果转换为符合MCP协议的格式 """ import os import json from typing import List, Dict, Any, Optional import logging from datetime import datetime logger = logging.getLogger(__name__) class McpFormatter: """ MCP响应格式化器类 将检索结果转换为符合MCP协议的格式，包含代码内容、位置、关系等信息 """ def __init__(self): """ 初始化MCP响应格式化器 Returns: 无返回值 """ logger.info("MCP响应格式化器初始化完成") def format_search_results(self, results: List[Dict[str, Any]], query: str, confidence_threshold: float = 0.7) -> Dict[str, Any]: """ 格式化搜索结果为MCP响应 Args: results: 搜索结果列表 query: 原始查询字符串 confidence_threshold: 置信度阈值，低于此值的结果将被标记为低置信度 Returns: MCP格式的响应字典 """ if not results: return self._create_empty_response(query) # 格式化代码块 code_blocks = [] for result in results: code_block = self._format_code_block(result, confidence_threshold) if code_block: code_blocks.append(code_block) # 创建MCP响应 response = { "mcp_version": "1.0", "timestamp": datetime.now().isoformat(), "query": query, "result_count": len(code_blocks), "code_blocks": code_blocks } return response def _format_code_block(self, result: Dict[str, Any], confidence_threshold: float) -> Optional[Dict[str, Any]]: """ 格式化单个代码块 Args: result: 代码块搜索结果 confidence_threshold: 置信度阈值 Returns: 格式化后的代码块字典，如果格式化失败则返回None """ try: # 提取基本信息 file_path = result.get('file_path', '') content = result.get('content', '') start_line = result.get('start_line', 1) end_line = result.get('end_line', 1) language = result.get('language', 'text') similarity = result.get('similarity', 0.0) # 计算置信度 confidence = similarity low_confidence = confidence < confidence_threshold # 创建代码块字典 code_block = { "id": self._generate_block_id(file_path, start_line, end_line), "file_path": file_path, "file_name": os.path.basename(file_path), "language": language, "start_line": start_line, "end_line": end_line, "content": content, "confidence": confidence, "low_confidence": low_confidence, "metadata": { "similarity": similarity, "type": result.get('type', 'code') } } return code_block except Exception as e: logger.error(f"格式化代码块失败: {str(e)}") return None def _generate_block_id(self, file_path: str, start_line: int, end_line: int) -> str: """ 生成代码块ID Args: file_path: 文件路径 start_line: 起始行号 end_line: 结束行号 Returns: 代码块ID字符串 """ file_name = os.path.basename(file_path) return f"{file_name}:{start_line}-{end_line}" def _create_empty_response(self, query: str) -> Dict[str, Any]: """ 创建空响应 Args: query: 原始查询字符串 Returns: 空的MCP响应字典 """ return { "mcp_version": "1.0", "timestamp": datetime.now().isoformat(), "query": query, "result_count": 0, "code_blocks": [], "message": "未找到结果" } def format_project_info(self, project_info: Dict[str, Any]) -> Dict[str, Any]: """ 格式化项目信息为MCP响应 Args: project_info: 项目信息字典 Returns: MCP格式的项目信息响应字典 """ return { "mcp_version": "1.0", "timestamp": datetime.now().isoformat(), "project_id": project_info.get("project_id"), "project_path": project_info.get("project_path"), "indexed_at": project_info.get("indexed_at"), "file_count": project_info.get("file_count"), "status": project_info.get("status"), "progress": project_info.get("progress", 0.0), "metadata": project_info.get("metadata", {}) } def format_code_context(self, code_context: Dict[str, Any], related_blocks: List[Dict[str, Any]] = None) -> Dict[str, Any]: """ 格式化代码上下文为MCP响应 Args: code_context: 代码上下文字典 related_blocks: 相关代码块列表 Returns: MCP格式的代码上下文响应字典 """ # 格式化主代码块 main_block = { "id": self._generate_block_id( code_context.get('file_path', ''), code_context.get('start_line', 1), code_context.get('end_line', 1) ), "file_path": code_context.get('file_path', ''), "file_name": os.path.basename(code_context.get('file_path', '')), "start_line": code_context.get('start_line', 1), "end_line": code_context.get('end_line', 1), "target_line": code_context.get('target_line', 1), "content": code_context.get('content', ''), "language": self._guess_language(code_context.get('file_path', '')) } # 格式化相关代码块 related_code_blocks = [] if related_blocks: for block in related_blocks: formatted_block = self._format_code_block(block, 0.0) if formatted_block: related_code_blocks.append(formatted_block) # 创建MCP响应 response = { "mcp_version": "1.0", "timestamp": datetime.now().isoformat(), "code_context": main_block, "related_blocks": related_code_blocks, "related_count": len(related_code_blocks) } return response def _guess_language(self, file_path: str) -> str: """ 根据文件扩展名猜测编程语言 Args: file_path: 文件路径 Returns: 编程语言字符串 """ ext_to_lang = { ".py": "python", ".js": "javascript", ".ts": "typescript", ".jsx": "javascript", ".tsx": "typescript", ".java": "java", ".c": "c", ".cpp": "cpp", ".h": "c", ".hpp": "cpp", ".cs": "csharp", ".go": "go", ".rb": "ruby", ".php": "php", ".swift": "swift", ".kt": "kotlin", ".rs": "rust", ".sh": "bash", ".html": "html", ".css": "css", ".scss": "scss", ".sql": "sql", ".md": "markdown", ".json": "json", ".xml": "xml", ".yaml": "yaml", ".yml": "yaml", ".toml": "toml" } _, ext = os.path.splitext(file_path.lower()) return ext_to_lang.get(ext, "text") def format_error(self, error_message: str, query: str = None) -> Dict[str, Any]: """ 格式化错误为MCP响应 Args: error_message: 错误消息 query: 原始查询字符串 Returns: MCP格式的错误响应字典 """ return { "mcp_version": "1.0", "timestamp": datetime.now().isoformat(), "query": query, "error": True, "error_message": error_message, "result_count": 0, "code_blocks": [] } def format_indexing_status(self, project_id: str, status: str, progress: float, message: str = None) -> Dict[str, Any]: """ 格式化索引状态为MCP响应 Args: project_id: 项目ID status: 索引状态 progress: 索引进度（0.0-1.0） message: 状态消息 Returns: MCP格式的索引状态响应字典 """ return { "mcp_version": "1.0", "timestamp": datetime.now().isoformat(), "project_id": project_id, "indexing_status": { "status": status, "progress": progress, "message": message or f"索引状态：{status} ({progress:.1%})" } }