Add phishing detection to your AI agent: Hermes Agent, LangChain, AutoGen, CrewAI (2026 tutorial)

Tested with: langchain-core 0.3.x · langgraph 0.2.x · mcp 1.0+ (Python SDK) · pyautogen 0.2.x (AutoGen 0.4 uses different imports - noted in the AutoGen section) · crewai 0.x · openai 1.x · anthropic 0.3x · agentcash 0.14.x (npm). Code follows official documentation patterns for each framework. Report bugs at info@safebrowz.com.

Why AI agents need phishing detection

Through 2025 and into 2026, AI agents stopped being toys and started being trusted with money. Agents now book travel, shop on Amazon, file expense reports, place crypto trades, and increasingly hold wallet keys delegated by their users. Every one of those workflows starts the same way: the agent reads a URL from somewhere and clicks it.

That single click is the new attack surface. A phishing page does not need to fool a human anymore. It needs to fool a token-bounded language model that has been told to "complete the task." The model does not pause at a slightly off domain or notice that the favicon is blurry. It fills the form, signs the message, and reports success.

Mandiant's M-Trends 2026 report flagged AI-driven account takeover as the fastest-growing intrusion vector across enterprise environments, and the FBI's IC3 2025 Internet Crime Report logged more than $16 billion in losses from phishing and impersonation scams combined. The defense is not asking the model to "be more careful." The defense is a pre-fetch URL safety check that returns a verdict before the agent's HTTP client ever opens the page.

What the SafeBrowz Detection API does

One endpoint, one job. Send a URL, get a safety verdict.

• Endpoint: POST https://api.safebrowz.com/v1/detect
• Output: verdict (safe / caution / danger), brand identified, reason, threat-type signals
• Latency: sub-500ms median (shared cache hits across all callers)
• Coverage: 550+ tracked brands, 60+ URL pattern signatures, 30+ disposable scam TLDs, 100+ languages on the AI deep-scan layer

Under the hood, every request passes through a 3-layer detection pipeline: local pattern matching, third-party reputation APIs, and AI content analysis - covered in detail at the bottom of this post.

Pricing models

Model	Cost	Best for	Signup
x402 pay-per-request	$0.001 USDC per call	Hobby agents, prototypes, anything under ~10k calls per month	None - just a wallet
Enterprise Bearer key	Flat monthly USDC invoice, negotiated	Wallet apps, large agent platforms, 10k+ calls per month	Email info@safebrowz.com

The x402 path is the default. It uses the HTTP 402 Payment Required status code and a USDC micropayment on Solana or Base. No account, no API key rotation, no billing portal. Your agent pays for what it uses, your wallet stays in your wallet, and the server-side ledger is on-chain.

The Bearer key path skips the on-chain dance for high-volume callers. You send Authorization: Bearer sk_live_... and the request settles instantly against your monthly quota. Usage is invoiced in USDC at the end of the period.

Framework integration cookbook

Below is one complete integration per framework. Pick the one that matches your stack and copy-paste. All examples assume Bearer-key auth for clarity - swap in the x402 client (covered in section "x402 payment integration" further down) if you want pay-per-request instead.

Hermes Agent (Nous Research)

Hermes Agent runs MCP servers as first-class capability sources. The cleanest integration is to expose SafeBrowz as a small MCP server the agent can call as a tool. The skeleton below uses Python and the official mcp package.

"""safebrowz-mcp-server.py - minimal MCP server exposing /v1/detect."""
import os
import httpx
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("safebrowz")
SB_KEY = os.environ["SAFEBROWZ_BEARER"]   # provisioned by SafeBrowz
SB_URL = "https://api.safebrowz.com/v1/detect"

@mcp.tool()
async def scan_url(url: str) -> dict:
    """Return a phishing verdict for the given URL.

    Result fields:
      - verdict: 'safe' | 'caution' | 'danger'
      - brand:   detected impersonated brand or null
      - reason:  short human-readable explanation
    """
    async with httpx.AsyncClient(timeout=10) as client:
        r = await client.post(
            SB_URL,
            headers={"Authorization": f"Bearer {SB_KEY}"},
            json={"url": url},
        )
        r.raise_for_status()
        return r.json()

if __name__ == "__main__":
    mcp.run()

Register the server in your Hermes Agent MCP config (see the Hermes Agent docs at hermes-agent.nousresearch.com):

{
  "mcpServers": {
    "safebrowz": {
      "command": "python",
      "args": ["/path/to/safebrowz-mcp-server.py"],
      "env": { "SAFEBROWZ_BEARER": "sk_live_..." }
    }
  }
}

