# MCP SQLite Server (Read-Only)

A production-ready [Model Context Protocol](https://modelcontextprotocol.io/) server that provides AI agents with **safe, read-only access** to a SQLite database (`shop.db`). Built with the official `mcp` Python SDK using the stdio transport.

## Features

- **3 MCP tools**: `list_tables`, `describe_table`, `query_database`
- **Defense-in-depth read-only safety**: SQLite URI read-only mode + `PRAGMA query_only` + SQL validator + EXPLAIN opcode inspection
- **Query validation**: Rejects `INSERT`/`UPDATE`/`DELETE`/`DROP`/`ALTER`/`CREATE`/`REPLACE`/`TRUNCATE`/`ATTACH`/`DETACH`, multi-statement queries (`;`), SQL comments (`--`, `/* */`), and modifying `PRAGMA` — without false positives on string literals
- **Pagination**: Default row limit (100), `limit`/`offset` parameters, truncated-output flag
- **Stderr-only logging**: All logs/tracebacks go to `sys.stderr`; `stdout` is reserved exclusively for JSON-RPC
- **Full type hints**: `mypy --strict` clean
- **TDD**: 105 tests covering security, DB layer, MCP tools, 8 benchmark queries, and the stderr guard

## Quick Start

### Prerequisites

- Python 3.10+
- A SQLite database file (default: `./shop.db`)

### Local Setup

```bash
python -m venv .venv
source .venv/bin/activate
pip install -e ".[dev]"
```

### Configure

Copy `.env.example` and set the database path:

```bash
cp .env.example .env
# Edit DATABASE_PATH to point to your SQLite file
```

Or set the environment variable directly:

```bash
export DATABASE_PATH=/abs/path/to/shop.db
```

### Run the Server

```bash
python -m mcp_server.server
```

The server communicates over stdin/stdout using the MCP stdio transport. You don't interact with it directly — an MCP client (e.g., Claude Desktop, your AI agent) connects to it.

## MCP Client Configurations

### Standard Python

Add this to your MCP client configuration (e.g., Claude Desktop's `claude_desktop_config.json`):

```json
{
  "mcpServers": {
    "sqlite-shop": {
      "command": "python",
      "args": ["-m", "mcp_server.server"],
      "env": {
        "DATABASE_PATH": "/abs/path/to/shop.db"
      }
    }
  }
}
```

### Docker

First build the image:

```bash
docker build -t mcp-shop:latest .
```

Then configure your MCP client:

```json
{
  "mcpServers": {
    "sqlite-shop": {
      "command": "docker",
      "args": [
        "run", "-i", "--rm",
        "-v", "/abs/path/to/shop.db:/app/shop.db",
        "-e", "DATABASE_PATH=/app/shop.db",
        "mcp-shop:latest"
      ]
    }
  }
}
```

### Docker Compose

```bash
docker compose up -d
```

## Tools

### `list_tables`

Lists all user tables and views in the database (excludes internal `sqlite_*` tables).

**Parameters**: none

**Returns**:
```json
{
  "tables": ["customers", "orders", "order_items", "products"],
  "count": 4
}
```

### `describe_table`

Describes the schema of a table: columns, foreign keys, row count, and the CREATE statement.

**Parameters**:
- `table` (string, required): Name of the table to describe.

**Returns**:
```json
{
  "table": "customers",
  "columns": [
    {"cid": 0, "name": "id", "type": "INTEGER", "notnull": 0, "default": null, "pk": 1},
    {"cid": 1, "name": "first_name", "type": "TEXT", "notnull": 1, "default": null, "pk": 0}
  ],
  "foreign_keys": [],
  "row_count": 150,
  "sql": "CREATE TABLE customers (...)"
}
```

### `query_database`

Executes a read-only SQL query with pagination support.

**Parameters**:
- `sql` (string, required): A single read-only SQL statement (`SELECT`, `WITH`, `EXPLAIN`, or read-only `PRAGMA`).
- `limit` (integer, optional): Maximum rows to return. Default: 100. Max: 1000.
- `offset` (integer, optional): Number of rows to skip. Default: 0.

**Returns**:
```json
{
  "columns": ["id", "first_name"],
  "rows": [{"id": 1, "first_name": "Alice"}, {"id": 2, "first_name": "Bob"}],
  "row_count": 2,
  "truncated": false,
  "limit": 100,
  "offset": 0
}
```

When `truncated` is `true`, more rows are available — increase `offset` to fetch the next page.

## Security

The server implements defense-in-depth to guarantee read-only access:

### Layer 1: SQLite Connection (URI read-only mode)

The database is opened with `file:<path>?mode=ro`, which prevents writes at the SQLite engine level. Additionally, `PRAGMA query_only = ON` is set on every connection.

### Layer 2: SQL Query Validator (`security.py`)

Before any query reaches SQLite, it passes through a multi-stage validator:

1. **String literal stripping**: String literals (`'...'`, `"..."`) are replaced with placeholders so keywords inside data (e.g., a product named "Deleted Item") don't trigger false positives.
2. **Comment detection**: SQL comments (`--`, `/* */`) are rejected to prevent comment-based bypasses.
3. **Multi-statement rejection**: Any semicolon (`;`) is rejected, preventing stacked queries.
4. **Keyword analysis**: The first real statement keyword must be `SELECT`, `WITH`, `EXPLAIN`, or `PRAGMA`. Destructive keywords (`INSERT`, `UPDATE`, `DELETE`, `DROP`, `ALTER`, `CREATE`, `REPLACE`, `TRUNCATE`, `ATTACH`, `DETACH`, `VACUUM`, etc.) are blocked.
5. **PRAGMA validation**: Read-only PRAGMAs (`table_info`, `database_list`, etc.) are allowed. Any PRAGMA with an assignment (`=`) or in the mutating-PRAGMA blocklist (`journal_mode`, `synchronous`, `foreign_keys`, etc.) is rejected.

### Layer 3: EXPLAIN Opcode Inspection

As a final defense, the query is run through SQLite's own parser via `EXPLAIN <query>`. The resulting opcode stream is inspected for write opcodes (`OpenWrite`, `Insert`, `Delete`, `Create`, `Drop`, etc.) and write-transaction flags. If any are found, the query is rejected.

### Layer 4: Sanitized Error Messages

All errors returned to the client are sanitized — filesystem paths and internal details are stripped to prevent information leakage.

## Testing

Tests use temporary/in-memory databases only — never the production `shop.db`.

```bash
# Run all tests
python -m pytest

# Run with verbose output
python -m pytest -v

# Run a specific test file
python -m pytest tests/test_security.py
```

### Test Coverage

| Test File | Coverage |
|-----------|----------|
| `tests/test_security.py` | 76 tests: valid queries, destructive statement rejection, PRAGMA validation, multi-statement rejection, comment bypass prevention, string literal handling |
| `tests/test_db.py` | 20 tests: read-only enforcement, table listing, schema description, pagination, truncation, all 8 benchmark queries |
| `tests/test_server.py` | 9 tests: MCP tool discovery, tool calls via SDK client, destructive query rejection, pagination, 7 benchmark queries via tools, stderr/no-stdout-pollution guard |

## Static Analysis

```bash
# Type checking
python -m mypy

# Linting
python -m ruff check src/ tests/
```

## Project Structure

```
.
├── .env.example          # Environment variable template
├── Dockerfile            # Docker containerization
├── docker-compose.yml    # Docker Compose config
├── pyproject.toml        # Package config, deps, tool settings
├── README.md             # This file
├── shop.db               # The SQLite database (not included in tests)
├── src/mcp_server/
│   ├── __init__.py
│   ├── config.py         # Configuration (DATABASE_PATH, limits, URI builder)
│   ├── db.py             # Read-only Database class with introspection + query
│   ├── security.py       # SQL validator (multi-layer defense-in-depth)
│   ├── server.py         # MCP server entrypoint (stdio transport)
│   ├── tools.py          # MCP tool definitions and handlers
│   └── py.typed          # PEP 561 marker
└── tests/
    ├── __init__.py
    ├── test_db.py        # Database layer + benchmark tests
    ├── test_security.py  # Query validator tests
    └── test_server.py    # MCP server/tool tests
```

## Benchmark Tasks

The server's tools enable an AI agent to perform these analytical tasks (validated by tests against a controlled fixture database):

1. **Table Discovery**: `list_tables` + `describe_table` — list all tables and describe schemas.
2. **Filtered Count**: `query_database` with `SELECT COUNT(*) FROM customers WHERE country = 'Germany'`.
3. **Country Aggregation**: `SELECT country, COUNT(*) ... GROUP BY country ORDER BY ... DESC LIMIT 1`.
4. **Customer LTV**: Join `customers` + `orders`, `SUM(total_amount)`, order by total.
5. **Product Performance**: Join `order_items` + `products`, aggregate by quantity and revenue, `LIMIT 5`.
6. **Category Aggregation**: Traverse `order_items` → `products` → `category`, aggregate revenue, `LIMIT 3`.
7. **Date Filtering**: `SUM(total_amount) WHERE substr(order_date,1,4) = '2025'`.
8. **Order Aggregation**: Join `customers` + `orders`, `COUNT(o.id)`, order by count.

## Configuration

| Environment Variable | Default | Description |
|---------------------|---------|-------------|
| `DATABASE_PATH` | `./shop.db` | Path to the SQLite database file |
| `ROW_LIMIT` | `100` | Default row limit for query results (max 1000) |

## License

This project is provided as-is for demonstration purposes.