Automating AI Workflows with Python APIs

The moment a language model stops being a chat window and starts being a step inside a background job, everything you already know about calling third-party APIs applies again — and a few things that do not apply anywhere else start to bite. Part of the Automating Side-Hustle Operations with APIs guide. This page builds one production-grade AI pipeline: a worker that pulls jobs off a queue, calls a model with a versioned prompt, forces structured output, validates it, retries the failures that deserve retrying, and refuses to spend more than the budget you gave it.

The framing that matters is commercial. A model call is the single most expensive request in your stack — often two to four orders of magnitude more expensive than a database query — and it is also the only one whose cost varies with the length of the input a customer happened to paste in. Treat it like an unmetered dependency and your margin evaporates on the day you get traction. Every pattern below exists to make that cost predictable.

Prerequisites

You need Python 3.11 or newer. The code below uses match statements, tomllib, and the X | None union syntax without a __future__ import, all of which land in 3.11. Install the model SDK, a validation library, a retry library, and a Redis client:

pip install "anthropic>=0.68" "pydantic>=2.7" "tenacity>=8.3" "redis>=5" "structlog>=24.1"

The Anthropic SDK ships its own async HTTP transport built on httpx, so you do not add httpx yourself for model calls. You still want httpx directly for the non-model hops in the same pipeline — pulling the source document, posting the result to a webhook — and the httpx versus requests trade-offs for async work apply to those hops unchanged.

Set configuration in the environment. Never hardcode a model name, a concurrency number, or a spend cap:

ANTHROPIC_API_KEY=sk-ant-...
LLM_MODEL=claude-opus-4-8
LLM_CHEAP_MODEL=claude-haiku-4-5
LLM_MAX_TOKENS=2048
LLM_REQUEST_TIMEOUT_S=90
LLM_CONCURRENCY=8
LLM_MAX_ATTEMPTS=4
PROMPT_FILE=prompts/lead_extract.toml
PROMPT_VERSION=v3
DAILY_BUDGET_USD=25.00
REDIS_URL=redis://localhost:6379/0

The baseline assumption is that this worker runs outside your web process. If you are calling a model inside a request handler, you are holding an HTTP connection open for the length of a generation, which is the wrong shape for anything but streaming — see streaming LLM responses through FastAPI for that case. Everything here targets background work, which is where the volume, and therefore the money, actually is.

Step 1: Build one configured client, not a client per call

The first mistake in almost every AI pipeline is constructing the SDK client inside the function that uses it. Each construction spins up a fresh connection pool, so a batch of fifty jobs opens fifty TLS handshakes to the same host and throws them away. Build one async client at import time, cache it, and give it an explicit timeout.

Turn the SDK's built-in retries off. The SDK retries 429s and 5xx errors twice by default with its own backoff, which silently multiplies your wall-clock timeout by three and hides the failure from your metrics. You want retries in one place, under your control, with your logging attached.

# llm/client.py
import os
from functools import lru_cache

from anthropic import AsyncAnthropic

MODEL = os.getenv("LLM_MODEL", "claude-opus-4-8")
CHEAP_MODEL = os.getenv("LLM_CHEAP_MODEL", "claude-haiku-4-5")
MAX_TOKENS = int(os.getenv("LLM_MAX_TOKENS", "2048"))
REQUEST_TIMEOUT_S = float(os.getenv("LLM_REQUEST_TIMEOUT_S", "90"))


@lru_cache(maxsize=1)
def get_client() -> AsyncAnthropic:
    """One pooled async client per process. API key comes from the environment."""
    return AsyncAnthropic(
        timeout=REQUEST_TIMEOUT_S,   # seconds in the Python SDK
        max_retries=0,               # we own the retry policy, see Step 4
    )

Ninety seconds sounds enormous next to the two-second timeout you would give a REST call. It is correct here. A model generating two thousand tokens of structured output genuinely takes twenty to sixty seconds under load, and a timeout tuned for a normal API produces a pipeline that fails constantly while the model is doing exactly what you asked. Set the timeout from measured p99 latency, not from habit.

Step 2: Force structured output with a schema, never a regex

The single highest-leverage change you can make to an AI pipeline is to stop parsing prose. Ask for JSON in the prompt and roughly 2% of responses will arrive wrapped in a markdown fence, prefixed with "Here is the JSON you requested", or truncated mid-object. At a thousand jobs a day that is twenty broken rows every day, forever.

