#!/usr/bin/env python3 """ NMP_Plus MCP Server - Рефакторенная версия Настоящий MCP сервер для Cursor (stdin/stdout протокол) """ import asyncio import json import logging import os import sys import traceback import urllib.request import urllib.parse from datetime import datetime from typing import Dict, List, Any, Optional, Callable from pathlib import Path # Добавляем путь к модулям sys.path.append(str(Path(__file__).parent)) # Импорт обработчиков команд from mcp_handlers.core_handlers import CoreHandlers from mcp_handlers.code_handlers import CodeHandlers from mcp_handlers.architecture_handlers import ArchitectureHandlers from mcp_handlers.workflow_handlers import WorkflowHandlers from mcp_handlers.git_handlers import GitHandlers from mcp_handlers.context_handlers import ContextHandlers # Настраиваем логирование LOG_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "Logwork", "mcp") os.makedirs(LOG_DIR, exist_ok=True) logging.basicConfig(level=logging.DEBUG) # MCP Protocol Implementation class MCPServer: """Настоящий MCP сервер для Cursor""" def __init__(self): self.tools = MCPTools() async def handle_request(self, request: Dict) -> Optional[Dict]: """Обработка MCP запроса в формате JSON-RPC 2.0""" try: # Проверяем обязательные поля JSON-RPC 2.0 if request.get("jsonrpc") != "2.0": return { "jsonrpc": "2.0", "id": request.get("id"), "error": { "code": -32600, "message": "Invalid Request" } } method = request.get("method") params = request.get("params", {}) request_id = request.get("id") # Если нет id, это уведомление - не отправляем ответ if request_id is None: return None if method == "initialize": return { "jsonrpc": "2.0", "id": request_id, "result": { "protocolVersion": "2024-11-05", "capabilities": { "tools": {} }, "serverInfo": { "name": "nmp-plus-mcp", "version": "2.4.1" } } } elif method == "tools/list": return { "jsonrpc": "2.0", "id": request_id, "result": { "tools": self.tools.get_tools_list() } } elif method == "tools/call": tool_name = params.get("name") arguments = params.get("arguments", {}) result = await self.tools.call_tool(tool_name, arguments) # ✅ ИСПРАВЛЕНИЕ: правильная обработка MCP формата ответов if result.get("isError"): # Это ошибка из call_tool в MCP формате error_text = "" if "content" in result and len(result["content"]) > 0: error_text = result["content"][0].get("text", "Tool execution failed") return { "jsonrpc": "2.0", "id": request_id, "error": { "code": -32000, "message": error_text } } else: # Успешный результат в MCP формате return { "jsonrpc": "2.0", "id": request_id, "result": result } else: return { "jsonrpc": "2.0", "id": request_id, "error": { "code": -32601, "message": f"Method not found: {method}" } } except Exception as e: logging.error(f"MCP Request Error: {str(e)}\n{traceback.format_exc()}") return { "jsonrpc": "2.0", "id": request.get("id"), "error": { "code": -32603, "message": "Internal error", "data": str(e) } } async def run(self): """Запуск MCP сервера""" logging.info("🚀 Запуск NMP_Plus MCP сервера...") while True: try: # Читаем запрос из stdin line = await asyncio.get_event_loop().run_in_executor(None, sys.stdin.readline) # ИСПРАВЛЕНИЕ: проверяем не только пустую строку, но и EOF if not line or line == '': logging.warning("⚠️ Получена пустая строка из stdin, продолжаем ожидание...") await asyncio.sleep(0.1) # Небольшая пауза continue line = line.strip() if not line: # Пропускаем пустые строки continue request = json.loads(line) response = await self.handle_request(request) # Отправляем ответ в stdout только если это не уведомление if response is not None: print(json.dumps(response)) sys.stdout.flush() except json.JSONDecodeError as e: logging.error(f"JSON Parse Error: {str(e)}") error_response = { "jsonrpc": "2.0", "id": None, "error": { "code": -32700, "message": "Parse error", "data": str(e) } } print(json.dumps(error_response)) sys.stdout.flush() except EOFError: logging.info("📡 EOF получен, но продолжаем работу...") await asyncio.sleep(0.1) continue except Exception as e: logging.error(f"MCP Server Error: {str(e)}") # Для других ошибок не отправляем ответ, так как не знаем id await asyncio.sleep(0.1) # Пауза перед следующей итерацией # Класс инструментов MCP class MCPTools: """Обработчик инструментов MCP""" def __init__(self): """Инициализация обработчика инструментов""" self.tools = {} # Настройка HTTP API подключения self.server_hosts = [ "localhost", # ОСНОВНОЙ локальный сервер "127.0.0.1", # Альтернативный локальный "192.168.88.21" # Резервный сетевой сервер NMP_Plus ] self.server_port = "8001" self.base_url = None self.connection_tested = False # Инициализация обработчиков команд self.core_handlers = CoreHandlers(self) self.code_handlers = CodeHandlers(self) self.architecture_handlers = ArchitectureHandlers(self) self.workflow_handlers = WorkflowHandlers(self) self.git_handlers = GitHandlers(self) self.context_handlers = ContextHandlers(self) # Регистрация всех инструментов self._register_all_tools() logging.info(f"Зарегистрировано {len(self.tools)} инструментов MCP") def test_connection(self) -> bool: """Тестирует соединение с разными API серверами""" if self.connection_tested: return True logging.info("🔍 Тестирование соединения с API серверами...") for host in self.server_hosts: try: url = f"http://{host}:{self.server_port}/api/status" logging.info(f"Попытка подключения к {url}") req = urllib.request.Request(url) with urllib.request.urlopen(req, timeout=10) as response: result = json.loads(response.read().decode('utf-8')) logging.info(f"Ответ от {url}: {result}") if result and result.get('success') is True: self.base_url = f"http://{host}:{self.server_port}" logging.info(f"✅ Соединение установлено с {self.base_url}") self.connection_tested = True return True except Exception as e: logging.warning(f"Не удалось подключиться к {host}: {str(e)}") logging.error("❌ Не удалось подключиться ни к одному API серверу") self.base_url = f"http://{self.server_hosts[0]}:{self.server_port}" return False def make_request(self, endpoint: str, data: Dict = None) -> Dict: """HTTP запрос к API серверу""" if not self.connection_tested: self.test_connection() try: url = f"{self.base_url}/{endpoint}" logging.info(f"🔍 Запрос к API: {url}") logging.info(f"🔍 Данные запроса: {data}") if endpoint == "api/status": req = urllib.request.Request(url) else: if data is None: data = {} json_data = json.dumps(data).encode('utf-8') req = urllib.request.Request( url, data=json_data, headers={'Content-Type': 'application/json'} ) with urllib.request.urlopen(req, timeout=10) as response: result = json.loads(response.read().decode('utf-8')) logging.info(f"🔍 Ответ от API: success={result.get('success')}") logging.info(f"🔍 Ключи ответа: {list(result.keys())}") if result.get('success') and 'data' in result: data_keys = list(result['data'].keys()) if isinstance(result['data'], dict) else 'NOT_DICT' logging.info(f"🔍 Ключи data: {data_keys}") # Детальная проверка для context команд if endpoint.startswith('api/context/'): logging.info(f"🔍 Context данные: {str(result['data'])[:300]}...") return result except Exception as e: logging.error(f"HTTP request failed: {str(e)}") return {"error": f"Server connection failed: {str(e)}"} def _register_all_tools(self): """Регистрация всех инструментов из обработчиков""" # Каждый обработчик регистрирует свои инструменты self.core_handlers.register_tools() self.code_handlers.register_tools() self.architecture_handlers.register_tools() self.workflow_handlers.register_tools() self.git_handlers.register_tools() self.context_handlers.register_tools() def register_tool(self, name: str, description: str, function: Callable, input_schema: Dict = None): """Регистрация нового инструмента""" self.tools[name] = { "name": name, "description": description, "function": function, "input_schema": input_schema } def get_tools_list(self) -> List[Dict]: """Получение списка всех инструментов в формате MCP""" tools_list = [] for name, tool in self.tools.items(): tool_info = { "name": name, "description": tool["description"], "inputSchema": tool.get("input_schema", { "type": "object", "properties": {}, "required": [] }) } tools_list.append(tool_info) return tools_list async def call_tool(self, name: str, arguments: Dict) -> Dict: """Вызов инструмента по имени""" logging.info(f"🔧 MCP call_tool: {name} with args: {arguments}") # ИСПРАВЛЕНИЕ: Обработка дублированных префиксов от Cursor original_name = name if name.startswith("mcp_nmp_mcp_nmp_"): # Тройное дублирование: mcp_nmp_mcp_nmp_ name = name.replace("mcp_nmp_mcp_nmp_", "mcp_nmp_") logging.info(f"🔧 Исправлен тройной префикс: {original_name} -> {name}") elif name.startswith("mcp_mcp_nmp_"): # Cursor добавил mcp_ к уже существующему mcp_nmp_ name = name.replace("mcp_mcp_nmp_", "mcp_nmp_") logging.info(f"🔧 Исправлен дублированный префикс: {original_name} -> {name}") elif name.startswith("mcp_nmp_nmp_"): # Другой вариант дублирования name = name.replace("mcp_nmp_nmp_", "mcp_nmp_") logging.info(f"🔧 Исправлен дублированный префикс: {original_name} -> {name}") if name not in self.tools: return { "content": [ { "type": "text", "text": f"Ошибка: Инструмент '{name}' не найден" } ], "isError": True } try: tool = self.tools[name] result = await tool["function"](**arguments) logging.info(f"🔧 MCP call_tool result type: {type(result)}") logging.info(f"🔧 MCP call_tool result length: {len(str(result))}") logging.info(f"🔧 MCP call_tool result preview: {str(result)[:200]}...") # ИСПРАВЛЕНИЕ: Обработчики уже возвращают отформатированные строки if isinstance(result, str): # Это уже отформатированная строка от ResponseFormatter mcp_result = { "content": [ { "type": "text", "text": result } ] } logging.info(f"🔧 MCP call_tool возвращает строку длиной: {len(result)}") return mcp_result elif isinstance(result, dict): # Если это HTTP ответ с success/error (не обработанный ResponseFormatter) if "success" in result: if result.get("success"): # Успешный ответ - извлекаем данные data = result.get("data", result.get("result", result)) # Форматируем для MCP if isinstance(data, (dict, list)): import json text_content = json.dumps(data, ensure_ascii=False, indent=2) else: text_content = str(data) mcp_result = { "content": [ { "type": "text", "text": text_content } ] } else: # Ошибка в HTTP ответе error_msg = result.get("error", "Unknown error") return { "content": [ { "type": "text", "text": f"Ошибка: {error_msg}" } ], "isError": True } else: # Обычный словарь - конвертируем в JSON import json text_content = json.dumps(result, ensure_ascii=False, indent=2) mcp_result = { "content": [ { "type": "text", "text": text_content } ] } elif result is None: mcp_result = { "content": [ { "type": "text", "text": "Success" } ] } else: # Любой другой тип - конвертируем в строку mcp_result = { "content": [ { "type": "text", "text": str(result) } ] } return mcp_result except Exception as e: logging.error(f"Ошибка вызова инструмента {name}: {str(e)}\n{traceback.format_exc()}") return { "content": [ { "type": "text", "text": f"Ошибка выполнения: {str(e)}" } ], "isError": True } if __name__ == "__main__": # Запуск настоящего MCP сервера server = MCPServer() asyncio.run(server.run())