mcp-nvd

Instructions

Implementation Reference

• mcp_nvd/server.py:148-160 (handler)

  The primary handler function for the 'get_cve' tool. Decorated with @mcp.tool() for MCP registration. Fetches specific CVE data from the NVD API and returns a formatted string response.

  @mcp.tool()
  async def get_cve(cve_id: str, concise: bool = False) -> str:
      """Get a CVE based on the ID and return a formatted string with detailed attributes."""
      url = f"{BASE_URL}?cveId={cve_id}"
      data = await make_nvd_request(url)
  
      if not data or "vulnerabilities" not in data or not data["vulnerabilities"]:
          return f"No data found for CVE ID: {cve_id}"
  
      cve = data["vulnerabilities"][0]["cve"]
      logger.info(f"Processing CVE: {cve_id}")
      return format_cve(cve, concise)

• mcp_nvd/server.py:67-146 (helper)

  Key helper function used by get_cve to format the raw CVE data into a human-readable string, supporting both concise and detailed output.

  def format_cve(cve: Dict[str, Any], concise: bool = False) -> str:
      """Helper function to format a single CVE entry, shared by get_cve and search_cve."""
      try:
          cve_id = cve["id"]
          source_identifier = cve["sourceIdentifier"]
          published = cve["published"]
          last_modified = cve["lastModified"]
          vuln_status = cve["vulnStatus"]
          description = next(
              (desc["value"] for desc in cve["descriptions"] if desc["lang"] == "en"),
              "No English description available",
          )
  
          # Extract CVSS v3.1 metrics
          cvss_v31_metric = next(
              (metric for metric in cve.get("metrics", {}).get("cvssMetricV31", []) if metric["type"] == "Primary"),
              None,
          )
          cvss_v31_data = cvss_v31_metric["cvssData"] if cvss_v31_metric else None
          cvss_v31_score = cvss_v31_data.get("baseScore", "N/A") if cvss_v31_data else "N/A"
          cvss_v31_severity = cvss_v31_data.get("baseSeverity", "N/A") if cvss_v31_data else "N/A"
          cvss_v31_vector = cvss_v31_data.get("vectorString", "N/A") if cvss_v31_data else "N/A"
          cvss_v31_exploitability = cvss_v31_metric.get("exploitabilityScore", "N/A") if cvss_v31_metric else "N/A"
          cvss_v31_impact = cvss_v31_metric.get("impactScore", "N/A") if cvss_v31_metric else "N/A"
  
          # Extract CVSS v2.0 metrics
          cvss_v2 = next(
              (metric["cvssData"] for metric in cve.get("metrics", {}).get("cvssMetricV2", []) if metric["type"] == "Primary"),
              None,
          )
          cvss_v2_score = cvss_v2.get("baseScore", "N/A") if cvss_v2 else "N/A"
          cvss_v2_severity = cvss_v2.get("baseSeverity", "N/A") if cvss_v2 else "N/A"
          cvss_v2_vector = cvss_v2.get("vectorString", "N/A") if cvss_v2 else "N/A"
  
          # Extract weaknesses (CWE IDs)
          weaknesses = [
              desc["value"] for weak in cve.get("weaknesses", []) for desc in weak["description"] if desc["lang"] == "en"
          ]
          weaknesses_str = ", ".join(weaknesses) if weaknesses else "None listed"
  
          # Extract references with tags
          references = [f"{ref['url']} ({', '.join(ref.get('tags', []))})" for ref in cve.get("references", [])]
          references_str = "\n  - " + "\n  - ".join(references) if references else "None listed"
  
          # Extract configurations (CPEs)
          cpe_matches = []
          for node in cve.get("configurations", [{}])[0].get("nodes", []):
              for match in node.get("cpeMatch", []):
                  if match.get("vulnerable", False):
                      cpe_matches.append(match["criteria"])
          configurations_str = "\n  - " + "\n  - ".join(cpe_matches) if cpe_matches else "None listed"
  
          # Format output
          if concise:
              return (
                  f"CVE ID: {cve_id}\n"
                  f"Description: {description}\n"
                  f"CVSS v3.1 Score: {cvss_v31_score} ({cvss_v31_severity})"
              )
          else:
              return (
                  f"CVE ID: {cve_id}\n"
                  f"Source Identifier: {source_identifier}\n"
                  f"Published: {published}\n"
                  f"Last Modified: {last_modified}\n"
                  f"Vulnerability Status: {vuln_status}\n"
                  f"Description: {description}\n"
                  f"CVSS v3.1 Score: {cvss_v31_score} ({cvss_v31_severity})\n"
                  f"CVSS v3.1 Vector: {cvss_v31_vector}\n"
                  f"CVSS v3.1 Exploitability Score: {cvss_v31_exploitability}\n"
                  f"CVSS v3.1 Impact Score: {cvss_v31_impact}\n"
                  f"CVSS v2.0 Score: {cvss_v2_score} ({cvss_v2_severity})\n"
                  f"CVSS v2.0 Vector: {cvss_v2_vector}\n"
                  f"Weaknesses (CWE): {weaknesses_str}\n"
                  f"References:\n{references_str}\n"
                  f"Affected Configurations (CPE):\n{configurations_str}"
              )
      except Exception as e:
          logger.error(f"Error formatting CVE {cve.get('id', 'unknown')}: {str(e)}")
          return f"Error processing CVE: {str(e)}"

• mcp_nvd/server.py:50-65 (helper)

  Shared helper function for making asynchronous HTTP requests to the NVD API, used by get_cve to retrieve CVE data.

  async def make_nvd_request(url: str) -> Dict[str, Any] | None:
      """Make a request to the NVD API with proper error handling."""
      async with httpx.AsyncClient() as client:
          try:
              response = await client.get(url, headers=HEADERS, timeout=30.0)
              response.raise_for_status()
              return response.json()
          except httpx.HTTPStatusError as e:
              logger.error(f"HTTP error: {e.response.status_code} - {e.response.text}")
              return None
          except httpx.RequestError as e:
              logger.error(f"Request error: {e}")
              return None
          except Exception as e:
              logger.error(f"Unexpected error: {e}")
              return None

• mcp_nvd/__init__.py:1-2 (registration)

  Export of the get_cve function for use in the package.

  from .server import (
      get_cve,