# Database read tools

Web page: https://academy.memberpress.ai/courses/tool-use/l4/

Lesson 4 of 9 in [Tool Use — From Calculator to Agent](https://academy.memberpress.ai/courses/tool-use/), 18 min.

Video: https://www.youtube.com/watch?v=ie236dOJqp8

The model can answer questions about your data — but only if you give it a channel into the database. The channel has to be narrow enough that the model cannot accidentally mutate anything, and transparent enough that when the model generates a bad query you can see it. Building that channel is what this lesson is about.

**Why read-only first**

Write access is earned. A read-only SQL tool has a blast radius of zero — the worst outcome is a slow query, not a corrupted table. Build the read boundary cleanly, then expand permissions deliberately when you can audit the full call record. Many production use cases never need write access at all.

**The tool definition**

The model needs to know what it can query, not how it connects. Give it a `query` parameter and document the schema in the description. Never include the connection string in any field the model can read.

```python
# scripts/db_tool.py
import anthropic, sqlite3, json

DB_PATH = "data/orders.db"     # create this with sqlite3 for the exercise
MAX_ROWS = 20

client = anthropic.Anthropic()

tools = [{
    "name": "query_db",
    "description": (
        "Execute a read-only SQL SELECT query against the orders database. "
        "Tables: orders(id, customer_id, status, total, created_at), "
        "customers(id, name, email). "
        "Returns up to 20 rows. For larger result sets, add a WHERE clause "
        "to narrow the query. Do not use INSERT, UPDATE, DELETE, or DROP."
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "query": {
                "type": "string",
                "description": "A valid SQLite SELECT statement."
            }
        },
        "required": ["query"]
    }
}]
```

**The handler**

```python
def handle_query_db(query: str) -> dict:
    # Block write statements before they reach the database
    normalized = query.strip().upper()
    for banned in ("INSERT", "UPDATE", "DELETE", "DROP", "CREATE", "ALTER"):
        if normalized.startswith(banned) or f" {banned} " in normalized:
            return {"error": f"Write statements are not permitted. Received: {query[:80]}"}

    try:
        conn = sqlite3.connect(DB_PATH)
        conn.row_factory = sqlite3.Row
        cursor = conn.execute(query)
        rows = cursor.fetchmany(MAX_ROWS)
        data = [dict(row) for row in rows]
        conn.close()
        return {
            "rows": data,
            "row_count": len(data),
            "truncated": len(data) == MAX_ROWS
        }
    except sqlite3.OperationalError as e:
        return {"error": str(e), "query": query}
```

Three things this handler does that matter: it blocks write statements before they reach SQLite, it caps the result at `MAX_ROWS`, and it returns errors as dicts rather than raising exceptions. Claude can read a structured error and adapt — if it receives `{"error": "no such table: order"}`, it will correct the table name on the next call. An uncaught exception crashes the loop.

**Handling the round-trip**

```python
def run_db_agent(user_question: str):
    messages = [{"role": "user", "content": user_question}]

    while True:
        r = client.messages.create(
            model="claude-opus-4-5",
            max_tokens=512,
            tools=tools,
            messages=messages
        )
        messages.append({"role": "assistant", "content": r.content})

        if r.stop_reason == "end_turn":
            return r.content[0].text

        tool_results = []
        for block in r.content:
            if block.type == "tool_use":
                result = handle_query_db(**block.input)
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": json.dumps(result)
                })
        messages.append({"role": "user", "content": tool_results})

print(run_db_agent("Which customers placed orders over $500 last month?"))
```

**Handling large result sets**

When `truncated` is `True` in the response, Claude will often follow up with a narrower query. This is the correct behavior. You do not need to handle it specially — just let the loop continue. The pattern of "broad query, see truncation, narrow query" is reliable because the model reads the result dict.

If the model keeps issuing broad queries that return 20 rows without narrowing, add a sentence to the tool description: `"If results are truncated, issue a more specific WHERE clause before summarizing."` One sentence in the description is enough — the model follows it.

**If you get stuck**

*`sqlite3.OperationalError: no such column`:* The model hallucinated a column name. Fix: add the exact column names to the tool description. The model reads them and will use them. A description that says "Tables: orders(id, customer_id, status, total, created_at)" gives the model the exact column list it needs.

*Model issues a SELECT but gets back 0 rows, then asserts "no orders exist":* The query has a logic error — likely a date format mismatch or a wrong join condition. Return the query in the error dict so you can inspect it. `{"error": str(e), "query": query}` gives you the exact SQL in the tool result, which you can then run manually to diagnose.

*Model tries to INSERT despite the ban:* The keyword-match logic above uses `startswith` which catches `INSERT INTO ...` but not `SELECT * FROM (INSERT ...)`. For production, use a proper SQL parser or open the SQLite connection with `conn.execute("PRAGMA query_only = ON")` before running the query. The pragma raises an `OperationalError` on any write attempt at the database level, which is harder to circumvent than string matching.

**Cross-reference**

The agent loop above is simplified — it has no iteration cap and no `dispatch()` abstraction. l7 covers the full dispatch table, loop termination conditions, and iteration caps. Build the read-only tool here and plug it into the full agent architecture in l7.

**Deliverable**

`scripts/db_tool.py` — a SQLite database with at least two tables, the `query_db` tool, the handler with write-blocking, and the agent loop. Run it with two different natural-language questions and confirm the model issues syntactically correct SQL both times. Check that a question phrased to require a JOIN actually produces a JOIN in the emitted query.

## Try these prompts

- Write a tool description for a read-only SQL query tool that documents its table schema inline and instructs the model not to use INSERT, UPDATE, or DELETE.
- Write the Python handler function that blocks write statements before they reach SQLite and returns a structured error dict instead of raising an exception.
- Explain why the handler returns {"error": str(e), "query": query} instead of just {"error": str(e)} and how the model uses the query field.

Next: [External API tools](https://academy.memberpress.ai/courses/tool-use/l5/)