CrewAI AIエージェントでセキュリティニュースの要約を作成

uv tool install crewai

uv tool update-shell

crewai create crew <your_project_name>

crewai install

   @agent
    def researcher(self) -> Agent:
        return Agent(
            config=self.agents_config['researcher'], # type: ignore[index]
            verbose=True
        )

    @agent
    def reporting_analyst(self) -> Agent:
        return Agent(
            config=self.agents_config['reporting_analyst'], # type: ignore[index]
            verbose=True
        )

   @task
    def research_task(self) -> Task:
        return Task(
            config=self.tasks_config['research_task'], # type: ignore[index]
        )

    @task
    def reporting_task(self) -> Task:
        return Task(
            config=self.tasks_config['reporting_task'], # type: ignore[index]
            output_file='report.md'
        )

def run():
    """
    Run the crew.
    """
    inputs = {
        'topic': 'AI LLMs',
        'current_year': str(datetime.now().year)
    }

    try:
        SecurityNewsSummary().crew().kickoff(inputs=inputs)
    except Exception as e:
        raise Exception(f"An error occurred while running the crew: {e}")

cve_curation_task:
    description: >
      Parse the HTML page to find the top most interesting CVEs reported in {security_vulnerabilities_url}. 

      You need to select 3 CVEs from the page and extract the link to their vulnerability page.
      For example, the link to the vulnerability for the package @trpc/server is: https://security.snyk.io/vuln/SNYK-JS-TRPCSERVER-10060256

      You should de-prioritize choosing CVEs for malicious packages. Instead, you should prioritize CVEs for popular packages and those that are likely
      to have a high impact on the open-source ecosystem.

    expected_output: >
      A total of 3 CVEs, each formatted as a bullet point in the following markdown format 
      ```
      - https://security.snyk.io/vuln/SNYK-JS-TRPCSERVER-10060256
      ```
    agent: cve_curator

cve_research_task:
    description: >  
      For each of the CVE you should extract information by visiting the link to the vulnerability and extracting the information from the page.
      For example, the link to the vulnerability for the package @trpc/server is: https://security.snyk.io/vuln/SNYK-JS-TRPCSERVER-10060256

      Once you have the link, visit the page and extract the following information:
      1) CVE ID (example: CVE-2025-12345)
      2) Type of Vulnerability (example: RCE, SQL Injection, etc.)
      3) CVE Publication Date (example: 2025-01-01)
      4) Link to the CVE vulnerability page (example: https://security.snyk.io/vuln/SNYK-JS-KIBANA-10339388). The link exists as the href attribute of the anchor tag in the HTML page, set on the name of the vulnerability.
      5) Package name that was found vulnerable (example: kibana)

    expected_output: >
      A total of 3 CVEs, each formatted as a bullet point in the following markdown format (strict markdown formatting for example links should not have space between the [] and () chars):
      ```
      **@trpc/server** found vulnerable to CVE-2025-43855 [Uncaught Exception](https://security.snyk.io/vuln/SNYK-JS-TRPCSERVER-10060256), 24 Apr 2025
      ```
    agent: cve_researcher

from .custom_tool import MyCustomTool
from .table_scraper import TableScraperTool

__all__ = ['MyCustomTool', 'TableScraperTool']

from typing import List, Dict, Any
from bs4 import BeautifulSoup
import requests
from crewai.tools import BaseTool

class TableScraperTool(BaseTool):
    name: str = "table_scraper"
    description: str = "Scrapes table data from a webpage, extracting rows, links, and other relevant information"

    def _run(self, url: str) -> List[Dict[str, Any]]:
        """
        Scrapes table data from the given URL.

        Args:
            url: The URL to scrape

        Returns:
            List of dictionaries containing table row data
        """
        try:
            response = requests.get(url)
            response.raise_for_status()
            soup = BeautifulSoup(response.text, 'html.parser')

            # Find all table rows
            rows = soup.find_all('tr', class_='table__row')
            table_data = []

            for row in rows:
                row_data = {}

                # Extract severity
                severity_elem = row.find('li', class_='severity__item')
                if severity_elem:
                    row_data['severity'] = severity_elem.get('class', [''])[0].replace('severity__item--', '')

                # Extract vulnerability title and link
                title_link = row.find('a', class_='anchor--underline')
                if title_link:
                    row_data['title'] = title_link.text.strip()
                    row_data['vuln_link'] = title_link.get('href', '')

                # Extract package name and link
                package_link = row.find('a', attrs={'data-snyk-test-package-manager': True})
                if package_link:
                    row_data['package_name'] = package_link.text.strip()
                    row_data['package_link'] = package_link.get('href', '')

                # Extract version ranges
                version_spans = row.find_all('span', class_='vulns-table__semver')
                if version_spans:
                    row_data['version_ranges'] = [span.text.strip() for span in version_spans]

                # Extract package manager and date
                package_manager = row.find('span', attrs={'type': True})
                if package_manager:
                    row_data['package_manager'] = package_manager.get('type', '')
                    row_data['published_date'] = package_manager.get('published', '')

                if row_data:  # Only add if we found some data
                    table_data.append(row_data)

            return table_data

        except Exception as e:
            return [{"error": f"Failed to scrape table data: {str(e)}"}]

    async def _arun(self, url: str) -> List[Dict[str, Any]]:
        """Async implementation of the tool"""
        return self._run(url)

dependencies = [
    ...
    "beautifulsoup4>=4.12.0"

uv pip install -e .

from crewai import Agent, Crew, Process, Task
from crewai.project import CrewBase, agent, crew, task
from crewai.agents.agent_builder.base_agent import BaseAgent
from crewai_tools import ScrapeWebsiteTool
from typing import List
from .tools import TableScraperTool

@agent
    def cve_curator(self) -> Agent:
        return Agent(
            config=self.agents_config['cve_curator'], # type: ignore[index]
            verbose=True,
            tools=[ScrapeWebsiteTool(), TableScraperTool()]
        )

    @agent
    def cve_researcher(self) -> Agent:
        return Agent(
            config=self.agents_config['cve_researcher'], # type: ignore[index]
            verbose=True,
            tools=[ScrapeWebsiteTool()]
        )

    # To learn more about structured task outputs,
    # task dependencies, and task callbacks, check out the documentation:
    # https://docs.crewai.com/concepts/tasks#overview-of-a-task
    @task
    def cve_curation_task(self) -> Task:
        return Task(
            config=self.tasks_config['cve_curation_task'], # type: ignore[index]
            output_file='cves.md'
        )

    @task
    def cve_research_task(self) -> Task:
        return Task(
            config=self.tasks_config['cve_research_task'], # type: ignore[index]
            output_file='cves.md'
        )

crewai run