Agent Knowledge MCP

Instructions

Implementation Reference

• src/elasticsearch/sub_servers/elasticsearch_snapshots.py:24-189 (handler)

  The primary handler for the 'create_snapshot' tool. This async function, decorated with @app.tool(), implements the complete logic: validates inputs via Annotated Fields (serving as schema), creates/checks repository, executes snapshot via Elasticsearch client, handles wait_for_completion, provides detailed success/error messages with emojis, suggestions, and technical details.

  @app.tool(
      description="Create a snapshot (backup) of Elasticsearch indices with comprehensive options and repository management",
      tags={"elasticsearch", "snapshot", "backup", "repository"}
  )
  async def create_snapshot(
          snapshot_name: Annotated[str, Field(description="Name for the snapshot (must be unique)")],
          repository: Annotated[str, Field(description="Repository name to store the snapshot")] = "backup_repository",
          indices: Annotated[Optional[str], Field(
              description="Comma-separated list of indices to backup (default: all indices)")] = None,
          ignore_unavailable: Annotated[bool, Field(description="Whether to ignore unavailable indices")] = True,
          include_global_state: Annotated[bool, Field(description="Whether to include cluster global state")] = True,
          wait_for_completion: Annotated[bool, Field(description="Whether to wait for snapshot completion")] = True,
          description: Annotated[Optional[str], Field(description="Optional description for the snapshot")] = None
  ) -> str:
      """Create a snapshot (backup) of Elasticsearch indices."""
      try:
          es = get_es_client()
  
          # Check if repository exists, create if not
          try:
              repo_info = es.snapshot.get_repository(repository=repository)
          except:
              # Repository doesn't exist, create default file system repository
              repo_body = {
                  "type": "fs",
                  "settings": {
                      "location": f"/usr/share/elasticsearch/snapshots/{repository}",
                      "compress": True
                  }
              }
              try:
                  es.snapshot.create_repository(repository=repository, body=repo_body)
                  repo_created = True
              except Exception as repo_error:
                  return (f"❌ Failed to create snapshot repository:\n\n" +
                          f"🔧 **Repository Error**: Cannot create repository '{repository}'\n" +
                          f"📍 **Issue**: {str(repo_error)}\n\n" +
                          f"💡 **Common Solutions**:\n" +
                          f"   1. Ensure Elasticsearch has write permissions to snapshot directory\n" +
                          f"   2. Add 'path.repo: [\"/usr/share/elasticsearch/snapshots\"]' to elasticsearch.yml\n" +
                          f"   3. Restart Elasticsearch after configuration change\n" +
                          f"   4. Or use existing repository name\n\n" +
                          f"🔍 **Technical Details**: {str(repo_error)}")
          else:
              repo_created = False
  
          # Parse indices parameter
          if indices:
              indices_list = [idx.strip() for idx in indices.split(',')]
              indices_param = ','.join(indices_list)
          else:
              indices_param = "*"  # All indices
              indices_list = ["*"]
  
          # Create snapshot metadata
          snapshot_body = {
              "indices": indices_param,
              "ignore_unavailable": ignore_unavailable,
              "include_global_state": include_global_state
          }
  
          if description:
              snapshot_body["metadata"] = {
                  "description": description,
                  "created_by": "AgentKnowledgeMCP",
                  "created_at": datetime.now().isoformat()
              }
  
          # Create the snapshot
          snapshot_result = es.snapshot.create(
              repository=repository,
              snapshot=snapshot_name,
              body=snapshot_body,
              wait_for_completion=wait_for_completion
          )
  
          # Format response based on completion
          if wait_for_completion:
              snapshot_info = snapshot_result.get('snapshot', {})
              state = snapshot_info.get('state', 'UNKNOWN')
  
              if state == 'SUCCESS':
                  status_emoji = "✅"
                  status_msg = "Successfully completed"
              elif state == 'PARTIAL':
                  status_emoji = "⚠️"
                  status_msg = "Partially completed with some issues"
              elif state == 'FAILED':
                  status_emoji = "❌"
                  status_msg = "Failed to complete"
              else:
                  status_emoji = "🔄"
                  status_msg = f"Status: {state}"
  
              result_message = (f"{status_emoji} Snapshot '{snapshot_name}' {status_msg}!\n\n" +
                                f"📸 **Snapshot Details**:\n" +
                                f"   📂 Repository: {repository}\n" +
                                f"   📋 Name: {snapshot_name}\n" +
                                f"   📊 State: {state}\n" +
                                f"   📦 Indices: {', '.join(indices_list)}\n" +
                                f"   🌐 Global State: {'Included' if include_global_state else 'Excluded'}\n")
  
              if snapshot_info.get('shards'):
                  shards = snapshot_info['shards']
                  result_message += (f"   🔢 Shards: {shards.get('total', 0)} total, " +
                                     f"{shards.get('successful', 0)} successful, " +
                                     f"{shards.get('failed', 0)} failed\n")
  
              if snapshot_info.get('start_time_in_millis') and snapshot_info.get('end_time_in_millis'):
                  duration = (snapshot_info['end_time_in_millis'] - snapshot_info['start_time_in_millis']) / 1000
                  result_message += f"   ⏱️ Duration: {duration:.2f} seconds\n"
  
              if description:
                  result_message += f"   📝 Description: {description}\n"
  
          else:
              result_message = (f"🔄 Snapshot '{snapshot_name}' started!\n\n" +
                                f"📸 **Snapshot Details**:\n" +
                                f"   📂 Repository: {repository}\n" +
                                f"   📋 Name: {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 repo_created:
              result_message += f"\n🆕 **Repository Created**: Created new repository '{repository}'\n"
  
          result_message += (f"\n✅ **Success Actions**:\n" +
                             f"   📸 Snapshot backup is {'completed' if wait_for_completion else 'in progress'}\n" +
                             f"   🔍 Use 'list_snapshots' to view all snapshots\n" +
                             f"   🔄 Use 'restore_snapshot' to restore from this backup\n" +
                             f"   📊 Check snapshot status with repository '{repository}'\n\n" +
                             f"💾 **Backup Strategy**:\n" +
                             f"   🕒 Regular snapshots help protect against data loss\n" +
                             f"   🏷️ Use descriptive snapshot names with dates\n" +
                             f"   📂 Monitor repository storage space\n" +
                             f"   🧹 Clean up old snapshots periodically")
  
          return result_message
  
      except Exception as e:
          error_message = "❌ Failed to create 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 or misconfigured\n"
              error_message += f"📍 Check repository configuration and permissions\n"
              error_message += f"💡 Try: Use different repository name or check path.repo settings\n\n"
          elif "invalid_snapshot_name" in error_str:
              error_message += f"🏷️ **Naming Error**: Invalid snapshot name '{snapshot_name}'\n"
              error_message += f"📍 Snapshot names must be lowercase and cannot contain certain characters\n"
              error_message += f"💡 Try: Use alphanumeric characters and hyphens only\n\n"
          elif "already_exists" in error_str:
              error_message += f"📋 **Conflict Error**: Snapshot '{snapshot_name}' already exists\n"
              error_message += f"📍 Each snapshot must have a unique name within the repository\n"
              error_message += f"💡 Try: Use different snapshot name or delete existing snapshot\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-53 (registration)

  Registration of the elasticsearch_snapshots sub-server (containing create_snapshot) into the main unified Elasticsearch FastMCP app via app.mount(snapshots_app). This mounts all 6 sub-servers, making the tool available in the primary 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
  
  print("✅ All 6 sub-servers mounted successfully! Total: 16 tools available")

• src/elasticsearch/sub_servers/elasticsearch_snapshots.py:28-37 (schema)

  Input schema defined via function parameters with typing.Annotated and pydantic.Field for validation: snapshot_name (required str), repository (str default), indices (opt str), ignore_unavailable (bool), include_global_state (bool), wait_for_completion (bool), description (opt str). Output: str status message.

  async def create_snapshot(
          snapshot_name: Annotated[str, Field(description="Name for the snapshot (must be unique)")],
          repository: Annotated[str, Field(description="Repository name to store the snapshot")] = "backup_repository",
          indices: Annotated[Optional[str], Field(
              description="Comma-separated list of indices to backup (default: all indices)")] = None,
          ignore_unavailable: Annotated[bool, Field(description="Whether to ignore unavailable indices")] = True,
          include_global_state: Annotated[bool, Field(description="Whether to include cluster global state")] = True,
          wait_for_completion: Annotated[bool, Field(description="Whether to wait for snapshot completion")] = True,
          description: Annotated[Optional[str], Field(description="Optional description for the snapshot")] = None
  ) -> str:

• src/elasticsearch/sub_servers/elasticsearch_snapshots.py:24-27 (registration)

  Initial tool registration on the sub-server FastMCP app instance via @app.tool() decorator with description and tags.

  @app.tool(
      description="Create a snapshot (backup) of Elasticsearch indices with comprehensive options and repository management",
      tags={"elasticsearch", "snapshot", "backup", "repository"}
  )