AI & LLMs Intermediate

Calling the Claude API from Django — Setup, Streaming, Error Handling

From pip install to a streaming chat view in production. Authentication, error handling, prompt caching, and cost-aware patterns for Django + Claude.

KC Ramo · DjangoZen Team May 09, 2026 31 min read 380 views

Adding AI to a Django application most often means calling a model provider's API, and doing it well — with proper setup, streaming for a good user experience, and robust error handling — is what separates a fragile demo from a production feature. This tutorial covers integrating a language model API into Django the right way, focusing on the practical concerns of configuration, making requests, streaming responses, handling failures, and managing the operational realities of an external AI dependency.

AI as an API integration

The reassuring reality for a Django developer is that calling an AI model is, at its core, an API integration — familiar territory. You send a request containing your prompt and configuration to the provider's endpoint, and you receive the model's generated response. The sophistication is in the model behind the API, but your side is the well-understood work of making HTTP requests, handling responses, and managing credentials and errors. Understanding that adding AI is fundamentally integrating an external API, not building anything exotic, makes the task approachable: you apply the same care you would to any important third-party integration, with a few AI-specific considerations layered on top.

Setup and configuration

The first step is configuration: obtaining an API key from the provider and storing it securely. The key is a credential that must never be hard-coded or committed to version control — it belongs in your environment configuration, loaded from an environment variable, exactly as you handle other secrets in Django. You install the provider's client library, which simplifies making requests, and configure it with your key. Getting this foundation right — secure key storage and proper client setup — is essential, both for security, since a leaked API key can be abused at your expense, and as the basis for everything else you build on the integration.

Installing the SDK and wiring it into settings

The concrete version of that setup for Claude is the official anthropic Python package. Add it to your requirements alongside Django, then read the key and your model choices in settings.py so that every other module gets them from one place:

# requirements.txt
Django>=5.2,<6.0
anthropic

# myproject/settings.py
import os

ANTHROPIC_API_KEY = os.environ["ANTHROPIC_API_KEY"]  # fail fast at startup if missing
CLAUDE_MODEL = os.environ.get("CLAUDE_MODEL", "claude-sonnet-5")
CLAUDE_FAST_MODEL = os.environ.get("CLAUDE_FAST_MODEL", "claude-haiku-4-5-20251001")
CLAUDE_TIMEOUT_SECONDS = float(os.environ.get("CLAUDE_TIMEOUT_SECONDS", "60"))
CLAUDE_MAX_RETRIES = int(os.environ.get("CLAUDE_MAX_RETRIES", "2"))

Using os.environ["..."] rather than .get() for the key is deliberate: a missing key should stop the process at boot, not surface as a confusing authentication error on the first user request. Keeping the model IDs in settings means you can move a feature from one model to another with an environment change instead of a code deploy.

One shared client per process

The client object holds a connection pool, so create it once and reuse it rather than constructing a new one inside every view. A small module with a cached factory keeps this tidy and gives tests a single seam to patch:

# myapp/claude.py
from functools import lru_cache

import anthropic
from django.conf import settings


@lru_cache(maxsize=1)
def get_client() -> anthropic.Anthropic:
    return anthropic.Anthropic(
        api_key=settings.ANTHROPIC_API_KEY,
        timeout=settings.CLAUDE_TIMEOUT_SECONDS,
        max_retries=settings.CLAUDE_MAX_RETRIES,
    )

Setting timeout explicitly matters. The SDK's default timeout is long, sized for big generations, and a web worker that waits minutes for a stalled connection is a worker that cannot serve anyone else. Pick a ceiling that fits the feature; you can override it for an individual call with get_client().with_options(timeout=...) when one code path legitimately needs more time. In separate environments (development, staging, production) use separate API keys, so that you can revoke one without touching the others and so that usage reports show where tokens are actually being spent.

Making your first request

With setup done, a basic request sends your prompt to the model and receives its response. You construct the request with the input and any configuration — which model, how the output should behave — send it through the client, and get back the generated text to use in your application. This basic request-response is the core of the integration, and everything else refines it. Understanding the shape of a simple call — provide a prompt, get a response — gives you the foundation, and from there you add the production concerns of streaming, error handling, and operational management that turn a basic call into a robust feature.

A minimal service function

