Build an AI-Powered SQL Assistant API

Build a FastAPI backend where an AI agent writes and runs SQL queries.

Introduction

30 Second Summary

Every company has data locked in databases that people want answers from. The most powerful pattern in AI right now is the AI agent: a system that takes a question, decides what action to take, and executes it autonomously.

In this project, you will build a REST API where an AI agent takes natural language questions about sales data, writes the SQL query, runs it against a real database, and returns a structured answer with a plain-English explanation.

What You'll Build

You type a question like "Which region had the highest revenue last quarter?" into FastAPI's auto-generated interactive docs page, hit send, and get back a JSON response containing the SQL the AI wrote, the raw query results, and a plain-English explanation of what the numbers mean.

By the end of this project, you'll have:

  • A live API endpoint where you send any natural-language question and get back the SQL query the AI wrote, the raw results, and a plain-English explanation of what the numbers mean.
  • An AI tool-calling loop running against your SQLite database, where Claude autonomously decides what query to run, executes it, and interprets the results for you.
  • Production guardrails that block dangerous queries and validate all inputs with Pydantic models, plus token usage tracking on every response so you can monitor costs.
  • Secret Mission: Add conversation memory so your agent handles follow-up questions like "Now break that down by product category" without repeating context.

Do I need any prerequisites?

You'll need Python and VS Code installed on your computer. You'll also need a Claude API key from platform.claude.com, which is free to create (you only pay for the tokens you use during the project).

Before We Start

Before We Start...

Before we dive in, let's lock in what you're about to build and why it matters.

Set Up Your Project and Get Your API Key

Your AI SQL assistant needs two things before you can write any code: a Python environment with the right packages installed, and a valid API key to communicate with Claude.

In this step, you'll create an isolated project folder, install the Anthropic SDK and FastAPI, then connect to Claude's API and confirm it responds.

In this step, get ready to:
  • Create a project folder with a Python virtual environment.
  • Install the Anthropic SDK and FastAPI packages.
  • Get a Claude API key and verify it connects successfully.
Create your project folder and install dependencies

A virtual environment keeps your project's packages isolated from the rest of your system. This prevents version conflicts between projects.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal and press Enter to open it.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project folder and move into it by running these commands:
mkdir sql-assistant
cd sql-assistant
  • Create a Python virtual environment by running:
python -m venv venv
  • Activate the virtual environment by running the command for your operating system:

macOS

source venv/bin/activate

Windows

venv\Scripts\activate

You should see (venv) appear at the beginning of your terminal prompt. This confirms the virtual environment is active.

Don't see (venv) in your prompt?

Make sure you ran the activate command for your correct operating system. On Windows, use the PowerShell command, not the macOS one.

If you see a permissions error on Windows, try running PowerShell as Administrator and entering Set-ExecutionPolicy RemoteSigned before activating again.

help me activate my Python virtual environment

  • Open your project folder in VS Code by running:
code .

What does code . do?

The code . command opens VS Code with the current folder as the workspace. The dot means "this folder".

  • In VS Code, create a new file called requirements.txt by clicking File in the top menu bar, then New File.
  • Paste the following content into the file:
anthropic==0.120.2
fastapi[standard]==0.140.0
  • Save the file with Cmd+S (macOS) or Ctrl+S (Windows).

Why pin exact versions?

Pinning to anthropic==0.120.2 and fastapi[standard]==0.140.0 ensures everyone gets the same behaviour. Without version pins, a future release could introduce breaking changes.

✔️ Awesome, I've got everything!

Great. Make sure you saved requirements.txt before continuing.

ⓧ I'd like to double check the full code

Your requirements.txt should contain exactly these two lines:

anthropic==0.120.2
fastapi[standard]==0.140.0
  • Back in your terminal (with (venv) still showing), install the dependencies by running:
pip install -r requirements.txt

You'll see pip downloading and installing both packages along with their dependencies. This may take a minute.

  • Verify both packages installed correctly by running:
python -c "import anthropic; import fastapi; print('Setup complete')"

You should see Setup complete printed with no errors.

Seeing ModuleNotFoundError?

Make sure your virtual environment is still active. You should see (venv) in your terminal prompt. If it disappeared, re-activate it with the platform-specific command above, then run pip install -r requirements.txt again.

help me fix ModuleNotFoundError after installing packages

Get your Claude API key

Claude is an AI model made by Anthropic. To use it in your code, you need an API key that authenticates your requests.

  • Open platform.claude.com in your browser.
  • Sign in or create a free account if you don't have one yet.
  • Navigate to Settings, then API keys.
  • Click Create key and give it a name like sql-assistant.
  • Copy the key immediately. It starts with sk-ant- and is only shown once.

What does an API key do?

An API key is like a password that identifies your application to Claude's servers. Each request you send includes this key so Anthropic knows who is making the call and can track usage for billing.

Claude charges per token (roughly per word). A typical question in this project costs less than $0.01.

Now you need to store the key as an environment variable so the Anthropic SDK can find it automatically.

macOS

  • Set the environment variable by running this in your terminal, replacing the placeholder with your actual key:
export ANTHROPIC_API_KEY="your-api-key-here"

This sets the variable for your current terminal session. The SDK reads ANTHROPIC_API_KEY automatically.

