Mini-Project: a SQL Agent Loop for Reservo

Description

It's time to assemble everything yourself. In this mini-project you'll build your complete SQL agent over Reservo: the run_sql and describe_table tools (with M4's validation and M5's guardrails inside), and the loop's runner that — given a script of the model's tool calls — actually runs the tools, feeds the results back, corrects itself when it fails, and produces the final answer.

And you'll put it to the test with the three questions that cover the whole module:

1. A one-step one: resolved with a single run_sql loop.
2. A two-step one: the agent explores the schema (describe_table) before querying.
3. One that fails and self-corrects: the first SQL uses a hallucinated column, the error goes back to the model, and the second SQL gets it right.

There's no new engineering here: it's the synthesis of the previous seven lessons in a single file that runs end to end. By the end you'll have a working SQL agent — the same one Module 8 will package into the complete safe assistant.

Connection to the module

This closes out Module 6. The lessons gave you the pieces: the cycle (02), tool-calling and its contract (03-04), multi-step (05), when to stop (06), and self-correction (07). The mini-project brings them together into an executable deliverable. It's your SQL agent, ready for Module 7 to measure it and Module 8 to integrate it.

What you're going to build

A Python program, sql_agent.py, with three parts:

   sql_agent.py
   ────────────
   [1] Tools                  run_sql (M4 + M5 inside), describe_table
   [2] The loop's runner       run_agent: while with a cap, tool_use -> tool_result
   [3] Three tests              1 step · 2 steps · fails-and-corrects

The model is concept: its turns come from a script (what claude-sonnet-5 would return). Everything else — the tools, the runner, the SQL — actually runs against Reservo.

Step 0 — Make sure Reservo is seeded

The agent queries reservo.db. If you're coming from the previous lessons you already have it; if not, or if you want to start fresh, work in a temporary directory and run lesson 01's seed:

WORK=$(mktemp -d)      # unique working directory
cd "$WORK"
sqlite3 reservo.db < seed.sql   # lesson 01's seed.sql

Verify it loaded:

sqlite3 reservo.db "SELECT COUNT(*) FROM bookings;"

What to expect:

23

With the 23 bookings (5 rooms, 8 members, 26 payments), your playing field is ready.

Step 1 — The tools (with guardrails)

Create sql_agent.py. Start with the read-only connection (M5) and validation (M4), which the tools reuse:

"""M6 mini-project: a SQL agent loop for Reservo."""
import sqlite3, re, json

# ---------- read-only connection (M5 guardrail) ----------
def open_reservo(path="reservo.db"):
    con = sqlite3.connect(path)
    con.execute("PRAGMA query_only = ON")   # the connection can't write
    return con

# ---------- validation (M4) ----------
def strip_sql(sql):
    sql = re.sub(r"--[^\n]*", "", sql)
    sql = re.sub(r"/\*.*?\*/", "", sql, flags=re.DOTALL)
    return sql.strip()

def validate(sql, con):
    """Validates shape + schema without executing (M4). Returns (ok, error)."""
    clean = strip_sql(sql)
    if not clean:
        return False, "empty query"
    body = clean.rstrip(";").strip()
    if ";" in body:
        return False, "more than one statement"
    first = body.split(None, 1)[0].upper()
    if first not in ("SELECT", "WITH"):
        return False, f"not a read (starts with {first})"
    try:
        con.execute("EXPLAIN " + body).fetchall()
    except sqlite3.Error as e:
        return False, f"{type(e).__name__}: {e}"
    return True, None

Now the two tools. run_sql is the only door to the database: it validates (M4), runs on the read-only connection (M5), applies a row cap (M5), and returns the uniform dictionary. describe_table lets the agent explore the schema:

# ---------- tools ----------
def run_sql(query, con, max_rows=100):
    ok, err = validate(query, con)              # M4: doesn't touch the DB if the SQL is invalid
    if not ok:
        return {"ok": False, "error": err}
    body = strip_sql(query).rstrip(";").strip()
    try:
        cur = con.execute(body)                 # M5: con is read-only
        rows = cur.fetchmany(max_rows + 1)      # M5: row cap
        return {"ok": True, "columns": [d[0] for d in cur.description],
                "rows": [list(r) for r in rows[:max_rows]],
                "truncated": len(rows) > max_rows}
    except sqlite3.Error as e:
        return {"ok": False, "error": f"{type(e).__name__}: {e}"}