Keep API calls out of views and put them in a service function that the rest of the project calls. That function is where error handling, logging and usage tracking will accumulate later, so it pays to have exactly one of them:

# myapp/services.py
from django.conf import settings

from .claude import get_client


def generate_text(prompt: str, *, system: str | None = None, max_tokens: int = 1024) -> str:
    kwargs = {
        "model": settings.CLAUDE_MODEL,
        "max_tokens": max_tokens,
        "messages": [{"role": "user", "content": prompt}],
    }
    if system:
        kwargs["system"] = system
    response = get_client().messages.create(**kwargs)
    return "".join(block.text for block in response.content if block.type == "text")

What comes back

The response is not a bare string. response.content is a list of content blocks, which is why the function joins the text blocks rather than blindly reading content[0]. Two other fields deserve attention on every call. response.stop_reason tells you why generation ended: "end_turn" means the model finished naturally, while "max_tokens" means it was cut off by your max_tokens limit and the text is incomplete. response.usage carries input_tokens and output_tokens, the numbers your bill is computed from. Note also that max_tokens is required: it is a hard ceiling on output length, and choosing it per feature (a short classification needs a few dozen tokens, a drafted email perhaps a thousand) is your first and cheapest cost control.

You can try the function immediately from python manage.py shell before building any UI around it. That quick loop is the best place to iterate on prompts, since you see raw output without templates or JavaScript in the way.

Where to put API calls

An architectural decision is where in your Django application the API calls live. Because model calls take time — often several seconds — making them directly in a request handler can make the request slow, so for many cases it is better to call the API from a background task rather than blocking a web request, or to use streaming to deliver the response progressively. Organizing your AI calls deliberately, keeping slow operations off the request path where appropriate, is important for a responsive application. Understanding that the latency of model calls shapes where and how you make them — not casually inline in a fast request — is part of integrating AI well into a Django application's architecture.

Choosing between inline, streamed and background calls

In practice there are three placements, and the right one depends on who is waiting for the result:

PlacementGood forWatch out for
Inline in the viewShort outputs (tags, a one-line summary) with a small max_tokensWorker time is tied up for the whole call; needs a tight timeout
Streamed from the viewChat and drafting UIs where a person is watchingLong-lived connections; proxy buffering; errors after headers are sent
Background task (Celery, RQ, Django tasks)Batch jobs, documents, anything nobody watches liveNeeds result storage and a way to notify or poll

A background task is also the natural home for retries on a longer schedule. With Celery, for example, a bound task can re-queue itself when the provider is overloaded, and the view that enqueued it returns immediately:

# myapp/tasks.py
import anthropic
from celery import shared_task

from .models import Document
from .services import generate_text


@shared_task(bind=True, max_retries=5)
def summarize_document(self, document_id: int) -> None:
    doc = Document.objects.get(pk=document_id)
    try:
        doc.summary = generate_text(doc.body[:20000], system="Summarize in five bullet points.")
    except (anthropic.RateLimitError, anthropic.APIConnectionError) as exc:
        raise self.retry(exc=exc, countdown=30 * (2 ** self.request.retries))
    doc.save(update_fields=["summary"])

The task takes an ID, not the object, and re-reads it from the database, so it always works on current data and the message on the queue stays small.

Streaming responses

Model responses are generated progressively, token by token, and can take several seconds to complete, so streaming — delivering the response to the user as it is generated rather than waiting for the whole thing — dramatically improves the user experience. Instead of staring at a blank screen, the user sees text appear and can start reading immediately. The provider's API supports streaming, and you handle the incoming chunks as they arrive, passing them to the user. Understanding that streaming makes AI features feel responsive rather than slow is important, because for any user-facing AI feature, the difference between waiting for a complete response and watching it appear is the difference between feeling broken and feeling fast.

Streaming through Django

To stream a model's response to the browser through Django, you use a streaming response that yields chunks as they arrive from the model, typically delivered to the front-end using a mechanism suited to one-directional server-to-client updates. As each piece comes back from the API, you send it on to the client immediately rather than buffering the whole response. This connects the model's progressive generation to the user's screen in real time. Understanding how to wire streaming through Django — a streaming response backed by the incoming model chunks — is the practical skill that lets you build the responsive, progressively-rendering AI interfaces that users expect from modern AI features.

A server-sent events view