Windows

  • Set the environment variable permanently by running this in your terminal, replacing the placeholder with your actual key:
setx ANTHROPIC_API_KEY "your-api-key-here"
  • Close your current terminal completely.
  • Open a new terminal in VS Code by clicking Terminal in the top menu, then New Terminal.
  • Re-activate your virtual environment by running:
cd ~/Desktop/sql-assistant
venv\Scripts\activate

Why open a new terminal?

The setx command saves the variable permanently, but it only takes effect in terminals opened after you run it. Your current terminal still uses the old environment.

Verify your API key works

The packages are installed and the key is set. Now confirm the full chain works end to end.

Before you run the next command, what do you expect to see if everything is configured correctly?

  • Test your connection to Claude by running:
python -c "from anthropic import Anthropic; client = Anthropic(); msg = client.messages.create(model='claude-sonnet-5', max_tokens=128, messages=[{'role': 'user', 'content': 'Say hello in one sentence.'}]); print(msg.content[0].text)"

You should see a short greeting from Claude printed in your terminal. The exact wording varies, but any response confirms your API key is working.

Seeing an authentication error?

If you see a message about an invalid API key, double-check that your key starts with sk-ant- and that you copied the full key without any trailing spaces.

On Windows, make sure you opened a new terminal after running setx. The variable is not available in the same terminal where you set it.

On macOS, confirm the variable is set by running echo $ANTHROPIC_API_KEY and checking that it prints your key.

help me fix my Claude API key authentication error

Your environment is ready and Claude is responding. Next up, you'll build a sales database and create your first API endpoint.

Build a Sales Database and Your First API Endpoint

Your environment is set up and your API key is working. Now you need something for the AI to actually query.

In this step, you will create a SQLite database filled with sales data and build your first FastAPI endpoint. Getting the database and API running first means you can prove everything works before adding AI complexity in the next step.

In this step, get ready to:
  • Create a Python script that builds and seeds a SQLite database with sales records.
  • Create a FastAPI app with a health-check endpoint.
  • Run the development server and test your endpoint in the browser.
Create the database setup script

Your AI agent will query a sales database with information about revenue across different regions, product categories, and quarters. You need a script that creates this database and fills it with sample data.

  • In VS Code, create a new file called database.py inside your project folder.
  • Add the imports and connection helper by pasting this code:
import sqlite3
from pathlib import Path

# Path to the SQLite database file
DB_PATH = Path("sales.db")


def get_connection():
    """Create and return a connection to the SQLite database."""
    return sqlite3.connect(DB_PATH)

What does this code do?

This sets up the foundation for your database module. DB_PATH defines where the database file lives on disk. get_connection() is a helper that other functions will call whenever they need to talk to the database.

  • Below get_connection(), add the init_database() function that creates the table and seeds it with sample data:
def init_database():
    """Create the sales table and seed it with sample data if empty."""
    conn = get_connection()
    cursor = conn.cursor()

    # Create the sales table if it doesn't already exist
    cursor.execute("""
        CREATE TABLE IF NOT EXISTS sales (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            region TEXT NOT NULL,
            product_category TEXT NOT NULL,
            quarter TEXT NOT NULL,
            revenue REAL NOT NULL,
            units_sold INTEGER NOT NULL
        )
    """)

    # Only insert sample data if the table is empty
    cursor.execute("SELECT COUNT(*) FROM sales")
    if cursor.fetchone()[0] == 0:
        sample_data = [
            ("North", "Electronics", "Q1 2025", 150000.00, 1200),
            ("North", "Electronics", "Q2 2025", 175000.00, 1400),
            ("North", "Clothing", "Q1 2025", 80000.00, 3200),
            ("North", "Clothing", "Q2 2025", 95000.00, 3800),
            ("South", "Electronics", "Q1 2025", 120000.00, 960),
            ("South", "Electronics", "Q2 2025", 140000.00, 1120),
            ("South", "Clothing", "Q1 2025", 65000.00, 2600),
            ("South", "Clothing", "Q2 2025", 72000.00, 2880),
            ("East", "Electronics", "Q1 2025", 200000.00, 1600),
            ("East", "Electronics", "Q2 2025", 220000.00, 1760),
            ("East", "Clothing", "Q1 2025", 110000.00, 4400),
            ("East", "Clothing", "Q2 2025", 125000.00, 5000),
            ("West", "Electronics", "Q1 2025", 180000.00, 1440),
            ("West", "Electronics", "Q2 2025", 195000.00, 1560),
            ("West", "Clothing", "Q1 2025", 90000.00, 3600),
            ("West", "Clothing", "Q2 2025", 105000.00, 4200),
        ]
        cursor.executemany(
            "INSERT INTO sales (region, product_category, quarter, revenue, units_sold) VALUES (?, ?, ?, ?, ?)",
            sample_data,
        )
        conn.commit()

    conn.close()

What does this code do?

This function does two things. First, it creates a sales table with columns for region, product category, quarter, revenue, and units sold. Second, it checks whether the table is empty and inserts 16 rows of sample data if so.

The IF NOT EXISTS clause and the count check make this function safe to call multiple times. It will not duplicate data or crash if the table already exists.

  • Below init_database(), add the query_database() function and a main block to run the script directly:
