← Back to DevBytes
AI August 16, 2026 11 min read

Building a SQL Query Agent with OpenAI Agents SDK: Complete Guide

Introduction to Building a SQL Query Agent with OpenAI Agents SDK

Modern applications increasingly rely on natural language interfaces to unlock data locked inside relational databases. Instead of forcing users to learn SQL or rely on a handful of pre-built dashboards, a SQL Query Agent lets them ask plain-English questions and receive accurate, structured answers. With the release of the OpenAI Agents SDK, building such an agent has become dramatically simpler. The SDK provides primitives for tools, handoffs, guardrails, and tracing — everything you need to construct a robust, production-grade data assistant.

In this guide, you'll learn what a SQL Query Agent is, why it matters, how to build one end-to-end with the OpenAI Agents SDK, and the best practices that separate a toy demo from a reliable production system.

What Is a SQL Query Agent?

A SQL Query Agent is an AI-powered assistant that translates natural language questions into SQL queries, executes those queries against a target database, and returns the results in a human-readable form. Unlike a simple text-to-SQL wrapper, an agent operates in a loop: it can inspect schema, generate a query, validate it, run it, observe errors, and self-correct before answering.

Key capabilities of a mature SQL Query Agent include:

Why It Matters

Democratizing data access is one of the highest-leverage applications of LLMs. Most business users — marketers, operations teams, executives — cannot write SQL, yet they constantly need answers that live in the database. A SQL Query Agent closes that gap without requiring engineering teams to build bespoke reports for every question.

Beyond convenience, an agent-based approach offers advantages over naive text-to-SQL pipelines:

Prerequisites and Setup

Before writing code, make sure you have the following:

Install the SDK and supporting libraries:

pip install openai-agents python-dotenv

Create a sample SQLite database with realistic data so the agent has something meaningful to query:

import sqlite3

conn = sqlite3.connect("company.db")
cur = conn.cursor()

cur.executescript("""
CREATE TABLE IF NOT EXISTS customers (
    id INTEGER PRIMARY KEY,
    name TEXT NOT NULL,
    country TEXT,
    signup_date TEXT,
    lifetime_value REAL
);

CREATE TABLE IF NOT EXISTS orders (
    id INTEGER PRIMARY KEY,
    customer_id INTEGER REFERENCES customers(id),
    order_date TEXT,
    total_amount REAL,
    status TEXT
);

CREATE TABLE IF NOT EXISTS products (
    id INTEGER PRIMARY KEY,
    name TEXT,
    category TEXT,
    price REAL
);

CREATE TABLE IF NOT EXISTS order_items (
    id INTEGER PRIMARY KEY,
    order_id INTEGER REFERENCES orders(id),
    product_id INTEGER REFERENCES products(id),
    quantity INTEGER,
    unit_price REAL
);
""")

# Insert sample data
cur.executemany("INSERT INTO customers VALUES (?,?,?,?,?)", [
    (1, "Acme Corp", "USA", "2023-01-15", 12500.00),
    (2, "Globex", "Germany", "2023-03-22", 8900.50),
    (3, "Initech", "USA", "2023-06-10", 4300.00),
    (4, "Umbrella SA", "France", "2024-02-01", 21000.00),
])

cur.executemany("INSERT INTO orders VALUES (?,?,?,?,?)", [
    (1, 1, "2024-01-20", 1500.00, "completed"),
    (2, 1, "2024-03-11", 2300.00, "completed"),
    (3, 2, "2024-02-18", 950.00, "completed"),
    (4, 4, "2024-04-05", 7800.00, "completed"),
    (5, 3, "2024-05-01", 200.00, "refunded"),
])

conn.commit()
conn.close()
print("Database initialized.")

Architecture of the Agent

Our agent will be composed of three tools and a single coordinating agent:

The agent loop works as follows: the user asks a question, the model calls list_schema (and optionally sample_rows), it generates SQL, calls run_sql, and if the result is an error it retries with a corrected query. Finally, it summarizes the rows into a natural-language answer.

Defining the Tools

Tools in the OpenAI Agents SDK are plain Python functions decorated with @function_tool. The SDK inspects type hints and docstrings to produce the JSON schema the model uses to call them.

import sqlite3
import json
from agents import function_tool

DB_PATH = "company.db"

def _connect():
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    return conn

