ms-sentinel-mcp-server

Instructions

Implementation Reference

• tools/incident_tools.py:33-124 (handler)

  The `run` method is the core handler that executes the tool logic. It extracts parameters like limit, severity, and status, queries the SecurityIncident table using KQL over the last 30 days, processes the results into a structured list of incidents, and handles errors gracefully.
  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 registers the SentinelIncidentListTool (sentinel_incident_list) with the MCP server instance.
  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)
• tools/incident_tools.py:21-32 (schema)

  The class defines the tool name 'sentinel_incident_list' and description, along with docstrings describing inputs (filters: limit, severity, status) and outputs (list of incidents).
  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"