def query_database(sql: str) -> list[dict]:
    """Execute a SQL query and return results as a list of dictionaries."""
    conn = get_connection()
    conn.row_factory = sqlite3.Row
    cursor = conn.cursor()
    cursor.execute(sql)
    rows = cursor.fetchall()
    results = [dict(row) for row in rows]
    conn.close()
    return results


if __name__ == "__main__":
    init_database()
    print("Database initialized with sample data.")

What does this code do?

The query_database() function takes a SQL string, runs it, and returns the results as a list of dictionaries. Setting row_factory = sqlite3.Row means each row comes back with column names as keys instead of anonymous tuples.

The if __name__ == "__main__" block lets you run this file directly to create and seed the database.

  • Save database.py.
  • Initialize the database by running this command in your terminal:
python database.py

You should see Database initialized with sample data. printed in your terminal.

Seeing a ModuleNotFoundError?

Make sure your virtual environment is activated. You should see the environment name in parentheses at the start of your terminal prompt. If not, activate it again with source venv/bin/activate (macOS/Linux) or venv\Scripts\activate (Windows).

Help me fix my database.py script

  • Confirm the database file was created by running:
ls

You should see sales.db listed alongside database.py and requirements.txt.

✔️ Awesome, I've got everything!

Great. Your database is ready with 16 rows of sales data across 4 regions and 2 product categories.

ⓧ I'd like to double check the full code

Scroll up and confirm your database.py has these three sections in order:

  • Imports, DB_PATH, and get_connection() at the top.
  • The init_database() function with the CREATE TABLE statement and 16 sample data tuples.
  • The query_database() function and the if __name__ == "__main__" block at the bottom.
Create your FastAPI app

Now you need an API server that can receive HTTP requests. FastAPI gives you automatic interactive documentation at /docs where you can test endpoints directly in your browser.

  • Create a new file called main.py in your project folder.
  • Paste this code to set up the FastAPI app with a health-check endpoint:
import logging
import time

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

from database import init_database

# Set up request logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Create the FastAPI application
app = FastAPI(
    title="AI SQL Assistant API",
    description="An AI-powered REST API that answers natural language questions about a sales database using Claude's tool-calling capabilities.",
    version="1.0.0",
)

# Initialize the database when the app starts
init_database()


@app.get("/health")
def health_check():
    """Simple endpoint to verify the API is running."""
    return {"status": "healthy"}

What does this code do?

This creates a FastAPI application and calls init_database() so the sales database is ready the moment the server starts. The /health endpoint returns a simple JSON object to confirm the API is alive.

The logging and time imports are set up now for later use when you add request tracking. HTTPException and BaseModel will be used when you add the AI endpoint in a later step.

  • Save main.py.

✔️ Awesome, I've got everything!

Great. Make sure you have saved main.py before moving on to run the server.

ⓧ I'd like to double check the full code

import logging
import time

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

from database import init_database

# Set up request logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Create the FastAPI application
app = FastAPI(
    title="AI SQL Assistant API",
    description="An AI-powered REST API that answers natural language questions about a sales database using Claude's tool-calling capabilities.",
    version="1.0.0",
)

# Initialize the database when the app starts
init_database()


@app.get("/health")
def health_check():
    """Simple endpoint to verify the API is running."""
    return {"status": "healthy"}
Run the development server and test

Time to see your API in action. FastAPI includes a development server with auto-reload, so any changes you save will take effect immediately.

  • Start the FastAPI development server by running this command:
fastapi dev main.py

What does this command do?

This starts a local development server at http://127.0.0.1:8000 with auto-reload enabled. The server watches your files and restarts automatically when you save changes. It will keep running in this terminal until you press Ctrl+C to stop it.

You should see output indicating the server is running, including the URL http://127.0.0.1:8000.

Server not starting?

Make sure your virtual environment is activated and you are in the project folder containing main.py. If you see a port-already-in-use error, another process is using port 8000. Close it or use a different terminal.

Help me get my FastAPI server running

  • Open your browser and go to http://127.0.0.1:8000/docs.

You should see the automatic interactive API documentation page (Swagger UI) with your /health endpoint listed.

  • Click on the GET /health row to expand it.
  • Click the Try it out button on the right.
  • Click Execute.

Before you look at the response, what do you expect to see returned from your /health endpoint?

In the response body below the button, you should see:

{"status": "healthy"}

That confirms your FastAPI server is running and your database was initialized without errors.

Seeing a 500 Internal Server Error?

Check the terminal where your server is running. If you see an error about database.py, make sure the file is saved and in the same folder as main.py. Stop the server with Ctrl+C and restart it with fastapi dev main.py.

Help me fix my /health endpoint

Your database is seeded and your API is live. Next up, you will connect Claude to this database and watch it autonomously write and execute SQL queries.

Connect Claude and Make It Query Your Database

Your database is ready and your API runs. Now it is time to build the core intelligence that makes this project an AI agent rather than a simple chatbot.

The pattern you are about to implement is called tool calling. You define a function (your SQL query), describe it to Claude in a schema, and let Claude autonomously decide when and how to call it. Claude writes the SQL, you execute it, and feed the results back. Claude then interprets the data and writes a human-friendly answer.

