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