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"} )