# Local Lakehouse + MCP Server

A fully local (zero-cloud-cost) data lakehouse: synthetic e-commerce events
land as partitioned Parquet in a **Bronze** layer, get cleaned and
contract-tested into **Silver**, aggregated into business-ready **Gold**
tables in DuckDB, and exposed live to Claude through an **MCP server** so
Claude can query the pipeline autonomously.

## Architecture

```mermaid
flowchart LR
    A["ingest.py<br/>(Faker-driven event generator)"] --> B["Bronze<br/>data/bronze/date=YYYY-MM-DD/*.parquet"]
    B --> C["Silver<br/>stg_transactions (dbt view)<br/>clean · dedupe · cast · filter"]
    C --> D1["Gold: fct_daily_sales<br/>(dbt table)"]
    C --> D2["Gold: dim_users<br/>(dbt table)"]
    D1 --> E["lakehouse.duckdb"]
    D2 --> E
    E --> F["server.py<br/>MCP Server"]
    F --> G["Claude Desktop / Claude"]

    style A  fill:#e8f0fe,stroke:#4285f4
    style B  fill:#fff3cd,stroke:#c9a227
    style C  fill:#e6d9f5,stroke:#8a4fd1
    style D1 fill:#d9f2e3,stroke:#2fa565
    style D2 fill:#d9f2e3,stroke:#2fa565
    style E  fill:#d9f2e3,stroke:#2fa565
    style F  fill:#fde3e3,stroke:#e05555
    style G  fill:#e0f0ff,stroke:#2b7fd6
```

**14 dbt data-quality tests enforced:** `transaction_id` uniqueness/not-null,
`platform` and `region` accepted-value sets, not-null checks on `amount`,
`user_id`, `transaction_date`, and `lifetime_value`.

## Project layout

```
lakehouse-mcp-project/
├── ingest.py                          # Bronze layer generator
├── data/bronze/date=.../*.parquet     # Bronze output (300 k rows, 45 partitions)
├── lakehouse_project/                 # dbt project
│   ├── dbt_project.yml
│   ├── profiles.yml                   # points at ../lakehouse.duckdb
│   └── models/
│       ├── staging/stg_transactions.sql
│       ├── marts/fct_daily_sales.sql
│       ├── marts/dim_users.sql
│       └── schema.yml                 # dbt tests / data contracts
├── lakehouse.duckdb                   # Materialised Silver/Gold warehouse
├── server.py                          # MCP server (3 tools)
├── claude_desktop_config.example.json
└── requirements.txt
```

## Setup

```bash
python3 -m venv env
source env/bin/activate        # Windows: env\Scripts\activate
pip install -r requirements.txt
```

## Step 1 — Regenerate the Bronze layer

```bash
python ingest.py
```

Generates 300,000 transactions across 6,000 users / 45 days with realistic
structure: each user has a permanent region and platform preference, purchase
volume follows a day-of-week cycle per region, spend follows a log-normal
distribution per platform, and ~0.7 % of rows are deliberately dirty
(negative amounts, null platforms, duplicate IDs) so the Silver-layer tests
have real signal to catch.

## Step 2 — Build & test the Silver/Gold layers

```bash
cd lakehouse_project
export DBT_PROFILES_DIR=$(pwd)     # Windows (PowerShell): $env:DBT_PROFILES_DIR = (Get-Location)
dbt build                          # runs models AND the 14 data-quality tests
```

Expect `Done. PASS=17 ... ERROR=0 SKIP=0`. Query the results directly:

```bash
python3 -c "
import duckdb
con = duckdb.connect('lakehouse.duckdb')
print(con.sql('SELECT * FROM fct_daily_sales ORDER BY transaction_date DESC LIMIT 5').df())
"
```

## Step 3 — Run the MCP server

```bash
python server.py       # starts a stdio MCP server
```

**Connect it to Claude Desktop:** copy `claude_desktop_config.example.json`'s
`lakehouse-mcp` entry into your own `~/Library/Application Support/Claude/claude_desktop_config.json`
(macOS) or `%APPDATA%\Claude\claude_desktop_config.json` (Windows), updating
the paths to match your environment. Restart Claude Desktop, then try:

> "Check the pipeline health, then tell me the top 3 regions by revenue."

Claude will call `check_pipeline_health`, then `get_schema_info`, then
`execute_sql` — no manual SQL required.

### The 3 exposed MCP tools

| Tool | Purpose |
| --- | --- |
| `get_schema_info()` | Describes Gold tables (columns, types, row counts) |
| `execute_sql(query)` | Runs a read-only `SELECT`/`WITH` query against DuckDB, returns markdown |
| `check_pipeline_health()` | Runs `dbt test` and reports pass/fail |

`execute_sql` rejects any query containing write/DDL keywords
(`INSERT`/`UPDATE`/`DELETE`/`DROP`/etc.) and connects read-only — Claude
can analyse the warehouse but cannot mutate it.

## MCP SDK version note

Built against `mcp==2.1.1`, where the `FastMCP` class from `mcp` v1 was
renamed to `MCPServer` (imported from `mcp.server.mcpserver`). If your
environment has `mcp<2` installed, change the import in `server.py` to:

```python
from mcp.server.fastmcp import FastMCP as MCPServer
```

The rest of the decorator-based `@mcp.tool()` API is unchanged between versions.

## Extending the project

- Add a `sensor.py` that appends a new day of Bronze data on a cron/timer to
  simulate a live feed.
- Add `dbt_utils`-style tests (surrogate key checks, freshness assertions) as
  the next layer of data quality enforcement.
- Add a 4th MCP tool, e.g. `explain_query_plan(query)`, wrapping DuckDB's
  `EXPLAIN ANALYZE`.