From e117c696d7aacba429f6c336b2f70cbc46af06a6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=BF=9F=E9=BE=99?= Date: Wed, 22 Jul 2026 11:15:56 +0800 Subject: [PATCH] feat(python): add x402 route decorator --- packages/sdk-python/README.md | 36 ++++++ packages/sdk-python/pyproject.toml | 16 +++ packages/sdk-python/src/nirium/__init__.py | 3 +- packages/sdk-python/src/nirium/x402.py | 142 +++++++++++++++++++++ packages/sdk-python/tests/test_x402.py | 107 ++++++++++++++++ 5 files changed, 303 insertions(+), 1 deletion(-) create mode 100644 packages/sdk-python/src/nirium/x402.py create mode 100644 packages/sdk-python/tests/test_x402.py diff --git a/packages/sdk-python/README.md b/packages/sdk-python/README.md index aa1cda3..9fcf849 100644 --- a/packages/sdk-python/README.md +++ b/packages/sdk-python/README.md @@ -71,6 +71,42 @@ agent.init_x402( response = await agent.x402_fetch("https://nirium-agent.fly.dev/api/v1/premium/signals") ``` +### Protect FastAPI or Flask routes + +Set the Stellar settlement destination outside application source, then use the +same decorator with either framework: + +```bash +export NIRIUM_X402_PAY_TO="G_YOUR_STELLAR_ADDRESS" +``` + +```python +from fastapi import FastAPI, Request +from nirium import x402_required + +app = FastAPI() + +@app.get("/premium") +@x402_required(price="0.02") +async def premium(request: Request): + return {"report": "paid content"} +``` + +```python +from flask import Flask +from nirium import x402_required + +app = Flask(__name__) + +@app.get("/premium") +@x402_required(price="0.02") +def premium(): + return {"report": "paid content"} +``` + +Missing or invalid `X-402-Signature` headers receive HTTP 402. The decorated +handler runs only after the facilitator verifies and settles the payment. + ### MPP — Session-Based Budget Delegation ```python agent.init_mpp( diff --git a/packages/sdk-python/pyproject.toml b/packages/sdk-python/pyproject.toml index 35fdcd8..1f73f70 100644 --- a/packages/sdk-python/pyproject.toml +++ b/packages/sdk-python/pyproject.toml @@ -17,6 +17,22 @@ dependencies = [ "stellar-sdk>=11.0.0", ] +[project.optional-dependencies] +frameworks = [ + "fastapi>=0.110.0", + "flask>=3.0.0", +] +test = [ + "fastapi>=0.110.0", + "flask[async]>=3.0.0", + "httpx>=0.27.0", + "pytest>=8.0.0", +] + +[tool.pytest.ini_options] +pythonpath = ["src"] +testpaths = ["tests"] + [project.urls] Homepage = "https://nirium.xyz" Repository = "https://github.com/nirium-protocol/nirium-sdk" diff --git a/packages/sdk-python/src/nirium/__init__.py b/packages/sdk-python/src/nirium/__init__.py index a3ddb2a..908623d 100644 --- a/packages/sdk-python/src/nirium/__init__.py +++ b/packages/sdk-python/src/nirium/__init__.py @@ -1,6 +1,7 @@ """Nirium — Official Python SDK for the Nirium autonomous DeFi agent.""" from .client import Agent # type: ignore +from .x402 import FacilitatorVerifier, PaymentVerifier, x402_required __version__ = "0.6.1" -__all__ = ["Agent"] +__all__ = ["Agent", "FacilitatorVerifier", "PaymentVerifier", "x402_required"] diff --git a/packages/sdk-python/src/nirium/x402.py b/packages/sdk-python/src/nirium/x402.py new file mode 100644 index 0000000..be81cd5 --- /dev/null +++ b/packages/sdk-python/src/nirium/x402.py @@ -0,0 +1,142 @@ +"""Framework adapters for protecting Python routes with x402 payments.""" + +from __future__ import annotations + +import asyncio +import inspect +import os +from functools import wraps +from typing import Any, Callable, Mapping, Optional, Protocol, TypeVar, cast + +import aiohttp + +F = TypeVar("F", bound=Callable[..., Any]) + + +class PaymentVerifier(Protocol): + """A verifier that confirms and settles an x402 payment signature.""" + + async def verify_and_settle(self, signature: str, *, price: str, pay_to: str) -> bool: + """Return true only after the facilitator verifies and settles payment.""" + + +class FacilitatorVerifier: + """Verify and settle x402 signatures through an HTTP facilitator.""" + + def __init__(self, url: str = "https://facilitator.x402.org") -> None: + self.url = url.rstrip("/") + + async def verify_and_settle(self, signature: str, *, price: str, pay_to: str) -> bool: + payload = { + "paymentPayload": signature, + "paymentRequirements": { + "scheme": "exact", + "price": price, + "network": "stellar:testnet", + "payTo": pay_to, + }, + } + async with aiohttp.ClientSession() as session: + async with session.post(f"{self.url}/verify", json=payload) as response: + if response.status != 200 or not (await response.json()).get("isValid", False): + return False + async with session.post(f"{self.url}/settle", json=payload) as response: + if response.status != 200: + return False + result = await response.json() + return bool(result.get("success") or result.get("transaction")) + + +def _request_from_call(args: tuple[Any, ...], kwargs: Mapping[str, Any]) -> Any: + request = kwargs.get("request") + if request is not None: + return request + for value in args: + if hasattr(value, "headers"): + return value + try: + from flask import request as flask_request + + return flask_request + except (ImportError, RuntimeError): + return None + + +def _signature(request: Any) -> Optional[str]: + if request is None: + return None + headers = getattr(request, "headers", {}) + return headers.get("X-402-Signature") or headers.get("x-402-signature") + + +def _payment_required(request: Any) -> Any: + payload = {"error": "Payment Required"} + module = request.__class__.__module__ if request is not None else "" + if module.startswith(("fastapi", "starlette")): + from starlette.responses import JSONResponse + + return JSONResponse(payload, status_code=402) + try: + from flask import jsonify + + return jsonify(payload), 402 + except (ImportError, RuntimeError): + return payload, 402 + + +def x402_required( + *, + price: str, + pay_to: Optional[str] = None, + verifier: Optional[PaymentVerifier] = None, +) -> Callable[[F], F]: + """Require a settled x402 payment before running a FastAPI or Flask route. + + ``pay_to`` defaults to ``NIRIUM_X402_PAY_TO`` so the concise + ``@x402_required(price="0.02")`` form can keep payout configuration out of + application source code. + """ + + try: + numeric_price = float(price) + except (TypeError, ValueError) as error: + raise ValueError("price must be a positive decimal string") from error + if numeric_price <= 0: + raise ValueError("price must be a positive decimal string") + + settlement_address = pay_to or os.getenv("NIRIUM_X402_PAY_TO", "") + payment_verifier = verifier or FacilitatorVerifier() + + def decorator(func: F) -> F: + async def authorize(args: tuple[Any, ...], kwargs: Mapping[str, Any]) -> tuple[bool, Any]: + request = _request_from_call(args, kwargs) + signature = _signature(request) + if not signature or not settlement_address: + return False, request + valid = await payment_verifier.verify_and_settle( + signature, + price=price, + pay_to=settlement_address, + ) + return valid, request + + if inspect.iscoroutinefunction(func): + @wraps(func) + async def async_wrapper(*args: Any, **kwargs: Any) -> Any: + valid, request = await authorize(args, kwargs) + if not valid: + return _payment_required(request) + return await func(*args, **kwargs) + + return cast(F, async_wrapper) + + @wraps(func) + def sync_wrapper(*args: Any, **kwargs: Any) -> Any: + valid, request = asyncio.run(authorize(args, kwargs)) + if not valid: + return _payment_required(request) + return func(*args, **kwargs) + + return cast(F, sync_wrapper) + + return decorator diff --git a/packages/sdk-python/tests/test_x402.py b/packages/sdk-python/tests/test_x402.py new file mode 100644 index 0000000..daa8061 --- /dev/null +++ b/packages/sdk-python/tests/test_x402.py @@ -0,0 +1,107 @@ +import asyncio +from dataclasses import dataclass + +import pytest +from fastapi import FastAPI, Request as FastAPIRequest +from fastapi.testclient import TestClient +from flask import Flask + +from nirium import x402_required + + +@dataclass +class Request: + headers: dict[str, str] + + +class Verifier: + def __init__(self, result: bool) -> None: + self.result = result + self.calls: list[tuple[str, str, str]] = [] + + async def verify_and_settle(self, signature: str, *, price: str, pay_to: str) -> bool: + self.calls.append((signature, price, pay_to)) + return self.result + + +def test_missing_signature_returns_402() -> None: + verifier = Verifier(True) + + @x402_required(price="0.02", pay_to="GDESTINATION", verifier=verifier) + async def endpoint(request: Request) -> dict[str, bool]: + return {"ok": True} + + response = asyncio.run(endpoint(Request(headers={}))) + assert response == ({"error": "Payment Required"}, 402) + assert verifier.calls == [] + + +def test_invalid_signature_returns_402() -> None: + verifier = Verifier(False) + + @x402_required(price="0.02", pay_to="GDESTINATION", verifier=verifier) + async def endpoint(request: Request) -> dict[str, bool]: + return {"ok": True} + + response = asyncio.run(endpoint(Request(headers={"X-402-Signature": "invalid"}))) + assert response == ({"error": "Payment Required"}, 402) + assert verifier.calls == [("invalid", "0.02", "GDESTINATION")] + + +def test_settled_signature_runs_fastapi_style_async_endpoint() -> None: + verifier = Verifier(True) + + @x402_required(price="0.02", pay_to="GDESTINATION", verifier=verifier) + async def endpoint(request: Request) -> dict[str, bool]: + return {"ok": True} + + response = asyncio.run(endpoint(request=Request(headers={"x-402-signature": "signed"}))) + assert response == {"ok": True} + + +def test_settled_signature_runs_flask_style_sync_endpoint() -> None: + verifier = Verifier(True) + + @x402_required(price="0.02", pay_to="GDESTINATION", verifier=verifier) + def endpoint(request: Request) -> dict[str, bool]: + return {"ok": True} + + assert endpoint(Request(headers={"X-402-Signature": "signed"})) == {"ok": True} + + +def test_fastapi_integration_returns_402_then_allows_settled_payment() -> None: + verifier = Verifier(True) + app = FastAPI() + + @app.get("/premium") + @x402_required(price="0.02", pay_to="GDESTINATION", verifier=verifier) + async def premium(request: FastAPIRequest) -> dict[str, bool]: + return {"ok": True} + + client = TestClient(app) + assert client.get("/premium").status_code == 402 + paid = client.get("/premium", headers={"X-402-Signature": "signed"}) + assert paid.status_code == 200 + assert paid.json() == {"ok": True} + + +def test_flask_integration_returns_402_then_allows_settled_payment() -> None: + verifier = Verifier(True) + app = Flask(__name__) + + @app.get("/premium") + @x402_required(price="0.02", pay_to="GDESTINATION", verifier=verifier) + def premium() -> dict[str, bool]: + return {"ok": True} + + client = app.test_client() + assert client.get("/premium").status_code == 402 + paid = client.get("/premium", headers={"X-402-Signature": "signed"}) + assert paid.status_code == 200 + assert paid.get_json() == {"ok": True} + + +@pytest.mark.parametrize("price", ["0", "-1", "not-a-number"]) +def test_invalid_price_is_rejected(price: str) -> None: + with pytest.raises(ValueError): + x402_required(price=price)