Agent Knowledge MCP

Instructions

Implementation Reference

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

  The primary handler function for the 'validate_config' MCP tool. Implements comprehensive validation of configuration structure, required sections, data types, and values with detailed error reporting and warnings. Uses FastMCP @app.tool decorator for registration. Handles both provided config objects and current loaded config.

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