Prefect Stripe Integration: Restricted API Keys, Spend Caps, and Agent Governance

Prefect is a Python-native workflow orchestration platform — flows are Python functions decorated with @flow, tasks are Python functions decorated with @task, and the whole thing runs wherever Python runs. Teams wire Prefect flows into Stripe billing to close the loop between usage data and SaaS charges: a scheduled billing flow queries a usage database, maps a billing task over each customer record, calls stripe.charges.create() per customer, and writes charge results back. Three Prefect-specific behaviors introduce Stripe double-charge failure modes that Prefect’s own observability — flow run state history, task run logs, Prefect Cloud dashboards — does not surface as billing risk.

This post covers those three failure modes with Prefect Python code and the two-layer governance pattern — content-hash idempotency keys plus per-flow-run vault keys via a spend-cap proxy — that eliminates all three without restructuring your flows.

Failure mode 1: @task(retries=N) re-executes the billing callable from line 1 after any downstream failure

Prefect’s @task decorator accepts a retries parameter. When a task raises an exception, Prefect waits retry_delay_seconds and re-invokes the task callable — starting at the function’s first line, with the same arguments that were passed originally. There is no partial continuation. The full function body executes again.

In a billing task, this means: if stripe.charges.create() succeeds on line 12, but the downstream db.execute("INSERT INTO billing_records ...") on line 18 raises a psycopg2.OperationalError (connection pool exhausted, transient network fault, replica lag), Prefect marks the task run as RETRYING and invokes the callable again. The second invocation reaches line 12 again. It calls stripe.charges.create() again with the same customer, the same amount, and — without an explicit idempotency key — a new Stripe-generated request ID. Stripe creates a second charge. With retries=3 and a persistent database failure, the customer receives four charges before Prefect marks the task as FAILED.

# UNSAFE: billing task with retries but no idempotency key
# A transient database failure after the Stripe call triggers a retry
# that re-executes stripe.charges.create() from line 1

import stripe
from prefect import task, flow

stripe.api_key = "sk_live_..."  # UNSAFE: unrestricted, shared across all task instances

@task(retries=3, retry_delay_seconds=10)
def charge_customer(customer_id: str, amount_cents: int,
                    stripe_customer_id: str, billing_period: str):
    # Stripe call succeeds -- ch_A created
    charge = stripe.Charge.create(
        amount=amount_cents,
        currency="usd",
        customer=stripe_customer_id,
        description=f"Billing for {billing_period}",
        # no idempotency_key -- Stripe generates a new request ID each call
    )
    # Database write raises psycopg2.OperationalError -> Prefect retries
    # Next invocation creates ch_B for the same customer + billing_period
    db.execute(
        "INSERT INTO billing_records (customer_id, charge_id, billing_period) "
        "VALUES (%s, %s, %s)",
        (customer_id, charge.id, billing_period)
    )

The Prefect UI shows the task run history as: RETRYING (attempt 1/3), RETRYING (attempt 2/3), COMPLETED. There is no billing-specific signal — Prefect does not know that stripe.charges.create() succeeded before the exception. Stripe’s dashboard shows four ch_ objects for the same customer in the same billing period, each with a different request ID and a different charge creation timestamp. The first three are orphaned — they have no corresponding row in billing_records.

# SAFE: billing task with content-hash idempotency key
# The same idempotency key on every retry causes Stripe to return
# the existing charge rather than creating a new one

import hashlib
import stripe
from prefect import task

VAULT_KEY = "vk_live_..."   # per-flow-run vault key from Keybrake
PROXY_BASE = "https://proxy.keybrake.com/stripe"

