Build a DuckDB Sales MCP Server

Build an MCP server that analyzes a local sales CSV with DuckDB SQL.

Introduction

30 Second Summary

Sales totals can look convincing while a missing label quietly changes the story. A regional report is only useful when every sale lands in a group you can explain.

In this project, you will build a local sales analyst that uses DuckDB to run SQL directly over a CSV file. MCP Inspector displays each structured result from the analyst's MCP server.

What You'll Build

Inside MCP Inspector, your 20-row sales file becomes headline totals, ranked regional revenue, a visible quality check, and a ready-to-use reporting prompt.

By the end of this project, you'll have:

  • Headline sales metrics that show 20 rows, 169 units, 10460 in total revenue, and 523.0 in average order revenue from one tool call.
  • Trustworthy regional rankings that expose one missing region before labeling it as Unknown in the final output.
  • Reusable analysis building blocks that reveal live column names and inferred DuckDB types. You can also render an executive-summary workflow for sales leadership.
  • Secret Mission: Add a product filter with a safely parameterized query that returns stable metrics for Monitor and unknown products.

Are there any prerequisites?

Basic Python and SQL familiarity is enough for this project.

You need Visual Studio Code, Node.js, Python 3.11 or newer, and uv installed locally.

Before We Start

Before the hands-on work begins, lock in what your local sales analyst will examine. This gives you a clear reason to detect missing dimensions before presenting grouped results.

Set Up the Local Analytics Project

A reproducible environment prevents dependency drift. A fixed dataset keeps every later SQL result predictable.

You will create a controlled workspace in Visual Studio Code with pinned DuckDB plus MCP dependencies. Windows PowerShell will verify that the managed Python environment satisfies version 3.11 or later.

In this step, get ready to:
  • Create the duckdb-mcp-analyst project structure in Visual Studio Code.
  • Populate the project artifacts with pinned configuration plus fixed sales data.
  • Sync the managed environment before confirming both packages import successfully.
Create the project structure

The workspace gives every file one predictable home. Placing it on your Desktop also makes the project easy to find from PowerShell.

  • Press the Windows key to open Windows search.
  • Type Visual Studio Code into the search field.
  • Press Enter to open Visual Studio Code.

You will see the Visual Studio Code welcome window. The project folder comes next.

Visual Studio Code may show a workspace trust prompt after the folder opens. Trusting this folder enables the integrated tools used in the project.

  • Click File in the top menu bar.
  • Select Open Folder....
  • Select your Desktop in the folder dialog.
  • Click New Folder.
  • Enter duckdb-mcp-analyst as the folder name.
  • Press Enter to create the folder.
  • Click Select Folder to use it as your workspace.
  • Choose the trust option if Visual Studio Code shows the workspace trust prompt.

The Explorer sidebar now shows duckdb-mcp-analyst as the top-level folder. Your project now has a dedicated workspace.

  • Click the New File icon in the Explorer toolbar.
  • Enter pyproject.toml in the file name field.
  • Click the New File icon in the Explorer toolbar.
  • Enter server.py in the file name field.

The Explorer now shows both top-level files. Selecting server.py should display a blank editor.

  • Click the New Folder icon in the Explorer toolbar.
  • Enter data in the folder name field.

The Explorer now shows a data folder beneath the two top-level files.

  • Select the data folder in the Explorer sidebar.
  • Click the New File icon in the Explorer toolbar.
  • Enter sales.csv in the file name field.

The Explorer now shows the complete project skeleton. The data folder contains sales.csv.

Add the pinned dependencies and sample data

The pyproject.toml file sets the managed Python requirement. It also pins every package version used by the project.

Why DuckDB for this project?

DuckDB runs SQL directly over the local CSV inside the Python process. This keeps your attention on analytics instead of database deployment.

Pandas is useful for DataFrame-focused work. DuckDB supports the SQL-first workflow used throughout this project.

  • Select pyproject.toml in the Explorer sidebar.
  • Populate the project configuration by pasting the code below:
[project]
name = "duckdb-mcp-analyst"
version = "0.1.0"
description = "A local MCP server for SQL analytics over a sales CSV."
requires-python = ">=3.11"
dependencies = [
    "duckdb==1.5.6",
    "mcp[cli]==2.3.0",
]

What Does This Configuration Do?

  • The project name identifies the environment as duckdb-mcp-analyst.
  • The requires-python setting requires Python 3.11 or later.
  • The dependency list pins DuckDB to 1.5.6.
  • The dependency list pins the MCP Python SDK CLI extra to 2.3.0.
  • Press Ctrl+S to save pyproject.toml.
  • Check the final line to confirm the dependency list has a closing square bracket.

Seeing a Configuration Problem?

  • Check that each dependency remains inside double quotation marks.
  • Check that the dependency list has one opening square bracket plus one closing square bracket.

Help me find the formatting problem in my project configuration.

The fixed CSV gives every learner the same 20 sales rows. Its literal NULL region creates the data-quality problem investigated later.

  • Select data/sales.csv in the Explorer sidebar.
  • Populate the sample dataset by pasting the content below:
date,region,product,units,revenue
2026-01-05,North,Laptop,12,360
2026-01-07,South,Headphones,8,640
2026-01-12,West,Keyboard,15,450
2026-01-18,East,Monitor,5,1250
2026-02-02,North,Monitor,4,1000
2026-02-10,NULL,Mouse,6,480
2026-02-16,South,Speakers,10,300
2026-02-25,West,Monitor,3,750
2026-03-03,East,Mouse,9,270
2026-03-08,North,Keyboard,7,560
2026-03-14,South,Speakers,6,480
2026-03-21,West,Notebook,10,300
2026-03-28,East,Headphones,9,720
2026-04-02,North,Keyboard,14,420
2026-04-07,South,Laptop,11,330
2026-04-12,West,Speakers,5,400
2026-04-18,East,Mouse,8,240
2026-04-21,North,Headphones,10,800
2026-04-25,South,Keyboard,4,320
2026-04-29,West,Notebook,13,390

What Does This Dataset Contain?

  • The header defines five fields used by the later SQL queries.
  • The dataset contains 20 sales records across four months.
  • The product column includes Notebook, Headphones, Monitor, Mouse, Keyboard, Laptop, and Speakers.
  • The row dated 2026-02-10 uses the literal NULL as its region value.
  • Press Ctrl+S to save data/sales.csv.
  • Count the records to confirm the header is followed by 20 sales rows.
  • Check the row dated 2026-02-10 to confirm its region remains NULL.

Dataset Rows Not Lining Up?

  • Check that each sales record occupies one line.
  • Check that each line contains five comma-separated values.
  • Check that the file contains one header row plus 20 sales rows.

Help me compare my sales dataset with the supplied 20-row CSV.

Use this checkpoint to compare the three project artifacts before syncing the environment.

✔️ Awesome, I've got everything!

Your pinned project metadata is saved. Your 20-row dataset is saved inside data.

The empty server.py file is ready for the MCP server code added in the next step.

ⓧ I'd like to double check the full code

  • Compare your complete pyproject.toml with this reference:
[project]
name = "duckdb-mcp-analyst"
version = "0.1.0"
description = "A local MCP server for SQL analytics over a sales CSV."
requires-python = ">=3.11"
dependencies = [
    "duckdb==1.5.6",
    "mcp[cli]==2.3.0",
]

What Should Match?

Your file should contain this project metadata plus these two pinned dependencies. The Python requirement should remain >=3.11.

  • Compare your complete data/sales.csv with this reference:
date,region,product,units,revenue
2026-01-05,North,Laptop,12,360
2026-01-07,South,Headphones,8,640
2026-01-12,West,Keyboard,15,450
2026-01-18,East,Monitor,5,1250
2026-02-02,North,Monitor,4,1000
2026-02-10,NULL,Mouse,6,480
2026-02-16,South,Speakers,10,300
2026-02-25,West,Monitor,3,750
2026-03-03,East,Mouse,9,270
2026-03-08,North,Keyboard,7,560
2026-03-14,South,Speakers,6,480
2026-03-21,West,Notebook,10,300
2026-03-28,East,Headphones,9,720
2026-04-02,North,Keyboard,14,420
2026-04-07,South,Laptop,11,330
2026-04-12,West,Speakers,5,400
2026-04-18,East,Mouse,8,240
2026-04-21,North,Headphones,10,800
2026-04-25,South,Keyboard,4,320
2026-04-29,West,Notebook,13,390

What Should Match?

Your file should contain this header plus these 20 records. The row dated 2026-02-10 should retain its literal NULL region.

The complete server.py artifact is an empty file at this point.

Sync and verify the managed environment

The saved configuration becomes a runnable environment when uv resolves the pinned packages. A successful sync creates the project-local .venv environment plus the uv.lock lockfile.

  • Click View in the Visual Studio Code menu bar.
  • Select Terminal to reveal the integrated terminal.
  • Open the terminal profile menu if the active shell is different from PowerShell.
  • Select PowerShell from the profile menu.
  • Check the prompt to confirm its path ends with duckdb-mcp-analyst.

PowerShell is now open at the project root. Package download activity during the first sync means the local environment is being assembled.

  • Create the managed environment plus lockfile by running this command:
uv sync

What Does This Command Do?

The sync reads the Python requirement plus the dependency pins from pyproject.toml. It installs the resolved packages into the project-local environment.

The generated lockfile records the resolved dependency set. Future syncs can reproduce that environment.

  • Check the Explorer sidebar to confirm uv.lock now exists.
  • Check the Explorer sidebar to confirm the .venv folder now exists.

Did the Sync Fail?

  • Check that the terminal path ends with duckdb-mcp-analyst.
  • Check pyproject.toml for missing quotation marks or brackets.
  • Read the terminal output for a Python version mismatch if the managed environment cannot satisfy >=3.11.

Help me diagnose my failed project sync.

Good progress. Your project now has an isolated environment plus a reproducible dependency lockfile.

Before you run the final check, do you expect both installed packages to import successfully inside the managed environment?

  • Verify the managed Python environment plus both imports by running this command:
uv run python -c "import duckdb; import mcp; print('Dependencies ready')"

What Should I See?

You should see Dependencies ready in the terminal. This confirms that DuckDB plus the MCP package import inside the managed project environment.

The successful run also confirms that the managed environment satisfies the project's Python >=3.11 requirement.

