前提条件

在使用此技能之前，请确保：

已安装Python 3.10+
x402-avm包 可用：pip install "x402-avm[avm,fastapi]"
Algorand私钥 作为Base64编码的64字节密钥（32字节种子+32字节公钥）在AVM_PRIVATE_KEY环境变量中可用
**通过AlgoNode（默认）或通过ALGOD_SERVER / ALGOD_TOKEN自定义节点访问Algod节点

核心工作流程：Facilitator支付生命周期

Facilitator位于资源服务器和区块链之间，处理支付验证和结算：

资源服务器              Facilitator                   区块链
     |                           |                             |
     |--- /verify (payload) ---->|                             |
     |                           |-- simulate_group() -------->|
     |                           |<-- 模拟结果 -------|
     |<-- {isValid: true} -------|                             |
     |                           |                             |
     |--- /settle (payload) ---->|                             |
     |                           |-- sign_group() ------------>|
     |                           |-- send_group() ------------>|
     |                           |<-- txid --------------------|
     |                           |-- confirm_transaction() --->|
     |<-- {success, txid} -------|                             |

如何进行

第1步：安装依赖项

pip install "x402-avm[avm,fastapi]"

这将安装py-algorand-sdk (algosdk)，FastAPI，uvicorn和x402-avm SDK。

第2步：实现FacilitatorAvmSigner

FacilitatorAvmSigner协议定义了六个方法。使用algosdk实现所有方法：

import os
import base64
from algosdk import encoding, transaction
from algosdk.v2client import algod
from x402.mechanisms.avm.constants import NETWORK_CONFIGS


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: bytes, fee_payer: str, network: str) -> bytes:
        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)

第3步：将签名者注册到x402Facilitator

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

signer = AlgorandFacilitatorSigner(
    private_key_b64=os.environ["AVM_PRIVATE_KEY"],
    algod_url=os.environ.get("ALGOD_SERVER", ""),
    algod_token=os.environ.get("ALGOD_TOKEN", ""),
)

facilitator = x402Facilitator()
register_exact_avm_facilitator(
    facilitator,
    signer,
    networks=[ALGORAND_TESTNET_CAIP2, ALGORAND_MAINNET_CAIP2],
)

第4步：创建FastAPI端点

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI(title="x402-avm Facilitator Service")

@app.get("/supported")
async def supported():
    return facilitator.get_supported_networks()

@app.post("/verify")
async def verify(request: Request):
    body = await request.json()
    try:
        result = await facilitator.verify(
            body["paymentPayload"], body["paymentRequirements"]
        )
        return result
    except Exception as e:
        return JSONResponse(status_code=400, content={"error": str(e)})

@app.post("/settle")
async def settle(request: Request):
    body = await request.json()
    try:
        result = await facilitator.settle(
            body["paymentPayload"], body["paymentRequirements"]
        )
        return result
    except Exception as e:
        return JSONResponse(status_code=400, content={"error": str(e)})

第5步：运行服务

AVM_PRIVATE_KEY="your-base64-key" uvicorn facilitator_service:app --port 4000

重要规则/指南

algosdk编码边界 – msgpack_decode()期望一个base64字符串，而不是原始字节。在解码前总是转换：base64.b64encode(raw_bytes).decode("utf-8")。msgpack_encode()返回一个base64字符串；用base64.b64decode()转换回来。
私钥格式 – Transaction.sign()期望一个base64编码的64字节密钥字符串，而不是原始字节。
send_group模式 – 使用send_raw_transaction(base64.b64encode(b"".join(group_bytes)))避免不必要的解码/重新编码开销。
simulate_group模式 – 用SignedTransaction(txn, None)包装未签名的交易，并使用allow_empty_signatures=True。
地址派生 – 使用encoding.encode_address(secret_key[32:])从64字节密钥派生地址。
一流的AVM – 无条件注册AVM，绝不在if (env.AVM_*)守卫之后。
网络常量 – 从SDK导入ALGORAND_TESTNET_CAIP2和ALGORAND_MAINNET_CAIP2；不要在应用程序代码中硬编码创世哈希。

生命周期钩子

x402Facilitator支持生命周期钩子，用于自定义逻辑：

钩子	调用时间	用例
`on_before_verify`	支付验证前	日志记录，速率限制
`on_after_verify`	验证完成后	分析，缓存
`on_before_settle`	提交结算前	最终验证
`on_after_settle`	结算完成后	通知，收据

常见错误/故障排除

错误	原因	解决方案
`msgpack_decode expects string`	向`msgpack_decode`传递原始字节	用`base64.b64encode(raw).decode()`包装
`Invalid key length`	错误的私钥格式	确保`AVM_PRIVATE_KEY`是base64编码的64字节
`Simulation failed`	交易将在链上失败	检查余额，资产选择加入，费用金额
`send_raw_transaction expects bytes-like`	错误的参数类型	使用`base64.b64encode(b"".join(group_bytes))`
`Transaction not confirmed`	节点超时或网络问题	增加`rounds`参数或检查节点连接
`send_transactions asserts NOT Transaction`	向`send_transactions`传递未签名的`Transaction`	使用`send_raw_transaction`代替

参考资料/进一步阅读

REFERENCE.md – FacilitatorAvmSigner协议的详细API参考
EXAMPLES.md – 完整的代码示例
Facilitator Example
x402-avm AVM Documentation