DEV Community

Kaushik N
Kaushik N

Posted on

Letting an LLM write to your database safely: previews, idempotency keys and ETags in a small FastAPI API

#api #python #fastapi #mcp

I built NutriScan, a PWA that reads nutrition labels from a photo and tracks macros. Then I noticed I was telling Claude about my meals all day instead of logging them in my own app. So I gave the app an API, and a Model Context Protocol (MCP) connector so a claude.ai chat can read my day and log meals.

The hard part was not wiring Claude up. It was deciding what the server must guarantee regardless of what the model does. This post walks through those guarantees with the real code (FastAPI, Pydantic v2, psycopg2 on Neon Postgres with row-level security, Supabase for auth). The whole thing is open source: MrTig-afk/NutritionalTracker, mostly backend/api_v1.py.

TL;DR

  • Every write can run as a preview: the real code path inside a transaction that is always rolled back.
  • Every real POST needs an Idempotency-Key; a repeat replays the stored response for 24 h.
  • Every PATCH/DELETE needs If-Match with a content-hash ETag (and Cloudflare will quietly weaken it).
  • Errors are RFC 9457 application/problem+json with a stable error_type.
  • The MCP connector never lets the model save directly: write tools return a preview plus a one-time code, and only confirm_change commits, using that code as the idempotency key.
  • A test fails if the published OpenAPI file drifts from the code.

Status: the API and connector run on my own account while I test them; opening them up is the next phase.

The surface

17 endpoints, 8 reads and 9 writes:

Method Path
GET /v1/me, /v1/context, /v1/days, /v1/goals, /v1/library
GET /v1/entries/{log_id}, /v1/meals/{group_id}, /v1/templates/{template_id}
POST /v1/batch, /v1/meals, /v1/entries, /v1/templates, /v1/templates/{template_id}/log
PATCH /v1/entries/{log_id}, /v1/templates/{template_id}
DELETE /v1/entries/{log_id}, /v1/meals/{group_id}

And 13 MCP tools: three reads (get_context, get_template, search_library), eight write tools that only preview (log_meal, log_template, edit_entry, delete_entry, delete_meal, save_template, update_template, save_to_library), confirm_change, and stop_asking_before_saving.

1. One error shape

Before any feature, I wanted every failure to look the same. One exception type, rendered as RFC 9457 problem details, with a stable machine-readable error_type and optional extension fields:

class Problem(Exception):
    def __init__(self, status: int, error_type: str, detail: str, headers: Optional[dict] = None, **extra):
        self.status, self.error_type, self.detail = status, error_type, detail
        self.headers, self.extra = headers or {}, extra

def problem_response(p: Problem) -> JSONResponse:
    body = {"type": f"urn:nutriscan:error:{p.error_type}", "title": HTTPStatus(p.status).phrase,
            "status": p.status, "detail": p.detail, "error_type": p.error_type, **p.extra}
    return JSONResponse(jsonable_encoder(body), status_code=p.status, headers=p.headers,
                        media_type="application/problem+json")
Enter fullscreen mode Exit fullscreen mode

Everything below raises Problem. A rate-limited client gets limit, used and resets_at as extension fields; an idempotency clash gets in_progress. Clients branch on error_type, humans read detail.

2. Distrust the request before you parse it

A middleware on /v1 checks size and type before a body is ever handed to Pydantic:

if request.method in ("POST", "PATCH", "PUT", "DELETE"):
    declared = request.headers.get("content-length")
    if declared is None and request.headers.get("transfer-encoding"):
        # a chunked body would be read whole before its size is known
        raise Problem(411, "length_required", "Send a Content-Length header.")
    if declared and declared.isdigit() and int(declared) > m.APP_BODY_CAP:
        raise Problem(413, "payload_too_large", "Body over 64 KB.")
    body = await request.body()
    if len(body) > m.APP_BODY_CAP:
        raise Problem(413, "payload_too_large", "Body over 64 KB.")
    ctype = request.headers.get("content-type", "").split(";")[0].strip().lower()
    if body and ctype != "application/json":
        raise Problem(415, "unsupported_media_type", "Send JSON with Content-Type: application/json.")
Enter fullscreen mode Exit fullscreen mode

