rugguard-langgraph-agent 0.1.1

LangGraph integration kit for RugGuard — pre-trade safety as a typed node. Pay-per-call via x402 on Base mainnet.

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

Policy	Blocks at	Cautions at	Allows below
conservative	score ≥ 51 (medium_risk)	score 26-50	score ≤ 25
balanced (default)	score ≥ 71 (high_risk)	score 51-70	score ≤ 50
aggressive	score ≥ 91 (critical)	score 71-90	score ≤ 70

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