Model Context Protocol (MCP): Connecting AI Assistants to Your Data and Tools

import asyncio, json, duckdb, pandas as pd
from mcp import Client
from mcp.server.mcpserver import MCPServer        # in MCP SDK 1.x this class was FastMCP (from mcp.server.fastmcp import FastMCP)
from mcp.server.mcpserver.exceptions import ToolError
import logging; logging.getLogger("mcp").setLevel(logging.CRITICAL)      # keep the demo output tidy

sales = pd.read_csv("superstore_sales.csv", parse_dates=["order_date"])
con = duckdb.connect(); con.register("sales", sales)
server = MCPServer("NovaMart sales analytics")

@server.tool()
def sales_summary(year: int, dimension: str = "region") -> list[dict]:
    """Total sales, profit and margin for one year, grouped by region, category, segment or sub_category."""
    if dimension not in {"region", "category", "segment", "sub_category"}:
        raise ToolError("dimension must be one of region, category, segment, sub_category")   # a clear message the model can act on
    q = f"""SELECT {dimension}, ROUND(SUM(sales)/1e6, 2) AS sales_m, ROUND(SUM(profit)/1e6, 2) AS profit_m,
            ROUND(100*SUM(profit)/SUM(sales), 1) AS margin_pct FROM sales WHERE YEAR(order_date) = ? GROUP BY 1 ORDER BY sales_m DESC"""
    return con.execute(q, [year]).df().to_dict("records")

@server.tool()
def top_products(category: str, n: int = 5) -> list[dict]:
    """The n most profitable products in a category (Furniture, Office Supplies, Technology)."""
    q = "SELECT product_name, ROUND(SUM(profit)) AS profit FROM sales WHERE category = ? GROUP BY 1 ORDER BY 2 DESC LIMIT ?"
    return con.execute(q, [category, min(n, 20)]).df().to_dict("records")

@server.resource("schema://sales")
def sales_schema() -> str:
    """Column names and types of the sales table."""
    return sales.dtypes.astype(str).to_json()

@server.prompt()
def monthly_review(month: str) -> str:
    """Prompt template for a monthly business review."""
    return f"Using the sales tools, write a monthly business review for {month}: headline numbers, three drivers, one risk and one action. Cite every number's source."

async def demo():
    async with Client(server) as client:                      # in-process connection, ideal for testing
        tools = await client.list_tools()
        for t in tools.tools: print("tool:", t.name, "| inputs:", list(t.input_schema["properties"]))
        r = await client.call_tool("sales_summary", {"year": 2025, "dimension": "category"})
        print("call result:", r.content[0].text[:160])
        bad = await client.call_tool("sales_summary", {"year": 2025, "dimension": "customer_name"})
        print("error is returned, not raised:", bad.is_error, "|", bad.content[0].text[:90])
        res = await client.read_resource("schema://sales"); print("resource:", res.contents[0].text[:80], "...")
        p = await client.get_prompt("monthly_review", {"month": "December 2025"}); print("prompt:", p.messages[0].content.text[:90], "...")
asyncio.run(demo())

if __name__ == "__main__":
    server.run()            # stdio transport (default): the host launches this script as a subprocess
    # server.run(transport="streamable-http")   # remote server at http://host:8000/mcp

{
  "mcpServers": {
    "novamart-sales": {
      "command": "python",
      "args": ["C:/projects/novamart/sales_server.py"]
    }
  }
}

response = client.beta.messages.create(
    model="claude-opus-5-5", max_tokens=16000, betas=["mcp-client-2025-11-20"],
    mcp_servers=[{"type": "url", "url": "https://mcp.novamart.example/mcp", "name": "novamart"}],
    tools=[{"type": "mcp_toolset", "mcp_server_name": "novamart"}],
    messages=[{"role": "user", "content": "Compare 2025 sales by category and name the weakest one."}],
)