Sample code for: AI Engineering Skills: Prompt, Context, Harness, Loop

ai-engineering-layers/README.md

@@ -0,0 +1,49 @@
+# AI Engineering Skills: Prompt, Context, Harness, Loop
+
+This folder contains the sample project for the Real Python tutorial [AI Engineering Skills: Prompt, Context, Harness, Loop](https://realpython.com/ai-engineering-layers/).
+
+It's a small Flask checkout API backed by SQLite, plus `loop.py`, a script that hands the failing tests to Claude Code one test at a time. The app is **broken on purpose**: it has four bugs and five failing tests for the agent to fix.
+
+- `app/api.py`: The `POST /checkout` endpoint, which logs every error to `app.log` and returns a 500 status code
+- `app/db.py`: The SQLite schema and item prices
+- `tests/test_checkout.py`: The five tests
+- `loop.py`: The loop that runs `pytest` and calls `claude -p` until the tests pass or it gives up after `MAX_RUNS` runs
+
+## Installation
+
+You need Python 3.12 or later. Create and activate a virtual environment, then install the pinned requirements:
+
+```sh
+$ python -m venv venv/
+$ source venv/bin/activate
+(venv) $ python -m pip install -r requirements.txt
+```
+
+This installs `flask==3.1.3` and `pytest==9.1.1`.
+
+## Run the Tests
+
+From this folder, run:
+
+```sh
+(venv) $ python -m pytest -q
+...
+FAILED tests/test_checkout.py::test_checkout_creates_order - assert 500 == 201
+FAILED tests/test_checkout.py::test_checkout_totals_the_quantity - KeyError: ...
+FAILED tests/test_checkout.py::test_unknown_item_is_rejected - assert 500 == 400
+FAILED tests/test_checkout.py::test_orders_endpoint_lists_orders - assert 404...
+FAILED tests/test_checkout.py::test_order_ids_increment - KeyError: 'id'
+5 failed in 0.45s
+(venv) $ tail -1 app.log
+sqlite3.OperationalError: table orders has no column named created_at
+```
+
+## Run the Loop
+
+Running `loop.py` needs [Claude Code](https://code.claude.com/docs/en/overview) installed and logged in, with `claude` on your `PATH`. The flags were verified with Claude Code 2.1.308. The script lets the agent edit files under `app/`, so work on a copy of this folder:
+
+```sh
+(venv) $ python loop.py
+```
+
+Each run prints how many tests fail and which one it hands over, and the script ends with `All tests pass after N agent run(s).` or `Gave up after 5 runs.` A real agent may fix more than one bug per run, so the number of runs varies from one try to the next.

ai-engineering-layers/app/api.py

@@ -0,0 +1,34 @@
+import logging
+import os
+
+from flask import Flask, jsonify, request
+
+from app import db
+
+logger = logging.getLogger("checkout")
+logger.addHandler(logging.FileHandler("app.log"))
+
+
+def create_app(db_path=None):
+    app = Flask(__name__)
+    path = db_path or os.environ.get("ORDERS_DB", "orders.db")
+
+    @app.post("/checkout")
+    def checkout():
+        data = request.get_json()
+        try:
+            price = db.PRICES[data["item"]]
+            total = price  # TODO: price per unit
+            conn = db.connect(path)
+            cur = conn.execute(
+                "INSERT INTO orders (item, quantity, total, created_at)"
+                " VALUES (?, ?, ?, datetime('now'))",
+                (data["item"], data["quantity"], total),
+            )
+            conn.commit()
+            return jsonify(id=cur.lastrowid, total=total), 201
+        except Exception:
+            logger.exception("checkout failed")
+            return jsonify(error="internal error"), 500
+
+    return app

ai-engineering-layers/app/db.py

@@ -0,0 +1,19 @@
+import sqlite3
+
+SCHEMA = """
+CREATE TABLE IF NOT EXISTS orders (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    item TEXT NOT NULL,
+    quantity INTEGER NOT NULL,
+    total REAL NOT NULL
+);
+"""
+
+PRICES = {"widget": 4.50, "gadget": 12.00}
+
+
+def connect(path):
+    conn = sqlite3.connect(path)
+    conn.row_factory = sqlite3.Row
+    conn.executescript(SCHEMA)
+    return conn

ai-engineering-layers/loop.py

@@ -0,0 +1,40 @@
+import pathlib
+import re
+import subprocess
+
+PROMPT = """Fix this one failing test: {test}
+Edit files under app/ only, never tests/. Leave the other tests alone."""
+MAX_RUNS = 5
+log = pathlib.Path("app.log")
+
+for run in range(MAX_RUNS + 1):
+    log.write_text("")  # Keep only this run's errors
+    tests = subprocess.run(
+        ["python", "-m", "pytest", "-q"], capture_output=True, text=True
+    )
+    if tests.returncode == 0:
+        print(f"All tests pass after {run} agent run(s).", flush=True)
+        break
+    if run == MAX_RUNS:
+        print(f"Gave up after {MAX_RUNS} runs.", flush=True)
+        break
+    failing = re.findall(r"^FAILED (\S+)", tests.stdout, re.M)
+    lines = log.read_text().splitlines()
+    error = lines[-1] if lines else "nothing logged"
+    context = f"{tests.stdout}\n\napp.log shows: {error}"
+    print(
+        f"Run {run + 1}: {len(failing)} failing, handing over {failing[0]}",
+        flush=True,
+    )
+    subprocess.run(
+        [
+            "claude",
+            "-p",
+            f"{PROMPT.format(test=failing[0])}\n\n{context}",
+            "--permission-mode",
+            "manual",
+            "--allowedTools",
+            "Read,Edit,Write",
+        ],
+        check=True,
+    )

ai-engineering-layers/requirements.txt

@@ -0,0 +1,2 @@
+flask==3.1.3
+pytest==9.1.1

ai-engineering-layers/tests/test_checkout.py

@@ -0,0 +1,37 @@
+import pytest
+
+from app.api import create_app
+
+
+@pytest.fixture
+def client(tmp_path):
+    app = create_app(str(tmp_path / "orders.db"))
+    return app.test_client()
+
+
+def test_checkout_creates_order(client):
+    response = client.post("/checkout", json={"item": "widget", "quantity": 1})
+    assert response.status_code == 201
+
+
+def test_checkout_totals_the_quantity(client):
+    response = client.post("/checkout", json={"item": "widget", "quantity": 3})
+    assert response.get_json()["total"] == 13.5
+
+
+def test_unknown_item_is_rejected(client):
+    response = client.post("/checkout", json={"item": "nope", "quantity": 1})
+    assert response.status_code == 400
+
+
+def test_orders_endpoint_lists_orders(client):
+    client.post("/checkout", json={"item": "gadget", "quantity": 2})
+    response = client.get("/orders")
+    assert response.status_code == 200
+    assert response.get_json()[0]["item"] == "gadget"
+
+
+def test_order_ids_increment(client):
+    first = client.post("/checkout", json={"item": "widget", "quantity": 1})
+    second = client.post("/checkout", json={"item": "widget", "quantity": 1})
+    assert second.get_json()["id"] == first.get_json()["id"] + 1