MCP Kafka Schema Reg

Instructions

Implementation Reference

• interactive_tools.py:345-435 (handler)

  The core handler function implementing the check_compatibility_interactive MCP tool. It first runs a standard compatibility check using the injected check_compatibility_tool. If issues are detected, it creates an elicitation request using create_compatibility_resolution_elicitation to gather user preferences for resolution, waits for response with fallback, and augments the result with resolution guidance. Handles errors gracefully.
  async def check_compatibility_interactive( subject: str, schema_definition: Dict, schema_type: str = "AVRO", context: Optional[str] = None, registry: Optional[str] = None, # Core tool dependencies injected check_compatibility_tool=None, registry_manager=None, registry_mode=None, auth=None, headers=None, schema_registry_url=None, ) -> Dict[str, Any]: """ Interactive compatibility checking with elicitation for resolution options. When compatibility issues are found, this tool will elicit resolution preferences from the user. """ try: # First, perform the standard compatibility check compatibility_result = check_compatibility_tool( subject=subject, schema_definition=schema_definition, registry_manager=registry_manager, registry_mode=registry_mode, schema_type=schema_type, context=context, registry=registry, auth=auth, headers=headers, schema_registry_url=schema_registry_url, ) # Check if there are compatibility issues if isinstance(compatibility_result, dict): is_compatible = compatibility_result.get("compatible", False) compatibility_errors = compatibility_result.get("messages", []) if not is_compatible and compatibility_errors: logger.info(f"Compatibility issues found for subject '{subject}': {compatibility_errors}") # Create elicitation request for resolution strategy elicitation_request = create_compatibility_resolution_elicitation( subject=subject, compatibility_errors=compatibility_errors ) # Store the request for processing await elicitation_manager.create_request(elicitation_request) # Attempt elicitation with fallback response = await elicit_with_fallback(elicitation_request) if response and response.complete: # Add resolution guidance to the result compatibility_result["resolution_guidance"] = { "strategy": response.values.get("resolution_strategy"), "compatibility_level": response.values.get("compatibility_level"), "notes": response.values.get("notes"), "elicitation_used": True, } logger.info( f"Added compatibility resolution guidance: {response.values.get('resolution_strategy')}" ) else: # Add default guidance compatibility_result["resolution_guidance"] = { "strategy": "modify_schema", "compatibility_level": "BACKWARD", "notes": "Default resolution strategy (elicitation failed)", "elicitation_used": False, } else: # No issues, add successful indicator compatibility_result["resolution_guidance"] = { "strategy": "none_needed", "notes": "Schema is compatible, no resolution required", "elicitation_used": False, } return compatibility_result except Exception as e: logger.error(f"Error in interactive compatibility check: {str(e)}") return create_error_response( f"Interactive compatibility check failed: {str(e)}", error_code="INTERACTIVE_COMPATIBILITY_ERROR", )