"""Flask application for the Gradle MCP Dashboard with WebSocket support."""
import asyncio
import socket
import threading
from pathlib import Path
from typing import Any
from flask import Flask, jsonify, render_template, request
from flask_socketio import SocketIO, emit
from .daemon_monitor import DaemonMonitor
from . import log_store
# Global SocketIO instance for emitting events from other modules
_socketio: SocketIO | None = None
_app: Flask | None = None
def get_socketio() -> SocketIO | None:
"""Get the global SocketIO instance."""
return _socketio
def emit_log(log_entry: dict) -> None:
"""Emit a new log entry to all connected clients.
Args:
log_entry: The log entry dictionary to emit.
"""
if _socketio:
# Use namespace=None and broadcast from background thread
_socketio.emit("new_log", log_entry, namespace="/")
def emit_builds_changed() -> None:
"""Notify clients that active builds have changed."""
if _socketio:
# Get builds immediately to capture current state
builds = log_store.get_active_builds()
# Emit from a background thread to ensure it works from async context
def do_emit():
if _socketio:
_socketio.emit("builds_update", builds, namespace="/")
threading.Thread(target=do_emit, daemon=True).start()
def emit_daemons_changed() -> None:
"""Notify clients that daemon status may have changed."""
if _socketio and _app:
# Schedule a delayed refresh to allow daemon status to update
import time
def delayed_emit():
time.sleep(0.5) # Small delay to let gradle daemon status update
if _socketio and _app:
monitor: DaemonMonitor = _app.config.get("daemon_monitor")
if monitor:
daemons = monitor.get_daemons()
_socketio.emit("daemons_update", [d.to_dict() for d in daemons], namespace="/")
# Run in a separate thread to not block
import threading
threading.Thread(target=delayed_emit, daemon=True).start()
def find_available_port(start_port: int = 3333) -> int:
"""Find an available port starting from the given port.
Args:
start_port: Port number to start searching from.
Returns:
An available port number.
"""
port = start_port
max_attempts = 100
for _ in range(max_attempts):
try:
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.bind(("127.0.0.1", port))
return port
except OSError:
port += 1
raise RuntimeError(f"Could not find an available port after {max_attempts} attempts")
def create_app(daemon_monitor: DaemonMonitor) -> tuple[Flask, SocketIO]:
"""Create and configure the Flask application with SocketIO.
Args:
daemon_monitor: DaemonMonitor instance for daemon management.
Returns:
Tuple of (Flask application, SocketIO instance).
"""
global _socketio, _app
# Get the template and static directories relative to this file
template_dir = Path(__file__).parent / "templates"
static_dir = Path(__file__).parent / "static"
app = Flask(
__name__,
template_folder=str(template_dir),
static_folder=str(static_dir),
)
app.config["SECRET_KEY"] = "gradle-mcp-dashboard"
# Initialize SocketIO with threading mode - important for background emits
socketio = SocketIO(
app,
async_mode="threading",
cors_allowed_origins="*",
logger=False,
engineio_logger=False,
)
_socketio = socketio
_app = app
# Store monitor in app config for access in routes
app.config["daemon_monitor"] = daemon_monitor
@app.route("/")
def index() -> str:
"""Dashboard index page."""
return render_template("index.html")
@app.route("/logs/<daemon_id>")
def log_viewer(daemon_id: str) -> str:
"""Log viewer page for a specific daemon."""
return render_template("logs.html", daemon_id=daemon_id)
@app.route("/api/daemons")
def api_daemons() -> Any:
"""API endpoint to get list of daemons."""
monitor: DaemonMonitor = app.config["daemon_monitor"]
daemons = monitor.get_daemons()
return jsonify([d.to_dict() for d in daemons])
@app.route("/api/sessions")
def api_sessions() -> Any:
"""API endpoint to get active sessions."""
monitor: DaemonMonitor = app.config["daemon_monitor"]
sessions = monitor.get_active_sessions()
return jsonify([s.to_dict() for s in sessions])
@app.route("/api/builds")
def api_builds() -> Any:
"""API endpoint to get active builds."""
builds = log_store.get_active_builds()
return jsonify(builds)
@app.route("/api/logs/<daemon_pid>")
def api_logs(daemon_pid: str) -> Any:
"""API endpoint to get build logs."""
limit = request.args.get("limit", 500, type=int)
# Use the shared log_store module
logs = log_store.get_logs(limit=limit)
return jsonify(logs)
@app.route("/api/logs/<daemon_pid>/clear", methods=["POST"])
def api_clear_logs(daemon_pid: str) -> Any:
"""API endpoint to clear logs."""
log_store.clear_logs()
socketio.emit("logs_cleared", {})
return jsonify({"success": True})
@app.route("/api/daemons/stop", methods=["POST"])
def api_stop_all_daemons() -> Any:
"""API endpoint to stop all daemons."""
monitor: DaemonMonitor = app.config["daemon_monitor"]
try:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
result = loop.run_until_complete(monitor.stop_daemon())
loop.close()
socketio.emit("daemons_changed", {})
except Exception as e:
result = {"success": False, "error": str(e)}
return jsonify(result)
@app.route("/api/daemons/<pid>/stop", methods=["POST"])
def api_stop_daemon(pid: str) -> Any:
"""API endpoint to stop a specific daemon."""
monitor: DaemonMonitor = app.config["daemon_monitor"]
try:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
result = loop.run_until_complete(monitor.stop_daemon(pid))
loop.close()
socketio.emit("daemons_changed", {})
except Exception as e:
result = {"success": False, "error": str(e)}
return jsonify(result)
@app.route("/api/daemons/<pid>/details")
def api_daemon_details(pid: str) -> Any:
"""API endpoint to get detailed daemon configuration and status."""
monitor: DaemonMonitor = app.config["daemon_monitor"]
# Get daemon info
daemons = monitor.get_daemons()
daemon = next((d for d in daemons if d.pid == pid), None)
if not daemon:
return jsonify({"error": "Daemon not found"}), 404
# Get Gradle configuration (including runtime JVM args for this daemon)
from gradle_mcp.gradle import GradleWrapper
wrapper = GradleWrapper(str(monitor.project_root))
try:
config = wrapper.get_daemon_specific_config(pid)
except Exception:
# Fall back to basic config if daemon-specific extraction fails
config = wrapper.get_config()
config["runtime_jvm_args"] = []
# Get builds for this daemon
builds = log_store.get_active_builds()
daemon_builds = [b for b in builds if b.get("daemon_pid") == pid]
return jsonify({
"daemon": daemon.to_dict(),
"config": config,
"active_builds": daemon_builds
})
@app.route("/daemon/<daemon_pid>")
def daemon_details(daemon_pid: str) -> str:
"""Daemon details page for a specific daemon."""
return render_template("daemon_details.html", daemon_pid=daemon_pid)
# SocketIO event handlers
@socketio.on("connect")
def handle_connect() -> None:
"""Handle client connection - send initial data."""
monitor: DaemonMonitor = app.config["daemon_monitor"]
daemons = monitor.get_daemons()
emit("daemons_update", [d.to_dict() for d in daemons])
emit("logs_update", log_store.get_logs(limit=500))
emit("builds_update", log_store.get_active_builds())
@socketio.on("request_daemons")
def handle_request_daemons() -> None:
"""Handle request for daemon status."""
monitor: DaemonMonitor = app.config["daemon_monitor"]
daemons = monitor.get_daemons()
emit("daemons_update", [d.to_dict() for d in daemons])
@socketio.on("request_logs")
def handle_request_logs() -> None:
"""Handle request for logs."""
emit("logs_update", log_store.get_logs(limit=500))
return app, socketio
class DashboardServer:
"""Server for the Gradle MCP Dashboard with WebSocket support.
Runs the Flask-SocketIO application in a background thread.
"""
def __init__(self, daemon_monitor: DaemonMonitor) -> None:
"""Initialize the dashboard server.
Args:
daemon_monitor: DaemonMonitor instance for daemon management.
"""
self.daemon_monitor = daemon_monitor
self.app, self.socketio = create_app(daemon_monitor)
self._server_thread: threading.Thread | None = None
self._refresh_thread: threading.Thread | None = None
self._port: int | None = None
self._running = False
def start(self, port: int | None = None) -> str:
"""Start the dashboard server in a background thread.
Args:
port: Optional port to use. If None, finds an available port.
Returns:
URL where the dashboard is running.
"""
if self._running:
return f"http://127.0.0.1:{self._port}"
self._port = port or find_available_port()
def run_server() -> None:
"""Run the Flask-SocketIO server."""
import logging
# Disable verbose logging
log = logging.getLogger("werkzeug")
log.setLevel(logging.ERROR)
engineio_log = logging.getLogger("engineio")
engineio_log.setLevel(logging.ERROR)
socketio_log = logging.getLogger("socketio")
socketio_log.setLevel(logging.ERROR)
self.socketio.run(
self.app,
host="127.0.0.1",
port=self._port,
debug=False,
use_reloader=False,
allow_unsafe_werkzeug=True,
)
def refresh_daemon_status() -> None:
"""Periodically refresh daemon status and push to clients."""
import time
last_daemons = []
while self._running:
time.sleep(15) # Check every 15 seconds
if not self._running:
break
try:
daemons = self.daemon_monitor.get_daemons()
daemon_data = [d.to_dict() for d in daemons]
# Only emit if changed
if daemon_data != last_daemons:
last_daemons = daemon_data
self.socketio.emit("daemons_update", daemon_data, namespace="/")
except Exception:
pass
self._server_thread = threading.Thread(
target=run_server,
daemon=True,
name="gradle-mcp-dashboard",
)
self._server_thread.start()
# Start background daemon status refresh
self._refresh_thread = threading.Thread(
target=refresh_daemon_status,
daemon=True,
name="gradle-mcp-daemon-refresh",
)
self._refresh_thread.start()
self._running = True
return f"http://127.0.0.1:{self._port}"
def stop(self) -> None:
"""Stop the dashboard server."""
self._running = False
@property
def url(self) -> str | None:
"""Get the dashboard URL if running."""
if self._running and self._port:
return f"http://127.0.0.1:{self._port}"
return None
@property
def is_running(self) -> bool:
"""Check if the server is running."""
return self._running
