ms-sentinel-mcp-server

Instructions

Implementation Reference

• tools/table_tools.py:291-482 (handler)

  The `async def run` method in `GetTableDetailsTool` class executes the tool logic. It retrieves table details including retention policies, metadata from Azure Management REST API, and row count/lastUpdated from KQL queries on the Log Analytics workspace.

  async def run(self, ctx: Context, **kwargs):
      """
      Get details (metadata, retention, row count, etc.) for a Log Analytics table.
  
      Args:
          ctx (Context): The MCP tool context.
          **kwargs: Must include 'table_name'.
  
      Returns:
          dict: Results as described in the class docstring.
      """
      table_name = self._extract_param(kwargs, "table_name")
      if not table_name:
          return {"error": "Missing required parameter: table_name"}
      logs_client, workspace_id = self.get_logs_client_and_workspace(ctx)
      cache_key = f"table_details_json:{workspace_id}:{table_name}"
      cached = cache.get(cache_key)
      if cached:
          return cached
      # Get Azure context
      resource_group = None
      workspace_name = None
      subscription_id = None
      if (
          hasattr(ctx, "request_context")
          and getattr(ctx, "request_context", None) is not None
      ):
          services_ctx = ctx.request_context.lifespan_context
          logs_client = getattr(services_ctx, "logs_client", None)
          workspace_id = getattr(services_ctx, "workspace_id", None)
          # We no longer need rest_client as we're using direct API calls
          resource_group = getattr(services_ctx, "resource_group", None)
          workspace_name = getattr(services_ctx, "workspace_name", None)
          subscription_id = getattr(services_ctx, "subscription_id", None)
      errors = []
      result = {"table": table_name}
      # --- REST API METADATA ---
      try:
          if resource_group and workspace_name and subscription_id:
              # We'll use the call_api method directly, no need to get the client separately
  
              # Construct the URL with API version 2017-04-26-preview as recommended
              # pylint: disable=line-too-long
              # ruff: noqa: E501
              url = (
                  f"https://management.azure.com/subscriptions/{subscription_id}/"
                  f"resourceGroups/{resource_group}/providers/Microsoft.OperationalInsights/"
                  f"workspaces/{workspace_name}/tables/{table_name}?api-version=2017-04-26-preview"
              )
  
              # Make the direct REST API call using the task manager
              try:
                  # Use the call_api method from the base class
                  table_data = await self.call_api(
                      ctx, "GET", url, name=f"get_table_details_{table_name}"
                  )
  
                  # Process the response
                  if table_data and "properties" in table_data:
                      props = table_data["properties"]
  
                      # Extract metadata properties (camelCase format from API)
                      result["retentionInDays"] = props.get("retentionInDays")
                      result["totalRetentionInDays"] = props.get(
                          "totalRetentionInDays"
                      )
                      if (
                          props.get("totalRetentionInDays") is not None
                          and props.get("retentionInDays") is not None
                      ):
                          result["archiveRetentionInDays"] = (
                              props["totalRetentionInDays"] - props["retentionInDays"]
                          )
                      else:
                          result["archiveRetentionInDays"] = None
  
                      # Extract other metadata fields
                      result["plan"] = props.get("plan")
                      result["provisioningState"] = props.get("provisioningState")
  
                      # Extract schema-related properties if available
                      if "schema" in props:
                          schema = props["schema"]
                          result["tableType"] = schema.get("tableType")
                          result["description"] = schema.get("description")
                      else:
                          result["tableType"] = props.get("tableType")
                          result["description"] = props.get("description")
  
                      # Extract other properties
                      result["isInherited"] = props.get("isInherited")
                      result["isTotalRetentionInherited"] = props.get(
                          "isTotalRetentionInherited"
                      )
                      self.logger.info(
                          "Successfully retrieved metadata for table: %s", table_name
                      )
                  else:
                      errors.append(
                          "REST API: No properties found in table metadata response."
                      )
                      self.logger.error(
                          "No properties found in table metadata response for: %s",
                          table_name,
                      )
  
              except StopIteration:
                  # Handle case where no data is returned
                  errors.append("REST API: No data returned for table metadata.")
                  self.logger.error(
                      "No data returned for table metadata: %s", table_name
                  )
              except Exception as e:
                  errors.append("REST API call error: %s" % str(e))
                  self.logger.error(
                      "Error during REST API call for table %s: %s", table_name, e
                  )
          else:
              errors.append(
                  "REST API: Missing required parameters for table metadata retrieval."
              )
              self.logger.error(
                  "Missing required parameters: resource_group=%s, workspace_name=%s, subscription_id=%s",
                  resource_group,
                  workspace_name,
                  subscription_id,
              )
      except Exception as e:
          errors.append("REST API client error: %s" % str(e))
      # --- KQL METADATA ---
      if logs_client:
          # Query for lastUpdated
          try:
              kql_last_updated = (
                  f"{table_name}\n| summarize lastUpdated=max(TimeGenerated)"
              )
              last_updated_resp = await run_in_thread(
                  logs_client.query_workspace,
                  workspace_id=workspace_id,
                  query=kql_last_updated,
                  timespan=timedelta(days=30),
                  name="get_table_last_updated",
              )
              if (
                  last_updated_resp
                  and last_updated_resp.tables
                  and len(last_updated_resp.tables[0].rows) > 0
              ):
                  row = last_updated_resp.tables[0].rows[0]
                  result["lastUpdated"] = row[0]
              else:
                  result["lastUpdated"] = None
          except TimeoutError:
              errors.append(
                  "KQL timeout: lastUpdated query exceeded time limit (30 days)"
              )
              result["lastUpdated"] = None
          except Exception as e:
              errors.append(f"KQL error (lastUpdated): {str(e)}")
              result["lastUpdated"] = None
          # Query for rowCount
          try:
              kql_row_count = f"{table_name}\n| count"
              row_count_resp = await run_in_thread(
                  logs_client.query_workspace,
                  workspace_id=workspace_id,
                  query=kql_row_count,
                  timespan=timedelta(days=30),
                  name="get_table_row_count",
              )
              if (
                  row_count_resp
                  and row_count_resp.tables
                  and len(row_count_resp.tables[0].rows) > 0
              ):
                  row = row_count_resp.tables[0].rows[0]
                  result["rowCount"] = row[0]
              else:
                  result["rowCount"] = 0
          except TimeoutError:
              errors.append(
                  "KQL timeout: rowCount query exceeded time limit (30 days)"
              )
              result["rowCount"] = 0
          except Exception as e:
              errors.append(f"KQL error (rowCount): {str(e)}")
              result["rowCount"] = 0
      else:
          errors.append("logs_client missing.")
      if errors:
          result["errors"] = errors
      return result

• tools/table_tools.py:485-495 (registration)

  The `register_tools` function in table_tools.py calls `GetTableDetailsTool.register(mcp)` to register the tool with the MCP instance.

  def register_tools(mcp):
      """
      Register all table tools with the given MCP instance.
  
      Args:
          mcp: The MCP instance to register tools with.
      """
      ListTablesTool.register(mcp)
      GetTableSchemaTool.register(mcp)
      GetTableDetailsTool.register(mcp)

• server.py:236-238 (registration)

  The server.py loads Python modules from the tools/ directory via `load_components` and invokes their `register_tools(mcp)` functions, thereby registering all tools including `sentinel_logs_table_details_get`.

  if os.path.exists(tools_dir):
      tools = load_components(mcp, tools_dir, "register_tools")
      logger.info("Auto-registered %d tool modules", len(tools))