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

Metaflow was built at Netflix to bring software engineering discipline to data science pipelines. Its core value is reproducibility: every step is versioned, every artifact is tracked, and every failed flow can be resumed from the exact point of failure. Those same properties make Stripe billing in Metaflow uniquely dangerous, because the retry, parallelism, and resume mechanics that make ML experiments robust can silently replay billing operations against customers who have already been charged.

This post covers three Metaflow-specific Stripe billing failure modes, the Python code that exposes each one, and the two-layer governance pattern — content-hash idempotency keys and per-branch vault keys via a spend-cap proxy — that eliminates them without restructuring your flow.

Failure mode 1: @retry decorator re-executes billing from line 1 on any downstream exception

Metaflow's @retry decorator (from metaflow or via @retry(times=N)) instructs Metaflow to re-run the entire step function from the beginning when the step raises any unhandled exception. There is no mid-step checkpoint: the step function is an atomic unit from Metaflow's perspective. If stripe.charges.create() succeeds on line 9 and write_charge_to_database() raises on line 15, the retry starts at line 1 and calls stripe.charges.create() again with identical parameters but no idempotency key.

# billing_flow.py — UNSAFE: @retry re-fires stripe.charges.create() on database error
import stripe
import os
from metaflow import FlowSpec, step, retry, Parameter

stripe.api_key = os.environ["STRIPE_SECRET_KEY"]  # unrestricted live key

class BillingFlow(FlowSpec):

    billing_period = Parameter("billing_period", help="Billing period e.g. 2026-07")

    @step
    def start(self):
        self.customers = load_customers_to_bill()
        self.next(self.charge_customer, foreach="customers")

    @retry(times=3)
    @step
    def charge_customer(self):
        customer_id = self.input
        # @retry re-runs from here — no idempotency key, no guard
        charge = stripe.charges.create(
            amount=2999,
            currency="usd",
            customer=customer_id,
            description=f"Subscription {self.billing_period}",
        )
        # Database write fails intermittently — raises on 15% of executions
        write_charge_to_database(customer_id, charge.id, self.billing_period)
        self.charge_id = charge.id
        self.next(self.join)

    @step
    def join(self, inputs):
        self.charge_ids = [i.charge_id for i in inputs]
        self.next(self.end)

    @step
    def end(self):
        pass

if __name__ == "__main__":
    BillingFlow()

On the first retry, Stripe has no record of the previous request — no idempotency key was sent, so Stripe creates a second charge (ch_B) for the same customer at the same amount. With times=3 and a 15% database failure rate, each affected customer accumulates 1.45 charges on average per billing run. Metaflow's task dashboard shows each retry as expected step behavior; the duplicate charges are invisible until a customer raises a dispute or a month-end reconciliation surfaces the discrepancy.

The fix: derive a content-hash idempotency key from the inputs that are stable across all retries — customer_id, the charge amount, and billing_period — and pass it with every Stripe call. Metaflow guarantees that all retries of the same step receive the same self.input value and the same Parameter values, so the key is identical on every attempt. Stripe's idempotency layer returns the original ch_A on all subsequent calls without creating a new charge object.

# billing_flow.py — SAFE: content-hash idempotency key survives all @retry retries
import stripe
import hashlib
import os
from metaflow import FlowSpec, step, retry, Parameter

def billing_idempotency_key(customer_id: str, amount_cents: int, billing_period: str) -> str:
    raw = f"{customer_id}:{amount_cents}:{billing_period}:metaflow-billing"
    return hashlib.sha256(raw.encode()).hexdigest()[:32]

class BillingFlow(FlowSpec):

    billing_period = Parameter("billing_period", help="Billing period e.g. 2026-07")

    @step
    def start(self):
        self.customers = load_customers_to_bill()
        self.next(self.charge_customer, foreach="customers")

    @retry(times=3)
    @step
    def charge_customer(self):
        customer_id = self.input
        amount_cents = 2999
        idempotency_key = billing_idempotency_key(customer_id, amount_cents, self.billing_period)

        stripe_client = stripe.StripeClient(
            os.environ["KEYBRAKE_BILLING_VAULT_KEY"],
            base_url="https://proxy.keybrake.com/stripe/",
        )
        # Same key on every retry — Stripe returns ch_A without creating ch_B, ch_C, ch_D
        charge = stripe_client.charges.create(
            params={
                "amount": amount_cents,
                "currency": "usd",
                "customer": customer_id,
                "description": f"Subscription {self.billing_period}",
            },
            options={"idempotency_key": idempotency_key},
        )
        write_charge_to_database(customer_id, charge.id, self.billing_period)
        self.charge_id = charge.id
        self.next(self.join)

    @step
    def join(self, inputs):
        self.charge_ids = [i.charge_id for i in inputs]
        self.next(self.end)

    @step
    def end(self):
        pass