@function_tool
def list_schema() -> str:
    """Return the schema of all tables in the database, including columns,
    types, and foreign key relationships. Call this first before writing SQL."""
    conn = _connect()
    cur = conn.cursor()
    cur.execute("SELECT name FROM sqlite_master WHERE type='table';")
    tables = [r[0] for r in cur.fetchall()]

    schema = {}
    for table in tables:
        cur.execute(f"PRAGMA table_info({table});")
        columns = [
            {"name": r[1], "type": r[2], "notnull": bool(r[3]), "pk": bool(r[5])}
            for r in cur.fetchall()
        ]
        cur.execute(f"PRAGMA foreign_key_list({table});")
        fks = [
            {"from": r[3], "table": r[2], "to": r[4]}
            for r in cur.fetchall()
        ]
        schema[table] = {"columns": columns, "foreign_keys": fks}

    conn.close()
    return json.dumps(schema, indent=2)

@function_tool
def sample_rows(table_name: str, limit: int = 3) -> str:
    """Return a few sample rows from the given table to understand the data.
    Args:
        table_name: The name of the table to sample.
        limit: Maximum number of rows to return (default 3).
    """
    # Guard against SQL injection in the table name
    allowed = {"customers", "orders", "products", "order_items"}
    if table_name not in allowed:
        return f"Error: unknown table '{table_name}'"

    conn = _connect()
    cur = conn.cursor()
    cur.execute(f"SELECT * FROM {table_name} LIMIT {int(limit)};")
    rows = [dict(r) for r in cur.fetchall()]
    conn.close()
    return json.dumps(rows, indent=2, default=str)

@function_tool
def run_sql(query: str) -> str:
    """Execute a read-only SQL query against the SQLite database and return
    the results as JSON. Only SELECT statements are allowed. If the query
    fails, the database error message is returned so you can fix the query.

    Args:
        query: A single SELECT SQL statement.
    """
    stripped = query.strip().rstrip(";")
    if not stripped.lower().startswith("select"):
        return "Error: only SELECT statements are permitted."

    conn = _connect()
    try:
        cur = conn.cursor()
        cur.execute(stripped)
        rows = [dict(r) for r in cur.fetchall()]
        return json.dumps(rows, indent=2, default=str)
    except Exception as e:
        return f"SQL_ERROR: {type(e).__name__}: {e}"
    finally:
        conn.close()

Notice the safety measures baked into run_sql: only SELECT statements are allowed, and any exception is returned as a string rather than raised. Returning errors as strings is critical — it lets the model observe the failure and self-correct within the same conversation turn.

Creating the Agent

With tools defined, we now create the agent itself. The system prompt is the most important configuration choice: it tells the model how to use the tools, what dialect to target, and how to format answers.

from agents import Agent, Runner

SQL_AGENT_INSTRUCTIONS = """You are a SQL data analyst assistant working against a SQLite database.

Follow this workflow for every user question:
1. Call list_schema to inspect available tables, columns, and foreign keys.
2. If a column's meaning is ambiguous, call sample_rows to inspect real data.
3. Write a single SELECT query that answers the question. Prefer JOINs over
   multiple queries. Use SQLite-compatible SQL syntax.
4. Call run_sql to execute the query.
5. If run_sql returns SQL_ERROR, read the error, fix the query, and retry.
   Retry at most 3 times before explaining the issue to the user.
6. Summarize the returned rows in plain English. Include specific numbers.
   Do not show raw JSON unless the user explicitly asks for it.

Rules:
- Only use SELECT statements. Never INSERT, UPDATE, DELETE, or DROP.
- If the question cannot be answered with the available schema, say so clearly.
- Be concise. Lead with the answer, then add brief context if helpful.
"""

sql_agent = Agent(
    name="SQLQueryAgent",
    instructions=SQL_AGENT_INSTRUCTIONS,
    model="gpt-4o-mini",
    tools=[list_schema, sample_rows, run_sql],
)

Running the Agent

The Runner class executes the agent loop. It handles tool dispatch, model calls, and the iterative reasoning cycle automatically.

import asyncio
from agents import Runner

async def ask(question: str) -> str:
    result = await Runner.run(sql_agent, question)
    return result.final_output

if __name__ == "__main__":
    questions = [
        "Which country has the highest total order revenue?",
        "How many customers signed up in 2023, and what is their average lifetime value?",
        "List all orders that were refunded.",
    ]

    for q in questions:
        print(f"\nQ: {q}")
        answer = asyncio.run(ask(q))
        print(f"A: {answer}")

When you run this script, the agent will autonomously call list_schema, generate SQL, execute it, and produce a natural-language summary. Behind the scenes, the SDK's tracing system records every step, which you can inspect for debugging.

Adding a Guardrail for Safety

In production, you'll want to prevent the agent from answering certain types of questions — for example, queries that attempt to extract personally identifiable information. The Agents SDK supports output guardrails that run after the agent produces its final answer.

from agents import GuardrailFunctionOutput, output_guardrail