def describe_table(name, con):
    tables = [r[0] for r in con.execute(
        "SELECT name FROM sqlite_master WHERE type='table' "
        "AND name NOT LIKE 'sqlite_%' ORDER BY name")]
    if name not in tables:
        return {"ok": False, "error": f"table '{name}' doesn't exist. Tables: {tables}"}
    cols = con.execute(f"PRAGMA table_info({name})").fetchall()
    return {"ok": True, "table": name,
            "columns": [{"name": c[1], "type": c[2]} for c in cols]}

Step 2 — The loop's runner

The agent's heart: the while with a cap that orchestrates the tool_use → tool_result cycle. A registry (make_tools) dispatches any tool the model requests; summarize builds the readable trace:

# ---------- the loop's runner ----------
def make_tools(con):
    return {"run_sql": lambda query: run_sql(query, con),
            "describe_table": lambda name: describe_table(name, con)}

def summarize(result):
    if not result.get("ok"):
        return f"ERROR: {result['error']}"
    if "rows" in result:
        return f"{len(result['rows'])} row(s): {result['rows']}"
    if "columns" in result:
        return f"columns: {[c['name'] for c in result['columns']]}"
    return str(result)

def run_agent(question, con, model_script, max_iters=6):
    tools = make_tools(con)
    messages = [{"role": "user", "content": question}]
    script = list(model_script)
    print(f"Question: {question}")

    for step in range(1, max_iters + 1):
        # [MODEL] concept: the next turn from the script (what claude-sonnet-5 would return)
        turn = script.pop(0) if script else {"type": "text", "text": "I couldn't continue."}

        if turn["type"] == "text":                       # the model answers -> END
            print(f"  [step {step}] the model answers -> END")
            print(f"Answer: {turn['text']}")
            return turn["text"]

        results = []
        for call in turn["tool_calls"]:                  # run each requested tool
            result = tools[call["name"]](**call["input"])
            arg = call["input"].get("query") or call["input"].get("name") or ""
            print(f"  [step {step}] tool_use {call['name']}({arg!r})")
            print(f"            -> {summarize(result)}")
            results.append({"type": "tool_result", "tool_use_id": call["id"],
                            "content": json.dumps(result, ensure_ascii=False)})
        messages.append({"role": "assistant", "content": turn["tool_calls"]})
        messages.append({"role": "user", "content": results})

    print(f"  [cap] {max_iters} iterations -> forced END")
    return "I couldn't solve it within the step cap."

That runner has everything from the module: the cycle (for step), the cap (max_iters), tool-calling (tool_use → tool_result), multi-step (the registry and the for call), and self-correction (the error travels in results with no special logic). Under 30 lines for a complete agent.

Step 3 — The three tests

Now the block that runs the agent with the three questions. Remember: the scripts are concept — what claude-sonnet-5 would have returned; the runner and the tools actually run.

if __name__ == "__main__":
    con = open_reservo()

    print("### Question 1: ONE STEP ###")
    run_agent("How many confirmed bookings are there?", con, [
        {"type": "tool_use", "tool_calls": [
            {"id": "t1", "name": "run_sql",
             "input": {"query": "SELECT COUNT(*) AS n FROM bookings WHERE status = 'confirmed'"}}]},
        {"type": "text", "text": "There are 20 confirmed bookings in Reservo."}])

    print("\n### Question 2: TWO STEPS (explore + query) ###")
    run_agent("How much has member Ana Torres paid in total?", con, [
        {"type": "tool_use", "tool_calls": [
            {"id": "t2a", "name": "describe_table", "input": {"name": "payments"}}]},
        {"type": "tool_use", "tool_calls": [
            {"id": "t2b", "name": "run_sql", "input": {"query": (
                "SELECT SUM(CASE WHEN p.kind='charge' THEN p.amount_cents "
                "ELSE -p.amount_cents END) AS net_cents "
                "FROM payments p JOIN bookings b ON b.id = p.booking_id "
                "JOIN members m ON m.id = b.member_id WHERE m.name = 'Ana Torres'")}}]},
        {"type": "text", "text": "Ana Torres has paid 24000 cents in total ($240.00)."}])

    print("\n### Question 3: FAILS and SELF-CORRECTS ###")
    run_agent("How much did Reservo earn in March?", con, [
        {"type": "tool_use", "tool_calls": [
            {"id": "t3a", "name": "run_sql", "input": {"query": (
                "SELECT SUM(b.total) AS revenue_cents FROM bookings b "
                "WHERE b.status='confirmed' AND b.start_at LIKE '2026-03%'")}}]},
        {"type": "tool_use", "tool_calls": [
            {"id": "t3b", "name": "run_sql", "input": {"query": (
                "SELECT SUM(b.price_cents) AS revenue_cents FROM bookings b "
                "WHERE b.status='confirmed' AND b.start_at LIKE '2026-03%'")}}]},
        {"type": "text", "text": "In March, Reservo earned 55400 cents ($554.00) in confirmed bookings."}])

    con.close()

