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