Agent Knowledge MCP

Instructions

Implementation Reference

• src/elasticsearch/sub_servers/elasticsearch_index.py:160-251 (handler)

  The core handler function for the 'delete_index' tool. It checks for index metadata, deletes the Elasticsearch index using es.indices.delete(), and provides detailed success/error messages with governance checks.

  @app.tool(
      description="Delete an Elasticsearch index and all its documents permanently",
      tags={"elasticsearch", "delete", "index", "destructive"}
  )
  async def delete_index(
          index: Annotated[str, Field(description="Name of the Elasticsearch index to delete")]
  ) -> str:
      """Delete an Elasticsearch index permanently."""
      try:
          es = get_es_client()
  
          # Check if metadata document exists for this index
          metadata_index = "index_metadata"
          try:
              # Search for existing metadata document
              search_body = {
                  "query": {
                      "term": {
                          "index_name.keyword": index
                      }
                  },
                  "size": 1
              }
  
              metadata_result = es.search(index=metadata_index, body=search_body)
  
              if metadata_result['hits']['total']['value'] > 0:
                  metadata_doc = metadata_result['hits']['hits'][0]
                  metadata_id = metadata_doc['_id']
                  metadata_source = metadata_doc['_source']
  
                  return (f"❌ Index deletion blocked - Metadata cleanup required!\n\n" +
                          f"🚨 **MANDATORY: Remove Index Metadata First**:\n" +
                          f"   📋 **Found Metadata Document**: {metadata_id}\n" +
                          f"   📝 **Index Description**: {metadata_source.get('description', 'No description')}\n" +
                          f"   🔧 **Required Action**: Delete metadata document before removing index\n\n" +
                          f"💡 **Cleanup Workflow**:\n" +
                          f"   1. Call 'delete_index_metadata' with index name '{index}'\n" +
                          f"   2. Then call 'delete_index' again to remove the actual index\n" +
                          f"   3. This ensures proper cleanup and audit trail\n\n" +
                          f"📊 **Metadata Details**:\n" +
                          f"   • Purpose: {metadata_source.get('purpose', 'Not specified')}\n" +
                          f"   • Data Types: {', '.join(metadata_source.get('data_types', []))}\n" +
                          f"   • Created: {metadata_source.get('created_date', 'Unknown')}\n" +
                          f"   • Usage: {metadata_source.get('usage_pattern', 'Not specified')}\n\n" +
                          f"🎯 **Why This Matters**:\n" +
                          f"   • Maintains clean metadata registry\n" +
                          f"   • Prevents orphaned documentation\n" +
                          f"   • Ensures proper audit trail for deletions\n" +
                          f"   • Confirms intentional removal with full context")
  
          except Exception as metadata_error:
              # If metadata index doesn't exist, warn but allow deletion
              if "index_not_found" in str(metadata_error).lower():
                  # Proceed with deletion but warn about missing metadata system
                  result = es.indices.delete(index=index)
  
                  return (f"⚠️ Index '{index}' deleted but metadata system is missing:\n\n" +
                          f"{json.dumps(result, indent=2, ensure_ascii=False)}\n\n" +
                          f"🚨 **Warning**: No metadata tracking system found\n" +
                          f"   📋 Consider setting up 'index_metadata' index for better governance\n" +
                          f"   💡 Use 'create_index_metadata' tool for future index documentation")
  
          # If we get here, no metadata found - proceed with deletion
          result = es.indices.delete(index=index)
  
          return f"✅ Index '{index}' deleted successfully:\n\n{json.dumps(result, indent=2, ensure_ascii=False)}"
  
      except Exception as e:
          # Provide detailed error messages for different types of Elasticsearch errors
          error_message = "❌ Failed to delete index:\n\n"
  
          error_str = str(e).lower()
          if "connection" in error_str or "refused" in error_str:
              error_message += "🔌 **Connection Error**: Cannot connect to Elasticsearch server\n"
              error_message += f"📍 Check if Elasticsearch is running at the configured address\n"
              error_message += f"💡 Try: Use 'setup_elasticsearch' tool to start Elasticsearch\n\n"
          elif (
                  "not_found" in error_str or "not found" in error_str) or "index_not_found_exception" in error_str or "no such index" in error_str:
              error_message += f"📁 **Index Not Found**: Index '{index}' does not exist\n"
              error_message += f"📍 Cannot delete an index that doesn't exist\n"
              error_message += f"💡 Try: Use 'list_indices' to see available indices\n\n"
          elif "permission" in error_str or "forbidden" in error_str:
              error_message += "🔒 **Permission Error**: Not allowed to delete index\n"
              error_message += f"📍 Insufficient permissions for index deletion\n"
              error_message += f"💡 Try: Check Elasticsearch security settings\n\n"
          else:
              error_message += f"⚠️ **Unknown Error**: {str(e)}\n\n"
  
          error_message += f"🔍 **Technical Details**: {str(e)}"
  
          return error_message

• src/elasticsearch/elasticsearch_server.py:24-51 (registration)

  Mounting of the elasticsearch_index sub-server into the unified Elasticsearch server, registering the delete_index tool among others.

  # Import sub-server applications for mounting
  from .sub_servers.elasticsearch_snapshots import app as snapshots_app
  from .sub_servers.elasticsearch_index_metadata import app as index_metadata_app
  from .sub_servers.elasticsearch_document import app as document_app
  from .sub_servers.elasticsearch_index import app as index_app
  from .sub_servers.elasticsearch_search import app as search_app
  from .sub_servers.elasticsearch_batch import app as batch_app
  
  # Create unified FastMCP application
  app = FastMCP(
      name="AgentKnowledgeMCP-Elasticsearch",
      version="2.0.0",
      instructions="Unified Elasticsearch tools for comprehensive knowledge management via modular server mounting"
  )
  
  # ================================
  # SERVER MOUNTING - MODULAR ARCHITECTURE
  # ================================
  
  print("🏗️ Mounting Elasticsearch sub-servers...")
  
  # Mount all sub-servers into unified interface
  app.mount(snapshots_app)           # 3 tools: snapshot management
  app.mount(index_metadata_app)      # 3 tools: metadata governance  
  app.mount(document_app)            # 3 tools: document operations
  app.mount(index_app)               # 3 tools: index management
  app.mount(search_app)              # 2 tools: search & validation
  app.mount(batch_app)               # 2 tools: batch operations

• src/main_server.py:71-85 (registration)

  Mounting of the unified elasticsearch_server_app into the main AgentKnowledgeMCP server, making the delete_index tool available in the primary server interface.

  print("🏗️ Mounting individual servers into main server...")
  
  # Mount Elasticsearch server with 'es' prefix
  # This provides: es_search, es_index_document, es_create_index, etc.
  app.mount(elasticsearch_server_app)
  
  # Mount Administrative operations server with 'admin' prefix
  # This provides: admin_get_config, admin_update_config, admin_server_status, etc.
  app.mount(admin_server_app)
  
  # Mount Prompt server for AgentKnowledgeMCP guidance
  # This provides: usage_guide, help_request (prompts for LLM assistance)
  app.mount(prompt_server_app)

• src/elasticsearch/sub_servers/elasticsearch_index.py:165-165 (schema)

  Pydantic schema definition for the 'delete_index' tool input parameter using Annotated and Field for validation and documentation.

  index: Annotated[str, Field(description="Name of the Elasticsearch index to delete")]

• src/elasticsearch/sub_servers/__init__.py:47-47 (helper)

  Documentation of tool distribution in sub-servers package, confirming delete_index is part of elasticsearch_index server.

  "elasticsearch_index": 3,          # list_indices, create_index, delete_index