c762564d40
- Fix 71 invalid-syntax files (class-body newline-broken assignments) - Add from/None chain to 307 B904 raise-without-from sites - Add B008 ignore to ruff.toml (already in pyproject.toml) - Noqa F401 on __init__.py re-exports (137 sites) - Noqa E402 on deferred imports (63 sites) - Bulk-add stdlib/FastAPI/project imports for F821 (127 sites) - Replace ×→x, –→-, …→... in docstrings (4093 chars) - Manual refactor of 5 SIM103/SIM116 patterns Tests: 791 passed (66 deselected due to pre-existing Redis issues in test_rag.py) Co-authored-by: opencode <opencode@rugmunch.io>
103 lines
3.1 KiB
Python
103 lines
3.1 KiB
Python
"""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 collections.abc import Callable
|
|
|
|
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: Callable | None = 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]}...)"
|
|
)
|