x402 Python 客户端实战用 payment-identifier 扩展实现支付幂等与请求安全重试【免费下载链接】x402A payments protocol for the internet. Built on HTTP.项目地址: https://gitcode.com/GitHub_Trending/x4/x402本文围绕 x402 仓库中examples/python/clients/payment-identifier示例讲解如何在 Python 客户端中通过payment-identifier扩展为每次逻辑请求生成唯一支付标识从而在支付场景下实现幂等重复请求命中服务端缓存、不再重复扣款。读完本文你将掌握该扩展的完整接入流程生成 ID、注入 extensions、挂钩支付流程、配套服务端的缓存实现方式以及从源码层面理解 ID 的格式约束与校验逻辑。一、示例定位支付幂等为什么是刚需在基于 HTTP 的支付协议x402 v2中客户端收到402 Payment Required后需要构造PaymentPayload并重新发起请求。此时如果网络抖动、响应超时或客户端崩溃重试就意味着可能重复发起支付。payment-identifier扩展就是为此设计的客户端在支付载荷中携带一个幂等键id资源服务器据此去重——同一个 payment ID 只结算一次后续相同 ID 的请求直接返回缓存响应。示例目录结构如下客户端示例examples/python/clients/payment-identifier/main.py配套说明即本文核心文档 README服务端示例examples/python/servers/payment-identifier/main.py演示按 payment ID 缓存响应的完整幂等实现SDK 扩展实现python/x402/extensions/payment_identifier/ 包扩展规范specs/extensions/payment_identifier.md。二、工作原理从支付 ID 生成到缓存命中按示例 README 的描述整个幂等流程分为四步客户端调用generate_payment_id()生成唯一支付 ID客户端通过append_payment_identifier_to_extensions()将该 ID 写入PaymentPayload的 extensions服务端以 payment ID 为键缓存响应携带相同 payment ID 的重试请求直接返回缓存响应不重复处理支付。README 给出的最小接入代码如下from x402 import x402Client from x402.extensions.payment_identifier import ( append_payment_identifier_to_extensions, generate_payment_id, ) from x402.http.clients import x402HttpxClient client x402Client() # ... register schemes ... # 为本次逻辑请求生成唯一 payment ID payment_id generate_payment_id() # 在 payload 创建前挂钩支付流程注入 payment ID async def before_payment_creation(context): extensions context.payment_required.extensions if extensions is not None: append_payment_identifier_to_extensions(extensions, payment_id) client.on_before_payment_creation(before_payment_creation) async with x402HttpxClient(client) as http: # 第一次请求 - 正常处理支付 response1 await http.get(url) # 相同 payment ID 重试 - 返回缓存响应不再支付 response2 await http.get(url)关键点在于on_before_payment_creation钩子它在支付载荷创建之前执行能够拿到服务端在PaymentRequired响应中声明的 extensions 并原地修改。2.1 生成 IDUUID v4 前缀generate_payment_id()的实现在 python/x402/extensions/payment_identifier/utils.pydef generate_payment_id(prefix: str pay_) - str: # 生成去掉连字符的 UUID v432 个十六进制字符 uuid_str uuid.uuid4().hex return f{prefix}{uuid_str}默认前缀为pay_因此生成的 ID 形如pay_7d5d747be160e280504c099d984bcfe0。该函数支持自定义前缀如generate_payment_id(txn_)和空字符串无前缀这正是 README最佳实践中建议使用描述性前缀如order_、sub_来区分支付类型的底层支撑。ID 格式约束定义在 python/x402/extensions/payment_identifier/types.py常量值含义PAYMENT_ID_MIN_LENGTH16ID 最小长度PAYMENT_ID_MAX_LENGTH128ID 最大长度PAYMENT_ID_PATTERN^[a-zA-Z0-9_-]$仅允许字母、数字、连字符、下划线PAYMENT_IDENTIFIERpayment-identifier扩展在 extensions 字典中的键名is_valid_payment_id()依据上述约束做长度与正则校验任何非法 ID 都会在客户端侧被提前拦截。2.2 注入 extensions只在服务端声明支持时生效append_payment_identifier_to_extensions()实现在 python/x402/extensions/payment_identifier/client.py其行为有两个值得注意的设计被动降级函数会先检查extensions.get(PAYMENT_IDENTIFIER)是否是服务端声明的合法扩展结构is_payment_identifier_extension。如果服务端根本没有声明该扩展extensions 原样返回、不做任何修改——客户端不会因为擅自加扩展而破坏与服务端的协议约定原地修改并校验校验通过后payment ID 被写入扩展的info.id字段info_dict[id] payment_id修改直接作用于传入的 extensions 字典同一引用随后该 extensions 会被包含进PaymentPayload。若传入的自定义 ID 不合法则抛出ValueError提示必须为 16-128 字符且仅含合法字符。从源码结构看扩展声明本身携带一份 JSON Schemapython/x402/extensions/payment_identifier/schema.py符合 Draft 2020-12要求required字段必为布尔值、id需满足上述长度与 pattern 约束客户端与服务端可据此互相校验。2.3 数据模型python/x402/extensions/payment_identifier/types.py 定义了两个 Pydantic 模型PaymentIdentifierInfo包含required: bool服务端是否强制要求客户端携带 ID为true时缺失将收到 400 Bad Request与id: str | None客户端提供的幂等键PaymentIdentifierExtension包含info与schema两部分同时用于服务端声明只有required无id和客户端载荷info中带id。三、完整客户端示例走读真实的 main.py 在 README 代码片段基础上补全了可运行的细节import os import sys from dotenv import load_dotenv from eth_account import Account from x402 import x402Client from x402.extensions.payment_identifier import ( append_payment_identifier_to_extensions, generate_payment_id, ) from x402.http import x402HTTPClient from x402.http.clients import x402HttpxClient from x402.mechanisms.evm import EthAccountSigner from x402.mechanisms.evm.exact.register import register_exact_evm_client from x402.schemas import PaymentCreationContext load_dotenv() private_key os.getenv(EVM_PRIVATE_KEY) if not private_key: print(Error: EVM_PRIVATE_KEY environment variable is required) sys.exit(1) base_url os.getenv(RESOURCE_SERVER_URL, http://localhost:4022) endpoint_path os.getenv(ENDPOINT_PATH, /weather) url f{base_url}{endpoint_path} account Account.from_key(private_key) client x402Client() register_exact_evm_client(client, EthAccountSigner(account)) payment_id generate_payment_id() async def before_payment_creation(context: PaymentCreationContext) - None: extensions context.payment_required.extensions if extensions is not None: # 仅当服务端声明了该扩展时才会真正追加 ID append_payment_identifier_to_extensions(extensions, payment_id) client.on_before_payment_creation(before_payment_creation) http_client x402HTTPClient(client) # 用于从响应头提取支付结算信息 async with x402HttpxClient(client) as http: start_time1 time.time() response1 await http.get(url) await response1.aread() duration1 int((time.time() - start_time1) * 1000) print(fResponse ({duration1}ms): {response1.text}) # 从响应头中提取支付结算响应若存在 try: settle_response http_client.get_payment_settle_response( lambda name: response1.headers.get(name) ) print(settle_response.model_dump_json(indent2)) except ValueError: pass # 第二次请求相同 payment ID应命中服务端缓存 ...从示例源码可以看到几个 README 未展开的实操细节客户端通过register_exact_evm_client(client, EthAccountSigner(account))注册 EVM exact 方案签名器基于eth_account.Account服务端地址与路径可用环境变量RESOURCE_SERVER_URL默认http://localhost:4022和ENDPOINT_PATH默认/weather覆盖每次响应都会尝试用x402HTTPClient.get_payment_settle_response()从响应头提取结算信息第一次请求能提取到支付已结算第二次请求提取失败抛ValueError示例借此判断响应来自缓存未发生支付两次请求耗时被计时并输出最终打印加速比例缓存命中通常比完整结算快 97% 左右具体取决于结算链上耗时。四、配套服务端payment ID 驱动的缓存实现客户端幂等能生效的前提是服务端正确实现缓存。examples/python/servers/payment-identifier/main.py 给出了一个 FastAPI 参考实现核心有三块1. 声明扩展支持。在路由配置中通过declare_payment_identifier_extension()向客户端宣告支持实现在 python/x402/extensions/payment_identifier/server.py返回结构为{info: {required: required}, schema: payment_identifier_schema}routes { GET /weather: RouteConfig( accepts[ PaymentOption( schemeexact, price$0.001, networkEVM_NETWORK, # eip155:84532 (Base Sepolia) pay_toEVM_ADDRESS, ), ], mime_typeapplication/json, # 宣告 payment-identifier 扩展requiredFalse 表示可选 extensions{ PAYMENT_IDENTIFIER: declare_payment_identifier_extension(requiredFalse), }, ), }注意客户端侧的append_payment_identifier_to_extensions()正是依赖这里声明的payment-identifier条目才会注入 ID。2. 结算后写入缓存。服务端通过server.on_after_settle()挂钩在支付结算成功后把响应按 payment ID 缓存async def after_settle(ctx: SettleContext) - None: payment_id extract_payment_identifier(ctx.payment_payload) if payment_id: idempotency_cache[payment_id] CachedResponse( timestamptime.time(), response{report: {weather: sunny, temperature: 70, cached: False}}, )3. 支付前检查缓存。一个自定义中间件在PaymentMiddlewareASGI之前执行解码X-Payment请求头、反序列化为PaymentPayload、用extract_payment_identifier()提取 ID命中且未过期的缓存直接以 200 返回并把响应体中的cached标记为true从而完全绕开支付处理流程。缓存实现是进程内字典TTL 为 1 小时CACHE_TTL_SECONDS 60 * 60每次请求先执行过期清理。源码注释中明确提示生产环境应使用 Redis 等分布式缓存——这也是 README使用场景中负载均衡同一请求可命中共享缓存的不同服务器这一条成立的前提共享缓存需自行引入外部存储。五、环境准备与运行步骤前置条件Python 3.10uv按 uv 官方文档安装一个运行中的 payment-identifier 服务端见 examples/python/servers/payment-identifier一把有效的 EVM 私钥用于支付Base Sepolia 网络 USDC。配置与运行安装依赖uv sync依赖声明见 examples/python/clients/payment-identifier/pyproject.tomlpython-dotenv1.0.0与x402[httpx,evm,extensions]其中x402通过[tool.uv.sources]以可编辑模式指向仓库内的python/x402包。复制.env-local为.env并填入私钥cp .env-local .env必需的环境变量EVM_PRIVATE_KEY— 用于 EVM 支付的以太坊私钥。在另一个终端启动 payment-identifier 服务端cd ../../servers/payment-identifier uv run python main.py服务端默认监听http://0.0.0.0:4022需要设置EVM_ADDRESS收款地址环境变量facilitator 默认为官方 facilitator 服务。运行客户端uv run python main.py预期输出客户端会依次打印生成的 payment ID、两次请求的耗时与响应体、以及汇总对比形态如下时间数值会随结算网络波动Generated Payment ID: pay_7d5d747be160e280504c099d984bcfe0 First Request (with payment ID: pay_7d5d747be160e280504c099d984bcfe0) Making request to: http://localhost:4022/weather Response (1523ms): {report: {weather: sunny, temperature: 70, cached: false}} Payment settled on eip155:84532 Second Request (SAME payment ID: pay_7d5d747be160e280504c099d984bcfe0) Making request to: http://localhost:4022/weather Expected: Server returns cached response without payment processing Response (45ms): {report: {weather: sunny, temperature: 70, cached: true}} No payment processed - response served from cache! Summary Payment ID: pay_7d5d747be160e280504c099d984bcfe0 First request: 1523ms (payment processed) Second request: 45ms (cached) Cached response was 97% faster!第一次请求cached: false且能提取到结算响应第二次请求cached: true且无支付响应头——幂等生效的两个可观测判据。六、典型使用场景README 总结了四类适用场景网络故障安全重试失败请求不产生重复支付客户端崩溃持久化 payment ID重启后凭同一 ID 恢复请求负载均衡同一请求可能落到不同服务器实例配合共享缓存如 Redis实现跨实例去重测试开发阶段重放请求而不消耗资金。七、最佳实践在逻辑请求粒度生成 payment ID而不是每次重试各生成一个——重试的意义恰恰是复用同一个 ID持久化 payment ID长任务应将 ID 落盘使其在进程重启后仍然有效使用描述性前缀如order_、sub_便于按业务类型识别与排查generate_payment_id(prefix)原生支持该用法不要跨不同逻辑请求复用 payment ID复用的后果是不同请求互相顶掉缓存导致错误的响应命中。八、小结x402 的payment-identifier扩展在协议层面为支付请求引入了标准幂等键服务端在PaymentRequired的extensions中声明可设required强制客户端在PaymentPayload的info.id中回填 16-128 位合法字符的唯一 ID。Python SDK 用generate_payment_id、append_payment_identifier_to_extensions、extract_payment_identifier三个函数覆盖了生成、注入、提取的完整闭环配合on_before_payment_creation客户端与on_after_settle服务端两个钩子即可落地重试不重复扣款。示例代码客户端 main.py 服务端 main.py提供了可直接运行的端到端参考生产化时只需把进程内缓存替换为 Redis 一类的共享存储。【免费下载链接】x402A payments protocol for the internet. Built on HTTP.项目地址: https://gitcode.com/GitHub_Trending/x4/x402创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考