MCP Kafka Schema Reg

Instructions

Implementation Reference

• statistics_tools.py:26-79 (handler)

  Core implementation of the count_contexts tool handler. Uses registry client to fetch contexts, counts them, and returns structured JSON with metadata and error handling.

  @structured_output("count_contexts", fallback_on_error=True)
  def count_contexts_tool(registry_manager, registry_mode: str, registry: Optional[str] = None) -> Dict[str, Any]:
      """
      Count the number of contexts in a registry.
  
      Args:
          registry: Optional registry name (ignored in single-registry mode)
  
      Returns:
          Dictionary containing context count and details with registry 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,
                  )
  
          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,
              )
  
          # Get registry metadata
          metadata = client.get_server_metadata()
  
          result = {
              "registry": (client.config.name if hasattr(client.config, "name") else "default"),
              "count": len(contexts),
              "scope": "contexts",
              "contexts": contexts,
              "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="CONTEXT_COUNT_FAILED", registry_mode=registry_mode)

• schema_definitions.py:591-608 (schema)

  JSON Schema definition (COUNT_SCHEMA) used for validating the structured output of the count_contexts tool.

  COUNT_SCHEMA = {
      "type": "object",
      "properties": {
          "count": {"type": "integer", "minimum": 0, "description": "Count result"},
          "scope": {
              "type": "string",
              "description": "What was counted (contexts, schemas, versions)",
          },
          "context": {
              "type": "string",
              "description": "Context name if scoped to context",
          },
          "registry": {"type": "string", "description": "Registry name"},
          **METADATA_FIELDS,
      },
      "required": ["count", "scope"],
      "additionalProperties": True,
  }

• schema_definitions.py:1014-1014 (schema)

  Mapping of the 'count_contexts' tool name to its output schema (COUNT_SCHEMA) in the TOOL_OUTPUT_SCHEMAS dictionary.

  "count_contexts": COUNT_SCHEMA,

• task_management.py:188-190 (registration)

  Registration metadata for the count_contexts operation, specifying it as a QUICK direct operation (no task queue). Used for MCP client guidance.

  "count_contexts": {
      "duration": OperationDuration.QUICK,
      "pattern": AsyncPattern.DIRECT,