Agent example: ask your database

The job: a manager asks a question in plain English, such as “who bought the most last month?”. The agent writes the SQL, runs it through your function, and answers with the numbers.

Tools used: function calling on /v1/responses. The database stays on your network; only the rows your function returns are seen by the model.

The code

import json, sqlite3
from openai import OpenAI

client = OpenAI(base_url="https://dummydomain/v1", api_key="YOUR_API_KEY")

# Your database. This example builds a small one in memory.
db = sqlite3.connect(":memory:")
db.executescript("""
CREATE TABLE customers (id INTEGER PRIMARY KEY, name TEXT, city TEXT);
CREATE TABLE invoices (id INTEGER PRIMARY KEY, customer_id INTEGER, invoice_date TEXT, amount_inr INTEGER);
INSERT INTO customers VALUES (1, 'Gupta Electricals', 'Jaipur'), (2, 'Mehta Hardware', 'Pune'),
  (3, 'Rao Lighting', 'Chennai'), (4, 'Khan Traders', 'Lucknow');
INSERT INTO invoices VALUES (1, 1, '2026-09-04', 240000), (2, 2, '2026-09-11', 95000),
  (3, 3, '2026-09-18', 410000), (4, 1, '2026-09-25', 180000), (5, 4, '2026-09-27', 60000),
  (6, 2, '2026-10-02', 300000);
""")
db.execute("PRAGMA query_only = ON")  # the model may read, never change

SCHEMA = "\n".join(row[0] for row in db.execute("SELECT sql FROM sqlite_master WHERE type = 'table'"))

TOOLS = [{
    "type": "function",
    "name": "run_sql",
    "description": "Run one read-only SQLite SELECT query and return the rows as JSON",
    "parameters": {"type": "object", "properties": {"query": {"type": "string"}},
                   "required": ["query"], "additionalProperties": False},
    "strict": True,
}]

def run_sql(query):
    if not query.lstrip().lower().startswith(("select", "with")):
        return json.dumps({"error": "only SELECT queries are allowed"})
    try:
        cursor = db.execute(query)
        columns = [c[0] for c in cursor.description]
        return json.dumps([dict(zip(columns, row)) for row in cursor.fetchmany(50)])
    except sqlite3.Error as exc:
        return json.dumps({"error": str(exc)})  # the model can fix its query and try again

def ask(question):
    items = [{"role": "user", "content": question}]
    instructions = ("You answer questions about the company's sales from its database. Use run_sql. "
                    "Dates are stored as YYYY-MM-DD. Amounts are in rupees. "
                    "Answer in plain English. Write amounts as Rs with Indian grouping, such as Rs 12,50,000.\n\nTables:\n" + SCHEMA)
    response = client.responses.create(model="Fast", instructions=instructions, input=items, tools=TOOLS)
    for _ in range(6):
        calls = [item for item in response.output if item.type == "function_call"]
        if not calls:
            break
        items += response.output
        for call in calls:
            query = json.loads(call.arguments)["query"]
            print("SQL:", " ".join(query.split()))
            items.append({"type": "function_call_output", "call_id": call.call_id, "output": run_sql(query)})
        response = client.responses.create(model="Fast", instructions=instructions, input=items, tools=TOOLS)
    return response.output_text

print(ask("Which customer had the highest sales in September 2026, and how much was it?"))

What it prints

SQL: SELECT c.name, SUM(i.amount_inr) AS total_sales FROM customers c JOIN invoices i ON c.id = i.customer_id WHERE i.invoice_date BETWEEN '2026-09-01' AND '2026-09-30' GROUP BY c.id ORDER BY total_sales DESC LIMIT 1;
The customer with the highest sales in September 2026 was Gupta Electricals, with a total of Rs 4,20,000.

How it works

The schema is in the instructions. The model sees the table definitions, so it writes queries that fit your database.
Errors go back to the model. A bad query returns the database error as the function result, and the model fixes its SQL on the next round.
The connection is read-only. PRAGMA query_only and the SELECT check mean the agent can never change data, whatever it writes.
The loop is capped. Six rounds are enough for a question that needs a few queries.

Where else this works

Sales and collections reports, stock levels across warehouses, student results, hospital bed occupancy, ticket queues. Point the same pattern at PostgreSQL, MySQL or SQL Server by changing run_sql.

Common questions

Can ZenithAI answer questions from our SQL database?

Yes. Give the model a function that runs read-only queries and the table definitions. It writes the SQL, your code runs it on your database, and the model explains the result in plain language.

Is it safe to let the model write SQL?

Only with limits in your code. Open the connection read-only, allow SELECT queries only, cap the rows returned, and use a database user that can see just the tables the agent needs.