ms-sentinel-mcp-server

Instructions

Implementation Reference

• tools/watchlist_tools.py:99-201 (handler)

  The async run method of SentinelWatchlistGetTool that executes the tool: extracts watchlist_alias, gets SecurityInsights client, calls client.watchlists.get(resource_group, workspace_name, watchlist_alias), extracts properties safely, and returns watchlist details.
  async def run(self, ctx: Context, **kwargs): logger = self.logger # Extract parameters using the centralized parameter extraction from MCPToolBase watchlist_alias = self._extract_param(kwargs, "watchlist_alias") if not watchlist_alias: return {"error": "watchlist_alias parameter is required"} # Get Azure context workspace_name, resource_group, subscription_id = self.get_azure_context(ctx) # Get security insights client client = None try: client = self.get_securityinsight_client(subscription_id) except Exception as e: logger.error("Error initializing Azure SecurityInsights client: %s", e) return { "error": ( "Azure SecurityInsights client initialization failed: %s" % str(e) ) } if client is None: return {"error": "Azure SecurityInsights client is not initialized"} try: # Get the specific watchlist watchlist = await run_in_thread( client.watchlists.get, resource_group_name=resource_group, workspace_name=workspace_name, watchlist_alias=watchlist_alias, ) # Log the watchlist object to understand its structure logger.debug("Watchlist object: %s", watchlist) # Create a basic info dictionary with guaranteed attributes watchlist_details = { "id": watchlist.id if hasattr(watchlist, "id") else None, "name": watchlist.name if hasattr(watchlist, "name") else None, } # Try to access properties directly from the watchlist object first try: # Check for direct properties on the watchlist object if hasattr(watchlist, "watchlist_alias"): watchlist_details["alias"] = watchlist.watchlist_alias if hasattr(watchlist, "display_name"): watchlist_details["displayName"] = watchlist.display_name if hasattr(watchlist, "description"): watchlist_details["description"] = watchlist.description if hasattr(watchlist, "provider"): watchlist_details["provider"] = watchlist.provider if hasattr(watchlist, "source"): watchlist_details["source"] = watchlist.source if hasattr(watchlist, "items_search_key"): watchlist_details["itemsSearchKey"] = watchlist.items_search_key if hasattr(watchlist, "created_time_utc"): watchlist_details["created"] = watchlist.created_time_utc if hasattr(watchlist, "updated_time_utc"): watchlist_details["updated"] = watchlist.updated_time_utc if hasattr(watchlist, "items_count"): watchlist_details["itemsCount"] = watchlist.items_count # If we couldn't find any direct properties, try the nested properties approach if len(watchlist_details) <= 2 and hasattr(watchlist, "properties"): props = watchlist.properties if hasattr(props, "watchlist_alias"): watchlist_details["alias"] = props.watchlist_alias if hasattr(props, "display_name"): watchlist_details["displayName"] = props.display_name if hasattr(props, "description"): watchlist_details["description"] = props.description if hasattr(props, "provider"): watchlist_details["provider"] = props.provider if hasattr(props, "source"): watchlist_details["source"] = props.source if hasattr(props, "items_search_key"): watchlist_details["itemsSearchKey"] = props.items_search_key if hasattr(props, "created_time_utc"): watchlist_details["created"] = props.created_time_utc if hasattr(props, "updated_time_utc"): watchlist_details["updated"] = props.updated_time_utc if hasattr(props, "items_count"): watchlist_details["itemsCount"] = props.items_count except Exception as prop_error: # Log the property access error but continue with basic details logger.error("Error accessing watchlist properties: %s", prop_error) return {"watchlist": watchlist_details, "valid": True} except Exception as e: logger.error( "Error retrieving watchlist details for alias %s: %s", watchlist_alias, e, ) return { "error": "Error retrieving watchlist details for alias %s: %s" % (watchlist_alias, e) }
• tools/watchlist_tools.py:382-393 (registration)

  Registers SentinelWatchlistGetTool (among other watchlist tools) with the FastMCP server by calling its register method.
  def register_tools(mcp: FastMCP): """ Register all Sentinel watchlist tools with the MCP server instance. Args: mcp (FastMCP): The MCP server instance to register tools with. """ SentinelWatchlistsListTool.register(mcp) SentinelWatchlistGetTool.register(mcp) SentinelWatchlistItemsListTool.register(mcp) SentinelWatchlistItemGetTool.register(mcp)
• tools/watchlist_tools.py:91-98 (schema)

  Tool class definition including name, description, and docstring indicating input parameter 'watchlist_alias'.
  class SentinelWatchlistGetTool(MCPToolBase): """ Tool for retrieving a specific Microsoft Sentinel watchlist by alias. """ name = "sentinel_watchlist_get" description = "Get a specific Sentinel watchlist"