#!/usr/bin/env python3.11
import asyncio
import json
import os
import sqlite3
from datetime import datetime, timedelta
from typing import List, Dict, Any, Optional
import httpx
from mcp.server.fastmcp import FastMCP
# Create the stevia store MCP server
mcp = FastMCP("Stevia Store Manager")
# Database setup
DATABASE_PATH = os.getenv("DATABASE_PATH", "stevia_store.db")
def init_database():
"""Initialize the SQLite database with required tables"""
conn = sqlite3.connect(DATABASE_PATH)
cursor = conn.cursor()
# Products table
cursor.execute("""
CREATE TABLE IF NOT EXISTS products (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
category TEXT NOT NULL,
price REAL NOT NULL,
cost REAL NOT NULL,
description TEXT,
sku TEXT UNIQUE,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
# Inventory table
cursor.execute("""
CREATE TABLE IF NOT EXISTS inventory (
id INTEGER PRIMARY KEY AUTOINCREMENT,
product_id INTEGER,
quantity INTEGER NOT NULL DEFAULT 0,
low_stock_threshold INTEGER DEFAULT 10,
location TEXT DEFAULT 'Main Warehouse',
last_updated TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (product_id) REFERENCES products (id)
)
""")
# Customers table
cursor.execute("""
CREATE TABLE IF NOT EXISTS customers (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
email TEXT UNIQUE,
phone TEXT,
address TEXT,
city TEXT,
registration_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
total_orders INTEGER DEFAULT 0,
total_spent REAL DEFAULT 0.0
)
""")
# Orders table
cursor.execute("""
CREATE TABLE IF NOT EXISTS orders (
id INTEGER PRIMARY KEY AUTOINCREMENT,
customer_id INTEGER,
order_date TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
status TEXT DEFAULT 'pending',
total_amount REAL NOT NULL,
shipping_address TEXT,
notes TEXT,
FOREIGN KEY (customer_id) REFERENCES customers (id)
)
""")
# Order items table
cursor.execute("""
CREATE TABLE IF NOT EXISTS order_items (
id INTEGER PRIMARY KEY AUTOINCREMENT,
order_id INTEGER,
product_id INTEGER,
quantity INTEGER NOT NULL,
unit_price REAL NOT NULL,
total_price REAL NOT NULL,
FOREIGN KEY (order_id) REFERENCES orders (id),
FOREIGN KEY (product_id) REFERENCES products (id)
)
""")
# Insert sample data if tables are empty
cursor.execute("SELECT COUNT(*) FROM products")
if cursor.fetchone()[0] == 0:
sample_products = [
("סטיביה טבעית אורגנית", "חמרי גלם", 45.90, 25.00, "סטיביה טבעית 100% אורגנית ללא תוספות", "STEV-ORG-100"),
("סטיביה בטבליות", "מוצרים מוכנים", 29.90, 15.00, "טבליות סטיביה נוחות לשימוש", "STEV-TAB-60"),
("סטיביה נוזלית", "מוצרים מוכנים", 39.90, 20.00, "תמצית סטיביה נוזלית טעם ונילה", "STEV-LIQ-50"),
("אבקת סטיביה לאפיה", "בישול ואפיה", 55.90, 30.00, "אבקת סטיביה מיוחדת לאפיה ובישול", "STEV-BAK-200"),
("חבילת טעימה", "מארזים", 89.90, 45.00, "מארז טעימה עם 4 סוגי סטיביה שונים", "STEV-PACK-MIX")
]
cursor.executemany("""
INSERT INTO products (name, category, price, cost, description, sku)
VALUES (?, ?, ?, ?, ?, ?)
""", sample_products)
# Add inventory for sample products
for i in range(1, 6):
cursor.execute("""
INSERT INTO inventory (product_id, quantity, low_stock_threshold)
VALUES (?, ?, ?)
""", (i, 50 + i * 10, 10))
conn.commit()
conn.close()
# Initialize database on startup
init_database()
@mcp.tool()
def add_product(name: str, category: str, price: float, cost: float,
description: str = "", sku: str = "") -> str:
"""
הוספת מוצר חדש למערכת
Args:
name: שם המוצר
category: קטגוריית המוצר
price: מחיר מכירה
cost: עלות המוצר
description: תיאור המוצר (אופציונלי)
sku: מק"ט (אופציונלי)
Returns:
הודעת אישור או שגיאה
"""
try:
conn = sqlite3.connect(DATABASE_PATH)
cursor = conn.cursor()
cursor.execute("""
INSERT INTO products (name, category, price, cost, description, sku)
VALUES (?, ?, ?, ?, ?, ?)
""", (name, category, price, cost, description, sku))
product_id = cursor.lastrowid
# Create inventory record
cursor.execute("""
INSERT INTO inventory (product_id, quantity, low_stock_threshold)
VALUES (?, 0, 10)
""", (product_id,))
conn.commit()
conn.close()
return f"✅ המוצר '{name}' נוסף בהצלחה עם ID: {product_id}"
except sqlite3.IntegrityError:
return f"❌ שגיאה: מק\"ט {sku} כבר קיים במערכת"
except Exception as e:
return f"❌ שגיאה בהוספת מוצר: {str(e)}"
@mcp.tool()
def get_products(category: str = "") -> str:
"""
קבלת רשימת המוצרים
Args:
category: סינון לפי קטגוריה (אופציונלי)
Returns:
רשימת המוצרים בפורמט JSON
"""
try:
conn = sqlite3.connect(DATABASE_PATH)
cursor = conn.cursor()
if category:
cursor.execute("""
SELECT p.id, p.name, p.category, p.price, p.cost, p.description, p.sku,
i.quantity, i.low_stock_threshold
FROM products p
LEFT JOIN inventory i ON p.id = i.product_id
WHERE p.category = ?
ORDER BY p.name
""", (category,))
else:
cursor.execute("""
SELECT p.id, p.name, p.category, p.price, p.cost, p.description, p.sku,
i.quantity, i.low_stock_threshold
FROM products p
LEFT JOIN inventory i ON p.id = i.product_id
ORDER BY p.category, p.name
""")
products = cursor.fetchall()
conn.close()
if not products:
return "📦 לא נמצאו מוצרים במערכת"
result = "🛍️ **רשימת המוצרים:**\n\n"
current_category = ""
for product in products:
if product[2] != current_category:
current_category = product[2]
result += f"\n**{current_category}:**\n"
stock_status = "⚠️ מלאי נמוך" if product[7] <= product[8] else "✅ במלאי"
profit_margin = ((product[3] - product[4]) / product[3] * 100) if product[3] > 0 else 0
result += f"• **{product[1]}** (#{product[0]})\n"
result += f" מק\"ט: {product[6]} | מחיר: ₪{product[3]:.2f} | רווח: {profit_margin:.1f}%\n"
result += f" מלאי: {product[7]} יחידות {stock_status}\n"
if product[5]:
result += f" תיאור: {product[5]}\n"
result += "\n"
return result
except Exception as e:
return f"❌ שגיאה בקבלת רשימת מוצרים: {str(e)}"
@mcp.tool()
def update_inventory(product_id: int, quantity: int, operation: str = "set") -> str:
"""
עדכון מלאי מוצר
Args:
product_id: מזהה המוצר
quantity: כמות
operation: סוג הפעולה - "set" (הגדרה), "add" (הוספה), "subtract" (הפחתה)
Returns:
הודעת אישור או שגיאה
"""
try:
conn = sqlite3.connect(DATABASE_PATH)
cursor = conn.cursor()
# Get current inventory
cursor.execute("SELECT quantity FROM inventory WHERE product_id = ?", (product_id,))
result = cursor.fetchone()
if not result:
return f"❌ לא נמצא מוצר עם ID: {product_id}"
current_quantity = result[0]
if operation == "set":
new_quantity = quantity
elif operation == "add":
new_quantity = current_quantity + quantity
elif operation == "subtract":
new_quantity = max(0, current_quantity - quantity)
else:
return "❌ פעולה לא חוקית. השתמש ב: set, add, או subtract"
cursor.execute("""
UPDATE inventory
SET quantity = ?, last_updated = CURRENT_TIMESTAMP
WHERE product_id = ?
""", (new_quantity, product_id))
# Get product name for confirmation
cursor.execute("SELECT name FROM products WHERE id = ?", (product_id,))
product_name = cursor.fetchone()[0]
conn.commit()
conn.close()
return f"✅ מלאי עודכן: {product_name}\n📊 מלאי קודם: {current_quantity} → מלאי חדש: {new_quantity}"
except Exception as e:
return f"❌ שגיאה בעדכון מלאי: {str(e)}"
@mcp.tool()
def get_low_stock_alerts() -> str:
"""
קבלת התראות על מוצרים עם מלאי נמוך
Returns:
רשימת מוצרים עם מלאי נמוך
"""
try:
conn = sqlite3.connect(DATABASE_PATH)
cursor = conn.cursor()
cursor.execute("""
SELECT p.name, p.sku, i.quantity, i.low_stock_threshold, p.category
FROM products p
JOIN inventory i ON p.id = i.product_id
WHERE i.quantity <= i.low_stock_threshold
ORDER BY (i.quantity::FLOAT / i.low_stock_threshold) ASC
""")
low_stock = cursor.fetchall()
conn.close()
if not low_stock:
return "✅ כל המוצרים במלאי תקין!"
result = "⚠️ **התראות מלאי נמוך:**\n\n"
for item in low_stock:
urgency = "🔴 קריטי" if item[2] == 0 else "🟡 נמוך"
result += f"{urgency} **{item[0]}**\n"
result += f"📦 מלאי נוכחי: {item[2]} | סף מינימום: {item[3]}\n"
result += f"🏷️ מק\"ט: {item[1]} | קטגוריה: {item[4]}\n\n"
return result
except Exception as e:
return f"❌ שגיאה בקבלת התראות מלאי: {str(e)}"
@mcp.tool()
def add_customer(name: str, email: str, phone: str = "", address: str = "", city: str = "") -> str:
"""
הוספת לקוח חדש
Args:
name: שם הלקוח
email: כתובת אימייל
phone: טלפון (אופציונלי)
address: כתובת (אופציונלי)
city: עיר (אופציונלי)
Returns:
הודעת אישור או שגיאה
"""
try:
conn = sqlite3.connect(DATABASE_PATH)
cursor = conn.cursor()
cursor.execute("""
INSERT INTO customers (name, email, phone, address, city)
VALUES (?, ?, ?, ?, ?)
""", (name, email, phone, address, city))
customer_id = cursor.lastrowid
conn.commit()
conn.close()
return f"✅ הלקוח '{name}' נוסף בהצלחה עם ID: {customer_id}"
except sqlite3.IntegrityError:
return f"❌ שגיאה: כתובת אימייל {email} כבר קיימת במערכת"
except Exception as e:
return f"❌ שגיאה בהוספת לקוח: {str(e)}"
@mcp.tool()
def get_sales_summary(days: int = 30) -> str:
"""
קבלת סיכום מכירות
Args:
days: מספר ימים אחורה לסיכום (ברירת מחדל: 30)
Returns:
סיכום מכירות מפורט
"""
try:
conn = sqlite3.connect(DATABASE_PATH)
cursor = conn.cursor()
date_limit = datetime.now() - timedelta(days=days)
# Total sales and orders
cursor.execute("""
SELECT COUNT(*) as total_orders, SUM(total_amount) as total_sales
FROM orders
WHERE order_date >= ?
""", (date_limit.isoformat(),))
total_stats = cursor.fetchone()
# Top selling products
cursor.execute("""
SELECT p.name, SUM(oi.quantity) as units_sold, SUM(oi.total_price) as revenue
FROM order_items oi
JOIN products p ON oi.product_id = p.id
JOIN orders o ON oi.order_id = o.id
WHERE o.order_date >= ?
GROUP BY p.id, p.name
ORDER BY units_sold DESC
LIMIT 5
""", (date_limit.isoformat(),))
top_products = cursor.fetchall()
# Category performance
cursor.execute("""
SELECT p.category, SUM(oi.quantity) as units_sold, SUM(oi.total_price) as revenue
FROM order_items oi
JOIN products p ON oi.product_id = p.id
JOIN orders o ON oi.order_id = o.id
WHERE o.order_date >= ?
GROUP BY p.category
ORDER BY revenue DESC
""", (date_limit.isoformat(),))
category_stats = cursor.fetchall()
conn.close()
result = f"📊 **סיכום מכירות - {days} ימים אחרונים**\n\n"
if total_stats[0]:
result += f"💰 **סה\"כ מכירות:** ₪{total_stats[1]:.2f}\n"
result += f"📦 **מספר הזמנות:** {total_stats[0]}\n"
result += f"💳 **ממוצע הזמנה:** ₪{(total_stats[1]/total_stats[0]):.2f}\n\n"
if top_products:
result += "🏆 **מוצרים מובילים:**\n"
for i, product in enumerate(top_products, 1):
result += f"{i}. {product[0]} - {product[1]} יחידות (₪{product[2]:.2f})\n"
result += "\n"
if category_stats:
result += "📈 **ביצועים לפי קטגוריה:**\n"
for category in category_stats:
result += f"• {category[0]}: {category[1]} יחידות (₪{category[2]:.2f})\n"
else:
result += "📭 לא נמצאו מכירות בתקופה זו"
return result
except Exception as e:
return f"❌ שגיאה בקבלת סיכום מכירות: {str(e)}"
@mcp.tool()
def create_order(customer_id: int, products: str, shipping_address: str = "", notes: str = "") -> str:
"""
יצירת הזמנה חדשה
Args:
customer_id: מזהה הלקוח
products: רשימת מוצרים בפורמט JSON - [{"product_id": 1, "quantity": 2}, ...]
shipping_address: כתובת משלוח (אופציונלי)
notes: הערות (אופציונלי)
Returns:
פרטי ההזמנה או הודעת שגיאה
"""
try:
# Parse products JSON
try:
product_list = json.loads(products)
except json.JSONDecodeError:
return "❌ שגיאה: פורמט מוצרים לא תקין. השתמש בפורמט JSON"
conn = sqlite3.connect(DATABASE_PATH)
cursor = conn.cursor()
# Verify customer exists
cursor.execute("SELECT name FROM customers WHERE id = ?", (customer_id,))
customer = cursor.fetchone()
if not customer:
return f"❌ לא נמצא לקוח עם ID: {customer_id}"
total_amount = 0
order_details = []
# Calculate total and prepare order details
for item in product_list:
product_id = item.get('product_id')
quantity = item.get('quantity', 1)
cursor.execute("SELECT name, price FROM products WHERE id = ?", (product_id,))
product = cursor.fetchone()
if not product:
return f"❌ לא נמצא מוצר עם ID: {product_id}"
item_total = product[1] * quantity
total_amount += item_total
order_details.append({
'product_id': product_id,
'name': product[0],
'quantity': quantity,
'unit_price': product[1],
'total_price': item_total
})
# Create order
cursor.execute("""
INSERT INTO orders (customer_id, total_amount, shipping_address, notes)
VALUES (?, ?, ?, ?)
""", (customer_id, total_amount, shipping_address, notes))
order_id = cursor.lastrowid
# Add order items
for item in order_details:
cursor.execute("""
INSERT INTO order_items (order_id, product_id, quantity, unit_price, total_price)
VALUES (?, ?, ?, ?, ?)
""", (order_id, item['product_id'], item['quantity'], item['unit_price'], item['total_price']))
# Update inventory
cursor.execute("""
UPDATE inventory
SET quantity = quantity - ?
WHERE product_id = ? AND quantity >= ?
""", (item['quantity'], item['product_id'], item['quantity']))
# Update customer stats
cursor.execute("""
UPDATE customers
SET total_orders = total_orders + 1, total_spent = total_spent + ?
WHERE id = ?
""", (total_amount, customer_id))
conn.commit()
conn.close()
result = f"✅ **הזמנה #{order_id} נוצרה בהצלחה!**\n\n"
result += f"👤 לקוח: {customer[0]}\n"
result += f"💰 סה\"כ: ₪{total_amount:.2f}\n\n"
result += "📦 **פריטים בהזמנה:**\n"
for item in order_details:
result += f"• {item['name']} x{item['quantity']} - ₪{item['total_price']:.2f}\n"
if shipping_address:
result += f"\n📍 כתובת משלוח: {shipping_address}"
return result
except Exception as e:
return f"❌ שגיאה ביצירת הזמנה: {str(e)}"
if __name__ == "__main__":
mcp.run()
