ZeroPath MCP Server

Instructions

Implementation Reference

• src/zeropath_mcp_server/server.py:84-105 (handler)

  The core handler function for the 'search_vulnerabilities' tool. It is registered via the @mcp.tool() decorator with FastMCP and implements the logic to search ZeroPath API for vulnerabilities using the provided search_query.

  @mcp.tool()
  def search_vulnerabilities(search_query=None):
      """
      Search for vulnerabilities using the Zeropath API with a simple search query.
      """
      payload = {}
      if search_query:
          payload["searchQuery"] = search_query
  
      response, error = make_api_request("issues/search", payload)
  
      if error:
          return error
  
      if response.status_code == 200:
          return process_vulnerability_response(response.json())
      elif response.status_code == 401:
          return "Error: Unauthorized - check API credentials"
      elif response.status_code == 400:
          return f"Error: Bad request - {response.text}"
      else:
          return f"Error: API returned status {response.status_code}: {response.text}"

• src/zeropath_mcp_server/server.py:21-47 (helper)

  Helper utility to make authenticated POST requests to the ZeroPath API, used by search_vulnerabilities to query the /issues/search endpoint.

  def make_api_request(endpoint, payload=None, include_org=True):
      """Make authenticated API request to ZeroPath."""
      if not token_id or not token_secret:
          return None, "Error: Zeropath API credentials not found in environment variables"
  
      headers = {
          "X-ZeroPath-API-Token-Id": token_id,
          "X-ZeroPath-API-Token-Secret": token_secret,
          "Content-Type": "application/json"
      }
  
      if payload is None:
          payload = {}
  
      if include_org and org_id:
          payload["organizationId"] = org_id
  
      try:
          response = requests.post(
              f"{API_BASE_URL}/{endpoint}",
              headers=headers,
              json=payload
          )
          return response, None
      except Exception as e:
          return None, f"Error: {str(e)}"

• src/zeropath_mcp_server/server.py:888-968 (helper)

  Helper function called by search_vulnerabilities to process and format the raw JSON response from ZeroPath API into a human-readable string summary of vulnerabilities.

  def process_vulnerability_response(raw_response):
      """
      Process the raw API response into a more readable format for LLMs.
      Extracts and organizes the most relevant information in plain text format.
      """
      if "error" in raw_response:
          return f"Error: {raw_response['error']}"
  
      if "issues" not in raw_response:
          return "No vulnerability issues found in the response."
  
      # Count totals and categorize issues
      total_issues = len(raw_response["issues"])
      patchable_count = sum(1 for issue in raw_response["issues"] if not issue.get("unpatchable", False))
      unpatchable_count = sum(1 for issue in raw_response["issues"] if issue.get("unpatchable", True))
  
      # Build a formatted text response
      result = f"Found {total_issues} vulnerability issues. {patchable_count} are patchable, {unpatchable_count} are unpatchable.\n\n"
  
      # Process each issue
      for i, issue in enumerate(raw_response["issues"], 1):
          result += f"Issue {i}:\n"
          result += f"ID: {issue.get('id')}\n"
          result += f"Status: {issue.get('status', 'unknown')}\n"
  
          # Include all fields that exist
          if issue.get("type"):
              result += f"Type: {issue.get('type')}\n"
  
          if issue.get("patchable") is not None:
              patchable = not issue.get("unpatchable", False)
              result += f"Patchable: {patchable}\n"
  
          if issue.get("language"):
              result += f"Language: {issue['language']}\n"
  
          if issue.get("score") is not None:
              result += f"Score: {issue['score']}\n"
  
          if issue.get("severity") is not None:
              result += f"Severity: {issue['severity']}\n"
  
          if issue.get("generatedTitle"):
              result += f"Title: {issue['generatedTitle']}\n"
  
          if issue.get("generatedDescription"):
              result += f"Description: {issue['generatedDescription']}\n"
  
          if issue.get("affectedFile"):
              result += f"Affected File: {issue['affectedFile']}\n"
  
          if issue.get("cwes"):
              result += f"CWEs: {', '.join(issue['cwes'])}\n"
  
          if issue.get("validated"):
              result += f"Validation Status: {issue['validated']}\n"
  
          if issue.get("triagePhase"):
              result += f"Triage Phase: {issue['triagePhase']}\n"
  
          # Add patch information if available
          if issue.get("vulnerabilityPatch") and not issue.get("unpatchable", False):
              patch = issue["vulnerabilityPatch"]
              result += "\n--- PATCH INFORMATION ---\n"
              result += f"PATCH ID: {patch.get('id', 'N/A')}\n"
              result += "------------------------\n"
              result += "Has Patch: Yes\n"
  
              if patch.get("pullRequestStatus"):
                  result += f"Patch Status: {patch['pullRequestStatus']}\n"
  
          # Add extra space between issues
          result += "\n"
  
      # Include pagination info if available
      if "currentPage" in raw_response or "pageSize" in raw_response:
          result += "Pagination Info:\n"
          result += f"Current Page: {raw_response.get('currentPage', 1)}\n"
          result += f"Page Size: {raw_response.get('pageSize', total_issues)}\n"
  
      return result