Now any time Hermes Agent reasons about visiting a URL, it can call scan_url first. A typical system prompt addition: "Before opening any URL with a browser tool, call safebrowz.scan_url first. If the verdict is danger, refuse and tell the user why."

LangChain (Python)

LangChain's @tool decorator is the cleanest path. Below is a sync and an async version - LangChain agents using LCEL/LangGraph almost always want async.

"""safebrowz_tool.py - LangChain tool wrapper."""
import os
import httpx
from langchain_core.tools import tool

SB_KEY = os.environ["SAFEBROWZ_BEARER"]
SB_URL = "https://api.safebrowz.com/v1/detect"

@tool
def check_url_safety(url: str) -> str:
    """Check whether a URL is a known phishing site or impersonates a brand.

    Returns a short string of the form 'verdict: reason'. Call this BEFORE
    fetching, clicking, or navigating to any URL provided by a user or
    discovered in a tool result.
    """
    r = httpx.post(
        SB_URL,
        headers={"Authorization": f"Bearer {SB_KEY}"},
        json={"url": url},
        timeout=10,
    )
    data = r.json()
    return f"{data['verdict']}: {data.get('reason', 'no detail')}"

@tool
async def check_url_safety_async(url: str) -> str:
    """Async version of check_url_safety. Use in LangGraph / LCEL pipelines."""
    async with httpx.AsyncClient(timeout=10) as client:
        r = await client.post(
            SB_URL,
            headers={"Authorization": f"Bearer {SB_KEY}"},
            json={"url": url},
        )
    data = r.json()
    return f"{data['verdict']}: {data.get('reason', 'no detail')}"

Wire the tool into an agent and instruct the model to call it first:

from langchain_anthropic import ChatAnthropic
from langgraph.prebuilt import create_react_agent
from safebrowz_tool import check_url_safety_async

llm   = ChatAnthropic(model="claude-opus-4-7")
agent = create_react_agent(
    llm,
    tools=[check_url_safety_async, your_browser_tool],
    state_modifier=(
        "You are a research agent. Before fetching ANY URL with the "
        "browser tool, call check_url_safety_async first. If the verdict "
        "is 'danger', stop and report the threat to the user instead of "
        "browsing."
    ),
)
result = await agent.ainvoke({"messages": [("user", "Visit trezor-support.com and summarise.")]} )

AutoGen (Microsoft Research)

AutoGen registers Python functions as callable tools and lets a UserProxyAgent execute them in response to function calls from an AssistantAgent. The pattern below uses the AutoGen 0.2.x API (the most widely deployed version). Note: AutoGen 0.4+ uses a different package layout (autogen-core + autogen-agentchat) - the function-registration concept is the same but imports change. The example below works with pip install pyautogen.

"""autogen_safebrowz.py - phishing-aware AutoGen pair."""
import os
import httpx
from autogen import AssistantAgent, UserProxyAgent, register_function

SB_KEY = os.environ["SAFEBROWZ_BEARER"]
SB_URL = "https://api.safebrowz.com/v1/detect"

def check_url_safety(url: str) -> dict:
    """Return a phishing verdict for a URL."""
    r = httpx.post(
        SB_URL,
        headers={"Authorization": f"Bearer {SB_KEY}"},
        json={"url": url},
        timeout=10,
    )
    return r.json()

assistant = AssistantAgent(
    name="researcher",
    llm_config={"config_list": [{"model": "gpt-4o", "api_key": os.environ["OPENAI_API_KEY"]}]},
    system_message=(
        "You research URLs the user asks about. ALWAYS call "
        "check_url_safety on a URL before recommending it. If the "
        "verdict is 'danger', warn the user and do not summarise the page."
    ),
)
user_proxy = UserProxyAgent(
    name="executor",
    human_input_mode="NEVER",
    code_execution_config=False,
)
register_function(
    check_url_safety,
    caller=assistant,
    executor=user_proxy,
    name="check_url_safety",
    description="Check whether a URL is a known phishing or scam site.",
)

user_proxy.initiate_chat(
    assistant,
    message="Is https://trezor-support.com safe to visit?",
)

The flow at runtime: the assistant emits a function call for check_url_safety, the user proxy executes it, the result returns to the assistant, and the assistant composes its reply based on the verdict.

CrewAI

CrewAI uses a @tool decorator nearly identical to LangChain's, but tools attach to individual Agent roles inside a Crew. The natural pattern is to give your security-conscious researcher agent the URL checker.

"""crew_safebrowz.py - CrewAI research crew with phishing safety."""
import os
import httpx
from crewai import Agent, Task, Crew
from crewai.tools import tool

