forked from coinbase/x402
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathmain.py
More file actions
119 lines (94 loc) · 3.41 KB
/
main.py
File metadata and controls
119 lines (94 loc) · 3.41 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
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, payment_middleware # noqa: F401
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")
# Response schemas
class WeatherReport(BaseModel):
weather: str
temperature: int
class WeatherResponse(BaseModel):
report: WeatherReport
class PremiumContentResponse(BaseModel):
content: str
# App
app = FastAPI()
# x402 Middleware
facilitator = HTTPFacilitatorClient(FacilitatorConfig(url=FACILITATOR_URL))
server = x402ResourceServer(facilitator)
server.register(EVM_NETWORK, ExactEvmServerScheme())
server.register(SVM_NETWORK, ExactSvmServerScheme())
routes = {
# a first way to pay for the weather report, it's a string
"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",
),
# a second way to pay for the premium content, it's an AssetAmount object
"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",
),
}
app.add_middleware(PaymentMiddlewareASGI, routes=routes, server=server)
# app.middleware("http")(payment_middleware(routes=routes, server=server)) # Alternative way to register the middleware
# Routes
@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)