stripe-integration

Stripe 集成

通过明确的服务器端授权和重试边界，实现并验证 Stripe 结账、订阅、Webhook 和退款。

以下情况不要使用此技能

• 任务与 Stripe 集成无关
• 你需要此范围之外的其他领域或工具

说明

• 明确目标、约束条件和必需输入。
• 应用相关最佳实践并验证结果。
• 提供可执行的步骤和验证方法。
• 检查已安装的 Stripe SDK 以及固定的 API/Webhook 版本。此单文件技能没有捆绑的 playbook 或生产环境封装。

以下情况使用此技能

• 在 Web/移动应用中实现支付处理
• 搭建订阅计费系统
• 处理一次性支付和周期性扣款
• 处理退款和争议
• 管理客户的支付方式
• 为欧洲支付实现 SCA（强客户认证）
• 使用 Stripe Connect 构建市场平台支付流程

核心概念

1. 支付流程

Checkout Session（托管式）

• Stripe 托管的支付页面
• 减少直接处理卡数据
• 实现速度最快
• 支持一次性支付和周期性支付

Payment Intents（自定义 UI）

• 完全掌控支付 UI
• 使用 Stripe.js/Elements 避免直接处理原始卡数据
• 实现更复杂
• 提供更好的自定义选项

Setup Intents（保存支付方式）

• 收集支付方式而不扣款
• 用于订阅和未来支付
• 需要客户确认

2. Webhook

关键事件：

• payment_intent.succeeded：支付完成
• payment_intent.payment_failed：支付失败
• customer.subscription.updated：订阅发生变更
• customer.subscription.deleted：订阅已取消
• charge.refunded：退款已处理
• invoice.payment_succeeded：订阅支付成功

3. 订阅

组成部分：

• Product：你要销售的内容
• Price：金额和频率
• Subscription：客户的周期性支付
• Invoice：每个计费周期生成的账单

4. 客户管理

• 创建和管理客户记录
• 存储多个支付方式
• 跟踪客户元数据
• 管理账单详情

输入与安全边界

使用明确授权的 Stripe 测试账户/沙盒、服务器拥有的订单和客户记录、已安装的 SDK/API 版本以及预期的 Webhook 类型。金额、货币、价格/客户 ID 和退款权限必须来自经过身份验证的服务器策略，而非任意客户端参数。Checkout/Elements 可以减少卡数据暴露；它们本身并不能确立 PCI 合规。

这些代码片段是集成示意。测试密钥和 Webhook 签名密钥不同；请从项目密钥机制中加载它们，并且绝不要打印它们。仅阅读此技能并不授权任何实时支付、退款、客户更新或账户配置。

快速开始

import os
import stripe

stripe.api_key = os.environ["STRIPE_SECRET_KEY"]

# Create a checkout session
session = stripe.checkout.Session.create(
    payment_method_types=['card'],
    line_items=[{
        'price_data': {
            'currency': 'usd',
            'product_data': {
                'name': 'Premium Subscription',
            },
            'unit_amount': 2000,  # $20.00
            'recurring': {
                'interval': 'month',
            },
        },
        'quantity': 1,
    }],
    mode='subscription',
    success_url='https://yourdomain.com/success?session_id={CHECKOUT_SESSION_ID}',
    cancel_url='https://yourdomain.com/cancel',
)

# Redirect user to session.url
print(session.url)

支付实现模式

模式 1：一次性支付（托管式 Checkout）

def create_checkout_session(amount, order_attempt_id, currency='usd'):
    """Create a one-time payment checkout session."""
    try:
        session = stripe.checkout.Session.create(
            payment_method_types=['card'],
            line_items=[{
                'price_data': {
                    'currency': currency,
                    'product_data': {
                        'name': 'Purchase',
                        'images': ['https://example.com/product.jpg'],
                    },
                    'unit_amount': amount,  # Amount in cents
                },
                'quantity': 1,
            }],
            mode='payment',
            success_url='https://yourdomain.com/success?session_id={CHECKOUT_SESSION_ID}',
            cancel_url='https://yourdomain.com/cancel',
            metadata={
                'order_id': 'order_123',
                'user_id': 'user_456'
            },
            idempotency_key=order_attempt_id
        )
        return session
    except stripe.error.StripeError as e:
        # Handle error
        print(f"Stripe error: {e.user_message}")
        raise

模式 2：自定义 Payment Intent 流程

def create_payment_intent(amount, order_attempt_id, currency='usd', customer_id=None):
    """Create a payment intent for custom checkout UI."""
    intent = stripe.PaymentIntent.create(
        amount=amount,
        currency=currency,
        customer=customer_id,
        automatic_payment_methods={
            'enabled': True,
        },
        metadata={
            'integration_check': 'accept_a_payment'
        },
        idempotency_key=order_attempt_id
    )
    return intent.client_secret  # Only to the authenticated client for this order; never log it

# Frontend (JavaScript)
"""
const stripe = Stripe('pk_test_...');
const elements = stripe.elements();
const cardElement = elements.create('card');
cardElement.mount('#card-element');

const {error, paymentIntent} = await stripe.confirmCardPayment(
    clientSecret,
    {
        payment_method: {
            card: cardElement,
            billing_details: {
                name: 'Customer Name'
            }
        }
    }
);

if (error) {
    // Handle error
} else if (paymentIntent.status === 'succeeded') {
    // Update display only; server fulfillment still verifies payment state
}
"""

模式 3：订阅创建契约

使用账户所固定 API 版本对应的文档化流程。不要假设每个版本中都存在 latest_invoice.payment_intent，也不要假设每张账单都有一个可立即确认的支付。解析出经过授权的客户和允许的价格，使用幂等键创建未完成的订阅，并通过该版本的 API 处理返回的确认状态。根据经过验证的订阅/账单状态授予访问权限；测试试用、零金额账单、延迟支付、取消和重试。

参见 Stripe 订阅集成。下面的客户门户在接受客户 ID 之前同样需要所有权检查。

模式 4：客户门户

def create_customer_portal_session(customer_id):
    """Create a portal session for customers to manage subscriptions."""
    session = stripe.billing_portal.Session.create(
        customer=customer_id,
        return_url='https://yourdomain.com/account',
    )
    return session.url  # Redirect customer here

Webhook 处理

安全的 Webhook 端点

import os
from flask import Flask, request
import stripe

app = Flask(__name__)

endpoint_secret = os.environ["STRIPE_WEBHOOK_SECRET"]

@app.route('/webhook', methods=['POST'])
def webhook():
    payload = request.get_data(cache=False)  # Exact raw bytes; enforce an ingress body-size limit
    sig_header = request.headers.get('Stripe-Signature')
    if not sig_header:
        return 'Missing signature', 400

    try:
        event = stripe.Webhook.construct_event(
            payload, sig_header, endpoint_secret
        )
    except ValueError:
        # Invalid payload
        return 'Invalid payload', 400
    except stripe.SignatureVerificationError:
        # Invalid signature
        return 'Invalid signature', 400

    # Integration sketch: durable deduplication/queueing below must precede effects.
    # Bind account, mode, order, amount/currency and expected current state first.
    # Handle the event
    if event['type'] == 'payment_intent.succeeded':
        payment_intent = event['data']['object']
        handle_successful_payment(payment_intent)
    elif event['type'] == 'payment_intent.payment_failed':
        payment_intent = event['data']['object']
        handle_failed_payment(payment_intent)
    elif event['type'] == 'customer.subscription.deleted':
        subscription = event['data']['object']
        handle_subscription_canceled(subscription)

    return 'Success', 200

def handle_successful_payment(payment_intent):
    """Process successful payment."""
    customer_id = payment_intent.get('customer')
    amount = payment_intent['amount']
    metadata = payment_intent.get('metadata', {})

    # Update your database
    # Send confirmation email
    # Fulfill order
    print(f"Payment succeeded: {payment_intent['id']}")

def handle_failed_payment(payment_intent):
    """Handle failed payment."""
    error = payment_intent.get('last_payment_error', {})
    print(f"Payment failed: {error.get('message')}")
    # Notify customer
    # Update order status

def handle_subscription_canceled(subscription):
    """Handle subscription cancellation."""
    customer_id = subscription['customer']
    # Update user access
    # Send cancellation email
    print(f"Subscription canceled: {subscription['id']}")

签名、重复与履约

使用 stripe.Webhook.construct_event，传入原始字节、完整的 Stripe-Signature 标头和正确的端点密钥。仅对请求体做裸 HMAC 并不符合 Stripe 带时间戳的标头格式或重放容忍机制。保留 SDK 的时间戳校验，并同时验证无效签名和过期签名。参见 Stripe webhook 文档。

Verify signature and event/account/mode before accepting the event.
Atomically insert event.id into a durable inbox with a unique constraint.
Return 2xx only after durable acceptance; retryable storage failure remains a failure.
A worker reconciles current payment/order state and applies a guarded transition.
Use a unique order/fulfillment key as well: different events can describe the same payment.
Commit the state change and an outbox entry together; downstream effects are idempotent.
Duplicate/concurrent delivery returns the recorded result without fulfilling twice.

先检查、再处理、后标记的序列并非原子操作，无法提供恰好一次的效果。事件可能乱序到达。上面的 Flask 处理器示例只是占位符，不是完整的持久化处理器；不要将它们作为履约逻辑部署。结账成功重定向只是 UI 信号，并非付款凭证。请参阅履约指南和请求幂等性。

