ms-sentinel-mcp-server

Instructions

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)