integrations / crewai

CrewAI Integration

Use Attestd as a CrewAI BaseTool to give agents real-time CVE risk and supply chain integrity data. These patterns target crewai >= 1.15, the current stable release.

Two correctness concerns apply to every agent that checks dependencies. First: outside_coverage is not a safety signal. Catch AttestdUnsupportedProductError and return an explicit unknown-risk response. Second: check typosquat_detected on that path. Attestd can detect AI-hallucinated or misspelled package names. An agent that ignores this signal may recommend a package that does not exist or resembles a known-malicious one.

installation

Install

bash

pip install "crewai>=1.15" attestd

tool definition

BaseTool definition

Subclass BaseTool from crewai.tools with an explicit args_schema. This gives the LLM a validated input schema and gives you typed parameters. Use PrivateAttr for the Attestd client so it is not included in the schema shown to the model, and initialise it in model_post_init.

attestd_tool.py

import os
from typing import Type

import attestd
from attestd import AttestdUnsupportedProductError
from crewai.tools import BaseTool
from pydantic import BaseModel, Field, PrivateAttr


class _AttestdInput(BaseModel):
    product: str = Field(..., description='Package slug, e.g. "nginx", "runc", "@bitwarden/cli"')
    version: str = Field(..., description='Exact version string, e.g. "1.0.0"')


class AttestdTool(BaseTool):
    name: str = "check_package_vulnerability"
    description: str = (
        "Check whether a software package version has known CVE vulnerabilities "
        "or supply chain compromise. Use before deploying or recommending any "
        "software dependency. outside_coverage=True means no data — treat as "
        "unknown risk, not safe. typosquat_detected=True means the package name "
        "itself is suspect; use resembles to recover the intended package."
    )
    args_schema: Type[BaseModel] = _AttestdInput

    # PrivateAttr keeps _client out of the Pydantic schema shown to the LLM.
    _client: attestd.Client = PrivateAttr()

    def model_post_init(self, __context: object) -> None:
        self._client = attestd.Client(api_key=os.environ["ATTESTD_API_KEY"])

    def _run(self, product: str, version: str) -> dict:
        try:
            result = self._client.check(product, version)
            sc = result.supply_chain
            return {
                "outside_coverage": False,
                "risk_state": result.risk_state,
                "remote_exploitable": result.remote_exploitable,
                "actively_exploited": result.actively_exploited,
                "patch_available": result.patch_available,
                "fixed_version": result.fixed_version,
                "supply_chain_compromised": sc.compromised if sc is not None else False,
                "supply_chain_malware_type": sc.malware_type if sc is not None else None,
                "supply_chain_still_live": (
                    (sc.removed_at is None) if (sc is not None and sc.compromised) else None
                ),
            }
        except AttestdUnsupportedProductError as e:
            if e.typosquat and e.typosquat.detected:
                return {
                    "outside_coverage": True,
                    "typosquat_detected": True,
                    "typosquat_kind": e.typosquat.kind,
                    "resembles": e.typosquat.resembles,
                    "likely_intended": list(e.typosquat.likely_intended),
                    "message": (
                        f"'{product}' fails package name integrity "
                        f"(kind={e.typosquat.kind}). "
                        f"Likely intended: {e.typosquat.resembles}. "
                        f"Do not install this package."
                    ),
                }
            return {
                "outside_coverage": True,
                "typosquat_detected": False,
                "risk_state": None,
                "message": f"No Attestd coverage for '{product}'. Treat as unknown risk.",
            }

single agent

Single-agent pattern

One Agent with the Attestd tool assigned to a Task. The task description must state the blocking policy explicitly — the model cannot infer correct behaviour from the tool description alone. Pass dependencies as a kickoff input so the task is reusable.

single_agent.py

from crewai import Agent, Crew, Process, Task

from attestd_tool import AttestdTool

security_agent = Agent(
    role="Security Gate",
    goal=(
        "Check every software dependency for CVE risk and supply chain compromise "
        "before approving it for deployment."
    ),
    backstory=(
        "You are a security-aware agent that uses deterministic risk data to "
        "approve or block software dependencies. You never approve a dependency "
        "without first checking it with check_package_vulnerability."
    ),
    tools=[AttestdTool()],
    verbose=True,
)