In this step, get ready to:
  • Create the agent file with a system prompt describing your database schema.
  • Define the tool schema and build the agentic loop that sends questions to Claude, handles tool calls, and returns answers.
  • Test the full loop with a real question and see Claude generate SQL, execute it, and explain the results.
Set up the agent file with a system prompt and tool schema

The agent needs two things before it can work: a system prompt that teaches Claude about your database, and a tool schema that tells Claude what function it can call.

  • Create a new file called agent.py in your project folder (the same folder where database.py lives) and paste the following code:
import json
import os
import time
import anthropic

from database import query_database

SYSTEM_PROMPT = """You are a data analyst assistant. You have access to a sales database with the following schema:

Table: sales
Columns:
- id (INTEGER, primary key)
- region (TEXT): North, South, East, West
- product_category (TEXT): Electronics, Clothing
- quarter (TEXT): Q1 2025, Q2 2025
- revenue (REAL): revenue in dollars
- units_sold (INTEGER): number of units sold

When asked a question about the data, use the query_database tool to run a SQL query and then provide a clear, concise answer based on the results. Always use SQLite-compatible SQL syntax."""

What does this code do?

The SYSTEM_PROMPT is a detailed description of your database that Claude reads before answering any question. It lists every column, its data type, and the possible values. The more precise this prompt is, the better SQL Claude writes.

The import from database import query_database pulls in the function you built in the previous step so the agent can execute SQL queries directly.

  • Add the tool schema and a SQL safety check below your SYSTEM_PROMPT in agent.py:
TOOLS = [
    {
        "name": "query_database",
        "description": "Execute a read-only SQL query against the sales database and return the results as a list of rows. Only SELECT statements are allowed.",
        "input_schema": {
            "type": "object",
            "properties": {
                "sql": {
                    "type": "string",
                    "description": "A valid SQLite SELECT query to execute against the sales database.",
                }
            },
            "required": ["sql"],
        },
    }
]

BLOCKED_KEYWORDS = ["INSERT", "UPDATE", "DELETE", "DROP", "ALTER", "CREATE", "TRUNCATE"]


def validate_sql(sql: str) -> str | None:
    sql_upper = sql.upper().strip()
    for keyword in BLOCKED_KEYWORDS:
        if keyword in sql_upper:
            return f"Blocked: SQL contains disallowed keyword '{keyword}'. Only SELECT queries are permitted."
    return None

What does this code do?

TOOLS is a JSON schema that tells Claude exactly what function it can call and what parameters it needs. Claude reads this schema and decides on its own when to use the tool. This is the contract between your code and the AI.

validate_sql is a guardrail that blocks any query containing write operations like DELETE or DROP. Even though Claude is instructed to only use SELECT, defence in depth means you verify before executing.

  • Save agent.py.
  • Confirm the safety check works by running this command:
python -c "from agent import validate_sql; print(validate_sql('DELETE FROM sales'))"

You should see the blocked message printed: Blocked: SQL contains disallowed keyword 'DELETE'. Only SELECT queries are permitted.

Seeing an import error?

Make sure your virtual environment is activated (you should see (venv) in your terminal prompt). Also check that agent.py is saved in the same folder as database.py.

help me debug an import error in my agent.py file

Build the agent loop

The agent loop is where the magic happens. It sends a question to Claude, checks if Claude wants to call a tool, executes the tool, feeds the result back, and repeats until Claude has a final answer.

  • Add the run_agent function below validate_sql in agent.py. Start with the function signature and the API call setup:
def run_agent(question: str, messages: list | None = None) -> dict:
    client = anthropic.Anthropic()

    if messages is None:
        messages = []

    messages.append({"role": "user", "content": question})

    total_input_tokens = 0
    total_output_tokens = 0
    generated_sql = None
    query_results = None
    start_time = time.time()

    while True:
        response = client.messages.create(
            model="claude-sonnet-5",
            max_tokens=1024,
            system=SYSTEM_PROMPT,
            tools=TOOLS,
            messages=messages,
        )

        total_input_tokens += response.usage.input_tokens
        total_output_tokens += response.usage.output_tokens

        if response.stop_reason == "tool_use":
            tool_use_block = next(
                block for block in response.content if block.type == "tool_use"
            )

What is happening in this code?

The while True loop is what makes this an agentic pattern. Claude might need multiple tool calls to answer one question. The loop keeps running until Claude stops asking for tools and gives a final text answer.

When response.stop_reason == "tool_use", that means Claude is requesting to call your function. You extract the tool call details and will handle them in the next section.

  • Continue the function by adding the tool handling logic directly below the last line you just typed. This code extracts what Claude wants to do and validates the SQL before executing:
            tool_name = tool_use_block.name
            tool_input = tool_use_block.input
            tool_use_id = tool_use_block.id

            messages.append({"role": "assistant", "content": response.content})

            if tool_name == "query_database":
                sql = tool_input["sql"]
                generated_sql = sql

                validation_error = validate_sql(sql)
                if validation_error:
                    messages.append(
                        {
                            "role": "user",
                            "content": [
                                {
                                    "type": "tool_result",
                                    "tool_use_id": tool_use_id,
                                    "content": validation_error,
                                    "is_error": True,
                                }
                            ],
                        }
                    )
  • Continue adding code directly below. This handles the early return for blocked queries and the SQL execution for valid ones:
                    return {
                        "question": question,
                        "query": sql,
                        "results": [],
                        "answer": validation_error,
                        "tokens_used": {
                            "input": total_input_tokens,
                            "output": total_output_tokens,
                        },
                        "response_time_ms": int((time.time() - start_time) * 1000),
                        "error": True,
                    }

                try:
                    query_results = query_database(sql)
                    tool_result_content = json.dumps(query_results)
                except Exception as e:
                    tool_result_content = f"Error executing query: {str(e)}"

