Week 9 — Real backend, deployed¶
Why this week: Streamlit on your laptop got you this far. Strangers need a proper service: an address on the internet that takes a message, runs your agent, keeps a safe record of the answer, and stays up when your laptop is closed. It is also the week you move from your Claude subscription to a paid API key, because a subscription login can't power an app other people use.
By Friday you have: an Anthropic API key with a spend limit; call_claude() running on the Anthropic SDK; a tested FastAPI /check endpoint that stores every check in Postgres with personal data removed; a Docker image; the API live on Google Cloud Run with its secrets in Secret Manager; a GitHub Actions job that deploys every merge to main; and your first public link.
Main online course: the FastAPI Tutorial — User Guide, from First Steps through Handling Errors and Dependencies. You don't need every page in between; each day below says which ones.
The heaviest week of the course, and a short one
The Ministry of Manpower lists Chinese New Year 2027 as Saturday 6 and Sunday 7 February, with Monday 8 February a public holiday in lieu for most people. Check the page for your own case. If Monday is off, start Monday's work on Tuesday and expect the week to spill over.
If it overruns, cut in this order:
- Deploy on merge (Friday). Deploy by hand with one command until week 11, then add it.
- The Streamlit page. Your first public link can be the API's own
/docspage for a week. - Never cut the spend limit, the tests, or the personal-data stripping. Those are what make it safe to put online.
Concepts¶
API and endpoint. An API (application programming interface) is a way for one program to ask another for something. A web API does it over HTTP, the same protocol your browser uses. An endpoint is one address on the API that does one job. Yours is POST /check: you send a message, you get a verdict back.
Request, response and status code. A request has a method (GET to read, POST to send data), a path such as /check, and usually a JSON body. The response has a status code and a JSON body. The codes you'll see this week: 200 (worked), 422 (your request was malformed, such as an empty message), 503 (the service can't help right now, try later), 500 (a bug on the server).
FastAPI and Pydantic. FastAPI turns a Python function into an endpoint. You describe the request and response as Pydantic models, as you did with CheckResult in week 1, and FastAPI checks every request against them, rejects bad ones with a 422, and writes an interactive documentation page at /docs for free.
Dependency. A function FastAPI calls to hand your endpoint something it needs, such as a database session or the agent. Because FastAPI does the handing over, a test can swap in a fake: a fake agent that never calls Claude, a throwaway database that disappears after the test.
Database, ORM and migration. Postgres stores rows in tables. An ORM (object-relational mapper; here SQLAlchemy) lets you describe a table as a Python class and save rows as objects, without writing SQL by hand. A migration is a small script that changes the database's structure, such as "create the checks table" or later "add a cost column". Alembic keeps them numbered in order, like commits for your tables, so your laptop's database and the live one end up identical.
Personal data, and keeping less of it. A pasted scam message often contains a name, a phone number or an NRIC. You need those to check the message, but you don't need them afterwards. So you strip them before anything is written to the database or a log. What you never store can't leak.
Image and container. A Dockerfile is a recipe. Building it produces an image: your code, Python and every package, frozen into one file. A container is a running copy of an image. "Works on my machine" stops being a problem, because the cloud runs the same image you ran.
Environment variables and secrets. Settings that change between your laptop and the cloud (the database address, the API key) live outside the code, in environment variables. This is one of the rules in The Twelve-Factor App. A secret is a setting that costs money or causes harm if someone else sees it. Your repository is public, so a secret must never be committed to it, not even once: deleting it later doesn't remove it from git history.
Serverless containers. Cloud Run takes your image and runs copies of it when requests arrive. When nobody is using it, it runs none, and you pay nothing for compute. The price of that is a "cold start": the first request after a quiet spell waits a few seconds while a copy starts.
Workload Identity Federation. The old way to let GitHub Actions deploy to Google Cloud was to download a permanent key file and store it in GitHub. If it leaked, anyone could deploy. With Workload Identity Federation, GitHub hands each job a short-lived signed token saying "I am a workflow in repo X". Google checks the signature and lets that repo, and only that repo, act as one service account for a few minutes. Nothing permanent is stored anywhere.
Why the subscription can't do this. Until now call_claude() ran the claude command on your laptop, logged in with your subscription. The Agent SDK overview says it plainly: "Unless previously approved, Anthropic does not allow third party developers to offer claude.ai login or rate limits for their products, including agents built on the Claude Agent SDK. Use the API key authentication methods described in the Quickstart instead." A public checker is exactly that kind of product, so it gets its own API key, billed per token.
Monday — The API key, and the one-function swap¶
Learn (75 min):
- Get your Claude API key (10 min)
- Rate limits, the "Spend limits" section only (10 min)
- Workspaces, the "Set workspace limits" part (10 min)
- Structured outputs: the Python examples and "JSON Schema limitations" (20 min). This is what
--json-schemawas doing for you. - Pricing, the model pricing table, and the models overview (15 min)
- Claude API errors again, now that you'll see them directly (10 min)
Build:
-
Create a Claude Console account and sign in. On the Billing page, add a small amount of credit. US$10 is plenty for this week: an agent check on Haiku costs a fraction of a US cent.
-
Put a ceiling on spending before you create any key. Two layers:
- Organisation: on the Billing page, in the Spend limits section, click Set limit and enter a monthly amount you'd be comfortable losing, such as US$20. When you reach a limit you set, requests fail with an error until you raise it or the month ends.
- Workspace: on the Workspaces page, click Create workspace and name it
scam-checker. Select it, click Spend limits, set a lower cap (say US$15), and add alerts at about half and three-quarters of it.
A workspace is a box for one project's keys and spending. If the public app is abused, the damage stops at the workspace limit.
-
On the API keys page, click Create key three times, each scoped to the
scam-checkerworkspace:laptop,ciandcloud-run. Give each an expiry date after the course ends. The Console shows a key once; paste it straight into a password manager. Separate keys mean you can cancel one (say, the CI key) without breaking the others, and the Console shows which one spent what. -
Keep the laptop key in a file git never sees. In the project root, add this line to
.gitignore, then create.env:Check before going further:
git statusmust not list.env. -
Add the SDK:
-
Replace the whole of
src/scam_checker/llm.py. The function's name, its arguments andClaudeResultstay exactly as they were, sochecker.py,agent.py, the reviewer and every script keep working untouched. That was the point of putting every call behind one function in week 1 (Calling Claude from Python)."""Call Claude through the Anthropic API, using your API key. Weeks 1-8 called the Claude Code CLI here. Week 9 swapped the body of call_claude() for the Anthropic SDK; its name, arguments and ClaudeResult are unchanged, so nothing that calls it needs to change. """ from __future__ import annotations import json import time from dataclasses import dataclass from functools import cache from typing import Any import anthropic # The short names used since week 1, mapped to API model IDs. # Check the models overview page when you start: IDs change as models are released. MODELS = { "haiku": "claude-haiku-5-5", "sonnet": "claude-sonnet-5-5", "opus": "claude-opus-5-5", } # US dollars per million tokens: (input, output). From the pricing page. # Haiku 5.5's price is for prompts up to 100,000 tokens; yours are far shorter. PRICES = { "claude-haiku-5-5": (0.10, 0.50), "claude-sonnet-5-5": (2.00, 10.00), "claude-opus-5-5": (4.00, 20.00), } MAX_TOKENS = 4096 # room for the model's thinking plus a short JSON answer class ClaudeError(RuntimeError): """The API call failed, timed out, or gave no usable answer.""" @dataclass class ClaudeResult: text: str data: dict[str, Any] | None # parsed JSON, only when you pass a schema input_tokens: int output_tokens: int cost_usd: float # estimated from PRICES; your Console bill is the real number duration_ms: int @cache def _client() -> anthropic.Anthropic: # Reads ANTHROPIC_API_KEY from the environment. max_retries=0 because # check_message() already retries; two retry layers multiply the waiting. return anthropic.Anthropic(max_retries=0) def estimate_cost(model_id: str, input_tokens: int, output_tokens: int) -> float: price_in, price_out = PRICES.get(model_id, (0.0, 0.0)) return (input_tokens * price_in + output_tokens * price_out) / 1_000_000 def call_claude( prompt: str, *, system: str | None = None, schema: dict[str, Any] | None = None, model: str = "haiku", timeout: float = 120, ) -> ClaudeResult: """Send one prompt to Claude and return the reply.""" model_id = MODELS.get(model, model) # a full model ID also works output_config: Any = anthropic.omit if schema: # transform_schema adds what the API insists on, such as # "additionalProperties": false, to a Pydantic-made schema. output_config = { "format": { "type": "json_schema", "schema": anthropic.transform_schema(schema), } } start = time.perf_counter() try: response = _client().messages.create( model=model_id, max_tokens=MAX_TOKENS, system=system or anthropic.omit, messages=[{"role": "user", "content": prompt}], output_config=output_config, timeout=timeout, ) except anthropic.APITimeoutError as exc: # check before APIConnectionError raise ClaudeError(f"no reply after {timeout} seconds") from exc except anthropic.APIConnectionError as exc: raise ClaudeError(f"could not reach the API: {exc}") from exc except anthropic.APIStatusError as exc: # 400, 401, 429, 529 and the rest raise ClaudeError(f"API error {exc.status_code}: {exc.message}") from exc duration_ms = int((time.perf_counter() - start) * 1000) if response.stop_reason == "refusal": raise ClaudeError("the model declined to answer") if response.stop_reason == "max_tokens": raise ClaudeError(f"the reply was cut off at {MAX_TOKENS} tokens") # The reply can start with thinking blocks; the answer is in the text blocks. text = "".join(block.text for block in response.content if block.type == "text") data = None if schema: try: data = json.loads(text) except json.JSONDecodeError as exc: raise ClaudeError(f"reply was not valid JSON: {text[:200]}") from exc usage = response.usage return ClaudeResult( text=text, data=data, input_tokens=usage.input_tokens, output_tokens=usage.output_tokens, cost_usd=estimate_cost(model_id, usage.input_tokens, usage.output_tokens), duration_ms=duration_ms, ) if __name__ == "__main__": reply = call_claude("Say hello in five words.") print(reply.text) print(f"{reply.input_tokens} in, {reply.output_tokens} out, {reply.duration_ms} ms") print(f"about US${reply.cost_usd:.6f}")What changed, line by line:
- Model names. The CLI's
haikumeant "the newest Haiku". The API wants an exact model ID, soMODELStranslates. In October 2026 the newest models were Haiku 5.5, Sonnet 5.5 and Opus 5.5. Check the models overview and the deprecations page when you start; if the IDs have moved on, updateMODELSandPRICESand re-run your eval before trusting the new numbers. - Structured output.
output_config={"format": {"type": "json_schema", ...}}is the API's version of--json-schema. The API insists that every object in the schema says"additionalProperties": false; Pydantic doesn't add that, andanthropic.transform_schema()does. - Errors. Each SDK error becomes a
ClaudeError, which is what your week 1 retry loop already catches. The timeout check comes first because a timeout is a special kind of connection error. - Thinking. Current models can think before answering, and the reply may start with "thinking" blocks. Only the text blocks are the answer. Thinking tokens are billed as output tokens, which is one reason
cost_usdis worth logging. - Cost. The SDK reports token counts in
response.usage;estimate_costmultiplies them by the published prices. For example, 1,500 input and 300 output tokens on Haiku 5.5 is 1,500 × 0.10 + 300 × 0.50 = 300 millionths of a dollar: US$0.0003.
- Model names. The CLI's
-
Try it.
--env-fileloads.envfor this one command:You should see a greeting, token counts and a cost of a few millionths of a dollar. Open the Console's Usage page: the call is there, under the
scam-checkerworkspace. -
Tests that never touch the network. The SDK sends its requests through an HTTP library called
httpx2, which lets you plug in a fake "transport" that answers instead of Anthropic. The SDK then does all its real work (building the request, parsing the reply, raising errors) against your fake. Createtests/test_llm.py:"""Test call_claude() with the HTTP layer faked: no key, no network, no cost.""" import json import anthropic import httpx2 import pytest from scam_checker import llm def api_message(text: str, stop_reason: str = "end_turn") -> dict: """The JSON the Messages API sends back, cut down to what the SDK needs.""" return { "id": "msg_test", "type": "message", "role": "assistant", "model": "claude-haiku-5-5", "content": [ {"type": "thinking", "thinking": "", "signature": "sig"}, {"type": "text", "text": text}, ], "stop_reason": stop_reason, "stop_sequence": None, "usage": {"input_tokens": 100, "output_tokens": 20}, } @pytest.fixture def fake_api(monkeypatch): """Point call_claude() at a fake API that answers with `handler`.""" def install(handler): client = anthropic.Anthropic( api_key="test-key", max_retries=0, http_client=anthropic.DefaultHttpxClient( transport=httpx2.MockTransport(handler) ), ) monkeypatch.setattr(llm, "_client", lambda: client) return install def test_structured_reply_is_parsed(fake_api): sent = {} def handler(request): sent.update(json.loads(request.content)) return httpx2.Response(200, json=api_message('{"verdict": "scam"}')) fake_api(handler) schema = {"type": "object", "properties": {"verdict": {"type": "string"}}} reply = llm.call_claude("Check this", system="Be careful", schema=schema) assert reply.data == {"verdict": "scam"} assert sent["model"] == "claude-haiku-5-5" assert sent["system"] == "Be careful" assert sent["output_config"]["format"]["schema"]["additionalProperties"] is False assert reply.cost_usd == pytest.approx((100 * 0.10 + 20 * 0.50) / 1_000_000) def test_rate_limit_becomes_claude_error(fake_api): error = { "type": "error", "error": {"type": "rate_limit_error", "message": "slow down"}, } fake_api(lambda request: httpx2.Response(429, json=error)) with pytest.raises(llm.ClaudeError, match="429"): llm.call_claude("hi")Notice what the first test checks: not only the parsed answer, but the request your code sent. That is how you catch a missing
additionalPropertiesbefore it costs a failed call. -
Your turn: add two more tests to the same file. One where the fake raises
httpx2.ReadTimeout, andcall_claudemust raiseClaudeErrormentioning "no reply". One where the API answers200but withstop_reasonset to"refusal", andcall_claudemust raiseClaudeErrortoo.Stuck? The two tests
def test_timeout_becomes_claude_error(fake_api): def handler(request): raise httpx2.ReadTimeout("timed out", request=request) fake_api(handler) with pytest.raises(llm.ClaudeError, match="no reply"): llm.call_claude("hi", timeout=1) def test_refusal_becomes_claude_error(fake_api): fake_api(lambda request: httpx2.Response(200, json=api_message("", "refusal"))) with pytest.raises(llm.ClaudeError, match="declined"): llm.call_claude("hi") -
Re-run your week 5 eval on the API (
scripts/run_eval.py), or week 1'scompare.pyif that's quicker. The numbers should be close to last week's; if they jump, the model behindhaikuhas changed, and that is worth one line in your README. Note the total cost: it is now real money, and it's your first data point for week 11. -
Move the week 7 CI gate onto the API. In
.github/workflows/ci.yml, delete the step that installs the Claude Code CLI and change the eval step's environment fromCLAUDE_CODE_OAUTH_TOKENto the new key:- name: Eval gate run: uv run python scripts/run_eval.py env: ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}Then store the
cikey as a repository secret and remove the old token:gh secret set ANTHROPIC_API_KEY # paste the ci key when asked gh secret delete CLAUDE_CODE_OAUTH_TOKENgh secret setreads the value from your keyboard, so the key never lands in your shell history. Every CI run now costs a little; if your eval runs on every push, restrict it to pull requests andmain, or to changes undersrc/and your prompts.
Done when: uv run pytest -q passes with no network, python -m scam_checker.llm answers with your laptop key, the eval number is in your README, and CI is green using ANTHROPIC_API_KEY.
Stuck?
401 authentication_error: the key is wrong or wasn't loaded. Run with--env-file .env, check there are no quotes or spaces around the key in.env, and check you copied the whole key.ClaudeError: API error 400: You have reached your specified API usage limits: that's your own spend limit working. Raise it on the Billing or Workspaces page if it's lower than you meant.400 invalid_request_errormentioning the schema: something in your schema isn't supported (see "JSON Schema limitations"). Printanthropic.transform_schema(CheckResult.model_json_schema())and read it.ClaudeError: the reply was cut off at 4096 tokens: the model thought for a long time. RaiseMAX_TOKENS, or shorten the prompt.ModuleNotFoundError: httpx2: you have an old 0.x version of the SDK. Runuv add "anthropic>=1".- A test is slow or asks for a key: the fake wasn't installed, so the real API was called. Check the test takes the
fake_apifixture and calls it beforecall_claude. - CI fails with
401: the secret name in the workflow must match exactly,ANTHROPIC_API_KEY.
If a key ever reaches GitHub
Treat it as stolen, even if you delete the commit a minute later. Automated scanners watch public repositories for keys. Go to the API keys page, disable that key, create a new one, and update .env and the secret. Then check the Usage page for spending you don't recognise.
Tuesday — What to store, and a database to store it in¶
Learn (75 min):
- Presidio: Getting started (15 min): what an analyser and an anonymiser do
- SQLAlchemy ORM Quick Start (25 min); the Unified Tutorial is the full reference, for later
- Alembic Tutorial, up to "Running our First Migration", then Auto Generating Migrations (35 min)
Build:
-
Install Presidio and a small English language model. The model comes from spaCy's GitHub releases rather than PyPI, so the full address goes into
uv add, anduv.lockrecords it like any other package:uv add presidio-analyzer presidio-anonymizer uv add "en_core_web_sm @ https://github.com/explosion/spacy-models/releases/download/en_core_web_sm-3.8.0/en_core_web_sm-3.8.0-py3-none-any.whl"Presidio's tutorial uses the large model (
en_core_web_lg, about 400 MB). The small one is about 13 MB, which keeps your Docker image and cold starts smaller. It finds fewer names; your tests below tell you whether that's good enough. -
src/scam_checker/pii.py. Presidio's default recognisers are tuned for the US and UK; a Singapore mobile number without+65slips through, and the NRIC recogniser isn't switched on. Two lines fix that:"""Strip personal data from text before it is stored or logged.""" from functools import cache from presidio_analyzer import AnalyzerEngine from presidio_analyzer.nlp_engine import NlpEngineProvider from presidio_analyzer.predefined_recognizers import PhoneRecognizer, SgFinRecognizer from presidio_anonymizer import AnonymizerEngine # What counts as personal data here. Links, bank and company names stay: # they are evidence about the scam, not about the person who received it. ENTITIES = ["PERSON", "PHONE_NUMBER", "EMAIL_ADDRESS", "SG_NRIC_FIN", "CREDIT_CARD"] NLP_CONFIG = { "nlp_engine_name": "spacy", "models": [{"lang_code": "en", "model_name": "en_core_web_sm"}], } @cache # loading the language model takes a second or two; do it once def _engines() -> tuple[AnalyzerEngine, AnonymizerEngine]: nlp_engine = NlpEngineProvider(nlp_configuration=NLP_CONFIG).create_engine() analyzer = AnalyzerEngine(nlp_engine=nlp_engine, supported_languages=["en"]) analyzer.registry.add_recognizer(PhoneRecognizer(supported_regions=["SG"])) analyzer.registry.add_recognizer(SgFinRecognizer()) return analyzer, AnonymizerEngine() def strip_pii(text: str) -> str: """Replace names, phone numbers, emails, NRICs and card numbers with <TYPE>.""" analyzer, anonymizer = _engines() findings = analyzer.analyze(text=text, language="en", entities=ENTITIES) # Presidio's two packages each define their own result class; they match at run time. return anonymizer.anonymize(text=text, analyzer_results=findings).text # type: ignore[arg-type]If your week 8
guardrails.pyalready has a redaction function, compare the two on the same messages and keep whichever catches more; then call that one everywhere. -
Try it on a message full of personal data:
uv run python -c "from scam_checker.pii import strip_pii; print(strip_pii('Hi Mrs Tan Mei Ling, call 9123 4567 or email lim@dbs-help.com. NRIC S1234567D. Visit https://dbs-secure.com'))"You should see
Hi <PERSON>, call <PHONE_NUMBER> or email <EMAIL_ADDRESS>. NRIC <SG_NRIC_FIN>. Visit https://dbs-secure.com. The link stays on purpose: it's evidence about the scam, not about the person. -
Your turn: run
strip_piion 20 messages from your labelled set and read every output. Write down in your README what it missed and what it removed that it shouldn't have. Decide about car plates such asSJK1234Z: personal data or not? A missed name is a privacy bug; treat it like one. -
Add the database packages.
psycopgis the driver that lets Python talk to Postgres: -
The table, as a Python class, in
src/scam_checker/api/db.py(create theapifolder with an empty__init__.py):"""The database: one table of past checks, with personal data already removed.""" import os from collections.abc import Iterator from datetime import datetime from functools import cache from sqlalchemy import JSON, DateTime, Engine, String, Text, create_engine, func from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column class Base(DeclarativeBase): pass class CheckRecord(Base): __tablename__ = "checks" id: Mapped[int] = mapped_column(primary_key=True) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), server_default=func.now() ) message: Mapped[str] = mapped_column(Text) # after strip_pii() verdict: Mapped[str] = mapped_column(String(16)) red_flags: Mapped[list[str]] = mapped_column(JSON) advice: Mapped[str] = mapped_column(Text) steps: Mapped[int] duration_ms: Mapped[int] @cache def get_engine() -> Engine: # pool_pre_ping checks a connection still works before using it; hosted # databases close idle connections, and this saves you a confusing error. return create_engine(os.environ["DATABASE_URL"], pool_pre_ping=True) def get_session() -> Iterator[Session]: """One database session per request; FastAPI closes it afterwards.""" with Session(get_engine()) as session: yield sessionMapped[int]says "an integer column that can't be empty".JSONstores the list of red flags as it is. The database address comes from theDATABASE_URLenvironment variable, so the same code talks to your laptop's Postgres today and to a hosted one on Thursday. -
Start the week 4 Postgres. If your
docker-compose.ymlhas nohealthcheckfor the database yet, add one now (it lets the API wait for Postgres on Thursday):services: db: image: pgvector/pgvector:pg17 environment: POSTGRES_USER: scam POSTGRES_PASSWORD: scam POSTGRES_DB: scam ports: - "5432:5432" volumes: - pgdata:/var/lib/postgresql/data healthcheck: test: ["CMD-SHELL", "pg_isready -U scam -d scam"] interval: 2s timeout: 5s retries: 15Use the user, password and database name from your own week 4 file. Add the matching address to
.env: -
Set up Alembic and point it at your models and your environment variable:
This creates
alembic.iniand analembic/folder. Leave thesqlalchemy.urlline inalembic.inias it is; you'll override it. Inalembic/env.py, addimport osat the top, then replace thetarget_metadata = Noneblock with:# Read the database address from the environment, never from alembic.ini, # so no password ends up in git. "%" must be doubled for the config parser. config.set_main_option("sqlalchemy.url", os.environ["DATABASE_URL"].replace("%", "%%")) # Your models, so `alembic revision --autogenerate` can compare them with the database. from scam_checker.api.db import Base target_metadata = Base.metadata def include_name(name, type_, parent_names): """Only manage tables defined in db.py; leave week 4's RAG tables alone.""" if type_ == "table": return name in target_metadata.tables return TrueThen add
include_name=include_name,to bothcontext.configure(...)calls further down the file, one inrun_migrations_offlineand one inrun_migrations_online.The
include_namefilter matters. Your week 4 RAG tables live in the same database but aren't indb.py. Without the filter, Alembic would see them as "tables that shouldn't exist" and write a migration that deletes them. -
Generate the first migration, read it, then apply it:
uv run --env-file .env alembic revision --autogenerate -m "create checks table" uv run ruff check --fix alembic && uv run ruff format alembic uv run --env-file .env alembic upgrade headOpen the new file in
alembic/versions/before you apply it. It should contain oneop.create_table('checks', ...)and nothing that drops a table. Generated code often fails your lint rules, hence theruffline. -
Look at the result:
You should see the eight columns, and a second table,
alembic_version, where Alembic notes which migration the database is at.
Done when: strip_pii handles your 20 test messages and the misses are written down, the migration file is committed, and \d checks shows the table.
Stuck?
OSError: [E050] Can't find model 'en_core_web_sm': the seconduv addline didn't run, or ran in another folder. Run it again in the project root.KeyError: 'DATABASE_URL'from Alembic: you left out--env-file .env.connection refusedon port 5432: Postgres isn't running.docker compose psshould showdbas healthy.- The generated migration is empty: Alembic couldn't see your model. Check the
from scam_checker.api.db import Baseline, and thattarget_metadata = Base.metadatareplaced the oldNone. - The migration wants to
drop_tablesomething: stop. Delete that file, checkinclude_nameis passed to bothconfigurecalls, and generate it again. ValueError: invalid interpolation syntax: your password contains%. The.replace("%", "%%")line handles it; check it's there.
Wednesday — The /check endpoint, tested¶
Learn (90 min): in the FastAPI tutorial: First Steps, Request Body, Handling Errors, Dependencies and Testing. Then Concurrency and async / await, for one reason: your agent is ordinary blocking code, so the endpoint is a plain def, and FastAPI runs it in a worker thread so one slow check doesn't stall everyone else.
Build:
-
Add FastAPI with its standard extras (the web server and the test client):
-
What goes in and out,
src/scam_checker/api/schemas.py. The response reuses your week 1CheckResultand adds the id it was stored under:"""What the API accepts and returns. FastAPI checks both against these.""" from pydantic import BaseModel, Field from scam_checker.models import CheckResult class CheckRequest(BaseModel): text: str = Field(min_length=1, max_length=2000, description="The message to check") class CheckResponse(CheckResult): """The checker's answer, plus the id it was stored under.""" check_id: intmax_length=2000is a cost control as much as a check: nobody can paste a novel and make you pay to read it. -
The endpoint,
src/scam_checker/api/main.py:"""The checker as a web API. Run locally with: uv run fastapi dev src/scam_checker/api/main.py """ import logging import time from collections.abc import Callable from typing import Annotated from fastapi import Depends, FastAPI, HTTPException from sqlalchemy.orm import Session from scam_checker.agent import AgentRun, run_agent from scam_checker.api.db import CheckRecord, get_session from scam_checker.api.schemas import CheckRequest, CheckResponse from scam_checker.llm import ClaudeError from scam_checker.pii import strip_pii logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) app = FastAPI(title="Scam checker API", version="0.1.0") Agent = Callable[[str], AgentRun] def get_agent() -> Agent: """The agent to use. A function, so tests can swap in a fake one.""" return run_agent @app.post("/check", response_model=CheckResponse) def check( request: CheckRequest, session: Annotated[Session, Depends(get_session)], agent: Annotated[Agent, Depends(get_agent)], ) -> CheckResponse: start = time.perf_counter() try: run = agent(request.text) except ClaudeError as exc: logger.warning("check failed: %s", exc) raise HTTPException(503, "The checker is busy. Please try again in a minute.") result = run.result record = CheckRecord( message=strip_pii(request.text), verdict=result.verdict, red_flags=[strip_pii(flag) for flag in result.red_flags], advice=strip_pii(result.advice), steps=len(run.steps), duration_ms=int((time.perf_counter() - start) * 1000), ) session.add(record) session.commit() logger.info("check %d: %s in %d ms", record.id, record.verdict, record.duration_ms) return CheckResponse(check_id=record.id, **result.model_dump())Read it in order. FastAPI checks the request against
CheckRequest, opens a database session, gets the agent, and calls your function. The agent sees the full message: it needs the phone number and link to investigate them. Only the stripped version is stored and logged. If the agent itself fails withClaudeError, the caller gets a503and a plain sentence, never a stack trace. (Your agent probably handles most failures itself and returns a fallback answer; this is for the ones that escape.) -
Run it and use the page FastAPI built for you:
Open
http://localhost:8000/docs, clickPOST /check, then Try it out, paste a message and Execute. Then try an empty message and see the422. Then check the row landed, without the personal data: -
Endpoint tests with no Claude and no Postgres,
tests/test_api.py. The fake agent replacesget_agent, and an in-memory SQLite database replacesget_session:"""Test the API with a fake agent and a throwaway database: no LLM, no Postgres.""" import pytest from fastapi.testclient import TestClient from sqlalchemy import create_engine, select from sqlalchemy.orm import Session from sqlalchemy.pool import StaticPool from scam_checker.agent import AgentRun from scam_checker.api.db import Base, CheckRecord, get_session from scam_checker.api.main import app, get_agent from scam_checker.llm import ClaudeError from scam_checker.models import CheckResult SCAM = CheckResult( verdict="scam", red_flags=["asks you to call 9123 4567 urgently"], advice="Don't call. Contact your bank using the number on your card.", ) def fake_agent(text: str) -> AgentRun: return AgentRun(result=SCAM, steps=[{"step": 1}, {"step": 2}]) def broken_agent(text: str) -> AgentRun: raise ClaudeError("API error 529: overloaded") @pytest.fixture def engine(): # An in-memory SQLite database, created fresh for each test. engine = create_engine( "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool ) Base.metadata.create_all(engine) return engine @pytest.fixture def client(engine): def test_session(): with Session(engine) as session: yield session app.dependency_overrides[get_session] = test_session app.dependency_overrides[get_agent] = lambda: fake_agent yield TestClient(app) app.dependency_overrides.clear() def test_check_returns_the_verdict(client): response = client.post("/check", json={"text": "DBS: call 9123 4567 now"}) assert response.status_code == 200 body = response.json() assert body["verdict"] == "scam" assert body["check_id"] == 1 def test_check_is_stored_without_personal_data(client, engine): client.post("/check", json={"text": "Hi Mrs Tan, call 9123 4567 now"}) with Session(engine) as session: record = session.scalars(select(CheckRecord)).one() assert "9123 4567" not in record.message assert "Tan" not in record.message assert "9123 4567" not in record.red_flags[0] assert record.steps == 2SQLite isn't Postgres, but for "does a row get saved, without the phone number" it behaves the same, and it needs no server, so CI can run it.
-
Your turn: add the error-path tests to the same file. Requests with an empty text, with 2,001 characters, and with the wrong field name must all get
422. Withbroken_agentswapped in, a request must get503with "try again" in the message.Stuck? The error-path tests
@pytest.mark.parametrize("text", ["", "x" * 2001]) def test_empty_or_huge_messages_are_rejected(client, text): response = client.post("/check", json={"text": text}) assert response.status_code == 422 def test_missing_field_is_rejected(client): assert client.post("/check", json={"message": "hi"}).status_code == 422 def test_llm_failure_gives_503(client): app.dependency_overrides[get_agent] = lambda: broken_agent response = client.post("/check", json={"text": "hello"}) assert response.status_code == 503 assert "try again" in response.json()["detail"]parametrizeruns the same test once per value in the list, so one function covers both bad lengths. -
Run everything, lint, commit:
Done when: the /docs page checks a real message, the stored row has no name or number in it, and pytest passes with tests for 200, 422 and 503.
Stuck?
Error loading ASGI app. Could not import module: run the command from the project root, and checksrc/scam_checker/api/__init__.pyexists.address already in useon port 8000: something else is using it. Add--port 8001and use that port.- Tests call the real agent: overrides must be set before the request, and the fixture must
yieldthe client sodependency_overrides.clear()runs afterwards. sqlite3.ProgrammingError ... thread: keepcheck_same_thread=FalseandStaticPoolexactly as shown; FastAPI's worker thread and the test use the same in-memory database.- The stored message still contains a name: that's Tuesday's
strip_piimissing it. Add the message to your README list.
Thursday — A container, then the cloud¶
Learn (75 min):
- Docker: Get started, the "What is an image?" and "What is a container?" pages, if week 4 felt hazy (20 min)
- uv: Using uv in Docker, "Installing a project" and "Intermediate layers" (20 min)
- Quickstart: deploy a FastAPI app to Cloud Run (20 min): read it; you'll do your own version below
- Configure secrets for services (15 min)
Build:
-
Tell Docker what to leave out, in
.dockerignore. Above all.env: a key baked into an image travels wherever the image goes. -
The
Dockerfile, following uv's guide. Pin the uv image to the same version as your laptop's (uv --version), so both readuv.lockthe same way:# The checker API as a container image. Build with: docker build -t scam-checker-api . FROM python:3.12-slim # Copy the uv program from its official image, pinned to one version. COPY --from=ghcr.io/astral-sh/uv:0.13.1 /uv /uvx /bin/ WORKDIR /app ENV UV_COMPILE_BYTECODE=1 UV_LINK_MODE=copy UV_NO_DEV=1 # 1. Dependencies only. This slow layer is reused until uv.lock changes. RUN --mount=type=cache,target=/root/.cache/uv \ --mount=type=bind,source=uv.lock,target=uv.lock \ --mount=type=bind,source=pyproject.toml,target=pyproject.toml \ uv sync --locked --no-install-project # 2. Your code, which changes often, then install the project itself. COPY . /app RUN --mount=type=cache,target=/root/.cache/uv \ uv sync --locked ENV PATH="/app/.venv/bin:$PATH" # Run as an ordinary user, not root: less damage if something goes wrong. RUN useradd --create-home app USER app # Cloud Run tells the container which port to listen on in $PORT. CMD ["sh", "-c", "uvicorn scam_checker.api.main:app --host 0.0.0.0 --port ${PORT:-8080}"]The two
uv syncsteps are a speed trick. Docker reuses any step whose inputs haven't changed, so the slow "install every package" step reruns only whenuv.lockchanges, not every time you edit your code. -
Build it:
The first build takes a few minutes, and the image is over 1 GB: it holds Python, spaCy's model and every package in
uv.lock. A rebuild after a code change should take seconds. -
Run the API and Postgres together. Add the
apiservice todocker-compose.yml, underservices:, next todb:api: build: . ports: - "8080:8080" environment: DATABASE_URL: postgresql://scam:scam@db:5432/scam ANTHROPIC_API_KEY: ${ANTHROPIC_API_KEY} depends_on: db: condition: service_healthyInside compose, the API reaches Postgres at the host name
db, notlocalhost. Compose fills in${ANTHROPIC_API_KEY}from your.envfile automatically.docker compose up -d --build curl -s -X POST localhost:8080/check -H 'content-type: application/json' -d '{"text":"DBS: your account is locked. Call 9123 4567 now"}'You should get a verdict as JSON. Then run your six hand-written messages from week 1 through it and check the answers still cite scam patterns and use the graph tool, as they did outside the container. If a tool goes quiet, it can't reach its data: see step 6. If it works here, it will work on Cloud Run: it's the same image.
-
A database the cloud can reach. Your laptop's Postgres can't serve the internet, and Google's own Cloud SQL bills by the hour even when idle. Neon has a free plan, with no card needed, that sleeps after a few idle minutes. Sign up, create a project in the AWS Asia Pacific (Singapore) region (
aws-ap-southeast-1), click Connect, and copy the connection string. Use the direct one, without-poolerin the host name; Alembic needs a direct connection. It looks like:postgresql://user:password@ep-something-123456.ap-southeast-1.aws.neon.tech/neondb?sslmode=require&channel_binding=requirePut it in your password manager, not in any file in the repo. Then apply your migration to it:
The single quotes stop your shell from misreading
&and?. The free plan is enough for this course; its limits are on the pricing page. -
The agent needs more than the
checkstable. Your week 4 scam patterns are in your laptop's Postgres, and the live agent will look for them in Neon. Neon supports pgvector: in Neon's SQL editor, runCREATE EXTENSION IF NOT EXISTS vector;, then run your week 4 loading script withDATABASE_URLset to the Neon string. While you're there, check thatretrieval.pyand the week 6 graph tool find their data throughDATABASE_URLor files inside the repository, not throughlocalhostor a laptop path that.dockerignoreleaves out of the image. -
Install the Google Cloud command line and sign in. Install the gcloud CLI (on a Mac with Homebrew:
brew install --cask gcloud-cli), then:It opens a browser to sign in and asks you to create or pick a project; call it something like
scam-checker-<yourname>. Cloud Run needs a billing account with a card, even inside the free allowance. New accounts get trial credit; the current amounts and the monthly Cloud Run free allowance are on the free features page. -
Before anything else in Google Cloud, create a budget with email alerts, say S$10 a month, alerting at 50% and 90%. A budget only warns you; it doesn't stop spending. The
--max-instancesflag below is what caps how much Cloud Run can run at once. -
Switch on the services you need and choose Singapore as your region:
-
Put the two secrets in Secret Manager.
read -rsreads what you paste without showing it or saving it in your shell history:read -rs KEY && printf '%s' "$KEY" | gcloud secrets create anthropic-api-key --data-file=- && unset KEY read -rs KEY && printf '%s' "$KEY" | gcloud secrets create database-url --data-file=- && unset KEYPaste the
cloud-runAPI key at the first prompt and press Enter; paste the Neon string at the second. -
Let Cloud Run's identity read them. By default your service runs as the project's "Compute Engine default service account", which is also the identity that builds your image:
PROJECT_ID=$(gcloud config get project) PROJECT_NUMBER=$(gcloud projects describe "$PROJECT_ID" --format='value(projectNumber)') RUNTIME_SA="${PROJECT_NUMBER}-compute@developer.gserviceaccount.com" for secret in anthropic-api-key database-url; do gcloud secrets add-iam-policy-binding "$secret" \ --member="serviceAccount:${RUNTIME_SA}" --role=roles/secretmanager.secretAccessor done gcloud projects add-iam-policy-binding "$PROJECT_ID" \ --member="serviceAccount:${RUNTIME_SA}" --role=roles/run.builderThis is IAM (identity and access management): a list of who may do what. The first loop says "this service may read these two secrets, and nothing else in Secret Manager".
-
Deploy.
--source .uploads your folder; Google builds it with your Dockerfile and runs it:gcloud run deploy scam-checker-api \ --source . \ --region asia-southeast1 \ --allow-unauthenticated \ --memory 1Gi \ --max-instances 2 \ --set-secrets ANTHROPIC_API_KEY=anthropic-api-key:latest,DATABASE_URL=database-url:latestSay yes when it offers to create an Artifact Registry repository.
--allow-unauthenticatedmakes it public, which is the point.--max-instances 2means that however many requests arrive, at most two copies run.:latestmeans "the newest version of the secret, read when a copy starts"; after you change a secret, deploy again to pick it up. -
Test the live service with the address it printed:
curl -s -X POST https://scam-checker-api-xxxx.asia-southeast1.run.app/check \ -H 'content-type: application/json' -d '{"text":"Your parcel is held. Pay $1.99 at singpost-redelivery.com"}'Open
/docson that address too. The first request after a quiet spell is slow; that's the cold start. Then check the stored row in Neon's SQL editor:select id, message, verdict from checks.
Done when: docker compose up works locally, and the Cloud Run address answers /check with the key read from Secret Manager, not from any file.
Stuck?
failed to solve ... uv syncmentioning the lockfile: your laptop's uv and the image's uv are different versions. Match the tag inCOPY --from=ghcr.io/astral-sh/uv:...touv --version, or runuv lockand rebuild.Bind for 0.0.0.0:8080 failed: port is already allocated: something else uses the port. Change the left number, such as"8081:8080".- The variable
ANTHROPIC_API_KEY"is not set" in compose:.envmust sit next todocker-compose.yml. - The deploy fails with
PERMISSION_DENIEDnaming Cloud Build or a builder: theroles/run.builderbinding in step 11 is missing, or needs a minute to take effect. - The service starts but every check gives
503and the logs say401: the secret has a trailing newline or the wrong key. Create a new version withread -rs KEY && printf '%s' "$KEY" | gcloud secrets versions add anthropic-api-key --data-file=- && unset KEYand deploy again. Permission denied on secret: thesecretAccessorbinding is missing for that secret.- Container failed to start, logs mention
DATABASE_URL: the secret wasn't attached. Check--set-secretsand the secret names. - Logs: in the Cloud Run console, open the service and the Logs tab, or run
gcloud run services logs read scam-checker-api --region asia-southeast1.
A public API costs money when strangers use it
Anyone who finds the address can call /check, and each call spends your credit. Your protection is layered: the 2,000-character limit, at most two running copies, and above all the Console spend limits from Monday, which stop the spending outright. Check the Console's Usage page daily this week.
Friday — Deploy on merge, a public link, and demo day¶
Learn (45 min):
- google-github-actions/auth, the "Workload Identity Federation through a Service Account" setup (20 min)
- OpenID Connect in GitHub Actions, the first two sections (10 min)
- Streamlit: Deploy your app, App dependencies and Secrets management (15 min)
Build (before 9:00 or after demo day):
-
Let GitHub deploy without a stored key. These commands follow the auth action's guide: one service account that may deploy, one identity pool, and one provider that admits only your repository:
PROJECT_ID=$(gcloud config get project) PROJECT_NUMBER=$(gcloud projects describe "$PROJECT_ID" --format='value(projectNumber)') REPO="<your-github-username>/scam-checker" DEPLOYER="github-deployer@${PROJECT_ID}.iam.gserviceaccount.com" gcloud iam service-accounts create github-deployer gcloud iam workload-identity-pools create github --location=global --display-name="GitHub Actions" gcloud iam workload-identity-pools providers create-oidc scam-checker \ --location=global --workload-identity-pool=github \ --issuer-uri="https://token.actions.githubusercontent.com" \ --attribute-mapping="google.subject=assertion.sub,attribute.repository=assertion.repository" \ --attribute-condition="assertion.repository == '${REPO}'" POOL=$(gcloud iam workload-identity-pools describe github --location=global --format='value(name)') gcloud iam service-accounts add-iam-policy-binding "$DEPLOYER" \ --role=roles/iam.workloadIdentityUser \ --member="principalSet://iam.googleapis.com/${POOL}/attribute.repository/${REPO}"Then give the deployer only what a source deploy needs:
for role in roles/run.sourceDeveloper roles/serviceusage.serviceUsageConsumer; do gcloud projects add-iam-policy-binding "$PROJECT_ID" --member="serviceAccount:${DEPLOYER}" --role="$role" done gcloud iam service-accounts add-iam-policy-binding "${PROJECT_NUMBER}-compute@developer.gserviceaccount.com" \ --member="serviceAccount:${DEPLOYER}" --role=roles/iam.serviceAccountUserThe last line lets the deployer start the service under the runtime identity, the one that can read the secrets. The deployer itself can't read them.
-
Tell GitHub the two names it needs. These aren't secrets, so they go in as variables; the Neon string is a secret, for the migration step:
-
.github/workflows/deploy.yml. It waits for your week 7 CI to pass onmain, applies any new migrations, then deploys:# Deploy the API to Cloud Run after CI passes on main. name: Deploy API on: workflow_run: workflows: ["CI"] # must match the name: line at the top of ci.yml types: [completed] branches: [main] permissions: contents: read id-token: write # lets this job ask GitHub for a short-lived identity token concurrency: group: deploy-api # one deploy at a time; a second waits its turn jobs: deploy: # Only when CI passed, and only for pushes (merges) to main, not pull requests. if: github.event.workflow_run.conclusion == 'success' && github.event.workflow_run.event == 'push' runs-on: ubuntu-latest steps: - uses: actions/checkout@v7 with: ref: ${{ github.event.workflow_run.head_sha }} # the exact commit CI tested - uses: google-github-actions/auth@v3 with: workload_identity_provider: ${{ vars.GCP_WIF_PROVIDER }} service_account: ${{ vars.GCP_DEPLOY_SA }} - uses: astral-sh/setup-uv@v10.3.0 - name: Apply database migrations run: uv run alembic upgrade head env: DATABASE_URL: ${{ secrets.DATABASE_URL }} - uses: google-github-actions/deploy-cloudrun@v3 with: service: scam-checker-api region: asia-southeast1 source: ./ secrets: |- ANTHROPIC_API_KEY=anthropic-api-key:latest DATABASE_URL=database-url:latestThe
workflows: ["CI"]line must match thename:at the top of yourci.yml. Commit and push tomain, then watch the Actions tab: CI first, then "Deploy API". Change the API'sversioninmain.py, push again, and check/docsshows the new number. -
Your first public link. Strangers need a page, not a JSON API. Make a small Streamlit page that only calls your API, in its own folder,
ui/streamlit_app.py:"""The public demo page. It only calls your API; no API key lives here.""" import httpx import streamlit as st API_URL = st.secrets["API_URL"] # set in Community Cloud's Secrets box st.title("Is this message a scam?") st.caption("A learning project. It can be wrong; when in doubt, don't click.") text = st.text_area("Paste the message here", height=150, max_chars=2000) if st.button("Check", type="primary") and text.strip(): with st.spinner("Checking... this can take up to a minute"): try: response = httpx.post(f"{API_URL}/check", json={"text": text}, timeout=90) response.raise_for_status() result = response.json() except httpx.HTTPError: st.error("The checker is not responding. Please try again later.") st.stop() if result["verdict"] == "scam": st.error("Likely a scam") elif result["verdict"] == "not_scam": st.success("Probably genuine") else: st.warning("Not sure") if result["red_flags"]: st.subheader("Warning signs") for flag in result["red_flags"]: st.write(f"- {flag}") st.subheader("What to do") st.write(result["advice"])And
ui/requirements.txt, so Community Cloud installs only these two packages instead of your whole project:Community Cloud looks for a dependency file next to the entrypoint first, so
ui/requirements.txtwins over your rootuv.lock. No API key lives in this app; it has nothing worth stealing. -
Deploy it: on Streamlit Community Cloud, create an app from your repo with
ui/streamlit_app.pyas the main file. In Advanced settings, paste into the Secrets box:Send the link to one friend. That's your first stranger-ready product; week 10 replaces this page with a streaming one.
-
Update your README: the public link, the API's
/docslink, and one paragraph on what is stored (stripped messages and verdicts) and what isn't.
Demo day (9:00–12:00):
- Record a 3-minute screen video: a check on the public page, the stored row with
<PHONE_NUMBER>in it, the green CI and deploy runs, and the Console's spend limit. - Write five lines on what you learned this week.
- Send both to Claude with the Friday check-in. Include your cost per check from the Console's Usage page.
Friday afternoon: the cloud side track starts this week. If deploy on merge isn't working yet, spend the afternoon on that instead.
Stuck?
Error: google-github-actions/auth failed ... unable to retrieve identity token: the workflow lacksid-token: writeunderpermissions.Permission 'iam.serviceAccounts.getAccessToken' denied: theworkloadIdentityUserbinding doesn't match your repo.REPOmust be exactlyowner/name, as in your GitHub URL.- The deploy workflow never starts:
workflow_runworkflows only run once the file is on your default branch, and theworkflows:name must match CI'sname:exactly. - The deploy step fails with a permission error naming a secret: also grant the deployer
roles/secretmanager.secretAccessoron the two secrets, as in Thursday's step 11. - The Streamlit app says "The checker is not responding": check
API_URLhas no trailing slash, then open the address's/docsin your browser to see whether the API is up. - Streamlit fails to install: the file must be named
requirements.txtand sit inui/, next tostreamlit_app.py.
Deliverable checklist¶
- A Console workspace with its own spend limit and alerts, and three separate keys
-
call_claude()on the Anthropic SDK, same signature, tested with a fake HTTP layer - The week 7 CI gate running on
ANTHROPIC_API_KEY -
strip_piiwith Singapore phone numbers and NRICs, and a written list of what it misses - A
checkstable created by an Alembic migration, locally and on Neon -
POST /checkwith tests for200,422and503, LLM and database faked - A Dockerfile, and the API plus Postgres running under docker compose
- The API on Cloud Run, secrets in Secret Manager, at most two instances
- Deploy on merge to
mainthrough Workload Identity Federation (or moved to week 11) - A public Streamlit link that calls the API
- Demo video, five lines, and the Friday check-in sent
If you have time left: the image is over 1 GB partly because it installs Streamlit and pandas, which the API itself doesn't import. Move them into a dependency group that the Dockerfile leaves out, rebuild, and compare the size and the cold-start time.