@output_guardrail
async def no_pii_guardrail(ctx, agent, output):
    """Block answers that appear to expose personal customer information."""
    forbidden = ["email", "phone", "ssn", "credit card", "password"]
    text = str(output).lower()
    flagged = [term for term in forbidden if term in text]
    return GuardrailFunctionOutput(
        output_info={"flagged_terms": flagged},
        tripwire_triggered=len(flagged) > 0,
    )

guarded_agent = Agent(
    name="SQLQueryAgentGuarded",
    instructions=SQL_AGENT_INSTRUCTIONS,
    model="gpt-4o-mini",
    tools=[list_schema, sample_rows, run_sql],
    output_guardrails=[no_pii_guardrail],
)

If the guardrail trips, the Runner raises a GuardrailTripwireTriggered exception, which you can catch and handle gracefully in your application layer.

Best Practices

Constrain the Database Surface

Don't expose every table in your warehouse to the agent. Curate a schema view containing only the tables relevant to the use case. This reduces hallucination, lowers token usage, and improves accuracy. You can implement this by filtering the tables returned from list_schema against an allowlist.

Enforce Read-Only Access at the Database Level

Application-level checks like the SELECT-only guard are useful, but they should not be your only defense. Connect to the database using a role with strictly read-only permissions. In PostgreSQL, for example, grant SELECT on specific tables and nothing else. Defense in depth is essential when LLMs generate SQL.

Limit Result Sizes

Large result sets consume tokens and slow down responses. Add a LIMIT clause in your instructions, or wrap queries server-side with a maximum row cap inside run_sql. For analytical questions, encourage aggregation (SUM, COUNT, AVG) over raw row retrieval.

Provide Rich Schema Metadata

Column names like ltv or is_actv are ambiguous to a model. Augment your schema tool with descriptions, enum values, and sample values. The more semantic context the model has, the fewer clarification rounds it needs.

Use Tracing for Debugging

The SDK's tracing is invaluable when the agent returns a wrong answer. Inspect the trace to see exactly which query was generated, what the database returned, and where the reasoning diverged. Add the trace processor early in development:

from agents import set_trace_processors
from agents.tracing import console_trace_processor

set_trace_processors([console_trace_processor()])

Cache Schema Lookups

Calling list_schema on every turn wastes tokens. Consider caching the schema string and injecting it into the system prompt, then offering sample_rows as the only dynamic tool. This is especially valuable for schemas with hundreds of tables.

Test with a Golden Question Set

Maintain a suite of representative questions with expected SQL or expected answer patterns. Run this suite whenever you change the prompt, model, or schema. Text-to-SQL accuracy is fragile; regression testing catches silent degradations.

Extending the Agent

Once the core loop is solid, you can extend the agent with additional tools and handoffs:

Here's a minimal example of adding a charting tool and a handoff:

import base64
import io
import matplotlib

matplotlib.use("Agg")
import matplotlib.pyplot as plt
from agents import function_tool, Agent, handoff

@function_tool
def make_bar_chart(categories: list[str], values: list[float], title: str) -> str:
    """Generate a bar chart and return it as a base64-encoded PNG string."""
    fig, ax = plt.subplots()
    ax.bar(categories, values)
    ax.set_title(title)
    ax.tick_params(axis='x', rotation=45)
    buf = io.BytesIO()
    fig.tight_layout()
    fig.savefig(buf, format="png")
    plt.close(fig)
    return base64.b64encode(buf.getvalue()).decode()

chart_agent = Agent(
    name="ChartAgent",
    instructions="You receive data and produce clear, labeled bar charts.",
    model="gpt-4o-mini",
    tools=[make_bar_chart],
)

sql_agent_with_handoff = Agent(
    name="SQLQueryAgentExtended",
    instructions=SQL_AGENT_INSTRUCTIONS + (
        "\nIf the user asks for a visualization, hand off to the ChartAgent "
        "with the relevant data."
    ),
    model="gpt-4o-mini",
    tools=[list_schema, sample_rows, run_sql],
    handoffs=[chart_agent],
)

Conclusion

Building a SQL Query Agent with the OpenAI Agents SDK is a practical and powerful way to give non-technical users direct access to their data. By combining schema-aware tools, a self-correcting execution loop, and the SDK's guardrail and tracing primitives, you can ship an assistant that is both capable and safe. Start with the minimal three-tool architecture shown here, enforce read-only access at every layer, and iterate by expanding your golden question set. With careful prompt engineering, curated schema metadata, and disciplined testing, your SQL Query Agent can become a trusted interface between your database and the people who need its answers.

— Ad —

Google AdSense will appear here after approval

← Back to all articles