Why send tool results back to Claude?

The tool_result message is what closes the loop. Claude asked to run a query, you executed it, and now you send the data back. Claude then interprets the raw numbers and writes a human-friendly explanation. Without this step, Claude would never see the query results.

  • Add the final section of the function. This sends the tool result back to Claude and handles the case where Claude gives its final text answer:
                messages.append(
                    {
                        "role": "user",
                        "content": [
                            {
                                "type": "tool_result",
                                "tool_use_id": tool_use_id,
                                "content": tool_result_content,
                            }
                        ],
                    }
                )
        else:
            answer = ""
            for block in response.content:
                if hasattr(block, "text"):
                    answer += block.text

            return {
                "question": question,
                "query": generated_sql,
                "results": query_results,
                "answer": answer,
                "tokens_used": {
                    "input": total_input_tokens,
                    "output": total_output_tokens,
                },
                "response_time_ms": int((time.time() - start_time) * 1000),
                "error": False,
            }

How does the loop end?

The else branch runs when response.stop_reason is NOT tool_use. That means Claude has finished using tools and is ready to give its final answer. The function extracts the text and returns a dictionary with everything: the original question, the SQL it generated, the raw query results, the explanation, token usage, and timing.

  • Save agent.py.
  • Verify the function is importable by running:
python -c "from agent import run_agent; print('Import OK')"

You should see Import OK printed. If you see a syntax error, double check your indentation matches the code above.

Seeing a syntax error on import?

The most common cause is mismatched indentation. The while True: loop is indented 4 spaces inside the function. The if response.stop_reason branch is indented 8 spaces. The else: must align with the if at 8 spaces.

help me fix a syntax error in my run_agent function

Test the full agent loop

The function is built. Now you need a way to run it from the command line to confirm the entire cycle works: question goes in, Claude generates SQL, the query runs against your database, and you get back a plain-English explanation.

  • Add a test block at the very bottom of agent.py (below the run_agent function, with no indentation):


if __name__ == "__main__":
    from database import init_database

    init_database()
    result = run_agent("What are the total sales by region?")
    print(f"SQL: {result['query']}")
    print(f"Answer: {result['answer']}")
    print(f"Tokens: {result['tokens_used']}")

What does this test do?

The if __name__ == "__main__" block only runs when you execute the file directly (not when it is imported by another file). It initializes the database, asks a real question, and prints the SQL Claude generated, its interpretation of the results, and how many tokens were used.

  • Save agent.py.

Before you run this, what do you think will happen? Will Claude write a simple SELECT with GROUP BY, or something more complex?

  • Run the agent test by executing:
python agent.py

You should see output similar to this (the exact wording from Claude may vary):

SQL: SELECT region, SUM(revenue) as total_revenue FROM sales GROUP BY region
Answer: Here are the total sales by region: East: $655,000, North: $500,000...
Tokens: {'input': 512, 'output': 198}

The exact numbers in your output depend on your sample data, and the token counts will vary slightly. The important thing is that you see a SQL query, a natural language answer with real numbers from your database, and a tokens count.

Seeing an authentication error?

If you see a message about an invalid API key, your ANTHROPIC_API_KEY environment variable is not set in this terminal session. On Windows, if you used setx earlier, you need to open a new terminal window for it to take effect. Reactivate your virtual environment in the new terminal before running again.

help me fix my Claude API authentication error

Why this approach over a framework like LangChain?

Building the tool-calling loop from scratch in about 50 lines teaches you the exact pattern that LangChain and CrewAI abstract away. In interviews, you will be asked how agents work under the hood. Understanding this loop (send message, detect tool_use, execute, feed result back, repeat) is the transferable skill.

Once you understand the raw loop, picking up any agent framework becomes trivial because they all implement this same cycle.

✔️ Awesome, I've got everything!

Great. Double check you have saved agent.py and that your test printed a SQL query and a natural language answer.

ⓧ I'd like to double check the full code

Your agent.py file should contain these sections in order. Scroll through your file and confirm each is present:

  • Imports (json, os, time, anthropic, from database import query_database).
  • SYSTEM_PROMPT with the full database schema description.
  • TOOLS list with one tool (query_database) and its input_schema.
  • BLOCKED_KEYWORDS list and validate_sql() function.
  • run_agent() function with the while True loop, tool_use handling, validate_sql check, try/except for query execution, and the else branch returning the final answer.
  • if __name__ == "__main__" block that runs a test question.

Your AI agent works end-to-end. It takes a plain English question, lets Claude decide what SQL to write, executes it, and explains the results. Next up, you will wrap this agent in a proper REST API endpoint so any service or frontend can call it over HTTP.

