sentinel_query_validate
Validate KQL query syntax locally to ensure proper structure before execution in Microsoft Sentinel.
Instructions
Validate KQL Query Syntax locally
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| kwargs | Yes |
Implementation Reference
- tools/kql_tools.py:42-92 (handler)The main handler function (run method) that implements the tool logic: extracts 'query' parameter, calls validate_kql helper, handles errors, and returns validation results.
async def run(self, ctx: Context, **kwargs): """ Validate a KQL query and return the result. Args: ctx (Context): The context of the MCP server. **kwargs: Additional keyword arguments. Returns: dict: A dictionary containing the validation result. """ # Extract query using the centralized parameter extraction from MCPToolBase query = self._extract_param(kwargs, "query") logger = self.logger if not query: logger.error("Missing required parameter: query") return { "error": "Missing required parameter: query", "valid": False, "errors": ["Missing required parameter: query"], } try: is_valid, errors = validate_kql(query) if is_valid: return { "result": ( "Query validation passed. " "The KQL syntax appears to be correct." ), "valid": True, "errors": [], } error_message = "KQL validation failed:\n" + "\n".join(errors) # Warn via context if available if hasattr(ctx, "warning") and callable(getattr(ctx, "warning", None)): await ctx.warning(error_message) # Special handling for initialization error if any("KQL validation unavailable" in err for err in errors): return {"error": error_message, "valid": False, "errors": errors} return {"error": error_message, "valid": False, "errors": errors} except Exception as e: logger.error("Error validating KQL query: %s", e, exc_info=True) return { "error": ( "An error occurred while validating the query. " "Try validating code by executing a KQL query against the " "workspace instead: %s" % str(e) ), "valid": False, "errors": [str(e)], } - tools/kql_tools.py:95-103 (registration)Registers the sentinel_query_validate tool (KQLValidateTool) with the FastMCP server instance.
def register_tools(mcp: FastMCP): """ Register KQL tools with the MCP server. Args: mcp (FastMCP): The MCP server instance to register tools with. """ KQLValidateTool.register(mcp) - utilities/kql_validator.py:176-193 (helper)Core helper function that performs KQL syntax validation using the KQLValidator singleton, which uses Kusto.Language.dll via pythonnet for offline syntax checking.
def validate_kql(query: str) -> Tuple[bool, List[str]]: """ Validate a KQL query. Args: query: The KQL query to validate. Returns: Tuple[bool, List[str]]: (is_valid, list_of_error_messages) """ validator = get_validator() if not validator.initialized: return False, [ "KQL validation unavailable: Could not initialize validator.", "For syntax validation, please use the query tool to validate against your workspace.", ] return validator.validate_query(query)