بناء MCP Server من الصفر مع Python — الدليل الشامل 2026

📋 فهرس المحتويات

• ما هو MCP؟ — المفهوم والسبب
• كيف يعمل MCP؟ — هندسة البروتوكول
• دورة حياة اتصال MCP — المصافحة والتفاوض
• البيئة والتثبيت — Python, UV, MCP SDKs
• المشروع 1: خادم الطقس (Anthropic MCP SDK)
• المشروع 2: خادم SQLite (OpenAI Agents SDK)
• المشروع 3: خادم نظام الملفات (FastMCP)
• المشروع 4: البوابة المتعددة الخوادم (Gateway)
• كيف تبني MCP Client مخصصاً
• نشر MCP مع SSE في الإنتاج
• مقارنة SDKs: متى تستخدم ماذا؟
• MCP في الإنتاج — أفضل الممارسات
• حل المشكلات الشائعة (Troubleshooting)
• القيود والتحديات الحالية
• مستقبل MCP والتوجهات
• خاتمة
• الأسئلة الشائعة (FAQ)

1. ما هو MCP؟ — المفهوم والسبب

Model Context Protocol (MCP) هو بروتوكول مفتوح المصدر طورته Anthropic في نوفمبر 2024، يهدف إلى توحيد طريقة تواصل نماذج الذكاء الاصطناعي مع الأدوات والخدمات الخارجية. الفكرة بسيطة وجوهرية: بدلاً من أن يبني كل مطور تكاملاً خاصاً لكل نموذج مع كل أداة (علاقة N×M)، يوفّر MCP واجهة موحدة شبيهة بـ USB-C للذكاء الاصطناعي — أي نموذج يتحدث مع أي خادم أدوات عبر بروتوكول معياري.

لماذا نحتاج MCP؟ قبل ظهوره، كان كل نموذج LLM يتصل بالأدوات بطريقته الخاصة:

• OpenAI: function calling مع JSON Schema مخصص
• Anthropic: tool use مع Blocks API خاص
• Google: tool_config مع بنية مختلفة
• Open-source: كل إطار يخترع عجلته الخاصة (LangChain tools, CrewAI tools, AutoGPT plugins)

نتيجة هذا التشتت: إذا أردت بناء وكيل يستخدم قاعدة بيانات SQLite و خدمة طقس و وصولاً لنظام الملفات، كان عليك كتابة 3 تكاملات مختلفة لكل نموذج تريد دعمه. MCP يحل هذه المشكلة بجعل كل خادم أدوات يتحدث JSON-RPC عبر معيار واحد، وكل عميل (Client LLM) يفهم هذا المعيار نفسه.

التشبيه الأفضل: فكر في MCP كـ USB-C للذكاء الاصطناعي. قبل USB-C، كل جهاز كان يحتاج كابل مختلف. الـ MCP هو المعيار الذي يسمح لأي LLM (جهاز) بالاتصال بأي خادم أدوات (طرفية) عبر كابل واحد.

منذ إطلاقه في نوفمبر 2024، تبنّى MCP بسرعة كبيرة. بحلول يونيو 2026، يدعمه:

• Claude Desktop — الدعم الأصلي الأول والأقوى
• OpenAI Agents SDK — دعم رسمي منذ مارس 2025
• GitHub Copilot — دعم رسمي لربط MCP Servers كأدوات
• Cline / Roo Code — أدوات VS Code/AI Coding تدعم MCP
• Continue.dev — إطار IDE المفتوح يدعم MCP
• LangChain — دعم عبر LangChain MCP Adapter
• n8n — منصة الأتمتة تدعم MCP كعقد (node)

2. كيف يعمل MCP؟ — هندسة البروتوكول

MCP يتبع بنية Client-Server بسيطة لكنها قوية:

┌──────────────┐        JSON-RPC (stdio/SSE)        ┌──────────────┐
│  MCP Client  │ ────────────────────────────────→ │  MCP Server  │
│  (LLM Host)  │ ←──────────────────────────────── │  (Tools)     │
└──────────────┘        Resources + Prompts         └──────────────┘
       │                                                    │
       │  HTTP/WebSocket                                     │  Local API/DB
       ▼                                                    ▼
  ┌──────────┐                                       ┌──────────┐
  │  Claude  │                                       │  SQLite  │
  │  GPT-5       │                                       │  REST    │
  │  Gemini 2.5  │                                    │  FS      │
  └──────────┘                                       └──────────┘

المكونات الأساسية

البروتوكول يحدد 3 أنواع رئيسية من الموارد التي يمكن للخادم تقديمها:

• 🛠️ Tools (الأدوات): دوال قابلة للاستدعاء تشبه API endpoints. الـ LLM يمكنه طلب استدعائها بمعاملات JSON. تستخدم عادةً للقراءة/الكتابة الفورية. كل أداة لها اسم فريد، وصف، و JSON Schema يحدد معاملاتها.
• 📄 Resources (الموارد): بيانات منظمة تشبه القراءة من ملفات. تدعم URI scheme مخصص (weather://دمشق, db://products/1) وتستخدم للإعلام والتحديثات المستمرة.
• 💬 Prompts (القوالب): قوالب جاهزة يمكن للـ LLM طلبها لتوجيه سلوكه في سياقات معينة. مفيدة لتوحيد أسلوب الإجابات.

طبقات النقل (Transport)

MCP يدعم طبقتين للنقل:

• stdio: الخادم يعمل كعملية فرعية (subprocess) ويتواصل عبر stdin/stdout. مثالي للتنمية المحلية والأدوات التي تعمل على نفس الجهاز. أبسط وأسرع — لا يحتاج خادم HTTP أو ميناء مفتوح.
• SSE (Server-Sent Events): الخادم يعمل كخادم HTTP ويتواصل عبر streams. مناسب للإنتاج والتوزيع عبر الشبكة. يدعم الاتصالات البعيدة والعديد من العملاء في وقت واحد.
• Streamable HTTP (جديد — v2025-06-18): طبقة نقل مبسّطة تعتمد على HTTP POST فقط — لا تحتاج لاتصال SSE دائم. العميل يرسل طلب POST مع رسالة JSON-RPC، والخادم يرد مباشرة. مثالي لبيئات Serverless و Lambda حيث لا يمكن الاحتفاظ باتصال طويل الأمد. يجمع بين بساطة stdio وفوائد HTTP. مدعوم في MCP SDK منذ v1.25+.

JSON-RPC 2.0 — لغة التواصل

جميع الرسائل بين العميل والخادم تتبع معيار JSON-RPC 2.0. كل رسالة تحتوي على:

{
  "jsonrpc": "2.0",
  "method": "tools/call",
  "params": {
    "name": "get_weather",
    "arguments": {
      "city": "دمشق",
      "units": "celsius"
    }
  },
  "id": "req-001"
}

الطرق الأساسية:

• initialize — المصافحة الأولية بين العميل والخادم
• tools/list — العميل يسأل: ما الأدوات المتوفرة؟
• tools/call — العميل يطلب استدعاء أداة محددة
• resources/list — العميل يسأل: ما الموارد المتوفرة؟
• resources/read — العميل يقرأ مورداً محدداً
• prompts/list — العميل يسأل: ما القوالب المتوفرة؟
• prompts/get — العميل يطلب قالباً معيناً

3. دورة حياة اتصال MCP — المصافحة والتفاوض

قبل أن ننتقل للتنفيذ العملي، من المهم فهم كيف يبدأ اتصال MCP ويتطور. الاتصال بين العميل (Client) والخادم (Server) يتبع تسلسلاً محدداً:

المرحلة 1: الإطلاق (Initialization)

عندما يبدأ الخادم، يرسل العميل طلب initialize يحتوي على:

• إصدار البروتوكول المدعوم (Protocol Version)
• قدرات العميل (Client Capabilities) — مثلاً: هل يدعم tools؟ resources؟ prompts؟
• معلومات التعريف (Client Info) — اسم العميل وإصداره

الخادم يردّ بقدراته الخاصة وإصدار البروتوكول الذي يختاره. هذه المصافحة تضمن أن الطرفين يتحدثان نفس اللغة قبل تبادل أي رسائل أخرى.

// الطلب — من العميل إلى الخادم
{
  "jsonrpc": "2.0",
  "method": "initialize",
  "params": {
    "protocolVersion": "2025-03-26",
    "capabilities": {
      "tools": { "listChanged": true },
      "resources": { "subscribe": true },
      "prompts": {}
    },
    "clientInfo": {
      "name": "claude-desktop",
      "version": "1.0.0"
    }
  },
  "id": "1"
}

// الاستجابة — من الخادم إلى العميل
{
  "jsonrpc": "2.0",
  "id": "1",
  "result": {
    "protocolVersion": "2025-03-26",
    "capabilities": {
      "tools": {},
      "resources": {},
      "prompts": {}
    },
    "serverInfo": {
      "name": "weather-server",
      "version": "1.0.0"
    }
  }
}

المرحلة 2: الإشعار بالتهيئة (Initialized Notification)

بعد المصافحة الناجحة، العميل يرسل إشعار initialized. هذا الإشعار ليس له رد — هو مجرد إعلام للخادم بأن العميل جاهز لاستقبال الرسائل. بعد هذه اللحظة، يبدأ التبادل الحقيقي للرسائل:

{
  "jsonrpc": "2.0",
  "method": "notifications/initialized",
  "params": {}
}

المرحلة 3: اكتشاف الأدوات (Tool Discovery)

بعد التهيئة، يقوم العميل باستدعاء tools/list لمعرفة الأدوات المتوفرة في الخادم:

// سؤال: ما الأدوات المتوفرة؟
{ "jsonrpc": "2.0", "method": "tools/list", "params": {}, "id": "2" }

// جواب: إليك الأدوات مع مواصفات كل منها
{
  "jsonrpc": "2.0", "id": "2",
  "result": {
    "tools": [{
      "name": "get_weather",
      "description": "الحصول على بيانات الطقس",
      "inputSchema": {
        "type": "object",
        "properties": {
          "city": { "type": "string", "description": "اسم المدينة" },
          "units": { "type": "string", "enum": ["celsius", "fahrenheit"] }
        },
        "required": ["city"]
      }
    }]
  }
}

يقوم العميل (مثل Claude Desktop) بتمرير هذه المواصفات للـ LLM، الذي يمكنه الآن اتخاذ قرار: هل يريد استدعاء هذه الأداة؟ إذا قرر ذلك:

{
  "jsonrpc": "2.0",
  "method": "tools/call",
  "params": {
    "name": "get_weather",
    "arguments": { "city": "دمشق", "units": "celsius" }
  },
  "id": "3"
}

الخادم ينفذ الأداة ويعيد النتيجة. هذا كل شيء! هذا النمط البسيط يسمح بتكامل غير محدود — أي أداة، أي خدمة، أي قاعدة بيانات — طالما تتبع هذا التسلسل.

المرحلة 4: دورة حياة رسالة كاملة — من المستخدم إلى الأداة والعودة

لنرى كيف تتدفق رسالة كاملة من المستخدم إلى النتيجة النهائية:

1. المستخدم يكتب: “كم درجة الحرارة في دمشق اليوم؟”
2. Claude Desktop (العميل) يرسل الرسالة لـ Claude LLM
3. Claude LLM يقرر: أحتاج أداة get_weather مع {"city": "دمشق", "units": "celsius"}
4. العميل يرسل tools/call عبر JSON-RPC للخادم
5. الخادم ينفذ الأداة (يبحث في قاعدة بيانات الطقس المحاكاة)
6. الخادم يعيد: {"content": [{"type": "text", "text": "🌤️ طقس دمشق: 28°C..."}]}
7. العميل يمرر النتيجة لـ Claude LLM
8. Claude يصيغ الرد النهائي: “درجة الحرارة في دمشق الآن 28°C…”

كل هذا يحدث في أقل من ثانية للمستخدم. هذا هو جمال MCP — فصل تام بين منطق الأداة وذكاء النموذج.

4. البيئة والتثبيت — Python, UV, MCP SDKs

قبل البدء في بناء الخوادم، تحتاج إلى تجهيز البيئة. سنستخدم Python 3.11+ و UV (مدير حزم سريع جداً) و MCP SDKs الثلاثة الرئيسية:

تثبيت UV (مدير الحزم فائق السرعة)

# تثبيت UV — مدير حزم أسرع 10-100x من pip
curl -LsSf https://astral.sh/uv/install.sh | sh

# تفعيله
source ~/.bashrc  # أو أعد فتح الطرفية

لماذا UV بدلاً من pip العادي؟ UV يركّب الحزم بشكل متوازٍ وقد يثبّت 30 حزمة في 2 ثانية حيث يحتاج pip العادي دقيقة كاملة. بالإضافة إلى أنه يدير إصدارات Python أيضاً.

تثبيت MCP SDKs الثلاثة

# 1. Anthropic MCP SDK (الأساسي والرسمي)
uv pip install "mcp[cli]"

# 2. OpenAI Agents SDK (يدعم MCP منذ مارس 2025)
uv pip install openai-agents

# 3. FastMCP — إطار مفتوح المصدر (يبسّط بناء الخوادم)
uv pip install fastmcp

# تحقق من التثبيت
python3 -c "import mcp; print(f'MCP SDK: {mcp.__version__}')"
python3 -c "import fastmcp; print(f'FastMCP: {fastmcp.__version__}')"
python3 -c "import agents; print(f'OpenAI Agents: {agents.__version__}')"

مقارنة سريعة بين طرق التثبيت

├── pip install mcp           ← 300+ حزمة، 3-5 دقائق
├── uv pip install mcp        ← نفس الحزم، 2-3 ثوانٍ
├── pipx install mcp          ← معزول، لكن بطيء
└── docker pull mcp/server    ← 500MB، يعمل فوراً

هيكل المشروع النموذجي لكل خادم

mcp-server/
├── server.py          # كود الخادم الرئيسي
├── requirements.txt   # التبعيات
├── README.md          # توثيق الاستخدام
├── setup.sh           # سكريبت التثبيت السريع
└── .gitignore         # تجاهل الملفات غير الضرورية

5. المشروع 1: خادم الطقس (Anthropic MCP SDK)

أول مشروع عملي: خادم MCP يوفّر بيانات الطقس لمدن متعددة. هذا الخادم يستخدم Anthropic MCP SDK الرسمي ويعمل كمثال تمهيدي لفهم أساسيات البروتوكول.

رابط المستودع الكامل: github.com/moaaz01/mcp-weather-server

الكود الكامل (weather_server.py)

from datetime import datetime
from typing import Any
from mcp.server.fastmcp import FastMCP

# ─── قاعدة بيانات الطقس المحاكاة ──────────────────────────────
WEATHER_DATA: dict[str, dict[str, Any]] = {
    "دمشق": {
        "temp": 28, "humidity": 35, "condition": "مشمس",
        "wind_speed": 12, "aqi": 65, "forecast": [
            {"day": "اليوم", "high": 30, "low": 18, "condition": "مشمس"},
            {"day": "غداً", "high": 28, "low": 17, "condition": "غائم جزئياً"},
        ]
    },
    "القاهرة": {
        "temp": 33, "humidity": 55, "condition": "حار",
        "wind_speed": 8, "aqi": 120, "forecast": [
            {"day": "اليوم", "high": 35, "low": 24, "condition": "حار"},
            {"day": "غداً", "high": 32, "low": 22, "condition": "مشمس"},
        ]
    },
    "دبي": {
        "temp": 38, "humidity": 40, "condition": "حار جداً",
        "wind_speed": 15, "aqi": 85, "forecast": [
            {"day": "اليوم", "high": 40, "low": 28, "condition": "حار جداً"},
            {"day": "غداً", "high": 39, "low": 27, "condition": "حار"},
        ]
    },
}

mcp = FastMCP(
    "Weather Server",
    description="خادم طقس عربي — بيانات محاكاة لمدن عربية",
    version="1.0.0",
)


# 🛠️ الأداة 1: الحصول على طقس مدينة
@mcp.tool(description="الحصول على بيانات الطقس الحالية لمدينة محددة")
def get_weather(city: str, units: str = "celsius") -> str:
    """الحصول على درجة الحرارة والرطوبة وحالة الطقس لمدينة عربية."""
    data = _find_city(city)
    if not data:
        return f"❌ المدينة '{city}' غير موجودة. المدن المتوفرة: {_city_list()}"

    temp = data["temp"]
    if units == "fahrenheit":
        temp = round(temp * 9 / 5 + 32)
    unit_label = "°C" if units == "celsius" else "°F"

    return (
        f"🌤️  طقس {city}:\n"
        f"   • درجة الحرارة: {temp}{unit_label}\n"
        f"   • الحالة: {data['condition']}\n"
        f"   • الرطوبة: {data['humidity']}%\n"
        f"   • سرعة الرياح: {data['wind_speed']} كم/س\n"
        f"   • جودة الهواء (AQI): {data['aqi']}"
    )


# 🛠️ الأداة 2: توقعات الطقس
@mcp.tool(description="الحصول على توقعات الطقس للأيام القادمة")
def get_forecast(city: str, days: int = 3) -> str:
    """توقعات الطقس لعدد محدد من الأيام (حد أقصى 7 أيام)."""
    data = _find_city(city)
    if not data:
        return f"❌ المدينة '{city}' غير موجودة"

    days = min(max(days, 1), 7)
    lines = [f"📅  توقعات الطقس لـ {city} (الـ {days} أيام القادمة):\n"]
    for f in data["forecast"][:days]:
        lines.append(f"   • {f['day']}: {f['high']}°C / {f['low']}°C — {f['condition']}")
    return "\n".join(lines)


# 🛠️ الأداة 3: مقارنة مدن
@mcp.tool(description="مقارنة الطقس بين مدينتين أو أكثر")
def compare_cities(cities: list[str]) -> str:
    """مقارنة درجات الحرارة والظروف الجوية بين قائمة مدن."""
    results = []
    for city in cities:
        found = _find_city(city)
        if found:
            results.append(f"🌆 {city}: {found['temp']}°C, {found['condition']}, رطوبة {found['humidity']}%")
        else:
            results.append(f"❌ {city}: غير موجودة")
    return "مقارنة المدن:\n" + "\n".join(results)


# 📄 المورد: معلومات الطقس كـ JSON منظم
@mcp.resource("weather://{city}/current")
def current_weather_resource(city: str) -> str:
    import json
    data = _find_city(city)
    if not data:
        return f"{{'error': 'المدينة {city} غير موجودة'}}"
    data["city"] = city
    data["last_updated"] = datetime.now().isoformat()
    return json.dumps(data, ensure_ascii=False, indent=2)


# ─── دوال مساعدة ──────────────────────────────────────────────
def _find_city(city: str) -> dict | None:
    """بحث مرن عن المدينة."""
    for c, d in WEATHER_DATA.items():
        if city.lower() in c.lower():
            return d
    return None


def _city_list() -> str:
    return "، ".join(WEATHER_DATA.keys())


# 🚀 تشغيل الخادم
if __name__ == "__main__":
    mcp.run(transport="stdio")

شرح الكود — FastMCP API

• السطر 1-2: استيراد datetime لتوقيت التحديث و FastMCP من حزمة MCP الرسمية — الخطاف الوحيد اللي تحتاجه.
• السطر 5-42: WEATHER_DATA — قاموس يخزّن بيانات الطقس المحاكاة لـ 3 مدن عربية. في الإنتاج، ستستبدل هذا بـ API حقيقي مثل OpenWeatherMap.
• السطر 44-48: mcp = FastMCP(...) — إنشاء كائن الخادم باسم فريد. FastMCP هي الطبقة العليا (high-level) من MCP SDK الرسمي من Anthropic/AAIF.
• السطر 51-90: تعريف 3 أدوات مع @mcp.tool() — نظام types تلقائي: city: str يصبح تلقائياً معلمة مطلوبة، units: str = "celsius" يصبح اختيارياً مع قيمة افتراضية. ممنوع الـ JSON Schema اليدوي!
• السطر 93-100: المورد (@mcp.resource) — يسمح للعميل بقراءة بيانات الطقس كاملة بتنسيق JSON منظم عن طريق URI متغير مثل weather://دمشق/current.
• السطر 103-112: الدوال المساعدة — _find_city للبحث المرن و _city_list لعرض المدن المتاحة.
• السطر 115-116: نقطة الدخول. سطر واحد: mcp.run(transport="stdio"). لا asyncio.run()، لا InitializationOptions، لا NotificationOptions.

كيفية تشغيل الخادم

# التشغيل عبر stdio (مع Claude Desktop)
python3 weather_server.py

# اختبار الخادم عبر MCP Inspector (أداة تصحيح رسمية)
npx @aaif/mcp-inspector python3 weather_server.py

# في Claude Desktop، أضف في claude_desktop_config.json:
# {
#   "mcpServers": {
#     "weather": {
#       "command": "python3",
#       "args": ["/path/to/weather_server.py"]
#     }
#   }
# }

ماذا تعلمنا من هذا المشروع؟

• كيفية تعريف الأدوات مع JSON Schema يدوياً
• الفرق بين list_tools() (اكتشاف) و call_tool() (تنفيذ)
• كيفية إضافة الموارد (Resources) كـ URI مخصص
• نظام types في Anthropic MCP SDK (TextContent, ImageContent, EmbeddedResource)
• كيفية التعامل مع الأخطاء (مدينة غير موجودة)

هذا المشروع يغطّي المفاهيم الأساسية: تعريف الأدوات (Tools) مع JSON Schema، الموارد (Resources) عبر URI مخصص، والتعامل مع أنواع البيانات المختلفة (نصوص، JSON). أي خادم MCP يبدأ من هذا الأساس.

6. المشروع 2: خادم SQLite (OpenAI Agents SDK)

المشروع الثاني يأخذنا لمستوى أعلى: خادم MCP يتصل بقاعدة بيانات SQLite حقيقية ويسمح للـ LLM بتنفيذ استعلامات SQL بشكل آمن. هذا المثال يوضح كيف يمكن استخدام MCP لتوفير وصول منظم ومأمون للبيانات.

المميز هنا أننا نستخدم OpenAI Agents SDK لبناء العميل، مما يثبت قابلية MCP للتشغيل عبر مختلف منصات LLM — ليس فقط Claude.

رابط المستودع: github.com/moaaz01/mcp-sqlite-server

الكود الكامل (sqlite_server.py)

import sqlite3
import os
from datetime import datetime
from mcp.server.fastmcp import FastMCP

DB_PATH = os.path.join(os.path.dirname(__file__), "sample.db")
mcp = FastMCP(
    "SQLite Server",
    description="خادم قاعدة بيانات SQLite — استعلامات وتحليلات",
    version="1.0.0",
)


# ─── تهيئة قاعدة البيانات ───────────────────────────
def init_db():
    """إنشاء الجداول وبيانات تجريبية إذا لم تكن موجودة"""
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()

    cursor.executescript("""
        CREATE TABLE IF NOT EXISTS products (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT NOT NULL,
            category TEXT NOT NULL,
            price REAL NOT NULL,
            stock INTEGER DEFAULT 0,
            created_at TEXT DEFAULT (datetime('now'))
        );

        CREATE TABLE IF NOT EXISTS orders (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            customer_id INTEGER NOT NULL,
            order_date TEXT DEFAULT (datetime('now')),
            total REAL NOT NULL,
            status TEXT DEFAULT 'pending'
        );

        CREATE TABLE IF NOT EXISTS order_items (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            order_id INTEGER NOT NULL,
            product_id INTEGER NOT NULL,
            quantity INTEGER NOT NULL,
            price REAL NOT NULL,
            FOREIGN KEY (order_id) REFERENCES orders(id),
            FOREIGN KEY (product_id) REFERENCES products(id)
        );

        CREATE TABLE IF NOT EXISTS customers (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            name TEXT NOT NULL,
            email TEXT UNIQUE NOT NULL,
            city TEXT,
            registered_at TEXT DEFAULT (datetime('now'))
        );

        -- بيانات تجريبية إذا كانت الجداول فارغة
        INSERT OR IGNORE INTO products (id, name, category, price, stock) VALUES
            (1, 'لابتوب ProBook', 'إلكترونيات', 4500, 15),
            (2, 'هاتف ذكي X10', 'إلكترونيات', 2200, 30),
            (3, 'سماعات لاسلكية', 'إكسسوارات', 350, 50),
            (4, 'لوحة مفاتيح ميكانيكية', 'إكسسوارات', 450, 25),
            (5, 'شاشة 27 بوصة', 'إلكترونيات', 1800, 10);

        INSERT OR IGNORE INTO customers (id, name, email, city) VALUES
            (1, 'أحمد علي', '[email protected]', 'دمشق'),
            (2, 'سارة خالد', '[email protected]', 'القاهرة'),
            (3, 'محمد عمر', '[email protected]', 'دبي');

        INSERT OR IGNORE INTO orders (id, customer_id, total, status) VALUES
            (1, 1, 4500, 'completed'),
            (2, 2, 2550, 'pending'),
            (3, 3, 1800, 'shipped');
    """)

    conn.commit()
    conn.close()


@mcp.tool(description="تنفيذ استعلام SQL للقراءة فقط (SELECT)")
def query(sql: str) -> str:
    """تنفيذ استعلام SQL آمن (SELECT فقط)."""
    sql_upper = sql.strip().upper()
    if not sql_upper.startswith("SELECT"):
        return "❌ يُسمح فقط باستعلامات SELECT للأمان"

    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    try:
        import json
        cursor = conn.execute(sql)
        rows = [dict(row) for row in cursor.fetchall()]
        return f"✅ الاستعلام أعاد {len(rows)} نتيجة:\n\n" + \
               json.dumps(rows, ensure_ascii=False, indent=2)
    finally:
        conn.close()


@mcp.tool(description="الحصول على ملخص شامل لجدول: عدد السجلات + المخطط")
def summarize_table(table: str) -> str:
    """ملخص تفصيلي لجدول — الأعمدة والأنواع وعدد السجلات."""
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    try:
        cursor = conn.execute(f"SELECT COUNT(*) as c FROM {table}")
        count = cursor.fetchone()["c"]
        cursor = conn.execute(f"PRAGMA table_info({table})")
        columns = [{"name": row["name"], "type": row["type"]}
                   for row in cursor.fetchall()]

        return (
            f"📊 ملخص جدول '{table}':\n"
            f"   • عدد السجلات: {count}\n"
            f"   • عدد الأعمدة: {len(columns)}\n"
            f"   • الأعمدة:\n" +
            "\n".join(f"      - {c['name']} ({c['type']})" for c in columns)
        )
    finally:
        conn.close()


@mcp.tool(description="تنفيذ تحليل مخصص على البيانات (JOINs, GROUP BY, إحصائيات)")
def run_analysis(analysis_type: str) -> str:
    """تحليل مبيعات، مخزون، أو إحصائيات طلبات."""
    analyses = {
        "top_products": """
            SELECT name, price, stock FROM products
            ORDER BY price DESC LIMIT 5
        """,
        "sales_by_category": """
            SELECT p.category, COUNT(oi.id) as units_sold,
                   SUM(oi.quantity * oi.price) as revenue
            FROM order_items oi
            JOIN products p ON oi.product_id = p.id
            GROUP BY p.category
        """,
        "customer_orders": """
            SELECT c.name, COUNT(o.id) as orders_count,
                   SUM(o.total) as total_spent
            FROM customers c
            LEFT JOIN orders o ON c.id = o.customer_id
            GROUP BY c.id
        """,
        "inventory_status": """
            SELECT name, stock,
                   CASE WHEN stock = 0 THEN 'نفذ'
                        WHEN stock < 20 THEN 'مخزون منخفض'
                        ELSE 'متوفر' END as status
            FROM products
        """,
        "order_summary": """
            SELECT status, COUNT(*) as count, SUM(total) as total
            FROM orders GROUP BY status
        """,
    }

    if analysis_type not in analyses:
        return f"❌ تحليل غير معروف: {analysis_type}"

    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    try:
        import json
        cursor = conn.execute(analyses[analysis_type])
        rows = [dict(row) for row in cursor.fetchall()]
        return f"📈 تحليل '{analysis_type}':\n\n" + \
               json.dumps(rows, ensure_ascii=False, indent=2)
    finally:
        conn.close()


# 🚀 التشغيل
if __name__ == "__main__":
    init_db()
    mcp.run(transport="stdio")

الاتصال من OpenAI Agents SDK

هذا الكود يركض كعميل MCP يتصل بالخادم أعلاه — ليس كخادم. هذا يثبت أن الخادم الذي بنيناه بأنثروبيك SDK يمكن استخدامه من OpenAI:

from agents import Agent, Runner
from agents.mcp import MCPServerStdio

async def main():
    async with MCPServerStdio(
        params={"command": "python3", "args": ["sqlite_server.py"]}
    ) as server:
        agent = Agent(
            name="SQLite Assistant",
            instructions="أنت مساعد قاعدة بيانات. استخدم الأدوات المتاحة للإجابة عن أسئلة المستخدم.",
            mcp_servers=[server],
            model="gpt-5",
        )
        result = await Runner.run(
            agent,
            input="ما هي أكثر فئة منتجات مبيعاً؟ وما إجمالي الإيرادات لكل فئة؟"
        )
        print(result.final_output)

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

ماذا يعني هذا للمطورين العرب؟

هذا المشروع يوضح نقطة محورية: MCP ليس حكراً على Claude. OpenAI Agents SDK يدعم MCP بشكل رسمي منذ مارس 2025، مما يعني أن نفس خادم SQLite يمكن استخدامه مع GPT-5 و Gemini 2.5 وحتى النماذج المفتوحة المصدر. أنت تبني الأداة مرة واحدة — وأي LLM في المستقبل سيتمكن من استخدامها.

7. المشروع 3: خادم نظام الملفات (FastMCP)

المشروع الثالث يستخدم FastMCP — إطار مفتوح المصدر يبسّط بناء خوادم MCP بشكل كبير. مع FastMCP، يمكنك تعريف أداة بسطر واحد باستخدام الديكوريتور @mcp.tool().

رابط المستودع: github.com/moaaz01/mcp-file-system-server

الكود الكامل (filesystem_server.py)

import os
import shutil
import fnmatch
from datetime import datetime
from pathlib import Path
from fastmcp import FastMCP

mcp = FastMCP("File System Server")
ALLOWED_DIRS = [os.path.expanduser("~"), "/tmp"]

# ─── دوال مساعدة ─────────────────────────────
def resolve_safe_path(user_path: str) -> Path | None:
    """تحويل مسار نسبي إلى مسار مطلق ضمن المسموح"""
    path = Path(user_path).expanduser().resolve()
    for allowed in ALLOWED_DIRS:
        allowed_path = Path(allowed).resolve()
        try:
            path.relative_to(allowed_path)
            return path
        except ValueError:
            continue
    return None

def format_size(size: int) -> str:
    for unit in ['B', 'KB', 'MB', 'GB']:
        if size < 1024:
            return f"{size:.1f} {unit}"
        size /= 1024
    return f"{size:.1f} TB"

def safe_pattern(pattern: str) -> list[str]:
    """تأمين نمط glob — يمنع الوصول للمسارات الحساسة"""
    sensitive = [".ssh", ".aws", ".config", ".git", "token", "secret", "key", "password"]
    if any(s in pattern.lower() for s in sensitive):
        raise ValueError(f"❌ النمط '{pattern}' يحتوي على مسارات حساسة—ممنوع")
    return [pattern]

# 🛠️ الأدوات
@mcp.tool()
async def list_directory(path: str = ".") -> str:
    """سرد محتويات مجلد"""
    resolved = resolve_safe_path(path)
    if not resolved or not resolved.is_dir():
        return f"❌ المسار غير موجود أو غير مسموح به: {path}"

    entries = []
    for entry in sorted(resolved.iterdir()):
        is_dir = entry.is_dir()
        size = ""
        if entry.is_file():
            size = format_size(entry.stat().st_size)
        modified = datetime.fromtimestamp(entry.stat().st_mtime).strftime("%Y-%m-%d %H:%M")
        icon = "📁" if is_dir else "📄"
        entries.append(f"{icon} {entry.name:<30} {size:>8}  {modified}")

    return f"📂 محتويات {resolved}:\n" + "\n".join(entries)

@mcp.tool()
async def read_file(path: str) -> str:
    """قراءة محتوى ملف نصي"""
    resolved = resolve_safe_path(path)
    if not resolved or not resolved.is_file():
        return f"❌ الملف غير موجود: {path}"
    if resolved.stat().st_size > 1024 * 1024:
        return f"❌ الملف كبير جداً (أكبر من 1MB): {path}"

    try:
        content = resolved.read_text(encoding="utf-8")
        return f"📄 {resolved.name} ({format_size(resolved.stat().st_size)}):\n\n{content[:10000]}"
    except UnicodeDecodeError:
        return f"❌ الملف ثنائي ولا يمكن قراءته كنص: {path}"

@mcp.tool()
async def search_files(pattern: str, path: str = ".") -> str:
    """البحث عن ملفات بنمط معين (glob wildcard)"""
    resolved = resolve_safe_path(path)
    if not resolved or not resolved.is_dir():
        return f"❌ المسار غير صالح: {path}"

    safe_pattern(pattern)
    matches = []
    for root, _, files in os.walk(resolved):
        for f in files:
            if fnmatch.fnmatch(f, pattern):
                full = Path(root) / f
                try:
                    full.relative_to(Path(ALLOWED_DIRS[0]).resolve())
                    matches.append(full)
                except ValueError:
                    continue
        if len(matches) >= 50:
            break

    if not matches:
        return f"🔍 لم يتم العثور على ملفات تطابق '{pattern}' في {resolved}"
    return f"🔍 تم العثور على {len(matches)} ملف:\n" + "\n".join(
        f"  📄 {m.relative_to(resolved)} ({format_size(m.stat().st_size)})"
        for m in matches
    )

@mcp.tool()
async def get_file_info(path: str) -> str:
    """الحصول على معلومات مفصلة عن ملف أو مجلد"""
    resolved = resolve_safe_path(path)
    if not resolved:
        return f"❌ المسار غير موجود: {path}"
    stat = resolved.stat()
    is_dir = resolved.is_dir()
    return (
        f"{'📁' if is_dir else '📄'} معلومات {resolved.name}:\n"
        f"  • المسار: {resolved}\n"
        f"  • {'مجلد' if is_dir else 'ملف'}\n"
        f"  • الحجم: {format_size(stat.st_size)}\n"
        f"  • آخر تعديل: {datetime.fromtimestamp(stat.st_mtime).strftime('%Y-%m-%d %H:%M:%S')}\n"
        f"  • الصلاحيات: {oct(stat.st_mode)[-3:]}"
        + (f"\n  • الملفات: {len(list(resolved.iterdir()))}" if is_dir else "")
    )

@mcp.tool()
async def disk_usage(path: str = "~") -> str:
    """فحص استخدام القرص لمجلد معين"""
    resolved = resolve_safe_path(path)
    if not resolved:
        return f"❌ المسار غير موجود: {path}"
    total = sum(f.stat().st_size for f in resolved.rglob("*") if f.is_file())
    count = len(list(resolved.rglob("*")))
    return (
        f"💾 استخدام القرص لـ {resolved}:\n"
        f"  • إجمالي الحجم: {format_size(total)}\n"
        f"  • عدد الملفات: {count}\n"
        f"  • المساحة المتوقعة: {format_size(total // 1024 * 1024)}"
    )

if __name__ == "__main__":
    mcp.run()

لماذا FastMCP؟ مقارنة مع Anthropic SDK

لاحظ كم هو أبسط! مع FastMCP:

• لا حاجة لتعريف list_tools() و call_tool() يدوياً — الديكوريتور @mcp.tool() يكفي
• نظام types تلقائي: FastMCP يستنتج JSON Schema من type hints — path: str = "." يصبح تلقائياً معلمة اختيارية بنوع string
• دعم async/await أصلي — كل أداة يمكن أن تكون async
• تشغيل بثلاثة أسطر: mcp.run(transport='stdio')
• يدعم طبقات نقل متعددة: stdio, SSE, WebSocket, HTTP
• إدارة الأخطاء التلقائية — إذا رفعت الأداة استثناء، FastMCP يحوله لرسالة خطأ JSON-RPC

الفرق واضح بين FastMCP و Anthropic SDK:

# FastMCP — 15 سطراً
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("my-server")
@mcp.tool()
def hello(name: str) -> str:
    return f"مرحباً {name}"
mcp.run()

# Anthropic SDK — 45+ سطراً
from mcp.server import Server
import mcp.types as types
server = Server("my-server")
@server.list_tools()
async def list_tools(): ...
@server.call_tool()
async def call_tool(name, args): ...
asyncio.run(main(server))

8. المشروع 4: البوابة المتعددة الخوادم (Gateway)

المشروع الرابع هو الأكثر تقدماً: بوابة MCP تدير خوادم متعددة تحت واجهة واحدة. هذا النمط ضروري في الإنتاج عندما تحتاج إلى 5 أو 10 أو 20 خادم MCP يعملون معاً.

رابط المستودع: github.com/moaaz01/mcp-multi-server-gateway

الكود الكامل (gateway.py)

import asyncio
import json
import subprocess
from typing import Any
from mcp.server.fastmcp import FastMCP

BACKEND_SERVERS = {
    "weather": {
        "command": "python3",
        "args": ["/path/to/weather_server.py"],
        "description": "خادم الطقس المحلي",
        "tools": ["get_weather", "compare_cities", "forecast_updates"],
    },
    "sqlite": {
        "command": "python3",
        "args": ["/path/to/sqlite_server.py"],
        "description": "قاعدة بيانات SQLite",
        "tools": ["query", "summarize_table", "run_analysis"],
    },
    "filesystem": {
        "command": "python3",
        "args": ["/path/to/filesystem_server.py"],
        "description": "نظام الملفات المحلي",
        "tools": ["list_directory", "read_file", "search_files",
                  "get_file_info", "disk_usage"],
    },
}

class BackendConnection:
    """إدارة اتصال واحد مع خادم خلفي عبر stdio"""

    def __init__(self, name: str, config: dict):
        self.name = name
        self.config = config
        self.process: subprocess.Popen | None = None
        self.available_tools: list[dict] = []

    async def start(self):
        self.process = subprocess.Popen(
            [self.config["command"]] + self.config["args"],
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )
        # المصافحة الأولية
        await self._send_request("initialize", {
            "protocolVersion": "2025-03-26",
            "capabilities": {},
            "clientInfo": {"name": "mcp-gateway", "version": "1.0.0"},
        })
        # جلب الأدوات المتاحة
        result = await self._send_request("tools/list", {})
        if result and "tools" in result:
            self.available_tools = result["tools"]

    async def _send_request(self, method: str, params: dict) -> dict | None:
        if not self.process:
            return None
        req = json.dumps({
            "jsonrpc": "2.0",
            "method": method,
            "params": params,
            "id": f"{self.name}-{id(self)}",
        })
        self.process.stdin.write((req + "\n").encode())
        self.process.stdin.flush()
        line = await asyncio.get_event_loop().run_in_executor(
            None, self.process.stdout.readline
        )
        if line:
            return json.loads(line).get("result")
        return None

    async def call_tool(self, name: str, arguments: dict) -> str:
        result = await self._send_request("tools/call", {
            "name": name, "arguments": arguments
        })
        if result and "content" in result:
            return "\n".join(
                c.get("text", "") for c in result["content"] if c.get("type") == "text"
            )
        return f"⚠️ خطأ في {self.name}: {result}"

    async def stop(self):
        if self.process:
            self.process.terminate()
            await asyncio.sleep(0.5)
            self.process.kill()

connections: dict[str, BackendConnection] = {}

async def get_connection(name: str) -> BackendConnection | None:
    if name not in connections:
        if name not in BACKEND_SERVERS:
            return None
        conn = BackendConnection(name, BACKEND_SERVERS[name])
        await conn.start()
        connections[name] = conn
    return connections[name]

mcp = FastMCP("MCP Gateway", description="بوابة موحّدة لإدارة خوادم MCP متعددة", version="2.0.0")

@mcp.tool(description="استدعاء أداة من أحد الخوادم الخلفية")
async def call_tool_on_backend(tool_name: str, arguments: dict = {}) -> str:
    """تنفيذ أداة على الخادم الخلفي المناسب. يوجّه الطلب للخادم الصحيح بناءً على اسم الأداة."""
    for backend_name, config in BACKEND_SERVERS.items():
        if tool_name in config["tools"]:
            conn = await get_connection(backend_name)
            if conn:
                return await conn.call_tool(tool_name, arguments)
            return f"❌ خادم {backend_name} غير متاح"
    return f"⚠️ أداة غير معروفة: {tool_name}"

@mcp.tool(description="الحصول على حالة البوابة والخوادم المتصلة")
async def gateway_status() -> str:
    """عرض حالة جميع الخوادم الخلفية — متصل/منفصل وعدد الأدوات."""
    report = {"servers": {}, "total_tools": 0}
    for name, config in BACKEND_SERVERS.items():
        conn = connections.get(name)
        status = "connected" if conn and conn.process else "disconnected"
        tools = len(conn.available_tools) if conn else len(config["tools"])
        report["servers"][name] = {"status": status, "description": config["description"], "tools_count": tools}
        report["total_tools"] += tools
    return json.dumps(report, ensure_ascii=False, indent=2)

# 🚀 التشغيل مع إيقاف آمن للخوادم
if __name__ == "__main__":
    try:
        mcp.run(transport="stdio")
    finally:
        for conn in connections.values():
            asyncio.run(conn.stop())

هيكل التكوين (gateway-config.json)

{
  "gateway": {
    "port": 8080,
    "transport": "sse",
    "max_connections": 10
  },
  "backends": {
    "weather": {
      "command": "python3",
      "args": ["weather_server.py"],
      "auto_start": true
    },
    "sqlite": {
      "command": "python3", 
      "args": ["sqlite_server.py"],
      "auto_start": true
    },
    "filesystem": {
      "command": "python3",
      "args": ["filesystem_server.py"],
      "auto_start": false
    }
  }
}

كيف تعمل البوابة؟

البوابة (Gateway) تعمل كطبقة وسيطة بين العميل والخوادم الخلفية. عندما يسأل العميل tools/list، البوابة تجمع الأدوات من جميع الخوادم الخلفية وتعيدها كقائمة واحدة. عندما يستدعي العميل أداة معينة، البوابة توجّه الطلب للخادم الصحيح. هذا يوفّر:

• تجريد (Abstraction): العميل يرى بواباً واحداً بدلاً 10 خوادم منفصلة
• إدارة دورة الحياة: البوابة تبدأ وتوقف الخوادم الخلفية تلقائياً
• التسامح مع الأخطاء: إذا فشل خادم واحد، البقية تعمل
• التوسع: أضف خوادم جديدة بتعديل ملف JSON

هذا المشروع هو الأقرب للإنتاج الفعلي. في أي نظام وكيل متعدد الأدوات، تحتاج إلى gateway لإدارة اتصالات متعددة مع خوادم مختلفة.

9. كيف تبني MCP Client مخصصاً — من الصفر

حتى الآن بنينا خوادم MCP. لكن ماذا لو أردت بناء عميل MCP مخصص يتصل بخادم موجود ويتحكم فيه؟ هذا القسم يعلّمك بناء عميل MCP من الصفر باستخدام Python، بدون أي SDK — فقط JSON-RPC عبر stdio.

العميل الأساسي (مشترك مع أي خادم MCP)

import asyncio
import json
import subprocess
from typing import Any

class MCPClient:
    """عميل MCP بسيط يتصل بأي خادم عبر stdio"""

    def __init__(self, command: str, args: list[str]):
        self.command = command
        self.args = args
        self.process: subprocess.Popen | None = None
        self.request_id = 0

    async def connect(self):
        """تشغيل الخادم وبدء المصافحة"""
        self.process = subprocess.Popen(
            [self.command] + self.args,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
        )
        # المصافحة — المرحلة 1: initialize
        result = await self._send("initialize", {
            "protocolVersion": "2025-03-26",
            "capabilities": {"tools": {}},
            "clientInfo": {"name": "mcp-custom-client", "version": "1.0.0"},
        })
        print(f"✅ متصل: {result['serverInfo']['name']} v{result['serverInfo']['version']}")

        # المصافحة — المرحلة 2: initialized notification
        await self._send_notification("notifications/initialized")

    async def _send(self, method: str, params: dict) -> dict:
        """إرسال طلب JSON-RPC وانتظار الرد"""
        self.request_id += 1
        req = json.dumps({
            "jsonrpc": "2.0",
            "method": method,
            "params": params,
            "id": str(self.request_id),
        })
        self.process.stdin.write((req + "\n").encode())
        self.process.stdin.flush()
        line = await asyncio.get_event_loop().run_in_executor(
            None, self.process.stdout.readline
        )
        resp = json.loads(line)
        if "error" in resp:
            raise Exception(f"خطأ JSON-RPC: {resp['error']}")
        return resp.get("result", {})

    async def _send_notification(self, method: str):
        """إرسال إشعار (لا ينتظر رد)"""
        req = json.dumps({
            "jsonrpc": "2.0",
            "method": method,
            "params": {},
        })
        self.process.stdin.write((req + "\n").encode())
        self.process.stdin.flush()

    async def list_tools(self) -> list[dict]:
        """الحصول على قائمة الأدوات المتوفرة"""
        result = await self._send("tools/list", {})
        return result.get("tools", [])

    async def call_tool(self, name: str, arguments: dict) -> Any:
        """استدعاء أداة محددة"""
        result = await self._send("tools/call", {
            "name": name,
            "arguments": arguments,
        })
        return result.get("content", [])

    async def close(self):
        if self.process:
            self.process.terminate()
            try:
                await asyncio.wait_for(
                    asyncio.get_event_loop().run_in_executor(None, self.process.wait),
                    timeout=5.0,
                )
            except asyncio.TimeoutError:
                self.process.kill()

# مثال استخدام: الاتصال بخادم الطقس الذي بنيناه
async def main():
    client = MCPClient("python3", ["weather_server.py"])
    await client.connect()

    # اكتشف الأدوات المتوفرة
    tools = await client.list_tools()
    print(f"\n🛠️ الأدوات المتوفرة ({len(tools)}):")
    for t in tools:
        print(f"  - {t['name']}: {t['description']}")

    # استدعِ أداة الطقس
    result = await client.call_tool("get_weather", {"city": "دمشق"})
    for item in result:
        print(f"\n🌤️  النتيجة:\n{item['text']}")

    await client.close()

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

تطبيقات عملية لعميل MCP مخصص

• Chatbot مع أدوات مباشرة: ابنِ واجهة محادثة يستخدم فيها المستخدم أدوات حقيقية (طقس، DB، ملفات)
• نظام مراقبة (Monitoring): عميل يفحص خوادم MCP كل دقيقة ويتحقق من صحتها
• واجهة إدارة مركزية: لوحة تحكم تدير 10+ خوادم MCP من واجهة واحدة
• اختبار تلقائي: عميل يختبر كل أداة في الخادم بعد كل تحديث
• سير عمل متعدد الخطوات: عميل ينفذ تسلسلاً معقداً: اسأل قاعدة البيانات ← حلّل ← احفظ النتيجة

10. نشر MCP مع SSE في الإنتاج

حتى الآن استخدمنا stdio للنقل — وهو مناسب للتطوير المحلي فقط. للإنتاج، تحتاج SSE (Server-Sent Events) لنشر خادم MCP عبر الشبكة. SSE يسمح للعملاء بالاتصال بالخادم عن بُعد عبر HTTP.

تشغيل خادم FastMCP مع SSE

# filesystem_server_sse.py — نفس الكود لكن مع transport=sse
from fastmcp import FastMCP

mcp = FastMCP("File System Server - Production")

@mcp.tool()
def hello(name: str) -> str:
    return f"مرحباً {name}"

# SSE مع uvicorn — يعمل كخادم HTTP حقيقي
if __name__ == "__main__":
    mcp.run(transport="sse")

# تشغيل الخادم (افتراضياً على port 8000)
python3 filesystem_server_sse.py

# يمكنك تحديد port مختلف
# mcp.run(transport="sse", port=8080)

نشر على Render.com (أبسط خيار)

Render.com يوفّر استضافة مجانية لخوادم Python مع دعم Web Service:

# 1. أنشئ ملف requirements.txt
echo "fastmcp>=0.5.0" > requirements.txt

# 2. أنشئ ملف render.yaml
cat > render.yaml << 'EOF'
services:
  - type: web
    name: mcp-filesystem-server
    runtime: python
    buildCommand: pip install -r requirements.txt
    startCommand: python filesystem_server_sse.py
    envVars:
      - key: ALLOWED_DIRS
        value: /home/user,/tmp
EOF

# 3. ادفع المستودع لـ GitHub واربطه بـ Render.com
# Render سيبني وينشر تلقائياً

نشر على Railway.app (مع HTTPS تلقائي)

# Railway يوفّر HTTPS تلقائي + Custom Domain
# فقط اربط مستودع GitHub ← Railway يكتشف requirements.py وينشر

# يمكنك استخدام nixpacks.toml للتخصيص:
[nixpacks]
builder = "python:3.11"
startCommand = "python filesystem_server_sse.py --port $PORT"

نشر مع Docker

FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["python", "filesystem_server_sse.py"]

الاتصال بالخادم البعيد من OpenAI Agents SDK

from agents.mcp import MCPServerSSE

async with MCPServerSSE(
    params={"url": "https://your-server.onrender.com/sse"}
) as server:
    agent = Agent(
        name="Remote Assistant",
        instructions="أنت وكيل ذكي متصل بأدوات MCP عن بُعد.",
        mcp_servers=[server],
    )
    result = await Runner.run(agent, "سرد محتويات المجلد /home")
    print(result.final_output)

اعتبارات الإنتاج لـ SSE

• HTTPS: تأكد من استخدام HTTPS دائماً — بيانات JSON-RPC قد تحتوي معلومات حساسة
• المصادقة: أضف API Key في رأس الطلب (Header) قبل السماح بالاتصال
• CORS: إذا كان العميل متصفحاً، هيئ CORS headers
• Rate Limiting: استخدم middleware للحد من عدد الطلبات لكل عميل
• Health Check: أضف نقطة نهاية /health لمراقبة الخادم
• Logging: سجّل كل طلب مع IP و timestamp واسم الأداة

11. مقارنة SDKs: متى تستخدم ماذا؟

بعد بناء 4 مشاريع بـ 3 SDKs مختلفة، إليك مقارنة عملية تساعدك على اختيار الأداة المناسبة:

Anthropic MCP SDK (رسمي)

• ✔️ المميزات: التحكم الكامل، يدعم جميع ميزات MCP (Tools, Resources, Prompts, Notifications)
• ✔️ يدعم stdio و SSE
• ✔️ أفضل لـ: خوادم إنتاجية تحتاج تحكماً دقيقاً، تطبيقات مع Claude Desktop
• ❌ العيوب: أكثر تعقيداً، يحتاج كود أكثر (list_tools + call_tool يدوياً)
• ❌ توثيقها يعتمد على TypeScript — أمثلة Python محدودة

FastMCP (مفتوح المصدر)

• ✔️ المميزات: أبسط بكثير — ديكوريتور واحد لكل أداة، type hints تلقائي
• ✔️ يدعم stdio, SSE, WebSocket, HTTP
• ✔️ أفضل لـ: النماذج الأولية السريعة، المشاريع الشخصية، المطورين الجدد في MCP
• ✔️ مجتمع نشط على GitHub (2000+ Star)
• ❌ العيوب: أقل تحكماً في البروتوكول، لا يدعم كل الميزات (بعض الإضافات التجريبية)

OpenAI Agents SDK (عميل MCP)

• ✔️ المميزات: يثبت أن MCP ليس حكراً على Claude، يعمل مع GPT-5
• ✔️ MCPServerStdio و MCPServerSSE للاتصال
• ✔️ أفضل لـ: تطبيقات OpenAI Agents التي تحتاج أدوات خارجية
• ❌ العيوب: فقط عميل (Client) — لا يمكن بناء خوادم MCP به
• ❌ بعض الميزات المتقدمة لا تزال تتطور (مثل Streamable HTTP في OpenAI)

متى تختار ماذا؟ — دليل سريع

• أريد بناء خادم بسرعة ← FastMCP
• أريد نشر خادم مع SSE للإنتاج ← FastMCP (لأنه يدعم 4 طرق نقل)
• أريد أن أتعلم MCP بعمق وأفهم كل شيء ← Anthropic SDK
• أريد استخدام GPT-5 مع أدوات خارجية ← OpenAI Agents SDK (كعميل)
• أريد بناء Gateway تدير خوادم متعددة ← Anthropic SDK (للتحكم الكامل)
• أريد تشغيل خادم في Docker مع uvicorn ← FastMCP (يدعم HTTP)
• مشروع شخصي صغير ← FastMCP
• مشروع مؤسسي كبير ← Anthropic SDK

12. MCP في الإنتاج — أفضل الممارسات

بناء خادم MCP هو البداية فقط. لتشغيله في الإنتاج، اتبع هذه الممارسات:

🔒 الأمان

• التحقق من المدخلات (Input Validation): لا تثق أبداً بالمعاملات القادمة من الـ LLM. تحقق من كل شيء — المسارات، الاستعلامات، الأسماء.
• مبدأ الأقل صلاحية: امنح الخادم فقط الصلاحيات التي يحتاجها. إذا كان خادم SQLite يحتاج قراءة فقط، استخدم اتصالاً للقراءة فقط.
• Rate Limiting: ضع حداً لعدد استدعاءات الأدوات لكل جلسة لمنع الاستغلال.
• سجل التدقيق (Audit Log): سجّل كل استدعاء أداة مع timestamp ومعاملاته.
• منع الحقن (Injection Prevention): استخدم parameterized queries لـ SQL، و path traversal checks للملفات.

⚡ الأداء

• Connection Pooling: لا تفتح اتصالاً جديداً لكل طلب. استخدم pool من الاتصالات (مثلاً: from sqlite3 import connect; conn = connect(DB, check_same_thread=False)).
• Caching: خزّن نتائج الاستعلامات المتكررة (مثلاً: بيانات الطقس المخزنة مؤقتاً لمدة 5 دقائق باستخدام functools.lru_cache).
• Timeout: ضع timeout لكل استعلام (مثلاً: 30 ثانية كحد أقصى).
• Profiling: استخدم cProfile أو py-spy لقياس أداء خوادمك.
• Lazy Loading: لا تحمّل كل شيء عند بدء التشغيل — استخدم Lazy connections للخوادم التي قد لا تُستخدم.

🔄 دورة الحياة (Lifecycle)

• Health Checks: أضف نقطة نهاية health ترجع حالة الخادم ووقت التشغيل.
• Graceful Shutdown: تأكد من أن الخادم يغلق الاتصالات بشكل نظيف عند الـ shutdown.
• Logging: استخدم structlog أو loguru بدلاً من print().
• Monitoring: أضف Prometheus metrics لعدد الاستدعاءات، وقت الاستجابة، ومعدل الأخطاء.
• Auto-restart: استخدم systemd أو supervisor لضمان بقاء الخادم قيد التشغيل.

🧪 الاختبار

• MCP Inspector: أداة رسمية من Anthropic لاختبار الخوادم — npx @aaif/mcp-inspector python3 server.py
• Unit Tests: اختبر كل أداة بشكل منفصل باستخدام pytest
• Integration Tests: اختبر تدفق كامل: قم بتشغيل الخادم، اتصل به كعميل، استدعِ الأدوات، تحقق من النتائج.
• Edge Cases: اختبر بمدخلات فارغة، قيم null، أحرف خاصة، وأحجام كبيرة.
• Stress Tests: اختبر الخادم تحت ضغط 100+ طلب متزامن لقياس الثبات.

13. حل المشكلات الشائعة (Troubleshooting)

أثناء بناء خوادم MCP، ستواجه مشكلات. إليك أكثرها شيوعاً وحلولها المختبرة:

المشكلة: الخادم لا يستجيب (Connection refused)

السبب: الخادم لم يبدأ بشكل صحيح أو انتهى قبل المصافحة.

الحل:

• تحقق من أن الأمر المستخدم لبدء الخادم صحيح
• أضف طباعة (print("Server started", file=sys.stderr)) لترى إن كان الخادم بدأ
• استخدم MCP Inspector لاختبار الاتصال أولاً
• تأكد من أن sys.stdin غير محجوب بشيء آخر

المشكلة: Tool not found

السبب: اسم الأداة غير مطابق بين list_tools() و call_tool().

الحل:

• استخدم ثابتاً (constant) لاسم الأداة بدلاً من كتابته مرتين:

TOOL_GET_WEATHER = "get_weather"

@server.list_tools()
async def list_tools():
    return [types.Tool(name=TOOL_GET_WEATHER, ...)]

@server.call_tool()
async def call_tool(name, args):
    if name == TOOL_GET_WEATHER: ...

المشكلة: JSON Schema خطأ (Invalid inputSchema)

السبب: خطأ في هيكل JSON Schema (مثلاً استخدام enum بدون type: string).

الحل:

• تحقق من صحة Schema عبر JSON Schema Validator
• تأكد من أن required يحتوي على أسماء الحقول المطلوبة فقط
• استخدم pydantic لإنشاء الـ Schema تلقائياً مع FastMCP

المشكلة: stdio transport يتجمد (Freeze)

السبب: الخادم يستخدم print() أو logging يكتب على stdout ويتعارض مع JSON-RPC.

الحل:

• استخدم stderr لكل طباعة تصحيح: print("Debug", file=sys.stderr)
• استخدم logging مع output مخصص لـ stderr
• لا تخلط أبداً stdout بين تنسيق JSON و plain text

المشكلة: خادم FastMCP لا يعمل مع Claude Desktop

السبب: Claude Desktop يتوقع نقل stdio بأنماط معينة.

الحل:

• تأكد من استخدام mcp.run() بدون تحديد transport (يختار تلقائياً)
• في claude_desktop_config.json، استخدم uv بدلاً من python3:

{
  "mcpServers": {
    "filesystem": {
      "command": "uv",
      "args": ["run", "--with", "fastmcp", "python3", "filesystem_server.py"]
    }
  }
}

المشكلة: TypeError: Object of type X is not JSON serializable

السبب: أنت تحاول إرجاع كائن Python غير مدعوم في JSON (مثل datetime, Decimal).

الحل:

• حوّل البيانات إلى أنواع JSON قياسية قبل الإرجاع:

def json_serializable(obj):
    if isinstance(obj, datetime):
        return obj.isoformat()
    if isinstance(obj, Decimal):
        return float(obj)
    if isinstance(obj, set):
        return list(obj)
    return str(obj)

المشكلة: الخادم يعلق بعد عدة طلبات (Memory Leak)

السبب: عدم إغلاق اتصالات قاعدة البيانات أو الملفات بعد كل طلب.

الحل:

• استخدم try/finally أو context managers with لكل مورد
• تأكد من إغلاق كل conn في النهاية
• راقب استخدام الذاكرة بـ psutil أو tracemalloc

14. القيود والتحديات الحالية

رغم قوة MCP، هناك تحديات يجب معرفتها:

• تعدد نسخ البروتوكول: الإصدارات المتاحة: 2025-03-26 (مستقر) و 2025-06-18 (يضيف Streamable HTTP + OAuth 2.1) و 2025-07-28 (قيد الإصدار التجريبي). يجب اختيار النسخة المناسبة عند بناء الخادم.
• نقص التوثيق العربي: لا توجد مصادر عربية كافية لـ MCP، معظم التوثيق بالإنكليزية.
• قيود النقل عبر stdio: stdio مناسب للتطوير المحلي لكنه غير قابل للتوسع. SSE يتطلب إعدادات إضافية (CORS, Proxy, Auth).
• الأمان: MCP لا يحدد طبقة مصادقة أو تفويض موحدة. كل خادم يطبقها بنفسه.
• حجم الحمولة: كل رسالة JSON-RPC قد تصبح كبيرة مع البيانات المعقدة (صور، جداول كبيرة).
• التوافق عبر المنصات: ليس كل LLM يدعم MCP بالكامل بعد. Claude يدعمه أصلياً، OpenAI رسمياً، والباقي عبر أطر خارجية.
• لغة Python: أمثلة Python الرسمية محدودة. معظم الأمثلة والتوثيق يركز على TypeScript/Node.js.
• إدارة الأخطاء: البروتوكول لا يوفر طريقة موحدة للإبلاغ عن الأخطاء — كل خادم يعيد خطأ JSON-RPC عادياً.
• التوافق مع الجوال: MCP صُمم أساساً لسطح المكتب والخوادم. تطبيقات الجوال ما زالت بحاجة لمكتبات خاصة.

15. مستقبل MCP والتوجهات

في يونيو 2026، MCP في نقطة تحول. إليك التوجهات الرئيسية:

• MCP أصبح معياراً مفتوحاً: تبرعت به Anthropic لمؤسسة AAIF/Linux Foundation في ديسمبر 2025 — الآن أي شركة تساهم في تطويره.
• Streamable HTTP + OAuth 2.1: الإصدار 2025-06-18 يضيف اتصال HTTP مستمر (بدون SSE) مع مصادقة OAuth مدمجة — هذا يسهل نشر خوادم MCP للمستخدمين النهائيين.
• اعتماد متزايد من OpenAI: إضافة دعم MCP في OpenAI Agents SDK (مارس 2025) كانت نقطة تحول — الآن ليس فقط Claude يستخدم MCP.
• Google Gemini: يخطط لدعم MCP رسمياً في Gen AI SDK بحلول نهاية 2026.
• A2A من Google: بروتوكول Agent-to-Agent (A2A) أطلقته Google في أبريل 2026 — يكمّل MCP بدلاً من التنافس معه. MCP للأدوات، A2A للوكلاء.
• سوق MCP: يتوقع ظهور أسواق (Marketplaces) لخوادم MCP مثل متجر تطبيقات الجوال — اكتب خادماً مرة وانشره للجميع.
• Proxies و Gateways: شركات مثل Portkey تطلق MCP Proxies لإدارة المصادقة، Rate Limiting، والتخزين المؤقت مركزيًا.
• MCP للتطبيقات الجوالة: دعم MCP على Android و iOS عبر SSE/WebSocket.
• MCP + WebSocket: إضافة WebSocket كطبقة نقل ثنائية الاتجاه بديلة عن SSE (أسرع، أقل latency).
• Streaming Tools: القدرة على إرجاع نتائج الأدوات كـ stream بدلاً من دفعة واحدة — مفيد للبيانات الكبيرة.
• Federated MCP: خوادم MCP تتصل بخوادم MCP أخرى — شبكة أدوات لا نهاية لها.
• MCP في التعليم: استخدام MCP لربط نماذج AI بمنصات التعليم (مثل Blackboard, Moodle) لإنشاء تجارب تعلم مخصصة.

16. خاتمة

MCP ليس مجرد بروتوكول تقني — إنه نقلة نوعية في طريقة تواصل الذكاء الاصطناعي مع العالم. عندما تبني خادم MCP اليوم، فأنت تستثمر في معيار سيفهمه أي وكيل ذكاء اصطناعي في المستقبل، سواء كان Claude أو GPT-5.6 أو Gemini أو أي نموذج لم يُولد بعد.

في هذا الدليل، بنينا 5 مشاريع عملية تغطي:

1. خادم الطقس — أساسيات الأدوات والموارد
2. خادم SQLite — التكامل مع قواعد البيانات والتحليلات
3. خادم نظام الملفات — FastMCP للسرعة والسهولة
4. البوابة المتعددة الخوادم — إدارة خوادم متعددة في الإنتاج
5. عميل MCP مخصص — التحكم الكامل من أي تطبيق Python

كل مشروع موجود على GitHub مع كود كامل وتوثيق:

• 🔗 mcp-weather-server — Weather Server (Anthropic SDK)
• 🔗 mcp-sqlite-server — SQLite Server (OpenAI Agents)
• 🔗 mcp-file-system-server — File System Server (FastMCP)
• 🔗 mcp-multi-server-gateway — Multi-Server Gateway

الخطوات التالية بعد هذا الدليل

• جرّب تعديل خادم الطقس لإضافة API حقيقي (مثلاً OpenWeatherMap API بدلاً من البيانات المحاكاة)
• وسّع خادم SQLite ليدعم كتابة البيانات بعد الموافقة (سير عمل Read ثم Write بعد التأكيد)
• أضف مصادقة (API Key أو JWT) للـ Gateway قبل توجيه الطلبات للخوادم الخلفية
• انشر خادم FastMCP مع SSE عبر Render.com أو Railway ليكون متاحاً عبر الإنترنت
• ابنِ Custom Client لـ MCP باستخدام Streamlit لتجربة تفاعلية
• حاول دمج MCP مع A2A من Google — استخدم MCP للأدوات و A2A للوكلاء المتعددين
• انضم لمجتمع MCP على GitHub و Discord — ساهم بتحسين التوثيق العربي

تذكّر: MCP ما زال جديداً. كل خادم تكتبه اليوم يسبق به السوق، ويبني خبرتك في معيار سيصبح أساسياً في السنوات القادمة. العالم يحتاج محتوى عربي متعمق في هذا المجال، وأنت الآن تملك الأدوات والمعرفة لتقديمه.

❓ الأسئلة الشائعة

هل MCP يعمل مع GPT-5 أم فقط Claude؟

MCP ليس حكراً على Claude. OpenAI أضافت دعم MCP في OpenAI Agents SDK منذ مارس 2025. يمكنك استخدام نفس خادم MCP مع GPT-5 عبر MCPServerStdio. أيضاً، Google تخطط لدعم MCP رسمياً قريباً.

ما الفرق بين MCP و A2A من Google؟

A2A (Agent-to-Agent) بروتوكول من Google لتواصل الوكلاء مع بعضهم، بينما MCP لتواصل الوكلاء مع الأدوات. الاثنان مكملان — يمكن استخدام MCP لجلب البيانات من قاعدة بيانات و A2A لتنسيق وكلاء متعددين عبر الخوادم.

هل أحتاج FastAPI أو Flask لبناء خادم MCP؟

لا. MCP يتواصل عبر stdin/stdout (محلياً) أو SSE (عن بعد). لا تحتاج لخادم HTTP تقليدي. لكن يمكنك استخدام uvicorn مع FastMCP إذا أردت تشغيل الخادم عبر HTTP.

هل MCP آمن؟

الأمان يعتمد على تنفيذك. MCP لا يحدد طبقة مصادقة موحدة — أنت مسؤول عن التحقق من المدخلات، تحديد الصلاحيات، وتسجيل النشاط. اتبع مبدأ الأقل صلاحية وتحقق من كل معامل.

ما أفضل SDK لبدء التعلم؟

ابدأ بـ FastMCP للسرعة والسهولة. ثم انتقل لـ Anthropic MCP SDK عندما تحتاج تحكماً أكبر. استخدم OpenAI Agents SDK إذا كنت تعمل ضمن نظام OpenAI البيئي.

كم تكلفة تشغيل خادم MCP؟

خادم MCP نفسه مجاني تماماً (مفتوح المصدر). التكاليف تأتي من: استضافة الخادم إذا استخدمت SSE (حوالي $5-20/شهر على VPS صغير)، ومفاتيح API للخدمات الخارجية (مثل OpenWeatherMap API).

هل يمكن ربط خادم MCP بخدمة REST API خارجية؟

نعم، هذا أحد الاستخدامات الرئيسية. خادم MCP يمكنه استدعاء أي REST API داخلياً وتقديم البيانات كأدوات للـ LLM. هذا مفيد عندما تريد جعل API معقد متاحاً للذكاء الاصطناعي بطريقة منظمة.

ما اللغات التي تدعم بناء خوادم MCP؟

اللغات الأساسية: Python (SDK رسمي) و TypeScript/JavaScript (SDK رسمي + مجتمع). هناك أيضاً مكتبات مجتمعية لـ Go, Rust, Kotlin, Java, C#.

هل يمكن استخدام MCP في تطبيقات الجوال؟

حالياً، MCP صُمم لتطبيقات سطح المكتب والخادم. لكن مع دعم SSE و WebSocket، يمكن لتطبيقات الجوال الاتصال بخوادم MCP البعيدة. يتوقع ظهور مكتبات رسمية للجوال في 2026-2027.

كيف أختبر خادم MCP بدون Claude Desktop؟

استخدم MCP Inspector: npx @aaif/mcp-inspector python3 server.py. هذه الأداة توفّر واجهة ويب تفاعلية لاختبار كل أداة يدوياً ورؤية الطلبات والاستجابات. بديل آخر: ابنِ عميل Python بسيط كما في القسم 9 من هذا الدليل.

ماذا لو أردت إضافة مصادقة لخادم MCP؟

أضف طبقة مصادقة قبل تشغيل الخادم. مثال: استخدم fastapi كـ wrapper مع Bearer token، ثم وجّه الطلبات لخادم MCP الداخلي. أو أضف Authorization: Bearer <api_key> كمعامل إضافي في دالة المصادحة الأولية.

تم كتابة هذا المقال بواسطة علي الرزق — متخصص في الذكاء الاصطناعي التطبيقي ووكلاء AI
جميع الأكواد متاحة على GitHub — يمكنك نسخها وتعديلها واستخدامها في مشاريعك