Expose the Agent as a REST API with Proper Validation

Your agent works as a Python script, but a script is not a backend service. No frontend, mobile app, or other microservice can call a script that only runs locally in your terminal.

In this step, you will wrap the agent in a proper POST endpoint with Pydantic request and response models. This transforms your script into a production-quality API that validates inputs, returns structured JSON, and handles errors gracefully.

In this step, get ready to:
  • Define Pydantic models that validate every request and shape every response.
  • Create a POST /ask endpoint that runs the agent loop and returns structured JSON.
  • Test the endpoint in the interactive Swagger docs at /docs.
Define Pydantic request and response models

Pydantic models act as contracts for your API. They define exactly what data comes in and what goes out, rejecting anything that does not match the expected shape.

  • In main.py, find the line init_database() and add the following model definitions directly below it:


class AskRequest(BaseModel):
    # The natural language question from the user
    question: str
    conversation_id: str | None = None


class TokenUsage(BaseModel):
    input: int
    output: int


class AskResponse(BaseModel):
    question: str
    query: str | None
    results: list | None
    answer: str
    tokens_used: TokenUsage
    response_time_ms: int

What do these models do?

  • AskRequest defines what the client sends. A question string is required. The conversation_id is optional (for the Secret Mission later).
  • TokenUsage is a nested model tracking how many input and output tokens each request consumed.
  • AskResponse defines the exact JSON shape clients receive. FastAPI uses this to validate the response before sending it.
  • Save main.py.
  • Start the dev server by running this command:
fastapi dev main.py
  • You should see Application startup complete in the terminal with no errors. This confirms your models are valid Python.

Seeing a syntax error?

  • Check that your model classes are placed AFTER the init_database() call and BEFORE the @app.get("/health") decorator.
  • Make sure each class inherits from BaseModel (already imported at the top of the file).

Why won't my FastAPI server start?

Create the POST /ask endpoint with error handling

The endpoint ties everything together. It validates the incoming request, runs the agent loop, catches failures, and returns a structured response.

  • In main.py, find the health_check() function. Add the first part of your /ask endpoint directly below it:


@app.post("/ask", response_model=AskResponse)
def ask_question(request: AskRequest):
    logger.info(f"Question received: {request.question}")
    start_time = time.time()

    # Load conversation history if continuing a conversation
    messages = None
    if request.conversation_id:
        messages = conversations.get(request.conversation_id, [])

    # Call the agent and handle API failures
    try:
        result = run_agent(request.question, messages=messages)
    except Exception as e:
        logger.error(f"Agent error: {str(e)}")
        raise HTTPException(status_code=500, detail=f"Agent error: {str(e)}")

    # Return 400 if the agent blocked the SQL query
    if result.get("error"):
        raise HTTPException(status_code=400, detail=result["answer"])

What does this code do?

  • The @app.post("/ask") decorator registers this function as a POST endpoint. The response_model tells FastAPI to validate the output against AskResponse.
  • The try/except catches cases where the Claude API times out or fails. It returns an HTTP 500 so the client knows the server had an internal error.
  • The result.get("error") check catches blocked SQL queries from your validate_sql() function and returns an HTTP 400 (bad request).
  • Continue the ask_question function by adding this code directly below the error check you just wrote:

    # Store updated conversation history for follow-up questions
    if request.conversation_id and messages is not None:
        conversations[request.conversation_id] = messages

    logger.info(
        f"SQL: {result['query']} | Time: {result['response_time_ms']}ms | Tokens: {result['tokens_used']}"
    )

    return AskResponse(
        question=result["question"],
        query=result["query"],
        results=result["results"],
        answer=result["answer"],
        tokens_used=TokenUsage(**result["tokens_used"]),
        response_time_ms=result["response_time_ms"],
    )

What does this section do?

  • The conversation storage saves the message history so follow-up questions can reference previous context (you will use this in the Secret Mission).
  • The logger.info line logs the generated SQL, response time, and token count to the console for every request.
  • The return statement maps the agent's raw dictionary into your validated AskResponse model. FastAPI converts this to JSON automatically.
  • Save main.py. The dev server auto-reloads. Check your terminal for Application startup complete with no errors.

Server showing an error after saving?

  • Check that the return AskResponse(...) block is indented inside the ask_question function (one level of indentation).
  • Make sure you did not accidentally place the second block of code outside the function. Every line after def ask_question should be indented with 4 spaces.
  • Confirm that TokenUsage and AskResponse are defined ABOVE the endpoint. Python reads top-to-bottom, so models must exist before the function references them.

Help me fix my FastAPI endpoint error.

Test the endpoint in Swagger UI

FastAPI auto-generates interactive API documentation where you can send real requests without needing a separate tool like Postman.

Before you test, what do you think will happen when you send a natural language question to the endpoint? Will the response include the actual SQL that Claude generated?

  • Open your browser and go to http://127.0.0.1:8000/docs.
  • Click on the POST /ask section to expand it.
  • Click Try it out in the top right of the expanded section.
  • Replace the request body with this JSON:
{
  "question": "Which product category generated the most revenue?"
}
  • Click Execute.

You should see a JSON response containing a query field with the SQL that Claude generated, a results field with the raw data, and an answer field with a plain-English explanation using real numbers from your database.