Here is a complete streaming endpoint. It uses the SDK's client.messages.stream helper, whose text_stream iterator yields text fragments as they arrive, and wraps each fragment in the server-sent events (SSE) wire format:

# myapp/views.py
import json
import logging

import anthropic
from django.conf import settings
from django.contrib.auth.decorators import login_required
from django.http import StreamingHttpResponse
from django.views.decorators.http import require_POST

from .claude import get_client

logger = logging.getLogger(__name__)


def sse(data: dict, event: str | None = None) -> str:
    prefix = f"event: {event}\n" if event else ""
    return f"{prefix}data: {json.dumps(data)}\n\n"


@login_required
@require_POST
def chat_stream(request):
    prompt = json.loads(request.body).get("prompt", "")[:4000]

    def event_stream():
        try:
            with get_client().messages.stream(
                model=settings.CLAUDE_MODEL,
                max_tokens=1024,
                messages=[{"role": "user", "content": prompt}],
            ) as stream:
                for text in stream.text_stream:
                    yield sse({"text": text})
                final = stream.get_final_message()
            yield sse({"stop_reason": final.stop_reason}, event="done")
        except anthropic.APIError:
            logger.exception("Claude stream failed")
            yield sse({"message": "The assistant is unavailable. Please try again."}, event="error")

    response = StreamingHttpResponse(event_stream(), content_type="text/event-stream")
    response["Cache-Control"] = "no-cache"
    response["X-Accel-Buffering"] = "no"  # ask nginx not to buffer this response
    return response

Notice where the error handling sits. By the time the generator runs, Django has already sent a 200 status and headers, so a failure halfway through cannot become a 500 page. The only way to tell the browser something went wrong is in-band, as an error event, and the front-end must be written to expect it. After the loop, get_final_message() gives you the assembled message including usage, which is the right place to record token counts for a streamed call.

Consuming the stream in the browser

The browser's built-in EventSource only issues GET requests, and you usually want to POST a prompt with a CSRF token, so read the stream with fetch instead:

const res = await fetch("/chat/stream/", {
  method: "POST",
  headers: {"Content-Type": "application/json", "X-CSRFToken": csrfToken},
  body: JSON.stringify({prompt}),
});
const reader = res.body.pipeThrough(new TextDecoderStream()).getReader();
let buffer = "";
while (true) {
  const {value, done} = await reader.read();
  if (done) break;
  buffer += value;
  const events = buffer.split("\n\n");
  buffer = events.pop();               // keep any incomplete event for the next read
  for (const evt of events) {
    const data = JSON.parse(evt.slice(evt.indexOf("data: ") + 6));
    if (evt.startsWith("event: error")) showError(data.message);
    else if (!evt.startsWith("event: done")) output.textContent += data.text;
  }
}

Network chunks do not line up with event boundaries, which is why the code buffers and splits on the blank line that terminates each SSE event. Append with textContent rather than innerHTML: model output is untrusted text, and if you later render Markdown, sanitize the resulting HTML.

WSGI, ASGI and long-lived responses

Under a WSGI server such as Gunicorn with sync workers, each streaming response occupies a whole worker for its entire duration, and Gunicorn's default 30-second worker timeout will kill a long generation mid-stream. Either raise the timeout and add workers for the streaming endpoints, or serve the project under ASGI (for example Uvicorn or Daphne) and write the view with anthropic.AsyncAnthropic. Since Django 4.2, StreamingHttpResponse accepts an async iterator, so an async def generator that iterates async for text in stream.text_stream inside async with client.messages.stream(...) works without tying up a thread per connection. Any database access inside that async view must use the async ORM methods or sync_to_async.

Robust error handling

An external API will sometimes fail — network issues, rate limits, service problems, timeouts — and production code must handle these gracefully rather than letting them break the user's experience. This means catching errors, handling rate limits by backing off and retrying transient failures, and degrading sensibly when the model is unavailable rather than showing a broken page. Because you depend on an external service you do not control, robust error handling is not optional. Understanding that an AI integration must anticipate and handle failure — that the model API will occasionally be slow or unavailable, and your application should cope cleanly — is what makes the difference between a feature that works reliably and one that breaks whenever the provider has a hiccup.

The SDK's exception hierarchy

