MCP Servers

# بناء MCP Server بسيط بـ Python
# يربط Claude بقاعدة بيانات SQLite محلية

# التثبيت: pip install mcp

import sqlite3
import asyncio
from mcp.server import Server
from mcp.types import Tool, TextContent, Resource, ReadResourceResult
import mcp.server.stdio

# إنشاء قاعدة بيانات تجريبية
def init_db():
    conn = sqlite3.connect("academy.db")
    conn.execute("""
        CREATE TABLE IF NOT EXISTS students (
            id TEXT PRIMARY KEY,
            name TEXT,
            email TEXT,
            courses_enrolled INTEGER DEFAULT 0,
            last_login TEXT
        )
    """)
    conn.execute("""
        INSERT OR IGNORE INTO students VALUES
        ('s001', 'أحمد محمد', 'ahmed@example.com', 3, '2026-06-01'),
        ('s002', 'سارة علي', 'sara@example.com', 5, '2026-06-05'),
        ('s003', 'محمد خالد', 'm.khalid@example.com', 1, '2026-05-28')
    """)
    conn.commit()
    conn.close()

init_db()

# إنشاء الـ MCP Server
server = Server("academy-db-server")

@server.list_tools()
async def list_tools() -> list[Tool]:
    """قائمة الأدوات التي يمكن لـ Claude استخدامها"""
    return [
        Tool(
            name="get_student",
            description="احصل على معلومات طالب بالـ ID",
            inputSchema={
                "type": "object",
                "properties": {
                    "student_id": {
                        "type": "string",
                        "description": "معرّف الطالب (مثل: s001)"
                    }
                },
                "required": ["student_id"]
            }
        ),
        Tool(
            name="list_students",
            description="احصل على قائمة كل الطلاب",
            inputSchema={"type": "object", "properties": {}}
        ),
        Tool(
            name="run_query",
            description="تشغيل استعلام SQL للقراءة فقط",
            inputSchema={
                "type": "object",
                "properties": {
                    "sql": {
                        "type": "string",
                        "description": "استعلام SELECT فقط"
                    }
                },
                "required": ["sql"]
            }
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    """تنفيذ الأداة المطلوبة"""
    conn = sqlite3.connect("academy.db")
    conn.row_factory = sqlite3.Row

    try:
        if name == "get_student":
            row = conn.execute(
                "SELECT * FROM students WHERE id = ?",
                (arguments["student_id"],)
            ).fetchone()

            if not row:
                return [TextContent(type="text", text="الطالب غير موجود")]

            result = dict(row)
            return [TextContent(type="text", text=str(result))]

        elif name == "list_students":
            rows = conn.execute("SELECT * FROM students").fetchall()
            result = [dict(r) for r in rows]
            return [TextContent(type="text", text=str(result))]

        elif name == "run_query":
            sql = arguments["sql"].strip().upper()
            # أمان: SELECT فقط
            if not sql.startswith("SELECT"):
                return [TextContent(type="text", text="❌ مسموح فقط بـ SELECT")]

            rows = conn.execute(arguments["sql"]).fetchall()
            result = [dict(r) for r in rows]
            return [TextContent(type="text", text=str(result))]

    finally:
        conn.close()

    return [TextContent(type="text", text="أداة غير معروفة")]

async def main():
    print("🚀 MCP Academy Server يعمل...")
    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await server.run(read_stream, write_stream, server.create_initialization_options())

if __name__ == "__main__":
    asyncio.run(main())

# ──────────────────────────────────────
# إضافة الـ Server لـ Claude Code:
# claude mcp add academy-db -- python path/to/this/server.py