What should I see?

The response code should be 200. The JSON body includes all the fields from your AskResponse model. The answer field will reference Electronics as the top revenue category based on your sample data.

Check your terminal too. You will see the logged SQL query, response time, and token count printed by the logger.info line.

Getting a 500 error or timeout?

  • Check that your ANTHROPIC_API_KEY environment variable is set in the terminal where the server is running. If you used setx earlier, you may need to restart your terminal.
  • Confirm that sales.db exists in your project folder. The init_database() call at the top of main.py should create it automatically.
  • If the response takes longer than 30 seconds, the Claude API may be under load. Wait a moment and try again.

Why is my /ask endpoint returning a 500 error?

✔️ Awesome, I've got everything!

Great. Double check you have saved main.py and the server is running without errors.

ⓧ I'd like to double check the full code

Scroll up and confirm your main.py contains these sections in order:

  • Imports (logging, time, FastAPI, HTTPException, BaseModel, init_database, run_agent).
  • Logger setup, FastAPI app declaration, conversations dict, and init_database() call.
  • Three Pydantic model classes: AskRequest, TokenUsage, AskResponse.
  • The GET /health endpoint (health_check function).
  • The POST /ask endpoint (ask_question function) with try/except, error check, conversation storage, logging, and AskResponse return.

Your agent is now a real REST API. Any client that can send HTTP requests can ask it questions and get structured JSON answers. Next up, you will add production guardrails to prevent dangerous queries and track token costs.

Add Guardrails and Make It Production-Ready

Your agent can generate and run SQL queries autonomously. But what stops someone from asking "delete all the data"? A raw agent that executes any SQL it writes is a security risk.

Production agents need safety boundaries. In this step, you'll verify the three guardrails already built into your code: SQL validation that blocks write operations, token tracking that reports API costs, and request logging that records every interaction to the console.

In this step, get ready to:
  • Verify SQL validation blocks dangerous write operations.
  • Confirm token tracking reports API usage per request.
  • Confirm request logging captures questions, queries, and timing to the console.
Test the SQL validation guardrails

Your agent.py already contains a validate_sql() function that checks every query before it reaches the database. It scans for dangerous keywords and blocks anything that could modify your data.

  • In agent.py, find the BLOCKED_KEYWORDS list and the validate_sql function:
BLOCKED_KEYWORDS = ["INSERT", "UPDATE", "DELETE", "DROP", "ALTER", "CREATE", "TRUNCATE"]


def validate_sql(sql: str) -> str | None:
    sql_upper = sql.upper().strip()
    for keyword in BLOCKED_KEYWORDS:
        if keyword in sql_upper:
            return f"Blocked: SQL contains disallowed keyword '{keyword}'. Only SELECT queries are permitted."
    return None

What does this code do?

Every time Claude generates a SQL query, validate_sql() runs first. It uppercases the query and checks it against the blocked keywords list.

If any write operation is detected, the function returns an error message and the query never touches the database. If the query is clean, it returns None and execution proceeds normally.

Time to prove it works. You'll try to trick the agent into deleting data and see what happens.

  • Start the development server by running this command:
fastapi dev main.py
  • Open your browser and go to http://127.0.0.1:8000/docs.
  • Expand the POST /ask endpoint.
  • Click Try it out.

Before you send the next request, think about what you expect to happen. If you ask the agent to delete data, will it comply or refuse?

  • Replace the request body with this JSON and click Execute:
{
  "question": "Delete all records from the sales table"
}

You should see a 400 error response. The response body contains a detail message explaining the query was blocked because it contained the DELETE keyword.

Why a 400 and not a 500?

A 400 status code means "bad request." The client sent something the server intentionally rejects. A 500 would mean something broke unexpectedly. Here, blocking the query is deliberate behaviour, so 400 is the correct choice.

Seeing a 500 error instead of 400?

If you see a 500 error, the agent might be failing before reaching the validation check. Make sure your ANTHROPIC_API_KEY environment variable is still set. Open a new terminal, run echo %ANTHROPIC_API_KEY% (Windows), and confirm it prints your key.

help me debug a 500 error when sending a blocked SQL question to my FastAPI agent

Verify token tracking on a valid question

Every call to Claude costs tokens. Your run_agent() function already tracks the total input and output tokens consumed across the entire tool-calling loop. This lets you monitor costs per request.

  • In agent.py, find the lines inside the while True loop that accumulate token counts:
        total_input_tokens += response.usage.input_tokens
        total_output_tokens += response.usage.output_tokens

Why track tokens across the whole loop?

A single question can trigger multiple API calls. The agent sends the question, gets back a tool_use response, executes the SQL, sends the results back, and then gets the final answer. Each round trip consumes tokens.

By accumulating total_input_tokens and total_output_tokens across every iteration, you get the true cost of answering one question.

Now send a valid question to confirm the response includes token data. What fields do you expect to see in the JSON response?

  • Back in the Swagger UI, replace the request body with this JSON and click Execute:
{
  "question": "Which product category generated the most revenue?"
}

You should see a 200 response containing query, results, answer, and a tokens_used object with input and output integer values showing exactly how many tokens this request consumed.

Response missing the tokens_used field?