@task(retries=3, retry_delay_seconds=10)
def charge_customer(customer_id: str, amount_cents: int,
                    stripe_customer_id: str, billing_period: str,
                    vault_key: str):
    # Content-hash idempotency key: stable across all task retries
    # Does NOT include the Prefect task run ID -- that changes on each retry
    raw_key = f"{customer_id}:{amount_cents}:{billing_period}:prefect-billing"
    idempotency_key = hashlib.sha256(raw_key.encode()).hexdigest()[:32]

    stripe.api_key  = vault_key
    stripe.api_base = PROXY_BASE

    try:
        charge = stripe.Charge.create(
            amount=amount_cents,
            currency="usd",
            customer=stripe_customer_id,
            description=f"Billing for {billing_period}",
            idempotency_key=idempotency_key,  # SAFE: same key on every retry
        )
    except stripe.error.CardError:
        return {"customer_id": customer_id, "charge_id": None, "error": "card_declined"}
    except stripe.error.InvalidRequestError as e:
        return {"customer_id": customer_id, "charge_id": None, "error": str(e)}

    # If the database write fails and Prefect retries, the next invocation
    # calls stripe.Charge.create() with the same idempotency_key ->
    # Stripe returns the original ch_A, no new charge created
    db.execute(
        "INSERT INTO billing_records (customer_id, charge_id, billing_period) "
        "VALUES (%s, %s, %s) ON CONFLICT DO NOTHING",
        (customer_id, charge.id, billing_period)
    )
    return {"customer_id": customer_id, "charge_id": charge.id}

The critical constraint: the idempotency key must be derived from the business inputs — customer ID, amount, billing period — not from Prefect’s task run ID or flow run ID. Both change on every retry. A key derived from the task run ID is a new key to Stripe on each retry and provides no deduplication.

Failure mode 2: .map() with ConcurrentTaskRunner dispatches N billing tasks simultaneously sharing one unrestricted STRIPE_SECRET_KEY

Prefect’s .map() method submits all N task instances to the task runner at once. With ConcurrentTaskRunner — the default for async-compatible flows in Prefect 3 — all N instances execute concurrently in the same Python process using asyncio. With ThreadPoolTaskRunner, they execute concurrently in a thread pool. In both cases, all N task instances share the same module-level stripe.api_key that was set at import time from the environment. There is no per-task spend cap. There is no rate limit per task instance.

The failure scenario: a data pipeline upstream of your billing flow has a bug. The amount_cents column in your usage aggregation table is computed as the monthly total in dollars, not cents — so a customer whose plan costs $149.95/month has amount_cents = 149 instead of 14995. Or the column was already in cents, but someone added a × 100 conversion in the SQL, doubling every charge. Your billing flow maps over 800 customers. ConcurrentTaskRunner dispatches all 800 task instances simultaneously. All 800 call stripe.charges.create() before any result returns. All 800 succeed — Stripe accepts any positive integer as a valid charge amount. By the time you notice the error, tens of thousands in incorrect charges have been submitted.

# UNSAFE: billing flow with .map() and a shared unrestricted STRIPE_SECRET_KEY
# A data error in amount_cents propagates to all N customers simultaneously
# with no mechanism to detect or halt mid-run

import stripe
from prefect import flow, task
from prefect.task_runners import ConcurrentTaskRunner

stripe.api_key = "sk_live_..."  # UNSAFE: shared, unrestricted, no spend cap

@task(retries=2)
def charge_customer(customer: dict) -> dict:
    charge = stripe.Charge.create(
        amount=customer["amount_cents"],  # data error here -> all N customers
        currency="usd",
        customer=customer["stripe_customer_id"],
    )
    return {"customer_id": customer["id"], "charge_id": charge.id}

@flow(task_runner=ConcurrentTaskRunner())
def billing_flow(billing_period: str):
    customers = fetch_billing_queue(billing_period)  # returns list of dicts
    results = charge_customer.map(customers)         # dispatches all N at once
    return [r.result() for r in results]

# SAFE: billing flow with per-flow vault key capped at expected total
# A data error in amount_cents causes the vault key to hit its spend cap
# before all N charges can be submitted -- the proxy rejects excess calls

import hashlib
import httpx
import stripe
from prefect import flow, task
from prefect.task_runners import ConcurrentTaskRunner
from prefect import unmapped

KEYBRAKE_API_KEY = "kb_..."
PROXY_BASE = "https://proxy.keybrake.com/stripe"

def issue_vault_key(billing_period: str, expected_total_cents: int) -> str:
    resp = httpx.post(
        "https://api.keybrake.com/vault-keys",
        headers={"Authorization": f"Bearer {KEYBRAKE_API_KEY}"},
        json={
            "label": f"prefect-billing-{billing_period}",
            "vendor": "stripe",
            "spend_cap_usd": round(expected_total_cents / 100 * 1.10, 2),
            "ttl_seconds": 1800,
            "allowed_endpoints": ["/v1/charges"],
        },
        timeout=10,
    )
    resp.raise_for_status()
    return resp.json()["vault_key"]

@task(retries=3, retry_delay_seconds=10)
def charge_customer(customer: dict, vault_key: str) -> dict:
    raw_key = f"{customer['id']}:{customer['amount_cents']}:{customer['billing_period']}:prefect-billing"
    idempotency_key = hashlib.sha256(raw_key.encode()).hexdigest()[:32]

    stripe.api_key  = vault_key   # per-flow-run vault key
    stripe.api_base = PROXY_BASE  # Keybrake proxy

    try:
        charge = stripe.Charge.create(
            amount=customer["amount_cents"],
            currency="usd",
            customer=customer["stripe_customer_id"],
            description=f"Billing for {customer['billing_period']}",
            idempotency_key=idempotency_key,
        )
    except stripe.error.CardError:
        return {"customer_id": customer["id"], "charge_id": None, "error": "card_declined"}
    except stripe.error.InvalidRequestError as e:
        return {"customer_id": customer["id"], "charge_id": None, "error": str(e)}

    db.execute(
        "INSERT INTO billing_records (customer_id, charge_id, billing_period) "
        "VALUES (%s, %s, %s) ON CONFLICT DO NOTHING",
        (customer["id"], charge.id, customer["billing_period"]),
    )
    return {"customer_id": customer["id"], "charge_id": charge.id}

@flow(task_runner=ConcurrentTaskRunner())
def billing_flow(billing_period: str):
    customers = fetch_billing_queue(billing_period)
    expected_total = sum(c["amount_cents"] for c in customers)

    # Issue vault key once per flow run -- capped at 110% of expected total
    # A data error that would exceed the cap causes the proxy to reject
    # subsequent charges after the cap is hit, not all N go through
    vault_key = issue_vault_key(billing_period, expected_total)

    results = charge_customer.map(customers, unmapped(vault_key))
    return [r.result() for r in results]

The vault key spend cap is a circuit breaker for data errors. If amount_cents is 100× too large, the first few charges consume the cap and the proxy starts returning 402 Payment Required on subsequent calls. The remaining task instances fail with a predictable error rather than submitting incorrect charges. The cap is set at 110% of the expected total to absorb legitimate per-customer variance without triggering false rejections on correct data.

Failure mode 3: Prefect Cloud Automation + manual deployment run create two concurrent flow runs for the same billing period

Prefect Cloud Automations can trigger a deployment run on a cron schedule — the Prefect-native equivalent of a cron job. When a billing flow run fails partway through (say, a database connection pool exhausted in the eighth batch of twelve), the Automation’s error trigger may fire a new run. Simultaneously, an on-call engineer sees the failure notification and triggers a manual run via the Prefect UI or prefect deployment run. Prefect Cloud does not enforce at-most-one-active-run-per-deployment semantics by default. Both runs exist in the Prefect backend. Both start executing. Both call fetch_billing_queue(billing_period) and get the same customer list.

# What the Prefect Cloud flow run list looks like during a double-execution incident:
#
# Flow run: billing-flow/green-fox (flow_run_id=abc123)
#   State:  RUNNING
#   Started: 2026-07-01T02:00:11Z   (Automation scheduled trigger)
#   Progress: 6/12 task runs COMPLETED, 1 RUNNING, 5 PENDING
#
# Flow run: billing-flow/bold-cat (flow_run_id=def456)
#   State:  RUNNING
#   Started: 2026-07-01T02:08:44Z   (manual deployment run by on-call engineer)
#   Progress: 3/12 task runs COMPLETED, 1 RUNNING, 8 PENDING
#
# Both runs are executing charge_customer.map() over the same billing_period.
# Customers in batches 1-3 are being processed by BOTH flow runs.
# Neither run has visibility into the other's task results.

The standard Prefect pattern for deduplicating across flow runs is a concurrency limit. Prefect Cloud supports concurrency tags — you can apply a tag to a deployment and set a concurrency limit of 1. But concurrency limits are enforced at the deployment level, not the billing-period level. If you want at-most-one run for a specific billing period (allowing simultaneous runs for different billing periods), Prefect’s native concurrency limits don’t express that constraint. The fix is an external distributed lock keyed on the billing period.

# SAFE: billing flow with a Redis distributed lock that prevents two
# concurrent flow runs from processing the same billing period

import redis
import hashlib
import stripe
from prefect import flow, task
from prefect.task_runners import ConcurrentTaskRunner

REDIS_URL = "redis://your-redis-host:6379/0"
LOCK_TTL_SECONDS = 7200   # 2 hours: long enough for any realistic billing run

def acquire_billing_lock(billing_period: str) -> bool:
    r = redis.from_url(REDIS_URL)
    lock_key = f"prefect-billing-lock:{billing_period}"
    # NX: set only if key does not exist -- atomically prevents two acquisitions
    # EX: auto-expire after TTL so a crashed flow run releases the lock eventually
    result = r.set(lock_key, "1", nx=True, ex=LOCK_TTL_SECONDS)
    return result is not None  # True if lock was acquired, False if already held

def release_billing_lock(billing_period: str) -> None:
    r = redis.from_url(REDIS_URL)
    r.delete(f"prefect-billing-lock:{billing_period}")

@flow(task_runner=ConcurrentTaskRunner())
def billing_flow(billing_period: str):
    if not acquire_billing_lock(billing_period):
        # Another flow run is active for this billing period --
        # exit immediately without touching Stripe
        print(f"Billing lock held for {billing_period} -- skipping this run")
        return {"skipped": True, "reason": "concurrent_run_active"}

    try:
        customers = fetch_billing_queue(billing_period)
        expected_total = sum(c["amount_cents"] for c in customers)
        vault_key = issue_vault_key(billing_period, expected_total)

        results = charge_customer.map(customers, unmapped(vault_key))
        return [r.result() for r in results]
    finally:
        # Always release the lock -- even if the flow raises an exception
        # The finally block runs before Prefect transitions the flow run to FAILED
        release_billing_lock(billing_period)

The Redis SET NX EX is atomic — there is no window between checking and setting. The first flow run acquires the lock; any subsequent flow run for the same billing period returns immediately without calling Stripe. The finally block releases the lock even when the flow fails, so a fresh run can be triggered after the root cause is fixed. The TTL guards against a flow run that crashes without reaching the finally block — after 2 hours the lock auto-expires and a new run can proceed.

The two-layer governance pattern for Prefect billing flows

All three failure modes are addressed by two complementary layers:

The layers are independent. The idempotency key closes the retry window. The vault key closes the data-error window. The Redis lock closes the concurrent-execution window. You need all three because each addresses a distinct failure path — you can have a correct idempotency key and still suffer a data error that the vault key catches, or have both and still run two concurrent flow runs that the Redis lock prevents.

Gap analysis: four Prefect billing edge cases that remain after the two-layer pattern

Vault key TTL sizing for large concurrent .map() runs. The vault key TTL must cover the full expected duration of the billing flow run — not just the first batch. With 2,000 customers at 150ms per Stripe API call and ConcurrentTaskRunner, the flow may run for several minutes depending on concurrency limits. Issue the vault key with a TTL of at least 3× the expected maximum run duration. Catch stripe.error.AuthenticationError in the billing task — a vault key expiry during a run surfaces as an authentication error, not a spend-cap error. Log the expiry and return an error dict rather than raising, so the task run completes and Prefect doesn’t trigger a retry with the expired key.

Prefect task run ID as idempotency key anti-pattern. The Prefect task run ID (accessible via prefect.context.get_run_context().task_run.id) changes on every retry attempt, making it useless as an idempotency key. Teams sometimes use it because it looks like a stable unique identifier per task invocation, but it is per-attempt, not per-business-event. Always derive the idempotency key from business inputs: customer ID, amount, billing period. A key containing any Prefect runtime ID is a per-attempt key that provides no deduplication.

Prefect Automations on-failure trigger firing a new run while the original is still executing. Prefect Cloud Automations support an on-failure trigger — useful for alerting, but dangerous if configured to trigger a new flow run. If the original flow run failed at batch 8 of 12, a new Automation-triggered run fires immediately. Batches 1–7 were already billed. The Redis lock prevents the new run from executing — but only if the original run properly released the lock in its finally block. If the original run was killed mid-execution (SIGKILL, infrastructure failure) before reaching finally, the lock remains held for the TTL duration. Size the TTL conservatively: long enough to survive any realistic run, short enough that a killed run releases the lock before the next scheduled billing window.

Pre-flight check bypass when usage data is re-queried each run. Some billing designs re-query usage data each time the flow runs rather than reading from a persistent queue table. In these designs, a customer who was billed in the previous run appears again in the usage query for the next run if the usage aggregation window overlaps. The idempotency key prevents Stripe from creating a second charge within the 24-hour window, but past that window, the same key will cause Stripe to charge again. Add an explicit pre-flight check against a durable billing_records table before calling Stripe, and skip customers with an existing record regardless of the idempotency key state.

Pytest enforcement suite

# tests/test_prefect_billing.py
import hashlib
import pytest
from unittest.mock import MagicMock, patch

# 1. Idempotency key is stable across task retries (same inputs -> same key)
def test_idempotency_key_stability():
    customer_id = "cus_abc123"
    amount_cents = 4995
    billing_period = "2026-07"

    raw = f"{customer_id}:{amount_cents}:{billing_period}:prefect-billing"
    key_attempt_1 = hashlib.sha256(raw.encode()).hexdigest()[:32]
    key_attempt_2 = hashlib.sha256(raw.encode()).hexdigest()[:32]
    key_attempt_3 = hashlib.sha256(raw.encode()).hexdigest()[:32]

    assert key_attempt_1 == key_attempt_2 == key_attempt_3

# 2. Idempotency key differs from adjacent billing periods (no cross-period collision)
def test_idempotency_key_period_isolation():
    customer_id = "cus_abc123"
    amount_cents = 4995

    def make_key(period):
        raw = f"{customer_id}:{amount_cents}:{period}:prefect-billing"
        return hashlib.sha256(raw.encode()).hexdigest()[:32]

    assert make_key("2026-07") != make_key("2026-06")
    assert make_key("2026-07") != make_key("2026-08")

# 3. CardError is caught and returned as error dict (no raise -> no retry storm)
def test_card_error_returns_dict_not_raises():
    import stripe
    with patch("stripe.Charge.create") as mock_charge:
        mock_charge.side_effect = stripe.error.CardError(
            "card_declined", "card_number", "card_declined"
        )
        # Simulate calling the task function directly (bypassing Prefect runtime)
        from billing.tasks import charge_customer
        result = charge_customer.fn(
            customer={"id": "cus_x", "amount_cents": 1000,
                      "stripe_customer_id": "cus_stripe_x", "billing_period": "2026-07"},
            vault_key="vk_live_test"
        )
    assert result["charge_id"] is None
    assert "card_declined" in result.get("error", "")

# 4. Redis lock prevents second concurrent flow run from acquiring the same lock
def test_redis_lock_blocks_concurrent_run():
    from billing.flow import acquire_billing_lock

    with patch("redis.from_url") as mock_redis_factory:
        mock_r = MagicMock()
        mock_redis_factory.return_value = mock_r

        # First acquisition succeeds (SET NX returns True)
        mock_r.set.return_value = True
        assert acquire_billing_lock("2026-07") is True

        # Second acquisition fails (SET NX returns None when key exists)
        mock_r.set.return_value = None
        assert acquire_billing_lock("2026-07") is False

# 5. Idempotency key does NOT include Prefect task run ID (it changes on retry)
def test_idempotency_key_excludes_runtime_ids():
    import uuid

    customer_id = "cus_abc123"
    amount_cents = 4995
    billing_period = "2026-07"
    raw = f"{customer_id}:{amount_cents}:{billing_period}:prefect-billing"
    stable_key = hashlib.sha256(raw.encode()).hexdigest()[:32]

    # A key derived from a random task run ID would differ on every retry
    task_run_id_1 = str(uuid.uuid4())
    task_run_id_2 = str(uuid.uuid4())
    key_with_id_1 = hashlib.sha256(f"{raw}:{task_run_id_1}".encode()).hexdigest()[:32]
    key_with_id_2 = hashlib.sha256(f"{raw}:{task_run_id_2}".encode()).hexdigest()[:32]

    # Stable key is same both times; ID-based keys differ
    assert stable_key == hashlib.sha256(raw.encode()).hexdigest()[:32]
    assert key_with_id_1 != key_with_id_2

Prefect vs other orchestration platforms: billing-specific comparison

FAQ

Can I use Prefect’s built-in concurrency limit instead of a Redis lock?

Prefect Cloud concurrency limits enforce at-most-N-concurrent-runs per tag across the full deployment — useful for preventing resource exhaustion but not for billing-period isolation. A concurrency limit of 1 on your billing deployment prevents two simultaneous billing flow runs for any period, which is overly restrictive if you need to run billing for July while August’s pro-ration run executes. The Redis lock keyed on prefect-billing-lock:{billing_period} allows concurrent runs for different periods while preventing duplicate runs for the same period. Use the Redis lock for billing isolation, and leave Prefect’s concurrency limit for infrastructure-level constraints like limiting concurrent database connections.

Does idempotency_key protect against charges submitted more than 24 hours apart?

No. Stripe’s idempotency key deduplication window is 24 hours. After that window, the same key submitted again will produce a new charge. For retry scenarios (task retries happen within minutes), the 24-hour window is more than sufficient. For re-run scenarios — running next month’s billing against last month’s data by mistake, or re-running a billing flow days after the original run — the idempotency key does not protect you. The pre-flight check against a durable billing_records table is the correct guard for period-level deduplication that persists beyond 24 hours.

Should I use stripe.PaymentIntent instead of stripe.Charge?

Yes, for new integrations — stripe.Charge is a legacy API; Stripe recommends PaymentIntents for new work. The idempotency key pattern is identical: pass idempotency_key to stripe.PaymentIntent.create(). The same content-hash approach applies: sha256(customer_id:amount_cents:billing_period:prefect-billing)[:32]. The vault key governance and Redis lock patterns are API-agnostic and apply equally to PaymentIntents, Charges, and Subscriptions.

How do I size the vault key TTL for a large concurrent .map() run?

Measure your median per-Stripe-call latency under load (typically 100–250ms per call), estimate the flow’s total wall-clock duration accounting for ConcurrentTaskRunner concurrency, and multiply by 3 for safety margin. For 2,000 customers at 150ms per call with concurrency of 50, the estimated run time is roughly 6 seconds of Stripe calls plus database and scheduling overhead. A 300-second (5-minute) TTL is a safe floor for flows under 500 customers; 1,800 seconds for flows over 2,000. Catch stripe.error.AuthenticationError in the billing task and treat it as a retriable soft error rather than raising — this prevents Prefect from triggering a task retry that uses the same expired vault key.

What is the Prefect unmapped() utility for in the .map() pattern?

unmapped() tells Prefect’s .map() to pass a single value to all mapped task instances rather than iterating over it. Without unmapped(vault_key), Prefect would try to iterate over the vault key string character by character and pass one character per task instance — which is not what you want. Always wrap shared arguments (vault key, billing period, configuration) in unmapped() when calling .map(). Only the argument you want distributed element-by-element (the customer list) should be passed without unmapped().