OpenAI SDK: Project: retail analyst assistant

Ask “which category earned the most from completed orders?” in plain English and get a correct answer, with the numbers, the SQL that produced them, and caveats about data quality. The assistant queries the SQL guide’s retail database through a read-only SQL tool, answers in a structured format, can search products by meaning, and tracks tokens and cost per question. Every piece is tested offline against the real SDK with a fake server.

1. Architecture

2. Run it

from the repository root

cd openai-sdk/project
python demo.py                 # offline: real SDK, real SQL, a stand-in "model"
python -m pytest tests         # 21 offline tests
# with a key (lesson 02):
python cli.py

python demo.py (real output)

Q: Which product category earned the most revenue from completed orders in 2024?
tools used: [('run_sql', 'SELECT p.category,\n       ROUND(SUM(oi.q…')]

Electronics earned the most revenue from completed orders: INR 65,639,080.77 across 309 orders, ahead of Furniture (INR 57,179,788.12).
  - Electronics revenue (INR): 65,639,080.77
  - Furniture revenue (INR): 57,179,788.12
  - Stationery revenue (INR): 54,165,571.05
  - Appliances revenue (INR): 50,146,578.61
caveats: Lines with zero or negative quantity were excluded.; Discounts with no value were treated as 0%.

tokens: 398 in / 235 out  ≈ $0.000362 (fake token counts, real price table)

3. The code that matters

Read-only gateway

analyst/db.py (core)

_ALLOWED = {sqlite3.SQLITE_SELECT, sqlite3.SQLITE_READ, sqlite3.SQLITE_FUNCTION,
            sqlite3.SQLITE_RECURSIVE}

self._con = sqlite3.connect(f"file:{path}?mode=ro", uri=True)     # layer 1: read-only file handle
self._con.set_authorizer(self._authorize)                          # layer 2: deny non-read operations

def query(self, sql):
    text = sql.strip().rstrip(";").strip()
    if ";" in text:
        raise QueryRejected("only one statement is allowed")
    if not text.lower().startswith(("select", "with")):
        raise QueryRejected("only SELECT (or WITH ... SELECT) queries are allowed")
    deadline = time.monotonic() + MAX_SECONDS                      # layer 3: time and row limits
    self._con.set_progress_handler(lambda: 1 if time.monotonic() > deadline else 0, 10_000)
    ...
    rows = cur.fetchmany(MAX_ROWS + 1)

Three independent layers, because the model’s SQL is untrusted text (part 16). The tests prove it: a WITH … DELETE passes the prefix check but is stopped by the authorizer, and the table still has 800 orders afterwards.

The loop

analyst/assistant.py (core)

response = self.client.responses.parse(model=MODEL, instructions=INSTRUCTIONS, input=question,
                                       tools=TOOL_SCHEMAS, text_format=Answer)
for _ in range(self.max_rounds):
    turn.input_tokens += response.usage.input_tokens
    turn.output_tokens += response.usage.output_tokens
    calls = [item for item in response.output if item.type == "function_call"]
    if not calls:
        return response.output_parsed, turn               # the structured Answer
    outputs = [self._run_tool(c, turn) for c in calls]    # errors become outputs, not exceptions
    response = self.client.responses.parse(model=MODEL, instructions=INSTRUCTIONS,
                                           previous_response_id=response.id, input=outputs,
                                           tools=TOOL_SCHEMAS, text_format=Answer)
raise RuntimeError(f"no answer after {self.max_rounds} tool rounds")

The answer schema

analyst/assistant.py

class KeyNumber(BaseModel):
    label: str
    value: str

class Answer(BaseModel):
    summary: str = Field(description="2-4 sentences answering the question")
    key_numbers: list[KeyNumber]
    sql: list[str] = Field(description="The SQL queries whose results support the answer")
    caveats: list[str]

Returning the SQL makes answers auditable: an analyst can re-run the query. The instructions require every number to come from a query result, and the caveats field gives the model a place to mention the dataset’s known problems (duplicate customers, zero quantities, guest orders).

4. Tests

python -m pytest tests (real output)

.....................                                                    [100%]
21 passed in 4.65s

5. Extend it

Recap