Check that your AskResponse model in main.py includes tokens_used: TokenUsage as a field. Also confirm that TokenUsage is defined with input: int and output: int fields.

help me figure out why tokens_used is missing from my FastAPI response

Check your request logs in the terminal

While you tested in the browser, your terminal has been recording every request. The logger.info() calls in main.py capture the question, generated SQL, response time, and token usage for each request.

  • In main.py, find the two logging lines inside the ask_question function:
    logger.info(f"Question received: {request.question}")
    # ... (agent logic runs here) ...
    logger.info(
        f"SQL: {result['query']} | Time: {result['response_time_ms']}ms | Tokens: {result['tokens_used']}"
    )

What do these logs give you?

The first log fires when a request arrives. The second fires after the agent finishes, recording what SQL it generated, how long the entire round trip took, and the token count. In production, these logs help you spot slow queries, track costs, and debug unexpected behaviour.

  • Switch to the terminal where your server is running.
  • Look for log lines from both requests you just sent.

You should see a line showing "Question received: Delete all records from the sales table" followed by a line for the valid question that includes the SQL query, response time in milliseconds, and the token counts.

Don't see any log lines?

Make sure main.py has logging.basicConfig(level=logging.INFO) near the top of the file. Without this line, Python's logging module suppresses INFO-level messages by default.

help me figure out why my FastAPI logging isn't printing to the console

✔️ Awesome, I've got everything!

Great. You have confirmed all three production guardrails are working: SQL validation blocks write operations, token tracking reports usage, and request logging records every interaction.

ⓧ I'd like to double check the full code

Scroll up and confirm these pieces are present in your files, in order:

  • In agent.py: the BLOCKED_KEYWORDS list and validate_sql() function above run_agent().
  • In agent.py: the total_input_tokens += response.usage.input_tokens lines inside the while True loop.
  • In main.py: logging.basicConfig(level=logging.INFO) at the top and two logger.info() calls inside ask_question().
  • In main.py: the if result.get("error"): check that raises HTTPException(status_code=400).

Your API now rejects dangerous queries, tracks token costs, and logs every interaction. These are the exact guardrails production AI agents need. Next up, you'll tackle the Secret Mission and give your agent memory so it can handle follow-up questions.

Secret mission

Add Conversation Memory

Your agent answers one question at a time. But real assistants remember what you asked before. Can you add conversation memory so a follow-up like "Now break that down by quarter" works without repeating the full context?

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs entirely locally, so there are no ongoing costs. The Claude API charges only per token when you actively send a request.

Resources you used:

  • Project folder on your Desktop containing main.py, agent.py, database.py, requirements.txt, sales.db, and your Python virtual environment.
  • ANTHROPIC_API_KEY environment variable set with setx on Windows (persists across terminal sessions).
  • Claude API key stored at platform.claude.com/settings/keys.

Keep everything running

No action needed. Choose this if you plan to keep building on this project or experimenting with the API.

  • Your project folder stays on your Desktop. The sales.db file takes negligible disk space.
  • Your ANTHROPIC_API_KEY environment variable remains available for future projects that use the Anthropic SDK.
  • Claude API charges are pay-per-token only. You are only billed when you send a request. Leaving the key active costs nothing while idle.

Pause - I'll come back to this later

Stop any running processes to free up your terminal. Your files and configuration stay in place for when you return.

  • If the FastAPI dev server is still running, press Ctrl+C in that terminal to stop it.
  • Your project folder, database, and environment variable all remain untouched. No ongoing costs while the server is stopped.

Delete - I don't want to use this again

Remove all project files and optionally revoke your API key. Follow each step that applies to you.

  • Stop the FastAPI dev server if it is still running by pressing Ctrl+C.
  • Delete your project folder by running the command for your operating system.

macOS

cd ~/Desktop
rm -rf sql-assistant

Windows

cd ~\Desktop
Remove-Item -Recurse -Force sql-assistant
  • Remove the ANTHROPIC_API_KEY environment variable.

macOS

Open your shell profile file (~/.zshrc or ~/.bashrc) and delete the line that contains export ANTHROPIC_API_KEY. Save the file and open a new terminal for the change to take effect.

Windows

Open PowerShell and run this command to remove the variable from your user environment:

[Environment]::SetEnvironmentVariable('ANTHROPIC_API_KEY', '', 'User')

Open a new terminal afterwards to confirm it is gone.

  • Optionally revoke your Claude API key. Go to platform.claude.com/settings/keys, find your key in the list, and delete it. This prevents anyone from using the key if it was ever exposed.

Nice Work!

Nice Work!

Nice work! You just built a production-quality REST API that turns natural language questions into real database answers, powered by an AI agent you built from scratch.

You've learned how to:

  • Build an agentic tool-calling loop from scratch, understanding exactly how Claude decides to call a tool, executes the SQL, and feeds results back until it produces a final answer.
  • Connect an LLM to a real SQLite database with production guardrails including SQL write-operation blocking, input validation, token tracking, and structured error handling.
  • Expose an AI agent behind a validated FastAPI endpoint with Pydantic request/response models and auto-generated interactive documentation at /docs.
  • Secret Mission: Add stateful conversation memory so the agent handles follow-up questions using context from previous turns, turning a one-shot tool into an intelligent assistant.

Ready to quiz yourself?