SB_KEY = os.environ["SAFEBROWZ_BEARER"]
SB_URL = "https://api.safebrowz.com/v1/detect"

@tool("Phishing URL Checker")
def url_safety(url: str) -> str:
    """Check whether a URL is a phishing or wallet-drainer site.

    Always run this before opening, summarising, or recommending a URL.
    """
    r = httpx.post(
        SB_URL,
        headers={"Authorization": f"Bearer {SB_KEY}"},
        json={"url": url},
        timeout=10,
    )
    d = r.json()
    return f"verdict={d['verdict']} brand={d.get('brand')} reason={d.get('reason')}"

researcher = Agent(
    role="Web Researcher",
    goal="Summarise web pages the user asks about, but never visit phishing sites.",
    backstory="A cautious researcher who always verifies URLs before opening them.",
    tools=[url_safety, your_browser_tool],
    verbose=True,
)
task = Task(
    description="Look up trezor-support.com and report what it is.",
    agent=researcher,
    expected_output="A safety verdict plus a summary or refusal.",
)
Crew(agents=[researcher], tasks=[task]).kickoff()

OpenAI Assistants API

The Assistants API uses a JSON tool spec attached to the assistant on creation, plus a run-loop that handles requires_action events when the model wants a function called. Below is the spec and the loop.

{
  "type": "function",
  "function": {
    "name": "check_url_safety",
    "description": "Check whether a URL is a known phishing site. Call before visiting any URL.",
    "parameters": {
      "type": "object",
      "properties": {
        "url": {
          "type": "string",
          "description": "Fully qualified URL including scheme (https://...)."
        }
      },
      "required": ["url"]
    }
  }
}

"""openai_assistant_safebrowz.py - Assistants API run loop."""
import json
import os
import time
import httpx
from openai import OpenAI

client = OpenAI()
SB_KEY = os.environ["SAFEBROWZ_BEARER"]

def check_url_safety(url: str) -> dict:
    r = httpx.post(
        "https://api.safebrowz.com/v1/detect",
        headers={"Authorization": f"Bearer {SB_KEY}"},
        json={"url": url},
        timeout=10,
    )
    return r.json()

assistant = client.beta.assistants.create(
    name="Safe Researcher",
    instructions=(
        "Before visiting any URL, call check_url_safety. If verdict is "
        "'danger', refuse to open the page and tell the user why."
    ),
    model="gpt-4o",
    tools=[{
        "type": "function",
        "function": {
            "name": "check_url_safety",
            "description": "Check if a URL is phishing.",
            "parameters": {
                "type": "object",
                "properties": {"url": {"type": "string"}},
                "required": ["url"],
            },
        },
    }],
)

thread = client.beta.threads.create()
client.beta.threads.messages.create(
    thread_id=thread.id, role="user",
    content="Should I trust trezor-support.com?",
)
run = client.beta.threads.runs.create(thread_id=thread.id, assistant_id=assistant.id)

while run.status in ("queued", "in_progress", "requires_action"):
    if run.status == "requires_action":
        outputs = []
        for call in run.required_action.submit_tool_outputs.tool_calls:
            args = json.loads(call.function.arguments)
            result = check_url_safety(args["url"])
            outputs.append({"tool_call_id": call.id, "output": json.dumps(result)})
        run = client.beta.threads.runs.submit_tool_outputs(
            thread_id=thread.id, run_id=run.id, tool_outputs=outputs,
        )
    else:
        time.sleep(0.5)
        run = client.beta.threads.runs.retrieve(thread_id=thread.id, run_id=run.id)

Anthropic Claude (tool use API)

Claude's tool-use API takes tool definitions inline on each messages.create call. There is no persistent assistant resource; you pass the tools every turn. Below is a complete one-shot example.

"""claude_tool_use.py - one-shot Claude tool-use with SafeBrowz."""
import json
import os
import httpx
from anthropic import Anthropic

claude = Anthropic()
SB_KEY = os.environ["SAFEBROWZ_BEARER"]

TOOLS = [{
    "name": "check_url_safety",
    "description": "Return a phishing verdict for the given URL. Call BEFORE opening any URL.",
    "input_schema": {
        "type": "object",
        "properties": {"url": {"type": "string", "description": "https://..."}},
        "required": ["url"],
    },
}]

def run_tool(name: str, args: dict) -> dict:
    if name == "check_url_safety":
        r = httpx.post(
            "https://api.safebrowz.com/v1/detect",
            headers={"Authorization": f"Bearer {SB_KEY}"},
            json={"url": args["url"]},
            timeout=10,
        )
        return r.json()
    raise ValueError(name)

