Billing Stripe¶
billing_stripe — mvp.billing_stripe
Cluster: Core Infrastructure | Type: component | MCP Tools: None
Overview¶
Pure Python Stripe SDK wrapper providing plan definitions, checkout session creation, customer portal management, webhook signature verification, local duplicate-event detection, credit budget-cap checking, and hardware fingerprinting for device licensing. Defines G6 pricing tiers (free, Researcher, Builder, PAYG, credits) with feature gating for subscription-only capabilities. No Django or web framework imports required.
Production Webhook Idempotency¶
process_webhook_event() exposes a duplicate flag backed by a process-local in-memory store. This is useful for local development, tests, and single-process callers, but it is not sufficient as the only production idempotency control in a multi-worker deployment.
For production Stripe webhooks, persist Stripe event IDs in a shared store before applying side effects. The Django billing webhook does this with the WebhookEvent database table and a unique stripe_event_id, which is the production-safe path. If another service calls process_webhook_event() directly, it must pass a shared idempotency store or perform equivalent database/Redis deduplication around billing side effects.
Do not rely on the pure component's default in-memory idempotency store across multiple web workers, container replicas, restarts, or rolling deploys.
Operator-only Introspection Surface¶
BillingStripeBlock exposes read-only introspection ops in addition to the four curated write/action ops (create_session, portal_url, webhook, get_info). These ops surface billing facts that previously required importing the helper modules directly: list_tiers, describe_tier, resolve_price_id, recommend_tier, estimate_monthly_cost, check_feature_available, check_budget_cap, check_stripe_mode, validate_billing_config, describe_webhook_contract, idempotency_status, and describe_billing_capabilities.
These ops are operator-only and fail closed: the caller must pass operator_authority=True, otherwise the block returns G6_E_BILLING_UNAUTHORIZED_SURFACE. They are excluded from generic discovery (get_info lists only the four base ops, and the fuzzy unknown-op suggester never reveals them), consistent with PA-33's "forbidden end-user capability" posture. They are read-only: none makes a Stripe network call, moves money, or verifies a webhook signature, and none echoes STRIPE_SECRET_KEY or a raw Stripe price-id value — only presence booleans and posture flags.
Webhook Entitlement Boundary (blocked-escalated)¶
process_webhook_event() verifies signatures and returns a normalized event summary only; it applies no entitlement side effects. The downstream component that owns subscription activation/deactivation, credit grants, and customer-state updates after a verified webhook event is not yet named as a concrete component. describe_webhook_contract reports this as downstream_entitlement_owner: null with status blocked-escalated until that product decision is made.
Public API¶
BillingStripeInput(BaseModel)¶
| Field | Type | Default |
|---|---|---|
op | str | required |
parameters | dict[str, Any] | Field(default_factory=dict) |
BillingStripeOutput(BaseModel)¶
Normalized billing envelope.
| Field | Type | Default |
|---|---|---|
op | str | '' |
result | dict[str, Any] | Field(default_factory=dict) |
message | str | '' |
next_steps | list[str] | Field(default_factory=list) |
ok | bool | True |
completion_state | str | 'qualified-draft' |
degraded | bool | False |
degradation_reason | str \| None | None |
code | str \| None | None |
backend | str | 'stripe' |
stripe_mode | str \| None | None |
requires_operator_authority | bool | False |
idempotency_backend | str \| None | None |
duplicate | bool | False |
evidence | dict[str, Any] | Field(default_factory=dict) |
warning_card | dict[str, Any] \| None | None |
request_id | str \| None | None |
task_id | str \| None | None |
run_id | str \| None | None |
BillingStripeBlock(AIBlock)¶
AIBlock wrapper for Stripe billing: sessions, portal URLs, and webhooks.
Methods:
infer(input: BillingStripeInput) -> Result[BillingStripeOutput]¶
Functions¶
estimate_monthly_cost(tool_calls: int, tier: str, include_credit_pack_rounding: bool = True) -> dict¶
Estimate monthly AUD cents for a tier at a given tool-call volume.
calculate_break_even(subscription_tier: str = 'basic') -> dict¶
Return the PAYG call volume where a subscription becomes cheaper.
recommend_tier(tool_calls: int, requested_features: set[str] | None = None, devices: int = 1, needs_dashboard: bool = False, needs_priority_support: bool = False) -> dict¶
Recommend a tier using the launch-plan pricing strategy.
check_budget_cap(usage_amount: int, cap_type: str, cap_value: int) -> bool¶
Check if usage_amount would exceed the budget cap.
compute_hardware_fingerprint(components: dict) -> str¶
Compute SHA-256 hash from hardware component data.
tier_for_price_id(price_id: str) -> str | None¶
Reverse-map a Stripe price ID to a tier key.
create_checkout_session(tier: str, customer_email: str, success_url: str, cancel_url: str, stripe_customer_id: str = '', trial_period_days: int = 0) -> str¶
Create a Stripe Checkout Session and return the URL.
create_portal_session(stripe_customer_id: str, return_url: str) -> str¶
Create a Stripe Customer Portal session and return the URL.
process_webhook_event(payload: bytes, sig_header: str, webhook_secret: str, idempotency: _IdempotencyStore | None = None) -> dict¶
Verify and parse a Stripe webhook event.