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