bulk_operations_wizard
Streamline schema management with an interactive wizard for bulk operations, enabling updates, migrations, cleanup, and configuration changes across multiple schemas. Supports dry runs for validation.
Instructions
Start the interactive Bulk Operations Wizard for admin tasks.
Guides through safe execution of operations across multiple schemas. Supports schema updates, migrations, cleanup, and configuration changes.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| dry_run | No | ||
| operation_type | No |
Implementation Reference
- Handler function for MCP bulk operations tools. For 'bulk_operations_wizard', extracts operation_type from arguments and calls wizard.start_wizard(). Handles multiple related tools.
async def handle_bulk_operations_tool( wizard: BulkOperationsWizard, tool_name: str, arguments: Dict[str, Any] ) -> Dict[str, Any]: """Handle bulk operations tool calls""" if tool_name == "bulk_operations_wizard": # Start the main wizard operation_type = arguments.get("operation_type") if operation_type: operation_type = BulkOperationType(operation_type) return await wizard.start_wizard(operation_type) elif tool_name == "bulk_schema_update": # Direct schema update with parameters # In real implementation, would pass parameters to wizard return await wizard.start_wizard(BulkOperationType.SCHEMA_UPDATE) elif tool_name == "bulk_schema_cleanup": # Direct cleanup with parameters return await wizard.start_wizard(BulkOperationType.CLEANUP) elif tool_name == "bulk_schema_migration": # Direct migration with parameters return await wizard.start_wizard(BulkOperationType.MIGRATION) elif tool_name == "bulk_configuration_update": # Direct configuration update return await wizard.start_wizard(BulkOperationType.CONFIGURATION) elif tool_name == "get_bulk_operation_status": # Get task status task_id = arguments["task_id"] return await wizard.task_manager.get_task_status(task_id) elif tool_name == "cancel_bulk_operation": # Cancel operation task_id = arguments["task_id"] rollback = arguments.get("rollback", True) # In real implementation, would cancel the task return {"status": "cancelled", "task_id": task_id, "rollback": rollback} elif tool_name == "preview_bulk_operation": # Preview operation operation_type = BulkOperationType(arguments["operation_type"]) items = arguments["items"] params = arguments.get("parameters", {}) preview = await wizard._generate_preview(operation_type, items, params) return { "preview": { "affected_items": preview.affected_items, "total_count": preview.total_count, "changes_summary": preview.changes_summary, "estimated_duration": preview.estimated_duration, "warnings": preview.warnings, "consumer_impact": preview.consumer_impact, } } else: raise ValueError(f"Unknown tool: {tool_name}") - bulk_operations_mcp_integration.py:22-44 (registration)MCP Tool registration for 'bulk_operations_wizard' including name, description, and input schema.
Tool( name="bulk_operations_wizard", description=( "Start the interactive Bulk Operations Wizard for admin tasks. " "Guides through safe execution of operations across multiple schemas. " "Supports schema updates, migrations, cleanup, and configuration changes." ), inputSchema={ "type": "object", "properties": { "operation_type": { "type": "string", "enum": ["schema_update", "migration", "cleanup", "configuration"], "description": "Pre-select operation type (optional)", }, "dry_run": { "type": "boolean", "default": True, "description": "Run in preview mode without making changes", }, }, }, ) - Input schema definition for the bulk_operations_wizard tool.
inputSchema={ "type": "object", "properties": { "operation_type": { "type": "string", "enum": ["schema_update", "migration", "cleanup", "configuration"], "description": "Pre-select operation type (optional)", }, "dry_run": { "type": "boolean", "default": True, "description": "Run in preview mode without making changes", }, }, - bulk_operations_wizard.py:86-106 (handler)Core implementation of the bulk operations wizard logic, called by the MCP tool handler. Dispatches to specific operation handlers based on type.
async def start_wizard(self, operation_type: Optional[BulkOperationType] = None) -> Dict[str, Any]: """ Start the bulk operations wizard Args: operation_type: Pre-selected operation type, or None to prompt user Returns: Operation result """ # Select operation type if not provided if not operation_type: operation_type = await self._elicit_operation_type() # Get operation handler handler = self._operation_handlers.get(operation_type) if not handler: raise ValueError(f"Unknown operation type: {operation_type}") # Execute the operation workflow return await handler() # type: ignore - bulk_operations_wizard.py:77-84 (helper)Registers the handler methods for different BulkOperationType in the wizard class.
def _register_handlers(self) -> Dict[BulkOperationType, Callable]: """Register operation handlers""" return { BulkOperationType.SCHEMA_UPDATE: self._handle_bulk_schema_update, BulkOperationType.MIGRATION: self._handle_bulk_migration, BulkOperationType.CLEANUP: self._handle_bulk_cleanup, BulkOperationType.CONFIGURATION: self._handle_bulk_configuration, }