Skip to main content

Documentation Index

Fetch the complete documentation index at: https://docs.opper.ai/llms.txt

Use this file to discover all available pages before exploring further.

This is a clean fit for the JSON API: you send the database schema and a question, and you get back a structured object with the query. Because the output is schema-constrained, you can pull the sql field straight out and execute it. The example uses an in-memory SQLite database so it runs as-is.
python app.py

You › Who are the two highest-paid people in Engineering?

SQL › SELECT name, salary FROM employees WHERE department = ‘Engineering’ ORDER BY salary DESC LIMIT 2;

→ [(‘Dan’, 151000), (‘Alice’, 145000)]

​
The assistant

import os, sqlite3
from pydantic import BaseModel, Field
from opperai import Opper

opper = Opper(http_bearer=os.getenv("OPPER_API_KEY", ""))

# A sample database
db = sqlite3.connect(":memory:")
db.executescript("""
CREATE TABLE employees (id INTEGER, name TEXT, department TEXT, salary INTEGER, hire_date TEXT);
INSERT INTO employees VALUES
 (1,'Alice','Engineering',145000,'2021-03-01'),
 (2,'Bjorn','Engineering',132000,'2022-07-15'),
 (3,'Chloe','Sales',98000,'2020-01-10'),
 (4,'Dan','Engineering',151000,'2019-11-20'),
 (5,'Eva','Sales',105000,'2023-02-05');
""")

def schema(con):
    rows = con.execute("SELECT sql FROM sqlite_master WHERE type='table'").fetchall()
    return "\n".join(r[0] for r in rows)

class Query(BaseModel):
    reasoning: str = Field(description="Brief reasoning about how to answer.")
    sql: str = Field(description="A single valid SQLite query.")

def to_sql(question, db_schema):
    result = opper.call(
        name="generate-sql",
        instructions="Given a SQLite schema and a question, write one SQLite query that answers it. Return only the query in `sql`.",
        input={"schema": db_schema, "question": question},
        output_schema=Query,
    )
    return result.json_payload

question = "Who are the two highest-paid people in Engineering? Show name and salary."
out = to_sql(question, schema(db))

print("SQL:", out["sql"])
print("Rows:", db.execute(out["sql"]).fetchall())
Run it: 
pip install opperai
export OPPER_API_KEY="your-api-key"
python app.py

​
How it works

  • The Query schema forces the model to return a sql string (plus its reasoning). With structured output you read result.json_payload["sql"] and run it directly, no parsing or regex.
  • Passing the live schema(db) means the model writes queries against your real tables, not guessed ones.
  • The reasoning field is optional but useful: it gives you a plain-English trace of why the query looks the way it does, which shows up on the call’s trace.

​
Make it more robust

  • Retry on error. Catch SQLite exceptions, send the error message back as input, and ask for a corrected query.
  • Read-only safety. Run the generated SQL against a read-only connection so a bad query can’t modify data.
  • Score it. Add an Observe rule that checks the query ran and returned rows.

​
What’s next

Schemas

Pydantic, JSON Schema, and field descriptions.

Hints

Optimize for cost or speed without naming a model.

Streaming

Stream the query as it generates.

Observe

Score generated SQL automatically.