That is the setup complete. Your controlled dataset plus pinned analytics environment are ready to produce reliable results.

Imports Still Failing?

  • Check that PowerShell is still at the folder containing pyproject.toml.
  • Run the dependency sync again if either package is missing from the managed environment.
  • Check the first terminal error for the package that failed to import.

Help me diagnose the failed dependency import.

Your local analytics project is ready. Next, you will connect DuckDB to the CSV and return the first headline sales metrics through MCP.

Return the Headline Sales Numbers

Your pinned environment is ready. The 20-row sales dataset now gives your local analyst a predictable starting point.

Before adding dimensions or quality rules, you need a visible baseline. That baseline proves the dataset can produce useful analytics.

DuckDB reads the CSV through SQL. MCP returns the headline numbers as structured data.

MCP Inspector provides the local client you need to call the tool and inspect its response.

In this step, get ready to:
  • Create an MCP server that opens the sales data through an in-memory DuckDB connection.
  • Add a typed overview tool that calculates four headline sales metrics.
  • Verify the structured metrics through MCP Inspector.
Create the MCP server connection

An in-memory connection reads the sales file without deploying a separate database. The helper creates a fresh connection whenever an analytics tool needs the data.

  • In Visual Studio Code, select server.py from the left file sidebar.
  • Build the server foundation by pasting this code into the empty server.py file:
from pathlib import Path

import duckdb
from mcp.server import MCPServer

PROJECT_ROOT = Path(__file__).parent
DATA_FILE = PROJECT_ROOT / "data" / "sales.csv"

mcp = MCPServer("DuckDB Sales Analyst")


def sales_connection():
    connection = duckdb.connect()
    connection.read_csv(str(DATA_FILE), na_values=["NULL"]).create_view("sales")
    return connection

What does this code do?

  • PROJECT_ROOT anchors file paths to the folder containing server.py.
  • DATA_FILE points to the existing data/sales.csv dataset.
  • MCPServer creates the server named DuckDB Sales Analyst.
  • sales_connection() opens a new in-memory database connection for each request.
  • na_values=["NULL"] makes the literal missing-region value behave like a SQL null.
  • create_view("sales") exposes the CSV relation as the sales view.
  • Save server.py.

The first Inspector launch can take about a minute while npx prepares the local app.

  • Test the server foundation from the existing PowerShell terminal by running:
uv run mcp dev server.py

What does this command do?

The command loads server.py inside the uv-managed environment. It starts the local server through MCP Inspector.

You'll see the Inspector page open. This confirms that the server foundation loads without a Python error.

  • Stop this temporary server by pressing Ctrl+C in PowerShell.

Inspector did not open?

  • Paste the printed Inspector URL into your browser if the page does not open automatically.
  • Confirm that npx is available if PowerShell cannot launch Inspector.
  • Check the indentation in sales_connection() if PowerShell reports a Python error.

Help me diagnose why my local MCP server is not opening in Inspector.

Add the sales overview tool

The overview query compresses all 20 rows into four headline metrics. A typed dictionary gives MCP clients stable field names for every result.

  • Append the overview tool below sales_connection() in server.py by pasting this code:
@mcp.tool()
def get_sales_overview() -> dict[str, int | float]:
    """Return headline row, unit, and revenue metrics for the sales dataset."""
    with sales_connection() as connection:
        row = connection.execute(
            """
            SELECT
                COUNT(*) AS row_count,
                SUM(units) AS total_units,
                SUM(revenue) AS total_revenue,
                AVG(revenue) AS average_revenue
            FROM sales
            """
        ).fetchone()

    if row is None:
        raise RuntimeError("The sales overview query returned no result.")

    return {
        "row_count": int(row[0]),
        "total_units": int(row[1]),
        "total_revenue": int(row[2]),
        "average_revenue": float(row[3]),
    }

How does the overview work?

  • @mcp.tool() exposes the typed Python function as an MCP tool.
  • COUNT measures how many sales rows the dataset contains.
  • SUM calculates the unit and revenue totals.
  • AVG calculates the average revenue per row.
  • fetchone() retrieves the single aggregate result as a tuple.
  • The empty-result check prevents the tool from returning an invalid response.
  • The final dictionary converts each tuple position into a named integer or floating-point value.
  • Save server.py.

✔️ Awesome, I've got everything!

Your server.py now contains the complete MCP server code for this step.

ⓧ I'd like to double check the full code

from pathlib import Path

import duckdb
from mcp.server import MCPServer

PROJECT_ROOT = Path(__file__).parent
DATA_FILE = PROJECT_ROOT / "data" / "sales.csv"

mcp = MCPServer("DuckDB Sales Analyst")


def sales_connection():
    connection = duckdb.connect()
    connection.read_csv(str(DATA_FILE), na_values=["NULL"]).create_view("sales")
    return connection


@mcp.tool()
def get_sales_overview() -> dict[str, int | float]:
    """Return headline row, unit, and revenue metrics for the sales dataset."""
    with sales_connection() as connection:
        row = connection.execute(
            """
            SELECT
                COUNT(*) AS row_count,
                SUM(units) AS total_units,
                SUM(revenue) AS total_revenue,
                AVG(revenue) AS average_revenue
            FROM sales
            """
        ).fetchone()

    if row is None:
        raise RuntimeError("The sales overview query returned no result.")

    return {
        "row_count": int(row[0]),
        "total_units": int(row[1]),
        "total_revenue": int(row[2]),
        "average_revenue": float(row[3]),
    }
