Stop the Bill Before It Balloons: Designing API Key Blast Radius for Unattended Pipelines

Rotating Keys with Zero Downtime

Separating scopes is pointless if you can't actually swap a key — periodically, or the instant a leak is suspected. The hard part is swapping without stopping production. Any job running at the moment you delete the old key fails with a 401.

I solved it with "two overlapping keys." Always keep a primary and a secondary valid, and treat rotation as a state transition. Issue the new key as secondary, switch the client to prefer the new one, then revoke the old one after a sufficient grace window. During that grace window both are valid, so in-flight jobs don't fail.

# key_state.py — zero-downtime key rotation via two overlapping keys
# A JSON state file holds "which one to prefer right now."
import json
import os
from pathlib import Path
 
STATE_PATH = Path(os.environ.get("KEY_STATE_PATH", "/secure/key_state.json"))
 
def load_state() -> dict:
    # First run prefers primary. secondary may be empty.
    if not STATE_PATH.exists():
        return {"active": "primary", "primary_env": "ANTHROPIC_API_KEY_PRIMARY",
                "secondary_env": "ANTHROPIC_API_KEY_SECONDARY"}
    return json.loads(STATE_PATH.read_text())
 
def active_api_key() -> str:
    """Resolve the currently preferred key from the environment.
    If the preferred side is unset, fall back to the other (grace-window safety)."""
    st = load_state()
    primary = os.environ.get(st["primary_env"], "")
    secondary = os.environ.get(st["secondary_env"], "")
    prefer = primary if st["active"] == "primary" else secondary
    fallback = secondary if st["active"] == "primary" else primary
    key = prefer or fallback
    if not key:
        raise RuntimeError("No valid API key in the environment (both unset)")
    return key
 
def promote_secondary() -> None:
    """Call after placing the new key in secondary. Tips preference to secondary.
    primary is still valid at this point, so in-flight jobs do not fail."""
    st = load_state()
    st["active"] = "secondary" if st["active"] == "primary" else "primary"
    STATE_PATH.write_text(json.dumps(st, ensure_ascii=False, indent=2))
    print(f"Preferred key switched to {st['active']}. Revoke the old key after the grace window.")
 
if __name__ == "__main__":
    # Sanity check: feed dummy env vars and confirm which key resolves
    os.environ.setdefault("ANTHROPIC_API_KEY_PRIMARY", "sk-ant-PRIMARY-PLACEHOLDER")
    os.environ.setdefault("ANTHROPIC_API_KEY_SECONDARY", "sk-ant-SECONDARY-PLACEHOLDER")
    print("active:", active_api_key())   # -> primary placeholder
    promote_secondary()
    print("active:", active_api_key())   # -> secondary placeholder

The operational flow: (1) issue a new key in the console and place it in ANTHROPIC_API_KEY_SECONDARY. (2) flip preference with promote_secondary(). (3) wait 30-60 minutes — longer than your longest job. (4) revoke the old key in the console, overwrite ANTHROPIC_API_KEY_PRIMARY with the new value, and reset state to primary. In an emergency where a leak is suspected, drop the grace window in step (3), revoke immediately, and let retries handle the 401s on in-flight jobs. Why hold state in a file? Because scheduled processes start fresh each time; in-process memory can't share "which one is current" across runs.

One more thing: read keys only from the watchdog and production jobs, and make sure generation code running in a sandbox can't read them. That shrinks the exposed surface further. To apply the same idea on the Claude Code side, the setting that blocks the sandbox from reading credential files is the counterpart.

Put a Watchdog on Usage Anomalies

Scope and rotation were the "make damage small and short" layers. The last one is "notice fast." The classic signature of a leaked key is token consumption at an unusual hour, an order of magnitude off. My production jobs run about 20 times a day, and each run's input/output tokens land in a fairly fixed range. That's exactly why deviation from normal is detectable.

A fixed threshold ("warn if more than a million tokens in an hour") stops working the moment normal shifts. I went with a rolling baseline plus median absolute deviation (MAD). Take the median and spread from recent history, and judge how many "MADs" the current value sits from there (a robust z-score). The reason for median and MAD instead of mean and standard deviation: the baseline isn't dragged by past spikes themselves (it resists contamination).

# usage_watchdog.py — robustly detect usage spikes (standard library only)
from statistics import median
 
def robust_z_scores(series):
    """Return how many MADs each point sits from the median.
    Add a tiny floor to avoid choking when MAD=0 (all points equal)."""
    if len(series) < 3:
        return [0.0] * len(series)
    med = median(series)
    abs_dev = [abs(x - med) for x in series]
    mad = median(abs_dev)
    scale = 1.4826 * mad if mad > 0 else 1e-9  # factor that matches sigma under normality
    return [(x - med) / scale for x in series]
 
def detect_spikes(usage_by_bucket, z_threshold=6.0, min_tokens=50_000):
    """usage_by_bucket: [(bucket_label, tokens), ...] in time order
    Returns the list of buckets to alert on.
    Values below min_tokens are ignored even if their ratio swings (false-alarm suppression)."""
    series = [t for _, t in usage_by_bucket]
    zs = robust_z_scores(series)
    alerts = []
    for (label, tokens), z in zip(usage_by_bucket, zs):
        if z >= z_threshold and tokens >= min_tokens:
            alerts.append({"bucket": label, "tokens": tokens, "robust_z": round(z, 1)})
    return alerts
 