Constrain the response format at the API level instead. Define the shape once as a Pydantic model, hand the generated JSON Schema to the request, and validate what comes back anyway — belt and braces, because the constraint guarantees shape, not truth. If Pydantic v2 is new to you, validating JSON with Pydantic v2 covers the model syntax in depth.

# llm/extract.py
import os
from typing import Literal

from pydantic import BaseModel, ValidationError

from .client import MODEL, MAX_TOKENS, get_client
from .prompts import load_prompt


class Lead(BaseModel):
    company: str
    contact_email: str
    budget_usd: int | None = None
    intent: Literal["demo", "pricing", "support", "other"]


SCHEMA = Lead.model_json_schema()
SCHEMA["additionalProperties"] = False   # required by the structured-output API


class ExtractionRefused(Exception):
    """Permanent: the model declined. Retrying burns money and never succeeds."""


class ExtractionTruncated(Exception):
    """Recoverable by raising max_tokens, not by retrying identically."""


async def extract_lead(raw: str) -> tuple[Lead, dict]:
    system_text, prompt_digest = load_prompt()
    resp = await get_client().messages.create(
        model=MODEL,
        max_tokens=MAX_TOKENS,
        system=[{
            "type": "text",
            "text": system_text,
            "cache_control": {"type": "ephemeral"},   # cache the stable prefix
        }],
        output_config={"format": {"type": "json_schema", "schema": SCHEMA}},
        messages=[{"role": "user", "content": raw}],
    )

    match resp.stop_reason:
        case "refusal":
            raise ExtractionRefused(str(getattr(resp, "stop_details", None)))
        case "max_tokens":
            raise ExtractionTruncated(f"hit max_tokens={MAX_TOKENS}")

    text = "".join(b.text for b in resp.content if b.type == "text")
    meta = {
        "prompt_digest": prompt_digest,
        "input_tokens": resp.usage.input_tokens,
        "output_tokens": resp.usage.output_tokens,
        "cache_read_tokens": resp.usage.cache_read_input_tokens,
    }
    return Lead.model_validate_json(text), meta

Two details earn their keep. The match on stop_reason separates the three ways a call can end badly, and each one wants a different response: a refusal is permanent, a truncation is a configuration problem, and everything else is a transient you can retry. And cache_control on the system block tells the API to cache that prefix, which is where most of the savings in Step 6 come from.

Step 3: Batch with a semaphore, not with a thread pool

Model calls are almost pure network wait. One job spends fifty seconds doing nothing but holding a socket open. That makes async the obvious concurrency model — you can hold sixteen of those sockets in a single process with essentially no CPU cost, where sixteen threads would cost you sixteen stacks and a GIL argument.

The rule is: fan out with asyncio.gather, but gate the fan-out with a semaphore. Unbounded gather over a thousand queue items sends a thousand simultaneous requests, trips the provider's rate limit within a second, and turns your whole batch into a retry storm. The semaphore is what converts "as fast as possible" into "as fast as allowed".

# llm/batch.py
import asyncio
import os
from dataclasses import dataclass

from pydantic import ValidationError

from .extract import ExtractionRefused, Lead, extract_lead

CONCURRENCY = int(os.getenv("LLM_CONCURRENCY", "8"))


@dataclass(slots=True)
class JobResult:
    job_id: str
    lead: Lead | None
    error: str | None
    meta: dict


async def _run_one(sem: asyncio.Semaphore, job_id: str, raw: str) -> JobResult:
    async with sem:
        lead, meta = await extract_lead(raw)
        return JobResult(job_id=job_id, lead=lead, error=None, meta=meta)


async def run_batch(jobs: dict[str, str]) -> list[JobResult]:
    sem = asyncio.Semaphore(CONCURRENCY)
    tasks = [_run_one(sem, jid, raw) for jid, raw in jobs.items()]
    settled = await asyncio.gather(*tasks, return_exceptions=True)

    out: list[JobResult] = []
    for job_id, result in zip(jobs, settled, strict=True):
        match result:
            case JobResult():
                out.append(result)
            case ExtractionRefused() as exc:
                out.append(JobResult(job_id, None, f"refused: {exc}", {}))
            case ValidationError() as exc:
                out.append(JobResult(job_id, None, f"invalid: {exc.error_count()}", {}))
            case BaseException() as exc:
                out.append(JobResult(job_id, None, f"{type(exc).__name__}: {exc}", {}))
    return out

