Building an API key rotation workflow with dual keys

Most LLM integrations break the moment a single API key hits a rate limit or gets revoked. A solid api key rotation workflow dual keys approach keeps two live credentials in flight, swapping them transparently when one fails. This tutorial builds that mechanism in Python with runnable code and checkpoints you can run locally.

Prerequisites

• Python 3.10 or newer
• requests library (pip install requests)
• Two valid API keys from the same provider (we’ll use OpenAI’s real endpoint https://api.openai.com/v1/chat/completions)
• Environment variables KEY_A and KEY_B exported before running
• Familiarity with HTTP 401/429 responses and basic threading

If you don’t have a second key, create one in your provider dashboard. Never embed keys in source; we load from env.

The dual-key model

The core idea: maintain a primary and a secondary key. Normal traffic uses primary. On a fatal error (auth failure, sustained rate limit), you demote primary and promote secondary atomically. The api key rotation workflow dual keys pattern differs from simple retry because it rotates credentials, not just replays the same doomed request.

We track per-key health: consecutive failure timestamp and cooldown. A key becomes eligible for promotion only after its cooldown expires.

Step 1: Thread-safe key vault

We need a vault readable and swappable from multiple worker threads. Use threading.Lock.

import os
import threading
import time

class DualKeyVault:
    def __init__(self, key_a: str, key_b: str, cooldown: int = 300):
        self._lock = threading.Lock()
        self.primary = key_a
        self.secondary = key_b
        self.cooldown = cooldown
        self.primary_fail_since = None
        self.secondary_fail_since = None

    def get_active(self) -> str:
        with self._lock:
            return self.primary

    def mark_failed(self, key: str):
        with self._lock:
            now = time.time()
            if key == self.primary:
                self.primary_fail_since = now
                if self.secondary_fail_since is None or \
                   (now - self.secondary_fail_since) > self.cooldown:
                    self.primary, self.secondary = self.secondary, self.primary
                    self.primary_fail_since, self.secondary_fail_since = \
                        self.secondary_fail_since, self.primary_fail_since
            elif key == self.secondary:
                self.secondary_fail_since = now

    def mark_ok(self, key: str):
        with self._lock:
            if key == self.primary:
                self.primary_fail_since = None
            elif key == self.secondary:
                self.secondary_fail_since = None

This vault swaps only when the secondary is outside its cooldown window. That prevents flapping when both keys are dead.

Step 2: API call with fallback

Wrap the chat completion call. Try primary; on 401 or 429, mark it failed (which may swap) and retry once with the now-active key.

import requests

ENDPOINT = "https://api.openai.com/v1/chat/completions"

def chat_completion(vault: DualKeyVault, model: str, prompt: str):
    payload = {
        "model": model,
        "messages": [{"role": "user", "content": prompt}],
        "max_tokens": 50,
    }
    key = vault.get_active()
    for attempt in range(2):
        resp = requests.post(
            ENDPOINT,
            headers={"Authorization": f"Bearer {key}"},
            json=payload,
            timeout=10,
        )
        if resp.status_code == 200:
            vault.mark_ok(key)
            return resp.json()
        if resp.status_code in (401, 429):
            vault.mark_failed(key)
            key = vault.get_active()
            continue
        resp.raise_for_status()
    raise RuntimeError("Both keys failed")

Expected output on a healthy run:

{
  "id": "chatcmpl-abc",
  "object": "chat.completion",
  "choices": [
    {"message": {"role": "assistant", "content": "Hello!"}}
  ]
}

If KEY_A is expired, the first attempt returns 401. The vault swaps, second attempt with KEY_B succeeds. You get the same JSON, no exception.

Step 3: Proactive rotation on a schedule

Reactive swapping handles outages, but compliance often requires periodic rotation. Add a daemon that forces a swap every N seconds, assuming both keys are valid.

import threading

def rotation_daemon(vault: DualKeyVault, interval: int):
    def loop():
        while True:
            time.sleep(interval)
            with vault._lock:
                vault.primary, vault.secondary = vault.secondary, vault.primary
                vault.primary_fail_since = None
                vault.secondary_fail_since = None
    t = threading.Thread(target=loop, daemon=True)
    t.start()

Start it after vault creation:

vault = DualKeyVault(os.environ["KEY_A"], os.environ["KEY_B"])
rotation_daemon(vault, interval=3600)  # swap hourly

This keeps the api key rotation workflow dual keys pattern ahead of key leaks. The daemon only swaps healthy keys; if one is failed, the cooldown logic in mark_failed still protects you.

Step 4: Integration with a gateway

If you front your LLM traffic through an inference gateway like n4n.ai, the OpenAI-compatible endpoint honors client routing directives and forwards provider cache-control hints. Your rotation layer only needs to change the Authorization header—no changes to request bodies or model names. The gateway’s automatic fallback complements your dual-key strategy: if your primary provider degrades, the gateway reroutes while your vault handles key-level failures.

For local testing, point ENDPOINT at your gateway URL instead of api.openai.com. The same code works unchanged.

Step 5: Test harness and verification

Write a small script that forces a failure by using a bogus primary key, then confirms swap.

if __name__ == "__main__":
    os.environ["KEY_A"] = "sk-bogus"
    os.environ["KEY_B"] = os.environ.get("REAL_KEY", "sk-good")
    v = DualKeyVault(os.environ["KEY_A"], os.environ["KEY_B"], cooldown=0)
    try:
        out = chat_completion(v, "gpt-3.5-turbo", "hi")
        print("Success:", out["choices"][0]["message"]["content"])
    except Exception as e:
        print("Failed:", e)

Run with REAL_KEY set to a valid key. Expected console output:

Success: Hello!

If both keys are invalid, you’ll get Failed: Both keys failed. That’s the circuit breaker doing its job.

Operational notes

• Metrics: Export primary_fail_since to your monitoring. A key that flips to failed more than once per hour warrants investigation.
• Secret storage: Use a secrets manager (AWS Secrets Manager, HashiCorp Vault) to populate KEY_A/KEY_B at boot. Don’t hardcode.
• Concurrency: The lock keeps swaps safe within one process. If you run many processes, lift the vault state into Redis or similar shared store with the same swap semantics.
• Cooldown tuning: 300s is sane for rate limits; auth failures should swap immediately (cooldown 0).

The api key rotation workflow dual keys design is deliberately simple. It avoids complex consensus and works in a single process. For distributed systems, externalize the vault state and keep the same swap rules.

Final checklist

• Two keys loaded from env, not code
• Vault swaps on 401/429 only when secondary healthy
• Caller retries exactly once after swap
• Background rotation daemon optional but recommended
• Gateway layer (if any) just sees header change

Ship this behind your LLM client and you’ll stop paging on key revocation events.