AI Sales Query Agent

An AI-powered sales analytics API that converts natural-language questions into SQL and executes them against a SQLite sales database through a secure MCP-style database server.

The application is built with FastAPI and supports both an optional Claude API integration and a deterministic offline SQL planner.

---

🏗️ Architecture

                         ┌─────────────────────┐
                         │       Client        │
                         │  Swagger / Browser  │
                         └──────────┬──────────┘
                                    │
                                    │ POST /query
                                    ▼
                         ┌─────────────────────┐
                         │      FastAPI        │
                         │       main.py       │
                         └──────────┬──────────┘
                                    │
                                    ▼
                         ┌─────────────────────┐
                         │      SQLAgent       │
                         │      agent.py       │
                         └──────────┬──────────┘
                                    │
                       ┌────────────┴────────────┐
                       │                         │
                       ▼                         ▼
              ┌────────────────┐       ┌──────────────────┐
              │ ClaudeSQLAgent │       │ LocalSQLPlanner  │
              │  Claude API    │       │   Offline Mode   │
              └────────────────┘       └──────────────────┘
                       │                         │
                       └────────────┬────────────┘
                                    │
                                    │ Generated SQL
                                    ▼
                         ┌─────────────────────┐
                         │      MCPServer      │
                         │   mcp_server.py     │
                         │                     │
                         │ list_tables()       │
                         │ describe_schema()   │
                         │ execute_query()     │
                         └──────────┬──────────┘
                                    │
                             Read-only SQL
                                    │
                                    ▼
                         ┌─────────────────────┐
                         │   SQLite Database   │
                         │    data/sales.db    │
                         └─────────────────────┘

---

🔄 How It Works

The application follows this flow:

User Question
     │
     ▼
POST /query
     │
     ▼
SQLAgent
     │
     ├── Claude API
     │
     └── LocalSQLPlanner
     │
     ▼
Generated SQL
     │
     ▼
MCPServer
     │
     ├── Validate SQL
     ├── Check read-only operation
     ├── Reject dangerous keywords
     └── SQLite Authorizer
     │
     ▼
SQLite Database
     │
     ▼
Query Results
     │
     ▼
JSON Response

---

🛡️ Database Security

The execute_query() method uses multiple security checks.

1. Only SELECT queries

The server accepts:

SELECT ...

and:

WITH ... SELECT ...

Write operations are rejected.

2. Multiple statements are rejected

For example:

SELECT * FROM customers;
DROP TABLE customers;

is rejected.

3. Dangerous keywords are blocked

The application checks for operations such as:

INSERT
UPDATE
DELETE
DROP
ALTER
CREATE
PRAGMA
ATTACH
DETACH

4. SQLite Authorizer

The application also uses SQLite's built-in authorizer callback.

This provides database-level protection against:

• INSERT
• UPDATE
• DELETE
• DROP
• ALTER
• CREATE
• Other restricted database operations

Therefore, SQL execution is protected by both application-level validation and the SQLite engine.

---

🤖 SQL Generation

The application supports two modes.

Claude Mode

If ANTHROPIC_API_KEY is configured:

ANTHROPIC_API_KEY
       │
       ▼
ClaudeSQLAgent
       │
       ▼
Claude API
       │
       ▼
Generated SQL
       │
       ▼
MCPServer

The Claude agent receives the database schema and instructions for generating safe SQL.

---

Offline Mode

If ANTHROPIC_API_KEY is not configured:

User Question
      │
      ▼
LocalSQLPlanner
      │
      ▼
Pattern Matching
      │
      ▼
SQL Query
      │
      ▼
MCPServer

The offline planner supports common sales queries including:

• Customer counts
• Revenue calculations
• Category revenue
• Top-N products
• Regional aggregations
• Group-by queries
• Products that were never ordered
• Basic filtering
• Aggregations

This means the project can run without an API key or internet connection.

---

📊 Database Schema

The project uses SQLite.

Database location:

data/sales.db

📁 Project Structure

partnr-sales-agent/
│
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── agent.py
│   └── mcp_server.py
│
├── data/
│   └── sales.db
│
├── tests/
│   └── test_query.py
│
├── generate_db.py
├── evaluator.sh
├── Dockerfile
├── docker-compose.yml
├── requirements.txt
├── .env.example
├── .gitignore
└── README.md

---

⚙️ Requirements

Required

• Python 3.10+
• pip
• SQLite

Optional

• Docker Desktop
• Docker Compose
• Anthropic API key

---

🚀 Installation

Windows PowerShell

Step 1 — Open the project

Step 2 — Create virtual environment

python -m venv .venv

Step 3 — Activate virtual environment

.\.venv\Scripts\Activate.ps1

Important: source .venv/bin/activate is a Linux/macOS command. Do not use it in Windows PowerShell.

Step 4 — Upgrade pip

python -m pip install --upgrade pip

Step 5 — Install dependencies

pip install -r requirements.txt

---

🗄️ Generate the Database

If data/sales.db does not exist, run:

python generate_db.py

Verify:

data/
└── sales.db

---

▶️ Run the Application

Start the FastAPI development server:

uvicorn app.main:app --reload

You should see:

Uvicorn running on http://127.0.0.1:8000

The API is now running at:

http://127.0.0.1:8000

---

📖 API Documentation

FastAPI automatically generates interactive documentation.

Open:

http://127.0.0.1:8000/docs

You can use Swagger UI to test the API without Postman.

Alternative documentation:

http://127.0.0.1:8000/redoc

---

🔎 API Usage

POST /query

The endpoint accepts a natural-language question.

Request