The anthropic package raises typed exceptions, all deriving from anthropic.APIError. Two branches matter most: anthropic.APIConnectionError (the request never got a response, including timeouts) and anthropic.APIStatusError (the API answered with an error status, available as exc.status_code). anthropic.RateLimitError is the subclass of APIStatusError raised for HTTP 429. The status code tells you whether a retry can help:

StatusMeaningRetry?What to do
400Invalid request (bad parameters, malformed messages)NoBug in your code; log the request shape and fix it
401 / 403Missing, invalid or unauthorized keyNoCheck the key in this environment; alert immediately
404Unknown resource, often a mistyped model IDNoCorrect the model setting
413Request too largeNoTrim input or conversation history
429Rate limit exceededYes, after a delayBack off; honour the retry-after header
500 / 529Provider error or overloadedYesBack off, then degrade gracefully

Translating SDK errors into application errors

Views should not need to know about HTTP status codes from a third party. Convert the retryable cases into one application exception and let genuine bugs propagate so they reach your error tracker:

# myapp/services.py (continued)
import logging

import anthropic

logger = logging.getLogger(__name__)


class AIUnavailable(Exception):
    """The model could not be reached; the caller should degrade gracefully."""


def safe_generate_text(prompt: str, **kwargs) -> str:
    try:
        return generate_text(prompt, **kwargs)
    except anthropic.RateLimitError as exc:
        logger.warning("Claude rate limited; retry-after=%s", exc.response.headers.get("retry-after"))
        raise AIUnavailable("busy") from exc
    except anthropic.APIConnectionError as exc:
        logger.warning("Claude unreachable: %s", exc)
        raise AIUnavailable("unreachable") from exc
    except anthropic.APIStatusError as exc:
        if exc.status_code >= 500:
            logger.warning("Claude server error %s", exc.status_code)
            raise AIUnavailable("provider error") from exc
        raise  # 4xx: our request is wrong, so let it surface as a real error

The order of the except clauses matters: RateLimitError must come before APIStatusError because it is a subclass. A view then has a single, clear decision to make:

def suggest_reply(request, ticket_id):
    ticket = get_object_or_404(Ticket, pk=ticket_id)
    try:
        draft = safe_generate_text(ticket.body, system="Draft a polite support reply.")
    except AIUnavailable:
        draft = ""
        messages.info(request, "Suggestions are temporarily unavailable. You can still reply manually.")
    return render(request, "tickets/reply.html", {"ticket": ticket, "draft": draft})

Degrading gracefully usually means the surrounding page keeps working without the AI part, as here. For features where AI is the whole point, show a clear retry message instead of a generic error page, and never cache a failure as though it were a result.

Handling rate limits

Providers impose rate limits on how many requests you can make, and exceeding them returns errors you must handle. The right response is to respect the limits — backing off and retrying after a delay when you hit one, rather than hammering the API — and to design your usage to stay within them. For higher volume, you manage request rates deliberately. Understanding that rate limits are a normal part of using an API, and that handling them gracefully with backoff and retry is expected, prevents the failure where a burst of usage trips the limit and breaks your feature. It is a specific, common case of the broader error handling that robust integration requires.

Retries the SDK already does for you

Before writing your own retry loop, know that the client retries some failures automatically. Connection errors, 429s and server errors are retried with exponential backoff up to max_retries times (the setting from the setup section), and only when those attempts are exhausted does the exception reach your code. Adding a second retry loop on top multiplies attempts: three SDK attempts inside three of your own means nine requests against a provider that is already telling you to slow down. For interactive requests keep retries low (zero to two) so users get a fast, honest answer; for background tasks, lean on the task queue's delayed retries as shown earlier, which space attempts out over minutes rather than seconds.

Rate-limiting your own users

Your provider limits apply to your whole organization, so one enthusiastic user or a script hammering an endpoint can exhaust the budget for everyone. A per-user limit using Django's cache framework stops that before any tokens are spent:

from django.core.cache import cache


def allow_ai_request(user_id: int, limit: int = 20, window_seconds: int = 60) -> bool:
    key = f"ai-rate:{user_id}"
    cache.add(key, 0, timeout=window_seconds)  # creates the counter only if absent
    try:
        return cache.incr(key) <= limit
    except ValueError:  # key expired between add() and incr()
        cache.set(key, 1, timeout=window_seconds)
        return True

Use a shared cache backend such as Redis for this; the default local-memory cache is per process, so each worker would keep its own count. When the check fails, return HTTP 429 from your own view with a friendly message, which is far cheaper than letting the request reach the provider.

