18 Practice Lab: Advanced MCP Concepts
18.1 What You’ll Build
A Business Intelligence MCP Server that provides an AI assistant with access to: - Live database queries - File system operations - Report generation - Real-time streaming responses
18.2 Step 1: Production MCP Server
# file: bi_mcp_server.py
from mcp.server import Server
from mcp.server.models import InitializationOptions
from mcp.types import Tool, TextContent, Resource
import mcp.server.stdio
import asyncio
import json
import sqlite3
from pathlib import Path
from datetime import datetime
server = Server("business-intelligence")
# --- Mock database setup ---
def setup_demo_db():
conn = sqlite3.connect(":memory:")
conn.execute("""CREATE TABLE sales (
id INTEGER PRIMARY KEY,
date TEXT, product TEXT,
revenue REAL, units INTEGER, region TEXT
)""")
# Insert sample data
import random
products = ["AI Platform", "Data Analytics", "Cloud Storage", "API Services"]
regions = ["Lagos", "Nairobi", "Cairo", "Accra", "Johannesburg"]
for i in range(100):
conn.execute("INSERT INTO sales VALUES (?, ?, ?, ?, ?, ?)", (
i, f"2026-{(i%12)+1:02d}-{(i%28)+1:02d}",
random.choice(products),
round(random.uniform(10000, 500000), 2),
random.randint(1, 50),
random.choice(regions)
))
conn.commit()
return conn
DB = setup_demo_db()
# --- Tools ---
@server.list_tools()
async def list_tools():
return [
Tool(
name="query_database",
description="Run a SQL query against the business database. Tables: sales(id, date, product, revenue, units, region)",
inputSchema={
"type": "object",
"properties": {
"sql": {"type": "string", "description": "SQL SELECT query"},
"limit": {"type": "integer", "default": 10, "maximum": 100}
},
"required": ["sql"]
}
),
Tool(
name="get_kpi_summary",
description="Get key business KPIs for a date range",
inputSchema={
"type": "object",
"properties": {
"start_date": {"type": "string", "description": "YYYY-MM format"},
"end_date": {"type": "string", "description": "YYYY-MM format"}
},
"required": []
}
),
Tool(
name="export_to_file",
description="Export query results to CSV",
inputSchema={
"type": "object",
"properties": {
"sql": {"type": "string"},
"filename": {"type": "string"}
},
"required": ["sql", "filename"]
}
),
Tool(
name="create_analysis",
description="Generate an AI analysis of a business metric",
inputSchema={
"type": "object",
"properties": {
"metric": {"type": "string", "description": "e.g. 'revenue trends', 'regional performance'"},
"depth": {"type": "string", "enum": ["brief", "detailed"], "default": "brief"}
},
"required": ["metric"]
}
)
]
@server.call_tool()
async def call_tool(name: str, arguments: dict):
"""Execute tool with full error handling."""
try:
if name == "query_database":
sql = arguments["sql"]
limit = arguments.get("limit", 10)
# Security: only allow SELECT
if not sql.strip().upper().startswith("SELECT"):
return [TextContent(type="text", text="Error: Only SELECT queries are allowed")]
# Add limit
if "LIMIT" not in sql.upper():
sql = f"{sql} LIMIT {limit}"
cursor = DB.execute(sql)
rows = cursor.fetchall()
columns = [d[0] for d in cursor.description]
# Format as table
result = f"Columns: {', '.join(columns)}\n"
result += f"Rows returned: {len(rows)}\n\n"
for row in rows:
result += " | ".join(str(v) for v in row) + "\n"
return [TextContent(type="text", text=result)]
elif name == "get_kpi_summary":
cursor = DB.execute("""
SELECT
COUNT(*) as total_transactions,
ROUND(SUM(revenue), 2) as total_revenue,
ROUND(AVG(revenue), 2) as avg_deal_size,
SUM(units) as total_units,
COUNT(DISTINCT product) as products_sold,
COUNT(DISTINCT region) as regions_active
FROM sales
""")
row = cursor.fetchone()
cols = [d[0] for d in cursor.description]
kpis = dict(zip(cols, row))
# Top product
top = DB.execute("""
SELECT product, ROUND(SUM(revenue), 2) as rev
FROM sales GROUP BY product
ORDER BY rev DESC LIMIT 1
""").fetchone()
summary = f"""📊 Business KPI Summary (All Time)
=====================================
Total Transactions: {kpis['total_transactions']:,}
Total Revenue: ${kpis['total_revenue']:,.2f}
Average Deal Size: ${kpis['avg_deal_size']:,.2f}
Total Units Sold: {kpis['total_units']:,}
Products Active: {kpis['products_sold']}
Regions Active: {kpis['regions_active']}
Top Product: {top[0]} (${top[1]:,.2f})
"""
return [TextContent(type="text", text=summary)]
elif name == "export_to_file":
sql = arguments["sql"]
filename = arguments.get("filename", f"export_{datetime.now().strftime('%Y%m%d')}.csv")
cursor = DB.execute(sql)
rows = cursor.fetchall()
columns = [d[0] for d in cursor.description]
content = ",".join(columns) + "\n"
for row in rows:
content += ",".join(str(v) for v in row) + "\n"
Path("./exports").mkdir(exist_ok=True)
filepath = f"./exports/{filename}"
Path(filepath).write_text(content)
return [TextContent(type="text", text=f"✅ Exported {len(rows)} rows to {filepath}")]
elif name == "create_analysis":
metric = arguments["metric"]
depth = arguments.get("depth", "brief")
# Get relevant data
data = DB.execute("""
SELECT product, region, SUM(revenue) as rev, SUM(units) as units
FROM sales GROUP BY product, region ORDER BY rev DESC LIMIT 20
""").fetchall()
analysis_prompt = f"""
Based on this sales data: {data}
Provide a {depth} analysis of: {metric}
Be specific with numbers. Format clearly.
"""
from openai import AsyncOpenAI
client = AsyncOpenAI()
response = await client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": analysis_prompt}]
)
return [TextContent(type="text", text=response.choices[0].message.content)]
except Exception as e:
return [TextContent(type="text", text=f"Error executing {name}: {str(e)}")]
async def main():
async with mcp.server.stdio.stdio_server() as (read, write):
await server.run(read, write,
InitializationOptions(server_name="business-intelligence", server_version="1.0.0"))
if __name__ == "__main__":
asyncio.run(main())18.3 Step 2: Test Your MCP Server
# Install MCP SDK
pip install mcp openai
# Test server directly
python bi_mcp_server.py
# Or use the MCP Inspector (development tool)
npx @modelcontextprotocol/inspector python bi_mcp_server.py18.4 Step 3: Connect to Claude Desktop
Add to your Claude Desktop config:
{
"mcpServers": {
"business-intelligence": {
"command": "python",
"args": ["/full/path/to/bi_mcp_server.py"]
}
}
}Now you can ask Claude: “Show me the top 5 products by revenue” and it will use your MCP tools!
18.5 Lab Challenges 🏆
- Easy: Add a
get_top_regionstool that returns ranked regions by revenue - Medium: Add authentication — require an API key header before allowing tool calls
- Hard: Add a
forecasttool that uses linear regression to project next month’s revenue