跳转至

快速开始

x402-mock 是一个基于 HTTP 402 状态码的支付协议实现,支持 EVM 区块链上的代币支付。本指南将帮助您快速上手。

安装

本项目使用 uv 作为包管理工具。

uv add x402-mock
uv sync

环境配置

在项目根目录创建 .env 文件,配置您的私钥和 RPC 服务密钥:

# 必需:EVM 私钥(用于签名和收款)
EVM_PRIVATE_KEY=your_private_key_here

# 可选:Infura 或 Alchemy 的 API 密钥(用于访问区块链网络)
EVM_INFURA_KEY=your_infura_key_here
EVM_ALCHEMY_KEY=your_alchemy_key_here

核心概念

支付流程概述

x402-mock 实现了职责分离的支付流程,类似于电影院的售票和验票系统:

  1. Server(收款方):提供服务并接受支付,类似于电影院
  2. Client(付款方):请求服务并完成支付,类似于观众
  3. 支付流程
  4. Client 请求受保护的资源
  5. Server 验证 Client 的访问令牌(类似验票)
  6. 如果令牌无效,返回 402 状态码 + 支付信息(类似指引去售票处)
  7. Client 根据支付信息完成签名支付(类似买票)
  8. Client 获取访问令牌后重新请求资源(类似持票入场)

状态码 402 的职责分离

HTTP 402 "Payment Required" 状态码在本项目中实现了职责分离: - Server 端:只负责验证访问令牌的有效性,不处理支付逻辑 - 支付验证:由独立的 /token 端点处理,接收支付签名并发放访问令牌 - Client 端:自动处理 402 响应,完成支付流程后重试请求

这种设计使得支付逻辑与业务逻辑解耦,提高了系统的可维护性和安全性。

Server(收款方)

Server 是提供服务并接受支付的一方。主要职责包括: 1. 定义接受的支付方式(链、网络、代币) 2. 验证支付签名的真实性和有效性 3. 完成链上转账结算 4. 发放访问令牌

创建 Server 实例

from x402_mock.servers import Http402Server, create_private_key
from x402_mock.adapters.evm.schemas import EVMPaymentComponent

# 生成访问令牌签名密钥
token_key = create_private_key()

# 创建 Server 实例(继承自 FastAPI)
app = Http402Server(
    token_key=token_key,      # 访问令牌签名密钥
    token_expires_in=300      # 令牌有效期(秒)
)

Http402Server 参数说明

参数 类型 必填 说明
token_key str 访问令牌签名密钥,可使用 create_private_key() 生成
token_expires_in int 访问令牌有效期(秒),默认 3600
enable_auto_settlement bool 是否自动结算支付,默认 True
token_endpoint str 令牌交换端点路径,默认 "/token"

添加支付方式

# 添加 EVM 支付方式
app.add_payment_method(
    EVMPaymentComponent(
        amount=0.5,          # 支付金额(人类可读单位)
        currency="USDC",     # 代币符号
        caip2="eip155:11155111",  # CAIP-2 链标识符
        token="0x1c7D4B196Cb0C7B01d743Fbc6116a902379C7238"  # 代币合约地址
    )
)

EVMPaymentComponent 参数说明

