File size: 3,238 Bytes
bde2f3a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 | """x402 LangChain adapter — transparent payment for LangChain tools.
Usage:
from x402_frameworks import x402LangChain
from langchain.tools import tool
# Initialize x402
x402 = x402LangChain(api_key="your-key", auto_pay=True)
# Wrap any tool — payment is handled automatically
@x402.wrap_tool("token_scan")
@tool
def scan_token(address: str) -> dict:
\"\"\"Scan a token for rug pull indicators.\"\"\"
return {"risk": "low", "score": 85}
# Use with any LangChain agent
from langchain.agents import create_react_agent
agent = create_react_agent(llm, [scan_token], prompt)
# When agent calls scan_token, x402 checks payment first
"""
from __future__ import annotations
import functools
import logging
from typing import Any, Callable, Optional
from .base import x402Adapter
logger = logging.getLogger("x402.langchain")
class x402LangChain(x402Adapter):
"""LangChain adapter — wraps tools with automatic x402 payment.
Every tool call checks:
1. Free trial available? → use it
2. Prepaid credits? → deduct
3. Auto-pay enabled? → create invoice (dev must fund wallet)
4. None? → raise PaymentRequired error
"""
def wrap_tool(self, tool_name: str, tool_func: Optional[Callable] = None) -> Callable:
"""Decorator: wrap a LangChain tool function with x402 payment.
Usage as decorator:
@x402.wrap_tool("token_scan")
@tool
def my_tool(address: str): ...
Usage as wrapper:
wrapped = x402.wrap_tool("token_scan")(my_tool_func)
"""
def decorator(func: Callable) -> Callable:
@functools.wraps(func)
def wrapper(*args, **kwargs):
available, price, source = self.check_available(tool_name)
if not available:
invoice = self.get_invoice(tool_name)
raise PaymentRequiredError(
tool=tool_name,
price_usd=price,
pay_to=invoice.get("pay_to", ""),
chain=invoice.get("chain", "base"),
invoice_id=invoice.get("id", ""),
)
result = func(*args, **kwargs)
if source == "paid":
self.record_usage(tool_name, price)
return result
return wrapper
if tool_func is not None:
return decorator(tool_func)
return decorator
class PaymentRequiredError(Exception):
"""Raised when a tool requires payment before execution.
The caller should display the invoice to the user for payment.
"""
def __init__(
self,
tool: str = "",
price_usd: float = 0.01,
pay_to: str = "",
chain: str = "base",
invoice_id: str = "",
):
self.tool = tool
self.price_usd = price_usd
self.pay_to = pay_to
self.chain = chain
self.invoice_id = invoice_id
super().__init__(
f"Payment required: ${price_usd} for {tool}. "
f"Send USDC on {chain} to {pay_to[:16]}... (invoice: {invoice_id[:16]}...)"
)
|