Managing cost

Because you pay per use, typically per token, cost is an operational concern to manage from the start. Usage scales with how much text flows through the model, so being efficient — sending only necessary context, choosing an appropriate model for the task, avoiding unnecessary calls — controls spend, and monitoring usage tells you where the cost goes. Setting limits guards against runaway costs from bugs or abuse. Understanding that an AI feature carries an ongoing, usage-driven cost that must be watched and controlled — unlike a fixed-cost server — is essential to running one sustainably in production, so the feature delivers value without producing surprising bills.

Recording token usage per call

Monitoring starts with recording what every call consumed. A small model, written from the service function after each response, lets you answer questions like "which feature uses the most tokens?" with an ordinary query:

# myapp/models.py
from django.conf import settings
from django.db import models


class AIUsage(models.Model):
    user = models.ForeignKey(settings.AUTH_USER_MODEL, null=True, on_delete=models.SET_NULL)
    feature = models.CharField(max_length=50)
    model = models.CharField(max_length=100)
    input_tokens = models.PositiveIntegerField()
    output_tokens = models.PositiveIntegerField()
    created_at = models.DateTimeField(auto_now_add=True, db_index=True)


# after each call:
AIUsage.objects.create(
    user=user, feature="ticket-reply", model=response.model,
    input_tokens=response.usage.input_tokens, output_tokens=response.usage.output_tokens,
)

With usage in the database, a daily budget check is a single aggregate. Summing today's tokens for a user with AIUsage.objects.filter(user=user, created_at__date=today).aggregate(total=Sum("output_tokens")) and refusing further requests above a threshold turns a runaway loop or an abusive account into a bounded problem. Convert tokens into money in your reporting using the provider's current published rates rather than hard-coding them.

Levers that actually reduce spend

  • Route by task. Classification, extraction and short rewrites often work well on a smaller, faster model such as settings.CLAUDE_FAST_MODEL; reserve the larger model for tasks where you have measured a quality difference.
  • Cap output. Set max_tokens per feature instead of one large global value.
  • Trim input. Input tokens are paid on every request, so long system prompts and full conversation histories are recurring costs, not one-off ones.
  • Cache deterministic work. If the same document is summarized repeatedly, store the summary keyed by a hash of the input and reuse it.

Taking it to production

Moving an AI integration from working to production-ready means addressing everything beyond the happy path: secure configuration, keeping slow calls off the request path or streaming them, comprehensive error and rate-limit handling, cost monitoring and controls, and validating or constraining the model's output where it matters. These operational concerns are the bulk of what makes an AI feature dependable, as opposed to a demo that works when everything goes right. Understanding that a robust integration is mostly about handling the realities of an external, paid, sometimes-imperfect dependency — and building those in deliberately — is what lets you ship an AI feature that holds up under real traffic rather than one that impresses briefly and then fails.

System prompts and configuration

Beyond the user's input, you typically provide a system instruction that sets the model's overall behavior, role, and constraints for the interaction — establishing how it should respond before it sees the specific request. You also configure parameters that shape the output. Setting these deliberately gives you control over the model's behavior across your feature. Understanding that you direct the model not just with the immediate input but with an overarching system instruction and configuration — defining its role and behavior for the whole interaction — is part of using the API well, because much of getting consistent, appropriate output comes from establishing the right framing and settings rather than only the per-request prompt.

In the Messages API the system instruction is a top-level system parameter, not a message in the messages list. Treat system prompts as code: keep them in version-controlled files or constants next to the feature that uses them, give each one a name that you also log (for example "ticket-reply-v3"), and change them through review like any other behavior change. When output quality shifts, being able to say exactly which prompt version produced a response is what makes the problem debuggable. Keep user-supplied text out of the system prompt; put it in the user message, clearly delimited, so the instructions you control and the content you do not stay separate.

Handling multi-turn conversations

For conversational features, the model needs the prior exchanges to respond in context, since it does not remember between calls. You maintain the conversation history and include it with each request, so the model sees what came before. Managing this history — keeping it, including it, and trimming it when it grows too long for the context window — is part of building a conversational feature. Understanding that conversation memory is something you manage by passing history into each request, not something the model retains on its own, is essential for building chat-like features correctly, where each response must account for the conversation so far that you supply as context.

