MCP Kafka Schema Reg

Instructions

Implementation Reference

• workflow_mcp_integration.py:191-242 (handler)

  The MCP tool handler for guided_schema_evolution. Decorated with @self.mcp.tool and implements the core logic by initializing context from inputs and starting the schema_evolution_assistant workflow via the MultiStepElicitationManager.
  @self.mcp.tool( description=( "Start the Schema Evolution Assistant workflow. " "This guided workflow helps you safely evolve schemas by analyzing changes, " "suggesting strategies, and coordinating consumer updates." ), ) async def guided_schema_evolution( subject: Optional[str] = None, current_schema: Optional[str] = None, ) -> str: """Start the Schema Evolution Assistant workflow.""" initial_context = {} if subject: initial_context["subject"] = subject if current_schema: try: initial_context["current_schema"] = json.loads(current_schema) except json.JSONDecodeError: initial_context["current_schema"] = current_schema workflow_id = "schema_evolution_assistant" try: request = await self.multi_step_manager.start_workflow( workflow_id=workflow_id, initial_context=initial_context ) if request: return json.dumps( { "status": "started", "workflow_id": workflow_id, "workflow_name": "Schema Evolution Assistant", "request_id": request.id, "first_step": request.title, "description": request.description, "message": ( "Schema Evolution Assistant started. This workflow will guide you through:\n" "1. Analyzing schema changes\n" "2. Detecting breaking changes\n" "3. Selecting evolution strategy\n" "4. Planning consumer coordination\n" "5. Setting up rollback procedures" ), } ) else: return json.dumps({"error": "Failed to start Schema Evolution workflow"}) except Exception as e: logger.error(f"Error starting Schema Evolution workflow: {str(e)}") return json.dumps({"error": f"Failed to start workflow: {str(e)}"})
• workflow_mcp_integration.py:489-573 (helper)

  Helper function invoked upon workflow completion to process responses and generate a comprehensive schema evolution execution plan, capturing change analysis, strategy configurations, compatibility overrides, consumer coordination details, rollback procedures, and execution options.
  def execute_schema_evolution(responses: Dict[str, Any]) -> Dict[str, Any]: """Execute schema evolution based on workflow responses.""" result = { "operation": "schema_evolution", "subject": responses.get("subject"), "status": "pending", } # Basic change information result["change_info"] = { "change_type": responses.get("change_type"), "description": responses.get("change_description"), "current_consumers": responses.get("current_consumers"), "production_impact": responses.get("production_impact"), "has_breaking_changes": responses.get("has_breaking_changes") == "true", } # Evolution strategy strategy = responses.get("evolution_strategy") result["evolution_strategy"] = strategy # Strategy-specific configuration if strategy == "multi_version_migration": result["migration_config"] = { "intermediate_versions": int(responses.get("intermediate_versions", 1)), "version_timeline": responses.get("version_timeline"), "deprecation_strategy": responses.get("deprecation_strategy"), } elif strategy == "dual_support": result["dual_support_config"] = { "support_duration": responses.get("support_duration"), "field_mapping": responses.get("field_mapping"), "conversion_logic": responses.get("conversion_logic"), } elif strategy == "gradual_migration": result["migration_phases"] = { "phase_count": responses.get("phase_count"), "phase_criteria": responses.get("phase_criteria"), "rollback_checkpoints": responses.get("rollback_checkpoints") == "true", } else: result["implementation"] = { "deployment_window": responses.get("deployment_window"), "validation_approach": responses.get("validation_approach"), } # Compatibility resolution (if breaking changes) change_info = result.get("change_info", {}) if isinstance(change_info, dict) and change_info.get("has_breaking_changes"): result["compatibility_resolution"] = { "approach": responses.get("resolution_approach"), "override_compatibility": responses.get("compatibility_override") == "true", "notes": responses.get("compatibility_notes"), } # Consumer coordination result["consumer_coordination"] = { "notification_method": responses.get("notification_method"), "testing_approach": responses.get("consumer_testing"), "support_period": responses.get("support_period"), } # Rollback planning result["rollback_plan"] = { "trigger": responses.get("rollback_trigger"), "max_time": responses.get("rollback_time"), "data_handling": responses.get("data_handling"), "test_rollback": responses.get("rollback_testing") == "true", } # Final options result["documentation"] = { "generate_migration_guide": responses.get("generate_migration_guide") == "true", "create_runbook": responses.get("create_runbook") == "true", "schedule_dry_run": responses.get("schedule_dry_run") == "true", "evolution_notes": responses.get("evolution_notes"), } # Execution settings result["execution"] = { "confirmed": responses.get("final_confirmation") == "true", "enable_monitoring": responses.get("monitor_execution") == "true", } return result
• workflow_mcp_integration.py:573-576 (helper)

  Core analysis helper that detects and categorizes schema changes (add_field, remove_field, type changes, etc.), determines if breaking, handles Avro complex types (records, arrays, maps, unions), and supports compatibility checking used in the schema evolution workflow.
  return result def analyze_schema_changes(current_schema: Dict[str, Any], proposed_schema: Dict[str, Any]) -> List[Dict[str, Any]]: