agents 插件市场 billing-automation 技能全解构建生产级订阅计费自动化系统【免费下载链接】agentsMulti-harness agentic plugin marketplace for Claude Code, Codex, Cursor, OpenCode, GitHub Copilot, and Google Antigravity项目地址: https://gitcode.com/GitHub_Trending/agents24/agents本篇文章围绕 agents 项目Multi-harness agentic plugin marketplace中 payment-processing 插件的billing-automation技能展开完整解读该技能提供的计费自动化模式从订阅生命周期管理、账单周期处理、催收dunning、按比例分摊proration、税务计算、发票生成到基于用量的计费usage-based billing。读完本文你将掌握一套可落地的 Python 计费引擎设计骨架并了解该技能在项目中的定位、安装方式以及与同插件支付集成技能的协同关系。技能定位与适用场景billing-automation是位于 plugins/payment-processing/skills/billing-automation/SKILL.md 的技能文件。其 frontmatter 给出的定位是--- name: billing-automation description: Build automated billing systems for recurring payments, invoicing, subscription lifecycle, and dunning management. Use when implementing subscription billing, automating invoicing, or managing recurring payment systems. ---即构建面向周期性付款、发票、订阅生命周期与催收管理的自动化计费系统。技能描述中明确其触发场景包括实现 SaaS 订阅计费自动化发票生成与投递管理失败支付恢复dunning计算套餐变更的按比例分摊费用处理销售税、VAT 与 GST处理基于用量的计费管理计费周期与续订作为插件体系中的技能skill组件它遵循项目的渐进式披露progressive disclosure设计SKILL.md 是导航层给出核心概念与快速上手更深层的完整模式与可运行示例存放在同目录的 references/details.md 中当导航层不足以覆盖需求时再加载。这正是该项目按需加载、节省上下文的插件设计原则的体现。核心概念四个必须理解的计费模型SKILL.md 用四个核心概念搭建了整个技能的知识骨架理解它们是实现任何计费系统的前提。1. 计费周期Billing Cycles订阅产品的价格锚定在固定的时间间隔上。技能给出了常见周期Monthly月付SaaS 中最常见的默认周期Annual年付通常配合折扣用于长期锁定Quarterly季付Weekly周付Custom自定义按用量usage-based或按席位per-seat计费周期的选择直接影响后续的续订日期计算见订阅生命周期管理中calculate_next_billing_date的实现。2. 订阅状态机Subscription States技能给出了订阅的核心状态流转图trial → active → past_due → canceled → paused → resumedtrial试用期通常可免费使用并自动记录试用截止时间active正常计费的活跃状态past_due支付失败后进入逾期状态等待催收流程处理paused/resumed暂停与恢复常用于账户冻结或用户主动暂停canceled终止支持周期末生效at period end与立即生效两种取消语义3. 催收管理Dunning ManagementDunning 是指对失败支付进行自动化恢复的流程技能归纳了四个组成部分Retry schedules重试计划按时间间隔多次尝试重新扣款Customer notifications客户通知每次重试前后向客户发送提醒邮件Grace periods宽限期在限制账户前留给客户修正支付方式的时间窗口Account restrictions账户限制多次失败后降级或限制账户功能4. 按比例分摊Proration当客户在计费周期中途变更套餐时需要对费用进行按天折算典型场景包括周期中途升级/降级套餐增加/移除席位变更计费频率快速上手最小可用计费引擎SKILL.md 提供了一个极简的入门示例SKILL.md展示了该技能推荐的顶层抽象——BillingEngine与Subscriptionfrom billing import BillingEngine, Subscription # Initialize billing engine billing BillingEngine() # Create subscription subscription billing.create_subscription( customer_idcus_123, plan_idplan_pro_monthly, billing_cycle_anchordatetime.now(), trial_days14 ) # Process billing cycle billing.process_billing_cycle(subscription.id)这段代码揭示了计费引擎的三个关键入口create_subscription创建订阅支持billing_cycle_anchor计费周期锚点决定每月扣款日与trial_days试用天数process_billing_cycle驱动单个订阅的完整计费流程生成发票 → 扣款 → 状态流转 → 失败进入催收Subscription/BillingEngine分别承载订阅状态与计费流程逻辑这也与后续 details.md 中两个类的职责划分一一对应注意示例中的billing模块是技能给出的抽象示意实际项目中需要你结合具体支付网关如 Stripe/PayPal实现BillingEngine的底层能力。同插件的 stripe-integration/SKILL.md 与 paypal-integration/SKILL.md 正是这些能力的实现参考。订阅生命周期管理状态机落地details.md 首先给出了完整的订阅生命周期实现details.md。它用SubscriptionStatus枚举严格约束状态集合并用Subscription类封装状态转换from datetime import datetime, timedelta from enum import Enum class SubscriptionStatus(Enum): TRIAL trial ACTIVE active PAST_DUE past_due CANCELED canceled PAUSED paused class Subscription: def __init__(self, customer_id, plan, billing_cycle_dayNone): self.id generate_id() self.customer_id customer_id self.plan plan self.status SubscriptionStatus.TRIAL self.current_period_start datetime.now() self.current_period_end self.current_period_start timedelta(daysplan.trial_days or 30) self.billing_cycle_day billing_cycle_day or self.current_period_start.day self.trial_end datetime.now() timedelta(daysplan.trial_days) if plan.trial_days else None def start_trial(self, trial_days): Start trial period. self.status SubscriptionStatus.TRIAL self.trial_end datetime.now() timedelta(daystrial_days) self.current_period_end self.trial_end def activate(self): Activate subscription after trial or immediately. self.status SubscriptionStatus.ACTIVE self.current_period_start datetime.now() self.current_period_end self.calculate_next_billing_date() def mark_past_due(self): Mark subscription as past due after failed payment. self.status SubscriptionStatus.PAST_DUE # Trigger dunning workflow def cancel(self, at_period_endTrue): Cancel subscription. if at_period_end: self.cancel_at_period_end True # Will cancel when current period ends else: self.status SubscriptionStatus.CANCELED self.canceled_at datetime.now() def calculate_next_billing_date(self): Calculate next billing date based on interval. if self.plan.interval month: return self.current_period_start timedelta(days30) elif self.plan.interval year: return self.current_period_start timedelta(days365) elif self.plan.interval week: return self.current_period_start timedelta(days7)值得注意的工程细节周期锚点billing_cycle_day以创建日的日为锚点保证每月在固定日期扣款未显式传入时默认取当前日试用期与周期打通试用期间current_period_end直接对齐trial_end试用结束即触发计费取消语义默认at_period_endTrue周期末生效客户可继续用到本周期结束False则立即取消并记录canceled_at状态转换留口mark_past_due中注释Trigger dunning workflow将逾期处理委托给独立的催收管理器保持订阅类职责单一实现时注意generate_id()与plan对象含trial_days、interval、amount、pricing_model等字段在示例中为示意你需要用自己的 ID 生成器与定价模型补齐。账单周期处理BillingEngine 主循环BillingEngine.process_billing_cycle是整个计费系统的心跳details.md。它定义了每个计费周期执行的标准流程class BillingEngine: def process_billing_cycle(self, subscription_id): Process billing for a subscription. subscription self.get_subscription(subscription_id) # Check if billing is due if datetime.now() subscription.current_period_end: return # Generate invoice invoice self.generate_invoice(subscription) # Attempt payment payment_result self.charge_customer( subscription.customer_id, invoice.total ) if payment_result.success: # Payment successful invoice.mark_paid() subscription.advance_billing_period() self.send_invoice(invoice) else: # Payment failed subscription.mark_past_due() self.start_dunning_process(subscription, invoice) def generate_invoice(self, subscription): Generate invoice for billing period. invoice Invoice( customer_idsubscription.customer_id, subscription_idsubscription.id, period_startsubscription.current_period_start, period_endsubscription.current_period_end ) # Add subscription line item invoice.add_line_item( descriptionsubscription.plan.name, amountsubscription.plan.amount, quantitysubscription.quantity or 1 ) # Add usage-based charges if applicable if subscription.has_usage_billing: usage_charges self.calculate_usage_charges(subscription) invoice.add_line_item( descriptionUsage charges, amountusage_charges ) # Calculate tax tax self.calculate_tax(invoice.subtotal, subscription.customer) invoice.tax tax invoice.finalize() return invoice def charge_customer(self, customer_id, amount): Charge customer using saved payment method. customer self.get_customer(customer_id) try: # Charge using payment processor charge stripe.Charge.create( customercustomer.stripe_id, amountint(amount * 100), # Convert to cents currencyusd ) return PaymentResult(successTrue, transaction_idcharge.id) except stripe.error.CardError as e: return PaymentResult(successFalse, errorstr(e))该流程的关键决策点到期判断datetime.now() current_period_end时直接返回保证流程可被定时任务反复调用而不产生重复扣费天然幂等的外层判断发票先行先generate_invoice再扣款发票作为扣款金额的唯一来源包含订阅行项目、用量行项目与税款金额换算int(amount * 100)将元换算为分避免浮点精度问题——这是支付网关集成中的通用约定成败分流成功则mark_paidadvance_billing_period推进到下一计费周期 投递发票失败则mark_past_due并启动催收流程网关异常隔离仅捕获stripe.error.CardError卡被拒等业务性失败其余异常向上抛出避免把系统故障误判为客户欠费催收管理DunningManager 的失败支付恢复当支付失败后DunningManager接管恢复流程details.md。其核心是重试计划 分级通知 最终处置class DunningManager: Manage failed payment recovery. def __init__(self): self.retry_schedule [ {days: 3, email_template: payment_failed_first}, {days: 7, email_template: payment_failed_reminder}, {days: 14, email_template: payment_failed_final} ] def start_dunning_process(self, subscription, invoice): Start dunning process for failed payment. dunning_attempt DunningAttempt( subscription_idsubscription.id, invoice_idinvoice.id, attempt_number1, next_retrydatetime.now() timedelta(days3) ) # Send initial failure notification self.send_dunning_email(subscription, payment_failed_first) # Schedule retries self.schedule_retries(dunning_attempt) def retry_payment(self, dunning_attempt): Retry failed payment. subscription self.get_subscription(dunning_attempt.subscription_id) invoice self.get_invoice(dunning_attempt.invoice_id) # Attempt payment again result self.charge_customer(subscription.customer_id, invoice.total) if result.success: # Payment succeeded invoice.mark_paid() subscription.status SubscriptionStatus.ACTIVE self.send_dunning_email(subscription, payment_recovered) dunning_attempt.mark_resolved() else: # Still failing dunning_attempt.attempt_number 1 if dunning_attempt.attempt_number len(self.retry_schedule): # Schedule next retry next_retry_config self.retry_schedule[dunning_attempt.attempt_number] dunning_attempt.next_retry datetime.now() timedelta(daysnext_retry_config[days]) self.send_dunning_email(subscription, next_retry_config[email_template]) else: # Exhausted retries, cancel subscription subscription.cancel(at_period_endFalse) self.send_dunning_email(subscription, subscription_canceled) def send_dunning_email(self, subscription, template): Send dunning notification to customer. customer self.get_customer(subscription.customer_id) email_content self.render_template(template, { customer_name: customer.name, amount_due: subscription.plan.amount, update_payment_url: fhttps://app.example.com/billing }) send_email( tocustomer.email, subjectemail_content[subject], bodyemail_content[body] )从源码结构可以提炼出催收管理的三个设计要点可配置的重试节奏retry_schedule以第 3 / 7 / 14 天三级递进每一级对应不同措辞的邮件模板首次失败 → 提醒 → 最终警告实际节奏可按业务调整明确的终止策略重试耗尽后cancel(at_period_endFalse)立即取消订阅防止无限重试造成网关费用与客户骚扰恢复闭环任一次重试成功后立刻mark_paid、将订阅状态置回ACTIVE、发送payment_recovered通知并标记催收单已解决保证状态机收敛邮件模板渲染render_template与send_email在示例中为占位实现生产环境可替换为你的邮件服务如 SES/SendGrid/内部邮件 API。按比例分摊ProrationCalculator 的两类折算套餐变更与席位增减的公平计费由ProrationCalculator负责details.mdclass ProrationCalculator: Calculate prorated charges for plan changes. staticmethod def calculate_proration(old_plan, new_plan, period_start, period_end, change_date): Calculate proration for plan change. # Days in current period total_days (period_end - period_start).days # Days used on old plan days_used (change_date - period_start).days # Days remaining on new plan days_remaining (period_end - change_date).days # Calculate prorated amounts unused_amount (old_plan.amount / total_days) * days_remaining new_plan_amount (new_plan.amount / total_days) * days_remaining # Net charge/credit proration new_plan_amount - unused_amount return { old_plan_credit: -unused_amount, new_plan_charge: new_plan_amount, net_proration: proration, days_used: days_used, days_remaining: days_remaining } staticmethod def calculate_seat_proration(current_seats, new_seats, price_per_seat, period_start, period_end, change_date): Calculate proration for seat changes. total_days (period_end - period_start).days days_remaining (period_end - change_date).days # Additional seats charge additional_seats new_seats - current_seats prorated_amount (additional_seats * price_per_seat / total_days) * days_remaining return { additional_seats: additional_seats, prorated_charge: max(0, prorated_amount), # No refund for removing seats mid-cycle effective_date: change_date }两个方法的业务含义套餐切换分摊以按天单价 × 剩余天数分别计算旧套餐的未使用部分记为负值 credit与新套餐的剩余部分正值 chargenet_proration为正表示补差价、为负表示退款。返回值中同时给出days_used与days_remaining供审计席位变更分摊仅对新增席位按剩余天数收费max(0, ...)明确注释了周期中途减少席位不退款的产品策略——这是许多 SaaS 的通行做法你可根据业务决定是否调整计算时需注意(period_end - period_start).days依赖datetime的日期差语义若计费周期横跨月末建议基于周期锚点如第 1 天 / 第 15 天而非固定 30 天来保证精度。税务计算TaxCalculator 的多法域支持面向全球客户的订阅系统需要同时处理销售税、VAT 与 GSTTaxCalculator给出了法域判定与税率映射的实现details.mdclass TaxCalculator: Calculate sales tax, VAT, GST. def __init__(self): # Tax rates by region self.tax_rates { US_CA: 0.0725, # California sales tax US_NY: 0.04, # New York sales tax GB: 0.20, # UK VAT DE: 0.19, # Germany VAT FR: 0.20, # France VAT AU: 0.10, # Australia GST } def calculate_tax(self, amount, customer): Calculate applicable tax. # Determine tax jurisdiction jurisdiction self.get_tax_jurisdiction(customer) if not jurisdiction: return 0 # Get tax rate tax_rate self.tax_rates.get(jurisdiction, 0) # Calculate tax tax amount * tax_rate return { tax_amount: tax, tax_rate: tax_rate, jurisdiction: jurisdiction, tax_type: self.get_tax_type(jurisdiction) } def get_tax_jurisdiction(self, customer): Determine tax jurisdiction based on customer location. if customer.country US: # US: Tax based on customer state return fUS_{customer.state} elif customer.country in [GB, DE, FR]: # EU: VAT return customer.country elif customer.country AU: # Australia: GST return AU else: return None def get_tax_type(self, jurisdiction): Get type of tax for jurisdiction. if jurisdiction.startswith(US_): return Sales Tax elif jurisdiction in [GB, DE, FR]: return VAT elif jurisdiction AU: return GST return Tax def validate_vat_number(self, vat_number, country): Validate EU VAT number. # Use VIES API for validation # Returns True if valid, False otherwise pass该实现的关键逻辑法域判定规则美国按国家 州二级定位如US_CA欧盟与澳大利亚按国家定位无法判定的法域返回0不征税税率配置集中化tax_rates字典是唯一的税率事实来源便于随税率调整更新税种区分get_tax_type根据法域返回Sales Tax/VAT/GST供发票展示与申报区分B2B 校验留口validate_vat_number注释标明可对接 VIES API 验证欧盟 VAT 号B2B 场景常需豁免增值税方法体为占位实现需要强调示例税率是文档编写时的静态参考值实际生产中税率会随政策变化应接入税务服务或定期更新配置。发票生成Invoice 的完整生命周期Invoice类details.md定义了发票从草稿到已支付的完整状态流转并提供 HTML/PDF 两种输出能力class Invoice: def __init__(self, customer_id, subscription_idNone): self.id generate_invoice_number() self.customer_id customer_id self.subscription_id subscription_id self.status draft self.line_items [] self.subtotal 0 self.tax 0 self.total 0 self.created_at datetime.now() def add_line_item(self, description, amount, quantity1): Add line item to invoice. line_item { description: description, unit_amount: amount, quantity: quantity, total: amount * quantity } self.line_items.append(line_item) self.subtotal line_item[total] def finalize(self): Finalize invoice and calculate total. self.total self.subtotal self.tax self.status open self.finalized_at datetime.now() def mark_paid(self): Mark invoice as paid. self.status paid self.paid_at datetime.now() def to_pdf(self): Generate PDF invoice. from reportlab.pdfgen import canvas # Generate PDF # Include: company info, customer info, line items, tax, total pass def to_html(self): Generate HTML invoice. template !DOCTYPE html html headtitleInvoice #{invoice_number}/title/head body h1Invoice #{invoice_number}/h1 pDate: {date}/p h2Bill To:/h2 p{customer_name}br{customer_address}/p table trthDescription/ththQuantity/ththAmount/th/tr {line_items} /table pSubtotal: ${subtotal}/p pTax: ${tax}/p h3Total: ${total}/h3 /body /html return template.format( invoice_numberself.id, dateself.created_at.strftime(%Y-%m-%d), customer_nameself.customer.name, customer_addressself.customer.address, line_itemsself.render_line_items(), subtotalself.subtotal, taxself.tax, totalself.total )发票状态机为draft → open → paid对应明细录入 → 冻结金额 → 收款完成。设计要点行项目模型add_line_item以unit_amount × quantity累加小计与账单周期处理中的generate_invoice配合使用金额分段subtotal / tax / total三段分离finalize()时才合成总额并打上finalized_at时间戳防止中途变更双输出通道to_html给出可直接模板化的 HTML 发票to_pdf使用reportlab并注释了应包含的内容清单公司信息、客户信息、行项目、税、总额基于用量的计费UsageBillingEngine 与分档定价对于按量付费usage-based billing产品UsageBillingEngine提供了用量追踪 → 周期汇总 → 分档定价的完整链路details.mdclass UsageBillingEngine: Track and bill for usage. def track_usage(self, customer_id, metric, quantity): Track usage event. UsageRecord.create( customer_idcustomer_id, metricmetric, quantityquantity, timestampdatetime.now() ) def calculate_usage_charges(self, subscription, period_start, period_end): Calculate charges for usage in billing period. usage_records UsageRecord.get_for_period( subscription.customer_id, period_start, period_end ) total_usage sum(record.quantity for record in usage_records) # Tiered pricing if subscription.plan.pricing_model tiered: charge self.calculate_tiered_pricing(total_usage, subscription.plan.tiers) # Per-unit pricing elif subscription.plan.pricing_model per_unit: charge total_usage * subscription.plan.unit_price # Volume pricing elif subscription.plan.pricing_model volume: charge self.calculate_volume_pricing(total_usage, subscription.plan.tiers) return charge def calculate_tiered_pricing(self, total_usage, tiers): Calculate cost using tiered pricing. charge 0 remaining total_usage for tier in sorted(tiers, keylambda x: x[up_to]): tier_usage min(remaining, tier[up_to] - tier[from]) charge tier_usage * tier[unit_price] remaining - tier_usage if remaining 0: break return charge该引擎支持三种定价模型由plan.pricing_model字段驱动per_unit按单位总用量 × 单价最简单直接tiered分档不同用量区间不同单价如前 1000 次 $0.01/次之后 $0.008/次。实现按up_to升序遍历分档用remaining逐档消化用量volume阶梯总量按总用量所在区间整段计价calculate_volume_pricing为占位方法逻辑与 tiered 的差异在于是否整段套用单档单价注意UsageRecord.create与UsageRecord.get_for_period在示例中为 ORM 风格占位你需要接入自己的用量事件存储如埋点数据库、数据仓库或专门的用量 API。同时用量计费通常与账单周期处理中的has_usage_billing标志配合周期结算时把用量费用作为独立行项目追加到发票上。与同插件技能的协同从计费到收款的完整闭环billing-automation技能解决的是账单怎么算、怎么催而真正把钱收回来还需要支付网关能力。在 payment-processing 插件中这两者天然互补payment-integration.mdagent定义支付集成专家的行为准则其中与计费强相关的要求包括所有支付操作必须实现幂等性、全面处理失败支付/争议/退款等边界情况、先测试环境再迁移生产、对异步事件做完善的 webhook 处理stripe-integration/SKILL.md 提供 Stripe 的订阅组件模型Product / Price / Subscription / Invoice与关键 webhook 事件如payment_intent.payment_failed、invoice.payment_succeeded这些事件正是驱动BillingEngine中扣款成功/失败分流与DunningManager重试的实际信号源paypal-integration/SKILL.md 提供 PayPal 订阅与 IPN 通知的对接模式可作为替代或补充支付渠道pci-compliance/SKILL.md 强调服务器永不接触原始卡号、使用 token 化 API这意味着 billing-automation 中的charge_customer应当基于 token/已保存支付方式而非卡号明文扣款一个典型的落地组合是billing-automation定义计费编排与状态机 →stripe-integration提供stripe.Charge.create/checkout.Session等真实扣款调用 → webhook 回调驱动状态更新 → 失败支付交给DunningManager按重试计划恢复全程遵守 payment-integration agent 的幂等与安全要求。在本仓库中的安装与使用billing-automation技能属于payment-processing插件。根据 docs/plugins.md 的说明插件采用渐进式披露设计skills/是可选的模块化知识包仅在需要时激活。方式一安装整个插件适用于 Claude Code 等原生支持插件市场的 harness/plugin marketplace add wshobson/agents /plugin install payment-processing安装后插件会将其agents/、commands/、skills/组件加载进上下文其中就包含billing-automation技能。方式二仅安装单个技能适用于任意 Agent不加载 agent/command/hook。项目 README 与 docs/harnesses.md 提到可通过 Agent Skills 安装器如 GitHub CLI 的gh skill、vercel-labs 的npx skills直接从plugins/*/skills/读取并安装到目标 Agent支持--agent claude-code等目标参数与选择器配置。将示例中的技能名替换为billing-automation即可单独安装。需要说明的是该技能在仓库中是文档型技能SKILL.md references/details.md即面向 Agent 的领域知识包本身不包含可执行的 Python 包其中的from billing import BillingEngine需要你在自己的代码库中依据本文梳理的类设计来实现。你也可以参考仓库中其他带可执行代码的技能如 plugin-eval/scripts/eval_all.py来理解本仓库技能文档 可执行脚本的混合组织方式。工程化实践建议基于本技能的全部模式将计费系统落地生产时建议注意以下几点以状态机为单一事实来源所有订阅状态变更收敛到SubscriptionStatus枚举与Subscription的方法中避免散落的if/else造成状态不一致金额统一用最小货币单位遵循charge_customer中int(amount * 100)的做法避免浮点误差重试与定时任务要幂等process_billing_cycle的到期判断天然支持重复调用催收重试建议记录next_retry并用后台任务扫描到期项网关事件驱动状态更新把 webhook如 Stripe 的invoice.payment_succeeded/payment_intent.payment_failed接入状态流转与本地定时结算互为校验合规先行卡号处理遵循 pci-compliance/SKILL.md 的 token 化与数据最小化原则税务配置保持可更新可观测与审计保留days_used / days_remaining、finalized_at / paid_at等时间与明细字段为对账和审计提供依据综上billing-automation技能以订阅状态机 周期结算主循环 催收恢复 分摊/税务/发票/用量计费为骨架给出了一个完整、可参照的计费自动化实现路径。结合 payment-processing 插件内的支付集成技能你可以用它快速搭建一套结构清晰、可演进的生产级订阅计费系统。【免费下载链接】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），仅供参考