Agent Knowledge MCP

Instructions

Implementation Reference

• src/elasticsearch/sub_servers/elasticsearch_index_metadata.py:162-165 (registration)

  Registers the 'update_index_metadata' tool with FastMCP app using @app.tool decorator, providing description and tags.
  @app.tool( description="Update existing metadata documentation for an Elasticsearch index", tags={"elasticsearch", "metadata", "update", "documentation"} )
• src/elasticsearch/sub_servers/elasticsearch_index_metadata.py:166-179 (schema)

  Defines the input schema using Annotated parameters with pydantic Field descriptions for validation and documentation.
  async def update_index_metadata( index_name: Annotated[str, Field(description="Name of the index to update metadata for")], description: Annotated[ Optional[str], Field(description="Updated description of the index purpose and content")] = None, purpose: Annotated[Optional[str], Field(description="Updated primary purpose and use case")] = None, data_types: Annotated[ Optional[List[str]], Field(description="Updated types of data stored in this index")] = None, usage_pattern: Annotated[Optional[str], Field(description="Updated access pattern")] = None, retention_policy: Annotated[Optional[str], Field(description="Updated data retention policy")] = None, related_indices: Annotated[ Optional[List[str]], Field(description="Updated related or dependent indices")] = None, tags: Annotated[Optional[List[str]], Field(description="Updated tags for categorization")] = None, updated_by: Annotated[str, Field(description="Person or team making this update")] = "Unknown" ) -> str:
• src/elasticsearch/sub_servers/elasticsearch_index_metadata.py:180-296 (handler)

  Executes the tool logic: retrieves ES client, searches for existing metadata, applies partial updates to specified fields, fetches updated data, generates change summary, and returns formatted response. Includes detailed error handling for connection issues and missing indices.
  """Update existing metadata documentation for an Elasticsearch index.""" try: es = get_es_client() metadata_index = "index_metadata" # Search for existing metadata 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: return (f"❌ No metadata found for index '{index_name}'!\n\n" + f"🚨 **Missing Metadata**: Cannot update non-existent documentation\n" + f" 💡 **Solution**: Use 'create_index_metadata' to create documentation first\n" + f" 📋 **Required**: Provide description, purpose, and data types\n" + f" ✅ **Then**: Use this update tool for future modifications\n\n" + f"🔍 **Alternative**: Use 'list_indices' to see all documented indices") # Get existing document existing_doc = existing_result['hits']['hits'][0] existing_id = existing_doc['_id'] existing_data = existing_doc['_source'] # Prepare update data - only update provided fields update_data = { "last_updated": datetime.now().isoformat(), "updated_by": updated_by } if description is not None: update_data["description"] = description if purpose is not None: update_data["purpose"] = purpose if data_types is not None: update_data["data_types"] = data_types if usage_pattern is not None: update_data["usage_pattern"] = usage_pattern if retention_policy is not None: update_data["retention_policy"] = retention_policy if related_indices is not None: update_data["related_indices"] = related_indices if tags is not None: update_data["tags"] = tags # Update the document result = es.update(index=metadata_index, id=existing_id, body={"doc": update_data}) # Get updated document to show changes updated_result = es.get(index=metadata_index, id=existing_id) updated_data = updated_result['_source'] # Build change summary changes_made = [] if description is not None: changes_made.append(f" 📝 Description: {existing_data.get('description', 'None')} → {description}") if purpose is not None: changes_made.append(f" 🎯 Purpose: {existing_data.get('purpose', 'None')} → {purpose}") if data_types is not None: old_types = ', '.join(existing_data.get('data_types', [])) new_types = ', '.join(data_types) changes_made.append(f" 📂 Data Types: {old_types or 'None'} → {new_types}") if usage_pattern is not None: changes_made.append(f" 🔄 Usage Pattern: {existing_data.get('usage_pattern', 'None')} → {usage_pattern}") if retention_policy is not None: changes_made.append(f" 📅 Retention: {existing_data.get('retention_policy', 'None')} → {retention_policy}") if related_indices is not None: old_related = ', '.join(existing_data.get('related_indices', [])) new_related = ', '.join(related_indices) changes_made.append(f" 🔗 Related: {old_related or 'None'} → {new_related}") if tags is not None: old_tags = ', '.join(existing_data.get('tags', [])) new_tags = ', '.join(tags) changes_made.append(f" 🏷️ Tags: {old_tags or 'None'} → {new_tags}") return (f"✅ Index metadata updated successfully!\n\n" + f"📋 **Updated Metadata for '{index_name}'**:\n" + (f"🔄 **Changes Made**:\n" + '\n'.join(changes_made) + "\n\n" if changes_made else "") + f"📊 **Current Metadata**:\n" + f" 📝 Description: {updated_data.get('description', 'No description')}\n" + f" 🎯 Purpose: {updated_data.get('purpose', 'No purpose')}\n" + f" 📂 Data Types: {', '.join(updated_data.get('data_types', [])) if updated_data.get('data_types') else 'None'}\n" + f" 🔄 Usage Pattern: {updated_data.get('usage_pattern', 'Unknown')}\n" + f" 📅 Retention: {updated_data.get('retention_policy', 'Not specified')}\n" + f" 🔗 Related Indices: {', '.join(updated_data.get('related_indices', [])) if updated_data.get('related_indices') else 'None'}\n" + f" 🏷️ Tags: {', '.join(updated_data.get('tags', [])) if updated_data.get('tags') else 'None'}\n" + f" 👤 Last Updated By: {updated_by}\n" + f" 📅 Last Updated: {update_data['last_updated']}\n\n" + f"✅ **Benefits**:\n" + f" • Index documentation stays current and accurate\n" + f" • Team has updated context for index usage\n" + f" • Change history is tracked with timestamps\n" + f" • Governance and compliance are maintained") except Exception as e: error_message = "❌ Failed to update 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" elif ("not_found" in error_str or "not found" in error_str) and "index" in error_str: error_message += f"📁 **Index Error**: Metadata index 'index_metadata' does not exist\n" error_message += f"📍 The metadata system has not been initialized\n" error_message += f"💡 Try: Use 'create_index_metadata' to set up metadata system\n\n" else: error_message += f"⚠️ **Unknown Error**: {str(e)}\n\n" error_message += f"🔍 **Technical Details**: {str(e)}" return error_message