messages = [{"role": "user", "content": "Is trezor-support.com safe?"}]
while True:
    resp = claude.messages.create(
        model="claude-opus-4-7",
        max_tokens=1024,
        tools=TOOLS,
        messages=messages,
    )
    if resp.stop_reason != "tool_use":
        print(resp.content[0].text)
        break
    messages.append({"role": "assistant", "content": resp.content})
    tool_results = []
    for block in resp.content:
        if block.type == "tool_use":
            result = run_tool(block.name, block.input)
            tool_results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": json.dumps(result),
            })
    messages.append({"role": "user", "content": tool_results})

Raw HTTP (any language)

If you are integrating into a framework not listed above (Hugging Face Agents, smol-developer, custom Go/Rust agents, anything self-rolled), the API is plain JSON over HTTPS. Two examples:

curl -X POST https://api.safebrowz.com/v1/detect \
  -H "Authorization: Bearer sk_live_..." \
  -H "Content-Type: application/json" \
  -d '{"url":"https://trezor-support.com"}'

// Node / Deno / Bun
const res = await fetch("https://api.safebrowz.com/v1/detect", {
  method: "POST",
  headers: {
    "Authorization": `Bearer ${process.env.SAFEBROWZ_BEARER}`,
    "Content-Type": "application/json",
  },
  body: JSON.stringify({ url: "https://trezor-support.com" }),
});
const verdict = await res.json();
console.log(verdict);

For x402 pay-per-request flows without a Bearer key, omit the Authorization header. The first response will be a 402 Payment Required with a JSON challenge describing accepted chains and tokens. Pay the $0.001 USDC on-chain and retry with an X-PAYMENT header containing the proof - covered next.

x402 payment integration (when you do not have a Bearer key)

x402 is the HTTP 402 Payment Required status code (reserved in the HTTP/1.1 spec since 1999), given a concrete payment protocol by Coinbase in 2024. The flow: client POSTs with no auth, server returns 402 with a JSON challenge listing accepted chains, client pays $0.001 USDC on-chain (SPL-memo nonce binding on Solana, EIP-191 signature on Base), client retries with an X-PAYMENT base64 header, server verifies and returns the verdict. End-to-end latency is 1 to 3 seconds.

You do not have to implement this by hand. The agentcash npm package wraps the full flow:

npx agentcash fetch https://api.safebrowz.com/v1/detect \
  --body '{"url":"https://trezor-support.com"}' \
  --network solana

For high-volume agents where per-call signing is impractical, email info@safebrowz.com for a Bearer key.

Response format

The response is intentionally compact. Every field is optional except verdict.

{
  "verdict": "danger",
  "brand": "Coinbase",
  "reason": "Lookalike domain with login form on free hosting",
  "url": "evil-coinbase.help.com",
  "domain": "evil-coinbase.help.com"
}

Field	Type	Meaning
verdict	string	safe, caution, or danger
brand	string or null	Detected impersonated brand from the 550+ brand database
reason	string	Short human-readable explanation, safe to show to end users
url	string	The URL as submitted
domain	string	Parsed domain

The full enterprise response also includes 23 boolean threat signals (brand_impersonation, wallet_drainer, fake_captcha, pastejacking, credential_phishing, and so on), a trust_score from 0 to 100, and a confidence float. See /api-docs for the complete schema or pull /openapi.json directly for code generation.

Best practices and patterns

• Always pre-fetch validate. The check has to happen before your HTTP client opens the URL. A scan after the page has rendered is too late if the page runs JavaScript drainers or pastejacks the clipboard.
• Cache verdicts for 24 hours. Domain reputation does not flip second by second. A simple client-side cache eliminates 80%+ of repeat calls. The server already caches across all callers, so client-side caching is purely a cost and latency optimisation.
• Show users the verdict and reason, not just a block. "I refused to open this URL because it impersonates Coinbase on a free-hosting subdomain" is dramatically more useful than "blocked." Pass the reason field through to your UI.
• Request the deep AI scan when needed. Pass "deep": true in the request body to force the AI content layer for novel domains the local engine has not seen.
• Rate-limit your own usage. Wrap your tool call in a per-session budget (e.g. max 50 URL checks per conversation) to bound cost when the agent misbehaves.
• Never put user PII in the URL field. Strip query strings containing emails, tokens, or session IDs before scanning.

Two patterns worth naming. Pre-fetch validation wraps every fetch(), requests.get(), or page.goto() call your agent makes:

async def safe_fetch(url: str, **kwargs):
    verdict = await check_url_safety_async.ainvoke({"url": url})
    if "danger" in verdict:
        raise PhishingDetected(url, verdict)
    return await httpx.AsyncClient().get(url, **kwargs)

