integrations / langchain

LangChain Integration

Use Attestd as a LangChain tool to give agents real-time CVE risk and supply chain integrity data. The tool returns structured, deterministic output — no CVSS score interpretation, no ambiguous text for the agent to parse.

The most important correctness concern: outside_coverage is not a safety signal. An agent that treats AttestdUnsupportedProductError as “no known vulnerabilities” is wrong. Catch it and return an explicit unknown-risk response.

installation

Install

bash

pip install attestd "langchain-core>=0.3" langchain langchain-openai

tool definition

StructuredTool definition

Use StructuredTool.from_function with an explicit Pydantic args_schema. This gives the LLM a validated schema and gives you typed inputs. Instantiate attestd.Client once at module level and capture it in the closure — never instantiate inside the tool function.

attestd_tool.py

from langchain_core.tools import StructuredTool
from pydantic import BaseModel
import attestd

# Instantiate once — captured in closure. Never instantiate inside the tool function.
_client = attestd.Client(api_key="YOUR_API_KEY")


class AttestdInput(BaseModel):
    product: str
    version: str


def check_vulnerability(product: str, version: str) -> dict:
    try:
        result = _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 attestd.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.",
        }


attestd_tool = StructuredTool.from_function(
    func=check_vulnerability,
    name="check_package_vulnerability",
    description=(
        "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 Attestd has no data — "
        "treat as unknown risk, not safe. "
        "Input: product slug (e.g. 'nginx', 'runc', 'log4j') and exact version string."
    ),
    args_schema=AttestdInput,
)

agent executor

Agent executor pattern

Use create_tool_calling_agent (LangChain 0.3+) with any function-calling LLM. The system prompt must state the deployment-blocking policy explicitly — the model cannot infer the correct behaviour from the tool description alone.

agent.py

from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

prompt = ChatPromptTemplate.from_messages([
    (
        "system",
        "You are a security-aware deployment assistant. "
        "Before approving any software dependency, check its vulnerability status "
        "using the check_package_vulnerability tool. "
        "Block deployment if risk_state is 'critical' or 'high', or if "
        "supply_chain_compromised is true. "
        "If typosquat_detected is true, block and report the resembles field. "
        "outside_coverage=true means unknown risk — state this explicitly, "
        "do not treat it as safe.",
    ),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

agent = create_tool_calling_agent(llm, [attestd_tool], prompt)
executor = AgentExecutor(agent=agent, tools=[attestd_tool], verbose=True)

# Example: check two dependencies before a deploy
response = executor.invoke({
    "input": "Is it safe to deploy with runc 1.0.0 and litellm 1.82.7?"
})
print(response["output"])

async

Async support

Use attestd.AsyncClient for async agent deployments. Instantiate at module level — not inside the coroutine. Pass coroutine= (not func=) to StructuredTool.from_function.

async_tool.py

import attestd
from langchain_core.tools import StructuredTool
from pydantic import BaseModel

# Instantiate once at module level — NOT inside the coroutine.
_async_client = attestd.AsyncClient(api_key="YOUR_API_KEY")


class AttestdInput(BaseModel):
    product: str
    version: str


async def acheck_vulnerability(product: str, version: str) -> dict:
    try:
        result = await _async_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 attestd.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.",
        }


# Pass coroutine= (not func=) for async tools
async_attestd_tool = StructuredTool.from_function(
    coroutine=acheck_vulnerability,
    name="check_package_vulnerability",
    description=(
        "Check whether a software package version has known CVE vulnerabilities "
        "or supply chain compromise. outside_coverage=true means unknown risk, not safe."
    ),
    args_schema=AttestdInput,
)

# Use with async agent executor
response = await executor.ainvoke({
    "input": "Check litellm 1.82.7 before we deploy."
})

fields

Return fields

The tool returns these fields. Design agent branching 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.