MCP server: let an AI answer from your own data

A small MCP server on top of a shop database. An AI model calls its tools to answer plain questions like "which products sold most last month".

The shop, products, customers and orders are all made up. The AI answer in the video is a real run.

What it does

• Five tools: search orders, top products, sales totals, product list, today's date.
• Read-only, so the AI can't change anything.
• Standard MCP, so other MCP apps can connect to it too.

I checked the answer against a plain database query and the numbers match.

The answer, full size

The server code

"""Read-only MCP server over shop.db (a made-up shop). Runs over stdio.

python server.py          (an MCP client starts this itself; see client.py)

The database is opened read-only, so no tool can change data.
"""
import sqlite3
from datetime import date
from pathlib import Path

from mcp.server.fastmcp import FastMCP

DB = Path(__file__).with_name("shop.db")
mcp = FastMCP("shop")


def q(sql, args=()):
    db = sqlite3.connect(f"file:{DB.as_posix()}?mode=ro", uri=True)
    db.row_factory = sqlite3.Row
    try:
        return [dict(r) for r in db.execute(sql, args)]
    finally:
        db.close()


def check_date(s):
    return date.fromisoformat(s).isoformat()


@mcp.tool()
def today() -> str:
    """Today's date (YYYY-MM-DD). Use it to work out ranges like 'last month'."""
    return date.today().isoformat()


@mcp.tool()
def list_products(category: str = "") -> list[dict]:
    """All products with id, name, category and price. Optionally filter by category."""
    return q("SELECT * FROM products WHERE (? = '' OR category = ?) ORDER BY id", (category, category))


@mcp.tool()
def top_products(start: str, end: str, limit: int = 5, by: str = "units") -> list[dict]:
    """Best-selling products between start and end (inclusive, YYYY-MM-DD).
    by = 'units' or 'revenue'. Refunded orders are left out."""
    order = "revenue" if by == "revenue" else "units"
    return q(f"""SELECT p.name, p.category, SUM(i.qty) AS units, ROUND(SUM(i.qty * i.unit_price), 2) AS revenue,
                        COUNT(DISTINCT o.id) AS orders
                 FROM order_items i JOIN orders o ON o.id = i.order_id JOIN products p ON p.id = i.product_id
                 WHERE o.ordered_on BETWEEN ? AND ? AND o.status != 'refunded'
                 GROUP BY p.id ORDER BY {order} DESC LIMIT ?""",
             (check_date(start), check_date(end), max(1, min(int(limit), 50))))


@mcp.tool()
def search_orders(start: str = "", end: str = "", customer: str = "", product: str = "", status: str = "",
                  limit: int = 20) -> list[dict]:
    """Find orders. All filters are optional: date range (YYYY-MM-DD), part of a customer name,
    part of a product name, status (paid, shipped, refunded). Newest first."""
    return q("""SELECT o.id, o.ordered_on, o.status, c.name AS customer, c.city,
                       GROUP_CONCAT(i.qty || ' x ' || p.name, '; ') AS items,
                       ROUND(SUM(i.qty * i.unit_price), 2) AS total
                FROM orders o JOIN customers c ON c.id = o.customer_id
                JOIN order_items i ON i.order_id = o.id JOIN products p ON p.id = i.product_id
                WHERE (? = '' OR o.ordered_on >= ?) AND (? = '' OR o.ordered_on <= ?)
                  AND (? = '' OR c.name LIKE '%' || ? || '%') AND (? = '' OR o.status = ?)
                GROUP BY o.id
                HAVING (? = '' OR items LIKE '%' || ? || '%')
                ORDER BY o.ordered_on DESC, o.id DESC LIMIT ?""",
             (start, start, end, end, customer, customer, status, status, product, product,
              max(1, min(int(limit), 100))))


@mcp.tool()
def sales_summary(start: str, end: str) -> dict:
    """Totals between start and end (inclusive): orders, units, revenue, refunds."""
    rows = q("""SELECT o.status, COUNT(DISTINCT o.id) AS orders, SUM(i.qty) AS units,
                       ROUND(SUM(i.qty * i.unit_price), 2) AS revenue
                FROM orders o JOIN order_items i ON i.order_id = o.id
                WHERE o.ordered_on BETWEEN ? AND ? GROUP BY o.status""", (check_date(start), check_date(end)))
    by = {r["status"]: r for r in rows}
    kept = [r for s, r in by.items() if s != "refunded"]
    return {"start": start, "end": end,
            "orders": sum(r["orders"] for r in kept), "units": sum(r["units"] for r in kept),
            "revenue": round(sum(r["revenue"] for r in kept), 2),
            "refunded_orders": by.get("refunded", {}).get("orders", 0)}


if __name__ == "__main__":
    mcp.run()