Agent Knowledge MCP

Instructions

Implementation Reference

• src/elasticsearch/sub_servers/elasticsearch_index_metadata.py:21-160 (handler)

  The core handler function decorated with @app.tool() that implements the create_index_metadata tool. It creates or updates metadata documentation for Elasticsearch indices, including input validation via Pydantic Annotated fields, ES client interaction, metadata storage in 'index_metadata' index, and comprehensive user feedback.
  @app.tool( description="Create metadata documentation for an Elasticsearch index to ensure proper governance and documentation", tags={"elasticsearch", "metadata", "documentation", "governance"} ) async def create_index_metadata( index_name: Annotated[str, Field(description="Name of the index to document")], description: Annotated[str, Field(description="Detailed description of the index purpose and content")], purpose: Annotated[str, Field(description="Primary purpose and use case for this index")], data_types: Annotated[List[str], Field( description="Types of data stored in this index (e.g., 'documents', 'logs', 'metrics')")] = [], usage_pattern: Annotated[ str, Field(description="How the index is accessed (e.g., 'read-heavy', 'write-heavy', 'mixed')")] = "mixed", retention_policy: Annotated[ str, Field(description="Data retention policy and lifecycle management")] = "No specific policy", related_indices: Annotated[List[str], Field(description="Names of related or dependent indices")] = [], tags: Annotated[List[str], Field(description="Tags for categorizing and organizing indices")] = [], created_by: Annotated[str, Field(description="Team or person responsible for this index")] = "Unknown" ) -> str: """Create comprehensive metadata documentation for an Elasticsearch index.""" try: es = get_es_client() # Check if metadata index exists metadata_index = "index_metadata" try: es.indices.get(index=metadata_index) except Exception: # Create metadata index if it doesn't exist metadata_mapping = { "properties": { "index_name": {"type": "keyword"}, "description": {"type": "text"}, "purpose": {"type": "text"}, "data_types": {"type": "keyword"}, "created_by": {"type": "keyword"}, "created_date": {"type": "date"}, "usage_pattern": {"type": "keyword"}, "retention_policy": {"type": "text"}, "related_indices": {"type": "keyword"}, "tags": {"type": "keyword"}, "last_updated": {"type": "date"}, "updated_by": {"type": "keyword"} } } try: es.indices.create(index=metadata_index, body={"mappings": metadata_mapping}) except Exception as create_error: if "already exists" not in str(create_error).lower(): return f"❌ Failed to create metadata index: {str(create_error)}" # Check if metadata already exists for this index search_body = { "query": { "term": { "index_name.keyword": index_name } }, "size": 1 } existing_result = es.search(index=metadata_index, body=search_body) if existing_result['hits']['total']['value'] > 0: existing_doc = existing_result['hits']['hits'][0] existing_id = existing_doc['_id'] existing_data = existing_doc['_source'] return (f"⚠️ Index metadata already exists for '{index_name}'!\n\n" + f"📋 **Existing Metadata** (ID: {existing_id}):\n" + f" 📝 Description: {existing_data.get('description', 'No description')}\n" + f" 🎯 Purpose: {existing_data.get('purpose', 'No purpose')}\n" + f" 📂 Data Types: {', '.join(existing_data.get('data_types', []))}\n" + f" 👤 Created By: {existing_data.get('created_by', 'Unknown')}\n" + f" 📅 Created: {existing_data.get('created_date', 'Unknown')}\n\n" + f"💡 **Options**:\n" + f" 🔄 **Update**: Use 'update_index_metadata' to modify existing documentation\n" + f" 🗑️ **Replace**: Use 'delete_index_metadata' then 'create_index_metadata'\n" + f" ✅ **Keep**: Current metadata is sufficient, proceed with 'create_index'\n\n" + f"🚨 **Note**: You can now create the index '{index_name}' since metadata exists") # Create new metadata document current_time = datetime.now().isoformat() metadata_doc = { "index_name": index_name, "description": description, "purpose": purpose, "data_types": data_types, "created_by": created_by, "created_date": current_time, "usage_pattern": usage_pattern, "retention_policy": retention_policy, "related_indices": related_indices, "tags": tags, "last_updated": current_time, "updated_by": created_by } # Generate a consistent document ID metadata_id = f"metadata_{index_name}" result = es.index(index=metadata_index, id=metadata_id, body=metadata_doc) return (f"✅ Index metadata created successfully!\n\n" + f"📋 **Metadata Details**:\n" + f" 🎯 Index: {index_name}\n" + f" 📝 Description: {description}\n" + f" 🎯 Purpose: {purpose}\n" + f" 📂 Data Types: {', '.join(data_types) if data_types else 'None specified'}\n" + f" 🔄 Usage Pattern: {usage_pattern}\n" + f" 📅 Retention: {retention_policy}\n" + f" 🔗 Related Indices: {', '.join(related_indices) if related_indices else 'None'}\n" + f" 🏷️ Tags: {', '.join(tags) if tags else 'None'}\n" + f" 👤 Created By: {created_by}\n" + f" 📅 Created: {current_time}\n\n" + f"✅ **Next Steps**:\n" + f" 🔧 You can now use 'create_index' to create the actual index '{index_name}'\n" + f" 📊 Use 'list_indices' to see this metadata in the index listing\n" + f" 🔄 Use 'update_index_metadata' if you need to modify this documentation\n\n" + f"🎯 **Benefits Achieved**:\n" + f" • Index purpose is clearly documented\n" + f" • Team collaboration is improved through shared understanding\n" + f" • Future maintenance is simplified with proper context\n" + f" • Index governance and compliance are maintained") except Exception as e: error_message = "❌ Failed to create index metadata:\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" else: error_message += f"⚠️ **Unknown Error**: {str(e)}\n\n" error_message += f"🔍 **Technical Details**: {str(e)}" return error_message
• src/elasticsearch/sub_servers/elasticsearch_index_metadata.py:26-37 (schema)

  Input schema defined via Pydantic Field descriptions and types in the function parameters.
  index_name: Annotated[str, Field(description="Name of the index to document")], description: Annotated[str, Field(description="Detailed description of the index purpose and content")], purpose: Annotated[str, Field(description="Primary purpose and use case for this index")], data_types: Annotated[List[str], Field( description="Types of data stored in this index (e.g., 'documents', 'logs', 'metrics')")] = [], usage_pattern: Annotated[ str, Field(description="How the index is accessed (e.g., 'read-heavy', 'write-heavy', 'mixed')")] = "mixed", retention_policy: Annotated[ str, Field(description="Data retention policy and lifecycle management")] = "No specific policy", related_indices: Annotated[List[str], Field(description="Names of related or dependent indices")] = [], tags: Annotated[List[str], Field(description="Tags for categorizing and organizing indices")] = [], created_by: Annotated[str, Field(description="Team or person responsible for this index")] = "Unknown"
• src/elasticsearch/sub_servers/elasticsearch_index_metadata.py:21-24 (registration)

  Registration of the tool using FastMCP @app.tool decorator with description and tags.
  @app.tool( description="Create metadata documentation for an Elasticsearch index to ensure proper governance and documentation", tags={"elasticsearch", "metadata", "documentation", "governance"} )