return_exceptions=True is what makes this batch survivable: one poisoned document does not cancel the other nine hundred and ninety-nine in flight. The match statement then sorts outcomes into buckets you can act on — refusals go to a human, validation failures go to a repair pass, everything else goes to the retry policy.

Pick the concurrency number from your provider's tokens-per-minute limit, not from your CPU count. If your account allows 400,000 input tokens per minute and each job sends 6,000, you can sustain roughly 66 jobs per minute; at 50 seconds per job that is about 55 concurrent calls before you saturate. Start at a quarter of your computed ceiling and raise it while watching your 429 rate. The mechanics of reading those headers are covered in debugging 429 Too Many Requests errors.

Step 4: Retry the transients, kill the permanents

Retry logic around a model call is different from retry logic around a normal REST call in one crucial way: a failed attempt may still have cost you money. A request that times out after fifty seconds of generation is billed for the tokens produced before you hung up. So the question is not "is this error retryable" but "is retrying this cheaper than failing".

Three classes of outcome exist. Rate limits, connection errors, and 5xx responses are transient and worth retrying with exponential backoff and jitter. A 400 is a bug in your request and will fail identically forever. A refusal is a policy decision and will also fail identically forever, but unlike a 400 it costs you tokens each time you ask. Retrying refusals is the most expensive no-op in the business.

# llm/retry.py
import os
import random

import anthropic
from tenacity import (
    AsyncRetrying,
    RetryError,
    retry_if_exception_type,
    stop_after_attempt,
    wait_exponential_jitter,
)

MAX_ATTEMPTS = int(os.getenv("LLM_MAX_ATTEMPTS", "4"))

TRANSIENT = (
    anthropic.RateLimitError,        # 429
    anthropic.InternalServerError,   # 500 and 529 overloaded
    anthropic.APIConnectionError,    # DNS, TLS, read timeout
)


def _retry_after_seconds(exc: BaseException) -> float | None:
    response = getattr(exc, "response", None)
    header = getattr(response, "headers", {}).get("retry-after") if response else None
    try:
        return float(header) if header else None
    except ValueError:
        return None


async def with_retry(coro_factory):
    """Run coro_factory() with backoff. Non-transient errors propagate at once."""
    async for attempt in AsyncRetrying(
        retry=retry_if_exception_type(TRANSIENT),
        stop=stop_after_attempt(MAX_ATTEMPTS),
        wait=wait_exponential_jitter(initial=1, max=30, jitter=2),
        reraise=True,
    ):
        with attempt:
            return await coro_factory()

Wrap the extraction call with with_retry(lambda: extract_lead(raw)) and the policy applies to every job without touching the extraction code. The jitter matters more than the backoff: without it, sixteen workers that all hit a rate limit at the same instant will all wake up at the same instant and hit it again. The general pattern, including the decorator form, is covered in retrying failed HTTP requests with tenacity.

Step 5: Version prompts like code, because they are code

A prompt is the most volatile piece of logic in the pipeline and the only one most teams edit without review. Change one sentence and your extraction accuracy moves three points in a direction nobody measures until a customer complains. Worse, an inline prompt string interpolated with datetime.now() or a job ID silently destroys prompt caching, because caching is a prefix match and any byte change invalidates everything after it.

Store prompts in a TOML file, keyed by version, loaded through tomllib. Hash the text and attach the digest to every log line and every stored row. Then rolling back a bad prompt is one environment variable, not a deploy.

# prompts/lead_extract.toml
[v2]
released = "2026-06-02"
system = """
You extract sales-lead fields from inbound email text.
Return only fields you can support with a direct quote from the input.
"""

[v3]
released = "2026-07-14"
system = """
You extract sales-lead fields from inbound email text.
Return only fields you can support with a direct quote from the input.
If a budget figure appears as a range, record the lower bound.
Never infer an intent that is not stated; use "other" when unsure.
"""

# llm/prompts.py
import hashlib
import os
import tomllib
from functools import lru_cache
from pathlib import Path

PROMPT_FILE = Path(os.getenv("PROMPT_FILE", "prompts/lead_extract.toml"))
ACTIVE_VERSION = os.getenv("PROMPT_VERSION", "v3")


@lru_cache(maxsize=8)
def load_prompt(version: str | None = None) -> tuple[str, str]:
    """Return (system_text, short_sha256) for a prompt version."""
    version = version or ACTIVE_VERSION
    data = tomllib.loads(PROMPT_FILE.read_text(encoding="utf-8"))
    if version not in data:
        raise KeyError(f"prompt version {version!r} not in {PROMPT_FILE}")
    text = data[version]["system"].strip()
    digest = hashlib.sha256(text.encode("utf-8")).hexdigest()[:12]
    return text, f"{version}:{digest}"

Storing prompt_digest alongside every extracted row is what makes the pipeline auditable. When accuracy drops, you can group your quality metric by digest and see exactly which edit did it. Without that column you are guessing, and guessing about a prompt change three weeks after the fact is an afternoon you never get back.

Step 6: Queue with a budget, and route the patient work to batch

Now the commercial part. Two levers separate a pipeline that costs $47 per thousand jobs from one that costs $5, and neither of them requires a worse model.

The first lever is prompt caching. Your system prompt, schema, and few-shot examples are identical on every job; only the user document changes. Marking the stable prefix with cache_control bills those tokens at roughly a tenth of the input rate on cache hits. The prefix must clear a minimum size to cache at all — 4,096 tokens on the larger models — so short system prompts silently do not cache. Check usage.cache_read_input_tokens on a real response before you believe it works.

The second lever is asynchronous batch submission. Work that does not need an answer this second — nightly enrichment, backfills, a weekly report — can go through the Batch API at half price. Combine both and you are at a quarter of the naive bill.

# llm/budget.py
import os
from datetime import date

import redis.asyncio as redis

from .client import CHEAP_MODEL, MODEL, get_client

# US dollars per million tokens: (input, output)
PRICES = {
    "claude-opus-4-8": (5.00, 25.00),
    "claude-haiku-4-5": (1.00, 5.00),
}
DAILY_BUDGET_USD = float(os.getenv("DAILY_BUDGET_USD", "25.00"))


class BudgetExceeded(Exception):
    """Stop the queue rather than discover the overspend on the invoice."""


async def estimate_usd(system: list, messages: list, model: str) -> float:
    counted = await get_client().messages.count_tokens(
        model=model, system=system, messages=messages
    )
    in_rate, out_rate = PRICES[model]
    projected_out = int(os.getenv("LLM_MAX_TOKENS", "2048")) * 0.6
    return (counted.input_tokens * in_rate + projected_out * out_rate) / 1_000_000


async def reserve(pool: redis.Redis, amount_usd: float) -> None:
    key = f"llm:spend:{date.today().isoformat()}"
    spent = await pool.incrbyfloat(key, amount_usd)
    await pool.expire(key, 172_800)
    if spent > DAILY_BUDGET_USD:
        await pool.incrbyfloat(key, -amount_usd)
        raise BudgetExceeded(f"{spent:.2f} would exceed {DAILY_BUDGET_USD:.2f}")


def choose_model(input_tokens: int, needs_reasoning: bool) -> str:
    """Route cheap, mechanical work to the small model. Measure before trusting it."""
    return MODEL if needs_reasoning or input_tokens > 20_000 else CHEAP_MODEL

Counting tokens before you send is the whole game. count_tokens is a cheap call that tells you what a job will cost before you commit to it, which means the queue can reject a 400,000-token document rather than silently spending four dollars on one customer's PDF dump. Pair the reservation counter with the same Redis instance you use for caching API responses with Redis — the budget ledger is a few keys, not a new dependency.

For the patient half of the workload, submit a batch and poll:

# llm/offline.py
import asyncio
import os

from .client import MODEL, MAX_TOKENS, get_client
from .prompts import load_prompt

POLL_SECONDS = int(os.getenv("BATCH_POLL_SECONDS", "60"))


async def submit_offline(jobs: dict[str, str]) -> str:
    system_text, _ = load_prompt()
    batch = await get_client().messages.batches.create(
        requests=[
            {
                "custom_id": job_id,
                "params": {
                    "model": MODEL,
                    "max_tokens": MAX_TOKENS,
                    "system": [{"type": "text", "text": system_text}],
                    "messages": [{"role": "user", "content": raw}],
                },
            }
            for job_id, raw in jobs.items()
        ]
    )
    return batch.id


async def collect_offline(batch_id: str) -> dict[str, str]:
    client = get_client()
    while True:
        batch = await client.messages.batches.retrieve(batch_id)
        if batch.processing_status == "ended":
            break
        await asyncio.sleep(POLL_SECONDS)

    out: dict[str, str] = {}
    async for entry in await client.messages.batches.results(batch_id):
        if entry.result.type == "succeeded":
            blocks = entry.result.message.content
            out[entry.custom_id] = "".join(b.text for b in blocks if b.type == "text")
    return out