Storing and trimming history

The messages list alternates "user" and "assistant" roles and must start with a user turn. Store each turn as a row, then rebuild the list for each request, keeping only the most recent turns when the history grows:

class ChatMessage(models.Model):
    conversation = models.ForeignKey("Conversation", on_delete=models.CASCADE, related_name="turns")
    role = models.CharField(max_length=10, choices=[("user", "user"), ("assistant", "assistant")])
    content = models.TextField()
    created_at = models.DateTimeField(auto_now_add=True)


def build_messages(conversation, max_turns: int = 20) -> list[dict]:
    turns = list(conversation.turns.order_by("-created_at")[:max_turns])[::-1]
    while turns and turns[0].role != "user":
        turns.pop(0)  # the trimmed history must still begin with a user message
    return [{"role": t.role, "content": t.content} for t in turns]

Save the assistant's reply only after the call succeeds; otherwise a failed request leaves two user turns in a row. Counting turns is a coarse limit, and for long documents in chat a better strategy is to summarize older turns into a short recap that you send in their place, which keeps context while bounding input tokens.

Validating and using the output

The model returns generated text, and how you handle it depends on your feature. For free-form text you may use it directly; for structured output you may have asked the model to produce a particular format that you then parse and validate before using. Because the model's output is not guaranteed to be perfectly formed, validating it where structure matters protects your application from malformed responses. Understanding that you should handle the model's output thoughtfully — using it directly where appropriate but validating it where your application depends on a specific structure — is part of robust integration, ensuring that the variability inherent in generated output does not break the features built on top of it.

Parsing structured output defensively

When you ask for JSON, parse it with the same suspicion you would apply to form input. The model may wrap it in prose or Markdown fences, omit a field, or run out of tokens mid-object:

import json

ALLOWED_CATEGORIES = {"billing", "bug", "feature", "other"}


class BadModelOutput(ValueError):
    pass


def parse_classification(response) -> dict:
    if response.stop_reason == "max_tokens":
        raise BadModelOutput("output truncated")
    text = "".join(b.text for b in response.content if b.type == "text").strip()
    start, end = text.find("{"), text.rfind("}")
    if start == -1 or end == -1:
        raise BadModelOutput("no JSON object found")
    try:
        data = json.loads(text[start:end + 1])
    except json.JSONDecodeError as exc:
        raise BadModelOutput("invalid JSON") from exc
    if data.get("category") not in ALLOWED_CATEGORIES:
        raise BadModelOutput(f"unexpected category {data.get('category')!r}")
    return data

Validating against an allow-list, rather than trusting whatever string comes back, is what keeps a model's creative answer from turning into a new, unexpected value in your database. On BadModelOutput, a single retry is reasonable; after that, fall back to a default such as "other" and log the raw text so you can improve the prompt. For richer schemas, a Django form or a validation library gives you field-level errors with little extra code.

Testing AI integrations

Testing code that calls an external AI API has its own considerations, since real calls are slow, cost money, and return varying output. For testing your integration logic — how you handle responses, errors, and rate limits — you often simulate the API rather than calling it for real, letting you test your code's behavior reliably and cheaply. Separately, you evaluate the AI feature's actual quality against representative cases. Understanding that testing an AI integration separates testing your handling code, where you simulate the API, from evaluating output quality, where you assess real results, lets you build confidence in both the robustness of your integration and the usefulness of the feature it powers.

Mocking the client in Django tests

Because every call goes through get_client(), a test can patch that one function and hand back a fake. The fake response only needs the attributes your code actually reads:

# myapp/tests/test_services.py
from types import SimpleNamespace
from unittest import mock

import anthropic
import httpx
from django.test import TestCase

from myapp.services import AIUnavailable, safe_generate_text


def fake_response(text: str, stop_reason: str = "end_turn"):
    return SimpleNamespace(
        content=[SimpleNamespace(type="text", text=text)],
        stop_reason=stop_reason,
        usage=SimpleNamespace(input_tokens=10, output_tokens=5),
        model="test-model",
    )


