MCP Vulnerability Checker Server

Instructions

Implementation Reference

• mcp_simple_tool/tools/vulnerability_search.py:16-292 (handler)

  Core handler function that executes the tool: validates inputs, queries NVD API v2.0, processes CVEs, handles errors, and returns formatted markdown results.
  async def search_vulnerabilities( keywords: Optional[str] = None, severity: Optional[str] = None, date_range: Optional[str] = None, ) -> List[types.TextContent | types.ImageContent | types.EmbeddedResource]: """ Search vulnerability databases with advanced filtering. Args: keywords: Keywords to search for in CVE descriptions (optional) severity: Filter by severity level: CRITICAL, HIGH, MEDIUM, LOW, NONE (optional) date_range: Date range filter in format "YYYY-MM-DD,YYYY-MM-DD" or predefined ranges like "30d", "90d" (optional) Note: NVD API has a 120-day maximum limit for date ranges Returns: List of content containing search results or error messages """ # Validate and parse inputs parsed_params = {} # Parse date range if date_range: try: if date_range.lower() in ["30d", "90d", "1y", "2y"]: # Predefined ranges days_map = {"30d": 30, "90d": 90, "1y": 365, "2y": 730} days = days_map[date_range.lower()] # NVD API has a 120-day maximum limit for date ranges if days > 120: return [ types.TextContent( type="text", text=f"Error: Date range '{date_range}' ({days} days) exceeds the NVD API limit of 120 days. Please use '30d', '90d', or a custom range within 120 days.", ) ] end_date = datetime.now() start_date = end_date - timedelta(days=days) parsed_params["pubStartDate"] = start_date.strftime( "%Y-%m-%dT00:00:00.000" ) parsed_params["pubEndDate"] = end_date.strftime("%Y-%m-%dT23:59:59.999") elif "," in date_range: # Custom range: YYYY-MM-DD,YYYY-MM-DD start_str, end_str = date_range.split(",", 1) start_date = datetime.strptime(start_str.strip(), "%Y-%m-%d") end_date = datetime.strptime(end_str.strip(), "%Y-%m-%d") # Check if custom range exceeds 120-day limit days_diff = (end_date - start_date).days if days_diff > 120: return [ types.TextContent( type="text", text=f"Error: Custom date range ({days_diff} days) exceeds the NVD API limit of 120 days. Please use a shorter range.", ) ] parsed_params["pubStartDate"] = start_date.strftime( "%Y-%m-%dT00:00:00.000" ) parsed_params["pubEndDate"] = end_date.strftime("%Y-%m-%dT23:59:59.999") else: return [ types.TextContent( type="text", text="Error: Invalid date_range format. Use 'YYYY-MM-DD,YYYY-MM-DD' or predefined ranges: 30d, 90d, 1y, 2y", ) ] except ValueError as e: return [ types.TextContent( type="text", text=f"Error: Invalid date format. Use YYYY-MM-DD. Details: {str(e)}", ) ] # Parse severity filter if severity: severity = severity.upper() valid_severities = ["CRITICAL", "HIGH", "MEDIUM", "LOW", "NONE"] if severity not in valid_severities: return [ types.TextContent( type="text", text=f"Error: Invalid severity '{severity}'. Valid options: {', '.join(valid_severities)}", ) ] # Map severity to CVSS score ranges for NVD API if severity == "CRITICAL": parsed_params["cvssV3Severity"] = "CRITICAL" elif severity == "HIGH": parsed_params["cvssV3Severity"] = "HIGH" elif severity == "MEDIUM": parsed_params["cvssV3Severity"] = "MEDIUM" elif severity == "LOW": parsed_params["cvssV3Severity"] = "LOW" # Add keywords if provided if keywords: parsed_params["keywordSearch"] = keywords.strip() # Set result limit parsed_params["resultsPerPage"] = "20" # Reasonable limit for display parsed_params["startIndex"] = "0" headers = { "User-Agent": "MCP Vulnerability Search Tool v1.0", "Accept": "application/json", } try: timeout = httpx.Timeout(30.0, connect=15.0) # Longer timeout for search async with httpx.AsyncClient( follow_redirects=True, headers=headers, timeout=timeout ) as client: # NVD API 2.0 endpoint for CVE search url = "https://services.nvd.nist.gov/rest/json/cves/2.0" response = await client.get(url, params=parsed_params) response.raise_for_status() data = response.json() total_results = data.get("totalResults", 0) vulnerabilities = data.get("vulnerabilities", []) results_per_page = data.get("resultsPerPage", 20) if total_results == 0: search_desc = [] if keywords: search_desc.append(f"keywords: '{keywords}'") if severity: search_desc.append(f"severity: {severity}") if date_range: search_desc.append(f"date range: {date_range}") search_text = " with " + ", ".join(search_desc) if search_desc else "" return [ types.TextContent( type="text", text=f"No vulnerabilities found{search_text}.", ) ] # Format the response result = "🔍 **Vulnerability Search Results**\n\n" result += "📊 **Search Summary:**\n" result += f" • Total Results: {total_results:,}\n" result += ( f" • Showing: {min(results_per_page, len(vulnerabilities))} results\n" ) if keywords: result += f" • Keywords: '{keywords}'\n" if severity: result += f" • Severity Filter: {severity}\n" if date_range: result += f" • Date Range: {date_range}\n" result += "\n📋 **Vulnerabilities Found:**\n\n" for i, vuln_data in enumerate(vulnerabilities[:20], 1): # Limit display cve = vuln_data.get("cve", {}) cve_id = cve.get("id", "Unknown") published = cve.get("published", "Unknown") cve.get("lastModified", "Unknown") # Get description descriptions = cve.get("descriptions", []) description = "No description available" for desc in descriptions: if desc.get("lang") == "en": description = desc.get("value", "")[ :200 ] # Truncate long descriptions if len(desc.get("value", "")) > 200: description += "..." break # Get CVSS scores metrics = cve.get("metrics", {}) cvss_info = "Not available" severity_emoji = "⚪" # Try CVSS 3.1 first, then 3.0, then 2.0 for version in [ "cvssMetricV31", "cvssMetricV30", "cvssMetricV2", ]: if version in metrics and metrics[version]: metric = metrics[version][0] # Take first metric cvss_data = metric.get("cvssData", {}) score = cvss_data.get("baseScore", "N/A") sev = cvss_data.get("baseSeverity", "N/A") if score != "N/A": cvss_info = f"{score} ({sev})" # Set emoji based on severity if sev == "CRITICAL": severity_emoji = "🔴" elif sev == "HIGH": severity_emoji = "🟠" elif sev == "MEDIUM": severity_emoji = "🟡" elif sev == "LOW": severity_emoji = "🟢" break # Format published date try: pub_date = datetime.fromisoformat(published.replace("Z", "+00:00")) pub_formatted = pub_date.strftime("%Y-%m-%d") except (ValueError, TypeError, AttributeError): pub_formatted = ( published[:10] if len(published) >= 10 else published ) result += f"**{i}. {severity_emoji} {cve_id}**\n" result += f" 📅 Published: {pub_formatted}\n" result += f" ⚠️ CVSS: {cvss_info}\n" result += f" 📝 Description: {description}\n" result += f" 🔗 Details: https://cve.mitre.org/cgi-bin/cvename.cgi?name={cve_id}\n\n" if total_results > results_per_page: result += f"📄 **Note:** Showing first {results_per_page} of {total_results:,} total results.\n" result += "💡 **Tip:** Use more specific keywords or severity filters to narrow results.\n\n" result += "📊 **Data Source:** National Vulnerability Database (NVD)\n" result += f"🕐 **Search performed:** {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}" return [types.TextContent(type="text", text=result)] except httpx.TimeoutException: return [ types.TextContent( type="text", text="Error: Request timed out while searching vulnerability database. Try using more specific search criteria.", ) ] except httpx.HTTPStatusError as e: if e.response.status_code == 403: return [ types.TextContent( type="text", text="Error: Access forbidden. The NVD API may be rate limiting requests. Please try again later.", ) ] elif e.response.status_code == 503: return [ types.TextContent( type="text", text="Error: NVD service temporarily unavailable. Please try again later.", ) ] return [ types.TextContent( type="text", text=f"Error: HTTP {e.response.status_code} error while searching vulnerabilities.", ) ] except json.JSONDecodeError: return [ types.TextContent( type="text", text="Error: Invalid JSON response from NVD API." ) ] except Exception as e: return [ types.TextContent( type="text", text=f"Error: Failed to search vulnerabilities: {str(e)}", ) ]