if __name__ == "__main__":
    BillingFlow()

Additionally, scope the vault key to POST /v1/charges only and set a per-key daily spend cap equal to the per-customer charge amount plus a small buffer. A unit calculation bug that passes dollars instead of cents is blocked at the proxy after the first attempt rather than silently multiplying across every retry and every concurrent branch.

Failure mode 2: foreach parallel branches share one unrestricted Stripe key with no per-branch spend cap

Metaflow's foreach fan-out — self.next(self.charge_step, foreach="customers") — creates one parallel branch per item in the input list, with each branch running as an independent task. All branch tasks run concurrently (on the local machine, AWS Batch, or Kubernetes, depending on the decorator) and inherit the same environment variables, including STRIPE_SECRET_KEY. There is no mechanism that stops a running branch when a sibling branch encounters a billing error.

# segmented_billing_flow.py — UNSAFE: foreach branches share one Stripe key, no per-branch cap
import stripe
import os
from metaflow import FlowSpec, step, batch, retry

stripe.api_key = os.environ["STRIPE_SECRET_KEY"]  # same key in all parallel branch containers

class SegmentedBillingFlow(FlowSpec):

    @step
    def start(self):
        # Each item is (customer_id, plan_amount_cents)
        self.billing_tasks = [
            ("cus_aaa", 999),
            ("cus_bbb", 2999),
            ("cus_ccc", 9999),
            # ... hundreds more
        ]
        self.next(self.charge_branch, foreach="billing_tasks")

    @retry(times=2)
    @batch(cpu=1, memory=512)
    @step
    def charge_branch(self):
        customer_id, amount_cents = self.input
        # No idempotency key, no spend cap — all branches run concurrently
        # A unit bug (amount_dollars not amount_cents) hits every branch simultaneously
        charge = stripe.charges.create(
            amount=amount_cents,
            currency="usd",
            customer=customer_id,
            description="Monthly subscription",
        )
        self.charge_id = charge.id
        self.next(self.join)

    @step
    def join(self, inputs):
        self.results = [(i.input, i.charge_id) for i in inputs]
        self.next(self.end)

    @step
    def end(self):
        pass

if __name__ == "__main__":
    SegmentedBillingFlow()

The blast-radius problem: when all branches execute simultaneously and one contains a billing bug — wrong amount, wrong customer, wrong billing period — all branches are already running by the time the first error surfaces. Sequential execution would expose the bug after the first customer; foreach parallel execution creates errors across every branch simultaneously before any branch result is joined. A calculation error passing amounts in dollars instead of cents charges all concurrent customers 100× with no circuit breaker between branches and no stop signal when the first bad charge is created.

The fix: issue per-branch vault keys in a setup step that runs before the foreach fan-out, and pass each vault key to its corresponding branch via Metaflow artifacts. Each vault key is scoped to POST /v1/charges and capped at the expected charge amount for that specific customer plus a small buffer. A wrong-amount request is rejected at the proxy immediately; because each branch has its own capped vault key, a bug in one branch does not exhaust the spend budget of any other branch.

# segmented_billing_flow.py — SAFE: per-branch vault keys issued before foreach fan-out
import stripe
import hashlib
import httpx
import os
from metaflow import FlowSpec, step, batch, retry

def issue_vault_key(label: str, max_amount_cents: int) -> str:
    resp = httpx.post(
        "https://proxy.keybrake.com/admin/vault_keys",
        headers={"Authorization": f"Bearer {os.environ['KEYBRAKE_ADMIN_KEY']}"},
        json={
            "label": label,
            "vendor": "stripe",
            "allowed_endpoints": ["POST /v1/charges"],
            "daily_usd_cap": round(max_amount_cents / 100 * 1.1, 2),
            "expires_in_seconds": 7200,
        },
        timeout=5.0,
    )
    resp.raise_for_status()
    return resp.json()["vault_key"]

def billing_idempotency_key(customer_id: str, amount_cents: int, billing_period: str) -> str:
    raw = f"{customer_id}:{amount_cents}:{billing_period}:metaflow-billing"
    return hashlib.sha256(raw.encode()).hexdigest()[:32]

class SegmentedBillingFlow(FlowSpec):

    billing_period = Parameter("billing_period", help="Billing period e.g. 2026-07")

    @step
    def start(self):
        raw_tasks = [("cus_aaa", 999), ("cus_bbb", 2999), ("cus_ccc", 9999)]
        # Issue one vault key per branch before fan-out; pass key alongside task input
        self.billing_tasks = [
            {
                "customer_id": cid,
                "amount_cents": amt,
                "vault_key": issue_vault_key(
                    f"metaflow-{self.billing_period}-{cid}", max_amount_cents=amt
                ),
            }
            for cid, amt in raw_tasks
        ]
        self.next(self.charge_branch, foreach="billing_tasks")

    @retry(times=2)
    @batch(cpu=1, memory=512)
    @step
    def charge_branch(self):
        task = self.input
        customer_id = task["customer_id"]
        amount_cents = task["amount_cents"]
        vault_key = task["vault_key"]

        idempotency_key = billing_idempotency_key(customer_id, amount_cents, self.billing_period)
        stripe_client = stripe.StripeClient(
            vault_key,
            base_url="https://proxy.keybrake.com/stripe/",
        )
        # Vault key is capped at this customer's expected charge — a unit bug is rejected immediately
        charge = stripe_client.charges.create(
            params={
                "amount": amount_cents,
                "currency": "usd",
                "customer": customer_id,
                "description": f"Monthly subscription {self.billing_period}",
            },
            options={"idempotency_key": idempotency_key},
        )
        write_charge_to_database(customer_id, charge.id, self.billing_period)
        self.charge_id = charge.id
        self.next(self.join)

    @step
    def join(self, inputs):
        self.results = [(i.input["customer_id"], i.charge_id) for i in inputs]
        self.next(self.end)

    @step
    def end(self):
        pass

if __name__ == "__main__":
    SegmentedBillingFlow()

Each vault key expires after two hours, so it cannot be reused if a subsequent flow run loads the artifact from a prior run's datastore entry. The per-branch spend caps ensure that a calculation error is contained to one branch: the proxy rejects the first wrong-amount charge request and the branch fails with a 402 rather than silently overcharging the customer.

Failure mode 3: resume re-runs a billing step that already charged customers

Metaflow's resume command is one of its most-used developer features: when a flow fails, you fix the bug and run python flow.py resume --origin-run-id <run-id> to replay the flow from the failed step, reusing successful steps' artifacts rather than re-computing them. This is the correct behavior for ML steps — you want to reuse the feature-engineering output rather than re-run it. For billing steps, it creates a critical trap.

The scenario: a billing step calls stripe.charges.create() successfully, then write_charge_to_database() raises a psycopg2.OperationalError (transient connection timeout). Metaflow marks the step as failed. The developer debugs the DB connection, sees the step error, runs resume. Metaflow re-executes the billing step from line 1 — stripe.charges.create() fires again. Without an idempotency key, Stripe creates ch_B for a customer who already has ch_A. The developer never sees a second charge in Metaflow's UI; from Metaflow's perspective, the step succeeded on the second attempt and the resume looks clean.

# billing_flow.py — UNSAFE: resume re-fires stripe.charges.create() after DB failure
from metaflow import FlowSpec, step, retry

class BillingFlow(FlowSpec):

    @step
    def start(self):
        self.customers = load_customers_to_bill()
        self.next(self.charge_customer, foreach="customers")

    @step  # no @retry — but developer will use `resume` when this fails
    def charge_customer(self):
        customer_id = self.input
        # Step succeeded here on first run — ch_A created
        charge = stripe.charges.create(
            amount=2999,
            currency="usd",
            customer=customer_id,
        )
        # DB write failed on first run — step marked failed
        # Developer runs: python billing_flow.py resume --origin-run-id 1234
        # Metaflow re-runs from here → stripe.charges.create() fires → ch_B
        write_charge_to_database(customer_id, charge.id)
        self.charge_id = charge.id
        self.next(self.join)

    @step
    def join(self, inputs):
        self.charge_ids = [i.charge_id for i in inputs]
        self.next(self.end)

    @step
    def end(self):
        pass

if __name__ == "__main__":
    BillingFlow()

The fix adds two layers. First, a content-hash idempotency key derived from stable business inputs closes the Stripe-level duplicate: even though resume calls stripe.charges.create() again, Stripe returns the original ch_A because the idempotency key matches. Second, a pre-flight audit check queries the Keybrake proxy's audit log before touching Stripe at all — this handles the case where the Stripe idempotency key's 24-hour window has expired but the customer was charged in a prior run months earlier. The pre-flight check operates on business logic (same customer + same amount + same period), not on Stripe's time-bounded idempotency layer.

# billing_flow.py — SAFE: idempotency key + pre-flight audit check survive resume
import stripe
import hashlib
import httpx
import os
from metaflow import FlowSpec, step, Parameter

def billing_idempotency_key(customer_id: str, amount_cents: int, billing_period: str) -> str:
    raw = f"{customer_id}:{amount_cents}:{billing_period}:metaflow-billing"
    return hashlib.sha256(raw.encode()).hexdigest()[:32]

def check_existing_charge(idempotency_key: str) -> str | None:
    resp = httpx.get(
        "https://proxy.keybrake.com/audit",
        headers={"Authorization": f"Bearer {os.environ['KEYBRAKE_AUDIT_KEY']}"},
        params={"idempotency_key": idempotency_key},
        timeout=5.0,
    )
    if resp.status_code == 200:
        entries = resp.json().get("entries", [])
        if entries:
            return entries[0].get("stripe_charge_id")
    return None

class BillingFlow(FlowSpec):

    billing_period = Parameter("billing_period", help="Billing period e.g. 2026-07")

    @step
    def start(self):
        self.customers = load_customers_to_bill()
        self.next(self.charge_customer, foreach="customers")

    @step
    def charge_customer(self):
        customer_id = self.input
        amount_cents = 2999
        idempotency_key = billing_idempotency_key(customer_id, amount_cents, self.billing_period)

        # Pre-flight: short-circuit if this customer was already charged
        # Catches resume, @retry, and any re-execution regardless of Stripe's 24h window
        existing = check_existing_charge(idempotency_key)
        if existing:
            self.charge_id = existing
            self.next(self.join)
            return

        stripe_client = stripe.StripeClient(
            os.environ["KEYBRAKE_BILLING_VAULT_KEY"],
            base_url="https://proxy.keybrake.com/stripe/",
        )
        charge = stripe_client.charges.create(
            params={
                "amount": amount_cents,
                "currency": "usd",
                "customer": customer_id,
                "description": f"Subscription {self.billing_period}",
            },
            options={"idempotency_key": idempotency_key},
        )
        write_charge_to_database(customer_id, charge.id, self.billing_period)
        self.charge_id = charge.id
        self.next(self.join)

    @step
    def join(self, inputs):
        self.charge_ids = [i.charge_id for i in inputs]
        self.next(self.end)

    @step
    def end(self):
        pass

if __name__ == "__main__":
    BillingFlow()

The pre-flight check adds one HTTPS round-trip before every billing call — typically 5–20 ms. For a billing flow processing hundreds of customers, that overhead is negligible compared to the Stripe API call itself. If the proxy is unreachable, the pre-flight check times out and the billing step fails fast with a ConnectError before creating any Stripe charge — a safe failure mode that the developer can address before re-running or resuming the flow.

Approach comparison

Gap analysis: four Metaflow edge cases not covered above

@batch container preemption creates a second retry layer outside @retry. When Metaflow steps run on AWS Batch with spot instances (@batch(cpu=4)), a container preemption mid-execution causes AWS Batch to restart the container independently of Metaflow's @retry decorator. If stripe.charges.create() succeeded in the preempted container and the container was terminated before the step wrote its output artifact, AWS Batch restarts the container and Metaflow re-runs the step from line 1 — without necessarily incrementing the @retry attempt counter. The result is a double-retry layer: Metaflow's @retry plus AWS Batch's spot restart. Content-hash idempotency keys close both layers, but teams should be aware that Metaflow task attempt counts may not reflect the true number of times the step function body executed when spot preemption is involved.

