Django AI agent API key management: scoped vault keys for agent views and tasks

The Django AI agent tool backend pattern

A typical Django AI agent tool backend looks like this:

from django.conf import settings
import stripe
from rest_framework.views import APIView
from rest_framework.response import Response

stripe.api_key = settings.STRIPE_API_KEY  # shared across ALL workers

class ChargeTool(APIView):
    def post(self, request):
        intent = stripe.PaymentIntent.create(
            amount=request.data["amount_cents"],
            currency="usd",
            customer=request.data["customer_id"]
        )
        return Response(intent)

This is clean and idiomatic Django. The problem: stripe.api_key is a module-level global. When gunicorn spins up four workers, all four share the same Stripe credential. An agent stuck in a retry loop in one user's session can rack up charges using any other user's customer data — there's no per-request spending boundary. And rotating the key to stop the runaway agent disrupts every other active session simultaneously.

Adding vault keys via Django middleware

Django middleware is the right integration point — it wraps every request and response, runs before your views, and has access to the request object that views already receive:

import requests as http
from django.conf import settings
from django.http import JsonResponse

class VaultKeyMiddleware:
    """Issues a per-request vault key and attaches it to request.vault_key."""

    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        # Skip non-agent endpoints (admin, static, health checks)
        if not request.path.startswith("/tools/"):
            return self.get_response(request)

        user_id = request.headers.get("X-User-Id", "anonymous")
        run_id = request.headers.get("X-Agent-Run-Id", "unknown")

        try:
            resp = http.post(
                "https://api.keybrake.com/v1/keys",
                headers={"Authorization": f"Bearer {settings.KEYBRAKE_TOKEN}"},
                json={
                    "label": f"django-{user_id}-{run_id}",
                    "vendor": "stripe",
                    "allowed_endpoints": [
                        "/v1/payment_intents",
                        "/v1/payment_intents/*"
                    ],
                    "daily_usd_cap": 500,
                    "expires_in": "5m"
                },
                timeout=5
            )
            resp.raise_for_status()
            key_data = resp.json()
            request.vault_key = key_data["token"]
            request.vault_key_id = key_data["id"]
        except Exception as e:
            # Log and fail open or closed depending on your risk tolerance
            import logging
            logging.error(f"Vault key issuance failed: {e}")
            request.vault_key = None
            request.vault_key_id = None

        response = self.get_response(request)

        # Revoke vault key after the view has finished
        if getattr(request, "vault_key_id", None):
            try:
                http.delete(
                    f"https://api.keybrake.com/v1/keys/{request.vault_key_id}",
                    headers={"Authorization": f"Bearer {settings.KEYBRAKE_TOKEN}"},
                    timeout=5
                )
            except Exception:
                pass  # TTL is the safety net if revocation fails

        return response

Register the middleware in settings.py:

MIDDLEWARE = [
    "django.middleware.security.SecurityMiddleware",
    # ... other middleware ...
    "yourapp.middleware.VaultKeyMiddleware",  # add after auth middleware
]

Now views receive the vault key on the request object:

import requests as http
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework import status

class ChargeTool(APIView):
    def post(self, request):
        if not request.vault_key:
            return Response({"error": "vault_key_unavailable"}, status=503)

        resp = http.post(
            "https://proxy.keybrake.com/stripe/v1/payment_intents",
            headers={"Authorization": f"Bearer {request.vault_key}"},
            json={
                "amount": request.data["amount_cents"],
                "currency": "usd",
                "customer": request.data["customer_id"]
            }
        )
        if resp.status_code == 429 and resp.json().get("code") == "cap_exhausted":
            return Response({"error": "spend_cap_exceeded"}, status=402)
        resp.raise_for_status()
        return Response(resp.json())

Vault keys with Django REST Framework serializers and viewsets

DRF viewsets and serializers work cleanly with the middleware approach because request is passed through the DRF call stack just as it is in class-based views. If you use DRF serializer context to pass the request into nested serializers, the vault key travels with it:

class BillingSerializer(serializers.Serializer):
    customer_id = serializers.CharField()
    amount_cents = serializers.IntegerField()

    def create(self, validated_data):
        request = self.context["request"]
        vault_key = request.vault_key

        resp = requests.post(
            "https://proxy.keybrake.com/stripe/v1/payment_intents",
            headers={"Authorization": f"Bearer {vault_key}"},
            json={
                "amount": validated_data["amount_cents"],
                "currency": "usd",
                "customer": validated_data["customer_id"]
            }
        )
        resp.raise_for_status()
        return resp.json()

class BillingViewSet(viewsets.GenericViewSet):
    serializer_class = BillingSerializer

    def get_serializer_context(self):
        ctx = super().get_serializer_context()
        ctx["request"] = self.request  # vault_key is on self.request
        return ctx

Django + Celery: vault keys for async agent tasks

Many Django AI agent backends enqueue Celery tasks from views. The vault key issued in Django middleware does not cross the Celery task boundary — Celery workers don't have access to the request object. Issue a fresh vault key at the start of the Celery task instead:

from celery import shared_task
import requests as http
from django.conf import settings

@shared_task(bind=True, max_retries=3)
def run_billing_agent(self, run_id: str, customer_ids: list):
    # Issue vault key at task start, not in the Django request
    resp = http.post(
        "https://api.keybrake.com/v1/keys",
        headers={"Authorization": f"Bearer {settings.KEYBRAKE_TOKEN}"},
        json={
            "label": f"celery-billing-{run_id}",
            "vendor": "stripe",
            "allowed_endpoints": ["/v1/payment_intents", "/v1/payment_intents/*"],
            "daily_usd_cap": 5000,
            "expires_in": "30m"  # longer TTL for async tasks
        }
    )
    resp.raise_for_status()
    vault_key = resp.json()["token"]
    key_id = resp.json()["id"]

    try:
        for customer_id in customer_ids:
            process_single_customer(customer_id, vault_key)
    except Exception as exc:
        # Revoke before retry to avoid duplicate active keys
        http.delete(
            f"https://api.keybrake.com/v1/keys/{key_id}",
            headers={"Authorization": f"Bearer {settings.KEYBRAKE_TOKEN}"}
        )
        raise self.retry(exc=exc, countdown=60)
    finally:
        http.delete(
            f"https://api.keybrake.com/v1/keys/{key_id}",
            headers={"Authorization": f"Bearer {settings.KEYBRAKE_TOKEN}"}
        )

Performance and concurrency considerations

Get early access

Related questions

Further reading

• Flask AI agent API key — the same vault key pattern for Flask's WSGI request model, using before_request hooks and Flask's g context object.
• FastAPI AI agent API key — async ASGI vault key integration for FastAPI using dependency injection, which avoids the middleware-level blocking HTTP call.
• Celery AI agent API key — vault key lifecycle for Celery workers used in async Django agent architectures, where keys must be issued at task start rather than at request start.
• AI agent API key lifecycle — the four lifecycle phases (issuance, enforcement, expiration, revocation) and how they map to Django's middleware request lifecycle.
• AI agent credential management — broader credential management architecture for Python-based agent backends, including Django settings management and secret rotation patterns.