Run it:

python3 sql_agent.py

What to expect:

### Question 1: ONE STEP ###
Question: How many confirmed bookings are there?
  [step 1] tool_use run_sql("SELECT COUNT(*) AS n FROM bookings WHERE status = 'confirmed'")
            -> 1 row(s): [[20]]
  [step 2] the model answers -> END
Answer: There are 20 confirmed bookings in Reservo.

### Question 2: TWO STEPS (explore + query) ###
Question: How much has member Ana Torres paid in total?
  [step 1] tool_use describe_table('payments')
            -> columns: ['id', 'booking_id', 'amount_cents', 'kind']
  [step 2] tool_use run_sql("SELECT SUM(CASE WHEN p.kind='charge' THEN p.amount_cents ELSE -p.amount_cents END) AS net_cents FROM payments p JOIN bookings b ON b.id = p.booking_id JOIN members m ON m.id = b.member_id WHERE m.name = 'Ana Torres'")
            -> 1 row(s): [[24000]]
  [step 3] the model answers -> END
Answer: Ana Torres has paid 24000 cents in total ($240.00).

### Question 3: FAILS and SELF-CORRECTS ###
Question: How much did Reservo earn in March?
  [step 1] tool_use run_sql("SELECT SUM(b.total) AS revenue_cents FROM bookings b WHERE b.status='confirmed' AND b.start_at LIKE '2026-03%'")
            -> ERROR: OperationalError: no such column: b.total
  [step 2] tool_use run_sql("SELECT SUM(b.price_cents) AS revenue_cents FROM bookings b WHERE b.status='confirmed' AND b.start_at LIKE '2026-03%'")
            -> 1 row(s): [[55400]]
  [step 3] the model answers -> END
Answer: In March, Reservo earned 55400 cents ($554.00) in confirmed bookings.

Your agent runs end to end! Read the three stories:

• Question 1 (one step). One run_sql loop ([[20]], 20 confirmed) and one answer. The simple case.
• Question 2 (two steps). The agent explores first (describe_table('payments') confirms amount_cents and kind exist), and only then writes the correct JOIN that adds charges and subtracts refunds: [[24000]], $240.00.
• Question 3 (self-correction). The first SQL uses b.total (hallucinated column) and fails; the error goes back to the model; the second SQL uses b.price_cents and gets it right: [[55400]], $554.00 confirmed revenue in March. The error didn't end anything: it was the clue that led to the answer.

Three questions, three of the agent's capabilities — resolving directly, exploring, and self-correcting — all in the same 30-line runner.

Your turn: three challenges

Now it's your turn. These challenges extend the mini-project; each has a suggested solution.

Challenge 1: A new one-step question (Easy)

Write a script for "Which room is the most expensive per hour?" with its run_sql (one loop) and its answer. Run it with run_agent.

con = open_reservo()
run_agent("Which room is the most expensive per hour?", con, [
    {"type": "tool_use", "tool_calls": [
        {"id": "r1", "name": "run_sql",
         "input": {"query": "SELECT name, hourly_cents FROM rooms ORDER BY hourly_cents DESC LIMIT 1"}}]},
    {"type": "text", "text": "The most expensive room is Boardroom, at 8000 cents ($80.00) per hour."}])
con.close()

Question: Which room is the most expensive per hour?
  [step 1] tool_use run_sql('SELECT name, hourly_cents FROM rooms ORDER BY hourly_cents DESC LIMIT 1')
            -> 1 row(s): [['Boardroom', 8000]]
  [step 2] the model answers -> END
Answer: The most expensive room is Boardroom, at 8000 cents ($80.00) per hour.

Challenge 2: An agent that explores before querying (Medium)

Write a two-step script for "How many bookings does the Focus room have?" where the model first calls describe_table('bookings') (to confirm the room is referenced by room_id), and then does the JOIN with rooms. Run it.

con = open_reservo()
run_agent("How many bookings does the Focus room have?", con, [
    {"type": "tool_use", "tool_calls": [
        {"id": "r2a", "name": "describe_table", "input": {"name": "bookings"}}]},
    {"type": "tool_use", "tool_calls": [
        {"id": "r2b", "name": "run_sql", "input": {"query": (
            "SELECT COUNT(*) AS n FROM bookings b JOIN rooms r ON r.id = b.room_id "
            "WHERE r.name = 'Focus'")}}]},
    {"type": "text", "text": "The Focus room has 8 bookings."}])
con.close()

