Agent Knowledge MCP

Instructions

Implementation Reference

• src/admin/admin_server.py:169-300 (handler)

  The core handler function implementing the 'validate_config' MCP tool. Performs comprehensive validation of configuration structure (required sections), data types (ports as int, booleans, lists, dicts), value ranges (port 1-65535), and existence checks (base directory). Reports detailed errors, warnings, and summary with current validation settings.
  @app.tool( description="Validate configuration object structure, types, and values with comprehensive error reporting", tags={"admin", "config", "validate", "check", "schema"} ) async def validate_config( config: Annotated[Optional[Dict[str, Any]], Field(description="Configuration object to validate. If not provided, validates current config")] = None ) -> str: """Validate configuration with comprehensive structure checking and detailed feedback.""" try: if config is None: config = load_config() errors = [] warnings = [] # Validate structure - required sections required_sections = ["elasticsearch", "security", "document_validation", "document_schema", "server"] for section in required_sections: if section not in config: errors.append(f"Missing required section: {section}") # Validate elasticsearch section if "elasticsearch" in config: es_config = config["elasticsearch"] if "host" not in es_config: errors.append("elasticsearch.host is required") if "port" not in es_config: errors.append("elasticsearch.port is required") elif not isinstance(es_config["port"], int): errors.append("elasticsearch.port must be an integer") elif es_config["port"] <= 0 or es_config["port"] > 65535: errors.append("elasticsearch.port must be between 1-65535") # Validate security section if "security" in config: sec_config = config["security"] if "allowed_base_directory" not in sec_config: errors.append("security.allowed_base_directory is required") else: base_dir = Path(sec_config["allowed_base_directory"]) if not base_dir.exists(): warnings.append(f"security.allowed_base_directory does not exist: {base_dir}") elif not base_dir.is_dir(): errors.append(f"security.allowed_base_directory is not a directory: {base_dir}") # Validate document_validation section if "document_validation" in config: doc_config = config["document_validation"] bool_fields = ["strict_schema_validation", "allow_extra_fields", "required_fields_only", "auto_correct_paths"] for field in bool_fields: if field in doc_config and not isinstance(doc_config[field], bool): errors.append(f"document_validation.{field} must be a boolean") # Validate document_schema section if "document_schema" in config: schema_config = config["document_schema"] required_schema_fields = ["required_fields", "field_types", "priority_values", "source_types"] for field in required_schema_fields: if field not in schema_config: errors.append(f"document_schema.{field} is required") # Validate specific schema field types if "required_fields" in schema_config: if not isinstance(schema_config["required_fields"], list): errors.append("document_schema.required_fields must be a list") elif not schema_config["required_fields"]: warnings.append("document_schema.required_fields is empty") if "field_types" in schema_config: if not isinstance(schema_config["field_types"], dict): errors.append("document_schema.field_types must be a dictionary") elif not schema_config["field_types"]: warnings.append("document_schema.field_types is empty") if "priority_values" in schema_config: if not isinstance(schema_config["priority_values"], list): errors.append("document_schema.priority_values must be a list") elif not schema_config["priority_values"]: warnings.append("document_schema.priority_values is empty") if "source_types" in schema_config: if not isinstance(schema_config["source_types"], list): errors.append("document_schema.source_types must be a list") elif not schema_config["source_types"]: warnings.append("document_schema.source_types is empty") # Validate server section if "server" in config: server_config = config["server"] if "name" in server_config and not isinstance(server_config["name"], str): errors.append("server.name must be a string") if "version" in server_config and not isinstance(server_config["version"], str): errors.append("server.version must be a string") # Prepare result message if errors: message = f"❌ Configuration validation failed!\n\n🚨 **Errors ({len(errors)}):**\n" for i, error in enumerate(errors, 1): message += f" {i}. {error}\n" else: message = "✅ Configuration validation passed!" if warnings: message += f"\n⚠️ **Warnings ({len(warnings)}):**\n" for i, warning in enumerate(warnings, 1): message += f" {i}. {warning}\n" # Show current validation settings summary if "document_validation" in config: doc_val = config["document_validation"] message += f"\n📋 **Current Document Validation Settings:**\n" message += f" 🔒 Strict schema validation: {doc_val.get('strict_schema_validation', False)}\n" message += f" 📝 Allow extra fields: {doc_val.get('allow_extra_fields', True)}\n" message += f" ✅ Required fields only: {doc_val.get('required_fields_only', False)}\n" message += f" 🔧 Auto correct paths: {doc_val.get('auto_correct_paths', True)}\n" # Show section summary sections_found = [s for s in required_sections if s in config] message += f"\n📊 **Configuration Summary:**\n" message += f" 📁 Sections found: {len(sections_found)}/{len(required_sections)}\n" message += f" ✅ Valid sections: {', '.join(sections_found)}\n" if len(sections_found) < len(required_sections): missing = [s for s in required_sections if s not in config] message += f" ❌ Missing sections: {', '.join(missing)}\n" return message except json.JSONDecodeError as e: return f"❌ JSON Error: Invalid JSON format in config data\n🔍 Details: {str(e)}\n💡 Check JSON syntax and structure" except Exception as e: return _format_admin_error(e, "validate configuration", "config structure validation")