Build with Lazaretto

const LZ = "https://lazaretto.dev";
type Decision = { decision: "allow" | "ask" | "deny"; why: string };

/** Decide whether an agent may install name@version. Only a definite answer allows. */
export async function preInstallCheck(name: string, version: string, apiKey?: string): Promise<Decision> {
  // 1. Free, no key: is this exact version listed as malware, or already attested?
  const subject = encodeURIComponent(`npm:${name}@${version}`);
  const res = await fetch(`${LZ}/v1/attestations/${subject}`);
  if (!res.ok) return { decision: "ask", why: `identity check unavailable (${res.status})` };
  const id = await res.json();
  if (id.found) {
    // Someone already paid for a scan of this version: a signed verdict exists.
    if (id.contradicted || id.verdict === "malicious") return { decision: "deny", why: "known bad" };
    if (id.risk === "critical" || id.risk === "high") return { decision: "deny", why: `attested risk ${id.risk}` };
    if (id.advisory_check) return { decision: "ask", why: "advisory recheck gave no answer" }; // never clear
  } else {
    const listed = id.identity_check?.listed_as_malware;
    if (listed === true) return { decision: "deny", why: "listed as malware: " + id.identity_check.advisory_ids.join(", ") };
    if (listed !== false) return { decision: "ask", why: "no definite answer" }; // null means unchecked, never clear
  }
  if (!apiKey) return { decision: "allow", why: "not listed; code not read" };

  // 2. With a key: read the code. One credit per verdict, nothing for an error.
  const r = await fetch(`${LZ}/v1/scan`, {
    method: "POST",
    headers: { "content-type": "application/json", "x-api-key": apiKey },
    body: JSON.stringify({ target: { type: "npm_package", ref: `${name}@${version}` }, depth: "full" }),
  });
  if (!r.ok) return { decision: "ask", why: `scan unavailable (${r.status})` };
  const scan = await r.json();
  if (scan.verdict === "error") return { decision: "ask", why: "scan did not complete" };
  if (scan.risk === "critical" || scan.risk === "high") return { decision: "deny", why: scan.risk_summary };
  if (scan.risk === "none" || scan.risk === "low") return { decision: "allow", why: scan.risk_summary };
  return { decision: "ask", why: scan.risk_summary }; // medium, unknown, or a value this code does not know
}

import requests
from urllib.parse import quote

LZ = "https://lazaretto.dev"

def pre_install_check(name: str, version: str, api_key: str | None = None) -> tuple[str, str]:
    """Return ("allow" | "ask" | "deny", why). Only a definite answer allows."""
    # 1. Free, no key: is this exact version listed as malware, or already attested?
    subject = quote(f"npm:{name}@{version}", "")  # encode "/" and "@" too
    r = requests.get(f"{LZ}/v1/attestations/{subject}", timeout=10)
    if not r.ok:
        return "ask", f"identity check unavailable ({r.status_code})"
    body = r.json()
    if body.get("found"):
        # Someone already paid for a scan of this version: a signed verdict exists.
        if body.get("contradicted") or body.get("verdict") == "malicious":
            return "deny", "known bad"
        if body.get("risk") in ("critical", "high"):
            return "deny", f"attested risk {body['risk']}"
        if body.get("advisory_check"):
            return "ask", "advisory recheck gave no answer"  # never clear
    else:
        ident = body.get("identity_check") or {}
        listed = ident.get("listed_as_malware")
        if listed is True:
            return "deny", "listed as malware: " + ", ".join(ident.get("advisory_ids", []))
        if listed is not False:
            return "ask", "no definite answer"  # None means unchecked, never clear
    if not api_key:
        return "allow", "not listed; code not read"

    # 2. With a key: read the code. One credit per verdict, nothing for an error.
    r = requests.post(
        f"{LZ}/v1/scan",
        headers={"x-api-key": api_key},
        json={"target": {"type": "npm_package", "ref": f"{name}@{version}"}, "depth": "full"},
        timeout=60,
    )
    if not r.ok:
        return "ask", f"scan unavailable ({r.status_code})"
    scan = r.json()
    if scan["verdict"] == "error":
        return "ask", "scan did not complete"
    if scan["risk"] in ("critical", "high"):
        return "deny", scan["risk_summary"]
    if scan["risk"] in ("none", "low"):
        return "allow", scan["risk_summary"]
    return "ask", scan["risk_summary"]  # medium, unknown, or a value this code does not know