Results come back in arbitrary order, so key them by custom_id and never by position. A batch usually finishes inside an hour and is guaranteed within twenty-four, which makes it perfect for anything driven by scheduled data pipelines and useless for anything a human is waiting on.

Configuration reference

Two of these deserve emphasis. LLM_MAX_TOKENS is not a safety limit — it is a truncation point, and setting it too low produces invalid JSON rather than a friendly error. DAILY_BUDGET_USD is the only line of defence between a runaway retry loop and a four-figure invoice, so it belongs in the environment of every deployment, including staging.

Gotchas and failure modes

A synchronous client inside an async worker. Symptom: you set LLM_CONCURRENCY=16 and throughput stays flat at one job at a time, with CPU near zero. Cause: someone imported the synchronous Anthropic client, so every call blocks the event loop for fifty seconds while fifteen other coroutines wait their turn. Fix: use AsyncAnthropic everywhere and grep the codebase for the sync class. If a legacy sync call must stay, wrap it in asyncio.to_thread.

Truncated JSON masquerading as a model quality problem. Symptom: ValidationError spikes on long input documents while short ones extract perfectly. Cause: the response hit max_tokens mid-object; the schema constraint guarantees the format of what was emitted, not that emission finished. Fix: check stop_reason == "max_tokens" explicitly before parsing, as Step 2 does, and raise LLM_MAX_TOKENS. Retrying the identical request just truncates again.

Prompt caching that never hits. Symptom: usage.cache_read_input_tokens is zero on every response and your bill matches the uncached estimate exactly. Cause: something volatile sits in the prefix — a timestamp, a job ID, a customer name interpolated into the system prompt — or the prefix is below the model's minimum cacheable size. Fix: freeze the system block, move all per-job context into the user message, and verify with a real response rather than assuming.

Retrying refusals and 400s. Symptom: a small number of jobs each consume four attempts and appear in the dead-letter queue anyway, and your cost per successful job is 15% higher than the arithmetic says it should be. Cause: a blanket except Exception around the retry wrapper. Fix: retry_if_exception_type(TRANSIENT) with an explicit tuple, as in Step 4. A refusal is an answer, not an error.

Thread-unsafe budget accounting. Symptom: the daily spend counter drifts below reality and the cap never triggers. Cause: read-modify-write against Redis from multiple workers. Fix: INCRBYFLOAT is atomic — reserve first, refund on failure, never GET then SET. The same reasoning behind building an idempotent webhook receiver applies to spend: assume every operation runs twice.

Verification

Prove three things before you call this done: that the schema constraint holds, that the budget gate fires, and that a real call produces the token counts you expect.

Start with a live smoke test that prints exactly what you are being billed for:

# scripts/smoke.py
import asyncio

from llm.extract import extract_lead

SAMPLE = "Hi - we're Northwind Ltd, budget is 4-6k, can we see a demo? tom@northwind.example"


async def main() -> None:
    lead, meta = await extract_lead(SAMPLE)
    print(lead.model_dump_json(indent=2))
    print(meta)


asyncio.run(main())

Run it with python -m scripts.smoke. Expected output is a valid Lead object plus a metadata dict whose cache_read_tokens is zero on the first run and non-zero on the second — that second run is the only proof that caching is working:

{"company": "Northwind Ltd", "contact_email": "tom@northwind.example",
 "budget_usd": 4000, "intent": "demo"}
{'prompt_digest': 'v3:e30f57a91c4d', 'input_tokens': 118,
 'output_tokens': 74, 'cache_read_tokens': 5211}

Then lock the budget gate down in a test that never touches the network. Mocking the model client rather than the HTTP layer keeps the test fast; the alternative approach, intercepting requests with respx, is better when you want to assert on the wire format itself:

# tests/test_budget.py
import pytest
import redis.asyncio as redis

from llm.budget import BudgetExceeded, reserve


@pytest.mark.asyncio
async def test_reserve_blocks_over_budget(monkeypatch):
    monkeypatch.setenv("DAILY_BUDGET_USD", "1.00")
    pool = redis.from_url("redis://localhost:6379/15", decode_responses=True)
    await pool.flushdb()

    await reserve(pool, 0.90)
    with pytest.raises(BudgetExceeded):
        await reserve(pool, 0.50)

    key_count = await pool.dbsize()
    assert key_count == 1          # the refund left no orphan keys

