MCP Kafka Schema Reg

Instructions

Implementation Reference

• core_registry_tools.py:279-384 (handler)

  Main handler function for the 'get_schema_versions' tool. Performs HTTP GET request to the Schema Registry API endpoint '/subjects/{subject}/versions' (with optional context prefix), handles single/multi-registry modes, formats response with metadata, HATEOAS links, and MCP protocol compliance.

  @structured_output("get_schema_versions", fallback_on_error=True)
  def get_schema_versions_tool(
      subject: str,
      registry_manager,
      registry_mode: str,
      context: Optional[str] = None,
      registry: Optional[str] = None,
      auth=None,
      headers=None,
      schema_registry_url: str = "",
  ) -> Dict[str, Any]:
      """
      Get all versions of a schema for a subject.
  
      Args:
          subject: The subject name
          context: Optional schema context
          registry: Optional registry name (ignored in single-registry mode)
  
      Returns:
          Dictionary containing version numbers with structured validation and resource links
      """
      try:
          if registry_mode == "single":
              # Single-registry mode: use secure session approach
              client = registry_manager.get_registry()
              if client is None:
                  return create_error_response(
                      "No default registry configured",
                      error_code="REGISTRY_NOT_FOUND",
                      registry_mode="single",
                  )
  
              url = client.build_context_url(f"/subjects/{subject}/versions", context)
  
              response = client.session.get(url, auth=client.auth, headers=client.headers)
  
              # Handle 404 specifically - subject doesn't exist
              if response.status_code == 404:
                  versions_list = []
              else:
                  response.raise_for_status()
                  versions_list = response.json()
  
              # Convert to enhanced response format
              result = {
                  "subject": subject,
                  "versions": versions_list,
                  "registry_mode": "single",
                  "mcp_protocol_version": "2025-06-18",
              }
  
              # Add resource links
              registry_name = _get_registry_name(registry_mode, registry)
              result = add_links_to_response(
                  result,
                  "schema_versions",
                  registry_name,
                  subject=subject,
                  context=context,
              )
  
              return result
          else:
              # Multi-registry mode: use client approach
              client = registry_manager.get_registry(registry)
              if client is None:
                  return create_error_response(
                      f"Registry '{registry}' not found",
                      error_code="REGISTRY_NOT_FOUND",
                      registry_mode="multi",
                  )
  
              url = client.build_context_url(f"/subjects/{subject}/versions", context)
  
              response = client.session.get(url, auth=client.auth, headers=client.headers)
  
              # Handle 404 specifically - subject doesn't exist
              if response.status_code == 404:
                  versions_list = []
              else:
                  response.raise_for_status()
                  versions_list = response.json()
  
              # Convert to enhanced response format
              result = {
                  "subject": subject,
                  "versions": versions_list,
                  "registry": client.config.name,
                  "registry_mode": "multi",
                  "mcp_protocol_version": "2025-06-18",
              }
  
              # Add resource links
              result = add_links_to_response(
                  result,
                  "schema_versions",
                  client.config.name,
                  subject=subject,
                  context=context,
              )
  
              return result
      except Exception as e:
          return create_error_response(str(e), error_code="VERSION_RETRIEVAL_FAILED", registry_mode=registry_mode)

• schema_definitions.py:154-179 (schema)

  JSON Schema definition for the 'get_schema_versions' tool output, including subject, versions array, optional registry and _links for HATEOAS.

  GET_SCHEMA_VERSIONS_SCHEMA = {
      "type": "object",
      "properties": {
          "subject": {
              "type": "string",
              "description": "Subject name for which versions are listed",
          },
          "versions": {
              "type": "array",
              "items": {"type": "integer", "minimum": 1},
              "description": "List of available schema versions",
          },
          "registry": {
              "type": "string",
              "description": "Registry name (multi-registry mode)",
          },
          "_links": {
              "type": "object",
              "description": "Navigation links to related resources",
              "additionalProperties": True,
          },
          **METADATA_FIELDS,
      },
      "required": ["subject", "versions"],
      "additionalProperties": True,
  }

• schema_definitions.py:917-919 (registration)

  Registration of the output schema for 'get_schema_versions' tool in the master TOOL_OUTPUT_SCHEMAS dictionary used for MCP tool validation.

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

