count_contexts
Determine the total number of contexts within a Kafka Schema Registry using this tool, enabling efficient registry management and schema organization.
Instructions
Count the number of contexts in a registry.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| registry | No |
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,