Finally, emit one structured log line per job carrying prompt_digest, input_tokens, output_tokens, cache_read_tokens, attempt, and usd. That single line is your entire AI observability story; structured logging with structlog covers wiring it into a JSON handler.

Cost and performance at scale

Here is the arithmetic that decides whether this pipeline is a business or a hobby. Take a realistic extraction job: a 5,000-token stable prefix (system prompt, schema, examples), a 1,000-token customer document, and 700 tokens of structured output. On a frontier model at $5 per million input tokens and $25 per million output tokens, the naive version costs 6,000 × $5 ÷ 1M for input plus 700 × $25 ÷ 1M for output — three cents plus 1.75 cents, or $47.50 per thousand jobs. Add cache_control to the prefix and those 5,000 shared tokens bill at roughly a tenth of the input rate, dropping input cost from $0.030 to $0.0075 and the total to $25.00 per thousand. Push the patient half through the Batch API at 50% off and it lands at $12.50 per thousand. Route the mechanical jobs to a small model at $1 and $5 per million and you are at $5.00 per thousand — a tenth of where you started, with the same code and the same prompt. Throughput follows the same shape: at eight concurrent workers and fifty seconds per call, one process sustains roughly 576 jobs per hour, so a single small container clears a 10,000-job backlog overnight. The number to instrument is cost per successful job, not cost per call — retries, refusals, and validation failures all inflate the denominator, and the difference between the two is usually 10 to 20%. The method for turning that into a price is in calculating cost per API request.

FAQ

What does an AI extraction pipeline actually cost to run at 10,000 jobs a month? Using the worked example above — a shared 5,000-token prefix, a 1,000-token document, and 700 tokens out — you land at roughly $250 a month on a frontier model with prompt caching, $125 if the work can go through the batch endpoint, and about $50 if the jobs are mechanical enough for a small model. Compute is noise next to that: a single small container handles the whole volume. Budget 15% on top for retries and dead-lettered jobs, and instrument cost per successful job so you notice when that overhead grows.

Should I resell this as a per-seat feature or meter it per job? Meter it, at least until your usage distribution stops surprising you. AI cost scales with input length, which scales with how much your heaviest customer pastes in, and a seat price averages that away in exactly the wrong direction — your worst-margin customer is also your happiest one. Start usage-based, cap the free tier by token volume rather than job count, and revisit once you have three months of data. The usage-based versus seat-based pricing trade-offs go deeper.

How risky is it to switch model providers later? Lower than it feels, if you keep three seams clean: the client construction, the prompt registry, and the schema. Those are the only three places provider specifics appear in the code above; the queueing, retry, budget, and validation layers are provider-agnostic. What does not port is calibration — your prompt version, your max_tokens value, and your quality baseline all need re-measuring, which is a week of eval work rather than a rewrite. Keep an eval set of a few hundred labelled examples and the migration becomes a number rather than an argument.

Do I need a job queue, or can cron and a script carry this? A script driven by cron is genuinely fine up to a few thousand jobs a day if the work is idempotent and a missed run is survivable. You need a real queue when you need per-job retry state, back-pressure, or partial progress across restarts — all three arrive together the first time a batch dies halfway. Running background jobs with Celery and the Celery versus RQ versus arq comparison cover the choice; arq is the natural fit for an async-native pipeline like this one.

How do I stop one customer's huge document from blowing the daily budget? Count tokens before you send, not after. count_tokens costs almost nothing and gives you the input size before you commit, so the queue can reject or downgrade an oversized job instead of discovering it on the invoice. Combine that with a per-customer daily ceiling in the same Redis ledger as the global cap, and the worst case becomes one rejected job with a clear error rather than an unbounded spend. Controlling LLM API costs in production covers the per-tenant accounting in detail.

Related

• Controlling LLM API Costs in Production — per-tenant ledgers, token budgets, and the alerting that catches an overspend on day one rather than day thirty.
• Streaming LLM Responses Through FastAPI — the interactive half of this problem, for when a human is waiting on the tokens.
• Running Background Jobs with Celery — the queue layer this pipeline sits on once cron stops being enough.
• Retrying Failed HTTP Requests with tenacity — the general retry patterns behind Step 4, including decorator and context-manager forms.
• Building Zapier Alternatives with Python — where an AI step becomes one node in a larger automation graph you own end to end.