MCP Code Indexer

  • server
""" MCP协议能力接口模块 提供符合MCP协议的能力描述和接口 """ import os import logging from typing import Dict, Any, List, Optional from flask import Flask, jsonify, request, current_app logger = logging.getLogger(__name__) def setup_mcp_routes(app: Flask) -> None: """ 设置MCP协议路由 Args: app: Flask应用实例 Returns: 无返回值 """ # MCP协议版本 MCP_VERSION = "0.1" # MCP协议根路径 @app.route('/mcp', methods=['GET']) def mcp_root(): """ MCP协议根路径 Returns: MCP协议信息 """ return jsonify({ "name": "MCP Code Indexer", "version": MCP_VERSION, "description": "基于MCP协议的代码检索工具,为AI大语言模型提供高效、准确的代码库检索能力", "capabilities": [ "tools", "resources" ] }) # MCP协议能力接口 @app.route('/mcp/capabilities', methods=['GET']) def mcp_capabilities(): """ MCP协议能力接口 Returns: MCP协议能力描述 """ return jsonify({ "version": MCP_VERSION, "capabilities": { "tools": { "list": "/mcp/tools", "call": "/mcp/tools/call" }, "resources": { "list": "/mcp/resources", "templates": "/mcp/resource-templates", "read": "/mcp/resources/read" } } }) # MCP工具列表接口 @app.route('/mcp/tools', methods=['GET']) def mcp_tools(): """ MCP工具列表接口 Returns: MCP工具列表 """ return jsonify({ "tools": [ { "name": "identify_project", "description": "识别代码项目,返回项目ID和状态", "inputSchema": { "type": "object", "properties": { "project_path": { "type": "string", "description": "项目路径" } }, "required": ["project_path"] } }, { "name": "index_project", "description": "索引代码项目,生成向量索引", "inputSchema": { "type": "object", "properties": { "project_path": { "type": "string", "description": "项目路径" }, "wait": { "type": "boolean", "description": "是否等待索引完成", "default": false } }, "required": ["project_path"] } }, { "name": "search_code", "description": "搜索代码,返回相关代码片段", "inputSchema": { "type": "object", "properties": { "query": { "type": "string", "description": "搜索查询字符串" }, "project_id": { "type": "string", "description": "项目ID,如果不提供则搜索所有项目" }, "language": { "type": "string", "description": "编程语言过滤" }, "limit": { "type": "integer", "description": "返回结果数量限制", "default": 10 } }, "required": ["query"] } }, { "name": "get_project_status", "description": "获取项目索引状态", "inputSchema": { "type": "object", "properties": { "project_id": { "type": "string", "description": "项目ID" } }, "required": ["project_id"] } }, { "name": "get_projects", "description": "获取所有已索引的项目列表", "inputSchema": { "type": "object", "properties": {} } }, { "name": "get_code_context", "description": "获取代码上下文", "inputSchema": { "type": "object", "properties": { "file_path": { "type": "string", "description": "文件路径" }, "line_number": { "type": "integer", "description": "行号" }, "context_lines": { "type": "integer", "description": "上下文行数", "default": 10 } }, "required": ["file_path", "line_number"] } } ] }) # MCP工具调用接口 @app.route('/mcp/tools/call', methods=['POST']) def mcp_call_tool(): """ MCP工具调用接口 Returns: 工具调用结果 """ data = request.json if not data or 'name' not in data or 'arguments' not in data: return jsonify({ "error": "Invalid request", "message": "Missing name or arguments" }), 400 tool_name = data['name'] arguments = data['arguments'] # 获取组件 indexer = current_app.config['mcp_indexer'] search_engine = current_app.config['mcp_search_engine'] formatter = current_app.config['mcp_formatter'] try: # 根据工具名称调用相应的功能 if tool_name == "identify_project": if 'project_path' not in arguments: return jsonify({ "error": "Invalid arguments", "message": "Missing project_path" }), 400 project_id, is_new, metadata = indexer.project_identifier.identify_project(arguments['project_path']) status, progress = indexer.get_indexing_status(project_id) return jsonify({ "content": [ { "type": "text", "text": f"项目识别成功。项目ID: {project_id}, 状态: {status}, 进度: {progress:.1%}" } ] }) elif tool_name == "index_project": if 'project_path' not in arguments: return jsonify({ "error": "Invalid arguments", "message": "Missing project_path" }), 400 wait = arguments.get('wait', False) if wait: # 识别项目 project_id, _, _ = indexer.project_identifier.identify_project(arguments['project_path']) # 启动索引 indexer.index_project(arguments['project_path']) # 等待索引完成 import time max_wait = 300 # 最多等待5分钟 start_time = time.time() while time.time() - start_time < max_wait: status, progress = indexer.get_indexing_status(project_id) if status == "completed" or status == "failed": break time.sleep(2) return jsonify({ "content": [ { "type": "text", "text": f"项目索引{status}。项目ID: {project_id}, 进度: {progress:.1%}" } ] }) else: # 启动索引但不等待 project_id = indexer.index_project(arguments['project_path']) return jsonify({ "content": [ { "type": "text", "text": f"项目索引已启动。项目ID: {project_id}" } ] }) elif tool_name == "search_code": if 'query' not in arguments: return jsonify({ "error": "Invalid arguments", "message": "Missing query" }), 400 # 构建过滤条件 filters = {} if 'language' in arguments: filters['language'] = arguments['language'] # 构建项目ID列表 project_ids = [arguments['project_id']] if 'project_id' in arguments else None # 执行搜索 limit = arguments.get('limit', 10) results = search_engine.search(arguments['query'], project_ids, filters, limit) # 格式化结果 formatted_results = [] for i, result in enumerate(results): file_path = result.get('file_path', '') language = result.get('language', 'text') start_line = result.get('start_line', 1) end_line = result.get('end_line', 1) content = result.get('content', '') formatted_results.append(f"### {i+1}. {os.path.basename(file_path)} (行 {start_line}-{end_line})") formatted_results.append(f"文件: {file_path}") formatted_results.append(f"```{language}") formatted_results.append(content) formatted_results.append("```") formatted_results.append("") if not formatted_results: formatted_results = ["未找到匹配的代码。"] return jsonify({ "content": [ { "type": "text", "text": "\n".join(formatted_results) } ] }) elif tool_name == "get_project_status": if 'project_id' not in arguments: return jsonify({ "error": "Invalid arguments", "message": "Missing project_id" }), 400 status, progress = indexer.get_indexing_status(arguments['project_id']) return jsonify({ "content": [ { "type": "text", "text": f"项目状态: {status}, 进度: {progress:.1%}" } ] }) elif tool_name == "get_projects": projects = indexer.get_indexed_projects() if not projects: return jsonify({ "content": [ { "type": "text", "text": "没有已索引的项目。" } ] }) formatted_projects = ["已索引的项目:"] for i, project in enumerate(projects): project_id = project.get('project_id', '') project_path = project.get('project_path', '') status = project.get('status', '') formatted_projects.append(f"{i+1}. ID: {project_id}") formatted_projects.append(f" 路径: {project_path}") formatted_projects.append(f" 状态: {status}") formatted_projects.append("") return jsonify({ "content": [ { "type": "text", "text": "\n".join(formatted_projects) } ] }) elif tool_name == "get_code_context": if 'file_path' not in arguments or 'line_number' not in arguments: return jsonify({ "error": "Invalid arguments", "message": "Missing file_path or line_number" }), 400 context_lines = arguments.get('context_lines', 10) context = search_engine.get_code_context( arguments['file_path'], arguments['line_number'], context_lines ) file_path = context.get('file_path', '') start_line = context.get('start_line', 1) end_line = context.get('end_line', 1) target_line = context.get('target_line', 1) content = context.get('content', '') language = search_engine._guess_language(file_path) return jsonify({ "content": [ { "type": "text", "text": f"文件: {file_path} (行 {start_line}-{end_line}, 目标行: {target_line})\n\n```{language}\n{content}\n```" } ] }) else: return jsonify({ "error": "Unknown tool", "message": f"Tool not found: {tool_name}" }), 404 except Exception as e: logger.error(f"工具调用失败: {str(e)}") return jsonify({ "content": [ { "type": "text", "text": f"工具调用失败: {str(e)}" } ], "isError": True }) # MCP资源列表接口 @app.route('/mcp/resources', methods=['GET']) def mcp_resources(): """ MCP资源列表接口 Returns: MCP资源列表 """ # 获取组件 indexer = current_app.config['mcp_indexer'] # 获取所有项目 projects = indexer.get_indexed_projects() resources = [] for project in projects: project_id = project.get('project_id', '') project_path = project.get('project_path', '') resources.append({ "uri": f"code://{project_id}/info", "name": f"项目信息: {os.path.basename(project_path)}", "description": f"项目 {project_path} 的基本信息", "mimeType": "application/json" }) return jsonify({ "resources": resources }) # MCP资源模板接口 @app.route('/mcp/resource-templates', methods=['GET']) def mcp_resource_templates(): """ MCP资源模板接口 Returns: MCP资源模板列表 """ return jsonify({ "resourceTemplates": [ { "uriTemplate": "code://{project_id}/info", "name": "项目信息", "description": "获取项目的基本信息", "mimeType": "application/json" }, { "uriTemplate": "code://{project_id}/file/{file_path}", "name": "文件内容", "description": "获取项目中指定文件的内容", "mimeType": "text/plain" } ] }) # MCP资源读取接口 @app.route('/mcp/resources/read', methods=['POST']) def mcp_read_resource(): """ MCP资源读取接口 Returns: 资源内容 """ data = request.json if not data or 'uri' not in data: return jsonify({ "error": "Invalid request", "message": "Missing uri" }), 400 uri = data['uri'] # 获取组件 indexer = current_app.config['mcp_indexer'] search_engine = current_app.config['mcp_search_engine'] try: # 解析URI if uri.startswith("code://"): # 项目信息: code://{project_id}/info if uri.endswith("/info"): project_id = uri[7:-5] # 去掉"code://"和"/info" # 获取项目信息 project = indexer.project_identifier.get_project_by_id(project_id) if not project: return jsonify({ "error": "Resource not found", "message": f"Project not found: {project_id}" }), 404 return jsonify({ "contents": [ { "uri": uri, "mimeType": "application/json", "text": json.dumps(project, indent=2) } ] }) # 文件内容: code://{project_id}/file/{file_path} elif "/file/" in uri: parts = uri[7:].split("/file/", 1) if len(parts) != 2: return jsonify({ "error": "Invalid URI", "message": f"Invalid file URI: {uri}" }), 400 project_id, file_path = parts # 获取项目信息 project = indexer.project_identifier.get_project_by_id(project_id) if not project: return jsonify({ "error": "Resource not found", "message": f"Project not found: {project_id}" }), 404 # 构建完整文件路径 project_path = project.get("project_path", "") full_path = os.path.join(project_path, file_path) # 读取文件内容 try: with open(full_path, 'r', encoding='utf-8', errors='replace') as f: content = f.read() return jsonify({ "contents": [ { "uri": uri, "mimeType": "text/plain", "text": content } ] }) except Exception as e: return jsonify({ "error": "File read error", "message": f"Failed to read file: {str(e)}" }), 500 return jsonify({ "error": "Invalid URI", "message": f"Unsupported URI: {uri}" }), 400 except Exception as e: logger.error(f"资源读取失败: {str(e)}") return jsonify({ "error": "Resource read error", "message": str(e) }), 500 logger.info("MCP协议路由设置完成")