demo.py•6.64 kB
#!/usr/bin/env python3
"""
Universal SQL MCP Server Demo
This script demonstrates the capabilities of the Universal SQL MCP Server
by creating a simple SQLite database and showing how the MCP tools work.
"""
import os
import sys
import json
import sqlite3
from pathlib import Path
# Add the current directory to the path
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
def create_demo_database():
"""Create a demo SQLite database with sample data"""
db_path = "./demo_database.db"
# Remove existing database
if os.path.exists(db_path):
os.remove(db_path)
# Create new database with sample data
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
# Create users table
cursor.execute("""
CREATE TABLE users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
email TEXT UNIQUE NOT NULL,
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
is_active BOOLEAN DEFAULT 1
)
""")
# Create orders table
cursor.execute("""
CREATE TABLE orders (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL,
product_name TEXT NOT NULL,
amount DECIMAL(10,2) NOT NULL,
order_date DATETIME DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (user_id) REFERENCES users (id)
)
""")
# Insert sample users
users_data = [
("Alice Johnson", "alice@example.com"),
("Bob Smith", "bob@example.com"),
("Charlie Brown", "charlie@example.com"),
("Diana Prince", "diana@example.com"),
("Eve Wilson", "eve@example.com")
]
cursor.executemany("INSERT INTO users (name, email) VALUES (?, ?)", users_data)
# Insert sample orders
orders_data = [
(1, "Laptop", 999.99),
(1, "Mouse", 29.99),
(2, "Keyboard", 79.99),
(3, "Monitor", 299.99),
(3, "Webcam", 89.99),
(4, "Headphones", 149.99),
(5, "Tablet", 399.99)
]
cursor.executemany("INSERT INTO orders (user_id, product_name, amount) VALUES (?, ?, ?)", orders_data)
# Create indexes
cursor.execute("CREATE INDEX idx_users_email ON users (email)")
cursor.execute("CREATE INDEX idx_orders_user_id ON orders (user_id)")
cursor.execute("CREATE INDEX idx_orders_date ON orders (order_date)")
conn.commit()
conn.close()
print(f"✅ Demo database created: {db_path}")
return db_path
def setup_environment(db_path):
"""Setup environment variables for the demo"""
os.environ['DB_TYPE'] = 'sqlite'
os.environ['DB_NAME'] = db_path
os.environ['ENABLE_WRITE_OPERATIONS'] = 'true'
os.environ['LOG_LEVEL'] = 'INFO'
print("✅ Environment configured for SQLite demo")
def demo_mcp_tools():
"""Demonstrate MCP tools functionality"""
from database import get_db_manager
from tools import get_database_schema, execute_sql_query, execute_write_operation, test_database_connection
print("\n" + "="*60)
print("🚀 Universal SQL MCP Server Demo")
print("="*60)
# Test 1: Database Connection
print("\n1️⃣ Testing Database Connection...")
result = test_database_connection()
print(f" Result: {result['message']}")
# Test 2: Get Database Schema
print("\n2️⃣ Getting Database Schema...")
schema = get_database_schema()
print(f" Found {len(schema)} tables:")
for table in schema:
print(f" 📋 {table['table_name']} ({len(table['columns'])} columns, {len(table['indexes'])} indexes)")
# Test 3: Execute SELECT Query
print("\n3️⃣ Executing SELECT Query...")
query_result = execute_sql_query("SELECT u.name, u.email, COUNT(o.id) as order_count FROM users u LEFT JOIN orders o ON u.id = o.user_id GROUP BY u.id, u.name, u.email ORDER BY order_count DESC")
if query_result['success']:
print(f" Query returned {query_result['row_count']} rows:")
for row in query_result['data'][:3]: # Show first 3 rows
print(f" 👤 {row['name']} ({row['email']}) - {row['order_count']} orders")
# Test 4: Execute Write Operation
print("\n4️⃣ Executing Write Operation...")
write_result = execute_write_operation("INSERT INTO users (name, email) VALUES ('Demo User', 'demo@example.com')")
if write_result['success']:
print(f" ✅ Inserted new user, affected rows: {write_result['affected_rows']}")
# Verify the insert
verify_result = execute_sql_query("SELECT * FROM users WHERE email = 'demo@example.com'")
if verify_result['success'] and verify_result['data']:
user = verify_result['data'][0]
print(f" ✅ Verified: User '{user['name']}' was created with ID {user['id']}")
# Test 5: Complex Query Example
print("\n5️⃣ Complex Query Example...")
complex_query = """
SELECT
u.name,
COUNT(o.id) as total_orders,
ROUND(SUM(o.amount), 2) as total_spent,
ROUND(AVG(o.amount), 2) as avg_order_value
FROM users u
LEFT JOIN orders o ON u.id = o.user_id
GROUP BY u.id, u.name
HAVING COUNT(o.id) > 0
ORDER BY total_spent DESC
"""
complex_result = execute_sql_query(complex_query)
if complex_result['success']:
print(f" Customer spending analysis ({complex_result['row_count']} customers):")
for row in complex_result['data']:
print(f" 💰 {row['name']}: {row['total_orders']} orders, ${row['total_spent']} total, ${row['avg_order_value']} avg")
print("\n" + "="*60)
print("✅ Demo completed successfully!")
print("="*60)
def main():
"""Main demo function"""
print("🎯 Universal SQL MCP Server Demo")
print("This demo will:")
print("1. Create a SQLite database with sample data")
print("2. Configure the environment")
print("3. Demonstrate all MCP tools")
print()
try:
# Create demo database
db_path = create_demo_database()
# Setup environment
setup_environment(db_path)
# Run demo
demo_mcp_tools()
print(f"\n📁 Demo database saved as: {db_path}")
print("🔧 You can now start the MCP server with: python main.py")
except Exception as e:
print(f"❌ Demo failed: {e}")
import traceback
traceback.print_exc()
return 1
return 0
if __name__ == "__main__":
sys.exit(main())