ms-sentinel-mcp-server

Instructions

Implementation Reference

• tools/threat_intel_tools.py:311-375 (handler)

  The async run method of SentinelThreatIntelligenceIndicatorGetTool that executes the core tool logic: extracts indicator_name parameter, validates Azure context, calls the Sentinel Threat Intelligence REST API to retrieve the specific indicator, processes the response properties into a structured dictionary, and returns it or an error.
  async def run(self, ctx: Context, **kwargs): """ Get a specific Sentinel Threat Intelligence indicator. Args: ctx (Context): The MCP tool context. **kwargs: Indicator name as 'indicator_name' parameter. Returns: dict: Results as described in the class docstring. """ indicator_name = self._extract_param(kwargs, "indicator_name") if not indicator_name: return {"error": "indicator_name parameter is required", "valid": False} workspace_name, resource_group, subscription_id = self.get_azure_context(ctx) valid = self.validate_azure_context( True, workspace_name, resource_group, subscription_id, self.logger ) if not valid: return {"error": "Missing required Azure context", "valid": False} try: url = ( f"https://management.azure.com/subscriptions/{subscription_id}/" f"resourceGroups/{resource_group}/providers/Microsoft.OperationalInsights/" f"workspaces/{workspace_name}/providers/Microsoft.SecurityInsights/" f"threatIntelligence/main/indicators/{indicator_name}?" f"api-version=2024-01-01-preview" ) indicator = await self.call_api(ctx, "GET", url, name="get_ti_indicator") if not indicator: return { "error": "Threat intelligence indicator '%s' not found" % indicator_name, "valid": False, } props = indicator.get("properties", {}) details = { "id": indicator.get("id"), "name": indicator.get("name"), "type": indicator.get("type"), "displayName": props.get("displayName"), "patternType": props.get("patternType"), "pattern": props.get("pattern"), "source": props.get("source"), "created": props.get("createdTimeUtc"), "confidence": props.get("confidence"), "threatTypes": props.get("threatTypes"), "validFrom": props.get("validFrom"), "validUntil": props.get("validUntil"), "description": props.get("description"), "killChainPhases": props.get("killChainPhases"), "labels": props.get("labels"), } return {"indicator": details, "valid": True} except Exception as e: self.logger.error( "Error retrieving threat intelligence indicator %s: %s", indicator_name, e, ) return { "error": "Error retrieving threat intelligence indicator %s: %s" % (indicator_name, e), "valid": False, }
• tools/threat_intel_tools.py:429-440 (registration)

  The register_tools function that registers the SentinelThreatIntelligenceIndicatorGetTool (sentinel_ti_indicator_get) with the MCP server by calling its register method, along with other TI tools.
  def register_tools(mcp: FastMCP): """ Register all Sentinel Threat Intelligence tools with the given MCP instance. Args: mcp (FastMCP): The MCP instance to register tools with. """ SentinelThreatIntelligenceIndicatorGetTool.register(mcp) SentinelThreatIntelligenceIndicatorMetricsCollectTool.register(mcp) SentinelIPGeodataGetTool.register(mcp) SentinelDomainWhoisGetTool.register(mcp)
• tools/threat_intel_tools.py:296-375 (handler)

  The full class definition of the tool handler, including name, description, input/output docstring, and the run method that implements the API call to retrieve the indicator.
  class SentinelThreatIntelligenceIndicatorGetTool(MCPToolBase): """ Tool to get a specific Sentinel Threat Intelligence indicator. Returns: dict: { 'indicator': dict, # Indicator details as returned by the API 'valid': bool, # True if successful 'error': str (optional) } """ name = "sentinel_ti_indicator_get" description = "Get a specific Sentinel threat intelligence indicator" async def run(self, ctx: Context, **kwargs): """ Get a specific Sentinel Threat Intelligence indicator. Args: ctx (Context): The MCP tool context. **kwargs: Indicator name as 'indicator_name' parameter. Returns: dict: Results as described in the class docstring. """ indicator_name = self._extract_param(kwargs, "indicator_name") if not indicator_name: return {"error": "indicator_name parameter is required", "valid": False} workspace_name, resource_group, subscription_id = self.get_azure_context(ctx) valid = self.validate_azure_context( True, workspace_name, resource_group, subscription_id, self.logger ) if not valid: return {"error": "Missing required Azure context", "valid": False} try: url = ( f"https://management.azure.com/subscriptions/{subscription_id}/" f"resourceGroups/{resource_group}/providers/Microsoft.OperationalInsights/" f"workspaces/{workspace_name}/providers/Microsoft.SecurityInsights/" f"threatIntelligence/main/indicators/{indicator_name}?" f"api-version=2024-01-01-preview" ) indicator = await self.call_api(ctx, "GET", url, name="get_ti_indicator") if not indicator: return { "error": "Threat intelligence indicator '%s' not found" % indicator_name, "valid": False, } props = indicator.get("properties", {}) details = { "id": indicator.get("id"), "name": indicator.get("name"), "type": indicator.get("type"), "displayName": props.get("displayName"), "patternType": props.get("patternType"), "pattern": props.get("pattern"), "source": props.get("source"), "created": props.get("createdTimeUtc"), "confidence": props.get("confidence"), "threatTypes": props.get("threatTypes"), "validFrom": props.get("validFrom"), "validUntil": props.get("validUntil"), "description": props.get("description"), "killChainPhases": props.get("killChainPhases"), "labels": props.get("labels"), } return {"indicator": details, "valid": True} except Exception as e: self.logger.error( "Error retrieving threat intelligence indicator %s: %s", indicator_name, e, ) return { "error": "Error retrieving threat intelligence indicator %s: %s" % (indicator_name, e), "valid": False, }