Agent Knowledge MCP

Instructions

Implementation Reference

• src/elasticsearch/sub_servers/elasticsearch_snapshots.py:195-387 (handler)

  Core handler implementation for the 'restore_snapshot' tool. Includes schema definition via Annotated[Field], registration via @app.tool decorator, and full logic using Elasticsearch client to restore snapshots with options for indices, renaming, settings, etc.
  @app.tool( description="Restore indices from an Elasticsearch snapshot with comprehensive options and conflict resolution", tags={"elasticsearch", "snapshot", "restore", "rollback"} ) async def restore_snapshot( snapshot_name: Annotated[str, Field(description="Name of the snapshot to restore from")], repository: Annotated[str, Field(description="Repository containing the snapshot")] = "backup_repository", indices: Annotated[Optional[str], Field( description="Comma-separated list of indices to restore (default: all from snapshot)")] = None, ignore_unavailable: Annotated[bool, Field(description="Whether to ignore unavailable indices")] = True, include_global_state: Annotated[bool, Field(description="Whether to restore cluster global state")] = False, wait_for_completion: Annotated[bool, Field(description="Whether to wait for restore completion")] = True, rename_pattern: Annotated[ Optional[str], Field(description="Pattern to rename restored indices (e.g., 'restored_%s')")] = None, index_settings: Annotated[Optional[str], Field(description="JSON string of index settings to override")] = None ) -> str: """Restore indices from an Elasticsearch snapshot.""" try: es = get_es_client() # Verify 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"📍 **Available Actions**:\n" + f" 1. Check repository name spelling\n" + f" 2. Use 'create_snapshot' to create repository first\n" + f" 3. Verify Elasticsearch path.repo configuration\n\n" + f"💡 **Tip**: Repositories must be configured before accessing snapshots") # Verify snapshot exists try: snapshot_info = es.snapshot.get(repository=repository, snapshot=snapshot_name) except: return (f"❌ Snapshot '{snapshot_name}' not found in repository '{repository}'!\n\n" + f"📸 **Snapshot Error**: Cannot find the specified snapshot\n" + f"📍 **Possible Issues**:\n" + f" 1. Snapshot name is incorrect\n" + f" 2. Snapshot was deleted or corrupted\n" + f" 3. Repository path has changed\n\n" + f"🔍 **Next Steps**:\n" + f" • Use 'list_snapshots' to see available snapshots\n" + f" • Check repository configuration and permissions\n" + f" • Verify backup storage accessibility") # Parse indices parameter if indices: indices_list = [idx.strip() for idx in indices.split(',')] indices_param = ','.join(indices_list) else: indices_param = None # Restore all indices from snapshot indices_list = ["all"] # Build restore body restore_body = { "ignore_unavailable": ignore_unavailable, "include_global_state": include_global_state } if indices_param: restore_body["indices"] = indices_param if rename_pattern: restore_body["rename_pattern"] = rename_pattern restore_body["rename_replacement"] = rename_pattern if index_settings: try: settings_dict = json.loads(index_settings) restore_body["index_settings"] = settings_dict except json.JSONDecodeError: return (f"❌ Invalid JSON in index_settings parameter!\n\n" + f"📋 **JSON Error**: Cannot parse index settings\n" + f"📍 **Provided**: {index_settings}\n" + f"💡 **Example**: '{{\"number_of_replicas\": 0, \"refresh_interval\": \"30s\"}}'") # Check for potential conflicts (existing indices) conflicts = [] if indices_list and indices_list != ["all"]: for index_name in indices_list: if rename_pattern: # If renaming, check the new name new_name = rename_pattern.replace('%s', index_name) try: es.indices.get(index=new_name) conflicts.append(f"{index_name} -> {new_name}") except: pass # Index doesn't exist, no conflict else: # Direct restore, check original name try: es.indices.get(index=index_name) conflicts.append(index_name) except: pass # Index doesn't exist, no conflict # Warn about conflicts conflict_warning = "" if conflicts and not rename_pattern: conflict_warning = (f"\n⚠️ **Warning - Existing Indices Will Be Overwritten**:\n" + f" 📋 Conflicting indices: {', '.join(conflicts)}\n" + f" 🔄 These indices will be closed and replaced\n" + f" 💡 Consider using rename_pattern to avoid conflicts\n\n") # Execute restore restore_result = es.snapshot.restore( repository=repository, snapshot=snapshot_name, body=restore_body, wait_for_completion=wait_for_completion ) # Get snapshot details for reporting snapshot_details = snapshot_info['snapshots'][0] if snapshot_info.get('snapshots') else {} snapshot_state = snapshot_details.get('state', 'UNKNOWN') # Format response based on completion if wait_for_completion: restore_info = restore_result.get('snapshot', {}) shards_info = restore_info.get('shards', {}) result_message = (f"✅ Snapshot '{snapshot_name}' restored successfully!\n\n" + f"🔄 **Restore Details**:\n" + f" 📂 Repository: {repository}\n" + f" 📸 Snapshot: {snapshot_name}\n" + f" 📊 Snapshot State: {snapshot_state}\n" + f" 📦 Restored Indices: {', '.join(indices_list)}\n" + f" 🌐 Global State: {'Restored' if include_global_state else 'Skipped'}\n") if rename_pattern: result_message += f" 🏷️ Rename Pattern: {rename_pattern}\n" if shards_info: result_message += (f" 🔢 Shards: {shards_info.get('total', 0)} total, " + f"{shards_info.get('successful', 0)} successful, " + f"{shards_info.get('failed', 0)} failed\n") else: result_message = (f"🔄 Snapshot restore started!\n\n" + f"🔄 **Restore Details**:\n" + f" 📂 Repository: {repository}\n" + f" 📸 Snapshot: {snapshot_name}\n" + f" 📊 Status: Running in background\n" + f" 📦 Indices: {', '.join(indices_list)}\n" + f" 🌐 Global State: {'Included' if include_global_state else 'Excluded'}\n") if conflict_warning: result_message += conflict_warning result_message += (f"\n✅ **Restore Complete**:\n" + f" 🔄 Data has been {'restored' if wait_for_completion else 'restore started'}\n" + f" 🔍 Use 'list_indices' to verify restored indices\n" + f" 📊 Check cluster health and index status\n" + f" 🧪 Test restored data integrity\n\n" + f"📋 **Post-Restore Checklist**:\n" + f" ✅ Verify all expected indices are present\n" + f" ✅ Check document counts match expectations\n" + f" ✅ Test search functionality on restored data\n" + f" ✅ Monitor cluster performance after restore") return result_message except Exception as e: error_message = "❌ Failed to restore snapshot:\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 path.repo settings\n\n" elif "snapshot" in error_str and ("not found" in error_str or "missing" in error_str): error_message += f"📸 **Snapshot Error**: Snapshot '{snapshot_name}' not found\n" error_message += f"📍 Check snapshot name and repository\n" error_message += f"💡 Try: Use 'list_snapshots' to see available snapshots\n\n" elif "index_not_found" in error_str: error_message += f"📋 **Index Error**: Some indices from snapshot no longer exist\n" error_message += f"📍 Original indices may have been deleted\n" error_message += f"💡 Try: Use ignore_unavailable=true or specify different indices\n\n" elif "already_exists" in error_str or "conflict" in error_str: error_message += f"🔄 **Conflict Error**: Cannot restore over existing indices\n" error_message += f"📍 Target indices already exist and are open\n" error_message += f"💡 Try: Use rename_pattern or close/delete conflicting indices\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:42-52 (registration)

  Mounting of the elasticsearch_snapshots sub-server app into the main Elasticsearch FastMCP app, effectively registering the restore_snapshot tool (among others) 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
• src/elasticsearch/sub_servers/elasticsearch_snapshots.py:200-210 (schema)

  Input schema definition using Pydantic Annotated and Field for the restore_snapshot tool parameters.
  snapshot_name: Annotated[str, Field(description="Name of the snapshot to restore from")], repository: Annotated[str, Field(description="Repository containing the snapshot")] = "backup_repository", indices: Annotated[Optional[str], Field( description="Comma-separated list of indices to restore (default: all from snapshot)")] = None, ignore_unavailable: Annotated[bool, Field(description="Whether to ignore unavailable indices")] = True, include_global_state: Annotated[bool, Field(description="Whether to restore cluster global state")] = False, wait_for_completion: Annotated[bool, Field(description="Whether to wait for restore completion")] = True, rename_pattern: Annotated[ Optional[str], Field(description="Pattern to rename restored indices (e.g., 'restored_%s')")] = None, index_settings: Annotated[Optional[str], Field(description="JSON string of index settings to override")] = None ) -> str:
• src/elasticsearch/sub_servers/__init__.py:44-44 (registration)

  Documentation of tool count including restore_snapshot in sub_servers package.
  "elasticsearch_snapshots": 3, # create_snapshot, restore_snapshot, list_snapshots