Call the tool in MCP Inspector

The completed server now has one callable analytics tool. Inspector lets you invoke that tool without building a separate client.

Before you relaunch, consider whether Inspector will list the new overview tool.

  • Launch the completed server from the existing PowerShell terminal by running:
uv run mcp dev server.py

What happens this time?

The command loads the saved tool definition into the running server. Inspector now provides a client for calling get_sales_overview.

  • Switch to the Inspector browser tab opened by the command.
  • Confirm that Inspector is connected to the preconfigured local server.
  • Open the server's tools view.
  • Select get_sales_overview from the available tools.

Before you call it, consider which four values the 20 rows should produce.

  • Call get_sales_overview from the selected tool panel.

You'll see row_count as 20. You'll see total_units as 169.

You'll see total_revenue as 10460. You'll see average_revenue as 523.0.

That's your first analytics tool working. The server now turns the CSV into structured MCP results.

Tool missing or numbers different?

  • Stop the server with Ctrl+C if get_sales_overview is missing.
  • Save server.py before relaunching the command above.
  • Compare data/sales.csv with the 20-row dataset from Step 1 if any metric differs.

Help me debug my MCP overview tool or its sales totals.

Your local analyst now returns a trustworthy headline baseline. Next, you'll break the revenue down by region.

Break Revenue Down by Region

Your MCP overview tool now proves DuckDB can read the CSV. However, its whole-dataset totals hide where revenue comes from.

A grouped SQL query turns those totals into a regional ranking. Testing the raw dimension shows whether the report is ready for decision-makers.

In this step, get ready to:
  • Add a regional revenue tool to server.py.
  • Rank raw region groups by total revenue.
  • Inspect every returned group for reporting readiness.
Add the regional ranking tool

A grouped query combines rows that share the same region. Sorting the results by revenue turns those groups into a useful ranking.

  • In server.py, paste the following code immediately below get_sales_overview():
@mcp.tool()
def revenue_by_region() -> list[dict[str, str | int]]:
    """Rank regions by total revenue."""
    with sales_connection() as connection:
        rows = connection.execute(
            """
            SELECT
                region,
                SUM(units) AS total_units,
                SUM(revenue) AS total_revenue
            FROM sales
            GROUP BY region
            ORDER BY total_revenue DESC
            """
        ).fetchall()

    return [
        {
            "region": region,
            "total_units": int(total_units),
            "total_revenue": int(total_revenue),
        }
        for region, total_units, total_revenue in rows
    ]

How does the regional ranking work?

  • GROUP BY region keeps each distinct raw region value in its own result row.
  • SUM(units) calculates units for each group.
  • SUM(revenue) calculates revenue for each group.
  • Descending revenue order puts the largest group first.
  • The list comprehension maps every DuckDB tuple to a dictionary.
  • Save server.py.
  • Return to MCP Inspector from earlier.
  • Reconnect the Inspector session to server.py.
  • View the server's available tools.

You'll now see revenue_by_region alongside get_sales_overview.

Regional tool missing?

  • Confirm that server.py is saved if the tool list still shows only get_sales_overview.
  • Reconnect the Inspector session so it reloads the server definition.

Help me find why the regional tool is missing.

✔️ Awesome, I've got everything!

Your saved file now contains both analytics tools.

ⓧ I'd like to double check the full code

  • Compare server.py with the complete file below.
from pathlib import Path

import duckdb
from mcp.server import MCPServer

PROJECT_ROOT = Path(__file__).parent
DATA_FILE = PROJECT_ROOT / "data" / "sales.csv"

mcp = MCPServer("DuckDB Sales Analyst")


def sales_connection():
    connection = duckdb.connect()
    connection.read_csv(str(DATA_FILE), na_values=["NULL"]).create_view("sales")
    return connection


@mcp.tool()
def get_sales_overview() -> dict[str, int | float]:
    """Return headline row, unit, and revenue metrics for the sales dataset."""
    with sales_connection() as connection:
        row = connection.execute(
            """
            SELECT
                COUNT(*) AS row_count,
                SUM(units) AS total_units,
                SUM(revenue) AS total_revenue,
                AVG(revenue) AS average_revenue
            FROM sales
            """
        ).fetchone()

    if row is None:
        raise RuntimeError("The sales overview query returned no result.")

    return {
        "row_count": int(row[0]),
        "total_units": int(row[1]),
        "total_revenue": int(row[2]),
        "average_revenue": float(row[3]),
    }


@mcp.tool()
def revenue_by_region() -> list[dict[str, str | int]]:
    """Rank regions by total revenue."""
    with sales_connection() as connection:
        rows = connection.execute(
            """
            SELECT
                region,
                SUM(units) AS total_units,
                SUM(revenue) AS total_revenue
            FROM sales
            GROUP BY region
            ORDER BY total_revenue DESC
            """
        ).fetchall()

    return [
        {
            "region": region,
            "total_units": int(total_units),
            "total_revenue": int(total_revenue),
        }
        for region, total_units, total_revenue in rows
    ]
