get_registry_info
Retrieve detailed information about a specific Kafka Schema Registry. Use this tool to access registry data efficiently, maintained for backward compatibility with the MCP server.
Instructions
Get detailed information about a specific registry.
NOTE: This tool is maintained for backward compatibility. Consider using the 'registry://info/{name}' resource instead for better performance.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| registry | No |
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)