#!/usr/bin/env python3 """ chat_with_data.py — "chat with your data" for the BudgetDB warehouse. Lets a non-technical leader ask a plain-English question ("which vendors are we paying the most for relative to how many teams actually use them?") and get a grounded, numbers-backed answer, without needing to know SQL or open the warehouse directly. HOW IT WORKS (the actual architecture, not a black box): 1. Claude is given the warehouse's real schema (table/column names, what each table means) and one tool: run_sql_query. 2. Claude decides what SQL to run to answer the question, calls the tool, gets real rows back, and can run further queries if the first result doesn't fully answer it. 3. Once it has enough to answer, it responds in plain English, citing the actual numbers it queried — not a hallucinated guess. SAFETY MODEL (this is the part that actually matters for a finance tool): - The tool only ever executes SELECT statements. Every generated query is checked against a strict allowlist (must start with SELECT/WITH, must not contain any of INSERT/UPDATE/ DELETE/DROP/ALTER/CREATE/TRUNCATE/GRANT/REVOKE) before it's run at all - defense in depth on top of point 2. - The demo database connection is opened in SQLite's own read-only mode (`mode=ro`), so even a query that somehow got past the allowlist would be rejected by the database driver itself, not just by application logic. - In production against the real Postgres warehouse, the equivalent is a dedicated read-only database role (GRANT SELECT ONLY) - the allowlist check here is what that role enforcement would look like at the SQL-string level; use both, don't rely on one. - Every query Claude runs is printed to the terminal before execution, so there's always a visible audit trail of what was actually asked of the database. USAGE: export ANTHROPIC_API_KEY=sk-ant-... python3 seed_demo_db.py # one-time: build the demo warehouse python3 chat_with_data.py "Which vendors are we paying the most for relative to how many teams actually use them?" python3 chat_with_data.py # no argument: interactive prompt loop NOTE ON THIS BEING A PORTFOLIO PIECE: This has been run and verified end-to-end against the seeded SQLite demo database's SQL execution and safety-check logic. The live Claude API call itself has NOT been tested in the environment this was built in (no API key was available there) - the request/response handling follows the Anthropic Python SDK's documented tool-use pattern exactly, but if you're the one running this, that's the one piece worth double-checking on first run. """ import argparse import json import os import re import sqlite3 import sys from pathlib import Path try: import anthropic except ImportError: print("Missing dependency. Run: pip3 install anthropic", file=sys.stderr) sys.exit(1) DB_PATH = Path(__file__).parent / "demo_warehouse.sqlite3" MODEL = "claude-sonnet-4-5" SCHEMA_DESCRIPTION = """ You are answering questions about a company's finance & workforce data warehouse. Tables: dim_employee(employee_id, full_name, team, role_title, employment_type, base_salary_annual, benefits_rate) -- one row per employee. employment_type is FTE / CON (contractor) / INT-CON (international contractor). vendor_spend(vendor_name, category, department_owner, total_spend_2025, avg_monthly_spend_2025, teams_using_tool, recurring) -- one row per software/vendor tool. teams_using_tool = how many internal teams actively use it -- (low teams_using_tool + high spend = a strong candidate to review before renewal). commission_joined(employee_id, month_date, commission_amount, match_method) -- one row per commission payout, already matched to dim_employee.employee_id. -- match_method explains HOW the match was made (MAP_OVERRIDE / NAME_KEY_EXACT / UNMATCHED) - -- mention this if a question is about data trustworthiness or matching confidence. fact_cost_monthly(month_date, payroll_total, vendors_total, commission_total, te_total, total_cost) -- one row per month: the unified source of truth for total company cost, already -- reconciled across payroll/vendors/commission/T&E. Prefer this table for "total cost" -- or "burn" questions over summing the other tables yourself. Only SELECT queries are permitted. Always ground your final answer in numbers you actually queried - if the schema can't answer the question, say so plainly instead of guessing. """.strip() _FORBIDDEN = re.compile(r"\b(INSERT|UPDATE|DELETE|DROP|ALTER|CREATE|TRUNCATE|GRANT|REVOKE|ATTACH|PRAGMA)\b", re.I) def is_safe_select(sql: str) -> bool: """Defense-in-depth check: only SELECT/WITH, no mutating keywords anywhere in the string. This runs in addition to (not instead of) opening the DB connection read-only - either layer alone should already stop a write, both together is the point. """ stripped = sql.strip().rstrip(";") if not re.match(r"^\s*(SELECT|WITH)\b", stripped, re.I): return False if _FORBIDDEN.search(stripped): return False if ";" in stripped: # no stacked statements return False return True def run_sql_query(sql: str) -> dict: if not is_safe_select(sql): return {"error": f"Refused: only a single read-only SELECT/WITH statement is permitted. Got: {sql!r}"} if not DB_PATH.exists(): return {"error": "Demo database not found - run seed_demo_db.py first."} print(f" \033[2m[SQL] {sql}\033[0m") try: # uri=True + mode=ro: the database driver itself refuses writes, independent of the # regex check above. In production this is a read-only Postgres role instead. conn = sqlite3.connect(f"file:{DB_PATH}?mode=ro", uri=True) conn.row_factory = sqlite3.Row rows = conn.execute(sql).fetchall() conn.close() return {"rows": [dict(r) for r in rows][:200]} # cap result size sent back to the model except sqlite3.Error as e: return {"error": str(e)} TOOLS = [{ "name": "run_sql_query", "description": "Execute a single read-only SQL SELECT query against the warehouse and return the rows.", "input_schema": { "type": "object", "properties": {"sql": {"type": "string", "description": "A single SELECT (or WITH ... SELECT) statement."}}, "required": ["sql"], }, }] def ask(client: "anthropic.Anthropic", question: str) -> str: messages = [{"role": "user", "content": question}] for _ in range(6): # hard cap on tool-call round trips, avoids a runaway loop response = client.messages.create( model=MODEL, max_tokens=1024, system=SCHEMA_DESCRIPTION, tools=TOOLS, messages=messages, ) if response.stop_reason != "tool_use": return "".join(block.text for block in response.content if block.type == "text") messages.append({"role": "assistant", "content": response.content}) tool_results = [] for block in response.content: if block.type == "tool_use" and block.name == "run_sql_query": result = run_sql_query(block.input.get("sql", "")) tool_results.append({ "type": "tool_result", "tool_use_id": block.id, "content": json.dumps(result), }) messages.append({"role": "user", "content": tool_results}) return "Stopped after too many query rounds without a final answer - the question may need to be narrower." def main(): parser = argparse.ArgumentParser(description="Chat with the BudgetDB warehouse in plain English.") parser.add_argument("question", nargs="*", help="Question to ask (omit for interactive mode)") args = parser.parse_args() if not os.environ.get("ANTHROPIC_API_KEY"): print("Set ANTHROPIC_API_KEY first: export ANTHROPIC_API_KEY=sk-ant-...", file=sys.stderr) sys.exit(1) if not DB_PATH.exists(): print("Demo database not found. Run: python3 seed_demo_db.py", file=sys.stderr) sys.exit(1) client = anthropic.Anthropic() if args.question: print(ask(client, " ".join(args.question))) return print("Chat with BudgetDB. Ctrl+C to exit.\n") while True: try: question = input("> ").strip() except (EOFError, KeyboardInterrupt): print() break if not question: continue print(ask(client, question), "\n") if __name__ == "__main__": main()