Question: How many bookings does the Focus room have?
  [step 1] tool_use describe_table('bookings')
            -> columns: ['id', 'room_id', 'member_id', 'start_at', 'end_at', 'status', 'price_cents']
  [step 2] tool_use run_sql("SELECT COUNT(*) AS n FROM bookings b JOIN rooms r ON r.id = b.room_id WHERE r.name = 'Focus'")
            -> 1 row(s): [[8]]
  [step 3] the model answers -> END
Answer: The Focus room has 8 bookings.

Challenge 3: A self-correction with a hallucinated table (Hard)

Write a three-step script for "How many pro members are there?" where the model first queries a nonexistent table (users), sees the error, corrects to members, and answers. Run it and confirm self-correction works with tables, not just columns.

con = open_reservo()
run_agent("How many pro members are there?", con, [
    {"type": "tool_use", "tool_calls": [
        {"id": "r3a", "name": "run_sql", "input": {"query": "SELECT COUNT(*) AS n FROM users WHERE tier = 'pro'"}}]},
    {"type": "tool_use", "tool_calls": [
        {"id": "r3b", "name": "run_sql", "input": {"query": "SELECT COUNT(*) AS n FROM members WHERE tier = 'pro'"}}]},
    {"type": "text", "text": "Reservo has 4 pro members."}])
con.close()

Question: How many pro members are there?
  [step 1] tool_use run_sql("SELECT COUNT(*) AS n FROM users WHERE tier = 'pro'")
            -> ERROR: OperationalError: no such table: users
  [step 2] tool_use run_sql("SELECT COUNT(*) AS n FROM members WHERE tier = 'pro'")
            -> 1 row(s): [[4]]
  [step 3] the model answers -> END
Answer: Reservo has 4 pro members.

Deliverable

Your Module 6 deliverable is the working agent. Check that you have:

• ✅ reservo.db seeded — 23 bookings (verified with COUNT(*)).
• ✅ sql_agent.py — with the tools (run_sql with M4+M5, describe_table), the runner (run_agent, with cap and self-correction), and the three tests.
• ✅ The one-step question — 20 confirmed bookings, resolved in one run_sql loop.
• ✅ The two-step question — Ana Torres $240.00, with exploration (describe_table) before querying.
• ✅ The self-correcting question — March $554.00, with the hallucinated b.total failing and b.price_cents getting it right.
• ✅ At least one of the challenges — a new question, an exploration, or a table self-correction.

If you have all six, you've completed Module 6: you have a SQL agent that answers questions by querying the database in a cycle, that explores when it doesn't know the schema, that corrects itself when it fails, and that always finishes — thanks to the cap.

Module summary

Close out Module 6 by recalling the map you built:

• An assistant becomes an agent when it closes the linear flow into a cycle: it decides, acts (invokes a tool), observes the result or the error, and decides again — until it answers or hits the cap.
• The engine of the cycle is tool-calling: the model doesn't execute SQL; it requests running run_sql (filling in its input_schema), and your program runs it and hands back the result in a tool_result. That separation lets you put validation (M4) and guardrails (M5) behind a single door.
• The real back-and-forth shape is Claude's Messages API tool_use → tool_result, paired by id (concept with claude-sonnet-5); the runner is that same cycle with the model's turns scripted.
• The agent explores the schema (describe_table/list_tables) when it needs to, stops when the model answers or the iteration cap runs out, and self-corrects when the SQL fails — the error comes back as a tool_result and the model fixes it.
• Finishing isn't being right. The cap guarantees termination, validation prevents broken SQL, and self-correction fixes what breaks; but none of them guarantee the answer is correct. SQL that runs and lies (241400 instead of 211900) passes through the loop without objection. Measuring accuracy is Module 7.

From here on: M7 measures how good your agent is (does the generated SQL return the same thing as the "gold" SQL? execution accuracy, test set, failure modes), and M8 packages everything — context (M2) + prompt (M3) + validation (M4) + guardrails (M5) + this loop (M6) + evaluation (M7) — into Reservo's safe SQL assistant. The agent is yours; now let's measure whether it gets things right.

Additional resources

1. Claude — Tool use (function calling) — The tool_use → tool_result cycle your runner implements; in production, script.pop(0) becomes a client.messages.create(...).
2. Claude — Building agents — How an agent with tools gets orchestrated; the boundary with the AI Engineering ecosystem.
3. SQLite — PRAGMA query_only — M5's read-only guardrail that open_reservo reuses.
4. Chen et al., "Teaching Large Language Models to Self-Debug" — Self-correction inside the loop: handing the error back to the model so it can fix its own code.
5. BIRD: execution accuracy in text-to-SQL — The next step: measuring whether the agent gets it right, not just whether the loop finishes (Module 7).