客户管理

def create_customer(email, name, payment_method_id=None):
    """Create a Stripe customer."""
    customer = stripe.Customer.create(
        email=email,
        name=name,
        payment_method=payment_method_id,
        invoice_settings={
            'default_payment_method': payment_method_id
        } if payment_method_id else None,
        metadata={
            'user_id': '12345'
        }
    )
    return customer

def attach_payment_method(customer_id, payment_method_id):
    """Attach a payment method to a customer."""
    stripe.PaymentMethod.attach(
        payment_method_id,
        customer=customer_id
    )

    # Set as default
    stripe.Customer.modify(
        customer_id,
        invoice_settings={
            'default_payment_method': payment_method_id
        }
    )

def list_customer_payment_methods(customer_id):
    """List all payment methods for a customer."""
    payment_methods = stripe.PaymentMethod.list(
        customer=customer_id,
        type='card'
    )
    return payment_methods.data

退款处理

def create_refund(payment_intent_id, refund_attempt_id, amount=None, reason=None):
    """Create a refund."""
    refund_params = {
        'payment_intent': payment_intent_id
    }

    if amount is not None:
        if type(amount) is not int or amount <= 0:
            raise ValueError("Refund amount must be a positive minor-unit integer")
        refund_params['amount'] = amount  # Partial refund

    if reason:
        refund_params['reason'] = reason  # 'duplicate', 'fraudulent', 'requested_by_customer'

    refund = stripe.Refund.create(**refund_params, idempotency_key=refund_attempt_id)
    return refund

def handle_dispute(dispute_id, evidence):
    """Update dispute with evidence."""
    stripe.Dispute.modify(
        dispute_id,
        evidence={
            'customer_name': evidence.get('customer_name'),
            'customer_email_address': evidence.get('customer_email'),
            'shipping_documentation': evidence.get('shipping_proof'),
            'customer_communication': evidence.get('communication'),
        }
    )

测试

# Use test mode keys
import os

stripe.api_key = os.environ["STRIPE_SECRET_KEY"]

# Test card numbers
TEST_CARDS = {
    'success': '4242424242424242',
    'declined': '4000000000000002',
    '3d_secure': '4000002500003155',
    'insufficient_funds': '4000000000009995'
}

def test_payment_flow():
    """Test complete payment flow."""
    # Create test customer
    customer = stripe.Customer.create(
        email="[email protected]"
    )

    # Create payment intent
    intent = stripe.PaymentIntent.create(
        amount=1000,
        currency='usd',
        customer=customer.id,
        payment_method_types=['card']
    )

    # Confirm with test card
    confirmed = stripe.PaymentIntent.confirm(
        intent.id,
        payment_method='pm_card_visa'  # Test payment method
    )

    assert confirmed.status == 'succeeded'

完整验证用例

对于一个已授权的测试订单，并发地投递两次相同的有效支付事件，然后投递一个过期/无效签名和一个更早的乱序状态事件。预期结果：一次持久化履约、无效签名被拒绝，且不会回滚更新的状态。在 inbox 提交之前模拟存储故障；它不得返回一个会丢失事件的成功响应。断言零金额退款被拒绝，而不是静默地变成全额退款。

这些是要在项目中实现的验收检查。本技能不记录任何真实交易结果，也不包含隐藏的客户端封装或额外参考文件。

最佳实践

1. 始终使用 Webhook：不要仅依赖客户端确认
2. 幂等性：以幂等方式处理 webhook 事件
3. 错误处理：优雅地处理所有 Stripe 错误
4. 测试模式：在生产环境之前使用测试密钥彻底测试
5. 元数据：使用元数据将 Stripe 对象关联到你的数据库
6. 监控：跟踪支付成功率和错误
7. PCI 合规：绝不在你的服务器上处理原始卡数据
8. SCA 就绪：为欧洲支付实现 3D Secure

常见陷阱

• 不验证 Webhook：始终验证 webhook 签名
• 遗漏 Webhook 事件：处理所有相关的 webhook 事件
• 硬编码金额：使用分/最小货币单位
• 没有重试逻辑：为 API 调用实现重试
• 忽视测试模式：使用测试卡测试所有边界情况

局限性

• API 版本、事件负载和 SDK 异常命名空间各不相同；请验证已安装的版本，而不是混合来自不同发行版的示例。
• 请求幂等键必须绑定到相同的逻辑操作和参数；每次重试使用新密钥可能会重复扣款或退款。
• 仅靠 webhook 去重无法防止不同事件导致的重复业务效果。
• 客户端成功、测试卡成功和签名检查并不能确立完整的履约、税务、合规或订阅正确性。
• 客户、支付方式、争议和退款变更需要服务端所有权/角色检查以及明确的任务授权。