MCP Hub

Instructions

Implementation Reference

• app.py:510-516 (handler)

  Synchronous handler function that runs the async sandbox pool status checker using asyncio.run. This is the direct entry point matching the tool name pattern.
  def get_sandbox_pool_status_sync() -> Dict[str, Any]: """Synchronous wrapper for sandbox pool status.""" try: import asyncio return asyncio.run(get_sandbox_pool_status()) except Exception as e: return {"error": f"Failed to get sandbox pool status: {str(e)}"}
• app.py:475-508 (handler)

  Core async implementation that instantiates CodeRunnerAgent, fetches pool stats, computes readiness status, and formats the response with user-friendly messages.
  async def get_sandbox_pool_status() -> Dict[str, Any]: """Get sandbox pool status and statistics.""" try: # Create a temporary code runner to get pool stats code_runner = CodeRunnerAgent() stats = await code_runner.get_pool_stats() # Add warmup status information pool_size = stats.get("pool_size", 0) target_size = stats.get("target_pool_size", 0) if pool_size == 0: status_message = "🔄 Sandbox environment is warming up... This may take up to 2 minutes for the first execution." status = "warming_up" elif pool_size < target_size: status_message = f"⚡ Sandbox pool partially ready ({pool_size}/{target_size} sandboxes). More sandboxes warming up..." status = "partially_ready" else: status_message = f"✅ Sandbox pool fully ready ({pool_size}/{target_size} sandboxes available)" status = "ready" return { "status": status, "sandbox_pool": stats, "message": status_message, "user_message": status_message } except Exception as e: return { "status": "error", "error": f"Failed to get sandbox pool status: {str(e)}", "message": "Sandbox pool may not be initialized yet", "user_message": "🔄 Code execution environment is starting up... Please wait a moment." }
• app.py:1085-1090 (registration)

  Gradio interface registration exposing the handler as an API/MCP tool via api_name when launched with mcp_server=True.
  sandbox_btn.click( fn=get_sandbox_pool_status_sync, inputs=[], outputs=sandbox_output, api_name="get_sandbox_pool_status_service" )
• mcp_hub/agents/code_runner.py:88-92 (helper)

  Delegation method in CodeRunnerAgent that forwards the stats request to the underlying WarmSandboxPool.
  async def get_pool_stats(self): """Get sandbox pool statistics.""" if self.sandbox_pool: return self.sandbox_pool.get_stats() return {"error": "Pool not initialized"}
• mcp_hub/sandbox_pool.py:575-586 (helper)

  Fundamental stats computation in WarmSandboxPool class providing pool size, target size, health status, failure counts, and other metrics used throughout the chain.
  def get_stats(self) -> Dict[str, Any]: """Get pool statistics including health metrics.""" return { **self._stats, "pool_size": self._sandbox_queue.qsize(), "target_pool_size": self.pool_size, "running": self._running, "consecutive_failures": self._consecutive_failures, "last_successful_creation": self._last_successful_creation, "time_since_last_success": time.time() - self._last_successful_creation, "health_status": "healthy" if self._consecutive_failures < 3 else "degraded" if self._consecutive_failures < self._pool_reset_threshold else "critical" }