Multi-Agent SQL Assistant (LangGraph + OpenAI + FastAPI)

A multi-agent system that connects to any SQL database and answers natural-language questions about it, with a human-approval step before any query actually runs.

Tested end-to-end against a live SQLite database with the LLM calls mocked (no API key required to verify the wiring/routing — see "How it was verified" below).

Setup

pip install -r requirements.txt
cp .env.example .env   # add your OPENAI_API_KEY
uvicorn app.main:app --reload

This single command runs everything — the API and the website. Open http://localhost:8000 in a browser. No separate frontend server, no build step — static/index.html is plain HTML/JS served directly by FastAPI.

Four ways to connect (the website has a tab for each)

Tab	Who it's for	What happens
Saved connection	End users with no technical knowledge	Pick a name (e.g. "Sales DB") an admin already configured. No credentials shown.
Database login	Anyone who knows their DB's host/port/username/password	Fill in plain fields, no connection-string syntax needed. Can optionally save it as a named profile for next time.
Upload a file	Users with no database at all, just spreadsheets	Upload a CSV/XLSX, it's auto-converted into a SQLite database in the background.
Advanced	Developers	Paste a raw SQLAlchemy connection string directly.

Driver note: "Database login" needs the matching driver installed — pip install psycopg2-binary for PostgreSQL, pip install pymysql for MySQL, pip install pyodbc for SQL Server (these are commented out in requirements.txt, uncomment what you need).

If you ever want the frontend hosted separately from the API (e.g. a real deployed website calling a separately-hosted backend), the CORS middleware in app/main.py is already set up for that — just point the frontend's fetch calls at the backend's public URL instead of using the same-origin "" base.

Usage (API directly, e.g. for testing without the browser)

# 1. Connect to a database (any SQLAlchemy connection string)
curl -X POST localhost:8000/connect \
  -H "Content-Type: application/json" \
  -d '{"connection_string": "sqlite:///mydata.db"}'
# -> {"db_id": "...", "schema": "Table: customers\n  columns: ..."}

# 2. Ask a question — returns the PROPOSED SQL, does not run it yet
curl -X POST localhost:8000/ask \
  -H "Content-Type: application/json" \
  -d '{"db_id": "<db_id>", "question": "How many customers per country?"}'
# -> {"thread_id": "...", "status": "awaiting_approval", "sql": "SELECT ...", "explanation": "..."}

# 3. Approve and execute
curl -X POST localhost:8000/confirm \
  -H "Content-Type: application/json" \
  -d '{"thread_id": "<thread_id>"}'
# -> {"answer": "...", "columns": [...], "rows": [...]}

Architecture

See app/graph.py for the LangGraph wiring. Flow:

retrieve_schema -> generate_sql -> validate_sql --(invalid, retries left)--> generate_sql
                                         |
                                  (valid) v
                                   [PAUSE for human approval]
                                         |
                                  execute_sql --(error, retries left)--> generate_sql
                                         |
                                  (success) v
                                  synthesize_answer -> END

• app/db.py — Connection Agent: generic SQLAlchemy connector + schema introspection/caching.
• app/agents.py — SQL Generator, Validator, Executor, Synthesizer agents + retry routing.
• app/graph.py — LangGraph StateGraph wiring, including the interrupt_before=["execute_sql"] human-approval gate.
• app/main.py — FastAPI endpoints: /connect, /connect/structured, /connect/upload, /connect/profile, /profiles, /ask, /confirm + serves static/index.html at /.
• app/profiles.py — saved connection profiles (name -> connection string), so non-technical users never see credentials.
• static/index.html — the website: tabbed connect UI, schema viewer, question box, SQL approval card, answer + result table. Plain HTML/JS, talks to the API via fetch.

Safety

• Validator agent rejects anything that isn't a SELECT, blocks DDL/write keywords, and force-adds a row LIMIT.
• Real execution only happens after explicit /confirm — never automatically.
• Query timeout enforced at execution.
• Saved profiles are stored in plain memory for this MVP — encrypt them at rest (e.g. cryptography.fernet.Fernet) before using this with real credentials in production.

Next steps (see architecture doc for full roadmap)

• Swap MemorySaver for a persistent checkpointer (Postgres/Redis) so paused conversations survive a server restart.
• Wrap the same frontend in Electron/Tauri if you also want a native desktop app — no backend changes needed, it would call the same three endpoints.
• Add the Schema Retriever (vector search) agent for databases with many tables.
• Deploy: put the backend on a host (Render/Fly/EC2/etc.), point DATABASE_URL-style connection strings at real production-like databases over a private network, and put the static site behind the same domain or a CDN with CORS pointed at the API.