mkdocs-mcp-plugin

Pattern "(?:delay|sleep|timeout|interval)\s*[:=]\s*(?:[^;]*(?:secret|token|password|credential|key|env))" matched in source_code: "timeout=5) print("MkDocs server stopped gracefully.", file=sys.stderr) except subprocess.TimeoutExpired: # Force kill if it doesn't terminate print("MkDocs server did not stop gracefully, forcing shutdown...", file=sys.stderr) _mkdocs_process.kill() _mkdocs_process.wait(timeout=2) print("MkDocs server forcefully stopped.", file=sys.stderr) except Exception as e: print(f"Error stopping MkDocs server: {e}", file=sys.stderr) # Try to kill the process if it still exists try: if _mkdocs_process.poll() is None: _mkdocs_process.kill() except: pass finally: _mkdocs_process = None # Also ensure the thread is properly joined if _mkdocs_thread and _mkdocs_thread.is_alive(): try: _mkdocs_thread.join(timeout=1) except: pass finally: _mkdocs_thread = None def initialize_mkdocs_integration(): """Initialize MkDocs integration by finding config and starting server.""" global _mkdocs_config, _project_root, _mkdocs_thread # Find MkDocs config config_path = find_mkdocs_config() if not config_path: print( "Warning: No mkdocs.yml found. MkDocs serve will not be started.", file=sys.stderr, ) print("The MCP server will run in standalone mode.", file=sys.stderr) return False print(f"Found MkDocs config at: {config_path}", file=sys.stderr) _project_root = config_path.parent # Load config _mkdocs_config = load_mkdocs_config(config_path) site_name = _mkdocs_config.get("site_name", "MkDocs") print(f"Loaded MkDocs project: {site_name}", file=sys.stderr) # Check if MkDocs is installed try: result = subprocess.run( ["mkdocs", "--version"], capture_output=True, text=True, timeout=5 ) if result.returncode == 0: print(f"MkDocs version: {result.stdout.strip()}", file=sys.stderr) else: print( "Warning: MkDocs command failed. Server will not be started.", file=sys.stderr, ) return False except (subprocess.TimeoutExpired, FileNotFoundError) as e: print(f"Warning: MkDocs not found or not responding: {e}", file=sys.stderr) print("Install MkDocs with: pip install mkdocs", file=sys.stderr) return False # Start MkDocs serve port = int(os.getenv("MKDOCS_PORT", "8000")) _mkdocs_thread = start_mkdocs_serve(_project_root, port) # Wait a moment for server to start time.sleep(2) print("\n" + "=" * 60, file=sys.stderr) print("MkDocs RAG Server initialized successfully!", file=sys.stderr) print(f" - MkDocs site: http://localhost:{port}", file=sys.stderr) print(f" - Project root: {_project_root}", file=sys.stderr) print(f" - Docs directory: {_project_root / 'docs'}", file=sys.stderr) print("=" * 60 + "\n", file=sys.stderr) return True @mcp.tool def get_mkdocs_info() -> dict[str, Any]: """ Get information about the current MkDocs project. Returns: Information about the MkDocs configuration and server status """ global _mkdocs_config, _project_root, _mkdocs_process if not _mkdocs_config: return {"success": False, "error": "No MkDocs project loaded"} server_running = _mkdocs_process is not None and _mkdocs_process.poll() is None port = int(os.getenv("MKDOCS_PORT", "8000")) return { "success": True, "project_root": str(_project_root), "config_path": str(_project_root / "mkdocs.yml"), "docs_dir": str(_project_root / _mkdocs_config.get("docs_dir", "docs")), "site_name": _mkdocs_config.get("site_name", "MkDocs"), "site_url": _mkdocs_config.get("site_url", ""), "theme": _mkdocs_config.get("theme", {}), "plugins": list(_mkdocs_config.get("plugins", [])), "server_running": server_running, "server_url": f"http://localhost:{port}" if server_running else None, } @mcp.tool def restart_mkdocs_server(port: int | None = None) -> dict[str, Any]: """ Restart the MkDocs development server. Args: port: Port to run the server on (default: 8000 or MKDOCS_PORT env var) Returns: Status of the restart operation """ global _mkdocs_thread, _project_root if not _project_root: return {"success": False, "error": "No MkDocs project loaded"} # Stop existing server stop_mkdocs_serve() # Start new server if port is None: port = int(os.getenv("MKDOCS_PORT", "8000")) _mkdocs_thread = start_mkdocs_serve(_project_root, port) time.sleep(2) # Wait for server to start return { "success": True, "message": f"MkDocs server restarted on port {port}", "server_url": f"http://localhost:{port}", } # Cleanup function def cleanup(): """Clean up resources on exit.""" global _searcher, _mkdocs_process print("\nShutting down MkDocs RAG Server...", file=sys.stderr) # Stop MkDocs server stop_mkdocs_serve() # Clean up search index and Milvus connection if _searcher: try: # Close Milvus connection if it exists if hasattr(_searcher, 'milvus_connected') and _searcher.milvus_connected: try: if _searcher.collection: _searcher.collection.release() connections.disconnect("default") print("Milvus connection closed.", file=sys.stderr) except Exception as e: print(f"Error closing Milvus connection: {e}", file=sys.stderr) # Clean up temporary files _searcher.cleanup() print("Search index cleaned up.", file=sys.stderr) except Exception as e: print(f"Error during searcher cleanup: {e}", file=sys.stderr) finally: _searcher = None # Final check to ensure MkDocs process is terminated if _mkdocs_process: try: if _mkdocs_process.poll() is None: _mkdocs_process.kill() print("Ensured MkDocs process termination.", file=sys.stderr) except: pass print("Cleanup complete.", file=sys.stderr) # Register cleanup atexit.register(cleanup) # Handle signals for graceful shutdown def signal_handler(signum, frame): """Handle shutdown signals gracefully.""" print(f"\nReceived signal {signum}, initiating graceful shutdown...", file=sys.stderr) cleanup() sys.exit(0) # Register signal handlers for graceful shutdown signal.signal(signal.SIGINT, signal_handler) # Ctrl+C signal.signal(signal.SIGTERM, signal_handler) # Termination signal # On Windows, also handle SIGBREAK if hasattr(signal, 'SIGBREAK'): signal.signal(signal.SIGBREAK, signal_handler) def main(): """Main entry point for the MCP server.""" try: # Initialize MkDocs integration mkdocs_initialized = initialize_mkdocs_integration() # Adjust default docs_dir based on project if _project_root and _mkdocs_config: docs_dir = _mkdocs_config.get("docs_dir", "docs") # Update the default docs_dir for all tools default_docs_dir = str(_project_root / docs_dir) # Monkey-patch the default parameter for tools # This ensures tools use the correct docs directory import inspect for name, func in inspect.getmembers(sys.modules[__name__]): if hasattr(func, "__mcp_tool__"): sig = inspect.signature(func) params = sig.parameters if "docs_dir" in params: # Update default value new_params = [] for param_name, param in params.items(): if param_name == "docs_dir": new_param = param.replace(default=default_docs_dir) new_params.append(new_param) else: new_params.append(param) func.__signature__ = sig.replace(parameters=new_params) # Run the MCP server mcp.run() except KeyboardInterrupt: print("\nReceived key" (at position 34743)

Pattern "(?:read|write|modify|delete).*(?:file|path|directory)(?!.*(?:lock|mutex|semaphore|flock|atomic))" matched in source_code: "read_document_impl(file_path" (at position 18694)

Dependency "fastmcp@2.11.1" has known CVEs:

Pattern "(?:query|execute|select|find).*(?:return|respond|result|rows|data)(?!.*(?:sanitize|escape|encode|map|filter|select|pick))" matched in source_code: "query: str, max_result" (at position 12342)