Add AgentTollSafetyTool: honeypot/rug checks for a Base token via x402 (#7227)

lib/crewai-tools/pyproject.toml

@@ -159,6 +159,9 @@ e2b = [
     "e2b~=2.20.0",
     "e2b-code-interpreter~=2.6.0",
 ]
+x402 = [
+    "x402[requests,evm]>=2.21.0",
+]
 
 
 [tool.uv]

lib/crewai-tools/src/crewai_tools/__init__.py

@@ -7,6 +7,9 @@
 )
 from crewai_tools.aws.s3.reader_tool import S3ReaderTool
 from crewai_tools.aws.s3.writer_tool import S3WriterTool
+from crewai_tools.tools.agenttoll_safety_tool.agenttoll_safety_tool import (
+    AgentTollSafetyTool,
+)
 from crewai_tools.tools.ai_mind_tool.ai_mind_tool import AIMindTool
 from crewai_tools.tools.apify_actors_tool.apify_actors_tool import ApifyActorsTool
 from crewai_tools.tools.arxiv_paper_tool.arxiv_paper_tool import ArxivPaperTool
@@ -226,6 +229,7 @@
 
 __all__ = [
     "AIMindTool",
+    "AgentTollSafetyTool",
     "ApifyActorsTool",
     "ArxivPaperTool",
     "BedrockInvokeAgentTool",

lib/crewai-tools/src/crewai_tools/tools/__init__.py

@@ -1,3 +1,6 @@
+from crewai_tools.tools.agenttoll_safety_tool.agenttoll_safety_tool import (
+    AgentTollSafetyTool,
+)
 from crewai_tools.tools.ai_mind_tool.ai_mind_tool import AIMindTool
 from crewai_tools.tools.apify_actors_tool.apify_actors_tool import ApifyActorsTool
 from crewai_tools.tools.arxiv_paper_tool.arxiv_paper_tool import ArxivPaperTool
@@ -213,6 +216,7 @@
 
 __all__ = [
     "AIMindTool",
+    "AgentTollSafetyTool",
     "ApifyActorsTool",
     "ArxivPaperTool",
     "BraveImageSearchTool",

lib/crewai-tools/src/crewai_tools/tools/agenttoll_safety_tool/README.md

@@ -0,0 +1,56 @@
+# AgentTollSafetyTool Documentation
+
+## Description
+
+This tool checks whether a Base (Coinbase's L2) token contract is a honeypot or a rug
+before you trade it or recommend it. It calls
+[AgentToll](https://agenttoll.app)'s `/api/base/safety` endpoint, which runs a
+simulated buy **and** sell, checks buy/sell tax, owner privileges, holder
+concentration, liquidity risk, and the deployer's own history — a token shipped from a
+wallet with a handful of transactions and dust is the shape most rugs share.
+
+There is no API key and no subscription. Every call is paid for inline, in USDC on
+Base, via the [x402 protocol](https://x402.org) (HTTP 402) — the agent's wallet pays
+$0.003 per call, and only once the response comes back successfully.
+
+## Installation
+
+```shell
+uv add crewai-tools --extra x402
+# or
+pip install 'crewai[tools]' 'x402[requests,evm]'
+```
+
+## Example
+
+```python
+from crewai_tools import AgentTollSafetyTool
+
+# EVM_PRIVATE_KEY must be set to a Base wallet holding a little USDC
+tool = AgentTollSafetyTool()
+
+result = tool.run(address="0x940181a94a35a4569e4529a3cdfb74e38fd98631")
+```
+
+## Steps to Get Started
+
+1. **Package installation**: install the `x402` extra as shown above.
+2. **Wallet funding**: the wallet behind `EVM_PRIVATE_KEY` needs a small amount of USDC
+   on Base mainnet (each call costs $0.003). To test without real funds, pass
+   `base_url="http://localhost:4021"` when constructing the tool to point at a
+   self-hosted AgentToll instance running on Base Sepolia, and fund the wallet with
+   free testnet USDC from [faucet.circle.com](https://faucet.circle.com).
+3. **Environment configuration**: `EVM_PRIVATE_KEY=0x...`
+
+## Arguments
+
+| Argument | Type | Description |
+|---|---|---|
+| `address` | `str` | Base token contract address to check, e.g. `0x1234...` |
+
+## Conclusion
+
+`AgentTollSafetyTool` gives an agent a real safety verdict on a Base token — clear,
+caution, high-risk, or insufficient-data — without an account, an API key, or a
+subscription: the agent pays for exactly the checks it runs, and nothing when a call
+fails.

lib/crewai-tools/src/crewai_tools/tools/agenttoll_safety_tool/agenttoll_safety_tool.py

@@ -0,0 +1,96 @@
+import os
+from typing import Any
+
+from crewai.tools import BaseTool, EnvVar
+from pydantic import BaseModel, Field
+
+
+class AgentTollSafetyToolInput(BaseModel):
+    """Input for AgentTollSafetyTool."""
+
+    address: str = Field(
+        ...,
+        description="Base (chain id 8453) token contract address to check, e.g. '0x1234...'",
+    )
+
+
+class AgentTollSafetyTool(BaseTool):
+    """
+    AgentTollSafetyTool - checks whether a Base token is a honeypot or rug before you trade it.
+
+    Calls AgentToll's /api/base/safety endpoint (https://agenttoll.app): a simulated buy
+    AND sell, taxes, owner privileges, holder concentration, liquidity risk, and the
+    deployer's own history. Paid per call in USDC on Base via the x402 protocol (HTTP 402)
+    -- no API key, no subscription, no signup. Costs $0.003, charged only once the
+    response comes back successfully.
+
+    Dependencies:
+        - x402[requests,evm]
+    """
+
+    name: str = "Check Base token safety"
+    description: str = (
+        "Check whether a Base token contract is a honeypot or rug: simulated buy and "
+        "sell, taxes, owner privileges, holder concentration, liquidity risk, and the "
+        "deployer's history. Use before trading or recommending an unfamiliar Base token."
+    )
+    args_schema: type[BaseModel] = AgentTollSafetyToolInput
+    base_url: str = "https://agenttoll.app"
+
+    env_vars: list[EnvVar] = Field(
+        default_factory=lambda: [
+            EnvVar(
+                name="EVM_PRIVATE_KEY",
+                description=(
+                    "Private key of a Base wallet holding a little USDC, used to pay "
+                    "$0.003 per call via x402"
+                ),
+                required=True,
+            ),
+        ]
+    )
+    package_dependencies: list[str] = Field(default_factory=lambda: ["x402"])
+
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super().__init__(*args, **kwargs)
+        try:
+            import x402  # noqa: F401
+        except ImportError as exc:
+            raise ImportError(
+                "Missing optional dependency 'x402'. Install with: \n"
+                "  uv add crewai-tools --extra x402\n"
+                "or\n"
+                "  pip install 'x402[requests,evm]'\n"
+            ) from exc
+
+        if "EVM_PRIVATE_KEY" not in os.environ:
+            raise ValueError(
+                "Environment variable EVM_PRIVATE_KEY is required for AgentTollSafetyTool"
+            )
+
+    def _paid_session(self) -> Any:
+        """A requests session that pays x402 quotes automatically, capped per call."""
+        from eth_account import Account
+        from x402 import x402ClientSync
+        from x402.http.clients import x402_requests
+        from x402.mechanisms.evm import EthAccountSigner
+        from x402.mechanisms.evm.exact.register import register_exact_evm_client
+
+        client = x402ClientSync().set_spend_controls(
+            {"max_amount_per_payment": "$0.05"}
+        )
+        account = Account.from_key(os.environ["EVM_PRIVATE_KEY"])
+        register_exact_evm_client(client, EthAccountSigner(account))
+        return x402_requests(client)
+
+    def _run(self, address: str) -> str:
+        try:
+            with self._paid_session() as session:
+                response = session.get(f"{self.base_url}/api/base/safety/{address}")
+                response.raise_for_status()
+                return str(response.text)
+        except Exception as e:
+            return f"Error checking token safety: {e}"
+
+    async def _arun(self, *args: Any, **kwargs: Any) -> str:
+        return self._run(*args, **kwargs)

lib/crewai-tools/tests/tools/agenttoll_safety_tool_test.py

@@ -0,0 +1,48 @@
+from unittest.mock import MagicMock, patch
+
+from crewai_tools.tools.agenttoll_safety_tool.agenttoll_safety_tool import (
+    AgentTollSafetyTool,
+)
+import pytest
+
+
+DUMMY_KEY = "0x" + "11" * 32  # syntactically valid, never funded
+
+
+@pytest.fixture
+def agenttoll_tool(monkeypatch):
+    monkeypatch.setenv("EVM_PRIVATE_KEY", DUMMY_KEY)
+    return AgentTollSafetyTool()
+
+
+def test_requires_env_var(monkeypatch):
+    monkeypatch.delenv("EVM_PRIVATE_KEY", raising=False)
+    with pytest.raises(ValueError):
+        AgentTollSafetyTool()
+
+
+def test_happy_path(agenttoll_tool):
+    mock_response = MagicMock()
+    mock_response.text = '{"verdict": "clear"}'
+    mock_response.raise_for_status = MagicMock()
+
+    mock_session = MagicMock()
+    mock_session.__enter__.return_value = mock_session
+    mock_session.__exit__.return_value = False
+    mock_session.get.return_value = mock_response
+
+    with patch.object(AgentTollSafetyTool, "_paid_session", return_value=mock_session):
+        result = agenttoll_tool.run(address="0x940181a94a35a4569e4529a3cdfb74e38fd98631")
+
+    assert "clear" in result
+    mock_session.get.assert_called_once_with(
+        "https://agenttoll.app/api/base/safety/0x940181a94a35a4569e4529a3cdfb74e38fd98631"
+    )
+
+
+def test_error_is_returned_not_raised(agenttoll_tool):
+    with patch.object(AgentTollSafetyTool, "_paid_session", side_effect=RuntimeError("boom")):
+        result = agenttoll_tool.run(address="0x940181a94a35a4569e4529a3cdfb74e38fd98631")
+
+    assert "Error checking token safety" in result
+    assert "boom" in result