""" Minecraft AI Agent Research — Monitoring Dashboard Gradio-based monitoring interface for the MC server. This also serves the API endpoints needed for bot discovery. """ import gradio as gr import subprocess import json import os import time import threading import requests import psutil # ── State ───────────────────────────────────────────────────────────────────── _state = { "mc_online": False, "ngrok_url": "", "ngrok_tcp": "", "players": [], "uptime": 0, "start_time": time.time(), "mc_logs": [], "tps": "N/A", } def get_ngrok_url(): """Fetch the ngrok tunnel URL from local API""" try: resp = requests.get("http://localhost:4040/api/tunnels", timeout=3) data = resp.json() for t in data.get("tunnels", []): url = t.get("public_url", "") if url.startswith("tcp://"): _state["ngrok_url"] = url _state["ngrok_tcp"] = url return url except: pass return _state.get("ngrok_url", "Not available") def get_mc_status(): """Check if MC server is running by reading latest log""" try: log_path = "/app/mc-server/logs/latest.log" if os.path.exists(log_path): with open(log_path, "r") as f: lines = f.readlines() _state["mc_logs"] = lines[-50:] if len(lines) > 50 else lines for line in lines: if "Done (" in line: _state["mc_online"] = True if "lost connection" in line or "Stopping server" in line: pass # don't mark offline for individual disconnects except: pass return _state["mc_online"] def parse_mc_address(tcp_url): """Parse tcp://host:port into host, port""" if not tcp_url or not tcp_url.startswith("tcp://"): return "N/A", "N/A" addr = tcp_url.replace("tcp://", "") parts = addr.rsplit(":", 1) if len(parts) == 2: return parts[0], parts[1] return addr, "25565" def check_server_status(): """Background status checker""" while True: get_mc_status() get_ngrok_url() _state["uptime"] = int(time.time() - _state["start_time"]) time.sleep(15) # Start background checker _status_thread = threading.Thread(target=check_server_status, daemon=True) _status_thread.start() # ── Gradio UI ───────────────────────────────────────────────────────────────── def refresh_status(): """Get current status for display""" get_mc_status() get_ngrok_url() mc_status = "🟢 Online" if _state["mc_online"] else "🔴 Offline" uptime_h = _state["uptime"] // 3600 uptime_m = (_state["uptime"] % 3600) // 60 ngrok_url = _state.get("ngrok_url", "Not available") host, port = parse_mc_address(ngrok_url) # Format connect info if ngrok_url != "Not available": connect_info = f"**Server Address:** `{host}:{port}`\n\nConnect with Minecraft 1.21.4 using this address to watch the AI agent play." else: connect_info = "Tunnel not available yet. Please wait..." # Resource usage try: mem = psutil.virtual_memory() cpu = psutil.cpu_percent(interval=0.5) resources = f"CPU: {cpu}% | RAM: {mem.percent}% ({mem.used // 1024 // 1024}MB / {mem.total // 1024 // 1024}MB)" except: resources = "N/A" return mc_status, f"{uptime_h}h {uptime_m}m", ngrok_url, connect_info, resources def get_logs(): """Get recent MC server logs""" get_mc_status() logs = _state.get("mc_logs", []) return "".join(logs[-30:]) def get_connect_info(): """Get connection info formatted for users""" get_ngrok_url() ngrok_url = _state.get("ngrok_url", "") if not ngrok_url: return "Server tunnel not ready yet. Check back in a minute." host, port = parse_mc_address(ngrok_url) return f"""## Connect to Watch **Minecraft Java Edition 1.21.4** Server: `{host}` Port: `{port}` Add this as a server in your Minecraft multiplayer menu to watch the AI agent play in real time. The agent (Zelin_) connects autonomously and plays survival mode — exploring, mining, building, and fighting mobs.""" # ── Build Gradio Interface ──────────────────────────────────────────────────── with gr.Blocks( title="Minecraft AI Agent Research", theme=gr.themes.Soft(), css=""" .status-card { padding: 20px; border-radius: 10px; margin: 10px 0; } .connect-box { background: #1a1a2e; color: #e0e0e0; padding: 20px; border-radius: 10px; font-family: monospace; } """ ) as demo: gr.Markdown(""" # 🤖 Minecraft AI Agent Research Autonomous AI agent navigating and interacting in a Minecraft world. """) with gr.Row(): with gr.Column(scale=1): gr.Markdown("### Server Status") status_text = gr.Textbox(label="MC Server", value="Checking...", interactive=False) uptime_text = gr.Textbox(label="Uptime", value="0h 0m", interactive=False) tunnel_text = gr.Textbox(label="Tunnel URL", value="Connecting...", interactive=False) resources_text = gr.Textbox(label="Resources", value="N/A", interactive=False) refresh_btn = gr.Button("🔄 Refresh", variant="primary") with gr.Column(scale=1): connect_md = gr.Markdown(value="Loading connection info...") refresh_btn2 = gr.Button("📋 Get Connection Info") with gr.Row(): with gr.Column(): gr.Markdown("### Recent Logs") logs_text = gr.Textbox(label="Server Logs", value="", lines=15, interactive=False, max_lines=15) logs_btn = gr.Button("📜 Refresh Logs") # Wire up buttons refresh_btn.click( fn=refresh_status, outputs=[status_text, uptime_text, tunnel_text, connect_md, resources_text] ) refresh_btn2.click(fn=get_connect_info, outputs=[connect_md]) logs_btn.click(fn=get_logs, outputs=[logs_text]) # Auto-refresh every 30s demo.load(fn=refresh_status, outputs=[status_text, uptime_text, tunnel_text, connect_md, resources_text]) # ── API Endpoints (via Gradio mounting) ─────────────────────────────────────── # We need to add custom API routes for the bot to discover the server import http.server import socketserver from functools import partial class APIHandler(http.server.BaseHTTPRequestHandler): """Simple API server for bot discovery""" def do_GET(self): if self.path == "/api/ngrok": self.send_json({"url": _state.get("ngrok_url", ""), "tcp": _state.get("ngrok_tcp", "")}) elif self.path == "/api/status": self.send_json({ "mc_server": "online" if _state.get("mc_online") else "offline", "ngrok_url": _state.get("ngrok_url", ""), "uptime": _state.get("uptime", 0), }) elif self.path == "/health": self.send_json({"status": "ok", "mc": _state.get("mc_online", False)}) else: self.send_response(404) self.end_headers() def send_json(self, data): self.send_response(200) self.send_header("Content-Type", "application/json") self.send_header("Access-Control-Allow-Origin", "*") self.end_headers() self.wfile.write(json.dumps(data).encode()) def log_message(self, format, *args): pass # Suppress API logs # Start API server on a different port (7861) — Gradio on 7860 def run_api(): with socketserver.TCPServer(("", 7861), APIHandler) as httpd: httpd.serve_forever() api_thread = threading.Thread(target=run_api, daemon=True) api_thread.start() # Mount API handler on the main port too via middleware # We'll use a custom Gradio launch with the API routes # Actually, let's use a simpler approach: patch Gradio's FastAPI app original_launch = demo.launch def patched_launch(*args, **kwargs): # Launch Gradio first result = original_launch(*args, **kwargs) # Add API routes to Gradio's FastAPI app try: from fastapi import FastAPI app = demo.app # Gradio creates a FastAPI app @app.get("/api/ngrok") async def api_ngrok(): get_ngrok_url() return {"url": _state.get("ngrok_url", ""), "tcp": _state.get("ngrok_tcp", "")} @app.get("/api/status") async def api_status(): return { "mc_server": "online" if _state.get("mc_online") else "offline", "ngrok_url": _state.get("ngrok_url", ""), "uptime": _state.get("uptime", 0), } @app.get("/health") async def health(): return {"status": "ok", "mc": _state.get("mc_online", False)} print("[API] Routes mounted on Gradio FastAPI app") except Exception as e: print(f"[API] Could not mount routes: {e}") return result demo.launch = patched_launch # Launch if __name__ == "__main__": demo.launch( server_name="0.0.0.0", server_port=7860, show_error=True, quiet=True, )