x402-avm Python Core and AVM Examples
ClientAvmSigner Protocol
from x402.mechanisms.avm.signer import ClientAvmSigner
class ClientAvmSigner(Protocol):
@property
def address(self) -> str:
"""58-character Algorand address."""
...
def sign_transactions(
self,
unsigned_txns: list[bytes],
indexes_to_sign: list[int],
) -> list[bytes | None]:
"""Sign specified transactions in a group.
Args:
unsigned_txns: List of unsigned transaction bytes (msgpack encoded).
indexes_to_sign: Indexes of transactions this signer should sign.
Returns:
List parallel to unsigned_txns, with signed bytes at
indexes_to_sign and None elsewhere.
"""
...
Client Signer Implementation
import os
import base64
from x402.mechanisms.avm.signer import ClientAvmSigner
from algosdk import encoding
class PrivateKeySigner:
"""
ClientAvmSigner implementation using a base64-encoded private key.
Key format: 64 bytes = [32-byte seed][32-byte public key]
algosdk expects base64-encoded 64-byte key for signing.
"""
def __init__(self, private_key_b64: str):
self._secret_key = base64.b64decode(private_key_b64)
if len(self._secret_key) != 64:
raise ValueError(
f"Invalid key length: expected 64, got {len(self._secret_key)}"
)
self._address = encoding.encode_address(self._secret_key[32:])
# algosdk.Transaction.sign() expects base64 string
self._signing_key = base64.b64encode(self._secret_key).decode()
@property
def address(self) -> str:
return self._address
def sign_transactions(
self,
unsigned_txns: list[bytes],
indexes_to_sign: list[int],
) -> list[bytes | None]:
result: list[bytes | None] = []
for i, txn_bytes in enumerate(unsigned_txns):
if i not in indexes_to_sign:
result.append(None)
continue
# IMPORTANT: algosdk.encoding.msgpack_decode expects base64 string
b64_txn = base64.b64encode(txn_bytes).decode("utf-8")
txn_obj = encoding.msgpack_decode(b64_txn)
# Sign: Transaction.sign() expects base64 private key string
signed_txn = txn_obj.sign(self._signing_key)
# IMPORTANT: algosdk.encoding.msgpack_encode returns base64 string
signed_b64 = encoding.msgpack_encode(signed_txn)
signed_bytes = base64.b64decode(signed_b64)
result.append(signed_bytes)
return result
# Usage:
signer = PrivateKeySigner(os.environ["AVM_PRIVATE_KEY"])
print(f"Signer address: {signer.address}")
FacilitatorAvmSigner Protocol
from x402.mechanisms.avm.signer import FacilitatorAvmSigner
class FacilitatorAvmSigner(Protocol):
def get_addresses(self) -> list[str]:
"""Get all managed fee payer addresses."""
...
def sign_transaction(
self, txn_bytes: bytes, fee_payer: str, network: str,
) -> bytes:
"""Sign a single transaction with the fee payer's key."""
...
def sign_group(
self,
group_bytes: list[bytes],
fee_payer: str,
indexes_to_sign: list[int],
network: str,
) -> list[bytes]:
"""Sign specified transactions in a group."""
...
def simulate_group(
self, group_bytes: list[bytes], network: str,
) -> None:
"""Simulate a transaction group (raises on failure)."""
...
def send_group(
self, group_bytes: list[bytes], network: str,
) -> str:
"""Send a transaction group, returns txid."""
...
def confirm_transaction(
self, txid: str, network: str, rounds: int = 4,
) -> None:
"""Wait for transaction confirmation."""
...
Facilitator Signer Implementation
import base64
from x402.mechanisms.avm.signer import FacilitatorAvmSigner
from x402.mechanisms.avm.constants import NETWORK_CONFIGS
from algosdk import encoding, transaction
from algosdk.v2client import algod
class AlgorandFacilitatorSigner:
def __init__(self, private_key_b64: str, algod_url: str = "", algod_token: str = ""):
self._secret_key = base64.b64decode(private_key_b64)
self._address = encoding.encode_address(self._secret_key[32:])
self._signing_key = base64.b64encode(self._secret_key).decode()
self._clients: dict[str, algod.AlgodClient] = {}
if algod_url:
self._default_client = algod.AlgodClient(algod_token, algod_url)
else:
self._default_client = None
def _get_client(self, network: str) -> algod.AlgodClient:
if network not in self._clients:
if self._default_client:
self._clients[network] = self._default_client
else:
config = NETWORK_CONFIGS.get(network, {})
url = config.get("algod_url", "https://testnet-api.algonode.cloud")
self._clients[network] = algod.AlgodClient("", url)
return self._clients[network]
def get_addresses(self) -> list[str]:
return [self._address]
def sign_transaction(self, txn_bytes, fee_payer, network):
b64 = base64.b64encode(txn_bytes).decode("utf-8")
txn_obj = encoding.msgpack_decode(b64)
signed = txn_obj.sign(self._signing_key)
return base64.b64decode(encoding.msgpack_encode(signed))
def sign_group(self, group_bytes, fee_payer, indexes_to_sign, network):
result = list(group_bytes)
for i in indexes_to_sign:
result[i] = self.sign_transaction(group_bytes[i], fee_payer, network)
return result
def simulate_group(self, group_bytes, network):
client = self._get_client(network)
stxns = []
for txn_bytes in group_bytes:
b64 = base64.b64encode(txn_bytes).decode("utf-8")
obj = encoding.msgpack_decode(b64)
if isinstance(obj, transaction.SignedTransaction):
stxns.append(obj)
else:
stxns.append(transaction.SignedTransaction(obj, None))
req = transaction.SimulateRequest(
txn_groups=[transaction.SimulateRequestTransactionGroup(txns=stxns)],
allow_empty_signatures=True,
)
result = client.simulate_raw_transactions(req)
for group in result.get("txn-groups", []):
if group.get("failure-message"):
raise Exception(f"Simulation failed: {group['failure-message']}")
def send_group(self, group_bytes, network):
client = self._get_client(network)
return client.send_raw_transaction(base64.b64encode(b"".join(group_bytes)))
def confirm_transaction(self, txid, network, rounds=4):
client = self._get_client(network)
transaction.wait_for_confirmation(client, txid, rounds)
Constants
Network Identifiers
from x402.mechanisms.avm.constants import (
ALGORAND_MAINNET_CAIP2, # "algorand:wGHE2Pwdvd7S12BL5FaOP20EGYesN73ktiC1qzkkit8="
ALGORAND_TESTNET_CAIP2, # "algorand:SGO1GKSzyE7IEPItTxCByw9x8FmnrCDexi9/cOUJOiI="
# CAIP-2 values as of x402-avm 2.0.2; note the TypeScript @x402/avm package >=2.20.0
# uses the 32-char form "algorand:SGO1GKSzyE7IEPItTxCByw9x8FmnrCDe" -- a TS server/client
# and a Python facilitator (or vice-versa) will not match until x402-avm adopts the same form.
SUPPORTED_NETWORKS, # [MAINNET_CAIP2, TESTNET_CAIP2]
MAINNET_GENESIS_HASH,
TESTNET_GENESIS_HASH,
V1_NETWORKS,
V1_TO_V2_NETWORK_MAP,
V2_TO_V1_NETWORK_MAP,
)
USDC Configuration
from x402.mechanisms.avm.constants import (
USDC_MAINNET_ASA_ID, # 31566704
USDC_TESTNET_ASA_ID, # 10458941
DEFAULT_DECIMALS, # 6
NETWORK_CONFIGS,
)
testnet_config = NETWORK_CONFIGS[ALGORAND_TESTNET_CAIP2]
usdc_info = testnet_config["default_asset"]
# {"asa_id": 10458941, "name": "USDC", "decimals": 6}
Algod Endpoints
from x402.mechanisms.avm.constants import (
MAINNET_ALGOD_URL,
TESTNET_ALGOD_URL,
FALLBACK_ALGOD_MAINNET, # "https://mainnet-api.algonode.cloud"
FALLBACK_ALGOD_TESTNET, # "https://testnet-api.algonode.cloud"
)
Transaction Limits
from x402.mechanisms.avm.constants import (
MAX_GROUP_SIZE, # 16
MIN_TXN_FEE, # 1000
)
Address Validation
from x402.mechanisms.avm.constants import AVM_ADDRESS_REGEX
import re
is_valid = bool(re.match(AVM_ADDRESS_REGEX, some_address))
Utility Functions
Address Validation
from x402.mechanisms.avm.utils import is_valid_address
is_valid_address("AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA")
# => True
is_valid_address("invalid")
# => False
Amount Conversion
from x402.mechanisms.avm.utils import to_atomic_amount, from_atomic_amount
to_atomic_amount(1.50) # => 1500000
to_atomic_amount(0.10) # => 100000
to_atomic_amount(0.000001) # => 1
from_atomic_amount(1500000) # => 1.5
from_atomic_amount(100000) # => 0.1
from_atomic_amount(1) # => 1e-06
Transaction Encoding/Decoding
from x402.mechanisms.avm.utils import (
decode_transaction_bytes,
decode_base64_transaction,
decode_payment_group,
encode_transaction_group,
)
# Decode raw bytes
info = decode_transaction_bytes(raw_txn_bytes)
print(info.type) # "axfer" or "pay"
print(info.sender) # Algorand address
print(info.fee) # in microAlgos
print(info.is_signed) # True/False
print(info.asset_amount) # for asset transfers
print(info.receiver) # for payments
# Decode from base64 string
info = decode_base64_transaction(base64_txn_string)
# Decode a full payment group
group_info = decode_payment_group(
payment_group=["base64txn1...", "base64txn2..."],
payment_index=0,
)
print(group_info.transactions)
print(group_info.group_id)
print(group_info.total_fee)
print(group_info.has_fee_payer)
# Encode to base64
encoded = encode_transaction_group([raw_bytes_1, raw_bytes_2])
Network Utilities
from x402.mechanisms.avm.utils import (
normalize_network,
is_valid_network,
get_network_config,
get_usdc_asa_id,
get_genesis_hash,
network_from_genesis_hash,
)
normalize_network("algorand-testnet")
# => "algorand:SGO1GKSzyE7IEPItTxCByw9x8FmnrCDexi9/cOUJOiI=" (== ALGORAND_TESTNET_CAIP2, as of x402-avm 2.0.2;
# TypeScript @x402/avm >=2.20.0 uses the 32-char form "algorand:SGO1GKSzyE7IEPItTxCByw9x8FmnrCDe" -- a TS
# server/client and a Python facilitator (or vice-versa) will not match until x402-avm adopts the same form)
is_valid_network("algorand-testnet") # True
is_valid_network("unknown-network") # False
config = get_network_config("algorand-testnet")
# {"algod_url": "...", "genesis_hash": "...", "default_asset": {...}}
get_usdc_asa_id("algorand-testnet") # 10458941
Security Validation
from x402.mechanisms.avm.utils import (
validate_no_security_risks,
validate_fee_payer_transaction,
is_blocked_transaction_type,
)
info = decode_transaction_bytes(txn_bytes)
error_code = validate_no_security_risks(info)
if error_code:
raise ValueError(f"Security risk: {error_code}")
error_code = validate_fee_payer_transaction(info, expected_fee_payer="ABC...")
if error_code:
raise ValueError(f"Invalid fee payer: {error_code}")
is_blocked_transaction_type("keyreg") # True
is_blocked_transaction_type("axfer") # False
Transaction Group Creation
Simple Payment Group
import base64
from algosdk import transaction, encoding
from algosdk.v2client import algod
from x402.mechanisms.avm.constants import (
ALGORAND_TESTNET_CAIP2,
USDC_TESTNET_ASA_ID,
NETWORK_CONFIGS,
)
def create_simple_payment(sender, receiver, amount):
config = NETWORK_CONFIGS[ALGORAND_TESTNET_CAIP2]
client = algod.AlgodClient("", config["algod_url"])
params = client.suggested_params()
txn = transaction.AssetTransferTxn(
sender=sender, sp=params, receiver=receiver,
amt=amount, index=USDC_TESTNET_ASA_ID,
)
return [base64.b64decode(encoding.msgpack_encode(txn))]
Fee-Abstracted Payment Group
import base64
from algosdk import transaction, encoding
from algosdk.v2client import algod
from x402.mechanisms.avm.constants import (
ALGORAND_TESTNET_CAIP2,
USDC_TESTNET_ASA_ID,
MIN_TXN_FEE,
NETWORK_CONFIGS,
)
from x402.mechanisms.avm.utils import encode_transaction_group
def create_fee_abstracted_payment(sender, receiver, fee_payer, amount):
config = NETWORK_CONFIGS[ALGORAND_TESTNET_CAIP2]
client = algod.AlgodClient("", config["algod_url"])
params = client.suggested_params()
# Transaction 0: USDC transfer (fee = 0)
payment_params = transaction.SuggestedParams(
fee=0, first=params.first, last=params.last,
gh=params.gh, gen=params.gen, flat_fee=True,
)
payment_txn = transaction.AssetTransferTxn(
sender=sender, sp=payment_params, receiver=receiver,
amt=amount, index=USDC_TESTNET_ASA_ID,
)
# Transaction 1: Fee payer (self-payment, covers both fees)
fee_params = transaction.SuggestedParams(
fee=MIN_TXN_FEE * 2, first=params.first, last=params.last,
gh=params.gh, gen=params.gen, flat_fee=True,
)
fee_payer_txn = transaction.PaymentTxn(
sender=fee_payer, sp=fee_params,
receiver=fee_payer, amt=0,
)
# Assign group ID
gid = transaction.calculate_group_id([payment_txn, fee_payer_txn])
payment_txn.group = gid
fee_payer_txn.group = gid
payment_bytes = base64.b64decode(encoding.msgpack_encode(payment_txn))
fee_payer_bytes = base64.b64decode(encoding.msgpack_encode(fee_payer_txn))
return {
"paymentGroup": encode_transaction_group([payment_bytes, fee_payer_bytes]),
"paymentIndex": 0,
"rawBytes": [payment_bytes, fee_payer_bytes],
}
ExactAvmScheme Registration
Client Registration
from x402 import x402Client
from x402.mechanisms.avm.exact import register_exact_avm_client
client = x402Client()
register_exact_avm_client(client, signer)
Server Registration
from x402 import x402ResourceServer
from x402.mechanisms.avm.exact import register_exact_avm_server
server = x402ResourceServer(facilitator_url="https://facilitator.example.com")
register_exact_avm_server(server)
Facilitator Registration
from x402 import x402Facilitator
from x402.mechanisms.avm.exact import register_exact_avm_facilitator
from x402.mechanisms.avm import ALGORAND_TESTNET_CAIP2, ALGORAND_MAINNET_CAIP2
facilitator = x402Facilitator()
# Single network
register_exact_avm_facilitator(
facilitator, signer, networks=[ALGORAND_TESTNET_CAIP2]
)
# Multiple networks
register_exact_avm_facilitator(
facilitator, signer, networks=[ALGORAND_TESTNET_CAIP2, ALGORAND_MAINNET_CAIP2]
)
Client-Side Fee Abstraction
from x402 import x402Client
from x402.mechanisms.avm.exact import register_exact_avm_client
signer = MyClientSigner(os.environ["AVM_PRIVATE_KEY"])
client = x402Client()
register_exact_avm_client(client, signer)
# Fee abstraction is automatic when PaymentRequirements include feePayer info
response = await client.fetch("https://api.example.com/paid-resource")
algosdk Encoding Boundary Patterns
import base64
from algosdk import encoding
# Raw bytes -> algosdk object (DECODE)
raw_bytes: bytes = ...
b64_string = base64.b64encode(raw_bytes).decode("utf-8")
txn_obj = encoding.msgpack_decode(b64_string)
# algosdk object -> raw bytes (ENCODE)
b64_string = encoding.msgpack_encode(txn_obj)
raw_bytes = base64.b64decode(b64_string)