Inspect the raw regional results

The tool now has everything it needs to rank the raw region values. Before you run it, predict whether every returned group will have a named region.

  • Select revenue_by_region from the available tools in Inspector.
  • Run the selected tool using Inspector's tool control.

You'll see five ranked groups. The fifth group has region set to null.

Its total_revenue is 480. You caught the exact reporting gap the raw query was meant to reveal.

Why did null become a region?

sales_connection() interprets the literal NULL field as missing. Grouping the untouched region column keeps that missing value in its own bucket.

This fifth bucket is the designed obstacle. It proves the ranking needs a data-quality rule before presentation.

Your raw grouped report now reveals an unnamed revenue bucket. Next, you'll quantify missing values before labeling the regional output honestly.

Diagnose and Fix the Data-Quality Gap

Your DuckDB regional tool now ranks revenue. MCP Inspector exposed the weak point through a fifth region bucket whose value is null.

A grouped report loses credibility when a missing dimension appears as an unnamed bucket. The server needs evidence that explains the gap before it changes the label.

This step exposes the dataset schema through an MCP resource. It uses SQL to quantify each missing field before labeling the missing region as Unknown.

In this step, get ready to:
  • Expose the live CSV schema through a dataset metadata resource.
  • Count missing values in the important sales fields.
  • Label missing regions honestly in the regional ranking.
Expose live dataset metadata

A metadata resource lets clients inspect the dataset before trusting its calculations. This resource reads column names plus inferred types directly from the live sales view.

  • Switch back to the server.py tab from earlier.
  • Locate the blank line immediately before the existing get_sales_overview() tool.
  • Add the metadata resource in that blank line by pasting the code below:
@mcp.resource("dataset://sales/metadata", mime_type="application/json")
def dataset_metadata() -> dict[str, object]:
    """Column names and inferred DuckDB types for the sales dataset."""
    with sales_connection() as connection:
        sales = connection.table("sales")
        columns = [
            {"name": name, "type": str(data_type)}
            for name, data_type in zip(sales.columns, sales.dtypes)
        ]

    return {
        "dataset": "sales",
        "source": str(DATA_FILE.relative_to(PROJECT_ROOT)),
        "columns": columns,
    }

What does this resource do?

  • The @mcp.resource() decorator exposes dataset context separately from callable tools.
  • The connection.table("sales") expression reads the existing view as a relation.
  • The columns list pairs every column name with its inferred data type.
  • The relative source path keeps the metadata tied to the duckdb-mcp-analyst folder.
  • Save server.py.
  • Return to MCP Inspector from earlier.
  • Reconnect its existing session to reload the saved server.
  • Switch to the resource view.
  • Select dataset://sales/metadata.
  • Read the selected resource.

You'll see data/sales.csv as the source. You'll also see five column entries with inferred DuckDB types.

Good progress. Your server can now explain the dataset it analyzes.

Resource missing from Inspector?

Confirm that dataset_metadata() sits above get_sales_overview() in server.py. Save the file before reconnecting Inspector.

Check that the resource URI is exactly dataset://sales/metadata.

Help me debug the missing dataset resource.

Quantify missing values

A quality check turns the suspicious regional bucket into a measurable result. COUNT(*) counts every row while COUNT(column) counts rows whose chosen column is present.

Subtracting those values reveals how many entries are missing from each field. The same query checks the dimension fields plus the revenue measure.

  • Scroll to the end of the existing revenue_by_region() function in server.py.
  • Add the quality tool below that function by pasting this code:
@mcp.tool()
def check_data_quality() -> dict[str, int]:
    """Count missing values in important sales dimensions and measures."""
    with sales_connection() as connection:
        row = connection.execute(
            """
            SELECT
                COUNT(*) AS row_count,
                COUNT(*) - COUNT(region) AS missing_region,
                COUNT(*) - COUNT(product) AS missing_product,
                COUNT(*) - COUNT(revenue) AS missing_revenue
            FROM sales
            """
        ).fetchone()

    if row is None:
        raise RuntimeError("The data-quality query returned no result.")

    return {
        "row_count": int(row[0]),
        "missing_region": int(row[1]),
        "missing_product": int(row[2]),
        "missing_revenue": int(row[3]),
    }

How does the quality check work?

  • The query counts all sales rows through COUNT(*).
  • Each subtraction calculates the missing count for one important field.
  • The row is None guard stops the tool from returning an invalid result.
  • The returned dictionary gives Inspector stable integer values for every check.
  • Save server.py.
  • Return to MCP Inspector.
  • Reconnect the existing session.
  • Call check_data_quality from the tool view.

You'll see a row count of 20. The result reports missing_region as 1.

Both missing_product and missing_revenue are 0. The regional dimension is the only detected gap.

That is a solid checkpoint: the server has quantified the exact problem behind the unnamed bucket.

Quality tool not returning four counts?

Check that every missing-value expression subtracts COUNT(column) from COUNT(*). Confirm that the three column names match region, product, and revenue.

Help me debug the data-quality result.

Label missing regions honestly

The quality tool confirms that the missing region is real. COALESCE(region, 'Unknown') preserves that sale while giving its group an honest label.