// npm i @modelcontextprotocol/client
import { Client, StreamableHTTPClientTransport } from "@modelcontextprotocol/client";

const client = new Client({ name: "my-agent", version: "1.0.0" });
await client.connect(
  new StreamableHTTPClientTransport(new URL("https://lazaretto.dev/mcp"), {
    // Omit requestInit for the free tools only.
    requestInit: { headers: { "X-API-Key": process.env.LAZARETTO_API_KEY! } },
  }),
);

const result = await client.callTool({
  name: "find_attestation",
  arguments: { subject: "npm:chalk@5.6.1" },
});
await client.close();

// npm i @modelcontextprotocol/sdk   (v1)
import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";

const client = new Client({ name: "my-agent", version: "1.0.0" });
await client.connect(new StreamableHTTPClientTransport(new URL("https://lazaretto.dev/mcp")));

const result = await client.callTool({
  name: "check_lockfile",
  arguments: { packages: ["chalk@5.6.1", "express@4.21.2"] },
});

# pip install "mcp>=2,<3"
import asyncio
from mcp import Client

async def main() -> None:
    async with Client("https://lazaretto.dev/mcp") as client:
        result = await client.call_tool("find_attestation", {"subject": "npm:chalk@5.6.1"})
        print(result)

asyncio.run(main())
# For the metered tools, pass an httpx2.AsyncClient carrying the X-API-Key header
# to mcp.client.streamable_http.streamable_http_client and hand that to Client.

# pip install openai-agents
import asyncio, os
from agents import Agent, Runner
from agents.mcp import MCPServerStreamableHttp

async def main() -> None:
    async with MCPServerStreamableHttp(
        name="lazaretto",
        params={
            "url": "https://lazaretto.dev/mcp",
            "headers": {"X-API-Key": os.environ["LAZARETTO_API_KEY"]},  # omit for the free tools
        },
        cache_tools_list=True,
    ) as lazaretto:
        agent = Agent(
            name="Installer",
            instructions="Before installing any package, check it with the lazaretto tools.",
            mcp_servers=[lazaretto],
        )
        result = await Runner.run(agent, "Is chalk@5.6.1 fine to install?")
        print(result.final_output)

asyncio.run(main())

from openai import OpenAI

client = OpenAI()
resp = client.responses.create(
    model="YOUR_MODEL",
    tools=[{
        "type": "mcp",
        "server_label": "lazaretto",
        "server_url": "https://lazaretto.dev/mcp",
        "headers": {"X-API-Key": "YOUR_KEY"},  # omit for the free tools
        "require_approval": "never",
    }],
    input="Check chalk@5.6.1 and debug@4.4.2 before I install them.",
)
print(resp.output_text)

# From Perplexity's docs. Not run by us.
from perplexity import Perplexity

client = Perplexity()
response = client.responses.create(
    model="YOUR_MODEL",
    tools=[{
        "type": "mcp",
        "server_label": "lazaretto",
        "server_url": "https://lazaretto.dev/mcp?ref=perplexity",
        # "headers": {"X-API-Key": "YOUR_KEY"},  # for the scans that read code
    }],
    input="Check chalk@5.6.1 and chalk@5.3.0 before I install them.",
)
print(response.output_text)

# pip install langchain-mcp-adapters   (pins mcp<2)
import asyncio
from langchain_mcp_adapters.client import MultiServerMCPClient

async def main() -> None:
    client = MultiServerMCPClient({
        "lazaretto": {
            "transport": "http",
            "url": "https://lazaretto.dev/mcp",
            "headers": {"X-API-Key": "YOUR_KEY"},  # omit for the free tools
        }
    })
    tools = await client.get_tools()  # hand these to your LangGraph agent
    print([t.name for t in tools])

asyncio.run(main())

// npm i ai @ai-sdk/mcp
import { createMCPClient } from "@ai-sdk/mcp";

const mcp = await createMCPClient({
  transport: {
    type: "http",
    url: "https://lazaretto.dev/mcp",
    headers: { "X-API-Key": process.env.LAZARETTO_API_KEY! }, // omit for the free tools
  },
});
try {
  const tools = await mcp.tools(); // pass to generateText or streamText
} finally {
  await mcp.close();
}