Agent Knowledge MCP

Instructions

Implementation Reference

• src/admin/admin_server.py:459-546 (handler)

  The main handler function for the 'elasticsearch_status' tool. It creates an ElasticsearchSetup instance, calls get_container_status(), and formats a comprehensive status report for Elasticsearch and Kibana Docker containers including existence, running status, URLs, configuration, and overall system readiness.

  @app.tool(
      description="Check status of Elasticsearch and Kibana containers with detailed configuration information",
      tags={"admin", "elasticsearch", "status", "docker", "monitoring"}
  )
  async def elasticsearch_status() -> str:
      """Check comprehensive status of Elasticsearch and Kibana Docker containers."""
      try:
          # Get config path and setup manager
          config_path = Path(__file__).parent / "config.json"
          setup_manager = ElasticsearchSetup(config_path)
  
          # Get detailed container status
          status = setup_manager.get_container_status()
  
          if "error" in status:
              return f"❌ **Container Status Check Failed!**\n\n🚨 **Error:** {status['error']}\n\n💡 **Troubleshooting:**\n   • Check if Docker is running\n   • Verify Docker daemon is accessible\n   • Ensure proper Docker permissions\n   • Try restarting Docker service"
  
          # Build comprehensive status message
          message = "📊 **Elasticsearch & Kibana Container Status Report**\n\n"
  
          # Elasticsearch detailed status
          es_status = status["elasticsearch"]
          message += f"🔍 **Elasticsearch Container** (`{es_status['container_name']}`):\n"
          message += f"   📦 Container Exists: {'✅ Yes' if es_status['exists'] else '❌ No'}\n"
          message += f"   🚀 Container Running: {'✅ Yes' if es_status['running'] else '❌ No'}\n"
  
          if es_status['running']:
              message += f"   🌐 Access URL: http://localhost:9200\n"
              message += f"   💚 Status: Ready for connections\n"
          elif es_status['exists']:
              message += f"   ⚠️ Status: Container exists but not running\n"
              message += f"   💡 Action: Start container or use setup_elasticsearch\n"
          else:
              message += f"   🔧 Status: Container not found\n"
              message += f"   💡 Action: Run setup_elasticsearch to create\n"
  
          message += f"\n"
  
          # Kibana detailed status
          kibana_status = status["kibana"]
          message += f"📊 **Kibana Container** (`{kibana_status['container_name']}`):\n"
          message += f"   📦 Container Exists: {'✅ Yes' if kibana_status['exists'] else '❌ No'}\n"
          message += f"   🚀 Container Running: {'✅ Yes' if kibana_status['running'] else '❌ No'}\n"
  
          if kibana_status['running']:
              message += f"   🌐 Access URL: http://localhost:5601\n"
              message += f"   💚 Status: Dashboard available\n"
          elif kibana_status['exists']:
              message += f"   ⚠️ Status: Container exists but not running\n"
              message += f"   💡 Action: Start container or use setup_elasticsearch\n"
          else:
              message += f"   🔧 Status: Container not found\n"
              message += f"   💡 Action: Run setup_elasticsearch with include_kibana=true\n"
  
          # Current configuration summary
          config = load_config()
          message += f"\n⚙️ **Current Configuration Settings:**\n"
          message += f"   🏠 Host: {config['elasticsearch']['host']}\n"
          message += f"   🔌 Port: {config['elasticsearch']['port']}\n"
          message += f"   📍 Full URL: http://{config['elasticsearch']['host']}:{config['elasticsearch']['port']}\n"
  
          # Overall system status assessment
          es_ready = es_status['running']
          kibana_ready = kibana_status['running']
  
          message += f"\n🎯 **Overall System Status:**\n"
          if es_ready and kibana_ready:
              message += f"   ✅ **Fully Operational** - Both Elasticsearch and Kibana running\n"
              message += f"   🚀 **Next Steps:** Start indexing documents and using search\n"
          elif es_ready and not kibana_ready:
              message += f"   🟡 **Partially Ready** - Elasticsearch running, Kibana stopped\n"
              message += f"   💡 **Suggestion:** Elasticsearch is functional, Kibana optional for visualization\n"
          elif not es_ready and kibana_ready:
              message += f"   🔴 **Incomplete Setup** - Kibana running but Elasticsearch stopped\n"
              message += f"   ⚠️ **Action Required:** Start Elasticsearch for system to function\n"
          else:
              message += f"   🔴 **System Down** - Both services stopped\n"
              message += f"   🛠️ **Action Required:** Run setup_elasticsearch to start services\n"
  
          return message
  
      except ImportError as e:
          return f"❌ Module Error: Missing Elasticsearch setup dependency\n🔍 Details: {str(e)}\n💡 Ensure ElasticsearchSetup module is properly installed"
      except FileNotFoundError as e:
          return f"❌ Configuration Error: Required config file not found\n🔍 Details: {str(e)}\n💡 Check that config.json exists in the source directory"
      except Exception as e:
          return _format_admin_error(e, "check Elasticsearch status", "Docker container status monitoring")