{
  "question": "What is the total number of customers?"
}

Example Response

{
  "sql": "SELECT COUNT(*) AS total_customers FROM customers;",
  "results": [
    {
      "total_customers": 500
    }
  ],
  "chart_data": {
    "labels": [
      "500"
    ],
    "values": [
      500
    ]
  }
}

---

🧪 Example Queries

1. Total Customers

What is the total number of customers?

2. Technology Revenue

What is the total revenue from the Technology category?

3. Top Products

What are the top 5 products by revenue?

4. Regional Sales

What is the total sales amount by region?

5. Never Ordered Products

Which products have never been ordered?

6. Unsupported Question

What is the weather today?

The unsupported question should return:

HTTP 400

with an explanatory error message.

---

🧪 Testing

Run the complete test suite:

pytest tests/ -v

The tests cover:

• API endpoint
• Response format
• Customer count
• Revenue queries
• Complex SQL joins
• Unsupported questions
• MCP security
• SQL injection protection
• list_tables()
• describe_schema()
• Read-only SQL execution

All tests use the offline planner and therefore do not require an API key.

---

📋 End-to-End Evaluation

The repository contains:

evaluator.sh

The script executes predefined questions against the API.

Git Bash

chmod +x evaluator.sh
./evaluator.sh

PowerShell

If you are using Git Bash on Windows:

./evaluator.sh

You can also test all queries manually using:

http://127.0.0.1:8000/docs

---

🐳 Docker

Docker can be used instead of installing Python dependencies locally.

Build and start

docker compose up --build -d

If your system uses the older Docker Compose command:

docker-compose up --build -d

Check the containers:

docker compose ps

View logs:

docker compose logs -f api

Open:

http://localhost:8000/docs

---

Run Tests in Docker

docker compose exec api pytest tests/ -v

---

Stop Docker

docker compose down

---

❌ Error Handling

The application does not guess when a question cannot be answered.

For unsupported questions, the agent raises:

UnanswerableQuestionError

The API converts this into:

HTTP 400 Bad Request

Example:

Question:
What is the weather today?

Response:
400 Bad Request

This prevents unrelated questions from producing meaningless SQL.

---

🔒 Security Architecture

The security model follows defense in depth:

Natural Language Question
          │
          ▼
       SQLAgent
          │
          ▼
     Generated SQL
          │
          ▼
   SQL Validation
          │
          ├── Single statement
          ├── SELECT / WITH only
          ├── Forbidden keyword check
          │
          ▼
   SQLite Authorizer
          │
          ├── Reject writes
          ├── Reject DDL
          └── Reject restricted actions
          │
          ▼
      SQLite DB

The important principle is:

The AI agent generates SQL, but it never directly controls the database.

---

🧩 Components

app/main.py

Responsible for:

• FastAPI application
• /query endpoint
• Request validation
• Agent orchestration
• Response formatting
• Error handling

app/agent.py

Responsible for:

• Natural-language processing
• SQL generation
• Claude integration
• Offline SQL planning
• Unsupported-question detection

app/mcp_server.py

Responsible for:

• Database connection
• Table listing
• Schema inspection
• SQL validation
• Read-only enforcement
• SQLite authorizer

generate_db.py

Responsible for:

• Creating the SQLite database
• Generating customers
• Generating orders
• Generating products
• Generating order items

tests/test_query.py

Responsible for:

• API tests
• SQL tests
• Security tests
• Schema tests

---

🛠️ Technology Stack

Technology	Purpose
Python	Application development
FastAPI	REST API
SQLite	Database
Anthropic Claude	Optional AI SQL generation
MCP-style Server	Secure database gateway
Pydantic	Data validation
Pytest	Testing
Docker	Containerization
Docker Compose	Container orchestration

---

📈 Example End-to-End Flow

For the question:

What is the total revenue from the Technology category?

The application performs:

1. User sends question
          ↓
2. FastAPI receives /query
          ↓
3. SQLAgent analyzes question
          ↓
4. SQL is generated
          ↓
5. MCPServer validates SQL
          ↓
6. SQLite authorizer checks operation
          ↓
7. Query executes
          ↓
8. Results are returned
          ↓
9. Chart data is generated

Example SQL:

SELECT
    SUM(p.price * oi.quantity) AS total_revenue
FROM order_items oi
JOIN products p
    ON p.id = oi.product_id
WHERE p.category = 'Technology';

---

🔮 Future Improvements

Potential improvements include:

• Ollama/local LLM integration
• Additional LLM providers
• PostgreSQL support
• Authentication
• Rate limiting
• Query caching
• Conversation history
• Advanced SQL generation
• Automatic chart selection
• Frontend dashboard
• Production logging
• Monitoring
• Streaming responses

---

🎯 Project Objective

The main objective of this project is to demonstrate a secure architecture for querying structured sales data using natural language.

Instead of manually writing SQL:

"What is the total revenue from Technology?"

the user can ask a natural-language question and receive a structured result.

User
 │
 ▼
FastAPI
 │
 ▼
SQL Agent
 │
 ▼
MCPServer
 │
 ▼
SQLite
 │
 ▼
Sales Result

The architecture keeps AI-generated SQL separate from database execution, making the system easier to test, secure, and extend.

---

👩‍💻 Running the Project — Quick Start

For an existing local setup, these are the only commands normally required:

cd D:\partnr-sales-agent

.\.venv\Scripts\Activate.ps1

uvicorn app.main:app --reload

Then open:

http://127.0.0.1:8000/docs

Test:

{
  "question": "What is the total number of customers?"
}

---

📄 License

This project is intended for educational, development, and demonstration purposes.