Dependency auditor agent built with the Claude Agent SDK and Firecrawl

audit.py

import asyncio
import json
import os
import sys
from pathlib import Path

from dotenv import load_dotenv

load_dotenv()

from claude_agent_sdk import (
    AgentDefinition,
    ClaudeAgentOptions,
    ResultMessage,
    query,
)
from pydantic import BaseModel
from rich.console import Console
from rich.table import Table
from rich.text import Text


class PackageReport(BaseModel):
    name: str
    pinned_version: str = ""
    latest_version: str = ""
    last_release: str = ""
    last_commit: str = ""
    status: str = ""        # active | slowing down | stale | abandoned
    risk: str = ""          # low | medium | high
    summary: str = ""
    replacement: str = ""


class AuditResult(BaseModel):
    packages: list[PackageReport]


RESEARCHER_PROMPT = """You are a Python package researcher. For the package you're
given, use the Firecrawl tools to check its PyPI page for latest version and
release dates, and its GitHub repo for recent commits and activity. Return a
detailed text summary of your findings."""

AUDITOR_PROMPT = """You are a Python dependency auditor coordinating a team of
researchers. You will receive a list of Python packages to audit.

For EACH package, dispatch a researcher subagent using the Task tool. Dispatch
ALL packages at once so they run in parallel.

After all researchers report back, classify each package:

- active: released within the last 6 months
- slowing down: last release 6-18 months ago
- stale: last release 18 months to 2 years ago
- abandoned: no release in 2+ years, or repo archived

Risk levels:
- low: active, no known issues
- medium: slowing down or minor concerns
- high: stale/abandoned, security risks, or no maintained alternative

If stale or abandoned, suggest a replacement. Keep summaries to one sentence.
Return the structured audit result with all packages."""


def parse_requirements(path: Path) -> list[tuple[str, str]]:
    packages = []
    for line in path.read_text().splitlines():
        line = line.strip()
        if not line or line.startswith("#") or line.startswith("-"):
            continue
        if "==" in line:
            name, version = line.split("==", 1)
            packages.append((name.strip(), version.strip()))
        else:
            packages.append((line.strip(), ""))
    return packages


async def run(path: Path) -> list[PackageReport]:
    packages = parse_requirements(path)
    console = Console(stderr=True, force_terminal=True)

    devnull = open(os.devnull, "w")  # noqa: SIM115
    options = ClaudeAgentOptions(
        model="claude-sonnet-4-6",
        system_prompt=AUDITOR_PROMPT,
        permission_mode="bypassPermissions",
        max_turns=25,
        debug_stderr=devnull,
        output_format={
            "type": "json_schema",
            "schema": AuditResult.model_json_schema(),
        },
        mcp_servers={
            "firecrawl": {
                "command": "npx",
                "args": ["-y", "firecrawl-mcp"],
                "env": {"FIRECRAWL_API_KEY": os.environ["FIRECRAWL_API_KEY"]},
            }
        },
        agents={
            "researcher": AgentDefinition(
                description="Researches a Python package's health using Firecrawl web tools",
                prompt=RESEARCHER_PROMPT,
            ),
        },
    )

    pkg_list = "\n".join(
        f"- {name}=={version}" if version else f"- {name}"
        for name, version in packages
    )
    prompt = f"Audit these Python packages:\n\n{pkg_list}"
    result = None

    with console.status("[bold blue]Auditing packages..."):
        async for message in query(prompt=prompt, options=options):
            if isinstance(message, ResultMessage) and message.structured_output:
                data = message.structured_output
                if isinstance(data, str):
                    data = json.loads(data)
                result = AuditResult.model_validate(data)

    if result:
        return result.packages
    return [
        PackageReport(name=name, pinned_version=version,
                      status="unknown", risk="medium", summary="Audit failed.")
        for name, version in packages
    ]


RISK_COLORS = {"low": "green", "medium": "yellow", "high": "red"}
RISK_ORDER = {"high": 0, "medium": 1, "low": 2, "unknown": 3}


def print_report(reports: list[PackageReport]) -> None:
    console = Console(stderr=True, force_terminal=True)
    reports.sort(key=lambda r: RISK_ORDER.get(r.risk, 3))

    healthy = sum(1 for r in reports if r.risk == "low")
    warning = sum(1 for r in reports if r.risk == "medium")
    critical = sum(1 for r in reports if r.risk == "high")
    console.print()
    console.print(
        f"  [green]{healthy} healthy[/]  "
        f"[yellow]{warning} warning[/]  "
        f"[red]{critical} critical[/]",
    )
    console.print()

    table = Table(show_header=True, header_style="bold")
    table.add_column("Package", min_width=14)
    table.add_column("Pinned")
    table.add_column("Latest")
    table.add_column("Status")
    table.add_column("Risk", justify="center")
    table.add_column("Summary", max_width=44)
    table.add_column("Replacement")

    for r in reports:
        color = RISK_COLORS.get(r.risk, "white")
        table.add_row(
            r.name,
            r.pinned_version,
            r.latest_version,
            r.status,
            Text(r.risk, style=f"bold {color}"),
            r.summary,
            r.replacement or "",
        )

    console.print(table)
    console.print()


def save_report(reports: list[PackageReport], out_path: Path) -> None:
    lines = ["# Dependency Audit Report\n"]
    for r in reports:
        lines.append(f"## {r.name}")
        lines.append(f"| Field | Value |")
        lines.append(f"|---|---|")
        lines.append(f"| Pinned | {r.pinned_version} |")
        lines.append(f"| Latest | {r.latest_version} |")
        lines.append(f"| Last release | {r.last_release} |")
        lines.append(f"| Last commit | {r.last_commit} |")
        lines.append(f"| Status | {r.status} |")
        lines.append(f"| Risk | {r.risk} |")
        lines.append(f"| Summary | {r.summary} |")
        if r.replacement:
            lines.append(f"| Replacement | {r.replacement} |")
        lines.append("")
    out_path.write_text("\n".join(lines))


def main() -> None:
    if len(sys.argv) < 2:
        print("Usage: python audit.py <requirements.txt>")
        sys.exit(1)

    path = Path(sys.argv[1])
    if not path.exists():
        print(f"File not found: {path}")
        sys.exit(1)

    if not os.environ.get("FIRECRAWL_API_KEY"):
        print("Set FIRECRAWL_API_KEY environment variable.")
        sys.exit(1)

    reports = asyncio.run(run(path))
    print_report(reports)

    out_path = path.with_suffix(".md")
    save_report(reports, out_path)
    print(f"Report saved to {out_path}")


if __name__ == "__main__":
    main()