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"
      }