Note the declared length is checked and the real length, because a header can lie.

Then the models are strict. Unknown fields are an error, not silently dropped, and every number and string is bounded:

Name = Annotated[str, Field(min_length=1, max_length=80), AfterValidator(_text)]
Servings = Annotated[float, Field(gt=0, le=100)]
LogDate = Annotated[str, Field(pattern=r"^\d{4}-\d{2}-\d{2}$"), AfterValidator(_log_date)]

class Strict(BaseModel):
    model_config = ConfigDict(extra="forbid")

class Macros(Strict):
    calories: float = Field(ge=0, le=MAX_KCAL)
    protein_g: float = Field(ge=0, le=1000)
    # ...
    sodium_mg: Optional[float] = Field(None, ge=0, le=100000)

class Item(Strict):
    name: Name
    portion: Optional[Portion] = None
    servings: Servings = 1
    macros: Macros
    save_to_library: bool = False
Enter fullscreen mode Exit fullscreen mode

_text strips control characters and _log_date refuses dates more than a day in the future. extra="forbid" matters more than it looks when the client is an LLM: a hallucinated field like "meal_type": "brunch" fails loudly instead of being ignored while the model believes it was saved.

3. Tokens that never wake the database for a guess

Personal tokens look like nsk_live_ plus 43 URL-safe characters (secrets.token_urlsafe(32)). They are shown once and only their sha256 is stored:

TOKEN_PREFIX = "nsk_live_"
_TOKEN_RE = re.compile(r"^nsk_live_[A-Za-z0-9_-]{43}$")
MAX_TOKENS = 10
EXPIRY_DAYS = {"30d": 30, "90d": 90, "1y": 365, "never": None}

def token_hash(token: str) -> str:
    return hashlib.sha256(token.encode()).hexdigest()
Enter fullscreen mode Exit fullscreen mode

The database is serverless and bills for time awake, so a bot spraying random tokens should not be able to keep it up. At startup the server loads the first 12 characters of every live token into a set, and a token whose prefix is not in it is refused without a query:

if token.startswith(TOKEN_PREFIX):
    known = _live_prefixes is None or token[:12] in _live_prefixes
    row = _TOKEN_RE.match(token) and known and _lookup_token(token_hash(token), request.headers.get("user-agent"))
Enter fullscreen mode Exit fullscreen mode

Valid lookups are cached for 300 seconds. And guessing is punished: the 20th bad token from one IP inside five minutes blocks that IP for ten minutes.

if not row:
    ip = m._client_ip(request)
    if m._spike(f"patfail:{ip}", 20, 300):
        m._blocked[ip] = time.time() + 600
    raise Problem(401, "unauthorized", "Invalid token.")
Enter fullscreen mode Exit fullscreen mode

Scopes are a FastAPI dependency per route, and write routes re-check scopes per change inside a batch:

def need(*scopes: str, write: bool = False):
    def dep(request: Request) -> Caller:
        budget_gate()
        caller = resolve_caller(request)
        _require(caller, *scopes)
        request.state.ratelimit = take_request(caller, write)
        # ...
Enter fullscreen mode Exit fullscreen mode

4. Rate limits a client can plan around