audit_task = Task(
    description=(
        "Check these dependencies before deployment: {dependencies}. "
        "For each one, call check_package_vulnerability. "
        "Block if risk_state is 'critical' or 'high', supply_chain_compromised is True, "
        "or typosquat_detected is True. "
        "If outside_coverage is True and typosquat_detected is False, state the unknown risk explicitly. "
        "End with a clear APPROVED or BLOCKED verdict and your reasoning."
    ),
    expected_output=(
        "A structured verdict for each dependency: "
        "APPROVED <package@version> — <reason>, or "
        "BLOCKED <package@version> — <reason>. "
        "Followed by a final summary: APPROVED or BLOCKED."
    ),
    agent=security_agent,
)

crew = Crew(
    agents=[security_agent],
    tasks=[audit_task],
    process=Process.sequential,
    verbose=True,
)

result = crew.kickoff(inputs={
    "dependencies": "runc 1.0.0, litellm 1.82.7, nginx 1.27.4"
})
print(result.raw)

multi-agent pattern

Multi-agent safety gate

A dedicated security_gate agent runs Attestd checks, then passes a structured verdict to a deployment_agent via context=[security_task]. The deployment agent receives the full security verdict as context before it acts. In multi-agent workflows, a compromised dependency approved by one agent propagates to every downstream agent that trusts its output.

multi_agent.py

from crewai import Agent, Crew, Process, Task

from attestd_tool import AttestdTool

security_gate = Agent(
    role="Security Gate",
    goal="Verify every software dependency for CVE risk and supply chain compromise.",
    backstory=(
        "You are a security gate agent. You call check_package_vulnerability for every "
        "dependency before any deployment decision is made. You never skip a check."
    ),
    tools=[AttestdTool()],
    verbose=True,
)

deployment_agent = Agent(
    role="Deployment Agent",
    goal="Execute deployment tasks after all dependencies have been cleared by the security gate.",
    backstory=(
        "You handle deployment logistics. You wait for the security gate to provide "
        "a verdict before proceeding. You do not deploy any BLOCKED dependency."
    ),
    verbose=True,
)

security_task = Task(
    description=(
        "Audit these dependencies: {dependencies}. "
        "Call check_package_vulnerability for each one. "
        "Block if risk_state is 'critical' or 'high', supply_chain_compromised is True, "
        "or typosquat_detected is True. "
        "Produce a verdict: APPROVED or BLOCKED, with reasoning for each package."
    ),
    expected_output=(
        "Per-package verdicts followed by a final overall verdict: APPROVED or BLOCKED."
    ),
    agent=security_gate,
)

deployment_task = Task(
    description=(
        "Review the security gate verdict from the previous task. "
        "If the overall verdict is APPROVED, proceed with the deployment plan. "
        "If the verdict is BLOCKED, halt and report which packages were blocked and why."
    ),
    expected_output=(
        "Either a deployment confirmation with the approved package list, "
        "or a halt report listing each blocked package and the block reason."
    ),
    agent=deployment_agent,
    context=[security_task],
)

crew = Crew(
    agents=[security_gate, deployment_agent],
    tasks=[security_task, deployment_task],
    process=Process.sequential,
    verbose=True,
)

result = crew.kickoff(inputs={
    "dependencies": "runc 1.0.0, litellm 1.82.7, node-ipc 9.1.6, nginx 1.27.4"
})
print(result.raw)

fields

Return fields

The tool returns these fields. Design agent task descriptions and blocking logic around them.

Field	Semantics
`outside_coverage`	true if Attestd has no data for this product. Not a safety signal — treat as unknown risk. Check typosquat_detected on this path.
`risk_state`	"critical" \| "high" \| "elevated" \| "low" \| "none" \| null (when outside_coverage). Block on critical or high.
`remote_exploitable`	true if the vulnerability has a network attack vector. Drives the high vs elevated split in risk_state.
`actively_exploited`	true if in the CISA KEV catalog. Block regardless of risk_state if true.
`patch_available`	true if a fixed version is known. Use with fixed_version to tell the agent what to recommend.
`fixed_version`	The earliest clean version, or null if no patch exists yet.
`supply_chain_compromised`	true if a malicious publish was detected on npm or PyPI. Block immediately.
`supply_chain_malware_type`	Malware classification if compromised, else null. e.g. "malware", "keylogger", "cryptominer".
`supply_chain_still_live`	true if the compromised version has not been removed from the registry. null when not compromised. A still-live compromise is more urgent.
`typosquat_detected`	Only present on outside_coverage responses. true means the package name fails integrity — do not install.
`typosquat_kind`	"hallucination" (AI-invented or conflated name) or "typosquat" (misspelling of a real package).
`resembles / likely_intended`	The canonical package(s) the requested name most closely matches. Use to recover the intended dependency.