• statistics_tools.py:177-384 (helper)

  Helper function get_schema_versions imported from main MCP server module and used in statistics tools for retrieving schema versions.

          from kafka_schema_registry_unified_mcp import get_schema_versions
  
          versions = get_schema_versions(subject, context, registry)
          if isinstance(versions, dict) and "error" in versions:
              return create_error_response(
                  f"Failed to get schema versions: {versions.get('error')}",
                  error_code="SCHEMA_VERSIONS_RETRIEVAL_FAILED",
                  registry_mode=registry_mode,
              )
  
          # Get registry metadata
          metadata = client.get_server_metadata()
  
          result = {
              "registry": (client.config.name if hasattr(client.config, "name") else "default"),
              "context": context or "default",
              "subject": subject,
              "count": len(versions),
              "scope": "versions",
              "versions": versions,
              "counted_at": datetime.now().isoformat(),
              "registry_mode": registry_mode,
              "mcp_protocol_version": "2025-06-18",
          }
  
          # Add metadata information, but preserve the scope field
          metadata_copy = metadata.copy()
          if "scope" in metadata_copy:
              # Preserve the simple string scope, but add server scope info separately
              metadata_copy["server_scope"] = metadata_copy.pop("scope")
          result.update(metadata_copy)
  
          return result
      except Exception as e:
          return create_error_response(str(e), error_code="VERSION_COUNT_FAILED", registry_mode=registry_mode)
  
  
  @structured_output("get_registry_statistics", fallback_on_error=True)
  def get_registry_statistics_tool(
      registry_manager,
      registry_mode: str,
      registry: Optional[str] = None,
      include_context_details: bool = True,
  ) -> Dict[str, Any]:
      """
      Get comprehensive statistics about a registry.
  
      Args:
          registry: Optional registry name (ignored in single-registry mode)
          include_context_details: Whether to include detailed context statistics
  
      Returns:
          Dictionary containing registry statistics with metadata and structured validation
      """
      try:
          if registry_mode == "single":
              client = get_default_client(registry_manager)
          else:
              client = registry_manager.get_registry(registry)
              if client is None:
                  return create_error_response(
                      f"Registry '{registry}' not found",
                      error_code="REGISTRY_NOT_FOUND",
                      registry_mode=registry_mode,
                  )
  
          # Get all contexts
          contexts = client.get_contexts()
          if isinstance(contexts, dict) and "error" in contexts:
              return create_error_response(
                  f"Failed to get contexts: {contexts.get('error')}",
                  error_code="CONTEXTS_RETRIEVAL_FAILED",
                  registry_mode=registry_mode,
              )
  
          total_schemas = 0
          total_versions = 0
          context_stats = []
  
          # Import the function here to avoid circular imports
          from kafka_schema_registry_unified_mcp import get_schema_versions
  
          # Get statistics for each context
          for context in contexts:
              subjects = client.get_subjects(context)
              if isinstance(subjects, dict) and "error" in subjects:
                  continue
  
              context_schemas = len(subjects)
              context_versions = 0
  
              # Count versions for each subject
              for subject in subjects:
                  versions = get_schema_versions(subject, context, registry)
                  if not isinstance(versions, dict):
                      context_versions += len(versions)
  
              total_schemas += context_schemas
              total_versions += context_versions
  
              if include_context_details:
                  context_stats.append(
                      {
                          "name": context,
                          "subject_count": context_schemas,
                          "schema_count": context_versions,
                      }
                  )
  
          # Get default context stats
          default_subjects = client.get_subjects()
          if not isinstance(default_subjects, dict):
              default_schemas = len(default_subjects)
              default_versions = 0
  
              for subject in default_subjects:
                  versions = get_schema_versions(subject, None, registry)
                  if not isinstance(versions, dict):
                      default_versions += len(versions)
  
              total_schemas += default_schemas
              total_versions += default_versions
  
              if include_context_details:
                  context_stats.append(
                      {
                          "name": "default",
                          "subject_count": default_schemas,
                          "schema_count": default_versions,
                      }
                  )
  
          # Get registry metadata
          metadata = client.get_server_metadata()
  
          result = {
              "registry": (client.config.name if hasattr(client.config, "name") else "default"),
              "total_contexts": len(contexts),
              "total_subjects": total_schemas,
              "total_schemas": total_versions,
              "contexts": context_stats if include_context_details else None,
              "generated_at": datetime.now().isoformat(),
              "registry_mode": registry_mode,
              "mcp_protocol_version": "2025-06-18",
          }
  
          # Add metadata information
          result.update(metadata)
  
          return result
      except Exception as e:
          return create_error_response(str(e), error_code="REGISTRY_STATISTICS_FAILED", registry_mode=registry_mode)
  
  
  # ===== OPTIMIZED ASYNC STATISTICS FUNCTIONS =====
  
  
  async def _count_schemas_async(
      registry_manager,
      registry_mode: str,
      context: Optional[str] = None,
      registry: Optional[str] = None,
  ) -> Dict[str, Any]:
      """
      Async version of count_schemas_tool with better performance.
      Uses parallel API calls when counting multiple contexts.
      Includes registry metadata information.
      """
      try:
          if registry_mode == "single":
              client = get_default_client(registry_manager)
          else:
              client = registry_manager.get_registry(registry)
              if client is None:
                  return {"error": f"Registry '{registry}' not found"}
  
          # Get registry metadata
          metadata = client.get_server_metadata()
  
          if context:
              # Single context - direct call
              subjects = client.get_subjects(context)
              if isinstance(subjects, dict) and "error" in subjects:
                  return subjects
  
              result = {
                  "registry": (client.config.name if hasattr(client.config, "name") else "default"),
                  "context": context,
                  "count": len(subjects),  # Use 'count' to match schema
                  "scope": "schemas",  # Add scope field as string
                  "total_schemas": len(subjects),
                  "schemas": subjects,
                  "counted_at": datetime.now(timezone.utc).isoformat(),
              }
  
              # Add metadata information, but preserve the scope field
              metadata_copy = metadata.copy()
              if "scope" in metadata_copy:
                  # Preserve the simple string scope, but add server scope info separately
                  metadata_copy["server_scope"] = metadata_copy.pop("scope")
              result.update(metadata_copy)
              return result
          else:
              # All contexts - parallel execution
              contexts = client.get_contexts()
              if isinstance(contexts, dict) and "error" in contexts:
                  return contexts