• src/elasticsearch/elasticsearch_setup.py:210-232 (helper)

  Helper method in ElasticsearchSetup class that returns the status dictionary for both Elasticsearch and Kibana containers (exists, running, container_name), used directly by the elasticsearch_status handler.

  def get_container_status(self) -> Dict[str, Any]:
      """Get status of Elasticsearch and Kibana containers."""
      try:
          client = self._get_docker_client()
          
          status = {
              "elasticsearch": {
                  "exists": self._container_exists(self.container_name),
                  "running": self._is_container_running(self.container_name),
                  "container_name": self.container_name
              },
              "kibana": {
                  "exists": self._container_exists(self.kibana_container_name),
                  "running": self._is_container_running(self.kibana_container_name),
                  "container_name": self.kibana_container_name
              }
          }
          
          return status
          
      except Exception as e:
          return {"error": str(e)}

• src/admin/admin_server.py:459-462 (registration)

  The @app.tool decorator registers the elasticsearch_status function on the admin FastMCP app, which is later mounted to the main server.

  @app.tool(
      description="Check status of Elasticsearch and Kibana containers with detailed configuration information",
      tags={"admin", "elasticsearch", "status", "docker", "monitoring"}
  )

• src/elasticsearch/elasticsearch_setup.py:14-74 (helper)

  The ElasticsearchSetup class provides container management utilities including get_container_status, used by the tool handler.

  class ElasticsearchSetup:
      """Manage Elasticsearch Docker container setup."""
      
      def __init__(self, config_path: Path):
          self.config_path = config_path
          self.docker_client = None
          self.container_name = "elasticsearch-mcp"
          self.kibana_container_name = "kibana-mcp"
          
      def _get_docker_client(self):
          """Get Docker client."""
          if self.docker_client is None:
              try:
                  self.docker_client = docker.from_env()
                  # Test connection
                  self.docker_client.ping()
              except DockerException as e:
                  raise ConnectionError(f"Cannot connect to Docker. Is Docker running? Error: {e}")
          return self.docker_client
      
      def _is_elasticsearch_running(self, host: str, port: int) -> bool:
          """Check if Elasticsearch is running at the given host:port."""
          try:
              response = requests.get(f"http://{host}:{port}", timeout=5)
              return response.status_code == 200
          except:
              return False
      
      def _wait_for_elasticsearch(self, host: str, port: int, timeout: int = 60) -> bool:
          """Wait for Elasticsearch to be ready."""
          print(f"Waiting for Elasticsearch at {host}:{port}...")
          start_time = time.time()
          
          while time.time() - start_time < timeout:
              if self._is_elasticsearch_running(host, port):
                  print("✅ Elasticsearch is ready!")
                  return True
              time.sleep(2)
              print("⏳ Still waiting...")
          
          print("❌ Timeout waiting for Elasticsearch")
          return False
      
      def _container_exists(self, container_name: str) -> bool:
          """Check if container exists."""
          try:
              client = self._get_docker_client()
              client.containers.get(container_name)
              return True
          except NotFound:
              return False
      
      def _is_container_running(self, container_name: str) -> bool:
          """Check if container is running."""
          try:
              client = self._get_docker_client()
              container = client.containers.get(container_name)
              return container.status == 'running'
          except NotFound:
              return False