API 参考
x402-mock 模块的完整 API 文档。
Servers
HTTP 402 支付协议服务端实现
Servers 模块提供了一个基于 FastAPI 的服务器框架,用于实现 HTTP 402 Payment Required 协议。它采用事件驱动架构,封装了所有支付收款逻辑,使支付接收方能够以最少的配置集成 web3 支付接受功能。
主要特性:
- FastAPI 集成:扩展的 FastAPI 应用程序,内置支付端点路由
- 令牌管理:安全的 HMAC 签名访问令牌生成和验证
- 事件驱动架构:订阅支付生命周期事件(请求、验证、结算)
- 多链支持:在不同区块链网络上注册多种支付方式
- 自动结算:验证成功后可选自动链上结算
- 安全工具:私钥生成、令牌签名和环境密钥管理
- 现代 EVM 签名:
- USDC:ERC-3009 (transferWithAuthorization)
- 通用 ERC20:Permit2 (permitTransferFrom)
主要组件:
- Http402Server:扩展 FastAPI 并支持支付协议的主要服务器类
- Api key相关工具:generate_token()、verify_token()、create_private_key()、save_key_to_env()
servers
Http402Server
Bases: FastAPI
FastAPI server with X402 payment protocol support.
__init__(token_key: str, adapter_hub: Optional[AdapterHub] = None, token_expires_in: int = 3600, enable_auto_settlement: bool = True, token_endpoint: str = '/token', **fastapi_kwargs)
Initialize X402 payment server.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
token_key
|
str
|
Secret key for signing access tokens |
required |
adapter_hub
|
Optional[AdapterHub]
|
Payment adapter hub (default: new instance) |
None
|
token_expires_in
|
int
|
Token lifetime in seconds (default: 3600) |
3600
|
enable_auto_settlement
|
bool
|
Auto-settle after verification (default: True) |
True
|
token_endpoint
|
str
|
Token endpoint path (default: /token) |
'/token'
|
**fastapi_kwargs
|
FastAPI arguments (title, version, etc.) |
{}
|
add_payment_method(payment_component: Union[PaymentComponentTypes, Dict[str, Any]]) -> None
Register a payment method.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
payment_component
|
Union[PaymentComponentTypes, Dict[str, Any]]
|
A |
required |
subscribe(event_class: type[BaseEvent], handler: Callable) -> None
Register event handler.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
event_class
|
type[BaseEvent]
|
Event type to handle |
required |
handler
|
Callable
|
Async function(event, deps) -> Optional[BaseEvent] |
required |
Example
async def my_handler(event: TokenIssuedEvent, deps: Dependencies):
# Custom logic
return None # or return another event
app.subscribe(TokenIssuedEvent, my_handler)
add_hook(event_class: type[BaseEvent], hook: Callable) -> None
Register event hook for side effects.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
event_class
|
type[BaseEvent]
|
Event type to hook into |
required |
hook
|
Callable
|
Async function(event, deps) -> None |
required |
Example
async def log_event(event, deps):
print(f"Event: {event}")
app.add_hook(TokenIssuedEvent, log_event)
hook(event_class: type[BaseEvent]) -> Callable
Decorator for registering event hooks.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
event_class
|
type[BaseEvent]
|
Event type to hook into |
required |
Example
@app.hook(TokenIssuedEvent) async def on_token_issued(event, deps): await send_analytics(event)
payment_required(route_handler)
Decorator to protect routes with payment verification.
Returns 402 response if payment required, otherwise executes handler with verified payload.
Example
@app.payment_required
@app.get("/data")
async def get_data(payload):
return {"user": payload["address"]}
generate_token(*, private_key: str, expires_in: int = 3600, nonce_length: int = 16) -> str
Generate a signed token.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
private_key
|
str
|
Secret key used to sign the token. |
required |
expires_in
|
int
|
Token lifetime in seconds. |
3600
|
nonce_length
|
int
|
Length of random nonce. |
16
|
Returns:
| Type | Description |
|---|---|
str
|
Signed token string. |
verify_token(*, token: str, private_key: str, leeway: int = 0) -> Dict[str, object]
Verify token signature and expiration.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
token
|
str
|
Token string. |
required |
private_key
|
str
|
Secret key used to verify the token. |
required |
leeway
|
int
|
Allowed clock skew in seconds. |
0
|
Returns:
| Type | Description |
|---|---|
Dict[str, object]
|
Decoded payload if valid. |
Raises:
| Type | Description |
|---|---|
TokenExpired
|
If token is expired. |
TokenInvalid
|
If token is malformed or signature mismatch. |
create_private_key(*, prefix: str = '', length: int = 32, use_special_chars: bool = False) -> str
Generate a private key with optional prefix and randomization rules.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
prefix
|
str
|
A custom string to prepend to the random key (semi-automatic rule). |
''
|
length
|
int
|
The number of random characters to generate. |
32
|
use_special_chars
|
bool
|
Whether to include special characters in the random part. |
False
|
Returns:
| Type | Description |
|---|---|
str
|
A secure private key string. |
save_key_to_env(key_name: str, key_value: str, env_file: str = '.env')
Save or update a key-value pair in a .env file.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
key_name
|
str
|
The environment variable name (e.g., "PRIVATE_KEY"). |
required |
key_value
|
str
|
The actual key string to save. |
required |
env_file
|
str
|
Path to the .env file. Defaults to ".env". |
'.env'
|
Clients
HTTP 402 支付客户端中间件
Clients 模块提供了一个智能 HTTP 客户端,能够透明地处理 HTTP 402 Payment Required 响应。它扩展了 httpx.AsyncClient,自动拦截支付挑战、生成签名的支付许可、将其交换为访问令牌并重试原始请求——所有这些都不需要用户显式干预。
主要特性: - 透明支付处理:自动处理 402 响应,无需手动干预 - httpx 兼容性:完全兼容的 httpx.AsyncClient 直接替代品 - 离线签名自动签名:使用注册的支付方法生成链/代币特定的离线授权(ERC-3009 / Permit2) - 令牌交换:自动在服务器端点将许可交换为访问令牌 - 请求重试:无缝重试获得授权后的原始请求 - 多链支持:在不同区块链网络上注册支付能力
主要组件:
- Http402Client:具有自动支付流程处理功能的扩展异步 HTTP 客户端
使用模式: 1. 初始化客户端并注册支付方法 2. 向受保护资源发出标准 HTTP 请求 3. 客户端自动处理 402 挑战并获取访问权限 4. 透明地接收成功响应
clients
Client module for x402 payment authorization.
Provides easy-to-use interfaces for accessing protected resources with automatic permit signing and token exchange.
Http402Client
Bases: AsyncClient
Extended httpx.AsyncClient with automatic 402 payment handling.
This client extends httpx.AsyncClient and automatically handles 402 Payment Required status codes by: 1. Parsing payment requirements 2. Generating signed permits 3. Exchanging permits for access tokens 4. Retrying the original request with authorization
Fully compatible with httpx.AsyncClient - supports all methods, properties, and can be used as an async context manager.
Usage
async with Http402Client() as client:
# Register a payment method (PaymentComponentTypes or dict)
payment_component = {
"payment_type": "eip155:11155111",
"amount": 100.0,
"token": "USDC"
}
client.add_payment_method(payment_component)
response = await client.get("https://api.example.com/data")
__init__(adapter_hub: Optional[AdapterHub] = None, **kwargs)
Initialize client with optional payment adapter.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
adapter_hub
|
Optional[AdapterHub]
|
Optional AdapterHub for payment handling |
None
|
**kwargs
|
All standard httpx.AsyncClient arguments (timeout, headers, etc.) |
{}
|
add_payment_method(payment_component: Union[PaymentComponentTypes, Dict[str, Any]]) -> None
Register local payment capability.
This enables the middleware to automatically generate payment permits when encountering 402 responses.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
payment_component
|
Union[PaymentComponentTypes, Dict[str, Any]]
|
A |
required |
request(method: str, url: httpx._types.URLTypes, **kwargs) -> httpx.Response
async
Execute HTTP request with automatic 402 handling.
Overrides httpx.AsyncClient.request() to intercept 402 responses. All other httpx methods (get, post, etc.) automatically use this.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
method
|
str
|
HTTP method (GET, POST, etc.) |
required |
url
|
URLTypes
|
Request URL |
required |
**kwargs
|
All standard httpx arguments |
{}
|
Returns:
| Type | Description |
|---|---|
Response
|
httpx.Response object |
Adapters
统一区块链适配器接口
Adapters 模块提供了一个统一的抽象层,弥合了各种区块链平台(EVM、Solana 等)之间的差异。它实现了基于插件的架构,具有自动区块链类型检测功能,能够在异构区块链生态系统中实现一致的支付许可签名、签名验证和链上结算操作。
主要特性: - 区块链抽象:EVM、SVM(Solana)和其他区块链平台的统一接口 - 自动类型检测:从链标识符(CAIP-2 格式)识别区块链类型 - 签名操作:生成和验证区块链特定的加密签名 - 授权验证:验证授权真实性、过期时间、随机数和链上条件 - 交易结算:执行链上转账并跟踪确认(EVM 上的 ERC-3009 / Permit2) - 余额查询:查询不同链上的代币余额和授权额度 - 可扩展架构:工厂模式便于轻松添加新的区块链适配器
主要组件:
- AdapterHub:将操作路由到适当区块链适配器的中央网关
- AdapterFactory:定义适配器接口契约的抽象基类
- PaymentRegistry:管理支付方法注册和检索
- 平台特定适配器:EVMAdapter(以太坊/EVM 链)、SVM 适配器(即将推出)
架构模式: 使用适配器模式结合工厂模式,提供一致的 API,同时在底层委托给区块链特定的实现。
adapters
AdapterHub
Unified Blockchain Adapter Hub.
Provides core blockchain adapter operations with automatic type detection and routing. Manages payment component registration and delegates to blockchain-specific adapters.
__init__(evm_private_key: str = None, request_timeout: int = 60)
Initialize the hub with a payment registry and chain-specific adapter instances.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
evm_private_key
|
str
|
Private key used by the EVM adapter for on-chain operations. |
None
|
request_timeout
|
int
|
HTTP request timeout (seconds) forwarded to each adapter. |
60
|
register_payment_methods(payment_component: Union[PaymentComponentTypes, Dict[str, Any]], client_role: bool = False) -> None
Register a payment component into the hub under the given role.
Server role (client_role=False, default): if pay_to is not set on
the component, it is automatically filled with the wallet address returned by
the matching chain adapter. Use this when the hub acts as the receiving party.
Client role (client_role=True): pay_to is left untouched. Use this
when the hub acts as the signing/paying party and the recipient address will
come from the remote server's payment requirements.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
payment_component
|
Union[PaymentComponentTypes, Dict[str, Any]]
|
A |
required |
client_role
|
bool
|
|
False
|
Raises:
| Type | Description |
|---|---|
TypeError
|
If the blockchain type cannot be determined from the component, or if no adapter is registered for that type. |
ValueError
|
If the component cannot be parsed or fails chain validation. |
get_payment_methods() -> List[PaymentComponentTypes]
Get all registered payment methods.
Returns:
| Type | Description |
|---|---|
List[PaymentComponentTypes]
|
List of registered payment components |
initialize(client_role: bool = False) -> None
async
One-time startup initialisation gated by caller role.
Must be called once before signature() when operating as the signing
party (client). For each registered adapter, the chain-specific
client_init() hook is invoked so it can ensure any required
on-chain state is in place (e.g. Permit2 ERC-20 allowances for EVM).
Server-side roles (verify_signature / settle) do not require this call
and may pass client_role=False to skip the adapter hooks while still
marking the hub as initialised for completeness.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
client_role
|
bool
|
|
False
|
Raises:
| Type | Description |
|---|---|
RuntimeError
|
If any adapter's client_init() fails. |
verify_signature(permit_payload: Union[PermitTypes, Dict[str, Any]]) -> Optional[Any]
async
Verify permit signature with automatic blockchain detection and component matching.
Converts permit payload to typed model, matches token with registered components, and calls corresponding adapter to verify signature.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
permit_payload
|
Union[PermitTypes, Dict[str, Any]]
|
Permit data (PermitTypes or dict) |
required |
Returns:
| Type | Description |
|---|---|
Optional[Any]
|
Verification result from adapter |
Raises:
| Type | Description |
|---|---|
TypeError
|
If blockchain type cannot be determined |
ValueError
|
If payload conversion or token matching fails |
settle(permit_payload: Union[PermitTypes, Dict[str, Any]]) -> Optional[Any]
async
Execute permit settlement with automatic type conversion.
Converts permit payload to typed model and calls corresponding adapter.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
permit_payload
|
Union[PermitTypes, Dict[str, Any]]
|
Permit data (PermitTypes or dict) |
required |
Returns:
| Type | Description |
|---|---|
Optional[Any]
|
Transaction confirmation from server adapter |
Raises:
| Type | Description |
|---|---|
TypeError
|
If blockchain type cannot be determined |
ValueError
|
If payload conversion fails |
signature(list_components: List[Union[PaymentComponentTypes, Dict[str, Any]]]) -> PermitTypes
async
Generate signed permit from remote payment components.
Matches remote components against local support list, converts to typed model, and delegates to blockchain-specific adapter for signing.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
list_components
|
List[Union[PaymentComponentTypes, Dict[str, Any]]]
|
Remote payment components (typed or dict) to match against locally registered ones. |
required |
Returns:
| Type | Description |
|---|---|
PermitTypes
|
Signed permit produced by the matching chain adapter. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If no matching component is found or type conversion fails. |
TypeError
|
If the blockchain type cannot be determined. |
PaymentRegistry
Central registry that aggregates payment components across all supported blockchain types.
Routes each submitted payment component to the appropriate chain-specific registry (e.g. EVM, and future chains such as Solana) and maintains a unified support list.
method_register(payment_component: Union[PaymentComponentTypes, Dict[str, Any]]) -> None
Register a payment component into the appropriate chain-specific registry.
Accepts either a validated payment component instance or a raw dict that will be coerced into the correct type via Pydantic's discriminated union.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
payment_component
|
Union[PaymentComponentTypes, Dict[str, Any]]
|
A |
required |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the dict cannot be parsed into a known payment component type, or if the component fails chain-specific validation. |
get_support_list() -> List[PaymentComponentTypes]
Return all registered payment components across all supported chain types.
Returns:
| Type | Description |
|---|---|
List[PaymentComponentTypes]
|
An ordered list of every payment component that has been successfully registered. |
get_adapter_type(obj: Union[PermitTypes, PaymentComponentTypes, SignatureTypes, VerificationResultTypes, TransactionConfirmationTypes, BasePermit, BasePaymentComponent, BaseVerificationResult, BaseTransactionConfirmation]) -> Optional[str]
Retrieve the unified adapter type identifier for a given permit, payment component, signature, verification result, or transaction confirmation.
Automatically extracts the type discriminator field (permit_type, payment_type, signature_type, verification_type, or confirmation_type) from the object and returns the standardized adapter type string that can be used as a key in AdapterHub._adapter_factories.
This function maps all blockchain-specific discriminator values to their unified adapter type identifiers (e.g., "EIP2612" -> "evm", "polygon" -> "evm", "spl" -> "svm").
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
obj
|
Union[PermitTypes, PaymentComponentTypes, SignatureTypes, VerificationResultTypes, TransactionConfirmationTypes, BasePermit, BasePaymentComponent, BaseVerificationResult, BaseTransactionConfirmation]
|
A permit, payment component, signature, verification result, or transaction confirmation instance containing a type discriminator field. Supports both typed instances and base classes. |
required |
Returns:
| Type | Description |
|---|---|
Optional[str]
|
The unified adapter type string (e.g., "evm", "svm") that corresponds to |
Optional[str]
|
the object's blockchain type. Returns None if no matching adapter type is found. |
Examples:
>>> # EIP-2612 permit maps to "evm"
>>> permit = EIP2612Permit(permit_type="EIP2612", owner="0x...", ...)
>>> adapter_type = get_adapter_type(permit) # Returns "evm"
>>> adapter = hub._adapter_factories[adapter_type]
>>> # EVM payment component maps to "evm"
>>> component = EVMPaymentComponent(payment_type="evm", token="0x...", ...)
>>> adapter_type = get_adapter_type(component) # Returns "evm"
>>> adapter = hub._adapter_factories[adapter_type]
>>> # EVM verification result maps to "evm"
>>> result = EVMVerificationResult(verification_type="evm", ...)
>>> adapter_type = get_adapter_type(result) # Returns "evm"
>>> # Polygon payment also maps to "evm"
>>> component = EVMPaymentComponent(payment_type="polygon", ...)
>>> adapter_type = get_adapter_type(component) # Returns "evm"
Note
The type mapping is lazily initialized on first call to avoid circular imports. To add support for new blockchains, update _initialize_adapter_type_mapping() with new type mappings in ADAPTER_TYPE_MAPPING.
AdapterFactory
Bases: ABC
Abstract Base Class for Server-Side Blockchain Adapters.
Defines the interface for server-side operations that interact directly with blockchain nodes. Server-side adapters are responsible for: - Verifying permit signatures and permit validity on-chain - Executing transactions (settle/send_transaction) on-chain - Querying token balances and other on-chain state
Server adapters use the application's private key to sign and send transactions. They act as the bridge between the x402 system and the blockchain.
Key Responsibilities: 1. verify_signature: Validate permit signature and check permit conditions on-chain 2. settle: Execute the permit transaction on-chain and return confirmation 3. get_balance: Query token balance of an address on-chain 4. signature: Support signing operations for blockchain-specific formats
Example Implementation
class EVMServerAdapter(AdapterServerFactory): # EVM/Ethereum-specific implementation pass
class SolanaServerAdapter(AdapterServerFactory): # Solana-specific implementation pass
verify_signature(permit: BasePermit, payment_requirement: BasePaymentComponent) -> BaseVerificationResult
abstractmethod
async
Verify permit signature validity and check permit conditions on-chain.
This is the critical security operation that ensures: 1. The permit signature is cryptographically valid 2. The recovered signer matches the permit owner 3. The permit has not expired 4. The nonce prevents replay attacks 5. The owner has sufficient balance and allowance 6. The payment amount matches or exceeds requirements
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
permit
|
BasePermit
|
BasePermit instance containing permit data and signature |
required |
payment_requirement
|
BasePaymentComponent
|
BasePaymentComponent specifying expected payment amount/conditions |
required |
Returns:
| Name | Type | Description |
|---|---|---|
BaseVerificationResult |
BaseVerificationResult
|
Detailed verification result including: - status: Verification result status (SUCCESS, INVALID_SIGNATURE, EXPIRED, etc.) - is_valid: Boolean indicating if permit is valid - permit_owner: Verified owner address - authorized_amount: Verified authorized amount - message: Human-readable status message - blockchain_state: Optional on-chain state data |
Implementation Notes
- Must recover the signer from signature
- Must verify signature against permit data hash
- Must check permit.is_expired()
- Must query on-chain state (nonce, allowance, balance)
- Must validate payment amount meets requirements
Example
result = await adapter.verify_signature(permit, payment_req) if result.is_success(): # Permit is valid, proceed to settlement await adapter.settle(permit) else: # Return verification error to client error_msg = result.get_error_message()
settle(permit: BasePermit) -> BaseTransactionConfirmation
abstractmethod
async
Execute permit transaction on-chain (settlement/transaction execution).
This method actually executes the token transfer on-chain using the permit signature. It should only be called after verify_signature has confirmed the permit is valid.
Steps: 1. Construct permit() call with signature components (v, r, s) 2. Build the complete transaction (gas estimation, nonce, etc.) 3. Sign transaction with server's private key 4. Broadcast transaction to blockchain 5. Wait for transaction confirmation 6. Return transaction confirmation with hash and receipt data
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
permit
|
BasePermit
|
BasePermit instance with valid signature (assumed verified) |
required |
Returns:
| Name | Type | Description |
|---|---|---|
BaseTransactionConfirmation |
BaseTransactionConfirmation
|
Transaction execution result including: - status: Transaction status (SUCCESS, FAILED, PENDING, etc.) - tx_hash: Transaction hash on-chain - block_number: Block number containing transaction - block_timestamp: Block timestamp - gas_used: Actual gas consumed - confirmations: Number of block confirmations - error_message: Error details if transaction failed |
Implementation Notes
- Estimate gas before sending
- Sign transaction with server private key
- Handle blockchain-specific transaction formats
- Wait for configurable number of confirmations
- Handle network errors gracefully
Example
result = await adapter.settle(permit) if result.is_success(): print(f"Settlement complete: {result.tx_hash}") # Update database with tx_hash else: print(f"Settlement failed: {result.error_message}")
get_balance(address: str) -> int
abstractmethod
async
Query token balance for an address on the blockchain.
Retrieves the current balance of the configured token (typically USDC) for the given address. This is used for: - Verification: Check owner has sufficient balance - Queries: Allow clients to check addresses' balances
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
address
|
str
|
Wallet address to query (blockchain format, e.g., "0x..." for EVM) |
required |
Returns:
| Name | Type | Description |
|---|---|---|
int |
int
|
Token balance in smallest units (e.g., wei for EVM where 1 USDC = 1e6) Returns 0 if address has no balance |
Implementation Notes
- Use blockchain node RPC call (balanceOf for ERC20)
- Handle address validation/checksum
- Cache results if possible for performance
- Handle blockchain-specific address formats
Example
balance = await adapter.get_balance("0x1234...5678")
balance = 1000000 (representing 1 USDC with 6 decimals)
signature(payment_component: BasePaymentComponent) -> BasePermit
abstractmethod
async
Generate complete signed permit from payment component.
This method builds the permit message, signs it with user's private key/wallet, and returns a fully signed and ready-to-submit permit object. All permit parameters are derived from the payment_component.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
payment_component
|
BasePaymentComponent
|
BasePaymentComponent specifying payment requirements, blockchain type, and all permit parameters |
required |
Returns:
| Name | Type | Description |
|---|---|---|
BasePermit |
BasePermit
|
Fully signed permit ready for server submission. For EVM, this is EIP2612Permit with: - owner: Token owner address - spender: Authorized spender address - token: Token contract address - value: Authorized amount - deadline: Permit expiration timestamp - nonce: Replay attack prevention nonce - signature: EIP2612PermitSignature with v, r, s components - permit_type: Blockchain-specific type (e.g., "EIP2612") |
Implementation Notes
- Extract all permit parameters from payment_component
- Must validate payment_component matches the adapter's blockchain type
- Must build blockchain-specific permit message internally
- Must sign with user's private key (not server's)
- Must validate signature format before returning
- Must return complete permit with signature components
- Should not leak private key
Raises:
| Type | Description |
|---|---|
TypeError
|
If payment_component blockchain type doesn't match adapter |
ValueError
|
If payment_component is invalid or signing fails |
Example
adapter = EVMClientAdapter() payment = EVMPaymentComponent(payment_type="evm", ...) permit = await adapter.signature(payment_component=payment)
permit now contains complete signed data ready to send to server
await server.settle(permit)
client_init(payment_components: List[BasePaymentComponent]) -> None
async
One-time client-side pre-signing initialisation hook.
Called by AdapterHub.initialize(role="client") once at startup, before any signature() calls are made. Concrete adapters should override this to perform chain-specific on-chain setup required by the signing role (e.g. ERC-20 allowance approval for Permit2 on EVM, SPL token delegation on SVM). The default implementation is a no-op so that server-only adapters and future adapters can inherit without modification.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
payment_components
|
List[BasePaymentComponent]
|
All payment components registered via register_payment_methods(). The implementation may filter these down to the subset it manages. |
required |
Raises:
| Type | Description |
|---|---|
RuntimeError
|
If any required on-chain setup fails. |
get_wallet_address() -> str
abstractmethod
Get server wallet address from private key.
Returns:
| Name | Type | Description |
|---|---|---|
str |
str
|
Server wallet address in checksum format |
EVMAdapter
Bases: AdapterFactory
EVM Blockchain Server Adapter Implementation.
Provides complete server-side functionality for EVM blockchains including: - EIP2612 permit signature verification - On-chain state validation (nonce, allowance, balance, expiration) - Token transfer execution via signed permits - Balance and allowance queries
This adapter validates all security constraints before executing any on-chain operations. All methods are designed to return clear, actionable error messages rather than raising exceptions.
Key Design Features: - Dynamic RPC URL selection based on permit's chain_id - Environment-aware infrastructure key handling (evm_infra_key for premium RPC, falls back to public) - Private key loaded from environment (evm_private_key) during initialization - Lazy Web3 instance creation per blockchain interaction (ensures correct RPC endpoint)
Attributes:
| Name | Type | Description |
|---|---|---|
account |
Server's account object (initialized from evm_private_key environment variable) |
|
wallet_address |
Checksum-formatted server account address (not 'address' as previously documented) |
|
_infra_key |
Optional infrastructure API key for premium RPC endpoints (note: this attribute is not defined in the class) |
Environment Variables
- evm_private_key: Server's EVM private key for signing transactions (required)
- evm_infra_key: Optional infrastructure API key (e.g., Alchemy/Infura key) If not set, falls back to public RPC endpoints
Example
Initialize with environment variables
adapter = EVMAdapter() # Loads evm_private_key automatically (correct class name is EVMAdapter, not EVMServerAdapter)
Or explicitly provide private key (for testing)
adapter = EVMAdapter(private_key="0x...")
Verify permit signature and on-chain state
result = await adapter.verify_signature(permit, payment_requirement) if result.is_success(): # Execute the permit on-chain confirmation = await adapter.settle(permit)
__init__(*, private_key: Optional[str] = None, request_timeout: int = 60)
Initialize EVM Server Adapter with environment-aware configuration.
This constructor implements a flexible initialization pattern:
1. Accepts optional private_key parameter (useful for testing).
2. Falls back to the evm_private_key environment variable if not provided.
3. Initializes the server account object from the resolved private key.
Token address and RPC URL are not stored at initialization time. They are
supplied call-by-call via the rpc_url field of :class:EVMPaymentComponent
(or the permit object) during :meth:verify_signature and :meth:settle. This
design lets a single adapter instance handle arbitrary EVM networks without
reconfiguration.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
private_key
|
Optional[str]
|
Server's private key in 0x-prefixed hex format. When omitted,
the value of the |
None
|
request_timeout
|
int
|
HTTP request timeout in seconds passed to every
:class: |
60
|
Raises:
| Type | Description |
|---|---|
ValueError
|
If neither |
signature(payment_component: EVMPaymentComponent) -> Union[ERC3009Authorization, Permit2Signature]
async
Generate a signed authorization for the given payment component.
Scheme selection is based on the payment component's currency:
- USDC, EURC (and other ERC-3009 compatible tokens): ERC-3009
transferWithAuthorization is preferred.
- All other tokens: Permit2 permitTransferFrom is used as fallback.
Signing is performed entirely in-process via sign_universal. The chain
ID is parsed from payment_component.caip2 via
parse_caip2_eip155_chain_id; no metadata dict is accessed.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
payment_component
|
EVMPaymentComponent
|
:class:
|
required |
Returns:
| Type | Description |
|---|---|
Union[ERC3009Authorization, Permit2Signature]
|
|
Union[ERC3009Authorization, Permit2Signature]
|
|
Raises:
| Type | Description |
|---|---|
ValueError
|
If the chain is unsupported or required fields are invalid. |
TypeError
|
If the private key is not available. |
verify_signature(permit: Union[ERC3009Authorization, Permit2Signature], payment_requirement: EVMPaymentComponent) -> EVMVerificationResult
async
Verify a signed authorization and validate payment constraints.
Accepts the direct output of :meth:signature — either an
ERC3009Authorization or a Permit2Signature — and verifies it
using :func:verify_universal.
Validation steps
- Type check -
permitmust beERC3009AuthorizationorPermit2Signature. - Receiver check - the
recipient/spenderfield must match the server's wallet address. - Amount check - the authorized value (smallest token unit) must
be
>=the required amount derived frompayment_requirement.amount(human-readable USDC). - Balance check - the on-chain token balance of the sender must cover the required amount.
- Signature + expiry - delegates to :func:
verify_universal, which reconstructs the EIP-712 struct, recovers the signer, and checks the time window / deadline.
Amount conversion
payment_requirement.amount is a human-readable USDC quantity
(e.g. 1.5 for 1.5 USDC). It is converted to the smallest token
unit via :func:amount_to_value using payment_requirement.token_decimals
(defaults to 6 for USDC).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
permit
|
Union[ERC3009Authorization, Permit2Signature]
|
ERC3009Authorization or Permit2Signature produced by
:meth: |
required |
payment_requirement
|
EVMPaymentComponent
|
EVMPaymentComponent describing the expected
payment (human-readable |
required |
Returns:
| Type | Description |
|---|---|
EVMVerificationResult
|
EVMVerificationResult with |
EVMVerificationResult
|
|
EVMVerificationResult
|
failure result. No exceptions are raised. |
settle(permit: Union[ERC3009Authorization, Permit2Signature]) -> EVMTransactionConfirmation
async
Execute on-chain token transfer to settle a payment authorization.
Dispatches to the appropriate settlement strategy based on permit type:
- ERC3009Authorization → calls
transferWithAuthorizationdirectly on the token contract (ERC-3009 path, supported by USDC / EURC). - Permit2Signature → calls
permitTransferFromon the Uniswap Permit2 singleton contract.
Both paths share the same receipt-polling and confirmation-building
logic via :meth:_send_and_confirm.
Chain handling is fully dynamic: the Web3 RPC instance is resolved from
permit.chain_id at call time, with optional premium infra key.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
permit
|
Union[ERC3009Authorization, Permit2Signature]
|
Signed authorization produced by :meth: |
required |
Returns:
| Type | Description |
|---|---|
EVMTransactionConfirmation
|
class: |
EVMTransactionConfirmation
|
populated receipt fields on success, or a descriptive failure result. |
EVMTransactionConfirmation
|
No exceptions are raised; all errors are captured in the return value. |
get_balance(address: str, token_address: Optional[str] = None, web3: Optional[AsyncWeb3] = None) -> int
async
Query token balance for an address on-chain.
This method is designed to work in two modes: 1. With explicit parameters (token_address and web3) - for internal use 2. With just address - for external API use (not recommended without context)
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
address
|
str
|
Wallet address to query (0x-prefixed hex format) |
required |
token_address
|
Optional[str]
|
Token contract address (optional, for explicit chain/token context) |
None
|
web3
|
Optional[AsyncWeb3]
|
AsyncWeb3 instance (optional, created dynamically if not provided) |
None
|
Returns:
| Name | Type | Description |
|---|---|---|
int |
int
|
Token balance in smallest units (0 if address has no balance or error occurs) |
Note
When called from external code without token_address and web3, the method cannot determine which chain to query. Consider refactoring external calls to provide these parameters explicitly.
get_wallet_address() -> str
Get server wallet address from private key.
Returns:
| Name | Type | Description |
|---|---|---|
str |
str
|
Server wallet address in checksum format |
permit2_approve(chain_id: int, token_addr: str, values: int) -> Tuple[str, Optional[TxReceipt]]
async
Asynchronously signs and broadcasts an ERC20 approve transaction for Permit2.
This method approves the Permit2 singleton contract to spend tokens on behalf
of the server wallet. The approval amount is typically set to _MAX_UINT256
(infinite approval) to avoid repeated approval transactions.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
chain_id
|
int
|
The EVM chain ID (e.g., 1 for Ethereum, 11155111 for Sepolia). |
required |
token_addr
|
str
|
The contract address of the ERC20 token. |
required |
values
|
int
|
The raw amount (in smallest token units) to approve. |
required |
Returns:
| Type | Description |
|---|---|
str
|
A tuple containing: |
Optional[TxReceipt]
|
|
Tuple[str, Optional[TxReceipt]]
|
|
client_init(payment_components: List[BasePaymentComponent]) -> None
async
EVM-specific client-side pre-signing initialisation.
Iterates all registered EVMPaymentComponents and, for each currency
that requires the Permit2 protocol (i.e. is NOT natively ERC-3009),
queries the current ERC-20 allowance granted to the Permit2 singleton.
If the allowance is below _LOW_ALLOWANCE_THRESHOLD, an on-chain
approve(permit2, uint256_max) transaction is broadcast and awaited
before returning.
This must be called once at startup (via AdapterHub.initialize()) before any signature() calls are made. Skipped automatically for ERC-3009 currencies (USDC, EURC, …) because those use gasless transferWithAuthorization and do not require a prior on-chain approval.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
payment_components
|
List[BasePaymentComponent]
|
All components returned by register_payment_methods(). Non-EVM items are silently skipped. |
required |
Raises:
| Type | Description |
|---|---|
RuntimeError
|
If any approval transaction is broadcast but reverts on-chain (propagated from approve_erc20). |
Web3Exception
|
If the RPC allowance query fails. |
EVMPaymentComponent
Bases: BasePaymentComponent
EVM-Specific Payment Component.
Extends BasePaymentComponent with EVM-specific payment requirements including token contract address, chain identifier (CAIP-2), and an optional recipient address. Typically used for USDC payments on EVM networks (Ethereum, Sepolia, etc.).
Attributes:
| Name | Type | Description |
|---|---|---|
payment_type |
Literal['evm']
|
Always "evm" for this implementation |
amount |
float
|
Payment amount for human readability (e.g., 1.0 for 1 USDC) |
currency |
str
|
Currency code (typically "USD" for stablecoins) |
metadata |
Dict[str, Any]
|
Additional payment metadata (may include gas limits, fees, etc.) |
created_at |
datetime
|
Timestamp when payment component was created |
token |
str | None
|
Token contract address on specific EVM chain (EVM-specific) |
caip2 |
str
|
CAIP-2 chain identifier (e.g., "eip155:1", "eip155:11155111") (EVM-specific) |
pay_to |
str | None
|
Optional recipient address to pay to (EVM-specific) |
rpc_url |
str | None
|
Optional EVM RPC URL for this payment (EVM-specific) |
token_name |
str | None
|
Optional token name (e.g., "USDC") (EVM-specific) |
token_decimals |
str | int | None
|
Optional token decimals (string or int) (EVM-specific) |
token_version |
str | int | None
|
Optional token version (string or int) (EVM-specific) |
Example
payment = EVMPaymentComponent( payment_type="evm", amount=1.0, # 1 USDC currency="USD", token="0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48", caip2="eip155:11155111", metadata={"gas_price": "20", "priority_fee": "2"} )
validate_payment() -> bool
Validate EVM payment specification.
Checks that:
- payment_type is "evm"
- amount is non-negative
- token is valid EVM address format (0x...)
- caip2 is a valid CAIP-2 identifier for EVM chains (eip155:
Returns:
| Name | Type | Description |
|---|---|---|
bool |
bool
|
True if payment specification is valid |
Raises:
| Type | Description |
|---|---|
ValueError
|
With descriptive message if validation fails |
EVMVerificationResult
Bases: BaseVerificationResult
Unified EVM signature verification result.
Covers all EVM signing schemes (ERC-3009, Permit2, EIP-2612). Uses
scheme-neutral field names that mirror sign_universal /
verify_universal:
Attributes:
| Name | Type | Description |
|---|---|---|
verification_type |
Literal['evm']
|
Always |
sender |
Optional[str]
|
Token owner / authorizer address that produced the signature. |
receiver |
Optional[str]
|
Destination / authorised spender address. |
authorized_amount |
Optional[int]
|
Transfer amount in the token's smallest unit. |
blockchain_state |
Optional[Dict[str, Any]]
|
Optional on-chain state snapshot (balance, nonce, allowance, etc.). |
EVMTransactionConfirmation
Bases: BaseTransactionConfirmation
EVM-Specific Transaction Confirmation.
Extends BaseTransactionConfirmation with EVM-specific transaction receipt data. Returned by EVMServerAdapter.settle() method.
Attributes:
| Name | Type | Description |
|---|---|---|
confirmation_type |
Literal['evm']
|
Always "evm" for this implementation |
status |
TransactionStatus
|
Transaction execution status (inherited from BaseTransactionConfirmation) |
execution_time |
Optional[float]
|
Time taken to confirm transaction (seconds) (inherited from BaseTransactionConfirmation) |
confirmations |
int
|
Number of block confirmations (inherited from BaseTransactionConfirmation) |
error_message |
Optional[str]
|
Error details if transaction failed (inherited from BaseTransactionConfirmation) |
logs |
Optional[List[Dict[str, Any]]]
|
Transaction logs/events (inherited from BaseTransactionConfirmation) |
created_at |
datetime
|
Timestamp when confirmation was recorded (inherited from BaseTransactionConfirmation) |
tx_hash |
str
|
Transaction hash (0x-prefixed hex string on EVM) |
block_number |
Optional[int]
|
Block number containing transaction |
block_timestamp |
Optional[int]
|
Block timestamp (Unix) |
gas_used |
Optional[int]
|
Actual gas consumed by transaction |
gas_limit |
Optional[int]
|
Gas limit specified for transaction |
transaction_fee |
Optional[int]
|
Amount of ETH/native token paid as transaction fee (in wei) |
from_address |
Optional[str]
|
Transaction sender address |
to_address |
Optional[str]
|
Transaction receiver/contract address |
Example
confirmation = await evm_adapter.settle(permit) if confirmation.is_success(): print(f"Settlement confirmed: {confirmation.tx_hash}") print(f"Gas used: {confirmation.gas_used}") else: print(f"Settlement failed: {confirmation.error_message}")
EVM
以太坊虚拟机(EVM)区块链适配器
EVM 模块提供了针对以太坊虚拟机兼容区块链的专用适配器实现。它包含了处理 EVM 链上支付授权、签名验证、交易结算和链配置管理的完整工具集。
主要特性:
- ERC-3009 支持:USDC 和其他兼容代币的 transferWithAuthorization 离线签名
- Permit2 支持:通用 ERC-20 代币的 permitTransferFrom 离线授权
- 多链配置管理:统一的链配置和资产信息管理
- 智能合约交互:完整的 ERC-20、ERC-3009 和 Permit2 ABI 定义
- 签名验证:链上和链下的签名验证机制
- 配置工具:从外部源获取链信息和代币列表的实用工具
主要组件:
- EVMAdapter:主要的 EVM 区块链适配器类
- EVMRegistry:EVM 支付方法注册表
- EVMECDSASignature、EVMTokenPermit、ERC3009Authorization、Permit2Signature:签名和授权数据结构
- EVMVerificationResult、EVMTransactionConfirmation:验证和交易结果模型
配置管理工具:
EvmPublicRpcFromChainList
实时从 Chainlist.org 拉取公共 RPC 端点。支持按协议(https / wss)和隐私等级(none / limited)过滤,帮助你快速为任意 EVM 链找到可用的无需 API Key 的 RPC 地址。
from x402_mock.adapters.evm import EvmPublicRpcFromChainList
rpc = EvmPublicRpcFromChainList()
# 直接获取一个可用的公共 HTTPS RPC
print(rpc.pick_public_rpc("eip155:1"))
# 只要无隐私跟踪的节点
print(rpc.pick_public_rpc("eip155:8453", tracking_type="none"))
EvmTokenListFromUniswap
从 Uniswap 官方代币列表 查询任意代币的合约地址和精度(decimals)。结果会自动缓存,避免重复请求网络。
from x402_mock.adapters.evm import EvmTokenListFromUniswap
tokens = EvmTokenListFromUniswap()
# 查询以太坊主网 USDC 的合约地址和精度
address, decimals = tokens.get_token_address_and_decimals("eip155:1", "USDC")
print(address, decimals)
EvmChainInfoFromEthereumLists
从 ethereum-lists 仓库获取权威链元数据,专门用于解析 Infura / Alchemy 的带 API Key 占位符的 RPC 模板,以及枚举无需密钥的公共 RPC 列表。
from x402_mock.adapters.evm import EvmChainInfoFromEthereumLists
chain = EvmChainInfoFromEthereumLists()
# 获取 Infura / Alchemy RPC 模板(含 {API_KEY} 占位符)
print(chain.get_infura_rpc_url("eip155:1"))
print(chain.get_alchemy_rpc_url("eip155:1"))
# 列出所有无需 API Key 的公共端点
print(chain.get_public_rpc_urls("eip155:137"))
其他实用函数:
get_private_key_from_env():从环境变量加载 EVM 服务器私钥get_rpc_key_from_env():从环境变量加载 EVM 基础设施 API 密钥amount_to_value()/value_to_amount():代币金额与链上最小单位之间的互转parse_caip2_eip155_chain_id():将 CAIP-2 标识符解析为整数链 IDfetch_erc20_name_version_decimals():从链上 RPC 读取代币名称、版本和精度
evm
EVMAdapter
Bases: AdapterFactory
EVM Blockchain Server Adapter Implementation.
Provides complete server-side functionality for EVM blockchains including: - EIP2612 permit signature verification - On-chain state validation (nonce, allowance, balance, expiration) - Token transfer execution via signed permits - Balance and allowance queries
This adapter validates all security constraints before executing any on-chain operations. All methods are designed to return clear, actionable error messages rather than raising exceptions.
Key Design Features: - Dynamic RPC URL selection based on permit's chain_id - Environment-aware infrastructure key handling (evm_infra_key for premium RPC, falls back to public) - Private key loaded from environment (evm_private_key) during initialization - Lazy Web3 instance creation per blockchain interaction (ensures correct RPC endpoint)
Attributes:
| Name | Type | Description |
|---|---|---|
account |
Server's account object (initialized from evm_private_key environment variable) |
|
wallet_address |
Checksum-formatted server account address (not 'address' as previously documented) |
|
_infra_key |
Optional infrastructure API key for premium RPC endpoints (note: this attribute is not defined in the class) |
Environment Variables
- evm_private_key: Server's EVM private key for signing transactions (required)
- evm_infra_key: Optional infrastructure API key (e.g., Alchemy/Infura key) If not set, falls back to public RPC endpoints
Example
Initialize with environment variables
adapter = EVMAdapter() # Loads evm_private_key automatically (correct class name is EVMAdapter, not EVMServerAdapter)
Or explicitly provide private key (for testing)
adapter = EVMAdapter(private_key="0x...")
Verify permit signature and on-chain state
result = await adapter.verify_signature(permit, payment_requirement) if result.is_success(): # Execute the permit on-chain confirmation = await adapter.settle(permit)
__init__(*, private_key: Optional[str] = None, request_timeout: int = 60)
Initialize EVM Server Adapter with environment-aware configuration.
This constructor implements a flexible initialization pattern:
1. Accepts optional private_key parameter (useful for testing).
2. Falls back to the evm_private_key environment variable if not provided.
3. Initializes the server account object from the resolved private key.
Token address and RPC URL are not stored at initialization time. They are
supplied call-by-call via the rpc_url field of :class:EVMPaymentComponent
(or the permit object) during :meth:verify_signature and :meth:settle. This
design lets a single adapter instance handle arbitrary EVM networks without
reconfiguration.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
private_key
|
Optional[str]
|
Server's private key in 0x-prefixed hex format. When omitted,
the value of the |
None
|
request_timeout
|
int
|
HTTP request timeout in seconds passed to every
:class: |
60
|
Raises:
| Type | Description |
|---|---|
ValueError
|
If neither |
signature(payment_component: EVMPaymentComponent) -> Union[ERC3009Authorization, Permit2Signature]
async
Generate a signed authorization for the given payment component.
Scheme selection is based on the payment component's currency:
- USDC, EURC (and other ERC-3009 compatible tokens): ERC-3009
transferWithAuthorization is preferred.
- All other tokens: Permit2 permitTransferFrom is used as fallback.
Signing is performed entirely in-process via sign_universal. The chain
ID is parsed from payment_component.caip2 via
parse_caip2_eip155_chain_id; no metadata dict is accessed.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
payment_component
|
EVMPaymentComponent
|
:class:
|
required |
Returns:
| Type | Description |
|---|---|
Union[ERC3009Authorization, Permit2Signature]
|
|
Union[ERC3009Authorization, Permit2Signature]
|
|
Raises:
| Type | Description |
|---|---|
ValueError
|
If the chain is unsupported or required fields are invalid. |
TypeError
|
If the private key is not available. |
verify_signature(permit: Union[ERC3009Authorization, Permit2Signature], payment_requirement: EVMPaymentComponent) -> EVMVerificationResult
async
Verify a signed authorization and validate payment constraints.
Accepts the direct output of :meth:signature — either an
ERC3009Authorization or a Permit2Signature — and verifies it
using :func:verify_universal.
Validation steps
- Type check -
permitmust beERC3009AuthorizationorPermit2Signature. - Receiver check - the
recipient/spenderfield must match the server's wallet address. - Amount check - the authorized value (smallest token unit) must
be
>=the required amount derived frompayment_requirement.amount(human-readable USDC). - Balance check - the on-chain token balance of the sender must cover the required amount.
- Signature + expiry - delegates to :func:
verify_universal, which reconstructs the EIP-712 struct, recovers the signer, and checks the time window / deadline.
Amount conversion
payment_requirement.amount is a human-readable USDC quantity
(e.g. 1.5 for 1.5 USDC). It is converted to the smallest token
unit via :func:amount_to_value using payment_requirement.token_decimals
(defaults to 6 for USDC).
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
permit
|
Union[ERC3009Authorization, Permit2Signature]
|
ERC3009Authorization or Permit2Signature produced by
:meth: |
required |
payment_requirement
|
EVMPaymentComponent
|
EVMPaymentComponent describing the expected
payment (human-readable |
required |
Returns:
| Type | Description |
|---|---|
EVMVerificationResult
|
EVMVerificationResult with |
EVMVerificationResult
|
|
EVMVerificationResult
|
failure result. No exceptions are raised. |
settle(permit: Union[ERC3009Authorization, Permit2Signature]) -> EVMTransactionConfirmation
async
Execute on-chain token transfer to settle a payment authorization.
Dispatches to the appropriate settlement strategy based on permit type:
- ERC3009Authorization → calls
transferWithAuthorizationdirectly on the token contract (ERC-3009 path, supported by USDC / EURC). - Permit2Signature → calls
permitTransferFromon the Uniswap Permit2 singleton contract.
Both paths share the same receipt-polling and confirmation-building
logic via :meth:_send_and_confirm.
Chain handling is fully dynamic: the Web3 RPC instance is resolved from
permit.chain_id at call time, with optional premium infra key.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
permit
|
Union[ERC3009Authorization, Permit2Signature]
|
Signed authorization produced by :meth: |
required |
Returns:
| Type | Description |
|---|---|
EVMTransactionConfirmation
|
class: |
EVMTransactionConfirmation
|
populated receipt fields on success, or a descriptive failure result. |
EVMTransactionConfirmation
|
No exceptions are raised; all errors are captured in the return value. |
get_balance(address: str, token_address: Optional[str] = None, web3: Optional[AsyncWeb3] = None) -> int
async
Query token balance for an address on-chain.
This method is designed to work in two modes: 1. With explicit parameters (token_address and web3) - for internal use 2. With just address - for external API use (not recommended without context)
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
address
|
str
|
Wallet address to query (0x-prefixed hex format) |
required |
token_address
|
Optional[str]
|
Token contract address (optional, for explicit chain/token context) |
None
|
web3
|
Optional[AsyncWeb3]
|
AsyncWeb3 instance (optional, created dynamically if not provided) |
None
|
Returns:
| Name | Type | Description |
|---|---|---|
int |
int
|
Token balance in smallest units (0 if address has no balance or error occurs) |
Note
When called from external code without token_address and web3, the method cannot determine which chain to query. Consider refactoring external calls to provide these parameters explicitly.
get_wallet_address() -> str
Get server wallet address from private key.
Returns:
| Name | Type | Description |
|---|---|---|
str |
str
|
Server wallet address in checksum format |
permit2_approve(chain_id: int, token_addr: str, values: int) -> Tuple[str, Optional[TxReceipt]]
async
Asynchronously signs and broadcasts an ERC20 approve transaction for Permit2.
This method approves the Permit2 singleton contract to spend tokens on behalf
of the server wallet. The approval amount is typically set to _MAX_UINT256
(infinite approval) to avoid repeated approval transactions.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
chain_id
|
int
|
The EVM chain ID (e.g., 1 for Ethereum, 11155111 for Sepolia). |
required |
token_addr
|
str
|
The contract address of the ERC20 token. |
required |
values
|
int
|
The raw amount (in smallest token units) to approve. |
required |
Returns:
| Type | Description |
|---|---|
str
|
A tuple containing: |
Optional[TxReceipt]
|
|
Tuple[str, Optional[TxReceipt]]
|
|
client_init(payment_components: List[BasePaymentComponent]) -> None
async
EVM-specific client-side pre-signing initialisation.
Iterates all registered EVMPaymentComponents and, for each currency
that requires the Permit2 protocol (i.e. is NOT natively ERC-3009),
queries the current ERC-20 allowance granted to the Permit2 singleton.
If the allowance is below _LOW_ALLOWANCE_THRESHOLD, an on-chain
approve(permit2, uint256_max) transaction is broadcast and awaited
before returning.
This must be called once at startup (via AdapterHub.initialize()) before any signature() calls are made. Skipped automatically for ERC-3009 currencies (USDC, EURC, …) because those use gasless transferWithAuthorization and do not require a prior on-chain approval.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
payment_components
|
List[BasePaymentComponent]
|
All components returned by register_payment_methods(). Non-EVM items are silently skipped. |
required |
Raises:
| Type | Description |
|---|---|
RuntimeError
|
If any approval transaction is broadcast but reverts on-chain (propagated from approve_erc20). |
Web3Exception
|
If the RPC allowance query fails. |
EVMECDSASignature
Bases: BaseSignature
Unified EVM ECDSA signature (v, r, s).
Shared base for all EVM standards that produce a three-component ECDSA
signature. Use signature_type to identify the signing standard:
"EIP2612"— EIP-2612permit()calls."ERC3009"— ERC-3009transferWithAuthorizationcalls."Permit2"— Uniswap Permit2permitTransferFromcalls.
Attributes:
| Name | Type | Description |
|---|---|---|
signature_type |
Literal['EIP2612', 'ERC3009', 'Permit2']
|
One of |
v |
int
|
ECDSA recovery ID (27 or 28). |
r |
str
|
r component — 32 bytes as a 64-char hex string (0x prefix optional). |
s |
str
|
s component — 32 bytes as a 64-char hex string (0x prefix optional). |
Example::
sig = EVMECDSASignature(signature_type="EIP2612", v=27, r="0x" + "a" * 64, s="0x" + "b" * 64)
sig.validate_format()
validate_format() -> bool
Validate v/r/s components.
Checks v is 27 or 28 and that r/s are valid 64-character hex strings (0x prefix stripped before length check).
Returns:
| Type | Description |
|---|---|
bool
|
True when all components pass. |
Raises:
| Type | Description |
|---|---|
ValueError
|
Descriptive message on the first failed check. |
to_packed_hex() -> str
Encode v/r/s into a packed 65-byte hex string (r || s || v).
This is the format expected by on-chain contracts such as Permit2's
permitTransferFrom which accept a raw bytes signature argument.
Returns:
| Type | Description |
|---|---|
str
|
0x-prefixed 132-character hex string. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If components do not pass |
EVMTokenPermit
Bases: BasePermit
EVM Token Approval Permit (EIP-2612).
Encapsulates a signed EIP-2612 permit() authorization, allowing a
spender to transfer tokens from the owner's account without a prior
on-chain approve() call. Pass to EVMServerAdapter.settle() for
settlement.
Use permit_type to distinguish permit standards at call sites.
Attributes:
| Name | Type | Description |
|---|---|---|
permit_type |
Literal['EIP2612']
|
Always |
owner |
str
|
Token owner's wallet address (0x-prefixed, 42 chars). |
spender |
str
|
Address authorized to spend; typically the settlement server. |
token |
str
|
ERC-20 token contract address. |
value |
int
|
Approved amount in the token's smallest unit. |
nonce |
int
|
On-chain nonce from the token contract (replay protection). |
deadline |
int
|
Unix timestamp after which the permit is invalid. |
chain_id |
int
|
EVM network ID (e.g. 1 = Mainnet, 11155111 = Sepolia). |
signature |
EVMECDSASignature
|
ECDSA signature with |
Example::
permit = EVMTokenPermit(
owner="0x1234...5678",
spender="0x8765...4321",
token="0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48",
value=1_000_000,
nonce=0,
deadline=1_900_000_000,
chain_id=11155111,
signature=EVMECDSASignature(signature_type="EIP2612", v=27, r="0x...", s="0x..."),
)
validate_structure() -> bool
Validate permit fields and embedded signature.
Checks that owner, spender, and token are valid 42-char
0x-prefixed addresses, that chain_id is positive, and that the
attached EVMECDSASignature passes its own format validation.
Returns:
| Type | Description |
|---|---|
bool
|
True when all checks pass. |
Raises:
| Type | Description |
|---|---|
ValueError
|
With a descriptive message on the first failed check. |
ERC3009Authorization
Bases: BasePermit
ERC-3009 Authorization (TransferWithAuthorization) container.
This model captures the canonical fields of an ERC-3009 authorization (also referred to as "TransferWithAuthorization" in the EIP). It is intended as a data-only representation for signing, transport and verification routines elsewhere in the system.
Notes
- The model does not perform cryptographic verification; it only describes the data structure expected by such verification.
Attributes:
| Name | Type | Description |
|---|---|---|
permit_type |
Literal['ERC3009']
|
Literal identifier for this authorization type ("ERC3009"). |
token |
str
|
Token contract address the authorization applies to. |
chain_id |
int
|
Numeric chain identifier where the authorization is valid. |
authorizer |
str
|
Address authorizing the transfer (field named |
recipient |
str
|
Address receiving tokens (maps to |
value |
int
|
Amount authorized for transfer (uint256 smallest units). |
validAfter |
int
|
Start timestamp (inclusive) for validity. |
validBefore |
int
|
Expiry timestamp (exclusive) for validity. |
nonce |
str
|
Unique nonce (bytes32 hex string) preventing replay. |
signature |
Optional[EVMECDSASignature]
|
ERC3009Signature container with v/r/s. |
Permit2Signature
Bases: BasePermit
Permit2 permitTransferFrom authorization payload.
Mirrors the structure of :class:ERC3009Authorization: inherits
:class:BasePermit and embeds the ECDSA signature as a separate
:class:EVMECDSASignature object rather than mixing v/r/s directly
into the permit fields.
EIP-712 domain: name="Permit2", chainId=chain_id,
verifyingContract=permit2_address.
Attributes:
| Name | Type | Description |
|---|---|---|
permit_type |
Literal['Permit2']
|
Always |
owner |
str
|
Token owner address that produced the signature. |
spender |
str
|
Address authorized to call |
token |
str
|
ERC-20 token contract address. |
amount |
int
|
Transfer amount in the token's smallest unit. |
nonce |
int
|
Permit2 contract nonce for |
deadline |
int
|
Unix timestamp after which the permit is invalid. |
chain_id |
int
|
EVM network ID (e.g. 1=Mainnet, 11155111=Sepolia). |
permit2_address |
str
|
Permit2 contract address (defaults to canonical Uniswap deployment). |
signature |
Optional[EVMECDSASignature]
|
ECDSA signature ( |
Example::
sig = Permit2Signature(
owner="0xAbCd...1234", spender="0xServer...Addr",
token="0xA0b8...eB48", amount=1_000_000,
nonce=0, deadline=1_900_000_000, chain_id=11155111,
signature=EVMECDSASignature(
signature_type="Permit2",
v=27, r="0x" + "a" * 64, s="0x" + "b" * 64,
),
)
validate_structure() -> bool
Validate permit fields and embedded signature.
Checks that owner, spender, token, and permit2_address
are valid 0x-prefixed 42-character EVM addresses, and that the
attached :class:EVMECDSASignature passes its own format validation.
Returns:
| Type | Description |
|---|---|
bool
|
True when all checks pass. |
Raises:
| Type | Description |
|---|---|
ValueError
|
Descriptive message on the first failed check. |
EVMPaymentComponent
Bases: BasePaymentComponent
EVM-Specific Payment Component.
Extends BasePaymentComponent with EVM-specific payment requirements including token contract address, chain identifier (CAIP-2), and an optional recipient address. Typically used for USDC payments on EVM networks (Ethereum, Sepolia, etc.).
Attributes:
| Name | Type | Description |
|---|---|---|
payment_type |
Literal['evm']
|
Always "evm" for this implementation |
amount |
float
|
Payment amount for human readability (e.g., 1.0 for 1 USDC) |
currency |
str
|
Currency code (typically "USD" for stablecoins) |
metadata |
Dict[str, Any]
|
Additional payment metadata (may include gas limits, fees, etc.) |
created_at |
datetime
|
Timestamp when payment component was created |
token |
str | None
|
Token contract address on specific EVM chain (EVM-specific) |
caip2 |
str
|
CAIP-2 chain identifier (e.g., "eip155:1", "eip155:11155111") (EVM-specific) |
pay_to |
str | None
|
Optional recipient address to pay to (EVM-specific) |
rpc_url |
str | None
|
Optional EVM RPC URL for this payment (EVM-specific) |
token_name |
str | None
|
Optional token name (e.g., "USDC") (EVM-specific) |
token_decimals |
str | int | None
|
Optional token decimals (string or int) (EVM-specific) |
token_version |
str | int | None
|
Optional token version (string or int) (EVM-specific) |
Example
payment = EVMPaymentComponent( payment_type="evm", amount=1.0, # 1 USDC currency="USD", token="0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48", caip2="eip155:11155111", metadata={"gas_price": "20", "priority_fee": "2"} )
validate_payment() -> bool
Validate EVM payment specification.
Checks that:
- payment_type is "evm"
- amount is non-negative
- token is valid EVM address format (0x...)
- caip2 is a valid CAIP-2 identifier for EVM chains (eip155:
Returns:
| Name | Type | Description |
|---|---|---|
bool |
bool
|
True if payment specification is valid |
Raises:
| Type | Description |
|---|---|
ValueError
|
With descriptive message if validation fails |
EVMVerificationResult
Bases: BaseVerificationResult
Unified EVM signature verification result.
Covers all EVM signing schemes (ERC-3009, Permit2, EIP-2612). Uses
scheme-neutral field names that mirror sign_universal /
verify_universal:
Attributes:
| Name | Type | Description |
|---|---|---|
verification_type |
Literal['evm']
|
Always |
sender |
Optional[str]
|
Token owner / authorizer address that produced the signature. |
receiver |
Optional[str]
|
Destination / authorised spender address. |
authorized_amount |
Optional[int]
|
Transfer amount in the token's smallest unit. |
blockchain_state |
Optional[Dict[str, Any]]
|
Optional on-chain state snapshot (balance, nonce, allowance, etc.). |
EVMTransactionConfirmation
Bases: BaseTransactionConfirmation
EVM-Specific Transaction Confirmation.
Extends BaseTransactionConfirmation with EVM-specific transaction receipt data. Returned by EVMServerAdapter.settle() method.
Attributes:
| Name | Type | Description |
|---|---|---|
confirmation_type |
Literal['evm']
|
Always "evm" for this implementation |
status |
TransactionStatus
|
Transaction execution status (inherited from BaseTransactionConfirmation) |
execution_time |
Optional[float]
|
Time taken to confirm transaction (seconds) (inherited from BaseTransactionConfirmation) |
confirmations |
int
|
Number of block confirmations (inherited from BaseTransactionConfirmation) |
error_message |
Optional[str]
|
Error details if transaction failed (inherited from BaseTransactionConfirmation) |
logs |
Optional[List[Dict[str, Any]]]
|
Transaction logs/events (inherited from BaseTransactionConfirmation) |
created_at |
datetime
|
Timestamp when confirmation was recorded (inherited from BaseTransactionConfirmation) |
tx_hash |
str
|
Transaction hash (0x-prefixed hex string on EVM) |
block_number |
Optional[int]
|
Block number containing transaction |
block_timestamp |
Optional[int]
|
Block timestamp (Unix) |
gas_used |
Optional[int]
|
Actual gas consumed by transaction |
gas_limit |
Optional[int]
|
Gas limit specified for transaction |
transaction_fee |
Optional[int]
|
Amount of ETH/native token paid as transaction fee (in wei) |
from_address |
Optional[str]
|
Transaction sender address |
to_address |
Optional[str]
|
Transaction receiver/contract address |
Example
confirmation = await evm_adapter.settle(permit) if confirmation.is_success(): print(f"Settlement confirmed: {confirmation.tx_hash}") print(f"Gas used: {confirmation.gas_used}") else: print(f"Settlement failed: {confirmation.error_message}")
ERC4337ValidationResult
Bases: BaseVerificationResult
Validation result container for ERC-4337 user operations.
This model holds the outcome of the validateUserOp call performed by
bundlers or entrypoints (using ERC-4337 terminology) and provides
diagnostic information useful for callers and logging. It purposely does
not attempt to re-create the full on-chain validation machinery.
Attributes:
| Name | Type | Description |
|---|---|---|
verification_type |
Literal['ERC4337']
|
Literal identifier ("ERC4337"). |
user_op_hash |
Optional[str]
|
Optional hex digest of the user operation. |
entry_point |
Optional[str]
|
Entrypoint contract address that validated the op. |
bundle_id |
Optional[str]
|
Optional bundler identifier that accepted/processed the op. |
is_valid |
Optional[bool]
|
Boolean indicating validation success when known. |
validation_gas_used |
Optional[int]
|
Optional gas used by the validateUserOp routine. |
error_details |
Optional[Dict[str, Any]]
|
Optional structured error information. |
ERC4337UserOpPayload
Bases: CanonicalModel
EIP-4337 UserOperation submission payload.
This model wraps a UserOperationModel together with the target entry
point contract address and chain id, forming the complete payload that
would be submitted to a bundler or entrypoint for account-abstraction
settlement. It does not represent a raw cryptographic signature; rather
it encapsulates the full account-abstraction authorization object used
during the settle phase.
Attributes:
| Name | Type | Description |
|---|---|---|
signature_type |
Literal['ERC4337']
|
Literal identifier ("ERC4337"). |
user_operation |
UserOperationModel
|
The |
entry_point |
str
|
Entry point contract address that will process the op. |
chain_id |
int
|
Numeric chain id where the operation is intended. |
UserOperationModel
Bases: CanonicalModel
Minimal EIP-4337 UserOperation structure as a Pydantic model.
This model is a Pydantic representation of the UserOperation fields
commonly used with account-abstraction and bundlers. It is provided here
for transport and validation of the user-operation payload; it does not
implement packing or signing logic.
Attributes mirror the canonical EIP-4337 fields: sender, nonce,
initCode, callData, gas fields, fee fields, paymasterAndData, and
the signature blob.
sign_erc3009_authorization(*, private_key: str, token: str, chain_id: int, authorizer: str, recipient: str, value: int, valid_after: int, valid_before: int, domain_name: str, domain_version: str, nonce: Optional[str] = None) -> ERC3009Authorization
Sign an ERC-3009 transferWithAuthorization payload and return a
complete ERC3009Authorization with the signature attached.
Signing is performed entirely in-process via eth_account; no RPC
endpoint or network connection is required. The EIP-712 structured-data
hash is computed from the token's domain separator and the authorization
message fields, then signed with the supplied private key to produce the
canonical (v, r, s) ECDSA components.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
private_key
|
str
|
Hex-encoded secp256k1 private key of the authorizer
(with or without |
required |
token
|
str
|
ERC-20 token contract address (0x-prefixed, 42 chars).
Also used as the EIP-712 |
required |
chain_id
|
int
|
EVM network ID (e.g. |
required |
authorizer
|
str
|
Address that owns the tokens and is authorising the
transfer ( |
required |
recipient
|
str
|
Address that will receive the tokens
( |
required |
value
|
int
|
Amount to transfer in the token's smallest unit. |
required |
valid_after
|
int
|
Unix timestamp after which the authorisation is valid.
Pass |
required |
valid_before
|
int
|
Unix timestamp before which the authorisation must be submitted on-chain. |
required |
domain_name
|
str
|
EIP-712 domain |
required |
domain_version
|
str
|
EIP-712 domain |
required |
nonce
|
Optional[str]
|
Optional bytes32 hex string for replay protection. A cryptographically random nonce is generated automatically when omitted. |
None
|
Returns:
| Type | Description |
|---|---|
ERC3009Authorization
|
|
Raises:
| Type | Description |
|---|---|
ValueError
|
If |
Example::
auth = sign_erc3009_authorization(
private_key="0xYOUR_PRIVATE_KEY",
token="0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48",
chain_id=1,
authorizer="0xYourAddress",
recipient="0xRecipientAddress",
value=1_000_000, # 1 USDC (6 decimals)
valid_after=0,
valid_before=1_900_000_000,
domain_name="USD Coin",
domain_version="2",
)
# auth.signature contains v, r, s — ready for on-chain submission
sign_permit2(*, private_key: str, owner: str, spender: str, token: str, amount: int, nonce: int, deadline: int, chain_id: int, permit2_address: str = '0x000000000022D473030F116dDEE9F6B43aC78BA3') -> Permit2Signature
Sign a Permit2 permitTransferFrom authorization and return a
Permit2Signature with v, r, s attached.
Signing is performed entirely in-process via eth_account. The
EIP-712 structured-data hash follows the canonical Permit2 domain
(name="Permit2", no version field) and the
PermitTransferFrom type with a nested TokenPermissions sub-struct.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
private_key
|
str
|
Hex-encoded secp256k1 private key of the token owner
(with or without |
required |
owner
|
str
|
Token owner address (0x-prefixed, 42 chars).
Must match the address derived from |
required |
spender
|
str
|
Address authorised to call |
required |
token
|
str
|
ERC-20 token contract address. |
required |
amount
|
int
|
Transfer amount in the token's smallest unit. |
required |
nonce
|
int
|
Permit2 nonce for |
required |
deadline
|
int
|
Unix timestamp after which the permit is invalid. |
required |
chain_id
|
int
|
EVM network ID (e.g. |
required |
permit2_address
|
str
|
Permit2 singleton contract address. Defaults to the
canonical Uniswap deployment
|
'0x000000000022D473030F116dDEE9F6B43aC78BA3'
|
Returns:
| Type | Description |
|---|---|
Permit2Signature
|
|
Permit2Signature
|
populated. Call |
Permit2Signature
|
|
Example::
sig = sign_permit2(
private_key="0xYOUR_PRIVATE_KEY",
owner="0xYourAddress",
spender="0xServerAddress",
token="0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48",
amount=1_000_000, # 1 USDC (6 decimals)
nonce=0,
deadline=1_900_000_000,
chain_id=1,
)
packed_sig = sig.signature.to_packed_hex() # ready for on-chain submission
approve_erc20(w3: AsyncWeb3, token_addr: str, private_key: str, spender: str, amount: int, wait: bool = True) -> Tuple[str, Optional[TxReceipt]]
async
Asynchronously signs and broadcasts an ERC20 approve transaction.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
w3
|
AsyncWeb3
|
An instance of AsyncWeb3. |
required |
token_addr
|
str
|
The contract address of the ERC20 token. |
required |
private_key
|
str
|
The hex string private key of the sender. |
required |
spender
|
str
|
The address authorized to spend the tokens. |
required |
amount
|
int
|
The raw amount (in wei) to approve. |
required |
wait
|
bool
|
If True, waits for the transaction receipt before returning. |
True
|
Returns:
| Type | Description |
|---|---|
Tuple[str, Optional[TxReceipt]]
|
A tuple of (transaction_hash_hex, transaction_receipt). |
is_erc3009_currency(currency: str) -> bool
Determines whether the given currency symbol natively supports ERC-3009.
ERC-3009 (transferWithAuthorization) is primarily implemented by Circle's stablecoins, allowing for gasless transfers via signed authorizations. Tokens not on this list will typically fall back to the Permit2 protocol.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
currency
|
str
|
Uppercase currency / token symbol (e.g., "USDC", "EURC"). |
required |
Returns:
| Name | Type | Description |
|---|---|---|
bool |
bool
|
True if the token natively supports ERC-3009; False otherwise. |
sign_universal(*, private_key: str, chain_id: int, token: str, sender: str, receiver: str, amount: int, scheme: Optional[Literal['erc3009', 'permit2']] = None, domain_name: Optional[str] = None, domain_version: str = '2', deadline: Optional[int] = None, valid_after: int = 0, nonce: Optional[Union[int, str]] = None, permit2_address: str = '0x000000000022D473030F116dDEE9F6B43aC78BA3') -> Union[ERC3009Authorization, Permit2Signature]
Unified entry point for ERC-3009 and Permit2 signing.
Callers supply only the five essential parameters
(private_key, chain_id, token, sender, receiver,
amount); everything else is inferred or defaulted automatically:
+-----------------+----------------------------------------------+
| Parameter | Default |
+=================+==============================================+
| deadline | int(time.time()) + 3600 (now + 1 hour) |
+-----------------+----------------------------------------------+
| valid_after | 0 (immediately valid) |
+-----------------+----------------------------------------------+
| domain_version | "2" |
+-----------------+----------------------------------------------+
| nonce | random bytes32 (ERC-3009) / 0 (Permit2) |
+-----------------+----------------------------------------------+
Scheme selection
domain_nameprovided, orscheme="erc3009"→ ERC-3009domain_nameabsent, orscheme="permit2"→ Permit2
Parameters
private_key : Hex-encoded secp256k1 private key (with or without 0x).
chain_id : EVM network ID.
token : ERC-20 token contract address.
sender : Token owner / authorizer address.
receiver : Destination / authorised spender address.
amount : Transfer amount in the token's smallest unit.
scheme : Optional explicit scheme override.
domain_name : EIP-712 domain name (required for ERC-3009,
e.g. "USD Coin" for USDC).
domain_version : EIP-712 domain version string; defaults to "2".
deadline : Expiry Unix timestamp; defaults to now + 3600 s.
valid_after : ERC-3009 start timestamp; defaults to 0.
nonce : Replay-protection nonce. ERC-3009 accepts a bytes32 hex
string or an int (zero-padded); auto-generated when
None. Permit2 accepts an int; defaults to 0.
permit2_address: Permit2 singleton contract address.
Returns
ERC3009Authorization or Permit2Signature depending on scheme.
Raises
ValueError
If ERC-3009 is selected but domain_name is not provided.
Examples
ERC-3009 (all defaults, only domain_name added)::
auth = sign_universal(
private_key="0xKEY", chain_id=1,
token="0xA0b869...", sender="0xFrom", receiver="0xTo",
amount=1_000_000, domain_name="USD Coin",
)
Permit2 (fully defaulted)::
sig = sign_universal(
private_key="0xKEY", chain_id=1,
token="0xA0b869...", sender="0xOwner", receiver="0xSpender",
amount=1_000_000,
)
build_erc3009_typed_data(authorization: ERC3009Authorization, *, domain_name: str, domain_version: str) -> ERC3009TypedData
Wrap an ERC3009Authorization in an EIP-712 ERC3009TypedData envelope
without signing.
Use this when signing is handled externally (e.g. a hardware wallet or
MPC service). For the common in-process case, prefer
sign_erc3009_authorization.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
authorization
|
ERC3009Authorization
|
An |
required |
domain_name
|
str
|
EIP-712 domain |
required |
domain_version
|
str
|
EIP-712 domain |
required |
Returns:
| Type | Description |
|---|---|
ERC3009TypedData
|
|
ERC3009TypedData
|
|
Example::
typed_data = build_erc3009_typed_data(
authorization,
domain_name="USD Coin",
domain_version="2",
)
payload = typed_data.to_dict() # hand off to external signer
verify_erc3009(*, token: str, chain_id: int, authorizer: str, recipient: str, value: int, valid_after: int, valid_before: int, nonce: str, v: int, r: str, s: str, domain_name: str, domain_version: str, owner_balance: Optional[int] = None, on_chain_nonce: Optional[str] = None, current_time: Optional[int] = None, w3=None) -> EVMVerificationResult
async
Verify an ERC-3009 transferWithAuthorization signed by an EOA or a smart-contract wallet (ERC-1271).
Performs the following checks in order, returning on the first failure:
- Address format --
authorizerandrecipientmust be 0x-prefixed, 42-character strings. - Time window --
current_timemust satisfyvalid_after < current_time < valid_before. - Balance -- when
owner_balanceis supplied, it must be>= value. - Nonce -- when
on_chain_nonceis supplied, it is compared case-insensitively againstnonce; a mismatch indicates the authorization has already been used. - ECDSA recovery -- reconstructs the EIP-712 struct hash from the
supplied fields and recovers the signer address from (v, r, s).
The recovered address must equal
authorizer(case-insensitive).
Cryptographic checks are performed in-process. When a web3.Web3 instance
is provided, the verifier may also perform on-chain validation steps such as
querying DOMAIN_SEPARATOR() to ensure the supplied EIP-712 domain
parameters match the token contract (preventing late settlement failures).
If the EOA ECDSA check fails and w3 is provided, an ERC-1271
isValidSignature call is attempted, enabling smart-contract wallet
(e.g. Safe, Argent) support.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
token
|
str
|
ERC-20 token contract address (0x-prefixed, 42 chars).
Also used as the EIP-712 |
required |
chain_id
|
int
|
EVM network ID (e.g. |
required |
authorizer
|
str
|
Address that signed the authorization ( |
required |
recipient
|
str
|
Address that will receive the tokens ( |
required |
value
|
int
|
Transfer amount in the token's smallest unit. |
required |
valid_after
|
int
|
Unix timestamp after which the authorization is valid. |
required |
valid_before
|
int
|
Unix timestamp before which it must be submitted. |
required |
nonce
|
str
|
bytes32 hex string used for replay protection. |
required |
v
|
int
|
ECDSA recovery ID (27 or 28). |
required |
r
|
str
|
Signature |
required |
s
|
str
|
Signature |
required |
domain_name
|
str
|
EIP-712 domain |
required |
domain_version
|
str
|
EIP-712 domain |
required |
owner_balance
|
Optional[int]
|
Optional current token balance of |
None
|
on_chain_nonce
|
Optional[str]
|
Optional on-chain nonce (bytes32 hex string) for
|
None
|
current_time
|
Optional[int]
|
Optional Unix timestamp used for time-window checks.
Defaults to |
None
|
w3
|
Optional |
None
|
Returns:
| Type | Description |
|---|---|
EVMVerificationResult
|
|
EVMVerificationResult
|
|
Example::
auth = sign_erc3009_authorization(
private_key="0x...",
token="0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48",
chain_id=1,
authorizer="0xYourAddress",
recipient="0xRecipientAddress",
value=1_000_000,
valid_after=0,
valid_before=1_900_000_000,
domain_name="USD Coin",
domain_version="2",
)
result = verify_erc3009_eoa(
token=auth.token,
chain_id=auth.chain_id,
authorizer=auth.authorizer,
recipient=auth.recipient,
value=auth.value,
valid_after=auth.validAfter,
valid_before=auth.validBefore,
nonce=auth.nonce,
v=auth.signature.v,
r=auth.signature.r,
s=auth.signature.s,
domain_name="USD Coin",
domain_version="2",
)
assert result.is_valid
verify_permit2(*, owner: str, spender: str, token: str, amount: int, nonce: int, deadline: int, chain_id: int, v: int, r: str, s: str, permit2_address: str = '0x000000000022D473030F116dDEE9F6B43aC78BA3', owner_balance: Optional[int] = None, current_time: Optional[int] = None, w3=None) -> EVMVerificationResult
async
Verify a Permit2 permitTransferFrom authorization.
Performs checks in order, returning on the first failure:
- Address format --
owner,spender,token, andpermit2_addressmust be 0x-prefixed 42-character strings. - Deadline --
current_timemust be strictly less thandeadline. - Balance -- when
owner_balanceis supplied it must be>= amount. - Signature -- reconstructs the Permit2 EIP-712 struct hash and
verifies via EOA ECDSA recovery. If that fails and
w3is provided, falls back to an ERC-1271isValidSignaturecall.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
owner
|
str
|
Token owner address (signer of the permit). |
required |
spender
|
str
|
Address authorised to call |
required |
token
|
str
|
ERC-20 token contract address. |
required |
amount
|
int
|
Transfer amount in the token's smallest unit. |
required |
nonce
|
int
|
Permit2 integer nonce for |
required |
deadline
|
int
|
Unix timestamp after which the permit is invalid. |
required |
chain_id
|
int
|
EVM network ID. |
required |
v
|
int
|
ECDSA recovery ID (27 or 28). |
required |
r
|
str
|
Signature |
required |
s
|
str
|
Signature |
required |
permit2_address
|
str
|
Permit2 singleton contract address. |
'0x000000000022D473030F116dDEE9F6B43aC78BA3'
|
owner_balance
|
Optional[int]
|
Optional current token balance of |
None
|
current_time
|
Optional[int]
|
Optional Unix timestamp for deadline checks.
Defaults to |
None
|
w3
|
Optional |
None
|
Returns:
| Type | Description |
|---|---|
EVMVerificationResult
|
|
query_erc20_allowance(w3: AsyncWeb3, token_addr: str, owner: str, spender: str) -> int
async
Retrieves the amount of tokens that an owner allowed a spender to withdraw.
This function calls the 'allowance(address,address)' constant method of an ERC20 smart contract. It performs checksum address conversion and handles potential exceptions during the RPC call.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
w3
|
AsyncWeb3
|
The Web3 instance connected to the target blockchain. |
required |
token_addr
|
str
|
The contract address of the ERC20 token. |
required |
owner
|
str
|
The address of the token holder. |
required |
spender
|
str
|
The address authorized to spend the tokens. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
int |
int
|
The remaining allowance amount in the token's base units (e.g., wei). |
Raises:
| Type | Description |
|---|---|
ValueError
|
If any provided address is not a valid hex address. |
Web3Exception
|
If the contract call fails or the node returns an error. |
Exception
|
For any other unexpected errors during execution. |
verify_universal(*, v: int, r: str, s: str, chain_id: int, token: str, sender: str, receiver: str, amount: int, deadline: int, nonce: Union[int, str], scheme: Optional[Literal['erc3009', 'permit2']] = None, domain_name: Optional[str] = None, domain_version: str = '2', valid_after: int = 0, on_chain_nonce: Optional[str] = None, permit2_address: str = '0x000000000022D473030F116dDEE9F6B43aC78BA3', owner_balance: Optional[int] = None, current_time: Optional[int] = None, w3=None) -> EVMVerificationResult
async
Unified entry point for ERC-3009 and Permit2 signature verification.
Uses the same homogeneous parameter names as sign_universal:
+-----------+-------------------+-----------+ | Unified | ERC-3009 | Permit2 | +===========+===================+===========+ | sender | authorizer | owner | +-----------+-------------------+-----------+ | receiver | recipient | spender | +-----------+-------------------+-----------+ | amount | value | amount | +-----------+-------------------+-----------+ | deadline | valid_before | deadline | +-----------+-------------------+-----------+ | nonce | bytes32 hex / int | int | +-----------+-------------------+-----------+
Scheme selection
domain_nameprovided, orscheme="erc3009"→ ERC-3009domain_nameabsent, orscheme="permit2"→ Permit2
Parameters
v, r, s : ECDSA signature components.
chain_id : EVM network ID.
token : ERC-20 token contract address.
sender : Signer / token owner address.
receiver : Destination / authorised spender address.
amount : Transfer amount in the token's smallest unit.
deadline : Expiry Unix timestamp (valid_before for ERC-3009).
nonce : Replay-protection nonce — bytes32 hex string or int.
ERC-3009 uses str; Permit2 uses int;
both formats are accepted and normalised internally.
scheme : Optional explicit scheme override.
domain_name : EIP-712 domain name (required for ERC-3009).
domain_version : EIP-712 domain version; defaults to "2".
valid_after : ERC-3009 start timestamp; defaults to 0.
on_chain_nonce : ERC-3009 on-chain nonce for replay detection (optional).
permit2_address: Permit2 singleton contract address.
owner_balance : Optional token balance for a sufficiency check.
current_time : Optional Unix timestamp (defaults to int(time.time())).
w3 : Optional web3.Web3 instance for ERC-1271 fallback.
Returns
EVMVerificationResult.
Raises
ValueError
If ERC-3009 is selected but domain_name is not provided.
Examples
ERC-3009::
result = verify_universal(
v=auth.signature.v, r=auth.signature.r, s=auth.signature.s,
chain_id=1, token="0xA0b869...",
sender="0xFrom", receiver="0xTo",
amount=1_000_000, deadline=1_900_000_000,
nonce=auth.nonce, domain_name="USD Coin",
)
Permit2::
result = verify_universal(
v=sig.v, r=sig.r, s=sig.s,
chain_id=1, token="0xA0b869...",
sender="0xOwner", receiver="0xSpender",
amount=1_000_000, deadline=1_900_000_000,
nonce=sig.nonce,
)
EvmAssetConfig
Bases: BaseModel
Token asset configuration.
EvmChainInfo
Bases: BaseModel
Subset of ethereum-lists chain metadata we rely on.
This model is designed to be compatible with the JSON files in:
ethereum-lists/chains (eip155-
Note
The upstream payload contains many more fields; we keep only the ones needed by the dynamic chain config builder below.
EvmChainConfig
Bases: BaseModel
EVM blockchain network configuration.
EvmPublicRpcFromChainList
Utility class for fetching and selecting public RPC URLs from Chainlist.org.
This class provides methods to fetch chain data from Chainlist.org, build EvmChainList objects for specific chains, and pick public RPC URLs based on tracking preferences and protocol.
Attributes:
| Name | Type | Description |
|---|---|---|
_cached_data |
Optional[List[Dict[str, Any]]]
|
Optional cached chainlist data to avoid repeated network requests. |
clear() -> None
Clear the cached chainlist data.
This method removes any cached data fetched from Chainlist.org, forcing the next request to fetch fresh data from the network.
fetch_data_from_chainlist() -> List[Dict[str, Any]]
Fetch chainlist data from Chainlist.org.
Returns:
| Type | Description |
|---|---|
List[Dict[str, Any]]
|
List[Dict[str, Any]]: A list of chain entries from Chainlist.org. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the response format is unexpected (not a list). |
HTTPStatusError
|
If the HTTP request fails. |
RuntimeError
|
If the response is not valid JSON. |
get_specific_chain_public_rpcs(caip2: str) -> Optional[EvmChainList]
Build an EvmChainList object for a given CAIP-2 chain identifier.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
caip2
|
str
|
CAIP-2 chain identifier (e.g., 'eip155:1'). |
required |
Returns:
| Type | Description |
|---|---|
Optional[EvmChainList]
|
Optional[EvmChainList]: EvmChainList object if chain is found, None otherwise. |
pick_public_rpc(caip2: str, start_with: Literal['https://', 'wss://'] = 'https://', tracking_type: Optional[Literal['none', 'limited', 'yes']] = None) -> Optional[str]
Pick a public RPC URL from Chainlist.org for a given CAIP-2 chain ID.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
caip2
|
str
|
CAIP-2 chain identifier (e.g., 'eip155:1'). |
required |
start_with
|
Literal['https://', 'wss://']
|
Protocol prefix to filter by (defaults to "https://"). |
'https://'
|
tracking_type
|
Optional[Literal['none', 'limited', 'yes']]
|
Tracking preference to filter by (None for any). |
None
|
Returns:
| Type | Description |
|---|---|
Optional[str]
|
Optional[str]: Public RPC URL if found, None otherwise. |
EvmTokenListFromUniswap
Utility class for fetching and parsing ERC-20 token metadata from the Uniswap token list.
This class provides methods to fetch token list data from Uniswap's official token list, cache the results, and extract token contract addresses and decimals for specific chains.
Attributes:
| Name | Type | Description |
|---|---|---|
_cached_token_list |
Optional[List[Dict[str, Any]]]
|
Optional cached token list data to avoid repeated network requests. |
clear() -> None
Clear the cached token list data.
This method removes any cached data fetched from Uniswap's token list, forcing the next request to fetch fresh data from the network.
fetch_token_list() -> List[Dict[str, Any]]
Fetches the official Uniswap token list and extracts token metadata.
Returns:
| Type | Description |
|---|---|
List[Dict[str, Any]]
|
List[Dict[str, Any]]: A list of token metadata objects. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the response is empty or the 'tokens' key is missing. |
HTTPError
|
If the network request fails. |
get_token_address_and_decimals(caip2: str, symbol: str) -> Tuple[str, int]
Locates a token's contract address and decimals within the Uniswap token list.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
caip2
|
str
|
The CAIP-2 compliant chain identifier (e.g., 'eip155:1'). |
required |
symbol
|
str
|
The token ticker symbol (e.g., 'WETH', 'USDC'). |
required |
Returns:
| Type | Description |
|---|---|
Tuple[str, int]
|
Tuple[str, int]: A tuple containing the (address, decimals). |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the token is not found on the specified chain, or if the found token contains invalid/missing address or decimals data. |
EvmChainInfoFromEthereumLists
Utility class for fetching and parsing EVM chain metadata from the ethereum-lists repository.
The primary purpose of this class is to resolve Infura and Alchemy RPC endpoints (with API key placeholders) for a given CAIP-2 chain identifier, as well as to enumerate public RPC URLs that require no API key.
Attributes:
| Name | Type | Description |
|---|---|---|
_cached_chain_info |
Dict[str, EvmChainInfo]
|
Per-chain cache to avoid redundant network requests. |
clear() -> None
Clear the cached chain info data.
Removes all cached entries fetched from ethereum-lists, forcing the next call to retrieve fresh data from the network.
fetch_chain_info(caip2: str) -> EvmChainInfo
Fetch chain configuration data from the ethereum-lists repository.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
caip2
|
str
|
CAIP-2 chain identifier (e.g., 'eip155:1'). |
required |
Returns:
| Name | Type | Description |
|---|---|---|
EvmChainInfo |
EvmChainInfo
|
Validated chain specification object. |
Raises:
| Type | Description |
|---|---|
HTTPError
|
If the upstream file is not found or unreachable. |
TypeError
|
If the returned payload does not match the EvmChainInfo schema. |
get_infura_rpc_url(caip2: str, start_with: Literal['https:', 'wss:'] = 'https:') -> Optional[str]
Return the first Infura RPC URL with an API key placeholder for a given chain.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
caip2
|
str
|
CAIP-2 chain identifier (e.g., 'eip155:1'). |
required |
start_with (Literal["https
|
", "wss:"]): Protocol prefix to filter by. Defaults to "https:". |
required |
Returns:
| Type | Description |
|---|---|
Optional[str]
|
Optional[str]: The first matching Infura RPC URL, or None if not found. |
get_alchemy_rpc_url(caip2: str, start_with: Literal['https:', 'wss:'] = 'https:') -> Optional[str]
Return the first Alchemy RPC URL with an API key placeholder for a given chain.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
caip2
|
str
|
CAIP-2 chain identifier (e.g., 'eip155:1'). |
required |
start_with (Literal["https
|
", "wss:"]): Protocol prefix to filter by. Defaults to "https:". |
required |
Returns:
| Type | Description |
|---|---|
Optional[str]
|
Optional[str]: The first matching Alchemy RPC URL, or None if not found. |
get_public_rpc_urls(caip2: str) -> List[str]
Return all public RPC URLs (without API key placeholders) for a given chain.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
caip2
|
str
|
CAIP-2 chain identifier (e.g., 'eip155:1'). |
required |
Returns:
| Type | Description |
|---|---|
List[str]
|
List[str]: Public RPC URLs that require no API key. |
PublicRpcType
Bases: BaseModel
Configuration model for a single public RPC endpoint. Defines the connection URL and privacy tracking policies of the provider.
EvmChainList
Bases: BaseModel
Configuration model for an EVM-compatible blockchain network. Aggregates chain identification and available public RPC endpoints.
get_private_key_from_env() -> Optional[str]
Load EVM server private key from environment variables.
This function retrieves the private key that the EVM server adapter uses for signing transactions and initializing the server account.
Environment Variable
- evm_private_key: The server's EVM private key (0x-prefixed hex format)
Returns:
| Name | Type | Description |
|---|---|---|
str |
Optional[str]
|
Private key from environment, or None if not configured |
Note
The private key should be stored securely in environment variables and never committed to version control.
Example
In your .env file or environment setup:
export EVM_PRIVATE_KEY="0x1234567890abcdef..."
pk = get_private_key_from_env() if pk: adapter = EVMServerAdapter(private_key=pk)
get_rpc_key_from_env(env_variable_name: str = 'EVM_INFURA_KEY') -> Optional[str]
Load EVM infrastructure API key from environment variables.
This function retrieves the optional infrastructure API key used to construct premium RPC endpoints (e.g., Alchemy, Infura keys). If not provided, the adapter will fall back to public RPC nodes.
Environment Variable
- EVM_INFURA_KEY: Infrastructure provider API key (e.g., Alchemy/Infura key)
Returns:
| Name | Type | Description |
|---|---|---|
str |
Optional[str]
|
Infra key from environment, or None if not configured |
Note
If EVM_INFURA_KEY is not set or empty, public RPC endpoints will be used. Public endpoints may have rate limits, but are free and don't require configuration.
Example
In your .env file or environment setup:
export EVM_INFURA_KEY="xyz123abc456..."
infra_key = get_infra_key_from_env()
Will be used to construct premium RPC URLs like:
"https://eth-mainnet.g.alchemy.com/v2/xyz123abc456..."
amount_to_value(*, amount: float | int | str | Decimal, decimals: int) -> int
Convert a human-readable token amount into smallest-unit integer value.
This is the canonical conversion used by permit signing / verification / transactions.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
amount
|
float | int | str | Decimal
|
Human-readable amount (e.g. 1.23 for USDC). Accepts float/int/str/Decimal. |
required |
decimals
|
int
|
Token decimals (e.g. 6 for USDC). |
required |
Returns:
| Name | Type | Description |
|---|---|---|
int |
int
|
Smallest-unit integer value. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If inputs are invalid or the amount cannot be represented in smallest units. |
value_to_amount(*, value: int | str | Decimal, decimals: int) -> float
Convert a smallest-unit integer value into human-readable token amount.
This is the canonical conversion used for user-facing display / matching / requests.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
value
|
int | str | Decimal
|
Smallest-unit integer value (e.g. 1230000 for 1.23 USDC). Accepts int/str/Decimal. |
required |
decimals
|
int
|
Token decimals (e.g. 6 for USDC). |
required |
Returns:
| Name | Type | Description |
|---|---|---|
float |
float
|
Human-readable amount. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If inputs are invalid. |
parse_caip2_eip155_chain_id(caip2: str) -> int
Parses a CAIP-2 identifier (e.g., 'eip155:1' or 'eip155-1') into an integer chain ID.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
caip2
|
str
|
The CAIP-2 string to parse. |
required |
Returns:
| Name | Type | Description |
|---|---|---|
int |
int
|
The extracted EIP-155 chain ID. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If the input format is invalid, the prefix is missing, or the chain ID is not a positive integer. |
fetch_erc20_name_version_decimals(*, rpc_url: str, token_address: str) -> Tuple[Optional[str], str]
Fetch token name() and optional EIP-712 version() from the chain RPC.
Notes
version()is not part of ERC-20 and may not exist (defaults to "0").- If
name()cannot be resolved, returns None for name.
fetch_json(url: str, timeout: float = 10.0) -> Dict[str, Any]
Fetches JSON data from a URL and raises detailed exceptions on failure.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
url
|
str
|
The target URL to request. |
required |
timeout
|
float
|
Connection timeout in seconds. |
10.0
|
Returns:
| Type | Description |
|---|---|
Dict[str, Any]
|
Dict[str, Any]: The parsed JSON response. |
Raises:
| Type | Description |
|---|---|
HTTPStatusError
|
If the server returns a 4xx or 5xx status code. |
RequestError
|
If a network-level error occurs (DNS, Connection Refused). |
RuntimeError
|
If the response is not valid JSON. |
fetch_evm_chain_info(caip2: str) -> EvmChainInfo
Retrieves EVM chain configuration from the ethereum-lists repository.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
caip2
|
str
|
The CAIP-2 compliant chain identifier (e.g., 'eip155:1'). |
required |
Returns:
| Name | Type | Description |
|---|---|---|
EvmChainInfo |
EvmChainInfo
|
A validated data object containing chain specifications. |
Raises:
| Type | Description |
|---|---|
HTTPError
|
If the chain configuration file is not found or unreachable. |
TypeError
|
If the returned payload does not match the EvmChainInfo schema. |
parse_private_url(urls: List[str], start_with: Literal['https:', 'wss:'] = 'https:') -> List[str]
Filters a list of URLs based on a protocol prefix and the presence of API key placeholders.
This function identifies infrastructure URLs that contain environment variable placeholders (e.g., '${INFURA_API_KEY}') and match the specified starting prefix.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
urls
|
List[str]
|
A list of RPC or WebSocket URL strings. |
required |
start_with (Literal["https
|
", "wss:"]): The protocol prefix to filter by. Defaults to "https:". |
required |
Returns:
| Type | Description |
|---|---|
List[str]
|
List[str]: A list of URLs matching the prefix and containing a '${' placeholder. Returns an empty list if no matches are found. |
Raises:
| Type | Description |
|---|---|
TypeError
|
If 'urls' is not a list or 'start_with' is not a Literal["https:", "wss:"]. |
extract_endpoint_info(url_list: List[str], endpoint_type: Literal['Infura', 'Alchemy'] = 'Infura') -> List[str]
Filter URLs by endpoint type and extract API key placeholders.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
url_list
|
List[str]
|
List of RPC endpoint URLs containing API key placeholders like ${KEY}. |
required |
endpoint_type
|
Literal['Infura', 'Alchemy']
|
Type of endpoint to filter, e.g., "Infura" or "Alchemy". Case-insensitive. Defaults to "Infura". |
'Infura'
|
Returns:
| Type | Description |
|---|---|
List[str]
|
List[str]: List of matched URLs. |
get_balance_abi() -> List[Dict[str, Any]]
Get ABI for querying USDC token balance.
Returns:
| Type | Description |
|---|---|
List[Dict[str, Any]]
|
List[Dict[str, Any]]: ABI for balanceOf function |
Example
abi = get_balance_abi()
Use with web3.py: web3.eth.contract(address=token_address, abi=abi)
Call: contract.functions.balanceOf(address).call()
get_allowance_abi() -> List[Dict[str, Any]]
Get ABI for ERC20 allowance(owner, spender).
Returns:
| Type | Description |
|---|---|
List[Dict[str, Any]]
|
List[Dict[str, Any]]: ABI for ERC20 |
Example
abi = get_allowance_abi() contract = web3.eth.contract(address=token_address, abi=abi) allowance = contract.functions.allowance(owner, spender).call()
get_approve_abi() -> List[Dict[str, Any]]
Get ABI for ERC20 approve(spender, amount).
Returns:
| Type | Description |
|---|---|
List[Dict[str, Any]]
|
List[Dict[str, Any]]: ABI for ERC20 |
Example: abi = get_approve_abi() contract = web3.eth.contract(address=token_address, abi=abi) tx = contract.functions.approve(spender, amount).build_transaction({...})
get_erc3009_abi() -> List[Dict[str, Any]]
Get ABI for ERC-3009 transferWithAuthorization.
Used to call the on-chain transferWithAuthorization function directly on
a token contract that implements ERC-3009 (e.g. USDC, EURC). The caller
passes the signed authorization values (from, to, value, validAfter,
validBefore, nonce, v, r, s) — matching the fields of
:class:~x402_mock.adapters.evm.schemas.ERC3009Authorization.
Returns:
| Type | Description |
|---|---|
List[Dict[str, Any]]
|
List[Dict[str, Any]]: ABI containing the |
List[Dict[str, Any]]
|
function entry. |
Example::
abi = get_erc3009_abi()
contract = web3.eth.contract(address=token_address, abi=abi)
tx = contract.functions.transferWithAuthorization(
from_addr, to_addr, value,
valid_after, valid_before, nonce_bytes32,
v, r_bytes32, s_bytes32,
).build_transaction({...})
get_permit2_abi() -> List[Dict[str, Any]]
Get ABI for Uniswap Permit2 permitTransferFrom.
Used to call the canonical Permit2 singleton contract
(0x000000000022D473030F116dDEE9F6B43aC78BA3) to settle a
:class:~x402_mock.adapters.evm.schemas.Permit2Signature.
The function signature on-chain::
function permitTransferFrom(
PermitTransferFrom calldata permit,
SignatureTransferDetails calldata transferDetails,
address owner,
bytes calldata signature
) external
where PermitTransferFrom = { TokenPermissions permitted; uint256 nonce; uint256 deadline }
and TokenPermissions = { address token; uint256 amount },
and SignatureTransferDetails = { address to; uint256 requestedAmount }.
Returns:
| Type | Description |
|---|---|
List[Dict[str, Any]]
|
List[Dict[str, Any]]: ABI containing the |
List[Dict[str, Any]]
|
function entry with fully-resolved tuple components. |
Example::
abi = get_permit2_abi()
contract = web3.eth.contract(address=permit2_address, abi=abi)
tx = contract.functions.permitTransferFrom(
((token_addr, amount), nonce, deadline), # PermitTransferFrom
(to_addr, amount), # SignatureTransferDetails
owner_addr,
sig_bytes,
).build_transaction({...})
EVMRegistry
Bases: AdapterRegistry
payment_method_register(*, payment_component: EVMPaymentComponent, rpc_url_start_with: Literal['https:', 'wss:'] = 'https:', private_rpc_type: Literal['Infura', 'Alchemy'] = 'Infura') -> None
Register a payment method by processing and validating the payment component.
Performs the following steps in order: 1. If RPC URL is missing, resolves it via the preferred private RPC provider (Infura or Alchemy) or falls back to a public RPC URL. 2. If token address is missing, looks it up from the Uniswap token list. 3. If token name is missing, fetches name, version, and decimals on-chain. 4. Validates the completed payment component. 5. Appends the validated component to the registry.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
payment_component
|
EVMPaymentComponent
|
The EVM payment component to register. |
required |
rpc_url_start_with
|
Literal['https:', 'wss:']
|
Preferred RPC URL protocol ("https:" or "wss:"). |
'https:'
|
private_rpc_type
|
Literal['Infura', 'Alchemy']
|
Preferred private RPC provider ("Infura" or "Alchemy"). |
'Infura'
|
Raises:
| Type | Description |
|---|---|
ValueError
|
If payment_component is not an EVMPaymentComponent instance, or if the component fails validation after processing. |
get_rpc_url_by_caip2(caip2: str) -> str
classmethod
Retrieve the RPC URL for the payment component matching the given CAIP-2 identifier.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
caip2
|
str
|
The CAIP-2 chain identifier (e.g. "eip155:1") to look up. |
required |
Returns:
| Type | Description |
|---|---|
str
|
The RPC URL string associated with the matching payment component. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If no payment component with the given CAIP-2 is found, or if the matching component has an empty RPC URL. |
clear() -> None
Clear all cached data held by the internal fetcher instances.
Forces subsequent lookups to fetch fresh data from the network.
EIP712Domain
dataclass
EIP-712 domain separator. Used to prevent signature replay across domains.
TransferWithAuthorizationMessage
dataclass
Represents the message payload for EIP-3009 "TransferWithAuthorization".
This dataclass mirrors the typed fields required by the EIP-3009
structured data specification. Note that the EIP defines the field
name from which is a Python reserved word; this class uses
authorizer as the attribute name and maps it to from in
to_dict().
Attributes:
| Name | Type | Description |
|---|---|---|
authorizer |
str
|
Address of the account authorizing the transfer (maps to |
recipient |
str
|
Address receiving the tokens (maps to |
value |
int
|
Amount of tokens to transfer (uint256). |
validAfter |
int
|
Unix timestamp after which the authorization becomes valid. |
validBefore |
int
|
Unix timestamp before which the authorization expires. |
nonce |
str
|
A unique nonce (bytes32 hex string) preventing replay. |
to_dict() -> Dict[str, Any]
Return a dictionary representation compatible with EIP-712 signing.
The returned keys follow the names required by the EIP-3009 typed
definition (i.e. from, to, value, validAfter, validBefore, nonce).
ERC3009TypedData
dataclass
Container for ERC-3009 typed data usable with EIP-712 signing routines.
This class provides the to_dict() helper producing a dict compatible
with most signTypedData implementations: it includes types,
primaryType, domain and message entries.
Attributes:
| Name | Type | Description |
|---|---|---|
domain |
EIP712Domain
|
EIP712Domain instance describing the signing domain. |
message |
TransferWithAuthorizationMessage
|
TransferWithAuthorizationMessage instance carrying the payload. |
primary_type |
str
|
The primary EIP-712 type (defaults to "TransferWithAuthorization"). |
types |
Dict[str, List[Dict[str, str]]]
|
The typed definitions required by EIP-712 (automatically set). |
to_dict() -> Dict[str, Any]
Return a dictionary compatible with EIP-712 structured signing.
The returned structure follows the conventional layout consumed by EIP-712 signing libraries: { types, primaryType, domain, message }.
ERC1271ABI
dataclass
ABI definition for the ERC-1271 isValidSignature function.
Encodes the single function entry required to call
isValidSignature(bytes32 _hash, bytes _signature) returns (bytes4)
on any contract implementing the ERC-1271 standard.
Use to_dict() to obtain the raw ABI entry dict, or to_list()
to get the full ABI list accepted by web3.eth.contract.
to_dict() -> Dict[str, Any]
Return the isValidSignature ABI entry as a dict.
to_list() -> List[Dict[str, Any]]
Return the full ABI as a list compatible with web3.eth.contract.
ERC1271Signature
dataclass
Representation of an on-chain signature verification request for a contract implementing ERC-1271.
This dataclass is a simple container describing the minimal pieces of information needed to express a signature verification request against a contract that conforms to ERC-1271. It purposely does not implement any verification logic — it only represents data.
Attributes:
| Name | Type | Description |
|---|---|---|
contract |
str
|
Address of the ERC-1271 contract that will verify the signature. |
message_hash |
str
|
Hash of the message (bytes32 hex string) passed to |
signature |
str
|
The signature bytes (hex string, 0x-prefixed) to be checked. |
message |
Optional[str]
|
Optional raw message data that was signed (as hex string or utf-8 string). |
to_dict() -> Dict[str, Any]
Return a plain dict suitable for serialization or logging.
ERC6492Proof
dataclass
Represents a generic proof object produced under EIP-6492-style schemes.
This dataclass models a proof bundle that accompanies a signature or authorization and demonstrates how the signature or key material can be constructed or verified on-chain. The exact semantics of each field are intentionally left generic to avoid coupling this representation to any single proof scheme; consumers of the object can serialize and interpret the inner values as needed.
Attributes:
| Name | Type | Description |
|---|---|---|
proof_type |
str
|
A short identifier describing the proof scheme (e.g. "merkle", "delegation"). |
root |
str
|
Root digest associated with the proof (hex string). |
siblings |
Optional[List[str]]
|
Optional list of sibling hashes or intermediate nodes (hex strings). |
signer |
Optional[str]
|
Address or identifier of the signer this proof is associated with. |
aux_data |
Optional[Dict[str, Any]]
|
Optional free-form auxiliary data needed to interpret the proof. |
to_dict() -> Dict[str, Any]
Return a plain dict representation of the proof.
ERC6492GeneratedObject
dataclass
Container for an object produced when generating ERC-6492-style proofs.
This object ties a specific data payload (for example, a signature or an authorization) to a proof describing why or how that payload is considered valid under a particular on-chain verification scheme.
Attributes:
| Name | Type | Description |
|---|---|---|
subject |
str
|
The hex-encoded subject of the proof (e.g. signature, key, or payload digest). |
proof |
ERC6492Proof
|
An |
metadata |
Optional[Dict[str, Any]]
|
Optional dictionary for any additional, arbitrary metadata. |
to_dict() -> Dict[str, Any]
Return a plain dict suitable for serialization or transport.
Permit2TypedData
dataclass
EIP-712 typed-data container for a Permit2 permitTransferFrom authorization.
Encapsulates all runtime fields (chain, addresses, amounts, nonce, deadline)
together with the canonical Permit2 type definitions. to_dict() produces
a structure directly consumable by eth_account.sign_typed_data and
eth_signTypedData_v4.
The domain follows the canonical Permit2 convention: name="Permit2" with
no version field. Types embed both PermitTransferFrom and its nested
TokenPermissions sub-struct.
Attributes:
| Name | Type | Description |
|---|---|---|
chain_id |
int
|
EVM network ID (e.g. |
verifying_contract |
str
|
Permit2 singleton contract address. |
spender |
str
|
Address authorised to call |
token |
str
|
ERC-20 token contract address. |
amount |
int
|
Transfer amount in the token's smallest unit. |
nonce |
int
|
Permit2 nonce for the owner; consumed on first use. |
deadline |
int
|
Unix timestamp after which the permit is invalid. |
types |
Dict[str, List[Dict[str, str]]]
|
EIP-712 type schema (pre-populated with the Permit2
domain, |
to_dict() -> Dict[str, Any]
Return a dict compatible with EIP-712 structured signing.
The returned structure follows the conventional layout consumed by EIP-712 signing libraries: { types, primaryType, domain, message }.
Schemas
基础模式模型和类型系统
Schemas 模块定义了支撑整个 x402_mock 框架的基础类型系统和数据模型。它提供了符合 RFC8785 的 Pydantic 模型用于加密操作、确保跨区块链实现类型安全的抽象基类,以及标准化的 HTTP 协议消息格式。
主要特性: - RFC8785 合规性:用于确定性签名生成的规范 JSON 序列化 - 类型安全:基于 Pydantic 的验证,具有全面的类型提示 - 抽象基类:定义许可、签名、验证结果和确认的契约 - 协议消息:标准化的 HTTP 402 请求/响应负载模式 - 版本管理:协议版本协商和兼容性处理 - 区块链无关:所有区块链特定实现继承的基础模型
主要组件:
- CanonicalModel:符合 RFC8785 的具有确定性 JSON 序列化的基础模型
- 抽象类型:BasePermit、BaseSignature、BaseVerificationResult、BaseTransactionConfirmation
- HTTP 协议:ClientRequestHeader、Server402ResponsePayload、ClientTokenRequest、ServerTokenResponse
- 支付模型:定义支付需求的 BasePaymentComponent
- 状态枚举:VerificationStatus、TransactionStatus
- 版本处理:ProtocolVersion、SupportedVersions
目的: 作为类型基础,确保服务器、客户端、适配器和引擎组件之间一致的数据结构和验证。
schemas
CanonicalModel
Bases: BaseModel
RFC8785-compliant Pydantic base model with canonical JSON serialization.
This model ensures consistent, deterministic JSON representation suitable for cryptographic operations, signature verification, and hashing.
Features
- Automatic conversion of Pydantic objects, enums, and Decimals to standard types
- Deterministic key sorting in JSON output
- No extra whitespace for consistent hashing and signature verification
- RFC8785 compliance for canonical JSON representation
All schema models should inherit from this class to ensure consistent serialization across the system.
Example
class MyModel(CanonicalModel): name: str value: int
model = MyModel(name="test", value=123) canonical_json = model.to_canonical_json() # Guaranteed consistent format
to_canonical_json() -> str
Convert model to RFC8785-compliant canonical JSON string.
This method ensures that the JSON representation is: 1. Deterministically ordered (sorted keys) 2. Whitespace-minimal (compact format) 3. Suitable for cryptographic operations
The conversion process: 1. model_dump(mode="json") converts Pydantic objects, enums, and Decimals to standard Python types (str, int, float, etc.) 2. json.dumps with separators and sort_keys ensures RFC8785 compliance
Returns:
| Name | Type | Description |
|---|---|---|
str |
str
|
RFC8785-compliant JSON string with sorted keys and no extra whitespace. |
Example
model = MyModel(name="test", value=123) json_str = model.to_canonical_json()
Returns: '{"name":"test","value":123}'
to_dict() -> Dict[str, Any]
Convert model to dictionary representation.
Returns:
| Type | Description |
|---|---|
Dict[str, Any]
|
Dict[str, Any]: Dictionary with all model fields. |
BaseSignature
Bases: CanonicalModel, ABC
Abstract base class for blockchain signature components.
This class defines the interface that all blockchain-specific signature implementations must follow. Different blockchains use different signature formats (e.g., EIP2612 for EVM uses v/r/s, Solana uses 64-byte signature).
All concrete signature classes should inherit from this base class and implement the abstract methods to ensure consistent signature handling across the system.
Attributes:
| Name | Type | Description |
|---|---|---|
signature_type |
str
|
The type of signature (e.g., "EIP2612", "Solana") |
created_at |
datetime
|
Timestamp when the signature was created |
Methods:
| Name | Description |
|---|---|
validate_format |
Check if signature format is valid for the blockchain |
to_dict |
Convert signature to dictionary (inherited from CanonicalModel) |
validate_format() -> bool
Validate the signature format for the specific blockchain.
This method should check that all signature components are in valid format for the target blockchain. For example: - EIP2612: Check v is 27 or 28, r and s are 64 hex chars - Solana: Check signature is 64 bytes
Returns:
| Name | Type | Description |
|---|---|---|
bool |
bool
|
True if signature format is valid, False otherwise. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If signature format is invalid with descriptive message. |
BasePermit
Bases: CanonicalModel, ABC
Abstract base class for blockchain permit/approval mechanisms.
A permit is a signed message that authorizes a spender to transfer tokens on behalf of the token owner. Different blockchains implement permits differently (EIP2612 for EVM, etc.).
This base class provides the minimal common fields across all permit types. Blockchain-specific implementations should extend this class and add their specific fields such as owner, spender, token, value, and nonce.
Attributes:
| Name | Type | Description |
|---|---|---|
permit_type |
str
|
Type of permit (e.g., "EIP2612", "Solana") |
signature |
Optional[BaseSignature]
|
Signature components for permit authorization |
created_at |
datetime
|
Timestamp when permit was created |
Methods:
| Name | Description |
|---|---|
validate_structure |
Validate permit structure (blockchain-specific) |
validate_structure() -> bool
Validate the permit structure and required fields for the blockchain.
Should check that all required fields are present and in valid format for the specific blockchain permit type.
Returns:
| Name | Type | Description |
|---|---|---|
bool |
bool
|
True if permit structure is valid, False otherwise. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If permit structure is invalid with descriptive message. |
BasePaymentComponent
Bases: CanonicalModel, ABC
Abstract base class for payment requirement specifications.
Payment components define what payment is expected: the amount, currency, and any additional payment-related constraints or metadata. Blockchain-specific implementations (e.g., EVM, SVM) should extend this class and add their specific fields such as token addresses.
Attributes:
| Name | Type | Description |
|---|---|---|
payment_type |
str
|
Type of payment (e.g., "evm", "svm") |
amount |
float
|
Payment amount for human readability |
currency |
str
|
Currency code (e.g., "USD", "ETH") |
metadata |
Dict[str, Any]
|
Additional payment-related metadata |
created_at |
datetime
|
Timestamp when payment component was created |
validate_payment() -> bool
Validate the payment specification.
Should check that payment type, amount, and token are valid and compatible with the system requirements.
Returns:
| Name | Type | Description |
|---|---|---|
bool |
bool
|
True if payment specification is valid, False otherwise. |
Raises:
| Type | Description |
|---|---|
ValueError
|
If payment specification is invalid with descriptive message. |
VerificationStatus
Bases: str, Enum
Enumeration of possible verification result statuses.
Attributes:
| Name | Type | Description |
|---|---|---|
SUCCESS |
Permit signature is valid and verification passed |
|
INVALID_SIGNATURE |
Signature is invalid or signer mismatch |
|
EXPIRED |
Permit deadline has passed |
|
INSUFFICIENT_ALLOWANCE |
Authorized amount is insufficient |
|
INSUFFICIENT_BALANCE |
Token balance insufficient for transaction |
|
REPLAY_ATTACK |
Nonce indicates potential replay attack |
|
BLOCKCHAIN_ERROR |
Error querying blockchain state |
|
UNKNOWN_ERROR |
Unexpected error during verification |
BaseVerificationResult
Bases: CanonicalModel, ABC
Abstract base class for permit signature verification results.
This class encapsulates the result of verifying a permit signature and checking permit validity on-chain. It provides a standard interface for reporting verification status, success/failure details, and diagnostic information.
Attributes:
| Name | Type | Description |
|---|---|---|
verification_type |
str
|
Type of verification (e.g., "evm", "svm") |
status |
VerificationStatus
|
Verification result status (VerificationStatus enum) |
is_valid |
bool
|
Boolean indicating if verification was successful |
message |
str
|
Human-readable status message |
error_details |
Optional[Dict[str, Any]]
|
Detailed error information if verification failed |
verified_at |
datetime
|
Timestamp when verification was performed |
Methods:
| Name | Description |
|---|---|
is_success |
Check if verification was successful |
get_error_message |
Get formatted error message |
is_success() -> bool
Check if verification was successful.
Convenience method to check if the verification passed.
Returns:
| Name | Type | Description |
|---|---|---|
bool |
bool
|
True if verification was successful, False otherwise. |
Example
result = await adapter.verify_signature(permit, payment) if result.is_success(): # Proceed with transaction else: # Handle verification failure
get_error_message() -> Optional[str]
Get formatted error message from verification result.
Returns a human-readable error message explaining why verification failed. Returns None if verification was successful.
Returns:
| Type | Description |
|---|---|
Optional[str]
|
Optional[str]: Error message if verification failed, None if successful. |
Example
if not result.is_success(): error_msg = result.get_error_message() print(f"Verification failed: {error_msg}")
TransactionStatus
Bases: str, Enum
Enumeration of possible transaction execution statuses.
Attributes:
| Name | Type | Description |
|---|---|---|
SUCCESS |
Transaction executed successfully on-chain |
|
FAILED |
Transaction reverted or failed on-chain |
|
PENDING |
Transaction is pending confirmation |
|
INSUFFICIENT_GAS |
Transaction failed due to insufficient gas |
|
TIMEOUT |
Transaction confirmation timed out |
|
NETWORK_ERROR |
Network error during transaction submission |
|
INVALID_TRANSACTION |
Transaction is malformed or invalid |
|
UNKNOWN_ERROR |
Unexpected error during transaction execution |
BaseTransactionConfirmation
Bases: CanonicalModel, ABC
Abstract base class for blockchain transaction confirmation/receipt data.
This class captures the result of executing a transaction on-chain, including execution status, timing, and confirmation information.
Attributes:
| Name | Type | Description |
|---|---|---|
confirmation_type |
str
|
Type of confirmation (e.g., "evm", "svm") |
status |
TransactionStatus
|
Transaction execution status (TransactionStatus enum) |
execution_time |
Optional[float]
|
Time taken to confirm transaction (in seconds) |
confirmations |
int
|
Number of block confirmations |
error_message |
Optional[str]
|
Error message if transaction failed |
logs |
Optional[List[Dict[str, Any]]]
|
Optional transaction logs/events |
created_at |
datetime
|
Timestamp when confirmation was recorded |
Methods:
| Name | Description |
|---|---|
is_success |
Check if transaction executed successfully |
get_confirmation_status |
Get human-readable confirmation status |
is_success() -> bool
Check if transaction executed successfully on-chain.
Returns True if the transaction was executed without errors and achieved the intended state change on the blockchain.
Returns:
| Name | Type | Description |
|---|---|---|
bool |
bool
|
True if transaction succeeded, False if failed or pending. |
Example
confirmation = await adapter.send_transaction(permit) if confirmation.is_success(): print(f"Transaction confirmed: {confirmation.tx_hash}") else: print(f"Transaction failed: {confirmation.error_message}")
get_confirmation_status() -> str
Get human-readable confirmation status message.
Returns:
| Name | Type | Description |
|---|---|---|
str |
str
|
Human-readable status message describing transaction state. |
Example
status_msg = confirmation.get_confirmation_status()
May return: "Transaction confirmed with 10 confirmations"
ClientRequestHeader
Bases: BaseModel
HTTP request headers sent by client.
Attributes:
| Name | Type | Description |
|---|---|---|
content_type |
str
|
MIME type of request body (default: application/json). |
authorization |
Optional[str]
|
Optional bearer token for authenticated requests. |
ServerPaymentScheme
Bases: BaseModel
Payment scheme configuration for client payment authorization.
Describes the specific payment requirements and supported payment methods that the client must fulfill to access the protected resource.
Attributes:
| Name | Type | Description |
|---|---|---|
payment_components |
List[PaymentComponentTypes]
|
List of supported payment options/requirements. |
protocol_version |
str
|
Version of the payment protocol being used. |
Server402ResponsePayload
Bases: BaseModel
Server response payload for 402 Payment Required status.
This is returned by the server when client attempts to access a protected resource without valid authorization. It instructs the client where to submit payment and what payment methods are accepted.
Attributes:
| Name | Type | Description |
|---|---|---|
access_token_endpoint |
str
|
URL endpoint for POST request to obtain access token. |
payment_scheme |
ServerPaymentScheme
|
Payment requirements and accepted payment methods. |
payment_instruction |
Optional[str]
|
Optional instruction explaining payment process and endpoint usage. |
ClientTokenRequest
Bases: BaseModel
Client request to exchange permit for access token.
The client sends this request to the server's access token endpoint after generating a valid permit that authorizes the payment. The server validates the permit and returns an access token if the permit is valid.
This request should be sent as POST with JSON body.
Attributes:
| Name | Type | Description |
|---|---|---|
version |
ProtocalVersion
|
Protocol version of the permit. |
permit |
PermitTypes
|
Signed permit authorizing the payment. |
ServerTokenResponse
Bases: BaseModel
Server response containing access token and metadata.
Returned when server successfully verifies and accepts a client's permit. The access token is used by client to access the protected resource.
Attributes:
| Name | Type | Description |
|---|---|---|
access_token |
str
|
Bearer token for authenticating subsequent requests. |
token_type |
str
|
Type of token (typically "Bearer"). |
expires_in |
Optional[int]
|
Token lifetime in seconds. None means token never expires. |
metadata |
Dict[str, Any]
|
Additional metadata about the token or authorization. |
Engine
事件驱动执行引擎
Engine 模块实现了一个复杂的事件驱动架构,用于编排支付协议工作流。它提供了一个类型化的事件系统,带有事件总线,允许订阅者挂钩到支付生命周期、监控执行流程、捕获错误并在关键执行点自定义行为。
主要特性:
- 类型化事件系统:强类型事件,代表支付处理的每个阶段
- 事件总线:用于解耦事件处理的发布-订阅模式
- 钩子订阅:使用 add_hook() 订阅特定事件类型的处理程序
- 事件链执行:具有状态转换的顺序事件处理
- 全面的事件:请求初始化、令牌交换、验证、结算、错误
- 依赖注入:业务逻辑与基础设施依赖的清晰分离
- 异常层次结构:用于细粒度错误处理的丰富异常类型
- 异步原生:为异步执行构建,支持 asyncio
关键事件类型:
- RequestInitEvent:带有可选授权令牌的初始请求
- RequestTokenEvent:用于令牌交换的支付许可提交
- Http402PaymentEvent:带有支付方案的支付要求响应
- VerifySuccessEvent / VerifyFailedEvent:签名验证结果
- SettleSuccessEvent / SettleFailedEvent:链上结算结果
- TokenIssuedEvent:成功的访问令牌生成
- AuthorizationSuccessEvent:成功的请求授权
主要组件:
- EventBus:具有订阅者管理的中央事件分发器
- EventChain:编排事件序列执行
- Dependencies:共享基础设施的不可变容器
- 类型化事件:所有事件都继承自 BaseEvent
- 自定义异常:针对不同失败场景的详细错误类型
使用模式:
开发人员可以使用 event_bus.subscribe(EventType, handler) 订阅自定义处理程序到事件,以拦截事件、记录交易、触发 webhook 或在支付流程的任何点实现自定义业务逻辑。
engine
BaseEvent
Bases: ABC
Base class for all events in the system.
__repr__() -> str
abstractmethod
String representation of the event.
RequestInitEvent
RequestTokenEvent
AuthorizationSuccessEvent
Http402PaymentEvent
VerifySuccessEvent
VerifyFailedEvent
SettleFailedEvent
SettleSuccessEvent
Dependencies
dataclass
Container for infrastructure dependencies (read-only).
EventBus
Event dispatcher for publishing and subscribing to events.
__init__() -> None
Initialize with empty subscribers and hooks.
subscribe(event_class: type[BaseEvent], handler: EventHandlerFunc) -> None
Register an async handler for the given event class. Multiple handlers can be subscribed to the same event type and run in parallel.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
event_class
|
type[BaseEvent]
|
The event class to subscribe to. |
required |
handler
|
EventHandlerFunc
|
The async handler function to call when the event is published. |
required |
Raises:
| Type | Description |
|---|---|
TypeError
|
If handler is not a coroutine function. |
hook(event_class: type[BaseEvent], hook_func: EventHookFunc) -> None
Register a hook for the given event class. Hooks are executed before subscribers when the event is dispatched.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
event_class
|
type[BaseEvent]
|
The event class to hook into. |
required |
hook_func
|
EventHookFunc
|
The hook function to call when the event is published. |
required |
dispatch(event: BaseEvent, deps: Dependencies) -> AsyncGenerator[Optional[BaseEvent], None]
async
Dispatch an event to all registered hooks and subscribers. Hooks run first (synchronously in order), then all subscribers run in parallel.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
event
|
BaseEvent
|
The event to dispatch. |
required |
deps
|
Dependencies
|
Dependencies container with injected services. |
required |
Yields:
| Type | Description |
|---|---|
AsyncGenerator[Optional[BaseEvent], None]
|
Results from all subscribers as they complete. Yields nothing if no subscribers are registered. |
TokenIssuedEvent
BreakEvent
EventChain
Executes event-driven workflows by chaining event handler results.
Supports early return mechanism: when an event matching early_return_on is encountered, execute() returns that event immediately while remaining processing continues in background.
__init__(event_bus: EventBus, deps: Dependencies) -> None
Initialize event chain executor.
Parameters:
| Name | Type | Description | Default |
|---|---|---|---|
event_bus
|
EventBus
|
The event bus to dispatch events through. |
required |
deps
|
Dependencies
|
Dependencies container to pass to handlers. |
required |
execute(initial_event: BaseEvent) -> AsyncGenerator[BaseEvent, None]
async
Execute event chain starting from initial event.
Returns:
| Type | Description |
|---|---|
AsyncGenerator[BaseEvent, None]
|
Yields events encountered during chain execution. |
Note
When early return is triggered, subsequent events continue processing asynchronously in background without blocking the return.
BaseException
Bases: Exception
Root exception class for all project-specific exceptions.
All custom exceptions should inherit from this class to enable unified exception handling and centralized error processing.
AuthenticationError
Bases: BaseException
Raised when authentication fails or credentials are invalid.
This includes scenarios such as: - Invalid or expired access tokens - Missing authentication headers - Authentication signature verification failure
PaymentMethodError
Bases: BaseException
Raised when a payment method is invalid or unsupported.
This includes scenarios such as: - Unsupported payment type (on-chain vs off-chain mismatch) - Missing required payment method configuration - Payment method not registered on server
PaymentSignatureError
Bases: BaseException
Raised when payment signature generation or processing fails.
This includes scenarios such as: - Signature format validation failure - Private key access issues - Signature encoding errors
TokenError
Bases: BaseException
Base exception for token-related errors.
Parent class for all token validation and management errors.
TokenExpiredError
Bases: TokenError
Raised when a token has expired and is no longer valid.
Attributes:
| Name | Type | Description |
|---|---|---|
expiration_time |
When the token expired |
InvalidTokenError
Bases: TokenError
Raised when a token is invalid or malformed.
This includes scenarios such as: - Corrupted token data - Invalid token signature - Unsupported token version
TokenNotFoundError
Bases: TokenError
Raised when a requested token cannot be found.
Typically occurs in token lookup or retrieval operations.
PaymentVerificationError
Bases: BaseException
Base exception for payment verification failures.
Parent class for all errors that occur during payment validation.
SignatureVerificationError
Bases: PaymentVerificationError
Raised when permit signature verification fails.
This includes scenarios such as: - Invalid ECDSA signature - Signature from wrong address - Tampered signature data - Signer address mismatch
Attributes:
| Name | Type | Description |
|---|---|---|
permit_type |
Type of permit being verified |
|
signer |
Expected signer address |
|
recovered |
Actually recovered address from signature |
PermitExpiredError
Bases: PaymentVerificationError
Raised when a permit has expired and can no longer be executed.
The permit deadline has passed on the blockchain.
Attributes:
| Name | Type | Description |
|---|---|---|
deadline |
The expired permit deadline |
|
current_time |
Current block timestamp |
PermitNonceError
Bases: PaymentVerificationError
Raised when permit nonce is invalid or already used.
This protects against replay attacks by ensuring each permit has a unique nonce that increments with each use.
Attributes:
| Name | Type | Description |
|---|---|---|
expected_nonce |
Nonce expected on-chain |
|
provided_nonce |
Nonce in the permit |
InsufficientFundsError
Bases: PaymentVerificationError
Raised when account balance is insufficient for the payment.
This includes scenarios such as: - Token balance less than permit amount - Insufficient gas for transaction execution
Attributes:
| Name | Type | Description |
|---|---|---|
required |
Amount required |
|
available |
Amount available |
ConfigurationError
Bases: BaseException
Raised when configuration is missing or invalid.
This includes scenarios such as: - Missing required configuration keys - Invalid configuration values - RPC URL unreachable - Unsupported network configuration
InvalidTransition
Bases: Exception
Raised when an invalid state transition occurs in event processing.
This indicates that the payment state machine received an event that is not valid for the current state.
Attributes:
| Name | Type | Description |
|---|---|---|
current_state |
Current payment/transaction state |
|
event_type |
Event that triggered the transition |
|
message |
Description of why transition is invalid |
BlockchainInteractionError
Bases: BaseException
Raised when blockchain interaction (RPC call) fails.
This includes scenarios such as: - RPC call timeout - Network connectivity issues - Invalid contract address - Contract call revert
Attributes:
| Name | Type | Description |
|---|---|---|
rpc_method |
RPC method that was called (e.g., 'eth_call') |
|
reason |
Error reason from blockchain node |
TransactionExecutionError
Bases: BlockchainInteractionError
Raised when blockchain transaction execution fails.
This includes scenarios such as: - Transaction reverted on-chain - Out of gas - Invalid transaction parameters - Nonce conflicts
Attributes:
| Name | Type | Description |
|---|---|---|
tx_hash |
Transaction hash if available |
|
revert_reason |
Reason transaction was reverted |
MCP
模型上下文协议(MCP)工具集成
MCP 模块将 x402-mock 的支付能力以 Model Context Protocol 工具的形式暴露出来,使 LLM Agent(如 GitHub Copilot、Claude、GPT 等)能够直接调用,无需手动编写支付流程代码即可完成完整的 402 支付交互。
安装依赖:MCP 支持作为可选依赖提供,使用以下命令安装:
uv sync --extra mcp
主要特性:
- 零代码支付:LLM Agent 通过自然语言指令即可触发完整的 402 支付流程
- 角色分离:Client 角色(签名 + 请求)与 Server 角色(验证 + 结算)各自独立注册工具
- stdio 传输:基于标准 I/O 的进程通信,与所有主流 MCP 宿主(VS Code、Claude Desktop 等)兼容
- 自动支付重试:source_request 工具封装了完整的 402 拦截 → 签名 → 重试流程
- 类型安全:工具参数和返回值均基于 Pydantic 类型系统
主要组件:
- FacilitorTools:核心类,按角色向 FastMCP 实例注册工具
FacilitorTools
构造函数参数
| 参数 | 类型 | 必填 | 说明 |
|---|---|---|---|
adapter_hub |
AdapterHub |
是 | 已配置支付方式的适配器中枢 |
mcp |
FastMCP |
是 | FastMCP 服务器实例,工具将注册到此实例上 |
client_role |
bool |
否 | True 注册客户端工具,False(默认)注册服务端工具 |
示例
from mcp.server.fastmcp import FastMCP
from x402_mock.adapters.adapters_hub import AdapterHub
from x402_mock.mcp.facilitor_tools import FacilitorTools
hub = AdapterHub(evm_private_key="0x...")
mcp = FastMCP("x402")
# 服务端角色:注册 verify_and_settle 工具
FacilitorTools(adapter_hub=hub, mcp=mcp, client_role=False)
# 客户端角色:注册 signature + source_request 工具
# FacilitorTools(adapter_hub=hub, mcp=mcp, client_role=True)
mcp.run()
MCP 工具一览
source_request(客户端)
访问受 402 保护的资源,自动完成签名与支付重试,是 LLM Agent 最常用的入口工具。
参数
| 参数 | 类型 | 默认值 | 说明 |
|---|---|---|---|
url |
str |
必填 | 目标资源 URL |
method |
str |
"GET" |
HTTP 方法 |
headers |
dict \| None |
None |
额外请求头 |
timeout |
float |
30.0 |
请求超时时间(秒) |
返回值
{
"status_code": 200, # HTTP 状态码
"headers": { ... }, # 响应头字典
"body": "..." # 响应体字符串
}
内部流程
发送请求
└─> 收到 402?
├─ 是 → 解析支付组件 → 签名 permit → 携带令牌重试 → 返回最终响应
└─ 否 → 直接返回响应
signature(客户端)
根据服务端 402 响应中的支付组件列表,从本地注册的支付方式中匹配并生成签名的 permit,供后续提交至 /token 端点。
参数
| 参数 | 类型 | 说明 |
|---|---|---|
list_components |
List[PaymentComponentTypes] |
服务端返回的支付组件列表 |
返回值:已签名的 PermitTypes 对象(EVMTokenPermit 或其他链的对应类型)
verify_and_settle(服务端)
验证支付 permit 签名并在链上完成结算,一步完成,无需单独的令牌发放流程。
参数
| 参数 | 类型 | 说明 |
|---|---|---|
permit |
PermitTypes |
已签名的支付 permit |
返回值(三选一)
| 返回类型 | 含义 |
|---|---|
SettleSuccessEvent |
permit 有效,链上结算已确认 |
SettleFailedEvent |
permit 有效,但链上结算失败 |
VerifyFailedEvent |
permit 签名无效 |
事件流
RequestTokenEvent
└─> 验证签名
├─ 成功 → VerifySuccessEvent → 链上结算
│ ├─ 成功 → SettleSuccessEvent
│ └─ 失败 → SettleFailedEvent
└─ 失败 → VerifyFailedEvent
MCP 配置示例(VS Code / GitHub Copilot)
将以下内容保存为项目根目录下的 .vscode/mcp.json,即可在 VS Code 的 Copilot Agent 模式中直接使用 x402-mock 支付工具:
{
"servers": {
"X402-Mock-Server": {
"type": "stdio",
"command": "uv",
"args": ["run", "example/mcp_server_example.py"],
"env": {
"X402_TOKEN_KEY": "dev-secret-change-me",
"EVM_PRIVATE_KEY": "your_private_key_here",
"EVM_INFURA_KEY": "your_infura_key_here"
}
},
"X402-Mock-Client": {
"type": "stdio",
"command": "uv",
"args": ["run", "example/mcp_client_example.py"],
"env": {
"EVM_PRIVATE_KEY": "your_private_key_here",
"EVM_INFURA_KEY": "your_infura_key_here"
}
}
}
}
mcp
FacilitorTools
FacilitorTools exposes exactly two MCP tools:
signature— generate a signed permit from a list of remote payment components (client-side signing).verify_and_settle— verify a permit signature and settle on-chain in a single workflow (no token issuance).