参数 类型 必填 说明
amount float 支付金额(人类可读单位,如 0.5 USDC)
currency str 代币符号(如 "USDC", "USDT", "ETH")
caip2 str CAIP-2 链标识符(格式:eip155:<chain_id>
token str 建议 代币合约地址(0x 开头,42 字符)
pay_to str 收款地址,默认使用环境变量私钥对应的地址
rpc_url str RPC 节点 URL,默认使用公共节点或根据环境变量自动配置
token_name str 代币名称,自动获取时可省略
token_decimals int/str 代币精度,自动获取时可省略
token_version int/str 代币版本,自动获取时可省略

注意tokentoken_nametoken_decimalstoken_version 如未提供,系统会根据 caip2currency 自动查询。

保护 API 端点

使用 @app.payment_required 装饰器保护需要支付的端点:

@app.get("/api/protected-data")
@app.payment_required
async def get_protected_data(payload):
    """需要支付才能访问的端点"""
    return {
        "message": "Payment verified successfully",
        "user_address": payload["address"]
    }

事件处理

Server 提供了事件系统,您可以监听支付流程中的各种事件:

from x402_mock.engine.events import SettleSuccessEvent

@app.hook(SettleSuccessEvent)
async def on_settle_success(event, deps):
    """支付成功时的处理逻辑"""
    print(f"✅ 支付成功: {event.settlement_result}")
    # 可以在这里记录日志、发送通知等

可用的事件类型: - RequestInitEvent: 请求初始化 - RequestTokenEvent: 令牌请求 - TokenIssuedEvent: 令牌发放 - VerifyFailedEvent: 验证失败 - AuthorizationSuccessEvent: 授权成功 - Http402PaymentEvent: 需要支付 - SettleSuccessEvent: 结算成功

Client(付款方)

Client 是请求服务并完成支付的一方。主要职责包括: 1. 注册支持的支付方式 2. 自动处理 402 响应 3. 生成支付签名 4. 交换签名获取访问令牌

创建 Client 实例

from x402_mock.clients.http_client import Http402Client
from x402_mock.adapters.evm.schemas import EVMPaymentComponent

async with Http402Client() as client:
    # 注册支付方式
    client.add_payment_method(
        EVMPaymentComponent(
            caip2="eip155:11155111",
            amount=0.8,          # 最大支付金额限制
            currency="USDC",
            token="0x1c7D4B196Cb0C7B01d743Fbc6116a902379C7238"
        )
    )

    # 发送请求(自动处理 402 支付流程)
    response = await client.get("http://localhost:8000/api/protected-data")
    print(response.json())

Client 端 EVMPaymentComponent 的特殊说明

在 Client 端,amount 参数表示最大支付金额限制。如果 Server 要求的金额超过此限制,Client 将拒绝签名支付。

支付流程自动化

Http402Client 继承自 httpx.AsyncClient,完全兼容其所有方法。当收到 402 响应时,Client 会自动:

  1. 解析支付要求
  2. 匹配已注册的支付方式
  3. 生成支付签名
  4. /token 端点交换访问令牌
  5. 使用新令牌重试原始请求

整个过程对开发者透明,只需正常使用 HTTP 客户端即可。

完整示例

Server 端完整示例

from x402_mock.servers import Http402Server, create_private_key
from x402_mock.adapters.evm.schemas import EVMPaymentComponent
from x402_mock.engine.events import SettleSuccessEvent

# 创建 Server
token_key = create_private_key()
app = Http402Server(token_key=token_key, token_expires_in=300)

# 添加支付方式
app.add_payment_method(
    EVMPaymentComponent(
        amount=0.5,
        currency="USDC",
        caip2="eip155:11155111",
        token="0x1c7D4B196Cb0C7B01d743Fbc6116a902379C7238"
    )
)

# 支付成功事件处理
@app.hook(SettleSuccessEvent)
async def log_payment_success(event, deps):
    print(f"💰 收到支付: {event.settlement_result.authorized_amount} USDC")

# 受保护的 API 端点
@app.get("/api/premium-content")
@app.payment_required
async def get_premium_content(payload):
    return {
        "content": "这是付费内容",
        "paid_by": payload["address"],
        "timestamp": payload.get("timestamp")
    }

# 运行 Server(使用 uvicorn)
# uvicorn server:app --host 0.0.0.0 --port 8000

Client 端完整示例

import asyncio
from x402_mock.clients.http_client import Http402Client
from x402_mock.adapters.evm.schemas import EVMPaymentComponent

async def main():
    async with Http402Client() as client:
        # 注册支付方式(支持多个)
        client.add_payment_method(
            EVMPaymentComponent(
                caip2="eip155:11155111",
                amount=1.0,      # 最多支付 1.0 USDC
                currency="USDC",
                token="0x1c7D4B196Cb0C7B01d743Fbc6116a902379C7238"
            )
        )

        # 请求受保护的内容(自动处理支付)
        response = await client.get("http://localhost:8000/api/premium-content")

        if response.status_code == 200:
            data = response.json()
            print(f"✅ 获取到内容: {data['content']}")
            print(f"   支付方: {data['paid_by']}")
        else:
            print(f"❌ 请求失败: {response.status_code}")

if __name__ == "__main__":
    asyncio.run(main())

高级配置

自定义 RPC 节点

# Server 端指定 RPC 节点
app.add_payment_method(
    EVMPaymentComponent(
        amount=0.5,
        currency="USDC",
        caip2="eip155:1",  # Ethereum 主网
        token="0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48",  # USDC
        rpc_url="https://eth-mainnet.g.alchemy.com/v2/YOUR_KEY"
    )
)

多链支持

# 支持多个链的支付方式
app.add_payment_method(
    EVMPaymentComponent(
        amount=0.1,
        currency="ETH",
        caip2="eip155:1",  # Ethereum 主网
        token=None  # 使用原生代币
    )
)

app.add_payment_method(
    EVMPaymentComponent(
        amount=1.0,
        currency="USDC",
        caip2="eip155:42161",  # Arbitrum
        token="0xFF970A61A04b1cA14834A43f5dE4533eBDDB5CC8"
    )
)

核心组件 AdapterHub

AdapterHub 是统一的区块链适配器网关,为用户不依赖外部 facilitors,自身完成验证签名和转账操作的工具。支持 EVM 区块链(SVM 开发中)。

核心方法

方法 功能
register_payment_methods() 注册支付方式(client_role=False 为 Server 角色,True 为 Client 角色)
signature() 生成支付签名(Client 调用)
verify_signature() 验证签名(Server 调用)
settle() 执行链上转账(Server 调用)
initialize() 客户端初始化(如 Permit2 授权)

代码示例

from x402_mock.adapters import AdapterHub
from x402_mock.adapters.evm.schemas import EVMPaymentComponent

# 1. 初始化(需要 EVM 私钥)
hub = AdapterHub(
    evm_private_key="0xyour_private_key",  # 或使用环境变量 EVM_PRIVATE_KEY
    request_timeout=60
)

# 2. 注册支付方式
hub.register_payment_methods(
    EVMPaymentComponent(
        amount=1.0,
        currency="USDC",
        caip2="eip155:11155111",
        token="0x1c7D4B196Cb0C7B01d743Fbc6116a902379C7238"
    ),
    client_role=True  # Client 角色设为 True
)

# 3. Client 生成签名
permit = await hub.signature(remote_components)

# 4. Server 验证签名
result = await hub.verify_signature(permit)
if result.is_valid():
    # 5. Server 执行转账
    confirmation = await hub.settle(permit)

未来规划

  • bundle_settle:批量转账,减少 Gas 费用
  • SVM 支持:Solana 适配器(开发中)

链上信息获取工具

在配置支付方式时,您可能需要获取链上的各种信息,如 RPC 节点地址、代币合约地址、代币精度和版本等。x402-mock 提供了一系列工具方法来简化这些信息的获取。

导入工具方法

from x402_mock.adapters.evm.constants import (
    EvmChainInfoFromEthereumLists,
    EvmPublicRpcFromChainList,
    EvmTokenListFromUniswap,
    fetch_erc20_name_version_decimals,
    get_rpc_key_from_env
)

方法说明

1. EvmChainInfoFromEthereumLists

功能:从 ethereum-lists 仓库获取 EVM 链的详细配置信息。

主要用途: - 获取链的 Infura/Alchemy RPC URL(包含 API 密钥占位符) - 获取公共 RPC 节点列表 - 获取链的基本信息(名称、浏览器地址等)

示例用法

chain_info = EvmChainInfoFromEthereumLists()

# 获取 Infura RPC URL(需要填充 API 密钥)
infura_url = chain_info.get_infura_rpc_url("eip155:1")
# 返回类似:https://mainnet.infura.io/v3/{RPC_KEYS}

# 获取 Alchemy RPC URL(需要填充 API 密钥)
alchemy_url = chain_info.get_alchemy_rpc_url("eip155:1")
# 返回类似:https://eth-mainnet.g.alchemy.com/v2/{RPC_KEYS}

# 获取公共 RPC 节点列表
public_rpcs = chain_info.get_public_rpc_urls("eip155:1")
# 返回:["https://api.mycryptoapi.com/eth", ...]

2. EvmPublicRpcFromChainList

功能:从 Chainlist.org 获取公共 RPC 节点信息。

主要用途: - 获取无跟踪或有限跟踪的公共 RPC 节点 - 根据隐私偏好选择 RPC 节点 - 支持 HTTPS 和 WebSocket 协议

示例用法

rpc_finder = EvmPublicRpcFromChainList()

# 获取无跟踪的公共 RPC 节点
public_rpc = rpc_finder.pick_public_rpc(
    caip2="eip155:1",
    start_with="https://",
    tracking_type="none"
)
# 返回类似:https://rpc.ankr.com/eth

# 获取特定链的所有公共 RPC 信息
chain_rpcs = rpc_finder.get_specific_chain_public_rpcs("eip155:1")

3. EvmTokenListFromUniswap

功能:从 Uniswap 官方代币列表获取代币信息。

主要用途: - 获取代币的合约地址和精度 - 支持多链代币查询 - 自动缓存数据,减少网络请求

示例用法

token_finder = EvmTokenListFromUniswap()

# 获取代币地址和精度
address, decimals = token_finder.get_token_address_and_decimals(
    caip2="eip155:1",
    symbol="USDC"
)
# 返回:("0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48", 6)

4. fetch_erc20_name_version_decimals

功能:直接从链上 RPC 查询代币的详细信息。

主要用途: - 查询代币的 name()version()decimals() 函数 - 验证代币合约的完整信息 - 获取最新的链上数据

示例用法

# 从链上查询代币信息
name, version, decimals = fetch_erc20_name_version_decimals(
    rpc_url="https://eth-mainnet.g.alchemy.com/v2/YOUR_KEY",
    token_address="0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48"
)
# 返回:("USD Coin", "2", 6)

5. get_rpc_key_from_env

功能:从环境变量获取 RPC 服务商的 API 密钥。

主要用途: - 安全地获取 Infura 或 Alchemy API 密钥 - 支持自定义环境变量名称 - 返回 None 时表示使用公共节点

示例用法

# 获取 Infura 密钥(默认)
infura_key = get_rpc_key_from_env("EVM_INFURA_KEY")

# 获取 Alchemy 密钥
alchemy_key = get_rpc_key_from_env("EVM_ALCHEMY_KEY")

# 使用密钥构建 RPC URL
if infura_key:
    rpc_url = f"https://mainnet.infura.io/v3/{infura_key}"
else:
    # 使用公共节点
    rpc_finder = EvmPublicRpcFromChainList()
    rpc_url = rpc_finder.pick_public_rpc("eip155:1")

自动填充支付组件

这些工具方法通常被 EVMPaymentComponent 内部使用,当您未提供完整信息时自动填充:

# 只需提供基本信息,系统会自动查询缺失数据
payment = EVMPaymentComponent(
    amount=0.5,
    currency="USDC",
    caip2="eip155:1"
    # token、token_name、token_decimals、token_version 会自动查询
)

# 系统内部会:
# 1. 使用 EvmTokenListFromUniswap 获取 USDC 合约地址和精度
# 2. 使用 fetch_erc20_name_version_decimals 获取代币名称和版本
# 3. 使用 EvmPublicRpcFromChainList 获取公共 RPC 节点
# 4. 使用 get_rpc_key_from_env 检查是否有私有 RPC 密钥

最佳实践

  1. 生产环境:建议提供完整的 tokenrpc_url 等信息,减少网络查询
  2. 开发环境:可以依赖自动查询,简化配置
  3. 性能考虑:首次查询会进行网络请求,后续使用缓存
  4. 错误处理:网络不可用时,自动回退到内置的默认配置

故障排除

常见问题

  1. 402 响应后支付失败
  2. 检查私钥配置是否正确
  3. 确认代币余额充足
  4. 验证链 ID 和代币地址匹配

  5. 令牌验证失败

  6. 检查 token_key 是否一致
  7. 确认令牌未过期
  8. 验证签名算法

  9. RPC 连接问题

  10. 检查网络连接
  11. 确认 RPC URL 有效
  12. 考虑使用备用节点

调试模式

x402-mock 使用 loguru 作为日志库,默认不输出任何日志(避免干扰您的业务日志)。通过 setup_logger 开启:

from x402_mock.utils import setup_logger

# 仅输出到控制台(DEBUG 级别)
setup_logger(level="DEBUG")

# 同时保存到文件(可选)
setup_logger(level="DEBUG", log_to_file=True, log_path="logs/x402-mock.log")

setup_logger 参数说明

参数 类型 默认值 说明
level str "INFO" 日志级别:"DEBUG""INFO""WARNING""ERROR"
log_to_file bool False 是否同时保存日志到本地文件
log_path str "logs/x402-mock.log" 日志文件路径,目录不存在时自动创建

说明:日志文件自动按 10 MB 轮转,保留最近 7 天。

MCP 工具集成(供 AI Agent / 大模型调用)

x402-mock 提供了原生的 MCP(Model Context Protocol) 工具支持,让 LLM Agent(如 GitHub Copilot、Claude、GPT 等)可以直接调用,自动完成完整的 402 支付流程,无需手动编写支付代码。

安装 MCP 依赖

MCP 支持作为可选依赖提供,使用以下命令安装:

uv sync --extra mcp

可用的 MCP 工具

工具名 角色 说明
source_request Client 访问受 402 保护的资源,自动完成签名与支付重试
signature Client 根据服务端返回的支付组件列表,生成签名的 permit
verify_and_settle Server 验证 permit 签名并在链上结算(一步完成)

Server 端 MCP 示例

# example/mcp_server_example.py
import os
from mcp.server.fastmcp import FastMCP
from x402_mock.adapters.adapters_hub import AdapterHub
from x402_mock.adapters.evm.schemas import EVMPaymentComponent
from x402_mock.mcp.facilitor_tools import FacilitorTools

TOKEN_KEY = os.environ.get("X402_TOKEN_KEY", "dev-secret-change-me")
EVM_PRIVATE_KEY = os.environ.get("EVM_PRIVATE_KEY")

hub = AdapterHub(evm_private_key=EVM_PRIVATE_KEY)
hub.register_payment_methods(
    EVMPaymentComponent(
        amount=0.8,
        currency="USDC",
        caip2="eip155:11155111",
        token="0x1c7D4B196Cb0C7B01d743Fbc6116a902379C7238",
    ),
    client_role=False,  # 服务端角色
)

mcp = FastMCP("x402")
FacilitorTools(adapter_hub=hub, mcp=mcp, client_role=False)
mcp.run()  # stdio 传输

Client 端 MCP 示例

# example/mcp_client_example.py
import os
from mcp.server.fastmcp import FastMCP
from x402_mock.adapters.adapters_hub import AdapterHub
from x402_mock.adapters.evm.schemas import EVMPaymentComponent
from x402_mock.mcp.facilitor_tools import FacilitorTools

EVM_PRIVATE_KEY = os.environ.get("EVM_PRIVATE_KEY")

hub = AdapterHub(evm_private_key=EVM_PRIVATE_KEY)

if EVM_PRIVATE_KEY:
    hub.register_payment_methods(
        EVMPaymentComponent(
            amount=0.5,
            currency="USDC",
            caip2="eip155:11155111",
            token="0x1c7D4B196Cb0C7B01d743Fbc6116a902379C7238",
        ),
        client_role=True,  # 客户端角色
    )

mcp = FastMCP("x402-client")
FacilitorTools(adapter_hub=hub, mcp=mcp, client_role=True)
mcp.run()  # stdio 传输

在 VS Code(GitHub Copilot)中配置

将以下配置添加到您项目的 .vscode/mcp.json(或用户级 MCP 配置文件)中,即可让 Copilot Agent 直接调用 x402-mock 的支付工具:

{
  "servers": {
    "X402-Mock-Server": {
      "type": "stdio",
      "command": "uv",
      "args": ["run", "example/mcp_server_example.py"],
      "env": {
        "X402_TOKEN_KEY": "dev-secret-change-me",
        "EVM_PRIVATE_KEY": "your_private_key_here",
        "EVM_INFURA_KEY": "your_infura_key_here"
      }
    },
    "X402-Mock-Client": {
      "type": "stdio",
      "command": "uv",
      "args": ["run", "example/mcp_client_example.py"],
      "env": {
        "EVM_PRIVATE_KEY": "your_private_key_here",
        "EVM_INFURA_KEY": "your_infura_key_here"
      }
    }
  }
}

配置完成后,在 VS Code 的 Copilot Chat(Agent 模式)中,即可直接向大模型发出自然语言指令,例如:

请用 source_request 工具访问 http://localhost:8000/api/protected-data

Copilot 将自动调用 source_request 工具,完成签名、支付、重试全流程并返回结果。

source_request 工具参数

参数 类型 默认值 说明
url str 必填 目标资源 URL
method str "GET" HTTP 方法
headers dict None 额外请求头
timeout float 30.0 请求超时时间(秒)

返回值为包含 status_codeheadersbody 的字典。

下一步


提示:在生产环境中,请确保: 1. 使用安全的密钥管理方案 2. 配置适当的超时和重试策略 3. 监控支付成功率和失败原因 4. 定期更新依赖包以获取安全修复