feat(python): add FastAPI and Flask x402 decorator

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(

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"

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"]

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

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)