Partial foreach branch failures leave charged customers in an ambiguous state on resume. When a foreach flow fails because some branches succeed and others fail, Metaflow's resume re-runs only the failed branches (the successful branches' artifacts are cloned from the original run). This is correct behavior for ML steps. For billing steps, the ambiguity is: which branches charged customers and which did not? Without a pre-flight audit check, the resumed branches that previously created a charge and then failed (due to a downstream DB write error) will attempt to charge again. The idempotency key + pre-flight pattern handles this correctly regardless of which combination of branches is re-run on resume — each branch either returns the existing charge or creates a new one, never both.

@schedule cron overlaps when a billing run exceeds its interval. Metaflow supports scheduled flows via the @schedule(cron="0 1 1 * *") decorator (on Argo Workflows or AWS Step Functions). If a monthly billing run takes longer than expected — large customer cohort, slow DB writes, AWS Batch queue saturation — and the scheduler fires again before the previous run completes, two billing runs execute concurrently with the same billing_period parameter. Both runs' foreach branches reach stripe.charges.create() with identical inputs. The idempotency key pattern closes this: the second run's charge requests return the first run's charge objects because the keys are content-derived from the same customer ID, amount, and billing period — not from any run-specific metadata.

Vault keys passed as Metaflow artifacts are readable across runs. Metaflow stores step outputs as serialized artifacts in its datastore (S3 for cloud runs, local filesystem for local runs). If a vault key is passed as a step output artifact — for example, returned from a setup step and stored in self.vault_key — it persists in the Metaflow datastore and is readable by any step in the current run and any resumed run that clones that artifact. Keep vault keys in memory only: issue them inside the billing step itself (or pass them via the billing_tasks list as shown in failure mode 2, where the list is not the vault key itself persisted as a separate artifact). Set short TTLs on vault keys (1–2 hours) so any leaked artifact value expires before it can be reused in a subsequent run.

Pytest enforcement suite

# test_metaflow_billing.py
import hashlib
import pytest

def billing_idempotency_key(customer_id, amount_cents, billing_period):
    raw = f"{customer_id}:{amount_cents}:{billing_period}:metaflow-billing"
    return hashlib.sha256(raw.encode()).hexdigest()[:32]

def test_idempotency_key_stability_across_retries():
    """Same inputs always produce same key — @retry and resume are safe."""
    k1 = billing_idempotency_key("cus_abc", 2999, "2026-07")
    k2 = billing_idempotency_key("cus_abc", 2999, "2026-07")
    assert k1 == k2

def test_idempotency_key_distinct_per_billing_period():
    """Different billing periods produce different keys — monthly runs are isolated."""
    k_jul = billing_idempotency_key("cus_abc", 2999, "2026-07")
    k_aug = billing_idempotency_key("cus_abc", 2999, "2026-08")
    assert k_jul != k_aug

def test_idempotency_key_distinct_per_customer():
    """Different customers produce different keys — foreach branches are isolated."""
    k_a = billing_idempotency_key("cus_aaa", 2999, "2026-07")
    k_b = billing_idempotency_key("cus_bbb", 2999, "2026-07")
    assert k_a != k_b

def test_vault_key_scope_blocks_refund_endpoint(monkeypatch):
    """Vault key scoped to POST /v1/charges cannot call POST /v1/refunds."""
    import httpx

    def mock_post(url, **kwargs):
        if "/v1/refunds" in url:
            raise httpx.HTTPStatusError("403", request=None, response=None)
        return type("R", (), {"status_code": 200, "json": lambda self: {"id": "ch_test"}, "raise_for_status": lambda self: None})()

    monkeypatch.setattr(httpx, "post", mock_post)
    with pytest.raises(httpx.HTTPStatusError):
        httpx.post("https://proxy.keybrake.com/stripe/v1/refunds")

