Dagster AI agent API key: scoping vendor calls in data pipeline assets

How Dagster AI agent assets call vendor APIs

In a Dagster pipeline, vendor API calls live inside @asset-decorated functions that receive resources via dependency injection. An AI billing pipeline that processes subscription renewals might use dynamic partitions to materialize one asset per customer:

from dagster import asset, DynamicPartitionsDefinition, RetryPolicy
import stripe
import os

customer_partitions = DynamicPartitionsDefinition(name="customers")

@asset(
    partitions_def=customer_partitions,
    retry_policy=RetryPolicy(max_retries=3, delay=1.0),
)
def charged_customer(context) -> dict:
    customer_id = context.partition_key
    stripe.api_key = os.environ["STRIPE_SECRET_KEY"]  # full-access key
    result = stripe.PaymentIntent.create(
        amount=2999,
        currency="usd",
        customer=customer_id,
    )
    return {"status": result.status, "id": result.id}

This is standard Dagster. Materializing the charged_customer asset for all customer partitions executes one materialization per customer. Dagster's executor dispatches them concurrently up to your configured parallelism. The problem: STRIPE_SECRET_KEY is a full-access key from the environment, there's no per-run cap on total charges, and RetryPolicy(max_retries=3) means every materialization failure can produce up to 4 vendor API calls for the same customer.

Three gaps Dagster's native tooling doesn't fill for vendor spend control

The dynamic partition risk: concurrent materializations and vendor call bursts

Dagster's dynamic partitions let you define a partition key set at runtime — useful for processing a dynamic list of customers, orders, or events. When you trigger a backfill or run on a large partition set, Dagster's multi-process executor dispatches up to max_concurrent materializations simultaneously. On a Dagster deployment with max_concurrent=50, 50 concurrent Stripe calls execute in the first wave — each one independent, each one counting against the same Stripe key's rate limits and your daily charge volume.

A vault key with a dollar cap turns this into a bounded operation. The cap is enforced atomically at the proxy layer across all concurrent materializations. Once the cap is hit, further calls return 429. Dagster materializations that receive a 429 raise exceptions, which Dagster records as materialization failures with the full exception traceback — visible in the asset catalog without any additional instrumentation.

Using Dagster resources to inject vault keys

Dagster resources are initialized once per run and injected into assets that declare them as dependencies. This is the idiomatic place to issue a vault key — once at run start, shared across all materializations in the run:

import httpx
from dagster import asset, resource, DynamicPartitionsDefinition, RetryPolicy, RunConfig
import stripe
import os
from dataclasses import dataclass

customer_partitions = DynamicPartitionsDefinition(name="customers")

@dataclass
class KeybrakeResource:
    vault_key: str
    base_url: str = "https://proxy.keybrake.com/stripe"

@resource(config_schema={"budget_usd": float})
def keybrake_stripe_resource(context):
    run_id = context.run_id
    r = httpx.post(
        "https://proxy.keybrake.com/vault/keys",
        headers={"Authorization": f"Bearer {os.environ['KEYBRAKE_API_KEY']}"},
        json={
            "vendor": "stripe",
            "daily_usd_cap": context.resource_config["budget_usd"],
            "allowed_endpoints": ["POST /v1/payment_intents"],
            "expires_in": "4h",
            "agent_run_label": f"dagster/{run_id}",
        },
    )
    key = r.json()["vault_key"]
    return KeybrakeResource(vault_key=key)

@asset(
    partitions_def=customer_partitions,
    retry_policy=RetryPolicy(max_retries=3, delay=1.0),
    required_resource_keys={"keybrake"},
)
def charged_customer(context) -> dict:
    kb = context.resources.keybrake
    stripe.api_key = kb.vault_key              # scoped vault key from resource
    stripe.api_base = kb.base_url
    customer_id = context.partition_key
    result = stripe.PaymentIntent.create(
        amount=2999,
        currency="usd",
        customer=customer_id,
        idempotency_key=f"{customer_id}-2999",  # stable on retry
    )
    return {"status": result.status, "id": result.id}

The keybrake_stripe_resource is initialized once when the Dagster run starts. All charged_customer materializations in the run share the same KeybrakeResource instance and its vault key. The vault key's cap accumulates across all concurrent materializations. The real Stripe secret stays in Keybrake, never in Dagster's environment or asset metadata. The audit log records each call with agent_run_label: "dagster/{run_id}" — queryable by run ID or time window.

How Keybrake fits

Keybrake is the proxy layer between your Dagster assets and Stripe, Twilio, or Resend. You wrap it in a Dagster resource, issue the vault key at resource initialization (run start), and inject it into any asset that makes vendor calls. The real Stripe secret stays in Keybrake, not in Dagster's configuration or asset code. Each run gets its own vault key with its own dollar cap, endpoint allowlist, and expiry. Concurrent partition materializations that exceed the cap return 429s — these surface as Dagster materialization failures with structured traceback data, not silent charges across thousands of concurrent partitions.

Get early access

Related questions

Further reading

• Airflow AI agent API key — similar per-run vault key pattern for Airflow DAGs; Dagster dynamic partitions map to Airflow's dynamic task mapping.
• AI agent secrets management — why secrets stores alone aren't enough and the four-layer model that closes the enforcement gap for data pipelines.
• AI agent kill switch patterns — the four ways to stop a runaway agent and their real stop latencies, including pipeline executor scenarios.
• AI agent credential management — architecture framework for managing credentials across multi-step agent pipelines.