Agent Knowledge MCP

Instructions

Implementation Reference

• src/elasticsearch/sub_servers/elasticsearch_index.py:254-412 (handler)

  Implementation of the list_indices tool handler. This FastMCP tool function lists all Elasticsearch indices, retrieves stats (doc count, size), and enriches with metadata from 'index_metadata' index if available. Formats output distinguishing documented vs undocumented indices with governance suggestions.
  @app.tool( description="List all available Elasticsearch indices with document count and size statistics", tags={"elasticsearch", "list", "indices", "stats"} ) async def list_indices() -> str: """List all available Elasticsearch indices with basic statistics.""" try: es = get_es_client() indices = es.indices.get_alias(index="*") # Get stats for each index indices_info = [] for index_name in indices.keys(): if not index_name.startswith('.'): # Skip system indices try: stats = es.indices.stats(index=index_name) doc_count = stats['indices'][index_name]['total']['docs']['count'] size = stats['indices'][index_name]['total']['store']['size_in_bytes'] # Initialize basic index info index_info = { "name": index_name, "docs": doc_count, "size_bytes": size, "description": "No description available", "purpose": "Not documented", "data_types": [], "usage_pattern": "Unknown", "created_date": "Unknown" } # Try to get metadata for this index try: metadata_search = { "query": { "term": { "index_name": index_name } }, "size": 1 } metadata_result = es.search(index="index_metadata", body=metadata_search) if metadata_result['hits']['total']['value'] > 0: metadata = metadata_result['hits']['hits'][0]['_source'] # Merge metadata into index info index_info.update({ "description": metadata.get('description', 'No description available'), "purpose": metadata.get('purpose', 'Not documented'), "data_types": metadata.get('data_types', []), "usage_pattern": metadata.get('usage_pattern', 'Unknown'), "created_date": metadata.get('created_date', 'Unknown'), "retention_policy": metadata.get('retention_policy', 'Not specified'), "related_indices": metadata.get('related_indices', []), "tags": metadata.get('tags', []), "created_by": metadata.get('created_by', 'Unknown'), "has_metadata": True }) else: index_info["has_metadata"] = False except Exception: # If metadata index doesn't exist or search fails, keep basic info index_info["has_metadata"] = False indices_info.append(index_info) except: indices_info.append({ "name": index_name, "docs": "unknown", "size_bytes": "unknown", "description": "Statistics unavailable", "has_metadata": False }) # Sort indices: metadata-documented first, then by name indices_info.sort(key=lambda x: (not x.get('has_metadata', False), x['name'])) # Format the output with metadata information result = "✅ Available indices with metadata:\n\n" # Count documented vs undocumented documented = sum(1 for idx in indices_info if idx.get('has_metadata', False)) undocumented = len(indices_info) - documented result += f"📊 **Index Overview**:\n" result += f" 📋 Total indices: {len(indices_info)}\n" result += f" ✅ Documented: {documented}\n" result += f" ❌ Undocumented: {undocumented}\n\n" if undocumented > 0: result += f"🚨 **Governance Alert**: {undocumented} indices lack metadata documentation\n" result += f" 💡 Use 'create_index_metadata' tool to document missing indices\n" result += f" 🎯 Proper documentation improves index management and team collaboration\n\n" # Group indices by documentation status documented_indices = [idx for idx in indices_info if idx.get('has_metadata', False)] undocumented_indices = [idx for idx in indices_info if not idx.get('has_metadata', False)] if documented_indices: result += f"📋 **Documented Indices** ({len(documented_indices)}):\n\n" for idx in documented_indices: size_mb = idx['size_bytes'] / 1048576 if isinstance(idx['size_bytes'], (int, float)) else 0 result += f"🟢 **{idx['name']}**\n" result += f" 📝 Description: {idx['description']}\n" result += f" 🎯 Purpose: {idx['purpose']}\n" result += f" 📊 Documents: {idx['docs']}, Size: {size_mb:.1f} MB\n" result += f" 📂 Data Types: {', '.join(idx.get('data_types', [])) or 'Not specified'}\n" result += f" 🔄 Usage: {idx.get('usage_pattern', 'Unknown')}\n" result += f" 📅 Created: {idx.get('created_date', 'Unknown')}\n" if idx.get('tags'): result += f" 🏷️ Tags: {', '.join(idx['tags'])}\n" if idx.get('related_indices'): result += f" 🔗 Related: {', '.join(idx['related_indices'])}\n" result += "\n" if undocumented_indices: result += f"❌ **Undocumented Indices** ({len(undocumented_indices)}) - Need Metadata:\n\n" for idx in undocumented_indices: size_mb = idx['size_bytes'] / 1048576 if isinstance(idx['size_bytes'], (int, float)) else 0 result += f"🔴 **{idx['name']}**\n" result += f" 📊 Documents: {idx['docs']}, Size: {size_mb:.1f} MB\n" result += f" ⚠️ Status: No metadata documentation found\n" result += f" 🔧 Action: Use 'create_index_metadata' to document this index\n\n" # Add metadata improvement suggestions if undocumented > 0: result += f"💡 **Metadata Improvement Suggestions**:\n" result += f" 📋 Document each index's purpose and data types\n" result += f" 🎯 Define usage patterns and access frequencies\n" result += f" 📅 Record creation dates and retention policies\n" result += f" 🔗 Link related indices for better organization\n" result += f" 🏷️ Add relevant tags for categorization\n" result += f" 👤 Track ownership and responsibility\n\n" return result except Exception as e: # Provide detailed error messages for different types of Elasticsearch errors error_message = "❌ Failed to list indices:\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 "timeout" in error_str: error_message += "⏱️ **Timeout Error**: Elasticsearch server is not responding\n" error_message += f"📍 Server may be overloaded or slow to respond\n" error_message += f"💡 Try: Wait and retry, or check server status\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:43-51 (registration)

  Mounting of the elasticsearch_index sub-server (containing list_indices) into the unified Elasticsearch FastMCP server app. The index_app from elasticsearch_index.py provides the list_indices tool among others.
  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