bulk_schema_migration

Migrate schemas between contexts or registries with pattern-based selection, preserving schema IDs. Preview changes and enable rollback for controlled migration processes.

Instructions

Migrate schemas between contexts or registries.

Supports pattern-based selection and maintains schema IDs. Includes preview and rollback capabilities.

Input Schema

TableJSON Schema

Name	Required	Description	Default
`dry_run`	No
`preserve_ids`	No
`schema_pattern`	No
`source_context`	No
`source_registry`	No
`target_context`	No
`target_registry`	No

Implementation Reference

bulk_operations_mcp_integration.py:128-152 (schema)

Input schema defining parameters for bulk schema migration tool, including source/target contexts/registries, pattern, and options.

inputSchema={
    "type": "object",
    "properties": {
        "source_context": {"type": "string", "description": "Source context (default: current context)"},
        "target_context": {"type": "string", "description": "Target context for migration"},
        "source_registry": {
            "type": "string",
            "description": "Source registry (for cross-registry migration)",
        },
        "target_registry": {
            "type": "string",
            "description": "Target registry (for cross-registry migration)",
        },
        "schema_pattern": {"type": "string", "description": "Pattern to match schemas for migration"},
        "preserve_ids": {
            "type": "boolean",
            "default": True,
            "description": "Preserve schema IDs during migration",
        },
        "dry_run": {
            "type": "boolean",
            "default": True,
            "description": "Preview migration without executing",
        },
    },

bulk_operations_mcp_integration.py:120-155 (registration)

Registers the 'bulk_schema_migration' MCP tool with name, description, and input schema.

tools.append(
    Tool(
        name="bulk_schema_migration",
        description=(
            "Migrate schemas between contexts or registries. "
            "Supports pattern-based selection and maintains schema IDs. "
            "Includes preview and rollback capabilities."
        ),
        inputSchema={
            "type": "object",
            "properties": {
                "source_context": {"type": "string", "description": "Source context (default: current context)"},
                "target_context": {"type": "string", "description": "Target context for migration"},
                "source_registry": {
                    "type": "string",
                    "description": "Source registry (for cross-registry migration)",
                },
                "target_registry": {
                    "type": "string",
                    "description": "Target registry (for cross-registry migration)",
                },
                "schema_pattern": {"type": "string", "description": "Pattern to match schemas for migration"},
                "preserve_ids": {
                    "type": "boolean",
                    "default": True,
                    "description": "Preserve schema IDs during migration",
                },
                "dry_run": {
                    "type": "boolean",
                    "default": True,
                    "description": "Preview migration without executing",
                },
            },
        },
    )
)

bulk_operations_mcp_integration.py:274-276 (handler)

MCP tool handler dispatcher that invokes the BulkOperationsWizard for MIGRATION operation.

elif tool_name == "bulk_schema_migration":
    # Direct migration with parameters
    return await wizard.start_wizard(BulkOperationType.MIGRATION)

bulk_operations_wizard.py:159-184 (handler)

Core implementation of bulk schema migration: elicits source/target, schemas, options; generates preview; confirms and executes via generic bulk operation executor.

async def _handle_bulk_migration(self) -> Dict[str, Any]:
    """Handle bulk migration operations"""
    # Step 1: Select source and target
    source_target = await self._elicit_migration_source_target()  # type: ignore

    # Step 2: Select schemas to migrate
    schemas = await self._elicit_schema_selection(
        "Select schemas to migrate", context=source_target["source_context"]
    )

    # Step 3: Migration options
    migration_options = await self._elicit_migration_options()  # type: ignore

    # Step 4: Preview migration
    preview = await self._generate_preview(
        BulkOperationType.MIGRATION, schemas, {**source_target, **migration_options}
    )

    # Step 5: Confirm operation
    if not await self._confirm_operation(preview):
        return {"status": "cancelled", "reason": "User cancelled operation"}

    # Step 6: Execute migration
    return await self._execute_bulk_operation(
        BulkOperationType.MIGRATION, schemas, {**source_target, **migration_options}, preview
    )

bulk_operations_wizard.py:25-32 (helper)

Enum defining operation types, including MIGRATION used by the bulk_schema_migration tool.

class BulkOperationType(Enum):
    """Supported bulk operation types"""

    SCHEMA_UPDATE = "schema_update"
    MIGRATION = "migration"
    CLEANUP = "cleanup"
    CONFIGURATION = "configuration"

MCP Kafka Schema Reg