The grouped expression must match the selected expression. Converting the returned region to a string keeps every dictionary entry consistent.

  • Select the complete revenue_by_region() function in server.py.
  • Replace that selection with the corrected function below:
@mcp.tool()
def revenue_by_region() -> list[dict[str, str | int]]:
    """Rank regions by total revenue while labeling missing regions honestly."""
    with sales_connection() as connection:
        rows = connection.execute(
            """
            SELECT
                COALESCE(region, 'Unknown') AS region,
                SUM(units) AS total_units,
                SUM(revenue) AS total_revenue
            FROM sales
            GROUP BY COALESCE(region, 'Unknown')
            ORDER BY total_revenue DESC
            """
        ).fetchall()

    return [
        {
            "region": str(region),
            "total_units": int(total_units),
            "total_revenue": int(total_revenue),
        }
        for region, total_units, total_revenue in rows
    ]

What changed in the regional tool?

  • The selected region now uses COALESCE(region, 'Unknown') to replace a missing label.
  • The GROUP BY clause uses the same expression to keep the grouped result valid.
  • The revenue ordering still ranks every group from highest revenue to lowest revenue.
  • The str(region) conversion gives each returned region a consistent string type.
  • Save server.py.

✔️ Awesome, I've got everything!

Great. Your server now exposes metadata, measures missing values, and labels the incomplete region honestly.

ⓧ I'd like to double check the full code

from pathlib import Path

import duckdb
from mcp.server import MCPServer

PROJECT_ROOT = Path(__file__).parent
DATA_FILE = PROJECT_ROOT / "data" / "sales.csv"

mcp = MCPServer("DuckDB Sales Analyst")


def sales_connection():
    connection = duckdb.connect()
    connection.read_csv(str(DATA_FILE), na_values=["NULL"]).create_view("sales")
    return connection


@mcp.resource("dataset://sales/metadata", mime_type="application/json")
def dataset_metadata() -> dict[str, object]:
    """Column names and inferred DuckDB types for the sales dataset."""
    with sales_connection() as connection:
        sales = connection.table("sales")
        columns = [
            {"name": name, "type": str(data_type)}
            for name, data_type in zip(sales.columns, sales.dtypes)
        ]

    return {
        "dataset": "sales",
        "source": str(DATA_FILE.relative_to(PROJECT_ROOT)),
        "columns": columns,
    }


@mcp.tool()
def get_sales_overview() -> dict[str, int | float]:
    """Return headline row, unit, and revenue metrics for the sales dataset."""
    with sales_connection() as connection:
        row = connection.execute(
            """
            SELECT
                COUNT(*) AS row_count,
                SUM(units) AS total_units,
                SUM(revenue) AS total_revenue,
                AVG(revenue) AS average_revenue
            FROM sales
            """
        ).fetchone()

    if row is None:
        raise RuntimeError("The sales overview query returned no result.")

    return {
        "row_count": int(row[0]),
        "total_units": int(row[1]),
        "total_revenue": int(row[2]),
        "average_revenue": float(row[3]),
    }


@mcp.tool()
def revenue_by_region() -> list[dict[str, str | int]]:
    """Rank regions by total revenue while labeling missing regions honestly."""
    with sales_connection() as connection:
        rows = connection.execute(
            """
            SELECT
                COALESCE(region, 'Unknown') AS region,
                SUM(units) AS total_units,
                SUM(revenue) AS total_revenue
            FROM sales
            GROUP BY COALESCE(region, 'Unknown')
            ORDER BY total_revenue DESC
            """
        ).fetchall()

    return [
        {
            "region": str(region),
            "total_units": int(total_units),
            "total_revenue": int(total_revenue),
        }
        for region, total_units, total_revenue in rows
    ]


@mcp.tool()
def check_data_quality() -> dict[str, int]:
    """Count missing values in important sales dimensions and measures."""
    with sales_connection() as connection:
        row = connection.execute(
            """
            SELECT
                COUNT(*) AS row_count,
                COUNT(*) - COUNT(region) AS missing_region,
                COUNT(*) - COUNT(product) AS missing_product,
                COUNT(*) - COUNT(revenue) AS missing_revenue
            FROM sales
            """
        ).fetchone()

    if row is None:
        raise RuntimeError("The data-quality query returned no result.")

    return {
        "row_count": int(row[0]),
        "missing_region": int(row[1]),
        "missing_product": int(row[2]),
        "missing_revenue": int(row[3]),
    }

What does this reference contain?

This is the cumulative server.py for this checkpoint. It contains the metadata resource plus all three analytics tools.

Before you reconnect, do you expect the corrected ranking to preserve all five region buckets while changing the missing label?

  • Return to MCP Inspector from earlier.
  • Reconnect its existing session to the saved server.
  • Read dataset://sales/metadata from the resource view.
  • Call check_data_quality from the tool view.
  • Call revenue_by_region from the tool view.

The resource shows data/sales.csv as its source. It lists the five live columns date, region, product, units, and revenue.

The quality tool reports one missing region. It reports zero missing product values plus zero missing revenue values.

The regional result still contains five ranked groups. The former null bucket now appears as Unknown with total revenue of 480.

That is the trust gap closed: your server now exposes the issue, measures it, and labels it honestly.

Still seeing a null region?