And in multi-agent crews (CrewAI, AutoGen), the cleanest architecture gives URL safety its own role. The Researcher asks the Security Officer for a scan before the Browser agent ever loads the page. Same idea applies to chat moderation: scan user-pasted URLs before the model reads them, so a poisoned support DM cannot smuggle a link past the model's attention.

How SafeBrowz blocks this threat

SafeBrowz runs a 3-layer detection architecture: Local + APIs + AI.

• Layer 1 - Local detection: 60+ URL pattern signatures + 550+ brand-specific signatures (including Cyrillic and Punycode homograph variants) + community whitelist and blacklist, all evaluated before any network call. Catches typosquats, free-hosting brand subdomains, drainer-path patterns, and homograph variants instantly.
• Layer 2 - API checks: aggregates Google Safe Browsing, PhishTank, URLhaus, ScamAdviser, and 30+ disposable scam TLDs for known-malicious domains.
• Layer 3 - AI deep scan (Premium): 100+ language content analysis catches novel variants in seconds. Used when local and API layers do not return a confident verdict, or when the caller passes "deep": true.

Detection signatures come from threat-intelligence research and brand database analysis, not from user browsing data. Per-user URL history is never stored.

Frequently asked questions

Why pay per request instead of a flat subscription?

Because most agents do not know up front how many URLs they will need to check. A research agent might scan 5 URLs in one conversation and 500 in the next. Per-request pricing at $0.001 USDC means you pay exactly for what the agent actually does. If your volume crosses the threshold where a flat rate is cheaper - typically around 10,000 calls per month - email info@safebrowz.com for a Bearer-key contract.

Is x402 actually production-ready?

Yes. HTTP 402 has been reserved in the HTTP/1.1 spec since 1999, and the x402 payload format Coinbase published in 2024 is now implemented by multiple production services including SafeBrowz. The agentcash CLI, Coinbase's Bazaar marketplace, and the CDP wallet SDK all speak it. Settlement is ~400ms on Solana and ~2 seconds on Base. End-to-end request latency is 1 to 3 seconds.

How does this compare to Google Safe Browsing?

Google Safe Browsing is one of the inputs to our Layer 2 reputation aggregation, alongside PhishTank, URLhaus, and ScamAdviser. SafeBrowz adds two layers Google does not: a local pattern engine tuned for brand-impersonation and wallet-drainer signatures (Layer 1), and an AI deep-scan layer that reasons about the page content in 100+ languages (Layer 3). For an AI agent specifically, the AI layer is the differentiator - Google's blocklist is fast on known phishing but slow on the day-zero domains agents will encounter most.

What if my agent makes thousands of calls per day?

Either turn on client-side caching (24-hour TTL eliminates ~80% of repeat lookups for typical workloads) or contact us for a Bearer key. The Bearer tier is designed for wallet apps and large agent platforms running tens of millions of checks per month. Rate limits and pricing are negotiated at provisioning.

Can I cache verdicts?

Yes, and you should. Domain reputation changes on the order of hours to days, not seconds. A 24-hour client-side cache is the recommended default. For wallet apps proxying user clicks, a shared 1-hour Redis cache is typical. The server already runs a shared cross-caller cache, so caching client-side is purely a cost and latency optimisation - never a correctness risk.

Does this work for non-crypto agents (just regular browsing)?

Absolutely. Wallet drainers are the most expensive failure mode, but credential phishing, fake-CAPTCHA pastejacking, and brand impersonation hit non-crypto users just as hard. The same endpoint flags fake Apple ID pages, Microsoft 365 login clones, Amazon order-confirmation scams, DHL delivery phish, and the rest of the everyday phishing surface. The 550+ brand database is roughly half crypto, half traditional.

Get started in 30 seconds

The fastest path is a single cURL against the live endpoint. Replace the URL with anything you want to test:

curl -X POST https://api.safebrowz.com/v1/detect \
  -H "Content-Type: application/json" \
  -d '{"url":"https://trezor-support.com"}'

You will get a 402 Payment Required back with the x402 challenge. Pipe that through agentcash or your own wallet to settle the $0.001 USDC, retry, and read the verdict. Or skip the dance and email info@safebrowz.com for a Bearer key.

Full reference at safebrowz.com/api-docs. Machine-readable OpenAPI spec at safebrowz.com/openapi.json. Agent skill file at safebrowz-api.skill.md - drop it into any Claude skills directory or MCP server config and your agent will know how to call us.

AI agents are about to do most of what humans currently do on the web. The ones that ship a URL safety check before every fetch will not be the ones in the post-mortem. Build that check in now, while it is twenty lines of code.