Agent Knowledge MCP

Instructions

Implementation Reference

• src/elasticsearch/sub_servers/elasticsearch_snapshots.py:394-547 (handler)

  Implementation of the 'list_snapshots' tool handler, including input schema via Pydantic Annotated fields and registration via FastMCP @app.tool decorator. This is the core execution logic that lists Elasticsearch snapshots with detailed status and information.
  @app.tool( description="List all snapshots in an Elasticsearch repository with detailed information and status", tags={"elasticsearch", "snapshot", "list", "repository"} ) async def list_snapshots( repository: Annotated[str, Field(description="Repository name to list snapshots from")] = "backup_repository", verbose: Annotated[bool, Field(description="Whether to show detailed information for each snapshot")] = True ) -> str: """List all snapshots in an Elasticsearch repository.""" try: es = get_es_client() # Check if repository exists try: repo_info = es.snapshot.get_repository(repository=repository) except: return (f"❌ Repository '{repository}' not found!\n\n" + f"📂 **Repository Error**: Cannot access snapshot repository\n" + f"📍 **Possible Issues**:\n" + f" 1. Repository name is incorrect\n" + f" 2. Repository was not created yet\n" + f" 3. Elasticsearch path.repo configuration issue\n\n" + f"💡 **Solutions**:\n" + f" • Use 'create_snapshot' to create repository\n" + f" • Check Elasticsearch configuration\n" + f" • Verify repository permissions") # Get repository details repo_details = repo_info.get(repository, {}) repo_type = repo_details.get('type', 'unknown') repo_settings = repo_details.get('settings', {}) # List all snapshots try: snapshots_result = es.snapshot.get(repository=repository, snapshot="_all") snapshots = snapshots_result.get('snapshots', []) except: snapshots = [] if not snapshots: return (f"📋 No snapshots found in repository '{repository}'\n\n" + f"📂 **Repository Information**:\n" + f" 📋 Name: {repository}\n" + f" 📊 Type: {repo_type}\n" + f" 📍 Location: {repo_settings.get('location', 'Not specified')}\n" + f" 📸 Snapshots: 0\n\n" + f"💡 **Next Steps**:\n" + f" • Use 'create_snapshot' to create your first backup\n" + f" • Regular snapshots help protect against data loss\n" + f" • Consider automated snapshot scheduling") # Sort snapshots by start time (newest first) snapshots.sort(key=lambda x: x.get('start_time_in_millis', 0), reverse=True) result_message = f"📸 Found {len(snapshots)} snapshot(s) in repository '{repository}'\n\n" # Repository information result_message += (f"📂 **Repository Information**:\n" + f" 📋 Name: {repository}\n" + f" 📊 Type: {repo_type}\n" + f" 📍 Location: {repo_settings.get('location', 'Not specified')}\n" + f" 📸 Total Snapshots: {len(snapshots)}\n\n") # List snapshots result_message += f"📋 **Available Snapshots**:\n\n" for i, snapshot in enumerate(snapshots, 1): name = snapshot.get('snapshot', 'Unknown') state = snapshot.get('state', 'UNKNOWN') # Status emoji based on state if state == 'SUCCESS': status_emoji = "✅" elif state == 'PARTIAL': status_emoji = "⚠️" elif state == 'FAILED': status_emoji = "❌" elif state == 'IN_PROGRESS': status_emoji = "🔄" else: status_emoji = "❓" result_message += f"{status_emoji} **{i}. {name}**\n" result_message += f" 📊 State: {state}\n" if verbose: # Detailed information indices = snapshot.get('indices', []) result_message += f" 📦 Indices: {len(indices)} ({', '.join(indices[:3])}{'...' if len(indices) > 3 else ''})\n" # Timestamps if snapshot.get('start_time'): result_message += f" 🕒 Started: {snapshot['start_time']}\n" if snapshot.get('end_time'): result_message += f" 🕒 Completed: {snapshot['end_time']}\n" # Duration if snapshot.get('start_time_in_millis') and snapshot.get('end_time_in_millis'): duration = (snapshot['end_time_in_millis'] - snapshot['start_time_in_millis']) / 1000 result_message += f" ⏱️ Duration: {duration:.2f} seconds\n" # Shards info if snapshot.get('shards'): shards = snapshot['shards'] total = shards.get('total', 0) successful = shards.get('successful', 0) failed = shards.get('failed', 0) result_message += f" 🔢 Shards: {successful}/{total} successful" if failed > 0: result_message += f" ({failed} failed)" result_message += "\n" # Metadata metadata = snapshot.get('metadata', {}) if metadata.get('description'): result_message += f" 📝 Description: {metadata['description']}\n" # Global state include_global_state = snapshot.get('include_global_state', False) result_message += f" 🌐 Global State: {'Included' if include_global_state else 'Excluded'}\n" result_message += "\n" # Usage instructions result_message += (f"🔧 **Usage Instructions**:\n" + f" • Use 'restore_snapshot <name>' to restore from any snapshot\n" + f" • Use 'create_snapshot <name>' to create new backups\n" + f" • Monitor storage space in repository location\n" + f" • Clean up old snapshots periodically\n\n" + f"💾 **Backup Best Practices**:\n" + f" ✅ Regular automated snapshots (daily/weekly)\n" + f" ✅ Test restore procedures periodically\n" + f" ✅ Monitor snapshot success/failure status\n" + f" ✅ Keep snapshots in multiple locations if possible") return result_message except Exception as e: error_message = "❌ Failed to list snapshots:\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 "repository" in error_str and ("not found" in error_str or "missing" in error_str): error_message += f"📂 **Repository Error**: Repository '{repository}' not found\n" error_message += f"📍 Check repository configuration and permissions\n" error_message += f"💡 Try: Create repository first or check Elasticsearch configuration\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-52 (registration)

  Mounting of the elasticsearch_snapshots sub-server app into the main Elasticsearch FastMCP app, which registers the 'list_snapshots' tool (among others) for use in the unified server interface.
  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