	In Browser
	StumbleUpon
	del.icio.us
	Google
	Google Buzz
	reddit
	LinkedIn

	Facebook
	Twitter
	Linkedin
	E-Mail

Generative AI > Google Gemini API > Vulnerability Prioritisation Agent

Vulnerability Prioritisation Agent

Author: Venkata Sudhakar

Security scanners generate hundreds of vulnerability findings every week. Without intelligent prioritisation, the ShopMax India security team wastes time patching low-risk findings while critical vulnerabilities in payment systems remain open. An agent that scores and ranks vulnerabilities by business impact enables the team to fix what matters most first.

This tutorial builds a Gemini ADK agent that takes a list of vulnerability findings, computes a risk priority score using CVSS severity, asset criticality, and exploitability status, and produces a prioritised remediation plan.

The below example shows a vulnerability prioritisation agent for ShopMax India in a business context.

import asyncio
from google.adk.agents import LlmAgent
from google.adk.tools import FunctionTool
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.genai import types

# Vulnerability findings from scanner
VULNERABILITIES = [
    {"id": "CVE-2024-001", "system": "Payment Gateway", "cvss": 9.8,
     "asset_criticality": "critical", "exploited_in_wild": True,
     "description": "Remote code execution in payment processing library"},
    {"id": "CVE-2024-002", "system": "Employee Portal", "cvss": 4.3,
     "asset_criticality": "medium", "exploited_in_wild": False,
     "description": "Reflected XSS in login form"},
    {"id": "CVE-2024-003", "system": "WMS Server", "cvss": 7.5,
     "asset_criticality": "high", "exploited_in_wild": False,
     "description": "SQL injection in warehouse query endpoint"},
    {"id": "CVE-2024-004", "system": "Marketing CMS", "cvss": 5.5,
     "asset_criticality": "low", "exploited_in_wild": False,
     "description": "Path traversal in file upload feature"},
]

CRITICALITY_WEIGHT = {"critical": 40, "high": 25, "medium": 15, "low": 5}

def prioritise_vulnerabilities() -> dict:
    results = []
    for v in VULNERABILITIES:
        cvss_score = int(v["cvss"] * 4)  # scale to 40
        crit_score = CRITICALITY_WEIGHT.get(v["asset_criticality"], 5)
        exploit_score = 20 if v["exploited_in_wild"] else 0
        total = cvss_score + crit_score + exploit_score
        if total >= 90:
            priority = "P1 - Patch within 24 hours"
        elif total >= 65:
            priority = "P2 - Patch within 7 days"
        elif total >= 40:
            priority = "P3 - Patch within 30 days"
        else:
            priority = "P4 - Schedule for next maintenance window"
        results.append({
            "cve_id": v["id"],
            "system": v["system"],
            "cvss": v["cvss"],
            "description": v["description"],
            "exploited_in_wild": v["exploited_in_wild"],
            "risk_score": total,
            "priority": priority,
        })
    results.sort(key=lambda r: r["risk_score"], reverse=True)
    p1_count = sum(1 for r in results if r["priority"].startswith("P1"))
    return {
        "total_findings": len(results),
        "p1_critical": p1_count,
        "prioritised": results,
    }

vuln_tool = FunctionTool(func=prioritise_vulnerabilities)

agent = LlmAgent(
    name="vuln_prioritisation_agent",
    model="gemini-2.0-flash",
    description="Prioritises security vulnerabilities for ShopMax India by business risk score.",
    instruction=(
        "You are a security agent for ShopMax India. "
        "When asked to prioritise vulnerabilities, call prioritise_vulnerabilities. "
        "Present findings ranked by risk score with clear patch timelines. "
        "Highlight any P1 findings that need immediate action today."
    ),
    tools=[vuln_tool],
)

async def main():
    session_service = InMemorySessionService()
    runner = Runner(agent=agent, app_name="vuln_app", session_service=session_service)
    session = await session_service.create_session(app_name="vuln_app", user_id="sec1")
    msg = types.Content(role="user", parts=[types.Part(
        text="Prioritise this week vulnerability findings and give me the remediation plan"
    )])
    async for event in runner.run_async(user_id="sec1", session_id=session.id, new_message=msg):
        if event.is_final_response() and event.content:
            print(event.content.parts[0].text)

asyncio.run(main())

It gives the following output,

Vulnerability Prioritisation Report - ShopMax India Security

Total Findings: 4  |  P1 Critical: 1

[P1 - Patch within 24 hours] CVE-2024-001 | Risk Score: 119
  System : Payment Gateway (CRITICAL asset)
  CVSS   : 9.8 | Exploited in the wild: YES
  Issue  : Remote code execution in payment processing library
  Action : Emergency patch deployment. Isolate service if patch unavailable.

[P2 - Patch within 7 days] CVE-2024-003 | Risk Score: 80
  System : WMS Server (HIGH asset)
  CVSS   : 7.5 | Exploited in the wild: No
  Issue  : SQL injection in warehouse query endpoint

[P3 - Patch within 30 days] CVE-2024-002 | Risk Score: 47
  System : Employee Portal (MEDIUM asset)
  CVSS   : 4.3 | Issue: Reflected XSS in login form

[P4 - Next maintenance] CVE-2024-004 | Risk Score: 27
  System : Marketing CMS (LOW asset)
  CVSS   : 5.5 | Issue: Path traversal in file upload

ShopMax India's security team can feed scanner output into this agent every Monday morning to generate the week's remediation plan. The P1 findings trigger an immediate Slack alert to the on-call security engineer, while P2 and P3 items are added to the sprint backlog with assigned owners and target dates.

Send your comments, suggestions or queries regarding this site to [email protected].