ms-sentinel-mcp-server

Instructions

Implementation Reference

• tools/incident_tools.py:21-125 (handler)

  The SentinelIncidentListTool class is the main handler for the 'sentinel_incident_list' tool. It defines the tool name, description, and the async run method that implements the core logic: extracts optional parameters (limit default 10, severity, status), initializes Azure logs client, constructs and executes a KQL query on SecurityIncident table ordered by TimeGenerated desc over last 30 days, with optional filters and take limit, projects key fields, parses rows into dicts, and returns the list or empty with message on no results.
  class SentinelIncidentListTool(MCPToolBase): """ Tool for listing security incidents in Microsoft Sentinel. Returns a list of recent incidents with summary fields. Supports filtering by severity and status. """ name = "sentinel_incident_list" description = "List security incidents in Microsoft Sentinel" async def run(self, ctx: Context, **kwargs): """ List recent security incidents in Microsoft Sentinel. Args: ctx (Context): MCP context object. **kwargs: Optional filters (limit, severity, status). Returns: dict: Contains 'incidents' (list), 'valid' (bool), 'errors' (list), and optional 'message'. """ logger = self.logger # Using centralized parameter extraction from MCPToolBase limit = self._extract_param(kwargs, "limit", 10) severity = self._extract_param(kwargs, "severity", None) status = self._extract_param(kwargs, "status", None) try: logs_client, workspace_id = self.get_logs_client_and_workspace(ctx) except Exception as e: logger.error("Error initializing Azure logs client: %s", e) return {"error": "Azure Logs client initialization failed"} if logs_client is None or workspace_id is None: return {"error": "Azure Logs client or workspace_id is not initialized"} try: query = """ SecurityIncident | order by TimeGenerated desc """ if severity: query += f"\n| where Severity =~ '{severity}'" if status: query += f"\n| where Status =~ '{status}'" query += f"\n| take {limit}" query += """ | project IncidentNumber, Title, Severity, Status, CreatedTime=TimeGenerated, LastModifiedTime, IncidentUrl """ response = await run_in_thread( logs_client.query_workspace, workspace_id=workspace_id, query=query, timespan=timedelta(days=30), name="get_recent_incidents", ) if response and response.tables and len(response.tables[0].rows) > 0: incidents = [] for row in response.tables[0].rows: incident = { "IncidentNumber": row[0], "Title": row[1], "Severity": row[2], "Status": row[3], "CreatedTime": row[4], "LastModifiedTime": row[5], "IncidentUrl": row[6] if len(row) > 6 else None, } incidents.append(incident) return {"incidents": incidents, "valid": True, "errors": []} filters = [] if severity: filters.append(f"severity={severity}") if status: filters.append(f"status={status}") filter_text = "" if filters: filter_text = f" with filters ({', '.join(filters)})" logger.info("No incidents found%s in the last 30 days.", filter_text) return { "incidents": [], "valid": True, "errors": [], "message": f"No incidents found{filter_text} in the last 30 days.", } except Exception as e: logger.error("Error retrieving incidents: %s", e) return { "incidents": [], "valid": False, "errors": ["Error retrieving incidents"], }
• tools/incident_tools.py:276-285 (registration)

  The register_tools function in incident_tools.py registers the SentinelIncidentListTool (sentinel_incident_list) with the MCP server instance. This module's register_tools is invoked by server.py's load_components.
  def register_tools(mcp: FastMCP): """ Register incident tools with the MCP server. Args: mcp (FastMCP): The MCP server instance. """ SentinelIncidentListTool.register(mcp) SentinelIncidentDetailsTool.register(mcp)