MCP Kafka Schema Reg

Instructions

Implementation Reference

• statistics_tools.py:214-329 (handler)

  Primary synchronous handler implementing the core logic for get_registry_statistics tool. Fetches contexts, iterates over subjects and versions to compute totals, includes optional detailed context stats, and adds registry metadata.

  @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)

• schema_definitions.py:611-652 (schema)

  JSON Schema definition (REGISTRY_STATISTICS_SCHEMA) for validating the structured output of the get_registry_statistics tool, specifying properties like total_contexts, total_subjects, total_schemas, and optional per-context details.

  REGISTRY_STATISTICS_SCHEMA = {
      "type": "object",
      "properties": {
          "registry": {"type": "string", "description": "Registry name"},
          "total_contexts": {
              "type": "integer",
              "minimum": 0,
              "description": "Total number of contexts",
          },
          "total_subjects": {
              "type": "integer",
              "minimum": 0,
              "description": "Total number of subjects",
          },
          "total_schemas": {
              "type": "integer",
              "minimum": 0,
              "description": "Total number of schema versions",
          },
          "contexts": {
              "type": "array",
              "items": {
                  "type": "object",
                  "properties": {
                      "name": {"type": "string"},
                      "subject_count": {"type": "integer", "minimum": 0},
                      "schema_count": {"type": "integer", "minimum": 0},
                  },
                  "required": ["name", "subject_count", "schema_count"],
              },
              "description": "Per-context statistics",
          },
          "generated_at": {
              "type": "string",
              "format": "date-time",
              "description": "When statistics were generated",
          },
          **METADATA_FIELDS,
      },
      "required": ["total_contexts", "total_subjects", "total_schemas"],
      "additionalProperties": True,
  }

• schema_definitions.py:1017-1017 (schema)

  Mapping of tool name 'get_registry_statistics' to its output schema in the TOOL_OUTPUT_SCHEMAS registry.

  "get_registry_statistics": REGISTRY_STATISTICS_SCHEMA,

• statistics_tools.py:643-716 (helper)

  Async task queue variant of the tool for long-running operations, returns task_id immediately and uses background execution with progress updates.

  @structured_output("get_registry_statistics_task_queue", fallback_on_error=True)
  def get_registry_statistics_task_queue_tool(
      registry_manager,
      registry_mode: str,
      registry: Optional[str] = None,
      include_context_details: bool = True,
  ) -> Dict[str, Any]:
      """
      Task queue version of get_registry_statistics for better performance.
      Returns task_id immediately for async execution with progress tracking.
  
      Returns:
          Task information with task_id for monitoring progress with structured validation
      """
      try:
          # Create async task
          task = task_manager.create_task(
              TaskType.STATISTICS,
              metadata={
                  "operation": "get_registry_statistics",
                  "registry": registry,
                  "include_context_details": include_context_details,
              },
          )
  
          # Start async execution
          try:
              asyncio.create_task(
                  task_manager.execute_task(
                      task,
                      _get_registry_statistics_async,
                      registry_manager=registry_manager,
                      registry_mode=registry_mode,
                      registry=registry,
                      include_context_details=include_context_details,
                      task_id=task.id,
                  )
              )
          except RuntimeError:
              # No running event loop, use thread pool
              def run_task():
                  asyncio.run(
                      task_manager.execute_task(
                          task,
                          _get_registry_statistics_async,
                          registry_manager=registry_manager,
                          registry_mode=registry_mode,
                          registry=registry,
                          include_context_details=include_context_details,
                          task_id=task.id,
                      )
                  )
  
              thread = threading.Thread(target=run_task)
              thread.start()
  
          return {
              "message": "Registry statistics analysis started as async task",
              "task_id": task.id,
              "task": task.to_dict(),
              "operation_info": {
                  "operation": "get_registry_statistics",
                  "expected_duration": "long",
                  "async_pattern": "task_queue",
                  "guidance": "Long-running operation. Returns task_id immediately. Use get_task_status() to monitor progress.",
                  "registry_mode": registry_mode,
              },
              "registry_mode": registry_mode,
              "mcp_protocol_version": "2025-06-18",
          }
  
      except Exception as e:
          return create_error_response(str(e), error_code="TASK_CREATION_FAILED", registry_mode=registry_mode)

• statistics_tools.py:499-606 (helper)

  Async helper function used by task queue version, implements parallel execution across contexts and subjects using ThreadPoolExecutor for improved performance on large registries.

  async def _get_registry_statistics_async(
      registry_manager,
      registry_mode: str,
      registry: Optional[str] = None,
      include_context_details: bool = True,
      task_id: Optional[str] = None,
  ) -> Dict[str, Any]:
      """
      Async version of get_registry_statistics_tool with parallel execution.
      """
      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"}
  
          # Update progress
          if task_id:
              task_manager.update_progress(task_id, 10.0)
  
          # Get all contexts
          contexts = client.get_contexts()
          if isinstance(contexts, dict) and "error" in contexts:
              return contexts
  
          if task_id:
              task_manager.update_progress(task_id, 20.0)
  
          total_schemas = 0
          total_versions = 0
          context_stats = []
  
          # Parallel execution for better performance
          with ThreadPoolExecutor(max_workers=8) as executor:
              # Submit all context analysis tasks
              future_to_context = {}
  
              # Add all contexts
              for context in contexts:
                  future = executor.submit(_analyze_context_parallel, client, context, registry)
                  future_to_context[future] = context
  
              # Add default context
              future = executor.submit(_analyze_context_parallel, client, None, registry)
              future_to_context[future] = "default"
  
              # Collect results
              completed = 0
              total_contexts = len(future_to_context)
  
              for future in as_completed(future_to_context):
                  context = future_to_context[future]
                  try:
                      context_result = future.result()
  
                      if not isinstance(context_result, dict) or "error" not in context_result:
                          total_schemas += context_result.get("schemas", 0)
                          total_versions += context_result.get("versions", 0)
  
                          if include_context_details:
                              context_stats.append(
                                  {
                                      "context": context,
                                      "schemas": context_result.get("schemas", 0),
                                      "versions": context_result.get("versions", 0),
                                  }
                              )
  
                      # Update progress
                      completed += 1
                      if task_id:
                          # Progress from 20% to 90% based on context completion
                          progress = 20.0 + (completed / total_contexts) * 70.0
                          task_manager.update_progress(task_id, progress)
  
                  except Exception as e:
                      if include_context_details:
                          context_stats.append({"context": context, "error": str(e)})
  
          if task_id:
              task_manager.update_progress(task_id, 95.0)
  
          # 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_schemas": total_schemas,
              "total_versions": total_versions,
              "average_versions_per_schema": round(total_versions / max(total_schemas, 1), 2),
              "contexts": context_stats if include_context_details else None,
              "counted_at": datetime.now().isoformat(),
          }
  
          # Add metadata information
          result.update(metadata)
  
          if task_id:
              task_manager.update_progress(task_id, 100.0)
  
          return result
  
      except Exception as e:
          return {"error": str(e)}

• task_management.py:200-204 (registration)

  Operation metadata registration classifying get_registry_statistics as a long-running operation that uses task queue pattern for async handling.

      "get_registry_statistics": {
          "duration": OperationDuration.LONG,
          "pattern": AsyncPattern.TASK_QUEUE,
      },
  }