Agent Knowledge MCP

Instructions

Implementation Reference

• src/admin/admin_server.py:352-453 (handler)

  Primary handler for the 'setup_elasticsearch' tool. Manages Docker container lifecycle for Elasticsearch (+Kibana), calls helper auto_setup_elasticsearch, updates config, and reinitializes client.

  @app.tool(
      description="Auto-setup Elasticsearch using Docker with optional Kibana and force recreate options",
      tags={"admin", "elasticsearch", "setup", "docker", "auto-install"}
  )
  async def setup_elasticsearch(
      include_kibana: Annotated[bool, Field(description="Also setup Kibana (default: true)")] = True,
      force_recreate: Annotated[bool, Field(description="Force recreate containers even if they exist")] = False
  ) -> str:
      """Auto-setup Elasticsearch using Docker with comprehensive Docker container management."""
      try:
          # Get config path for setup
          config_path = Path(__file__).parent.parent / "config.json"
          config = load_config()
  
          # Handle force recreate - stop existing containers first
          if force_recreate:
              try:
                  setup_manager = ElasticsearchSetup(config_path)
                  stop_result = setup_manager.stop_containers()
  
                  # Give containers time to stop gracefully
                  import time
                  time.sleep(5)
  
                  message = "🔄 **Force Recreate Mode:**\n"
                  message += f"   🛑 Stopped existing containers\n"
                  message += f"   ⏳ Waiting for graceful shutdown...\n\n"
              except Exception as e:
                  # Continue with setup even if stop fails
                  message = f"⚠️ **Container Stop Warning:** {str(e)}\n"
                  message += f"   💡 Continuing with setup anyway...\n\n"
          else:
              message = ""
  
          # Run comprehensive auto setup
          result = auto_setup_elasticsearch(config_path, config)
  
          if result["status"] == "already_configured":
              message += "✅ **Elasticsearch Already Configured!**\n\n"
              message += f"📍 **Running at:** http://{result['host']}:{result['port']}\n"
              message += f"🔍 **Status:** Ready and operational\n"
              message += f"💡 **Action:** No setup needed - already working perfectly\n\n"
              message += f"🚀 **Next Steps:**\n"
              message += f"   • Test connection with search tools\n"
              message += f"   • Start indexing your knowledge base\n"
              message += f"   • Use force_recreate=true to rebuild if needed"
  
          elif result["status"] == "setup_completed":
              es_info = result["elasticsearch"]
              kibana_info = result.get("kibana")
  
              message += "🎉 **Elasticsearch Setup Completed Successfully!**\n\n"
              message += f"📍 **Elasticsearch:** http://{es_info['host']}:{es_info['port']}\n"
  
              # Handle Kibana setup results
              if kibana_info:
                  if kibana_info.get("status") in ["running", "already_running"]:
                      message += f"📊 **Kibana:** http://{kibana_info['host']}:{kibana_info['port']}\n"
                      message += f"   ✅ Status: {kibana_info['status'].replace('_', ' ').title()}\n"
                  elif "error" in kibana_info:
                      message += f"⚠️ **Kibana Warning:** {kibana_info['error']}\n"
                      message += f"   💡 Elasticsearch is ready, Kibana setup had issues\n"
              else:
                  message += f"📊 **Kibana:** Skipped (include_kibana=false)\n"
  
              message += f"\n🔧 **Configuration Updated:**\n"
              message += f"   📝 config.json automatically updated\n"
              message += f"   🔄 Elasticsearch client reinitialized\n"
              message += f"   ✅ All components ready for use\n"
  
              message += f"\n🚀 **Next Steps:**\n"
              message += f"   • Test search functionality\n"
              message += f"   • Index your first documents\n"
              message += f"   • Explore Kibana dashboard (if enabled)\n"
  
              # Reload configuration to use new Elasticsearch setup
              new_config = load_config()
              init_elasticsearch(new_config)
              reset_es_client()
  
          else:
              # Setup failed
              error_msg = result.get("error", "Unknown setup error")
              message += f"❌ **Elasticsearch Setup Failed!**\n\n"
              message += f"🚨 **Error Details:** {error_msg}\n\n"
              message += f"🔧 **Troubleshooting Steps:**\n"
              message += f"   1. Check Docker is running and accessible\n"
              message += f"   2. Verify ports 9200 (ES) and 5601 (Kibana) are available\n"
              message += f"   3. Try force_recreate=true to rebuild containers\n"
              message += f"   4. Check Docker logs for detailed error information\n"
              message += f"   5. Ensure sufficient disk space and memory\n\n"
              message += f"💡 **Need Help?** Check Docker status and container logs for more details"
  
          return message
  
      except ImportError as e:
          return f"❌ Module Error: Missing required dependency\n🔍 Details: {str(e)}\n💡 Ensure all Elasticsearch setup modules are properly installed"
      except FileNotFoundError as e:
          return f"❌ File Error: Required configuration or setup file not found\n🔍 Details: {str(e)}\n💡 Check config.json exists and setup modules are in place"
      except Exception as e:
          return _format_admin_error(e, "setup Elasticsearch", f"Docker setup with include_kibana={include_kibana}, force_recreate={force_recreate}")

