search_vulnerabilities

Search vulnerability databases with advanced filtering by keywords, severity levels, and date ranges to identify security threats and analyze vulnerability landscapes.

Instructions

Search vulnerability databases with advanced filtering

Input Schema

TableJSON Schema

Name	Required	Description
`keywords`	No	Search vulnerability databases with advanced filtering capabilities to find relevant security threats. Filter by keywords (e.g., 'apache', 'sql injection'), severity levels (CRITICAL, HIGH, MEDIUM, LOW), and date ranges (30d, 90d, 1y, 2y). Enables comprehensive threat research and vulnerability landscape analysis across multiple CVE and vulnerability databases.
`severity`	No	Filter by severity level: CRITICAL, HIGH, MEDIUM, LOW, NONE
`date_range`	No	Date range filter. Use predefined ranges (30d, 90d, 1y, 2y) or custom format YYYY-MM-DD,YYYY-MM-DD

Implementation Reference

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

The main handler function that executes the tool logic: searches NVD vulnerability database with optional filters for keywords, severity, and date range, returns formatted results or errors.

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:223-243 (registration)

Registers the search_vulnerabilities tool in the MCP server with name, description, and input schema definition.

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 calls to the search_vulnerabilities handler function based on tool name in the server's fetch_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)

mcp_simple_tool/server.py:54-60 (schema)

Defines the detailed description string used in the tool's schema.

vuln_search_description: str = (
    "Search vulnerability databases with advanced filtering capabilities to find "
    "relevant security threats. Filter by keywords (e.g., 'apache', 'sql injection'), "
    "severity levels (CRITICAL, HIGH, MEDIUM, LOW), and date ranges (30d, 90d, 1y, 2y). "
    "Enables comprehensive threat research and vulnerability landscape analysis across "
    "multiple CVE and vulnerability databases."
)

MCP Vulnerability Checker Server