Confirm that both the selected region and the grouped region use COALESCE(region, 'Unknown'). Save server.py before reconnecting Inspector.

Check that the returned dictionary uses str(region) for the region value.

Help me fix the remaining null region.

Your analytics now carries its own evidence about schema quality. Next, you'll package these trusted results into a reusable executive-summary workflow.

Package the Analysis as a Reusable Prompt

Your server now exposes live dataset metadata. Its quality checks turn the missing region into an honest Unknown group.

The analytics still arrive as separate facts. An MCP prompt packages the reporting sequence for an AI host.

In this step, get ready to:
  • Register a reusable executive-summary prompt.
  • Direct an AI host to call all three analytics tools before writing.
  • Render the prompt for sales leadership in MCP Inspector.
Add the executive-summary prompt

A prompt provides instructions that an MCP client can discover separately from tools or resources. The audience argument lets the same workflow adapt its wording for different readers.

  • Return to server.py in the editor.
  • Find the closing brace of the dictionary returned by check_data_quality().
  • Add the reusable prompt below that function by pasting this code:
@mcp.prompt()
def draft_executive_summary(audience: str = "sales leadership") -> str:
    """Create a repeatable workflow for drafting an executive sales summary."""
    return (
        f"Prepare a concise sales performance summary for {audience}. "
        "Call get_sales_overview, check_data_quality, and revenue_by_region first. "
        "Report the headline metrics, compare regions, disclose every data-quality "
        "limitation, and finish with two evidence-based recommendations."
    )

What does this code do?

  • The @mcp.prompt() decorator exposes draft_executive_summary() as an MCP prompt.
  • The audience argument defaults to sales leadership.
  • The returned message tells the host to call get_sales_overview, check_data_quality, plus revenue_by_region before writing.
  • The workflow requires headline metrics plus regional comparisons.
  • It requires every data-quality limitation to be disclosed.
  • It finishes with two evidence-based recommendations.
  • Save server.py.
  • Switch back to the PowerShell terminal from earlier.
  • Stop the running Inspector process by pressing Ctrl+C.
  • Restart the server with the new prompt by running this command:
uv run mcp dev server.py

PowerShell prints a fresh Inspector URL. Inspector can now discover the prompt you added.

Inspector not restarting?

  • Confirm the previous Inspector process has stopped before running the command again.
  • Check that PowerShell still points to the duckdb-mcp-analyst folder.
  • Confirm server.py was saved before restarting the process.

Help me troubleshoot why MCP Inspector cannot restart with my updated server.

✔️ Awesome, I've got everything!

Your saved server.py now contains the complete core analytics server.

ⓧ I'd like to double check the full code

from pathlib import Path

import duckdb
from mcp.server import MCPServer

PROJECT_ROOT = Path(__file__).parent
DATA_FILE = PROJECT_ROOT / "data" / "sales.csv"

mcp = MCPServer("DuckDB Sales Analyst")


def sales_connection():
    connection = duckdb.connect()
    connection.read_csv(str(DATA_FILE), na_values=["NULL"]).create_view("sales")
    return connection


@mcp.resource("dataset://sales/metadata", mime_type="application/json")
def dataset_metadata() -> dict[str, object]:
    """Column names and inferred DuckDB types for the sales dataset."""
    with sales_connection() as connection:
        sales = connection.table("sales")
        columns = [
            {"name": name, "type": str(data_type)}
            for name, data_type in zip(sales.columns, sales.dtypes)
        ]

    return {
        "dataset": "sales",
        "source": str(DATA_FILE.relative_to(PROJECT_ROOT)),
        "columns": columns,
    }


@mcp.tool()
def get_sales_overview() -> dict[str, int | float]:
    """Return headline row, unit, and revenue metrics for the sales dataset."""
    with sales_connection() as connection:
        row = connection.execute(
            """
            SELECT
                COUNT(*) AS row_count,
                SUM(units) AS total_units,
                SUM(revenue) AS total_revenue,
                AVG(revenue) AS average_revenue
            FROM sales
            """
        ).fetchone()

    if row is None:
        raise RuntimeError("The sales overview query returned no result.")

    return {
        "row_count": int(row[0]),
        "total_units": int(row[1]),
        "total_revenue": int(row[2]),
        "average_revenue": float(row[3]),
    }


@mcp.tool()
def revenue_by_region() -> list[dict[str, str | int]]:
    """Rank regions by total revenue while labeling missing regions honestly."""
    with sales_connection() as connection:
        rows = connection.execute(
            """
            SELECT
                COALESCE(region, 'Unknown') AS region,
                SUM(units) AS total_units,
                SUM(revenue) AS total_revenue
            FROM sales
            GROUP BY COALESCE(region, 'Unknown')
            ORDER BY total_revenue DESC
            """
        ).fetchall()

    return [
        {
            "region": str(region),
            "total_units": int(total_units),
            "total_revenue": int(total_revenue),
        }
        for region, total_units, total_revenue in rows
    ]


