ms-sentinel-mcp-server

Instructions

Implementation Reference

• tools/hunting_tools.py:214-281 (handler)

  The async run method implements the core handler logic: extracts query_id or name parameters, lists all saved searches in the workspace, finds the matching one, extracts tags/tactics/techniques using helper, and returns structured details or error.

  async def run(self, ctx: Context, **kwargs):
      """
      Get full details of a Sentinel hunting query (saved search) by name or ID.
      Extracts all tags, tactics, and techniques using shared utility.
      """
      # Extract parameters using the centralized parameter extraction from MCPToolBase
      query_id = self._extract_param(kwargs, "query_id")
      name = self._extract_param(kwargs, "name")
      if not query_id and not name:
          return {
              "valid": False,
              "error": (
                  "Must provide either 'query_id' or 'name' to identify "
                  "the hunting query."
              ),
              "results": None,
              "errors": [
                  (
                      "Must provide either 'query_id' or 'name' to identify "
                      "the hunting query."
                  )
              ],
          }
      workspace_name, resource_group, subscription_id = self.get_azure_context(ctx)
      client = self.get_loganalytics_client(subscription_id)
      try:
          searches = client.saved_searches.list_by_workspace(
              resource_group, workspace_name
          )
          match = None
          for s in getattr(searches, "value", []):
              if (query_id and getattr(s, "id", None) == query_id) or (
                  name and getattr(s, "name", None) == name
              ):
                  match = s
                  break
          if not match:
              return {
                  "valid": False,
                  "error": "No matching hunting query found.",
                  "results": None,
                  "errors": ["No matching hunting query found."],
              }
          tags, tactics, techniques = extract_tags_tactics_techniques(match)
          details = {
              "id": getattr(match, "id", None),
              "name": getattr(match, "name", None),
              "display_name": getattr(
                  match, "display_name", getattr(match, "name", None)
              ),
              "category": getattr(match, "category", None),
              "query": getattr(match, "query", None),
              "tags": tags,
              "tactics": tactics,
              "techniques": techniques,
              "description": getattr(match, "description", None),
              "version": getattr(match, "version", None),
          }
          return {"valid": True, "error": None, "results": details, "errors": []}
      except Exception as e:
          self.logger.error("Error in %s: %s", self.__class__.__name__, str(e))
          return {
              "valid": False,
              "error": str(e),
              "results": None,
              "errors": [str(e)],
          }

• tools/hunting_tools.py:187-213 (schema)

  Class definition with tool name, description, and detailed docstring defining input parameters (query_id, name) and output format, serving as the schema.

  class SentinelHuntingQueryGetTool(MCPToolBase):
      """
      MCP-compliant tool to retrieve the full details of a Sentinel hunting query
      (saved search) by name or ID.
  
      Parameters:
          query_id (str, optional): The full resource ID or GUID of the saved search.
          name (str, optional): The display name or name of the saved search.
  
      Returns:
          dict: Details of the hunting query, or error if not found. Output keys:
              - valid (bool): True if successful, False otherwise
              - error (str or None): Error message if any
              - results (dict or None): Full hunting query details if found
              - errors (list): List of error messages
  
      Error Cases:
          - If neither query_id nor name is provided, returns an error.
          - If no matching hunting query is found, returns an error.
          - Azure API or credential errors are reported in the error field.
      """
  
      name = "sentinel_hunting_query_get"
      description = (
          "Get full details of a Sentinel hunting query (saved search) by name or ID."
      )

• tools/hunting_tools.py:283-287 (registration)

  The register_tools function registers the SentinelHuntingQueryGetTool (along with related tools) with the MCP server.

  def register_tools(mcp: FastMCP):
      SentinelHuntingQueriesListTool.register(mcp)
      SentinelHuntingQueriesCountByTacticTool.register(mcp)
      SentinelHuntingQueryGetTool.register(mcp)

• tools/hunting_tools.py:12-74 (helper)

  Shared utility function to robustly extract tags, tactics, and techniques from Sentinel hunting query objects, used in the tool's handler.

  def extract_tags_tactics_techniques(obj):
      """
      Extracts tags, tactics, and techniques from a hunting query object.
      Returns:
          tags: List of {name, value} dicts.
          tactics: List of tactics (from tags or legacy fields).
          techniques: List of techniques (from tags or legacy fields).
      Extraction precedence:
        - Tactics/techniques: Prefer tags with name 'tactics'/'techniques'
          (case-insensitive, split on comma). Fallback to legacy fields.
        - Tags: All tags as {name, value} pairs (robust to SDK object, dict, or string).
      """
      tags = []
      tactics = []
      techniques = []
      # Extract tags as list of {name, value}
      raw_tags = getattr(obj, "tags", None)
      if raw_tags:
          for tag in raw_tags:
              tag_name = None
              tag_value = None
              # Tag as dict
              if isinstance(tag, dict):
                  tag_name = tag.get("name") or tag.get("Name")
                  tag_value = tag.get("value") or tag.get("Value")
              # Tag as object with .name/.value
              elif hasattr(tag, "name") and hasattr(tag, "value"):
                  tag_name = getattr(tag, "name", None)
                  tag_value = getattr(tag, "value", None)
              # Tag as string: treat as name only
              elif isinstance(tag, str):
                  tag_name = tag
                  tag_value = None
              # Fallback: try string conversion
              else:
                  try:
                      tag_name = str(tag)
                  except Exception:
                      continue
              if tag_name is not None:
                  tags.append({"name": tag_name, "value": tag_value})
      # Tactics/techniques from tags (case-insensitive match)
      for tag in tags:
          if tag["name"] and isinstance(tag["name"], str):
              if tag["name"].lower() == "tactics" and tag["value"]:
                  tactics += [t.strip() for t in tag["value"].split(",") if t.strip()]
              elif tag["name"].lower() == "techniques" and tag["value"]:
                  techniques += [t.strip() for t in tag["value"].split(",") if t.strip()]
      # Fallback: legacy fields
      legacy_tactics = getattr(obj, "tactics", None)
      if legacy_tactics:
          tactics += [
              t.strip() for t in legacy_tactics if isinstance(t, str) and t.strip()
          ]
      legacy_techniques = getattr(obj, "techniques", None)
      if legacy_techniques:
          techniques += [
              t.strip() for t in legacy_techniques if isinstance(t, str) and t.strip()
          ]
      # Deduplicate and preserve order
      tactics = list(dict.fromkeys([t for t in tactics if t]))
      techniques = list(dict.fromkeys([t for t in techniques if t]))
      return tags, tactics, techniques