MCP Kafka Schema Reg

Instructions

Implementation Reference

• registry_management_tools.py:83-126 (handler)

  Handler function for the get_registry_info tool, decorated with @structured_output to enable structured responses and schema validation.

  @structured_output("get_registry_info", fallback_on_error=True)
  def get_registry_info_tool(registry_manager, registry_mode: str, registry: Optional[str] = None) -> Dict[str, Any]:
      """
      Get detailed information about a specific registry with structured validation and resource links.
  
      Args:
          registry_manager: The registry manager instance
          registry_mode: Current registry mode (single/multi)
          registry: Optional registry name
  
      Returns:
          Dictionary containing detailed registry information with structured validation and navigation links
      """
      try:
          if registry_mode == "single" and not registry:
              registry = registry_manager.get_default_registry()
  
          info = registry_manager.get_registry_info(registry)
          if info is None:
              return create_error_response(
                  f"Registry '{registry}' not found",
                  error_code="REGISTRY_NOT_FOUND",
                  registry_mode=registry_mode,
              )
  
          # Add structured output metadata
          info["registry_mode"] = registry_mode
          info["mcp_protocol_version"] = "2025-06-18"
  
          # Add additional metadata for better context
          info["_metadata"] = {
              "queried_registry": registry,
              "is_default": registry == registry_manager.get_default_registry(),
              "query_timestamp": __import__("datetime").datetime.now().isoformat(),
          }
  
          # Add resource links
          registry_name_for_linking = _get_registry_name_for_linking(registry_mode, registry)
          info = add_links_to_response(info, "registry", registry_name_for_linking)
  
          return info
      except Exception as e:
          return create_error_response(str(e), error_code="REGISTRY_INFO_FAILED", registry_mode=registry_mode)

• schema_definitions.py:248-277 (schema)

  JSON Schema definition for the output of the get_registry_info tool (REGISTRY_INFO_SCHEMA). Mapped in TOOL_OUTPUT_SCHEMAS at line 924.

  REGISTRY_INFO_SCHEMA = {
      "type": "object",
      "properties": {
          "name": {"type": "string", "description": "Registry name"},
          "url": {"type": "string", "format": "uri", "description": "Registry URL"},
          "status": {
              "type": "string",
              "enum": ["connected", "disconnected", "error", "unknown"],
              "description": "Connection status",
          },
          "auth_type": {
              "type": "string",
              "enum": ["none", "basic", "oauth", "ssl"],
              "description": "Authentication type",
          },
          "viewonly": {
              "type": "boolean",
              "description": "Whether registry is in viewonly mode",
          },
          "version": {"type": "string", "description": "Schema Registry server version"},
          "capabilities": {
              "type": "array",
              "items": {"type": "string"},
              "description": "Server capabilities",
          },
          **METADATA_FIELDS,
      },
      "required": ["name", "url"],
      "additionalProperties": True,
  }

• schema_definitions.py:914-959 (schema)

  TOOL_OUTPUT_SCHEMAS dictionary mapping "get_registry_info" to its output schema (line 924). Used for MCP tool schema validation.

  # Schema Operations
  "register_schema": REGISTER_SCHEMA_SCHEMA,
  "get_schema": GET_SCHEMA_SCHEMA,
  "get_schema_versions": GET_SCHEMA_VERSIONS_SCHEMA,
  "get_schema_by_id": GET_SCHEMA_BY_ID_SCHEMA,
  "get_subjects_by_schema_id": GET_SUBJECTS_BY_SCHEMA_ID_SCHEMA,
  "check_compatibility": CHECK_COMPATIBILITY_SCHEMA,
  "list_subjects": LIST_SUBJECTS_SCHEMA,
  # Registry Management
  "list_registries": LIST_REGISTRIES_SCHEMA,
  "get_registry_info": REGISTRY_INFO_SCHEMA,
  "test_registry_connection": TEST_CONNECTION_SCHEMA,
  "test_all_registries": TEST_ALL_REGISTRIES_SCHEMA,
  # Configuration Management
  "get_global_config": CONFIG_SCHEMA,
  "update_global_config": CONFIG_SCHEMA,
  "get_subject_config": CONFIG_SCHEMA,
  "update_subject_config": CONFIG_SCHEMA,
  "delete_subject_alias": {
      "oneOf": [
          {
              "type": "object",
              "properties": {
                  "alias": {"type": "string", "description": "Alias subject name"},
                  "deleted": {"type": "boolean", "description": "Whether alias was deleted"},
                  "registry": {"type": "string", "description": "Registry name (multi-registry mode)"},
                  **METADATA_FIELDS,
              },
              "required": ["alias", "deleted"],
              "additionalProperties": True,
          },
          ERROR_RESPONSE_SCHEMA,
      ],
  },
  "add_subject_alias": {
      "oneOf": [
          {
              "type": "object",
              "properties": {
                  "alias": {"type": "string", "description": "New alias subject name"},
                  "target": {"type": "string", "description": "Existing subject target"},
                  "registry": {"type": "string", "description": "Registry name (multi-registry mode)"},
                  **METADATA_FIELDS,
              },
              "required": ["alias", "target"],
              "additionalProperties": True,

• registry_management_tools.py:18-83 (registration)

  Decorator @structured_output("get_registry_info") registers the tool name and enables structured output with schema validation per MCP spec.

  )
  
  
  def _get_registry_name_for_linking(registry_mode: str, registry_name: Optional[str] = None) -> str:
      """Helper function to get registry name for linking."""
      if registry_mode == "single":
          return "default"
      elif registry_name:
          return registry_name
      else:
          return "unknown"
  
  
  @structured_output("list_registries", fallback_on_error=True)
  def list_registries_tool(registry_manager, registry_mode: str) -> Dict[str, Any]:
      """
      List all configured Schema Registry instances with structured validation and resource links.
  
      Args:
          registry_manager: The registry manager instance
          registry_mode: Current registry mode (single/multi)
  
      Returns:
          Dictionary containing registry information with structured validation and navigation links
      """
      try:
          registries_list = []
          for name in registry_manager.list_registries():
              info = registry_manager.get_registry_info(name)
              if info:
                  # Add structured output metadata
                  info["registry_mode"] = registry_mode
                  info["mcp_protocol_version"] = "2025-06-18"
  
                  # Add resource links for each registry
                  info = add_links_to_response(info, "registry", name)
  
                  registries_list.append(info)
  
          # Convert to enhanced response format
          result = {
              "registries": registries_list,
              "total_count": len(registries_list),
              "registry_mode": registry_mode,
              "mcp_protocol_version": "2025-06-18",
          }
  
          # If no registries found, add helpful message
          if not registries_list:
              result["message"] = "No registries configured"
              result["help"] = {
                  "single_mode": "Set SCHEMA_REGISTRY_URL environment variable",
                  "multi_mode": "Set SCHEMA_REGISTRY_NAME_1, SCHEMA_REGISTRY_URL_1 environment variables",
              }
  
          # Add default registry information
          default_registry = registry_manager.get_default_registry()
          if default_registry:
              result["default_registry"] = default_registry
  
          return result
      except Exception as e:
          return create_error_response(str(e), error_code="REGISTRY_LIST_FAILED", registry_mode=registry_mode)
  
  
  @structured_output("get_registry_info", fallback_on_error=True)