forked from coinbase/x402
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathpaywall.py
More file actions
129 lines (102 loc) · 3.48 KB
/
paywall.py
File metadata and controls
129 lines (102 loc) · 3.48 KB
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
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
"""Paywall example with browser-based payment UI."""
import os
from dotenv import load_dotenv
from fastapi import FastAPI
from pydantic import BaseModel
from x402.http import FacilitatorConfig, HTTPFacilitatorClient, PaymentOption
from x402.http.middleware.fastapi import PaymentMiddlewareASGI
from x402.http.paywall import create_paywall, evm_paywall, svm_paywall
from x402.http.types import RouteConfig
from x402.mechanisms.evm.exact import ExactEvmServerScheme
from x402.mechanisms.svm.exact import ExactSvmServerScheme
from x402.schemas import AssetAmount, Network
from x402.server import x402ResourceServer
load_dotenv()
# Config
EVM_ADDRESS = os.getenv("EVM_ADDRESS")
SVM_ADDRESS = os.getenv("SVM_ADDRESS")
EVM_NETWORK: Network = "eip155:84532" # Base Sepolia
SVM_NETWORK: Network = "solana:EtWTRABZaYq6iMfeYKouRu166VU2xqa1" # Solana Devnet
FACILITATOR_URL = os.getenv("FACILITATOR_URL", "https://x402.org/facilitator")
if not EVM_ADDRESS or not SVM_ADDRESS:
raise ValueError("Missing required environment variables (EVM_ADDRESS, SVM_ADDRESS)")
class WeatherReport(BaseModel):
weather: str
temperature: int
class WeatherResponse(BaseModel):
report: WeatherReport
class PremiumContentResponse(BaseModel):
content: str
app = FastAPI()
facilitator = HTTPFacilitatorClient(FacilitatorConfig(url=FACILITATOR_URL))
server = x402ResourceServer(facilitator)
server.register(EVM_NETWORK, ExactEvmServerScheme())
server.register(SVM_NETWORK, ExactSvmServerScheme())
routes = {
"GET /weather": RouteConfig(
accepts=[
PaymentOption(
scheme="exact",
pay_to=EVM_ADDRESS,
price="$0.01",
network=EVM_NETWORK,
),
PaymentOption(
scheme="exact",
pay_to=SVM_ADDRESS,
price="$0.01",
network=SVM_NETWORK,
),
],
mime_type="application/json",
description="Weather report",
),
"GET /premium/*": RouteConfig(
accepts=[
PaymentOption(
scheme="exact",
pay_to=EVM_ADDRESS,
price=AssetAmount(
amount="10000", # $0.01 USDC
asset="0x036CbD53842c5426634e7929541eC2318f3dCF7e",
extra={"name": "USDC", "version": "2"},
),
network=EVM_NETWORK,
),
PaymentOption(
scheme="exact",
pay_to=SVM_ADDRESS,
price="$0.01",
network=SVM_NETWORK,
),
],
mime_type="application/json",
description="Premium content",
),
}
# Paywall provider for browser-based payment UI
paywall = (
create_paywall()
.with_network(evm_paywall)
.with_network(svm_paywall)
.with_config(app_name="x402 Paywall Demo", testnet=True)
.build()
)
app.add_middleware(
PaymentMiddlewareASGI,
routes=routes,
server=server,
paywall_provider=paywall,
)
@app.get("/health")
async def health_check() -> dict[str, str]:
return {"status": "ok"}
@app.get("/weather")
async def get_weather() -> WeatherResponse:
return WeatherResponse(report=WeatherReport(weather="sunny", temperature=70))
@app.get("/premium/content")
async def get_premium_content() -> PremiumContentResponse:
return PremiumContentResponse(content="This is premium content")
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=4021)