ms-sentinel-mcp-server

Instructions

Implementation Reference

• tools/incident_tools.py:136-273 (handler)

  The async run method of the SentinelIncidentDetailsTool class, which executes the tool logic. It extracts the incident_number parameter, queries the SecurityIncident table for details using KQL, parses the response, and optionally fetches related SecurityAlert details.
  async def run(self, ctx: Context, **kwargs): """ Get detailed information about a specific Sentinel incident. Args: ctx (Context): MCP context object. **kwargs: Must include 'incident_number'. Returns: dict: Contains 'incident' (dict), 'related_alerts' (list), 'valid' (bool), 'errors' (list). """ logger = self.logger # Using centralized parameter extraction from MCPToolBase incident_number = self._extract_param(kwargs, "incident_number") if incident_number is None: return { "incident": None, "related_alerts": [], "valid": False, "errors": ["incident_number is required"], } 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: details_query = f""" SecurityIncident | where IncidentNumber == '{incident_number}' | extend AlertsCount = iif(isnull(AlertIds), 0, array_length(AlertIds)) | extend BookmarksCount = iif(isnull(BookmarkIds), 0, array_length(BookmarkIds)) | extend CommentsCount = iif(isnull(Comments), 0, array_length(Comments)) """ details_response = await run_in_thread( logs_client.query_workspace, workspace_id=workspace_id, query=details_query, timespan=timedelta(days=90), name=f"get_incident_details_{incident_number}", ) if ( not details_response or not details_response.tables or not details_response.tables[0].rows ): return { "incident": None, "related_alerts": [], "valid": True, "errors": [ f"No incident found for incident number: {incident_number}" ], } row = details_response.tables[0].rows[0] columns = [ col.name if hasattr(col, "name") else col for col in details_response.tables[0].columns ] incident_details = {col: row[idx] for idx, col in enumerate(columns)} alert_ids = incident_details.get("AlertIds") if isinstance(alert_ids, str): try: alert_ids = json.loads(alert_ids) except Exception: alert_ids = [] result = { "incident": incident_details, "related_alerts": [], "valid": True, "errors": [], } if alert_ids and isinstance(alert_ids, list) and len(alert_ids) > 0: alert_id_list = ",".join([f"'{aid}'" for aid in alert_ids if aid]) alerts_query = f""" SecurityAlert | where TimeGenerated > ago(90d) | where SystemAlertId in ({alert_id_list}) | project TimeGenerated, AlertName, AlertSeverity, Description, Status, Entities | sort by TimeGenerated desc | take 5 """ alerts_response = await run_in_thread( logs_client.query_workspace, workspace_id=workspace_id, query=alerts_query, timespan=timedelta(days=90), name=f"get_incident_alerts_{incident_number}", ) if ( alerts_response and alerts_response.tables and alerts_response.tables[0].rows ): for alert_row in alerts_response.tables[0].rows: alert_time = alert_row[0] alert_name = alert_row[1] alert_severity = alert_row[2] alert_description = ( alert_row[3] if alert_row[3] else "No description" ) alert_status = alert_row[4] alert_entities = alert_row[5] result["related_alerts"].append( { "Time": alert_time, "Name": alert_name, "Severity": alert_severity, "Status": alert_status, "Description": alert_description, "Entities": alert_entities, } ) else: # No alert IDs associated with this incident result["related_alerts"] = [] return result except Exception as e: logger.error("Error retrieving incident details: %s", e) return { "incident": None, "related_alerts": [], "valid": False, "errors": ["Error retrieving incident details"], }
• tools/incident_tools.py:276-284 (registration)

  The register_tools function that registers the SentinelIncidentDetailsTool (named 'sentinel_incident_get') and the list tool with the MCP server instance. This function is called dynamically 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)
• tools/incident_tools.py:126-135 (schema)

  The class definition including the tool name 'sentinel_incident_get', description, and docstring outlining inputs ('incident_number') and outputs.
  class SentinelIncidentDetailsTool(MCPToolBase): """ Tool for retrieving detailed information about a specific Sentinel incident. Returns incident details and related alerts if available. """ name = "sentinel_incident_get" description = "Get detailed information about a specific Sentinel incident"