def test_preflight_short_circuits_on_existing_charge(monkeypatch):
    """Pre-flight audit check returns existing charge ID — resume does not re-fire."""
    import httpx

    def mock_get(url, **kwargs):
        return type("R", (), {
            "status_code": 200,
            "json": lambda self: {"entries": [{"stripe_charge_id": "ch_existing_456"}]},
        })()

    monkeypatch.setattr(httpx, "get", mock_get)

    import httpx as h
    resp = h.get("https://proxy.keybrake.com/audit", params={"idempotency_key": "abc"})
    entries = resp.json()["entries"]
    assert entries[0]["stripe_charge_id"] == "ch_existing_456"

Frequently asked questions

Can I use Metaflow's run ID or task attempt ID as the idempotency key?

No. Metaflow assigns a new run ID to every flow execution, and a new attempt ID to every retry of a step. Using either as the idempotency key means each attempt sends a different key, and Stripe creates a new charge on every retry — exactly the failure mode you are trying to prevent. The idempotency key must be derived from stable business inputs (customer_id, amount_cents, billing_period) that are identical on every attempt and on every resume of the same logical billing operation.

How does resume interact with idempotency keys in practice?

When you run python flow.py resume --origin-run-id 1234, Metaflow re-executes the billing step from line 1. Your billing step computes the same content-hash idempotency key from the same customer ID and billing period, then sends it to Stripe. Stripe looks up its idempotency cache and returns the original charge object (ch_A) without creating ch_B. If the original charge happened more than 24 hours ago (Stripe's idempotency window), the pre-flight audit check at the Keybrake proxy catches it instead — your billing step short-circuits with the existing charge ID and never contacts Stripe.

How should vault keys be distributed to foreach branches?

Issue one vault key per branch in the setup step (start or a dedicated prepare_keys step) and include it in the input list passed to foreach. Each branch reads its vault key from self.input["vault_key"]. Avoid issuing a single shared vault key and storing it in a flow-level artifact (self.vault_key) that all branches read — shared vault keys defeat the per-branch spend cap isolation. The goal is one vault key per customer or per branch, each capped at that branch's expected charge amount, so a bug in one branch is contained and does not exhaust the cap for other branches running in parallel.

What happens if the Keybrake proxy is unreachable during a Metaflow step?

The billing step raises an httpx.ConnectError before creating any Stripe charge. Metaflow's @retry decorator will retry the step, and the pre-flight check will time out again on each retry until the proxy becomes available. Because no charge was created, the retries are safe — once the proxy is reachable, the first successful attempt creates the charge with the stable idempotency key, and subsequent retries return the same charge. For flows without @retry, the step fails cleanly and the developer can resume after the proxy recovers.

Does this pattern work with @batch and @kubernetes?

Yes. The content-hash idempotency key is computed from business inputs, not from Metaflow or cloud runtime state — it produces the same string whether the step runs locally, in an AWS Batch container, or in a Kubernetes pod. The vault key issuance and Stripe proxy calls are standard HTTPS requests that work identically across all Metaflow execution environments. The only configuration requirement is ensuring the KEYBRAKE_ADMIN_KEY and KEYBRAKE_BILLING_VAULT_KEY secrets are available in the cloud execution environment — via AWS Secrets Manager with the @secrets decorator, or as Kubernetes secrets mounted as environment variables.

Should I use @retry on billing steps at all?

Yes, with idempotency keys in place. @retry is valuable for handling transient failures — DB connection timeouts, network blips, Stripe API 500s — and with a content-hash idempotency key, retries are safe by construction. Without idempotency keys, @retry on billing steps is dangerous and should be removed. With them, it is a useful resilience mechanism. Set minutes_between_retries to at least 30 seconds to give transient errors time to resolve rather than hammering the Stripe API immediately. Avoid @catch on billing steps — catching exceptions silently suppresses step failures and makes it impossible to detect when the billing step partially succeeded before raising.

Related reading

• Flyte Stripe Integration: Restricted API Keys, Spend Caps, and Agent Governance
• ZenML Stripe Integration: Restricted API Keys, Spend Caps, and Agent Governance
• Dagster Stripe Integration: Restricted API Keys, Spend Caps, and Agent Governance
• Airflow Stripe Integration: Restricted API Keys, Spend Caps, and Agent Governance
• Ray Stripe Integration: Restricted API Keys, Spend Caps, and Agent Governance
• Stripe Idempotency Keys for AI Agents: Three Failure Modes and the Fix