if __name__ == "__main__":
    # Normal is around 80k tokens. Only the last hour jumps, as if from a leak.
    data = [
        ("06-27T00", 78_000), ("06-27T01", 81_000), ("06-27T02", 79_500),
        ("06-27T03", 77_000), ("06-27T04", 82_000), ("06-27T05", 80_000),
        ("06-27T06", 1_900_000),   # <- a surge modeling a leaked-key third party
    ]
    for a in detect_spikes(data):
        print(f"SPIKE: {a['bucket']} tokens={a['tokens']:,} robust_z={a['robust_z']}")
    # Expected output:
    # SPIKE: 06-27T06 tokens=1,900,000 robust_z=... (a large positive value)

This script runs as-is with no external dependencies. Run it locally and the six normal points form the baseline while only the final surge is flagged with a large robust z. A fixed threshold would false-alarm just because "normal rose from 80k to 120k," but the MAD-based one follows the rising baseline automatically. To wire it to real data, build usage_by_bucket from the Admin usage/cost report at hourly-bucket granularity. I've walked through using that report API in making consumption visible with the usage and cost API, so you can lean the fetch part on that.

Wire the Watchdog into a Scheduled Run

Once the detection logic exists, run it more often than human eyes would. I made it a light task separate from the production job: pull the usage report every hour, pass it through detect_spikes, and notify if anything fires. The destination can be anything, but because this is unattended, the point is to route it somewhere you'll notice immediately (for me, a chat notification).

# watchdog_run.py — the hourly watchdog body (swap the stubbed report fetch for a real one)
import json
import urllib.request
from usage_watchdog import detect_spikes
 
def fetch_usage_buckets(admin_key: str):
    """Fetch the Admin usage report by time bucket and shape it into
    [(bucket_label, total_tokens), ...].
    Note: confirm the endpoint and response shape against the latest official docs."""
    # Implementation sketch (verify the official spec):
    # req = urllib.request.Request(USAGE_REPORT_URL, headers={"x-api-key": admin_key})
    # raw = json.loads(urllib.request.urlopen(req, timeout=30).read())
    # return [(b["starting_at"], b["input_tokens"] + b["output_tokens"]) for b in raw["data"]]
    raise NotImplementedError("Swap in a real report fetch")
 
def notify(message: str):
    """For unattended ops, route to somewhere you will notice immediately."""
    print(message)  # in production, send to chat etc.
 
def main():
    import os
    admin_key = os.environ["ANTHROPIC_ADMIN_KEY"]  # a least-privilege key just for the watchdog
    buckets = fetch_usage_buckets(admin_key)
    alerts = detect_spikes(buckets, z_threshold=6.0, min_tokens=50_000)
    if alerts:
        lines = [f"{a['bucket']}: {a['tokens']:,} tokens (z={a['robust_z']})" for a in alerts]
        notify("API usage spike detected. Consider revoking keys as a suspected leak:\n" + "\n".join(lines))
    else:
        print("Usage is within the normal range.")
 
if __name__ == "__main__":
    main()

To pick the threshold, run the last two weeks of buckets through robust_z_scores and see how tight normal stays. In my case normal mostly fell within +/-2, so z_threshold=6.0 is a safe setting that only catches "clearly wrong." For watchdog frequency and avoiding scheduled jobs eating each other's limits, the thinking in designing rate-limit headroom for shared scheduled jobs applies directly.

Pitfalls That Are Easy to Hit

Building this for real, I stepped on a few. Sharing them.

Skimp on the rotation grace window and long in-flight jobs die with 401. At first I revoked the old key right after switching, and a generation that happened to be running died midway. The grace window should be "your longest job plus buffer"; the only time to shorten it is an emergency revocation.

Don't contaminate the watchdog baseline with your own legitimate spikes. If you batch a big generation run at the start of the month, that day's bucket lifts the baseline. MAD is robust to it, but a clearly different "planned bulk run" is better treated as a separate series or temporarily muted, to cut false alarms.

Over-alerting breeds the habit of ignoring. I set the threshold too low at first, it rang repeatedly on normal jitter, and I stopped looking. Aim the watchdog at "rarely rings, but real when it does," and shave false alarms with the two-stage z_threshold and min_tokens.

Last, the plainest and most important point: don't log the key. Leave a print(active_api_key()) in for debugging and a plaintext key flows into your logging stack. You can design a small blast radius and still lose it all through the observability path. Even in the watchdog code, I never mix the key itself into output.

In Closing

If you do only one thing, carve the key of your heaviest production job into a dedicated workspace and put a spend limit on that workspace. Just separating scope and placing a limit drops the maximum possible loss from a leak sharply. The watchdog and rotation can be stacked on that foundation gradually.

Running an API on a schedule unattended is convenient, but I feel it sits right next to the fear of damage spreading without my noticing. As much as the effort not to leak, design for after the leak — that perspective is what gives me peace of mind running several pipelines alone. Thank you for reading.