agents24 仓库 paypal-integration Skill 实战指南:Express Checkout、IPN、订阅与退款全流程实现
【免费下载链接】agentsMulti-harness agentic plugin marketplace for Claude Code, Codex, Cursor, OpenCode, GitHub Copilot, and Google Antigravity项目地址: https://gitcode.com/GitHub_Trending/agents24/agents
本指南以agents24/agents仓库中 paypal-integration Skill 及其详细模式文档为主体,系统讲解 PayPal 支付集成的六大核心场景:OAuth 鉴权与 Express Checkout 服务端订单、IPN 异步通知验证与处理、订阅计费(Billing Plans/Subscriptions)、退款工作流、统一错误处理以及沙箱测试。读完本文,你将掌握一套可直接复制运行的服务端 PayPal 集成代码骨架,并理解 webhook 安全、幂等处理与 sandbox/live 环境切换等生产级细节。
一、Skill 定位:何时使用 paypal-integration
该 Skill 位于仓库 plugins/payment-processing 插件目录下,属于 Payment Processing(支付处理)插件家族的四个 Skill 之一(其余为 stripe-integration、pci-compliance、billing-automation)。
依据 SKILL.md 中的 frontmatter 声明,该 Skill 的激活场景为:
- 将 PayPal 作为支付选项接入
- 实现 Express Checkout 快速结账流程
- 用 PayPal 搭建周期性订阅计费(recurring billing)
- 处理退款与支付争议(disputes)
- 处理 PayPal webhook(即 IPN 异步通知)
- 支持国际支付
- 实现 PayPal Subscriptions 订阅产品
在仓库的文档体系里,Skill 采用「渐进式披露」(Progressive Disclosure)三层架构:Frontmatter 元数据(名称与激活条件)始终加载,核心指导在 SKILL.md 激活时加载,而references/details.md属于按需加载的第三层资源,存放完整的模式与可运行示例,即本文主体内容的来源。
1.1 三种支付产品
| 产品 | 用途 |
|---|---|
| PayPal Checkout | 一次性支付、Express Checkout 体验、支持访客与 PayPal 账户支付 |
| PayPal Subscriptions | 周期性计费、订阅计划、自动续费 |
| PayPal Payouts | 向多个收款人批量打款,适用于市场与平台支付 |
1.2 两种集成方式
- 客户端集成(JavaScript SDK):使用 Smart Payment Buttons 托管支付流程,后端代码最少;
- 服务端集成(REST API):对支付流程拥有完全控制权,可定制结账 UI,支持高级功能。
details.md中的示例全部采用服务端 REST API路线,通过requests直接调用 PayPal 官方 HTTP 接口,不依赖第三方 SDK 封装,便于理解底层请求结构。
二、Express Checkout 服务端实现:从 OAuth 到订单捕获
2.1 PayPalClient 基类:环境切换与 OAuth 访问令牌
details.md给出的核心类是PayPalClient,它同时承担环境路由与令牌管理两个职责:
import requests import json class PayPalClient: def __init__(self, client_id, client_secret, mode='sandbox'): self.client_id = client_id self.client_secret = client_secret self.base_url = 'https://api-m.sandbox.paypal.com' if mode == 'sandbox' else 'https://api-m.paypal.com' self.access_token = self.get_access_token() def get_access_token(self): """Get OAuth access token.""" url = f"{self.base_url}/v1/oauth2/token" headers = {"Accept": "application/json", "Accept-Language": "en_US"} response = requests.post( url, headers=headers, data={"grant_type": "client_credentials"}, auth=(self.client_id, self.client_secret) ) return response.json()['access_token']关键点拆解:
mode参数:'sandbox'路由到https://api-m.sandbox.paypal.com,其余值路由到生产环境https://api-m.paypal.com。这是隔离测试与线上流量的核心开关;- OAuth2 客户端凭证模式:向
/v1/oauth2/token发起POST,携带grant_type=client_credentials,并以(client_id, client_secret)作为 HTTP Basic Auth(requests.post的auth参数会自动编码)。返回 JSON 中的access_token即为后续所有请求的Bearer凭证; - 仓库中 payment-integration 智能体 强调「测试凭证必须不能在线上生效」,因此在多环境部署时应把
mode、CLIENT_ID、CLIENT_SECRET全部纳入环境变量管理,避免测试卡在线上站点被接受而触发 PCI 违规。
2.2 创建订单(v2/checkout/orders)
def create_order(self, amount, currency='USD'): """Create a PayPal order.""" url = f"{self.base_url}/v2/checkout/orders" headers = { "Content-Type": "application/json", "Authorization": f"Bearer {self.access_token}" } payload = { "intent": "CAPTURE", "purchase_units": [{ "amount": { "currency_code": currency, "value": str(amount) } }] } response = requests.post(url, headers=headers, json=payload) return response.json()要点说明:
- 订单创建接口是 PayPalOrders v2API,请求体至少包含
intent(CAPTURE表示创建后直接捕获)与purchase_units(采购单元,含金额); - 金额
value必须转为字符串(str(amount)),这是 PayPal API 对金额字段的类型约束,避免浮点精度问题; - 创建成功后返回的 JSON 中
links数组内rel == 'approve'的链接即为用户批准支付页(见下方订阅部分对同一模式的复用),这也被 SKILL.md 的测试示例所验证:next((link['href'] for link in order['links'] if link['rel'] == 'approve'), None)。
2.3 捕获订单与查询订单详情
def capture_order(self, order_id): """Capture payment for an order.""" url = f"{self.base_url}/v2/checkout/orders/{order_id}/capture" headers = { "Content-Type": "application/json", "Authorization": f"Bearer {self.access_token}" } response = requests.post(url, headers=headers) return response.json() def get_order_details(self, order_id): """Get order details.""" url = f"{self.base_url}/v2/checkout/orders/{order_id}" headers = { "Authorization": f"Bearer {self.access_token}" } response = requests.get(url, headers=headers) return response.json()生产建议:捕获动作必须在服务端完成。客户端(Smart Buttons 的onApprove回调)只负责把orderID回传后端,由后端调用capture_order并向 PayPal 再次确认订单状态。这正对应 payment-integration.md 中「服务端验证:从提供商 API 重新拉取支付状态,永远不要只信任 webhook 负载或客户端响应」的安全要求。
2.4 客户端入口:Smart Buttons 快速开始
完整的客户端-服务端链路可参考 SKILL.md 的 Quick Start:前端通过 PayPal JS SDK 渲染按钮,createOrder中声明purchase_units,onApprove中调用actions.order.capture()成功后把orderID发往后端/api/paypal/capture进行服务端捕获与校验:
// Frontend - PayPal Smart Buttons <div id="paypal-button-container"></div> <script src="https://www.paypal.com/sdk/js?client-id=YOUR_CLIENT_ID¤cy=USD"></script> <script> paypal.Buttons({ createOrder: function(data, actions) { return actions.order.create({ purchase_units: [{ amount: { value: '25.00' } }] }); }, onApprove: function(data, actions) { return actions.order.capture().then(function(details) { // Payment successful console.log('Transaction completed by ' + details.payer.name.given_name); // Send to backend for verification fetch('/api/paypal/capture', { method: 'POST', headers: {'Content-Type': 'application/json'}, body: JSON.stringify({orderID: data.orderID}) }); }); } }).render('#paypal-button-container'); </script>三、IPN(Instant Payment Notification)处理:验证与业务分发
IPN 是 PayPal 的异步通知机制:支付状态变化时,PayPal 向商户配置的端点推送表单数据。details.md给出一个完整的 Flask 端点示例,其核心分为「回验」与「分发」两段。
3.1 端点与消息分发
from flask import Flask, request import requests from urllib.parse import parse_qs app = Flask(__name__) @app.route('/ipn', methods=['POST']) def handle_ipn(): """Handle PayPal IPN notifications.""" # Get IPN message ipn_data = request.form.to_dict() # Verify IPN with PayPal if not verify_ipn(ipn_data): return 'IPN verification failed', 400 # Process IPN based on transaction type payment_status = ipn_data.get('payment_status') txn_type = ipn_data.get('txn_type') if payment_status == 'Completed': handle_payment_completed(ipn_data) elif payment_status == 'Refunded': handle_refund(ipn_data) elif payment_status == 'Reversed': handle_chargeback(ipn_data) return 'IPN processed', 200分发逻辑依据payment_status字段路由到三个处理器:Completed(支付完成)、Refunded(退款)、Reversed(退单/撤销)。注意txn_type字段同样被读取,可用于更细粒度的事件识别。
3.2 回验机制(VERIFIED / INVALID)
def verify_ipn(ipn_data): """Verify IPN message authenticity.""" # Add 'cmd' parameter verify_data = ipn_data.copy() verify_data['cmd'] = '_notify-validate' # Send back to PayPal for verification paypal_url = 'https://ipnpb.sandbox.paypal.com/cgi-bin/webscr' # or production URL response = requests.post(paypal_url, data=verify_data) return response.text == 'VERIFIED'IPN 安全模型的核心是回环验证:商户把收到的完整 IPN 数据原样加cmd=_notify-validate后回传给 PayPal 的 IPN 端点,PayPal 返回VERIFIED才视为可信。生产环境应将 URL 替换为https://ipnpb.paypal.com/cgi-bin/webscr。
这一点与仓库智能体 payment-integration.md 强调的 webhook 安全要求完全一致:
- 签名验证:必须使用官方机制验证通知真实性,绝不处理未验证的 webhook;
- 原始 Body 保留:验证前不得修改请求体,JSON 中间件会破坏校验;
- 幂等处理:把事件 ID 存入数据库,处理前检查去重——webhook 失败会重试,提供商不保证单次投递;
- 快速响应:应在执行数据库写入等昂贵操作之前返回
2xx(文档中的示例先返回200,处理器内部完成业务);超时触发重试会导致重复处理。
3.3 三个业务处理器
def handle_payment_completed(ipn_data): """Process completed payment.""" txn_id = ipn_data.get('txn_id') payer_email = ipn_data.get('payer_email') mc_gross = ipn_data.get('mc_gross') item_name = ipn_data.get('item_name') # Check if already processed (prevent duplicates) if is_transaction_processed(txn_id): return # Update database # Send confirmation email # Fulfill order print(f"Payment completed: {txn_id}, Amount: ${mc_gross}") def handle_refund(ipn_data): """Handle refund.""" parent_txn_id = ipn_data.get('parent_txn_id') mc_gross = ipn_data.get('mc_gross') # Process refund in your system print(f"Refund processed: {parent_txn_id}, Amount: ${mc_gross}") def handle_chargeback(ipn_data): """Handle payment reversal/chargeback.""" txn_id = ipn_data.get('txn_id') reason_code = ipn_data.get('reason_code') # Handle chargeback print(f"Chargeback: {txn_id}, Reason: {reason_code}")字段语义说明:
txn_id:本次交易 ID(退款场景下是退款交易 ID);parent_txn_id:退款对应的原始交易 ID,退款处理应以它为键关联原订单;mc_gross:交易总额(含费用);reason_code:退单原因码,用于风控分析。
handle_payment_completed中的is_transaction_processed(txn_id)幂等检查不可省略——这正是仓库智能体所列「Out-of-order webhooks breaking Lambda functions (no idempotency) → production failures」这一真实故障案例的防御措施。
四、订阅与周期性计费:Billing Plans 与 Subscriptions
4.1 创建订阅计划(v1/billing/plans)
def create_subscription_plan(name, amount, interval='MONTH'): """Create a subscription plan.""" client = PayPalClient(CLIENT_ID, CLIENT_SECRET) url = f"{client.base_url}/v1/billing/plans" headers = { "Content-Type": "application/json", "Authorization": f"Bearer {client.access_token}" } payload = { "product_id": "PRODUCT_ID", # Create product first "name": name, "billing_cycles": [{ "frequency": { "interval_unit": interval, "interval_count": 1 }, "tenure_type": "REGULAR", "sequence": 1, "total_cycles": 0, # Infinite "pricing_scheme": { "fixed_price": { "value": str(amount), "currency_code": "USD" } } }], "payment_preferences": { "auto_bill_outstanding": True, "setup_fee": { "value": "0", "currency_code": "USD" }, "setup_fee_failure_action": "CONTINUE", "payment_failure_threshold": 3 } } response = requests.post(url, headers=headers, json=payload) return response.json()参数解析:
| 参数 | 取值/默认 | 含义 |
|---|---|---|
product_id | 需预先创建 | PayPal 要求先创建 Product(产品),再在计划中引用其 ID |
frequency.interval_unit | MONTH/YEAR/WEEK/DAY | 计费周期单位 |
frequency.interval_count | 整数 | 每个计费周期的单位数量 |
tenure_type | REGULAR(常规)/TRIAL(试用) | 计费期类型 |
total_cycles | 0表示无限期 | 该 tenure 的总周期数 |
auto_bill_outstanding | True | 是否自动补收欠款 |
setup_fee | 金额对象 | 一次性设置费,"0"表示免设置费 |
setup_fee_failure_action | CONTINUE | 设置费收取失败后的动作 |
payment_failure_threshold | 3 | 支付连续失败多少次后暂停订阅(与下方 dunning 思路呼应) |
4.2 为客户创建订阅并获取批准链接
def create_subscription(plan_id, subscriber_email): """Create a subscription for a customer.""" client = PayPalClient(CLIENT_ID, CLIENT_SECRET) url = f"{client.base_url}/v1/billing/subscriptions" headers = { "Content-Type": "application/json", "Authorization": f"Bearer {client.access_token}" } payload = { "plan_id": plan_id, "subscriber": { "email_address": subscriber_email }, "application_context": { "return_url": "https://yourdomain.com/subscription/success", "cancel_url": "https://yourdomain.com/subscription/cancel" } } response = requests.post(url, headers=headers, json=payload) subscription = response.json() # Get approval URL for link in subscription.get('links', []): if link['rel'] == 'approve': return { 'subscription_id': subscription['id'], 'approval_url': link['href'] }模式要点:订阅创建后同样返回一个links数组,其中rel == 'approve'的href是订阅批准页。后端应将用户重定向到该 URL;用户批准后,PayPal 回调return_url(携带subscription_id等参数),此时订阅才正式激活。return_url与cancel_url需要替换为业务方真实域名。
4.3 订阅生命周期与自动化计费
订阅本身不产生代码,但它与仓库中另一 Skill billing-automation 的订阅生命周期管理紧密配合。billing-automation 定义的典型状态机为:
trial → active → past_due → canceled → paused → resumed其BillingEngine.process_billing_cycle展示了完整的周期处理流程:判断是否到账期 → 生成发票 → 尝试扣款 → 成功则标记已付并推进账期,失败则标记past_due并进入 dunning(催缴)流程。而 PayPal 侧的payment_failure_threshold: 3与auto_bill_outstanding: True正是把「自动重试 + 失败上限」下沉到支付服务商侧的配置化实现,二者互为补充。
五、退款工作流:部分退款与全额退款
def create_refund(capture_id, amount=None, note=None): """Create a refund for a captured payment.""" client = PayPalClient(CLIENT_ID, CLIENT_SECRET) url = f"{client.base_url}/v2/payments/captures/{capture_id}/refund" headers = { "Content-Type": "application/json", "Authorization": f"Bearer {client.access_token}" } payload = {} if amount: payload["amount"] = { "value": str(amount), "currency_code": "USD" } if note: payload["note_to_payer"] = note response = requests.post(url, headers=headers, json=payload) return response.json() def get_refund_details(refund_id): """Get refund details.""" client = PayPalClient(CLIENT_ID, CLIENT_SECRET) url = f"{client.base_url}/v2/payments/refunds/{refund_id}" headers = { "Authorization": f"Bearer {client.access_token}" } response = requests.get(url, headers=headers) return response.json()- 退款以
capture_id(捕获交易 ID)为操作对象,调用/v2/payments/captures/{capture_id}/refund; amount与note_to_payer均为可选参数:不传amount即全额退款;传amount即部分退款;- 若只做全款退款,
payload可以保持为空对象({}),PayPal 默认退还全部捕获金额; - 退款完成后可用
get_refund_details按refund_id查询退款明细,用于对账与审计。
退款通常由两类场景触发:商户主动发起(本节的create_refund路径),以及支付服务商主动回调(上一节 IPN 的Refunded状态)。两者都需要在业务系统中记录refund_id与parent_txn_id的关联关系,保证幂等、防止重复退款。
六、统一错误处理:PayPalError 封装
class PayPalError(Exception): """Custom PayPal error.""" pass def handle_paypal_api_call(api_function): """Wrapper for PayPal API calls with error handling.""" try: result = api_function() return result except requests.exceptions.RequestException as e: # Network error raise PayPalError(f"Network error: {str(e)}") except Exception as e: # Other errors raise PayPalError(f"PayPal API error: {str(e)}") # Usage try: order = handle_paypal_api_call(lambda: client.create_order(25.00)) except PayPalError as e: # Handle error appropriately log_error(e)该封装的价值在于错误归一化:无论底层是网络异常(requests.exceptions.RequestException,如超时、连接失败、DNS 解析错误)还是 PayPal 返回的业务错误,统一包装为自定义PayPalError,业务层只需捕获一种异常类型即可统一处理(记录日志、重试、通知用户)。这与仓库智能体 payment-integration.md 中「Payment integration code with error handling」「实现所有支付操作的幂等性」「处理所有边界情况(支付失败、争议、退款)」的输出要求一致。
七、沙箱测试与上线迁移
SKILL.md 的 Testing 章节给出了完整的沙箱验证路径:
# Use sandbox credentials SANDBOX_CLIENT_ID = "..." SANDBOX_SECRET = "..." # Test accounts # Create test buyer and seller accounts at developer.paypal.com def test_payment_flow(): """Test complete payment flow.""" client = PayPalClient(SANDBOX_CLIENT_ID, SANDBOX_SECRET, mode='sandbox') # Create order order = client.create_order(10.00) assert 'id' in order # Get approval URL approval_url = next((link['href'] for link in order['links'] if link['rel'] == 'approve'), None) assert approval_url is not None # After approval (manual step with test account) # Capture order # captured = client.capture_order(order['id']) # assert captured['status'] == 'COMPLETED'测试与上线要点:
- 沙箱凭证:在 developer.paypal.com 创建沙箱应用获取
SANDBOX_CLIENT_ID/SANDBOX_SECRET,同时创建测试买家和卖家账户; - 全链路验证:订单创建(
assert 'id' in order)→ 批准链接存在(assert approval_url is not None)→ 用测试买家账户手动完成批准 → 服务端捕获(captured['status'] == 'COMPLETED'); - 环境隔离:
PayPalClient的mode参数即切换开关,生产环境必须使用mode='live'、生产 API 域(https://api-m.paypal.com)与真实凭证。参照 payment-integration.md 的要求,测试凭证必须确保在线上站点失效,防止测试卡被线上接受。
八、生产级 Checklist 汇总
结合details.md的模式与仓库配套文档,落地 PayPal 集成时建议逐项核对:
- OAuth 令牌管理:
access_token有有效期,长生命周期应用中应按官方建议缓存并在过期前刷新(当前示例为每次实例化时获取,生产应升级为带过期时间的缓存); - 金额处理:
value一律str()化,金额计算使用定点数避免浮点误差; - 服务端捕获:客户端只回传
orderID,捕获与状态校验必须发生在服务端; - IPN 回验:所有通知先回传
cmd=_notify-validate验证,VERIFIED才处理;生产端点替换为ipnpb.paypal.com; - 幂等去重:以
txn_id/parent_txn_id为键落库去重,webhook 重试与重复投递不会造成重复发货/重复退款; - 订阅失败策略:
payment_failure_threshold与服务端 dunning 流程联动,避免长期欠费; - 错误归一化:所有 PayPal 调用经
handle_paypal_api_call包装,业务层只捕获PayPalError; - 环境隔离:sandbox/live 的凭证、域名、回调地址全部环境变量化,测试与生产严格分离。
参考与延伸阅读
- Skill 入口与激活条件:plugins/payment-processing/skills/paypal-integration/SKILL.md
- 本文主体(详细模式与完整代码):plugins/payment-processing/skills/paypal-integration/references/details.md
- 支付集成智能体(安全要求与常见故障):plugins/payment-processing/agents/payment-integration.md
- 相关 Skill:Stripe 集成 stripe-integration、PCI DSS 合规 pci-compliance、订阅生命周期与催缴 billing-automation
- Skill 体系与渐进式披露说明:docs/agent-skills.md
- 插件安装方式:
/plugin install payment-processing(详见 docs/plugins.md)
【免费下载链接】agentsMulti-harness agentic plugin marketplace for Claude Code, Codex, Cursor, OpenCode, GitHub Copilot, and Google Antigravity项目地址: https://gitcode.com/GitHub_Trending/agents24/agents
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考