A GPT-5 self-serve analytics agent turns natural-language questions into governed SQL queries, letting product managers pull their own funnel numbers without filing a ticket. This tutorial builds a minimal but production-shaped version using Python, SQLite, and the OpenAI tool-calling protocol.
Prerequisites
- Python 3.11+ with
pip install openai sqlalchemy - A local SQLite file (we’ll create it) or any Postgres/MySQL URL
- An API key for a provider that serves
gpt-5via an OpenAI-compatible endpoint - Familiarity with Python async/sync and basic SQL
SELECTstatements
Set your key in the environment:
export OPENAI_API_KEY="sk-..." # or use n4n.ai's key if routing through that gateway
export BASE_URL="https://api.openai.com/v1" # swap for n4n.ai's URL if desiredStep 1: Seed a realistic analytics schema
We need a small warehouse with customers and orders. This script builds it and inserts rows so the agent has something to query.
# seed.py
from sqlalchemy import create_engine, text
engine = create_engine("sqlite:///analytics.db")
with engine.begin() as conn:
conn.execute(text("""
CREATE TABLE IF NOT EXISTS customers (
id INTEGER PRIMARY KEY,
name TEXT,
signup_date TEXT,
plan TEXT
);
CREATE TABLE IF NOT EXISTS orders (
id INTEGER PRIMARY KEY,
customer_id INTEGER,
amount REAL,
created_at TEXT
);
"""))
conn.execute(text("DELETE FROM customers; DELETE FROM orders;"))
conn.execute(text("""
INSERT INTO customers (id, name, signup_date, plan) VALUES
(1, 'Acme', '2024-01-12', 'pro'),
(2, 'Globex', '2024-02-03', 'free'),
(3, 'Initech', '2024-03-21', 'pro');
INSERT INTO orders (id, customer_id, amount, created_at) VALUES
(1, 1, 99.50, '2024-01-15'),
(2, 1, 20.00, '2024-02-01'),
(3, 2, 5.00, '2024-02-10'),
(4, 3, 150.00, '2024-04-01');
"""))
print("seeded")Run it:
python seed.py
# seededStep 2: Expose safe data tools
The agent must never run DROP or INSERT. We expose two functions: one to list tables/columns, one to execute a read-only query.
# tools.py
from sqlalchemy import create_engine, text
import re
engine = create_engine("sqlite:///analytics.db")
def list_tables() -> str:
with engine.connect() as conn:
rows = conn.execute(text(
"SELECT name FROM sqlite_master WHERE type='table'"
)).fetchall()
return ", ".join(r[0] for r in rows)
def execute_readonly_sql(query: str) -> str:
normalized = query.strip().lower()
if not normalized.startswith("select"):
return "ERROR: only SELECT queries are permitted"
if re.search(r"\b(insert|update|delete|drop|alter|create)\b", normalized):
return "ERROR: forbidden keyword detected"
with engine.connect() as conn:
try:
rows = conn.execute(text(query)).fetchall()
except Exception as e:
return f"ERROR: {e}"
return str(rows[:10]) # truncate for context safetyStep 3: Define the GPT-5 tool schemas
GPT-5 consumes JSON Schema descriptions. Keep them tight.
[
{
"type": "function",
"function": {
"name": "list_tables",
"description": "List all available tables in the analytics database.",
"parameters": {"type": "object", "properties": {}}
}
},
{
"type": "function",
"function": {
"name": "execute_readonly_sql",
"description": "Run a read-only SELECT query against the analytics DB. Returns rows.",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "Valid SQLite SELECT statement"}
},
"required": ["query"]
}
}
}
]Step 4: Build the agent loop
We use the OpenAI Python client. Point it at n4n.ai’s OpenAI-compatible endpoint (or OpenAI’s) to reach GPT-5 with automatic fallback when a provider is rate-limited.
# agent.py
import os, json
from openai import OpenAI
from tools import list_tables, execute_readonly_sql
client = OpenAI(api_key=os.environ["OPENAI_API_KEY"], base_url=os.environ["BASE_URL"])
MODEL = "gpt-5"
TOOLS = json.load(open("tools.json")) # the JSON from Step 3
DISPATCH = {
"list_tables": lambda _: list_tables(),
"execute_readonly_sql": lambda a: execute_readonly_sql(a["query"]),
}
def run_agent(question: str):
messages = [{"role": "user", "content": question}]
while True:
resp = client.chat.completions.create(
model=MODEL,
messages=messages,
tools=TOOLS,
tool_choice="auto",
)
msg = resp.choices[0].message
if not msg.tool_calls:
return msg.content
messages.append(msg)
for call in msg.tool_calls:
fn = DISPATCH[call.function.name]
args = json.loads(call.function.arguments)
result = fn(args)
messages.append({
"role": "tool",
"tool_call_id": call.id,
"content": result,
})
if __name__ == "__main__":
print(run_agent("What is the total revenue from pro customers?"))Step 5: Run and verify
Execute the agent against the seeded DB:
python agent.pyExpected output (model-generated text will vary slightly):
The total revenue from pro customers is 269.5, coming from Acme (119.5) and Initech (150.0).Checkpoint: if you instead ask “delete all orders”, the execute_readonly_sql guard returns an ERROR string, and GPT-5 should report it cannot perform that action.
Step 6: Add guardrails and observe metering
The loop above is functional but missing three production concerns:
- Token budgeting – truncate tool results before returning them to the model.
- Cache hints – if your gateway supports provider cache-control, pass
extra_headers={"cache-control": "max-age=300"}on repeated schema calls. - Usage metering – if you route through n4n.ai, per-token usage metering is returned in response headers, letting you attribute cost per question.
A hardened call looks like:
resp = client.chat.completions.create(
model=MODEL,
messages=messages,
tools=TOOLS,
extra_headers={"cache-control": "max-age=300"}, # forwarded to provider
)
print(resp.usage.total_tokens) # or read headers via httpx if neededWrapping up
You now have a GPT-5 self-serve analytics agent that converts questions to SQL, enforces read-only access, and degrades safely on bad input. Swap SQLite for your warehouse, add column-level descriptions to the list_tables output, and put the loop behind a FastAPI endpoint to ship it.