API Reference
Comprehensive API documentation for x402_mock module.
Servers
HTTP 402 Payment Protocol Server Implementation
The Servers module provides a FastAPI-based server framework for implementing HTTP 402 Payment Required protocol. It offers an event-driven architecture that encapsulates all payment collection logic, allowing payment receivers to integrate cryptocurrency payment acceptance with minimal configuration.
Key Features:
- FastAPI Integration: Extended FastAPI application with built-in payment endpoint routes
- Token Management: Secure HMAC-signed access token generation and verification
- Event-Driven Architecture: Subscribe to payment lifecycle events (request, verification, settlement)
- Multi-Chain Support: Register multiple payment methods across different blockchain networks
- Auto-Settlement: Optional automatic on-chain settlement after successful verification
- Security Utilities: Private key generation, token signing, and environment key management
- Modern EVM Signing:
- USDC: ERC-3009 (transferWithAuthorization)
- Generic ERC20: Permit2 (permitTransferFrom)
Main Components:
- Http402Server: Main server class extending FastAPI with payment protocol support
- Security helpers: 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 Payment Client Middleware
The Clients module provides an intelligent HTTP client that transparently handles HTTP 402 Payment Required responses. It extends httpx.AsyncClient to automatically intercept payment challenges, generate signed payment permits, exchange them for access tokens, and retry the original request—all without requiring explicit user intervention.
Key Features: - Transparent Payment Handling: Automatically processes 402 responses without manual intervention - httpx Compatibility: Fully compatible drop-in replacement for httpx.AsyncClient - Offline Signature Auto-Signing: Generates chain/token-specific offline authorizations (ERC-3009 / Permit2) using registered payment methods - Token Exchange: Automatically exchanges permits for access tokens at server endpoints - Request Retry: Seamlessly retries original requests with obtained authorization - Multi-Chain Support: Register payment capabilities across different blockchain networks
Main Components:
- Http402Client: Extended async HTTP client with automatic payment flow handling
Usage Pattern: 1. Initialize client and register payment methods 2. Make standard HTTP requests to protected resources 3. Client automatically handles 402 challenges and obtains access 4. Receive successful responses transparently
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
Unified Blockchain Adapter Interface
The Adapters module provides a unified abstraction layer that bridges differences between various blockchain platforms (EVM, Solana, etc.). It implements a plugin-based architecture with automatic blockchain type detection, enabling consistent payment permit signing, signature verification, and on-chain settlement operations across heterogeneous blockchain ecosystems.
Key Features: - Blockchain Abstraction: Unified interface for EVM, SVM (Solana), and other blockchain platforms - Automatic Type Detection: Identifies blockchain type from chain identifiers (CAIP-2 format) - Signature Operations: Generate and verify blockchain-specific cryptographic signatures - Authorization Validation: Verify authorization authenticity, expiration, nonce, and on-chain conditions - Transaction Settlement: Execute on-chain transfers with confirmation tracking (ERC-3009 / Permit2 on EVM) - Balance Queries: Query token balances and allowances across different chains - Extensible Architecture: Factory pattern enables easy addition of new blockchain adapters
Main Components:
- AdapterHub: Central gateway routing operations to appropriate blockchain adapters
- AdapterFactory: Abstract base class defining adapter interface contracts
- PaymentRegistry: Manages payment method registration and retrieval
- Platform-specific adapters: EVMAdapter (Ethereum/EVM chains), SVM adapter (coming soon)
Architecture Pattern: Uses the Adapter pattern combined with Factory pattern to provide a consistent API while delegating to blockchain-specific implementations under the hood.
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
Ethereum Virtual Machine (EVM) Blockchain Adapter
The EVM module provides a specialized adapter implementation for EVM-compatible blockchains. It includes a complete toolkit for handling on-chain payment authorizations, signature verification, transaction settlement, and chain configuration management.
Key Features:
- ERC-3009 Support: Offline transferWithAuthorization signing for USDC and other compatible tokens
- Permit2 Support: Offline permitTransferFrom authorization for generic ERC-20 tokens
- Multi-Chain Configuration: Unified chain configuration and asset information management
- Smart Contract Interaction: Complete ABI definitions for ERC-20, ERC-3009, and Permit2
- Signature Verification: Both on-chain and off-chain signature verification mechanisms
- Configuration Utilities: Tools to fetch chain info and token lists from external sources
Main Components:
- EVMAdapter: Main EVM blockchain adapter class
- EVMRegistry: EVM payment method registry
- EVMECDSASignature, EVMTokenPermit, ERC3009Authorization, Permit2Signature: Signature and authorization data structures
- EVMVerificationResult, EVMTransactionConfirmation: Verification and transaction result models
Configuration Utilities:
EvmPublicRpcFromChainList
Fetches public RPC endpoints in real-time from Chainlist.org. Supports filtering by protocol (https / wss) and privacy level (none / limited), making it easy to find a usable, API-Key-free RPC for any EVM chain.
from x402_mock.adapters.evm import EvmPublicRpcFromChainList
rpc = EvmPublicRpcFromChainList()
# Get any available public HTTPS RPC
print(rpc.pick_public_rpc("eip155:1"))
# Only nodes with no privacy tracking
print(rpc.pick_public_rpc("eip155:8453", tracking_type="none"))
EvmTokenListFromUniswap
Queries any token's contract address and decimals from the Uniswap official token list. Results are automatically cached to avoid redundant network requests.
from x402_mock.adapters.evm import EvmTokenListFromUniswap
tokens = EvmTokenListFromUniswap()
# Look up USDC contract address and decimals on Ethereum mainnet
address, decimals = tokens.get_token_address_and_decimals("eip155:1", "USDC")
print(address, decimals)
EvmChainInfoFromEthereumLists
Fetches authoritative chain metadata from the ethereum-lists repository. Primarily used to resolve Infura / Alchemy RPC templates with API Key placeholders, and to enumerate public RPC endpoints that require no key.
from x402_mock.adapters.evm import EvmChainInfoFromEthereumLists
chain = EvmChainInfoFromEthereumLists()
# Get Infura / Alchemy RPC templates (with {API_KEY} placeholder)
print(chain.get_infura_rpc_url("eip155:1"))
print(chain.get_alchemy_rpc_url("eip155:1"))
# List all public endpoints requiring no API Key
print(chain.get_public_rpc_urls("eip155:137"))
Other Utility Functions:
get_private_key_from_env(): Load the EVM server private key from environment variablesget_rpc_key_from_env(): Load the EVM infrastructure API key from environment variablesamount_to_value()/value_to_amount(): Convert between human-readable token amounts and on-chain smallest unitsparse_caip2_eip155_chain_id(): Parse a CAIP-2 identifier into an integer chain IDfetch_erc20_name_version_decimals(): Read token name, version, and decimals from on-chain 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
Base Schema Models and Type System
The Schemas module defines the foundational type system and data models that underpin the entire x402_mock framework. It provides RFC8785-compliant Pydantic models for cryptographic operations, abstract base classes ensuring type safety across blockchain implementations, and standardized HTTP protocol message formats.
Key Features: - RFC8785 Compliance: Canonical JSON serialization for deterministic signature generation - Type Safety: Pydantic-based validation with comprehensive type hints - Abstract Base Classes: Define contracts for permits, signatures, verification results, and confirmations - Protocol Messages: Standardized HTTP 402 request/response payload schemas - Version Management: Protocol version negotiation and compatibility handling - Blockchain Agnostic: Base models inherited by all blockchain-specific implementations
Main Components:
- CanonicalModel: RFC8785-compliant base model with deterministic JSON serialization
- Abstract types: BasePermit, BaseSignature, BaseVerificationResult, BaseTransactionConfirmation
- HTTP protocol: ClientRequestHeader, Server402ResponsePayload, ClientTokenRequest, ServerTokenResponse
- Payment models: BasePaymentComponent defining payment requirements
- Status enums: VerificationStatus, TransactionStatus
- Version handling: ProtocolVersion, SupportedVersions
Purpose: Serves as the type foundation ensuring consistent data structures and validations across servers, clients, adapters, and engine components.
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
Event-Driven Execution Engine
The Engine module implements a sophisticated event-driven architecture for orchestrating payment protocol workflows. It provides a typed event system with an event bus that allows subscribers to hook into the payment lifecycle, monitor execution flow, capture errors, and customize behavior at critical execution points.
Key Features:
- Typed Event System: Strongly-typed events representing each stage of payment processing
- Event Bus: Publish-subscribe pattern for decoupled event handling
- Hook Subscription: Use add_hook() to subscribe handlers to specific event types
- Event Chain Execution: Sequential event processing with state transitions
- Comprehensive Events: Request initialization, token exchange, verification, settlement, errors
- Dependency Injection: Clean separation of business logic from infrastructure dependencies
- Exception Hierarchy: Rich exception types for granular error handling
- Async-Native: Built for asynchronous execution with asyncio support
Key Event Types:
- RequestInitEvent: Initial request with optional authorization token
- RequestTokenEvent: Payment permit submission for token exchange
- Http402PaymentEvent: Payment required response with payment schemes
- VerifySuccessEvent / VerifyFailedEvent: Signature verification results
- SettleSuccessEvent / SettleFailedEvent: On-chain settlement outcomes
- TokenIssuedEvent: Successful access token generation
- AuthorizationSuccessEvent: Successful request authorization
Main Components:
- EventBus: Central event dispatcher with subscriber management
- EventChain: Orchestrates event sequence execution
- Dependencies: Immutable container for shared infrastructure
- Typed events: All events inherit from BaseEvent
- Custom exceptions: Detailed error types for different failure scenarios
Usage Pattern:
Developers can subscribe custom handlers to events using event_bus.subscribe(EventType, handler) to intercept events, log transactions, trigger webhooks, or implement custom business logic at any point in the payment flow.
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
Model Context Protocol (MCP) Tool Integration
The MCP module exposes x402-mock's payment capabilities as Model Context Protocol tools, enabling LLM Agents (such as GitHub Copilot, Claude, GPT, etc.) to call them directly and complete the full 402 payment interaction without writing any payment flow code.
Install dependencies: MCP support is provided as an optional extra. Install it with:
uv sync --extra mcp
Key Features:
- Zero-Code Payments: LLM Agents trigger the complete 402 payment flow via natural language instructions
- Role Separation: Client-role tools (sign + request) and Server-role tools (verify + settle) are registered independently
- stdio Transport: Process communication over standard I/O, compatible with all mainstream MCP hosts (VS Code, Claude Desktop, etc.)
- Automatic Payment Retry: The source_request tool encapsulates the full 402 intercept → sign → retry flow
- Type Safety: Tool arguments and return values are based on the Pydantic type system
Main Components:
- FacilitorTools: Core class that registers tools onto a FastMCP instance according to the configured role
FacilitorTools
Constructor Parameters
| Parameter | Type | Required | Description |
|---|---|---|---|
adapter_hub |
AdapterHub |
Yes | Adapter hub with payment methods already configured |
mcp |
FastMCP |
Yes | FastMCP server instance to register tools onto |
client_role |
bool |
No | True registers client-side tools; False (default) registers server-side tools |
Example
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")
# Server role: register verify_and_settle tool
FacilitorTools(adapter_hub=hub, mcp=mcp, client_role=False)
# Client role: register signature + source_request tools
# FacilitorTools(adapter_hub=hub, mcp=mcp, client_role=True)
mcp.run()
MCP Tool Reference
source_request (Client)
Access a 402-protected resource with automatic signing and payment retry. This is the primary entry-point tool for LLM Agents.
Parameters
| Parameter | Type | Default | Description |
|---|---|---|---|
url |
str |
required | Target resource URL |
method |
str |
"GET" |
HTTP method |
headers |
dict \| None |
None |
Additional request headers |
timeout |
float |
30.0 |
Request timeout in seconds |
Return Value
{
"status_code": 200, # HTTP status code
"headers": { ... }, # Response headers dict
"body": "..." # Response body string
}
Internal Flow
Send request
└─> 402 received?
├─ Yes → Parse payment components → Sign permit → Retry with token → Return final response
└─ No → Return response directly
signature (Client)
Matches a compatible local payment method against the server's 402 payment component list and generates a signed permit, ready to be submitted to the /token endpoint.
Parameters
| Parameter | Type | Description |
|---|---|---|
list_components |
List[PaymentComponentTypes] |
Payment component list returned by the server in the 402 response |
Returns: A signed PermitTypes object (EVMTokenPermit or the equivalent for other chains)
verify_and_settle (Server)
Verifies a payment permit signature and settles on-chain in a single step, with no separate token issuance flow required.
Parameters
| Parameter | Type | Description |
|---|---|---|
permit |
PermitTypes |
Signed payment permit |
Return Value (one of three)
| Return Type | Meaning |
|---|---|
SettleSuccessEvent |
Permit valid; on-chain settlement confirmed |
SettleFailedEvent |
Permit valid; on-chain settlement failed |
VerifyFailedEvent |
Permit signature invalid |
Event Flow
RequestTokenEvent
└─> Verify signature
├─ Success → VerifySuccessEvent → On-chain settlement
│ ├─ Success → SettleSuccessEvent
│ └─ Failure → SettleFailedEvent
└─ Failure → VerifyFailedEvent
MCP Configuration Example (VS Code / GitHub Copilot)
Save the following as .vscode/mcp.json in your project root to use x402-mock payment tools directly in VS Code's Copilot Agent mode:
{
"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).