rugguard-langgraph-agent

LangGraph integration kit for RugGuard. Drop a typed pretrade_check node into any StateGraph to gate every trade behind RugGuard's prescriptive pre-trade safety check. Every call returns a Pydantic model with a block | caution | allow recommendation, plus a clamped max_suggested_exposure_usd and a signed JSON report (Ed25519). Pay-per-call via x402 micropayments on Base mainnet.

Why

If your LangGraph agent buys tokens, it should run a pre-trade check first. RugGuard wraps 14 heuristics on Base + 5 on Solana SPL into a single $0.01 USDC call. The node returns a typed result so your conditional edges can branch cleanly on state["decision"] without parsing strings.

Two surfaces in this kit:

• make_pretrade_check_node(...) — returns an async LangGraph node ready to plug into StateGraph.add_node. Reads chain + contract + intended_trade_usd from state, returns pretrade_result, decision, max_exposure_usd, and pretrade_error.
• pretrade_check_async(...) — framework-agnostic typed async function. Call it directly from any runtime.

Install

pip install rugguard-langgraph-agent

60-second tour

from langgraph.graph import END, START, StateGraph
from rugguard_langgraph_agent import (
    TradingState,
    make_pretrade_check_node,
    DecisionCache,
)

async def execute_buy(state):
    # Use state["max_exposure_usd"] — RugGuard already clamped it.
    size = state["max_exposure_usd"]
    return {"executed": True, "note": f"bought ${size:.2f}"}

async def skip(state):
    return {"executed": False, "note": "blocked by RugGuard"}

graph = StateGraph(TradingState)
graph.add_node("pretrade_check", make_pretrade_check_node(
    policy="balanced", cache=DecisionCache(),
))
graph.add_node("execute_buy", execute_buy)
graph.add_node("skip", skip)

graph.add_edge(START, "pretrade_check")
graph.add_conditional_edges(
    "pretrade_check",
    lambda state: state.get("decision") or "block",
    {"allow": "execute_buy", "caution": "execute_buy", "block": "skip"},
)
graph.add_edge("execute_buy", END)
graph.add_edge("skip", END)

app = graph.compile()
final = await app.ainvoke({
    "chain": "base",
    "contract": "0x4ed4E862860beD51a9570b96d89aF5E1B0Efefed",
    "intended_trade_usd": 250.0,
})
print(final)

The pretrade node automatically populates state["decision"] (which the conditional edge reads), state["max_exposure_usd"] (clamped size your buy node should use), and state["pretrade_result"] (full typed Pydantic model for logging / auditing).

Run the demo (no LLM, no network)

pip install rugguard-langgraph-agent
rugguard-langgraph-demo --demo

Runs a 3-node StateGraph (pretrade_check → execute_buy | skip) through 3 representative tokens. No payment, no LLM key needed.

Run the live demo (real $0.01 payment, no LLM)

# Get a funded x402 wallet (Base mainnet, ≥ $0.05 USDC)
# Generate one with: python -m rugguard_mcp init  (from rugguard-mcp)
export RUGGUARD_X402_PRIVATE_KEY=0xYOUR_PRIVATE_KEY_HEX

rugguard-langgraph-demo --live --chain base \
  --contract 0x4ed4E862860beD51a9570b96d89aF5E1B0Efefed \
  --size 100

The graph routes through pretrade_check against rugguard.redfleet.fr, then through either execute_buy or skip based on the verdict.

State shape

class TradingState(TypedDict, total=False):
    # --- Set by the caller ---
    chain: str                       # "base" | "solana"
    contract: str                    # token address
    intended_trade_usd: float        # > 0
    policy: Policy                   # optional override of factory default

    # --- Set by make_pretrade_check_node ---
    pretrade_result: PreTradeCheckResult | None
    decision: PolicyRecommendation | None       # "block" | "caution" | "allow"
    max_exposure_usd: float | None
    pretrade_error: PreTradeCheckError | None

pretrade_error is set on a recoverable failure (missing creds, payment rejected, non-200, network error). When set, the node sets decision="block" and max_exposure_usd=0.0 so the conservative branch fires. Inspect pretrade_error.error if you want to distinguish retry-worthy from terminal failures.

Policy modes

An uncertain verdict (sparse data) returns caution in all modes.

Signed reports

When the deployment has Ed25519 signing configured (production rugguard.redfleet.fr does as of 2026-05-17, fingerprint a0c71156d8747078), state["pretrade_result"] carries signature and key_fingerprint. Verify offline:

pip install rugguard-verify
python -c "
import asyncio, json
from rugguard_verify import fetch_pubkey, verify_signed_report
result = state['pretrade_result'].model_dump(mode='json')
pubkey = fetch_pubkey()['pubkey_base64']
print(verify_signed_report(result, pubkey))
"

The disclaimer field is inside the signed canonical bytes. Stripping or rewriting it breaks signature verification by design.

Safety

This kit is intentionally minimal (~400 LOC across all modules) so you can read it end-to-end before forking. It is not spend-capped. For production use:

• Install rugguard-mcp and import from rugguard_mcp.x402_client import paid_post instead of the inline x402_pay.paid_post — that ships session caps + 24h caps + EIP-712 domain enforcement + file-locked TOCTOU-safe charge reservation.
• Add your own retry policy + circuit breaker upstream of the pretrade node.
• Use a dedicated x402 wallet, funded only with the USDC you are willing to spend.

The asset whitelist IS enforced in this kit (USDC on Base / Base Sepolia only). A malicious 402 trying to drain a different EIP-3009 token in your wallet is rejected before signing.

How the node works under the hood

1. Node reads chain, contract, intended_trade_usd, policy from state.
2. Checks the in-memory DecisionCache. Cache hit → return cache_hit=True, no payment.
3. POST https://rugguard.redfleet.fr/v1/pretrade/check. Server returns 402 Payment Required with x402 spec body.
4. Signs an EIP-3009 TransferWithAuthorization for $0.01 USDC. Retries the POST with X-Payment header.
5. Server settles via Coinbase CDP facilitator, returns 200 with the typed signed response.
6. Node returns the state update dict. LangGraph applies the update.

Round trip: ~300-500ms cache miss, ~1ms cache hit.

Self-host / testnet

make_pretrade_check_node(
    api_url="https://my-rugguard.example.com",  # or http://localhost:8000
)

Also reads RUGGUARD_API_URL from the environment if no api_url argument is passed.

License

MIT. See LICENSE.

See also

• RugGuard — the pre-trade safety API
• rugguard-pydantic-ai-agent — Pydantic AI variant of this kit (typed tool instead of graph node)
• rugguard-mcp — MCP server for Claude Desktop / Cursor
• rugguard-verify — stand-alone Ed25519 signed-report verifier CLI
• LangGraph — the graph framework this kit plugs into