• mcp_simple_tool/server.py:226-242 (schema)

  JSON schema defining optional input parameters for the tool: keywords (str), severity (str), date_range (str). No required fields.
  inputSchema={ "type": "object", "properties": { "keywords": { "type": "string", "description": vuln_search_description, }, "severity": { "type": "string", "description": "Filter by severity level: CRITICAL, HIGH, MEDIUM, LOW, NONE", }, "date_range": { "type": "string", "description": "Date range filter. Use predefined ranges (30d, 90d, 1y, 2y) or custom format YYYY-MM-DD,YYYY-MM-DD", }, }, },
• mcp_simple_tool/server.py:223-243 (registration)

  Registers the tool in the MCP server's list_tools() method, providing name, description, and schema.
  types.Tool( name="search_vulnerabilities", description="Search vulnerability databases with advanced filtering", inputSchema={ "type": "object", "properties": { "keywords": { "type": "string", "description": vuln_search_description, }, "severity": { "type": "string", "description": "Filter by severity level: CRITICAL, HIGH, MEDIUM, LOW, NONE", }, "date_range": { "type": "string", "description": "Date range filter. Use predefined ranges (30d, 90d, 1y, 2y) or custom format YYYY-MM-DD,YYYY-MM-DD", }, }, }, ),
• mcp_simple_tool/server.py:126-131 (registration)

  Dispatches tool calls to the search_vulnerabilities handler function in the @app.call_tool() method.
  elif name == "search_vulnerabilities": # All parameters are optional for search keywords = arguments.get("keywords") severity = arguments.get("severity") date_range = arguments.get("date_range") return await search_vulnerabilities(keywords, severity, date_range)