class SafeGenerateTextTests(TestCase):
    @mock.patch("myapp.services.get_client")
    def test_returns_text(self, get_client):
        get_client.return_value.messages.create.return_value = fake_response("Hello")
        self.assertEqual(safe_generate_text("Hi"), "Hello")

    @mock.patch("myapp.services.get_client")
    def test_rate_limit_becomes_ai_unavailable(self, get_client):
        request = httpx.Request("POST", "https://api.anthropic.com/v1/messages")
        error = anthropic.RateLimitError(
            "rate limited", response=httpx.Response(429, request=request), body=None
        )
        get_client.return_value.messages.create.side_effect = error
        with self.assertRaises(AIUnavailable):
            safe_generate_text("Hi")

Patch the name where it is looked up (myapp.services.get_client), not where it is defined. Also guard against accidental real calls in the test suite: setting ANTHROPIC_API_KEY to a dummy value in your test settings means a missed mock fails fast with an authentication error instead of silently spending tokens.

Evaluating quality separately

For output quality, keep a small set of representative inputs with the properties a good answer must have (valid category, mentions the order number, stays under a length limit) and run it against the real API as a separate command, not in the normal test suite. Run it whenever you change a prompt or switch models, and compare results before deploying. Even twenty well-chosen cases catch most regressions that a prompt tweak introduces.

Protecting the integration

An AI endpoint is an endpoint that spends money on every request, which makes it an attractive target for abuse and a liability when left open. A few defensive habits cover most of the risk:

  • Require authentication on every view that reaches the API, and apply the per-user rate limit shown earlier. An anonymous endpoint that forwards prompts is effectively free model access for anyone who finds it.
  • Bound input size in the view (the streaming example truncates the prompt) so that a single request cannot send an enormous payload.
  • Treat model output as untrusted. Escape it in templates (Django's autoescaping does this unless you use |safe), sanitize any HTML produced from Markdown, and never pass it to eval, raw SQL or shell commands.
  • Assume prompts can be manipulated. Users can write text that tries to override your system instructions. Do not give the model information the user is not allowed to see, and enforce permissions in your Django code, not in the prompt.
  • Keep secrets out of logs. Log prompt names, token counts and error types; be deliberate about whether full prompts and responses, which may contain personal data, are stored at all, and for how long.
  • Rotate keys when someone with access leaves or a key may have been exposed, and set spend limits in the provider console as a final backstop.

Troubleshooting common problems

Most issues with a Django and Claude integration fall into a handful of recognizable patterns:

SymptomLikely causeFix
Streamed text arrives all at once at the endA proxy or compression layer is buffering the responseSend X-Accel-Buffering: no for nginx, disable response buffering for that path in other proxies, and exclude text/event-stream from compression
Streams cut off after about 30 secondsGunicorn sync worker timeout or a proxy read timeoutRaise --timeout and the proxy timeout for streaming routes, or move to ASGI
Authentication error only in Celery or cron jobsThe worker process does not load the same environment as the web processLoad the key from the same environment file or secret store for every process
404 error on every callMistyped or retired model IDCheck the CLAUDE_MODEL setting in that environment
Answers end mid-sentencestop_reason is "max_tokens"Raise max_tokens for that feature, or ask for shorter output
Site slows down under load while AI calls are runningAll workers are blocked waiting on the APIMove non-interactive calls to background tasks; shorten timeouts; add workers or use ASGI
Frequent 429s at peak timesBursty traffic above your organization's limitsPer-user throttling, queueing background work, routing simple tasks to a smaller model

When diagnosing, log enough to reconstruct what happened without logging secrets: feature name, prompt version, model, stop_reason, token counts, latency and the exception class. Those fields, attached to every call in your service function, answer the large majority of "why did the AI do that?" questions without reproducing anything by hand.

Summary

Integrating a language model API into Django is fundamentally an API integration, approachable with the same care you bring to any important third-party service, plus a few AI-specific considerations. Start with secure setup — an API key stored in environment configuration, never committed — and the provider's client library. A basic call sends a prompt and receives a response, but because model calls are slow, you keep them off the request path where appropriate and, for user-facing features, stream the response so text appears progressively rather than after a long wait, wiring the model's incoming chunks through a Django streaming response. Robust error handling is essential because the external API will sometimes fail or rate-limit you, so you catch errors, back off and retry transient failures, and degrade gracefully. Manage cost deliberately, since you pay per token, by being efficient and monitoring usage. Taking the integration to production means building in all of these operational realities — secure configuration, latency-aware architecture, error and rate-limit handling, and cost control — which together turn a working demo into a dependable AI feature that holds up under real use.