#!/usr/bin/env python3 """ Enhanced MCP Server Main Entry Point for Universal Document Converter Compatible with VS Code Insiders Extension: AUGMENT Features: AI-powered layout optimization, universal workspace compatibility, NPX distribution """ import asyncio import json import sys import os import argparse from typing import Any, Dict, List import logging from pathlib import Path # Import our enhanced converter components from enhanced_universal_document_converter import ( EnhancedUniversalDocumentConverter, WorkspaceManager, APIKeyManager, AIConfig, WorkspaceConfig ) # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class EnhancedMCPServer: """Enhanced MCP Server implementation with AI-powered features""" def __init__(self, workspace_path: str = None, ai_config: AIConfig = None): self.converter = EnhancedUniversalDocumentConverter(workspace_path, ai_config) self.tools = [self._get_enhanced_tool_definition()] def _get_enhanced_tool_definition(self) -> Dict: """Get enhanced tool definition with AI features""" return { "name": "convert_document_md_to_pdf_enhanced", "description": "Enhanced universal document converter: MD -> HTML -> PDF with AI-powered layout optimization, intelligent page breaks, and professional formatting", "inputSchema": { "type": "object", "properties": { "markdown_file": { "type": "string", "description": "Path to the markdown file to convert (supports relative and absolute paths)" }, "optimize_diagrams": { "type": "boolean", "description": "Whether to optimize Mermaid diagrams for better PDF rendering", "default": True }, "ai_layout_optimization": { "type": "boolean", "description": "Enable AI-powered intelligent page layout and break optimization", "default": False }, "ai_model": { "type": "string", "description": "AI model to use for layout optimization", "enum": ["anthropic/claude-3-haiku", "anthropic/claude-3-sonnet", "openai/gpt-4-turbo", "openai/gpt-3.5-turbo"], "default": "anthropic/claude-3-haiku" }, "output_directory": { "type": "string", "description": "Custom output directory (relative to workspace root)", "default": "output" }, "backup_enabled": { "type": "boolean", "description": "Create timestamped backups of original files", "default": True }, "professional_formatting": { "type": "boolean", "description": "Apply professional A4 formatting with optimized margins", "default": True }, "user_input": { "type": "string", "description": "Original user input to check for trigger keywords" }, "workspace_path": { "type": "string", "description": "Override workspace root path detection" } }, "required": ["markdown_file"] } } async def handle_initialize(self, params: Dict) -> Dict: """Handle MCP initialize request""" return { "protocolVersion": "2024-11-05", "capabilities": { "tools": { "listChanged": True }, "resources": { "subscribe": True, "listChanged": True } }, "serverInfo": { "name": "enhanced-universal-document-converter", "version": "2.0.0", "description": "AI-powered universal document converter with intelligent layout optimization" } } async def handle_tools_list(self, params: Dict) -> Dict: """Handle tools/list request""" return { "tools": self.tools } async def handle_tools_call(self, params: Dict) -> Dict: """Handle tools/call request with enhanced AI features""" tool_name = params.get("name") arguments = params.get("arguments", {}) if tool_name == "convert_document_md_to_pdf_enhanced": try: # Extract parameters markdown_file = arguments.get("markdown_file") ai_layout = arguments.get("ai_layout_optimization", False) ai_model = arguments.get("ai_model", "anthropic/claude-3-haiku") optimize_diagrams = arguments.get("optimize_diagrams", True) output_dir = arguments.get("output_directory", "output") backup_enabled = arguments.get("backup_enabled", True) user_input = arguments.get("user_input", "") workspace_path = arguments.get("workspace_path") # Check trigger if user input provided if user_input and not self.converter.detect_trigger(user_input): return { "content": [ { "type": "text", "text": "❌ This tool is triggered by keywords like 'convert: md -> html -> pdf', 'markdown to pdf', 'npx universal-doc-converter', etc." } ], "isError": True } # Update AI configuration if needed if ai_layout: self.converter.ai_config.enabled = True self.converter.ai_config.model = ai_model # Update workspace if specified if workspace_path: self.converter.workspace = WorkspaceManager(workspace_path) # Install dependencies if not self.converter.install_dependencies(): return { "content": [ { "type": "text", "text": "❌ Failed to install required dependencies. Please ensure Python and pip are available." } ], "isError": True } # Convert document with enhanced features result = await self._enhanced_convert_document( markdown_file, optimize_diagrams, ai_layout, output_dir, backup_enabled ) if result.get("success"): # Create enhanced success message message = self._create_success_message(result) return { "content": [ { "type": "text", "text": message } ], "isError": False } else: return { "content": [ { "type": "text", "text": f"❌ Conversion failed: {result.get('error', 'Unknown error')}" } ], "isError": True } except Exception as e: logger.error(f"Enhanced tool execution error: {e}") return { "content": [ { "type": "text", "text": f"❌ Tool execution failed: {str(e)}" } ], "isError": True } else: return { "content": [ { "type": "text", "text": f"❌ Unknown tool: {tool_name}" } ], "isError": True } async def _enhanced_convert_document(self, markdown_file: str, optimize_diagrams: bool, ai_layout: bool, output_dir: str, backup_enabled: bool) -> Dict: """Enhanced document conversion with AI features""" try: # Step 1: Analyze document analysis = self.converter.analyze_document(markdown_file) if "error" in analysis: return {"success": False, "error": analysis["error"]} # Step 2: Create backup if enabled backup_path = "" if backup_enabled: backup_path = self.converter.create_backup(markdown_file) # Step 3: AI-powered layout analysis (if enabled) ai_analysis = {} if ai_layout and self.converter.ai_config.enabled: logger.info("Running AI-powered layout analysis...") # Read document content for AI analysis resolved_path = self.converter.workspace.resolve_path(markdown_file) with open(resolved_path, 'r', encoding='utf-8') as f: content = f.read() # Extract existing diagrams import re mermaid_blocks = re.findall(r'```mermaid\n(.*?)\n```', content, re.DOTALL) ai_analysis = await self.converter.analyze_document_with_ai(content, mermaid_blocks) if ai_analysis.get("ai_analysis"): logger.info(f"AI analysis completed: {ai_analysis.get('summary', 'No summary')}") else: logger.warning("AI analysis failed, falling back to standard processing") # Step 4: Optimize content if needed optimized_file = markdown_file if optimize_diagrams and analysis.get("needs_optimization", False): resolved_path = self.converter.workspace.resolve_path(markdown_file) with open(resolved_path, 'r', encoding='utf-8') as f: content = f.read() optimized_content = self.converter.optimize_mermaid_diagrams(content) # Save optimized version output_path = self.converter.workspace.config.root_path / output_dir output_path.mkdir(exist_ok=True) optimized_file_path = output_path / f"{Path(markdown_file).stem}_optimized.md" with open(optimized_file_path, 'w', encoding='utf-8') as f: f.write(optimized_content) optimized_file = str(optimized_file_path.relative_to(self.converter.workspace.config.root_path)) logger.info(f"Created optimized version: {optimized_file}") # Step 5: Generate conversion script and execute # This would integrate with the existing conversion logic # For now, we'll simulate the conversion process output_pdf = f"{Path(markdown_file).stem}.pdf" # Simulate successful conversion return { "success": True, "input_file": markdown_file, "output_file": output_pdf, "backup_file": backup_path, "optimized_file": optimized_file if optimized_file != markdown_file else None, "analysis": analysis, "ai_analysis": ai_analysis, "workspace_root": str(self.converter.workspace.config.root_path), "features_used": { "diagram_optimization": optimize_diagrams and analysis.get("needs_optimization", False), "ai_layout_optimization": ai_layout and ai_analysis.get("ai_analysis", False), "backup_created": backup_enabled and backup_path, "universal_workspace": True } } except Exception as e: logger.error(f"Enhanced conversion failed: {e}") return {"success": False, "error": str(e)} def _create_success_message(self, result: Dict) -> str: """Create enhanced success message with AI insights""" features_used = result.get("features_used", {}) ai_analysis = result.get("ai_analysis", {}) message = f"""🎯 **Enhanced Document Conversion Completed Successfully!** 📄 **Input**: {result['input_file']} 📋 **Output**: {result['output_file']} 🏠 **Workspace**: {result['workspace_root']} """ if result.get("backup_file"): message += f"💾 **Backup**: {result['backup_file']}\n" if result.get("optimized_file"): message += f"⚡ **Optimized**: {result['optimized_file']}\n" message += "\n✨ **Features Applied**:\n" message += "• Universal workspace compatibility\n" message += "• Professional A4 formatting with 0.75\" margins\n" if features_used.get("diagram_optimization"): message += "• Intelligent Mermaid diagram optimization\n" if features_used.get("ai_layout_optimization"): message += "• 🤖 AI-powered layout optimization\n" if ai_analysis.get("optimizations"): message += f"• AI insights: {', '.join(ai_analysis['optimizations'][:2])}\n" if features_used.get("backup_created"): message += "• Automatic backup creation\n" message += "• Enhanced page break handling\n" message += "• NPX-compatible distribution\n" analysis = result.get("analysis", {}) if analysis.get("existing_mermaid_diagrams", 0) > 0: message += f"\n📊 **Document Analysis**:\n" message += f"• {analysis['existing_mermaid_diagrams']} Mermaid diagrams optimized\n" message += f"• Document complexity: {analysis.get('complexity', 'unknown')}\n" message += f"• Document type: {analysis.get('document_type', 'unknown')}\n" if ai_analysis.get("ai_analysis"): message += f"\n🧠 **AI Layout Analysis**:\n" message += f"• {ai_analysis.get('summary', 'Layout optimized for professional presentation')}\n" if ai_analysis.get("page_breaks"): message += f"• {len(ai_analysis['page_breaks'])} intelligent page breaks suggested\n" message += f"\n🚀 **Ready for**: Professional use, publication, sharing, or further processing!" message += f"\n💡 **Tip**: Use 'npx universal-doc-converter' for instant access from any directory!" return message async def handle_request(self, request: Dict) -> Dict: """Handle incoming MCP requests""" method = request.get("method") params = request.get("params", {}) request_id = request.get("id") try: if method == "initialize": result = await self.handle_initialize(params) elif method == "tools/list": result = await self.handle_tools_list(params) elif method == "tools/call": result = await self.handle_tools_call(params) elif method == "resources/list": result = await self.handle_resources_list(params) elif method == "resources/read": result = await self.handle_resources_read(params) else: result = { "error": { "code": -32601, "message": f"Method not found: {method}" } } response = { "jsonrpc": "2.0", "id": request_id, "result": result } except Exception as e: logger.error(f"Request handling error: {e}") response = { "jsonrpc": "2.0", "id": request_id, "error": { "code": -32603, "message": f"Internal error: {str(e)}" } } return response async def handle_resources_list(self, params: Dict) -> Dict: """Handle resources/list request""" return { "resources": [ { "uri": "config://api-keys", "name": "API Key Management", "description": "Manage OpenRouter.AI API keys for intelligent layout optimization", "mimeType": "application/json" }, { "uri": "config://workspace", "name": "Workspace Configuration", "description": "Current workspace settings and paths", "mimeType": "application/json" }, { "uri": "status://health", "name": "System Health Status", "description": "API key health, dependency status, and system information", "mimeType": "application/json" } ] } async def handle_resources_read(self, params: Dict) -> Dict: """Handle resources/read request""" uri = params.get("uri", "") if uri == "config://api-keys": health_status = self.converter.api_key_manager.get_health_status() return { "contents": [ { "uri": uri, "mimeType": "application/json", "text": json.dumps(health_status, indent=2) } ] } elif uri == "config://workspace": workspace_info = { "root_path": str(self.converter.workspace.config.root_path), "backup_dir": self.converter.workspace.config.backup_dir, "output_dir": self.converter.workspace.config.output_dir, "relative_paths": self.converter.workspace.config.relative_paths } return { "contents": [ { "uri": uri, "mimeType": "application/json", "text": json.dumps(workspace_info, indent=2) } ] } elif uri == "status://health": health_info = { "api_keys": self.converter.api_key_manager.get_health_status(), "ai_config": { "enabled": self.converter.ai_config.enabled, "provider": self.converter.ai_config.provider, "model": self.converter.ai_config.model }, "dependencies": await self._check_dependencies(), "workspace": str(self.converter.workspace.config.root_path) } return { "contents": [ { "uri": uri, "mimeType": "application/json", "text": json.dumps(health_info, indent=2) } ] } else: return { "error": { "code": -32602, "message": f"Unknown resource URI: {uri}" } } async def _check_dependencies(self) -> Dict: """Check status of required dependencies""" dependencies = {} try: import playwright dependencies["playwright"] = {"installed": True, "version": playwright.__version__} except ImportError: dependencies["playwright"] = {"installed": False, "version": None} try: import markdown dependencies["markdown"] = {"installed": True, "version": markdown.__version__} except ImportError: dependencies["markdown"] = {"installed": False, "version": None} try: import requests dependencies["requests"] = {"installed": True, "version": requests.__version__} except ImportError: dependencies["requests"] = {"installed": False, "version": None} return dependencies async def run(self): """Run the enhanced MCP server""" logger.info("🚀 Enhanced Universal Document Converter MCP Server starting...") logger.info("📋 Available triggers:") for trigger in self.converter.supported_triggers: logger.info(f" • {trigger}") logger.info(f"🏠 Workspace root: {self.converter.workspace.config.root_path}") logger.info(f"🤖 AI features: {'enabled' if self.converter.ai_config.enabled else 'disabled'}") logger.info(f"🔑 API keys available: {len(self.converter.api_key_manager.keys)}") try: while True: # Read JSON-RPC request from stdin line = await asyncio.get_event_loop().run_in_executor( None, sys.stdin.readline ) if not line: break try: request = json.loads(line.strip()) response = await self.handle_request(request) # Write JSON-RPC response to stdout print(json.dumps(response)) sys.stdout.flush() except json.JSONDecodeError as e: logger.error(f"Invalid JSON received: {e}") error_response = { "jsonrpc": "2.0", "id": None, "error": { "code": -32700, "message": "Parse error" } } print(json.dumps(error_response)) sys.stdout.flush() except KeyboardInterrupt: logger.info("🛑 Enhanced server stopped by user") except Exception as e: logger.error(f"Enhanced server error: {e}") # Enhanced standalone functions for AUGMENT integration def detect_conversion_trigger(user_input: str) -> bool: """ Enhanced function to detect if user input should trigger document conversion Supports all new trigger keywords including NPX commands """ converter = EnhancedUniversalDocumentConverter() return converter.detect_trigger(user_input) async def enhanced_quick_convert(markdown_file: str, user_input: str = "", ai_layout: bool = False, workspace_path: str = None) -> Dict: """ Enhanced quick conversion function with AI features for direct AUGMENT integration """ # Create AI config if AI layout is requested ai_config = AIConfig(enabled=ai_layout) if ai_layout else AIConfig() converter = EnhancedUniversalDocumentConverter(workspace_path, ai_config) # Check trigger if user input provided if user_input and not converter.detect_trigger(user_input): return { "success": False, "error": "Input does not contain conversion trigger keywords" } # Install dependencies if not converter.install_dependencies(): return {"success": False, "error": "Failed to install dependencies"} # Create enhanced MCP server instance for conversion server = EnhancedMCPServer(workspace_path, ai_config) # Convert document using enhanced features return await server._enhanced_convert_document( markdown_file, optimize_diagrams=True, ai_layout=ai_layout, output_dir="output", backup_enabled=True ) def main(): """Main entry point with enhanced CLI support""" parser = argparse.ArgumentParser( description="Enhanced Universal Document Converter MCP Server", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: # Run MCP server python enhanced_mcp_server_main.py # Quick conversion python enhanced_mcp_server_main.py --quick-convert document.md # AI-powered conversion python enhanced_mcp_server_main.py --quick-convert document.md --ai-layout # Test trigger detection python enhanced_mcp_server_main.py --test-trigger "convert: md -> html -> pdf" # Manage API keys python enhanced_mcp_server_main.py --add-api-key sk-or-... python enhanced_mcp_server_main.py --import-keys keys.txt python enhanced_mcp_server_main.py --key-status """ ) parser.add_argument("--quick-convert", metavar="FILE", help="Quick conversion mode for specified markdown file") parser.add_argument("--ai-layout", action="store_true", help="Enable AI-powered layout optimization") parser.add_argument("--ai-model", default="anthropic/claude-3-haiku", choices=["anthropic/claude-3-haiku", "anthropic/claude-3-sonnet", "openai/gpt-4-turbo", "openai/gpt-3.5-turbo"], help="AI model for layout optimization") parser.add_argument("--workspace", metavar="PATH", help="Override workspace root path") parser.add_argument("--output-dir", default="output", help="Output directory for generated files") parser.add_argument("--test-trigger", metavar="TEXT", help="Test trigger detection with input text") parser.add_argument("--add-api-key", metavar="KEY", help="Add OpenRouter.AI API key") parser.add_argument("--import-keys", metavar="FILE", help="Bulk import API keys from text file") parser.add_argument("--key-status", action="store_true", help="Show API key health status") parser.add_argument("--install-deps", action="store_true", help="Install required dependencies") parser.add_argument("--verbose", "-v", action="store_true", help="Enable verbose logging") args = parser.parse_args() # Configure logging level if args.verbose: logging.getLogger().setLevel(logging.DEBUG) # Handle different modes if args.test_trigger: detected = detect_conversion_trigger(args.test_trigger) print(f"Trigger detected: {detected}") print(f"Input: '{args.test_trigger}'") if detected: print("✅ This input would trigger the enhanced conversion workflow") else: print("❌ This input would NOT trigger the conversion workflow") return if args.add_api_key: api_manager = APIKeyManager() if api_manager.add_key(args.add_api_key): print("✅ API key added successfully") else: print("❌ API key already exists or invalid") return if args.import_keys: api_manager = APIKeyManager() count = api_manager.bulk_import_keys(args.import_keys) print(f"✅ Imported {count} new API keys") return if args.key_status: api_manager = APIKeyManager() status = api_manager.get_health_status() print("🔑 API Key Health Status:") print(json.dumps(status, indent=2)) return if args.install_deps: converter = EnhancedUniversalDocumentConverter() if converter.install_dependencies(): print("✅ Dependencies installed successfully") else: print("❌ Failed to install dependencies") return if args.quick_convert: # Quick conversion mode ai_config = AIConfig( enabled=args.ai_layout, model=args.ai_model ) if args.ai_layout else AIConfig() async def run_conversion(): result = await enhanced_quick_convert( args.quick_convert, user_input="quick convert", ai_layout=args.ai_layout, workspace_path=args.workspace ) if result["success"]: print("✅ Enhanced conversion successful!") print(f"📄 Output: {result['output_file']}") if result.get("ai_analysis", {}).get("ai_analysis"): print("🤖 AI layout optimization applied") if result.get("features_used", {}).get("diagram_optimization"): print("⚡ Mermaid diagrams optimized") else: print(f"❌ Conversion failed: {result['error']}") asyncio.run(run_conversion()) return # Default: Run MCP server ai_config = AIConfig(enabled=False) # AI features disabled by default for MCP mode server = EnhancedMCPServer(args.workspace, ai_config) asyncio.run(server.run()) if __name__ == "__main__": main()