Agent Knowledge MCP

Instructions

Implementation Reference

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

  Primary handler implementation for the 'create_index_metadata' tool. Uses FastMCP @app.tool decorator for registration and schema definition via Annotated parameters. Handles metadata index creation/check, existing metadata check, and new metadata document insertion into Elasticsearch.

  @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/__init__.py:45-45 (registration)

  TOOL_DISTRIBUTION mapping registers the sub-server containing 'create_index_metadata' tool, indicating 3 tools including this one.

  "elasticsearch_index_metadata": 3, # create_index_metadata, update_index_metadata, delete_index_metadata

• src/elasticsearch/sub_servers/elasticsearch_index_metadata.py:26-37 (schema)

  Input schema defined using Pydantic Annotated fields with descriptions for tool 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"