• src/admin/admin_server.py:356-359 (schema)

  Pydantic schema definitions for tool inputs using Annotated[Field]. Defines parameters for Kibana inclusion and force recreate.

  async def setup_elasticsearch(
      include_kibana: Annotated[bool, Field(description="Also setup Kibana (default: true)")] = True,
      force_recreate: Annotated[bool, Field(description="Force recreate containers even if they exist")] = False
  ) -> str:

• src/main_server.py:105-105 (registration)

  Lists 'setup_elasticsearch' as available tool from Admin Server in CLI output, confirming registration via mounting.

  print("    └─ Tools: get_config, update_config, server_status, server_upgrade, setup_elasticsearch, elasticsearch_status, validate_config, reset_config, reload_config")

• src/elasticsearch/elasticsearch_setup.py:271-310 (helper)

  Core helper function called by the tool handler. Performs Docker setup via ElasticsearchSetup class, updates config.json, returns setup status.

  def auto_setup_elasticsearch(config_path: Path, config: Dict[str, Any]) -> Dict[str, Any]:
      """Auto-setup Elasticsearch if not configured or not accessible."""
      
      # Check if Elasticsearch is already configured and accessible
      if check_elasticsearch_config(config):
          host = config["elasticsearch"]["host"]
          port = config["elasticsearch"]["port"]
          print(f"Elasticsearch already running at {host}:{port}")
          return {"status": "already_configured", "host": host, "port": port}
      
      print("Elasticsearch not accessible, starting auto-setup...")
      
      # Setup Elasticsearch using Docker
      setup = ElasticsearchSetup(config_path)
      
      try:
          result = setup.setup_elasticsearch(include_kibana=True)
          
          # Update config file
          es_host = result["elasticsearch"]["host"]
          es_port = result["elasticsearch"]["port"]
          setup.update_config(es_host, es_port)
          
          print("🎉 Elasticsearch setup completed!")
          print(f"📍 Elasticsearch: http://{es_host}:{es_port}")
          
          if result["kibana"] and result["kibana"]["status"] in ["running", "already_running"]:
              kibana_host = result["kibana"]["host"]
              kibana_port = result["kibana"]["port"]
              print(f"📊 Kibana: http://{kibana_host}:{kibana_port}")
          
          return {
              "status": "setup_completed",
              "elasticsearch": result["elasticsearch"],
              "kibana": result["kibana"]
          }
          
      except Exception as e:
          print(f"❌ Failed to setup Elasticsearch: {e}")
          return {"status": "setup_failed", "error": str(e)}

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

  Supporting class with Docker operations: container management (start/stop/status), config updates, health checks for ES and Kibana.

  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
      
      def start_elasticsearch_container(self) -> Dict[str, Any]:
          """Start Elasticsearch container."""
          client = self._get_docker_client()
          
          # Check if container already exists
          if self._container_exists(self.container_name):
              if self._is_container_running(self.container_name):
                  print(f"Container {self.container_name} is already running")
                  return {"status": "already_running", "host": "localhost", "port": 9200}
              else:
                  # Start existing container
                  print(f"🔄 Starting existing container {self.container_name}")
                  container = client.containers.get(self.container_name)
                  container.start()
          else:
              # Create new container
              print(f"🚀 Creating new Elasticsearch container {self.container_name}")
              
              environment = {
                  "discovery.type": "single-node",
                  "ES_JAVA_OPTS": "-Xms512m -Xmx512m",
                  "xpack.security.enabled": "false",
                  "xpack.security.enrollment.enabled": "false"
              }
              
              ports = {"9200/tcp": 9200, "9300/tcp": 9300}
              
              container = client.containers.run(
                  "docker.elastic.co/elasticsearch/elasticsearch:8.14.1",
                  name=self.container_name,
                  environment=environment,
                  ports=ports,
                  detach=True,
                  restart_policy={"Name": "unless-stopped"}
              )
              
              print(f"📦 Container {self.container_name} created")
          
          # Wait for Elasticsearch to be ready
          if self._wait_for_elasticsearch("localhost", 9200):
              return {"status": "running", "host": "localhost", "port": 9200}
          else:
              raise RuntimeError("Failed to start Elasticsearch within timeout")
      
      def start_kibana_container(self) -> Dict[str, Any]:
          """Start Kibana container."""
          client = self._get_docker_client()
          
          # Check if Elasticsearch is running first
          if not self._is_container_running(self.container_name):
              raise RuntimeError("Elasticsearch container must be running before starting Kibana")
          
          # Check if Kibana container already exists
          if self._container_exists(self.kibana_container_name):
              if self._is_container_running(self.kibana_container_name):
                  print(f"✅ Container {self.kibana_container_name} is already running")
                  return {"status": "already_running", "host": "localhost", "port": 5601}
              else:
                  # Start existing container
                  print(f"🔄 Starting existing container {self.kibana_container_name}")
                  container = client.containers.get(self.kibana_container_name)
                  container.start()
          else:
              # Create new container
              print(f"🚀 Creating new Kibana container {self.kibana_container_name}")
              
              environment = {
                  "ELASTICSEARCH_HOSTS": "http://elasticsearch-mcp:9200"
              }
              
              ports = {"5601/tcp": 5601}
              
              container = client.containers.run(
                  "docker.elastic.co/kibana/kibana:8.14.1",
                  name=self.kibana_container_name,
                  environment=environment,
                  ports=ports,
                  links={self.container_name: "elasticsearch-mcp"},
                  detach=True,
                  restart_policy={"Name": "unless-stopped"}
              )
              
              print(f"📦 Container {self.kibana_container_name} created")
          
          # Wait a bit for Kibana to start
          print("⏳ Waiting for Kibana to start...")
          time.sleep(10)
          
          return {"status": "running", "host": "localhost", "port": 5601}
      
      def setup_elasticsearch(self, include_kibana: bool = True) -> Dict[str, Any]:
          """Complete Elasticsearch setup with optional Kibana."""
          print("🔧 Setting up Elasticsearch...")
          
          # Start Elasticsearch
          es_result = self.start_elasticsearch_container()
          
          result = {
              "elasticsearch": es_result,
              "kibana": None
          }
          
          # Start Kibana if requested
          if include_kibana:
              try:
                  kibana_result = self.start_kibana_container()
                  result["kibana"] = kibana_result
              except Exception as e:
                  print(f"⚠️  Warning: Failed to start Kibana: {e}")
                  result["kibana"] = {"status": "failed", "error": str(e)}
          
          return result
      
      def update_config(self, host: str, port: int) -> None:
          """Update configuration file with Elasticsearch connection details."""
          try:
              # Load current config
              with open(self.config_path, 'r', encoding='utf-8') as f:
                  config = json.load(f)
              
              # Update Elasticsearch settings
              config["elasticsearch"]["host"] = host
              config["elasticsearch"]["port"] = port
              config["elasticsearch"]["auto_setup"] = True
              
              # Save updated config
              with open(self.config_path, 'w', encoding='utf-8') as f:
                  json.dump(config, f, indent=2, ensure_ascii=False)
              
              print(f"✅ Updated config file: {host}:{port}")
              
          except Exception as e:
              print(f"❌ Failed to update config: {e}")
              raise
      
      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)}
      
      def stop_containers(self) -> Dict[str, Any]:
          """Stop Elasticsearch and Kibana containers."""
          try:
              client = self._get_docker_client()
              results = {}
              
              # Stop Kibana first
              if self._container_exists(self.kibana_container_name):
                  container = client.containers.get(self.kibana_container_name)
                  container.stop()
                  results["kibana"] = "stopped"
              
              # Stop Elasticsearch
              if self._container_exists(self.container_name):
                  container = client.containers.get(self.container_name)
                  container.stop()
                  results["elasticsearch"] = "stopped"
              
              return results
              
          except Exception as e:
              return {"error": str(e)}