Limits are per user across all their tokens: 20 requests a minute, 10 writes a minute, 200 a day, with the day resetting at Melbourne midnight (where the app's users are). Every answer carries RateLimit-* headers, and a 429 says exactly when to come back:

BURST_PER_MIN, WRITES_PER_MIN, DAILY_CAP = 20, 10, 200
MELBOURNE = ZoneInfo("Australia/Melbourne")
# ...
retry = max(1, int(resets.timestamp() - now) + 1)
raise Problem(429, "rate_limited", f"Over the limit of {limit}. Try again after {_zulu(resets)}.",
              headers={"Retry-After": str(retry), "RateLimit-Limit": str(limit), "RateLimit-Remaining": "0"},
              limit=limit, used=used, resets_at=_zulu(resets))
Enter fullscreen mode Exit fullscreen mode

Honest wrinkle: on a successful response RateLimit-Limit always reports the daily cap, not whichever per-minute window is closest. The 429 is precise; the happy-path headers are a simplification.

5. Previews by rollback

This is the core idea. A preview should not estimate what a write would do; it should do it and throw the result away. So every write handler runs inside one database helper, and preview just turns off the commit:

@contextmanager
def db(user_id: Optional[str], commit: bool = True):
    conn = None
    try:
        conn = m.get_db(user_id)
        cur = conn.cursor()
        yield cur
        if commit:
            conn.commit()
        cur.close()
    # ...
Enter fullscreen mode Exit fullscreen mode 
with db(caller.user_id, commit=not preview) as cur:
    ...  # the exact same writes, triggers, RLS policies and validation
Enter fullscreen mode Exit fullscreen mode

When the connection goes back to the pool, release_db always calls conn.rollback() (it also resets the transaction-local app.user_id setting that the RLS policies read), so an uncommitted preview leaves nothing behind.

Because it is the same code path, the preview includes things an estimate would miss: database defaults, trigger side effects, constraint violations. Any /v1 write accepts ?preview=true.

To be precise about what is tested: the tests assert that a preview produces zero commits and the real call exactly one. There is no test yet that diffs the preview body against the committed body; the equivalence comes from sharing the code path, not from an assertion.

6. Idempotency keys

A real POST must carry an Idempotency-Key header. The key, a hash of the request, and eventually the response are stored in the same transaction as the write:

if not preview and request.method == "POST":
    key = (idempotency_key or "").strip()
    if not key or len(key) > 255:
        raise Problem(422, "validation_error", "Idempotency-Key header required (1-255 characters) on a real write.")
rhash = _request_hash(request.url.path, payload)
with db(caller.user_id, commit=not preview) as cur:
    if key:
        cur.execute("DELETE FROM api_idempotency WHERE user_id = %s AND key = %s "
                    "AND created_at < now() - interval '24 hours'", ...)
        cur.execute("""INSERT INTO api_idempotency (user_id, key, request_hash, status_code, response)
                       VALUES (%s, %s, %s, 0, '{}') ON CONFLICT (user_id, key) DO NOTHING RETURNING 1""", ...)
        if not cur.fetchone():
            old = ...  # the stored row
            if old[0] != rhash:
                raise Problem(409, "idempotency_conflict", "This Idempotency-Key was used for a different request.")
            if not old[1]:
                raise Problem(409, "idempotency_conflict", "A request with this Idempotency-Key is still running.",
                              in_progress=True)
            replay = (old[1], old[2])
Enter fullscreen mode Exit fullscreen mode

Three cases fall out of the INSERT ... ON CONFLICT DO NOTHING RETURNING 1:

  • New key: the placeholder row is claimed, the write runs, the response is stored.
  • Same key, same request: the stored status and body are replayed.
  • Same key, different request (the hash is sha256 of path plus payload with sorted keys): 409, because silently replaying a different request is worse than failing.

A status of 0 marks "claimed but not finished", which is how a concurrent duplicate gets in_progress: true instead of a second insert. The test that pins it:

def test_same_key_twice_gives_one_set_of_rows(self):
    body = {"date": TODAY, "name": "Apple", "macros": MACROS}
    first = self.post("/v1/entries", body)
    second = self.post("/v1/entries", body)
    self.assertEqual(second.json()["entry"]["log_id"], first.json()["entry"]["log_id"])
    self.assertEqual(len(self.statements("INSERT INTO daily_log")), 1)
Enter fullscreen mode Exit fullscreen mode

7. ETags, If-Match, and the proxy that rewrote them

Edits and deletes use optimistic concurrency. The ETag is a hash of the content itself, so no version column is needed:

def etag_of(*parts) -> str:
    return '"' + hashlib.sha256(json.dumps(parts, sort_keys=True, default=str).encode()).hexdigest()[:16] + '"'
Enter fullscreen mode Exit fullscreen mode 
def update_entry(self, c: UpdateEntryChange):
    log_id, name, servings, nutrition, d = self.entry(c.log_id)
    n = log_service.load_nutrition(nutrition)
    if etag_of(name, servings, n, str(d)) != c.if_match:
        raise Problem(412, "precondition_failed", "The entry changed since you read it. Read it again.")
Enter fullscreen mode Exit fullscreen mode

This worked perfectly on my machine and failed on every edit in production. The backend sits behind Cloudflare, and when Cloudflare compresses a response it weakens the ETag to W/"...". Clients faithfully echoed back the weak form, which never matched. The fix:

def _if_match(value: Optional[str]) -> str:
    if not value:
        raise Problem(422, "validation_error", "If-Match header required: send the ETag you read.",
                      errors=[{"field": "If-Match", "message": "required"}])
    # Cloudflare weakens ETag to W/"..." when it compresses a response
    return value.strip().removeprefix("W/")
Enter fullscreen mode Exit fullscreen mode

Strictly, RFC 9110 says If-Match uses strong comparison, so a weak tag should never match. Here the ETag is a hash of content I generate myself, and the only thing that weakens it is my own proxy, so accepting the stripped form is safe. It is a deliberate deviation, commented where it lives. (A missing If-Match returns 422; 428 Precondition Required would arguably be the more precise status.)

8. An API that switches itself off before the bill

The database runs on a free monthly compute allowance. The app estimates its own usage, and at 90% the API pauses while the app itself keeps working:

def budget_gate():
    """At 90% of the estimated free Neon month, /v1 pauses; the app keeps working."""
    if m.budget_used() >= 0.9:
        resume = m.neon_next_period_start(datetime.now(timezone.utc).date())
        retry = int(datetime(resume.year, resume.month, resume.day, tzinfo=timezone.utc).timestamp() - time.time())
        raise Problem(503, "api_paused_budget", f"The API is paused to protect the database budget until {resume.isoformat()}.",
                      headers={"Retry-After": str(max(retry, 1))}, resets_at=f"{resume.isoformat()}T00:00:00Z")
Enter fullscreen mode Exit fullscreen mode

On /v1 it runs in the need() dependency before anything else. On /mcp it only gates tools/call, so initialize and tools/list still answer and Claude sees a tool error rather than a dead server.

9. The MCP connector: the model never saves directly

Auth

The connector uses Supabase's OAuth 2.1 server. Claude discovers it from the protected-resource metadata, and every unauthenticated call gets a challenge pointing there, which is what makes claude.ai (re)start the OAuth flow:

@mcp_router.get("/.well-known/oauth-protected-resource/mcp")
@mcp_router.get("/.well-known/oauth-protected-resource")
def protected_resource() -> dict:
    return {"resource": MCP_URL, "authorization_servers": [mcp_issuer()], "scopes_supported": ["email"],
            "bearer_methods_supported": ["header"], "resource_name": "NutriScan"}

def _mcp_challenge() -> Problem:
    return Problem(401, "unauthorized", "Connect through claude.ai to use NutriScan.",
                   headers={"WWW-Authenticate": f'Bearer resource_metadata="{MCP_PRM_URL}"'})
Enter fullscreen mode Exit fullscreen mode

A token is accepted only if it has a client_id, the right issuer and the authenticated audience. A personal token or a normal app login gets the challenge, not access.

One subtle ordering bug I hit: if /mcp declared a typed JSON body, FastAPI would validate it first and a tokenless probe would get a 422 instead of the 401 challenge, so discovery never started. So the route reads the raw body and parses it only after the origin allowlist and auth:

@mcp_router.api_route("/mcp", methods=["GET", "POST", "DELETE"])
async def mcp(request: Request):
    # parsed only after the origin and auth checks: a declared JSON body would 422 before them
    raw = await request.body()
    return await run_in_threadpool(_mcp, request, raw)

def _mcp(request: Request, raw: bytes):
    origin = request.headers.get("origin")
    if origin and origin not in MCP_ORIGINS:   # ("https://claude.ai", "https://claude.com")
        raise Problem(403, "forbidden_origin", "This origin may not call /mcp.")
    caller, client_id = mcp_caller(request)
    # ... only now: payload = json.loads(raw)
Enter fullscreen mode Exit fullscreen mode

Preview, one-time code, confirm

Every write tool calls the same run_changes as /v1, with preview=True. The result goes back to the model with a one-time code:

PENDING_TTL, PENDING_MAX = 600, 20

_, body = run_changes(request, caller, [change], True, None)
code, now = secrets.token_urlsafe(16), time.time()
with _pending_lock:
    for k in [k for k, v in _pending.items() if now - v[3] > PENDING_TTL]:
        del _pending[k]
    mine = sorted((v[3], k) for k, v in _pending.items() if v[0] == caller.user_id)
    for _, k in mine[:max(0, len(mine) - PENDING_MAX + 1)]:
        del _pending[k]
    _pending[code] = (caller.user_id, client_id, [change], now, guard)
return json.dumps({"preview": jsonable_encoder(body), "confirm_code": code,
                   "expires_in_minutes": 10, "next": NEXT_STEP}, default=str)
Enter fullscreen mode Exit fullscreen mode

The model shows the preview to the user. Only confirm_change writes, and it reuses the code as the idempotency key, so a retried confirm replays instead of inserting twice:

entry = _pending.get(code) if isinstance(code, str) else None
if (not entry or entry[0] != caller.user_id or entry[1] != client_id
        or time.time() - entry[3] > PENDING_TTL):
    raise Problem(422, "confirmation_invalid", "This confirmation expired or is not valid. Ask again.")
if entry[4] and _template_etag(caller, entry[2][0].template_id) != entry[4]:
    raise Problem(412, "precondition_failed", "This changed since the preview. Ask again.")
_, body = run_changes(request, caller, entry[2], False, code)   # the code is the idempotency key
Enter fullscreen mode Exit fullscreen mode

The code is bound to the user and the OAuth client, expires in ten minutes, and at most 20 are pending per user (oldest evicted). The guard covers the one change type that carries no If-Match (logging a template), so editing the template between preview and confirm is caught too.

def test_retry_replays_instead_of_saving_twice(self):
    code = self.preview()
    self.conn.executed.clear()   # the preview ran (and rolled back) its own INSERT
    first = self.confirm(code)
    second = self.confirm(code)   # e.g. the first answer was lost and claude.ai retried
    self.assertEqual(json.loads(second[1]), json.loads(first[1]))
    self.assertEqual(len(self.statements("INSERT INTO daily_log")), 1)
Enter fullscreen mode Exit fullscreen mode

Tool annotations

Since a preview cannot save anything, there is no reason for the client to ask permission before one. Only the tool that actually commits is marked destructive, so the user gets one prompt, for the save:

def _annotations(name: str) -> dict:
    saves = name == "confirm_change"
    return {"readOnlyHint": not saves, "destructiveHint": saves, "idempotentHint": False, "openWorldHint": False}
Enter fullscreen mode Exit fullscreen mode

(stop_asking_before_saving, which turns off one setting, is hand-annotated as not read-only but not destructive.)

The rule behind all of this: the guarantees live in the server, not in the prompt. Tool descriptions tell Claude to show the preview and wait, and it does. But if it did not, a write without a valid code is still not a write.

10. Documentation that cannot drift

The OpenAPI document is generated from the code and also committed as openapi-v1.json. A test fails if they differ, and pins the route count:

def test_published_openapi_matches_the_code(self):
    path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "openapi-v1.json")
    with open(path, encoding="utf-8") as f:
        published = json.load(f)
    self.assertEqual(published, json.loads(json.dumps(api_v1.openapi_v1())),
                     "regenerate backend/openapi-v1.json from api_v1.openapi_v1()")
    routes = sum(len(ops) for ops in published["paths"].values())
    self.assertEqual(routes, 17)
Enter fullscreen mode Exit fullscreen mode

Known ceilings

This runs as a single instance on a small host, and some choices only hold there:

  • Rate-limit windows and pending confirm codes live in process memory. A restart forgets pending codes (the user just asks again), and a second instance would need Redis or a table.
  • The token-prefix set is loaded at startup and updated by the create endpoint, so a token inserted straight into the database is refused until a restart.
  • Preview equivalence is guaranteed by the shared code path, not yet by a test that diffs bodies.

Wrap-up

None of these patterns are new: Stripe popularised idempotency keys, ETags are older than most of us, and RFC 9457 exists so we stop inventing error formats. What was new to me was how much they matter once the client is a language model. An LLM retries, guesses field names and acts confidently on stale reads. Every one of those is a problem a boring HTTP mechanism already solves, as long as the server enforces it.

Questions, or things you would have done differently? I would genuinely like to hear them in the comments.

Top comments (0)