@mcp.tool()
def check_data_quality() -> dict[str, int]:
    """Count missing values in important sales dimensions and measures."""
    with sales_connection() as connection:
        row = connection.execute(
            """
            SELECT
                COUNT(*) AS row_count,
                COUNT(*) - COUNT(region) AS missing_region,
                COUNT(*) - COUNT(product) AS missing_product,
                COUNT(*) - COUNT(revenue) AS missing_revenue
            FROM sales
            """
        ).fetchone()

    if row is None:
        raise RuntimeError("The data-quality query returned no result.")

    return {
        "row_count": int(row[0]),
        "missing_region": int(row[1]),
        "missing_product": int(row[2]),
        "missing_revenue": int(row[3]),
    }


@mcp.prompt()
def draft_executive_summary(audience: str = "sales leadership") -> str:
    """Create a repeatable workflow for drafting an executive sales summary."""
    return (
        f"Prepare a concise sales performance summary for {audience}. "
        "Call get_sales_overview, check_data_quality, and revenue_by_region first. "
        "Report the headline metrics, compare regions, disclose every data-quality "
        "limitation, and finish with two evidence-based recommendations."
    )
Find the prompt in Inspector

Inspector lists prompts separately from callable tools or readable resources. This separation helps clients discover reporting workflows without treating them as analytics functions.

  • Return to Inspector using the fresh URL printed in PowerShell.
  • Select the Prompts tab.
  • Confirm draft_executive_summary appears in the prompt list.

That completes the server's core capabilities. Inspector now exposes analytics tools, dataset metadata, plus a reusable reporting workflow.

Render the executive summary

The prompt does not generate the report itself. It gives an AI host a consistent sequence for gathering facts before writing conclusions.

Before you render it, which three analytics tools do you expect the message to name?

  • Select draft_executive_summary from the prompt list.
  • Enter sales leadership for the audience argument.
  • Use the prompt detail panel to render the selected prompt.

You'll see a user message addressed to sales leadership. The workflow names get_sales_overview, check_data_quality, plus revenue_by_region.

The remaining instructions require headline metrics plus regional comparisons. They also require data-quality disclosure plus two evidence-based recommendations.

Prompt missing from Inspector?

  • Confirm draft_executive_summary() appears below check_data_quality() in the saved server.py file.
  • Restart the Inspector process if the Prompts tab still shows stale server capabilities.
  • Check that the audience value contains text before rendering the prompt.

Help me find or render my MCP prompt in Inspector.

You did it. Your complete server now packages trustworthy sales analytics as a workflow an AI host can follow.

Secret mission

Filter Analytics by Product

Add a product filter that returns sales totals for one exact product. You will use parameterized SQL to keep the product value separate from the query text and preserve a stable response for unknown products.

Clean Up Your Resources

Clean Up Your Resources

Choose how you want to leave your local project. Every option remains free because this project uses local files only.

Resources you used:

  • The running MCP Inspector session connected to your local server.
  • The duckdb-mcp-analyst folder containing pyproject.toml, server.py, data/sales.csv, uv.lock, and .venv.

Keep everything running

Keep the server active if you want to continue testing its analytics. The local process has no ongoing cost.

  • Continue calling the tools from MCP Inspector.
  • Leave the duckdb-mcp-analyst folder in its current location.
  • Keep the Visual Studio Code workspace available for future changes.

Pause - I'll come back to this later

Pausing stops the local process while preserving every project file. You can restart from the same folder later.

  • In the PowerShell terminal from earlier, press Ctrl+C to stop MCP Inspector.
  • Close Visual Studio Code.
  • Keep the duckdb-mcp-analyst folder on your machine.

The server process stops. DuckDB uses in-memory connections here, so no database remains running.

Delete - I don't want to use this again

Deletion permanently removes your code and dataset. Choose this option only when you have finished using the project.

  • In the VS Code PowerShell terminal from earlier, press Ctrl+C to stop MCP Inspector.
  • Remove the entire project from its parent folder by running these commands:
cd ..
Remove-Item -Recurse -Force duckdb-mcp-analyst

What do these commands remove?

  • The first line moves PowerShell out of the project folder.
  • The second line permanently deletes the project folder plus every file inside it.
  • Confirm the deletion by listing the parent folder's remaining contents with this command:
Get-ChildItem

What should you see?

The output lists the parent folder's remaining contents. The duckdb-mcp-analyst folder should be absent.

Folder still listed?

  • Stop any terminal process that still uses the project folder.
  • Run the deletion commands again from the parent folder.
  • Help me remove the local duckdb-mcp-analyst folder from PowerShell.

Cleanup complete. The duckdb-mcp-analyst folder and everything inside it are gone.

Nice Work!

Nice Work!

You did it! Your local sales analyst now turns a sales file into trustworthy metrics without a separate database server.

What you learned:

  • Built a local MCP server backed by DuckDB. It reads the CSV dataset in data/sales.csv through an in-memory view.
  • Used SQL aggregation to return headline metrics. Ranked regional revenue as structured tool results. Quantified the missing region through a data-quality check. Replaced the ambiguous null bucket with Unknown.
  • Exposed live column names through a dataset metadata resource. Included each column's inferred DuckDB type. Packaged the analysis sequence as a reusable executive-summary prompt for sales leadership.
  • Secret Mission: Added a safely parameterized product filter through sales_for_product. Confirmed Monitor returns 3 rows. The same result reports 12 units. Its total revenue is 3000. Confirmed unknown products keep the same response shape with zero totals.

Ready to quiz yourself?