#!/usr/bin/env python3 """ NMP_Plus MCP Server - полная версия Поддержка 36 команд через MCP для интеграции с Cursor """ import asyncio import json import logging import sys import os from datetime import datetime from typing import Dict, List, Optional, Any import urllib.request import urllib.parse # Настраиваем логирование log_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'logs') os.makedirs(log_dir, exist_ok=True) logging.basicConfig( level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s', handlers=[ logging.FileHandler(os.path.join(log_dir, 'mcp_python.log')) ] ) class NMPMCPServer: """Полный MCP сервер для NMP_Plus с 36 командами""" def __init__(self): # Приоритетный список серверов для подключения self.server_hosts = [ "192.168.88.21", "localhost", "127.0.0.1" ] self.server_port = "8001" self.base_url = None self.connection_tested = False logging.info("🧠 NMP_Plus 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=5) as response: result = json.loads(response.read().decode('utf-8')) logging.info(f"Ответ от {url}: {result}") # Важно: проверяем success: true, а не status: ok if result and result.get('success') == 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 серверу") # Устанавливаем базовый URL на localhost в случае неудачи 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.debug(f"🔍 MCP REQUEST: {url}") # Специальные GET endpoints get_endpoints = ["api/status", "api/collections/list", "api/memory/stats"] if endpoint in get_endpoints and (data is None or not data): # GET запрос для статусных endpoint'ов req = urllib.request.Request(url) else: # POST запрос с JSON данными для остальных 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')) return result except Exception as e: logging.error(f"HTTP request failed: {str(e)}") return {"error": f"Server connection failed: {str(e)}"} # ============ CORE TOOLS ============ def nmp_status(self, params: Dict) -> Dict: """Статус системы NMP""" return self.make_request("api/status", None) def nmp_memory_save(self, params: Dict) -> Dict: """Сохранить информацию в память""" return self.make_request("api/memory/save", params) def nmp_memory_search(self, params: Dict) -> Dict: """Поиск в памяти""" return self.make_request("api/memory/search", params) def nmp_memory_list(self, params: Dict) -> Dict: """Список коллекций""" return self.make_request("api/collections/list", None) def nmp_memory_create(self, params: Dict) -> Dict: """Создать коллекцию""" return self.make_request("api/memory/create", params) def nmp_memory_delete(self, params: Dict) -> Dict: """Удалить коллекцию""" return self.make_request("api/memory/delete", params) def nmp_memory_stats(self, params: Dict) -> Dict: """Статистика системы""" return self.make_request("api/memory/stats", None) # ============ CODE TOOLS ============ def nmp_code_error_save(self, params: Dict) -> Dict: """Сохранить ошибку + решение""" return self.make_request("api/code/error/save", params) def nmp_code_pattern_save(self, params: Dict) -> Dict: """Сохранить паттерн кода""" return self.make_request("api/code/pattern/save", params) def nmp_refactor_track(self, params: Dict) -> Dict: """Отслеживание рефакторинга""" return self.make_request("api/refactor/track", params) def nmp_solution_pattern(self, params: Dict) -> Dict: """Паттерны решений""" return self.make_request("api/code/solution/pattern", params) def nmp_debug_journey(self, params: Dict) -> Dict: """История отладки""" return self.make_request("api/debug/journey", params) def nmp_code_snippet_save(self, params: Dict) -> Dict: """Сохранить полезный код""" return self.make_request("api/code/snippet/save", params) # ============ ARCHITECTURE TOOLS ============ def nmp_architecture_decision(self, params: Dict) -> Dict: """Архитектурные решения""" return self.make_request("api/architecture/decision", params) def nmp_dependency_map(self, params: Dict) -> Dict: """Карта зависимостей""" return self.make_request("api/dependency/map", params) def nmp_config_snapshot(self, params: Dict) -> Dict: """Снимок конфигурации""" return self.make_request("api/config/snapshot", params) def nmp_tech_choice_log(self, params: Dict) -> Dict: """Логирование выбора технологий""" return self.make_request("api/tech/choice/log", params) def nmp_design_pattern(self, params: Dict) -> Dict: """Паттерны проектирования""" return self.make_request("api/design/pattern", params) # ============ WORKFLOW TOOLS ============ def nmp_session_context(self, params: Dict) -> Dict: """Контекст рабочей сессии""" return self.make_request("api/session/context", params) def nmp_file_relations(self, params: Dict) -> Dict: """Связи между файлами""" return self.make_request("api/file/relations", params) def nmp_todo_extract(self, params: Dict) -> Dict: """Извлечение TODO""" return self.make_request("api/todo/extract", params) def nmp_hotspot_analysis(self, params: Dict) -> Dict: """Анализ горячих точек""" return self.make_request("api/hotspot/analysis", params) def nmp_productivity_track(self, params: Dict) -> Dict: """Отслеживание продуктивности""" return self.make_request("api/productivity/track", params) def nmp_task_completion(self, params: Dict) -> Dict: """Завершение задач""" return self.make_request("api/task/completion", params) # ============ GIT TOOLS ============ def nmp_git_commit_analyze(self, params: Dict) -> Dict: """Анализ коммита""" return self.make_request("api/git/commit/analyze", params) def nmp_git_pattern_extract(self, params: Dict) -> Dict: """Извлечение паттернов из истории""" return self.make_request("api/git/pattern/extract", params) def nmp_commit_wisdom(self, params: Dict) -> Dict: """Мудрость из коммитов""" return self.make_request("api/commit/wisdom", params) def nmp_branch_strategy(self, params: Dict) -> Dict: """Стратегии ветвления""" return self.make_request("api/branch/strategy", params) # ============ CHAT TOOLS ============ def nmp_chat_save(self, params: Dict) -> Dict: """Сохранить чат""" return self.make_request("api/chat/save", params) def nmp_chat_search(self, params: Dict) -> Dict: """Поиск по чатам""" return self.make_request("api/chat/search", params) def nmp_chat_global(self, params: Dict) -> Dict: """Глобальный поиск по всем чатам""" return self.make_request("api/chat/global", params) def nmp_learning_capture(self, params: Dict) -> Dict: """Захват обучающих моментов""" return self.make_request("api/learning/capture", params) def nmp_insight_save(self, params: Dict) -> Dict: """Сохранить инсайт""" return self.make_request("api/insight/save", params) def nmp_knowledge_link(self, params: Dict) -> Dict: """Связать знания""" return self.make_request("api/knowledge/link", params) # ============ ANALYTICS TOOLS ============ def nmp_pattern_analyze(self, params: Dict) -> Dict: """Анализ паттернов""" return self.make_request("api/pattern/analyze", params) def nmp_trend_detect(self, params: Dict) -> Dict: """Обнаружение трендов""" return self.make_request("api/trend/detect", params) def nmp_skill_progress(self, params: Dict) -> Dict: """Прогресс навыков""" return self.make_request("api/skill/progress", params) def nmp_recommendation(self, params: Dict) -> Dict: """Рекомендации""" return self.make_request("api/recommendation", params) # ============ WEB TOOLS ============ # ============ WORKFLOW ADVANCED TOOLS ============ def nmp_workflow_discover(self, params: Dict) -> Dict: """Анализ проекта и поиск похожих""" return self.make_request("api/workflow/discover", params) def nmp_workflow_plan(self, params: Dict) -> Dict: """Планирование с опытом""" return self.make_request("api/workflow/plan", params) # ============ MCP Protocol Handlers ============ def list_tools(self) -> Dict: """Список всех доступных инструментов для MCP""" logging.info("Предоставление списка инструментов для MCP") return { "tools": [ # CORE TOOLS (5) { "name": "mcp_nmp_status", "description": "NMP system status" }, { "name": "mcp_nmp_memory_save", "description": "Save information to NMP memory" }, { "name": "mcp_nmp_memory_search", "description": "Search in NMP memory" }, { "name": "mcp_nmp_memory_list", "description": "List NMP collections" }, { "name": "mcp_nmp_memory_stats", "description": "NMP system statistics" }, # CODE TOOLS (6) { "name": "mcp_nmp_code_error_save", "description": "Save code error + solution" }, { "name": "mcp_nmp_code_pattern_save", "description": "Save code pattern" }, { "name": "mcp_nmp_refactor_track", "description": "Track refactoring" }, { "name": "mcp_nmp_solution_pattern", "description": "Solution patterns" }, { "name": "mcp_nmp_debug_journey", "description": "Debug journey history" }, { "name": "mcp_nmp_code_snippet_save", "description": "Save useful code snippet" }, # ARCHITECTURE TOOLS (5) { "name": "mcp_nmp_architecture_decision", "description": "Architecture decisions" }, { "name": "mcp_nmp_dependency_map", "description": "Dependency mapping" }, { "name": "mcp_nmp_config_snapshot", "description": "Configuration snapshot" }, { "name": "mcp_nmp_tech_choice_log", "description": "Technology choice logging" }, { "name": "mcp_nmp_design_pattern", "description": "Design patterns" }, # WORKFLOW TOOLS (7) { "name": "mcp_nmp_workflow_discover", "description": "Discover project patterns" }, { "name": "mcp_nmp_session_context", "description": "Work session context" }, { "name": "mcp_nmp_file_relations", "description": "File relations" }, { "name": "mcp_nmp_todo_extract", "description": "Extract TODO items" }, { "name": "mcp_nmp_hotspot_analysis", "description": "Hotspot analysis" }, { "name": "mcp_nmp_productivity_track", "description": "Track productivity" }, { "name": "mcp_nmp_task_completion", "description": "Task completion" }, # GIT TOOLS (4) { "name": "mcp_nmp_git_commit_analyze", "description": "Analyze git commit" }, { "name": "mcp_nmp_git_pattern_extract", "description": "Extract patterns from git history" }, { "name": "mcp_nmp_commit_wisdom", "description": "Wisdom from commits" }, { "name": "mcp_nmp_branch_strategy", "description": "Branch strategies" }, # CHAT TOOLS (5) { "name": "mcp_nmp_chat_save", "description": "Save chat" }, { "name": "mcp_nmp_chat_search", "description": "Search in chats" }, { "name": "mcp_nmp_learning_capture", "description": "Capture learning moments" }, { "name": "mcp_nmp_insight_save", "description": "Save insight" }, { "name": "mcp_nmp_knowledge_link", "description": "Link knowledge" }, # ANALYTICS TOOLS (4) { "name": "mcp_nmp_pattern_analyze", "description": "Analyze patterns" }, { "name": "mcp_nmp_trend_detect", "description": "Detect trends" }, { "name": "mcp_nmp_skill_progress", "description": "Skill progress" }, { "name": "mcp_nmp_recommendation", "description": "Recommendations" } ] } def call_tool(self, name: str, arguments: Dict) -> Dict: """Вызов инструмента по имени и аргументам""" logging.info(f"Вызов инструмента: {name}") # Удаляем префикс mcp_ clean_name = name.replace("mcp_", "") # Находим метод по имени method = getattr(self, clean_name, None) if method: try: result = method(arguments) return { "content": [{ "type": "text", "text": json.dumps(result, ensure_ascii=False, indent=2) }] } except Exception as e: logging.error(f"Ошибка при вызове {clean_name}: {str(e)}") return { "content": [{ "type": "text", "text": f"Error: {str(e)}" }] } else: logging.error(f"Инструмент не найден: {clean_name}") return { "content": [{ "type": "text", "text": f"Error: Tool {clean_name} not found" }] } async def handle_request(self, request: Dict) -> Dict: """Обработка MCP запроса""" method = request.get('method') params = request.get('params', {}) logging.info(f"Получен MCP запрос: {method}") if method == 'initialize': return { "protocolVersion": "2024-11-05", "capabilities": { "tools": { "list": True, "call": True } }, "serverInfo": { "name": "nmp-plus-mcp", "version": "1.0.14" } } elif method == 'tools/list': return self.list_tools() elif method == 'tools/call': return self.call_tool(params.get('name', ''), params.get('arguments', {})) else: logging.warning(f"Неизвестный метод: {method}") return {"error": f"Unknown method: {method}"} async def run(self): """Запуск MCP сервера""" logging.info("🧠 NMP_Plus MCP Server запущен и ожидает запросы") while True: try: line = await asyncio.get_event_loop().run_in_executor(None, sys.stdin.readline) if not line: break request = json.loads(line) response = await self.handle_request(request) print(json.dumps({ "jsonrpc": "2.0", "id": request.get('id'), "result": response })) sys.stdout.flush() except Exception as e: logging.error(f"Ошибка обработки запроса: {str(e)}") print(json.dumps({ "jsonrpc": "2.0", "id": None, "error": {"code": -1, "message": str(e)} })) sys.stdout.flush() async def main(): """Основная функция""" logging.info("Запуск NMP_Plus MCP Server") server = NMPMCPServer() await server.run() if __name__ == "__main__": asyncio.run(main())