مشاركة عبر


التعليم: نشر خادم MCP بلغة بايثون على Azure Container Apps

في هذا الدرس، تبني خادم بروتوكول السياق النموذجي (MCP) الذي يعرض أدوات إدارة المهام باستخدام FastAPI وحزمة تطوير البرمجيات MCP Python. تقوم بنشر الخادم إلى Azure Container Apps وتتصل به من GitHub Copilot Chat في VS Code.

في هذا البرنامج التعليمي، سوف تتعلّم:

  • أنشئ تطبيق FastAPI يعرض أدوات MCP
  • اختبار خادم MCP محليا باستخدام GitHub Copilot
  • Containerize ونشر التطبيق في Azure Container Apps
  • توصيل GitHub Copilot بخادم MCP المنشور

المتطلبات المسبقه

أنشئ منصة التطبيق

في هذا القسم، تنشئ مشروع بايثون جديد باستخدام FastAPI وحزمة تطوير البرمجيات MCP بايثون.

  1. أنشئ دليل المشروع وقم بإعداد بيئة افتراضية:

    mkdir tasks-mcp-server && cd tasks-mcp-server
    python -m venv .venv
    source .venv/bin/activate
    
  2. إنشاء requirements.txt:

    fastapi>=0.115.0
    uvicorn>=0.30.0
    mcp[cli]>=1.2.0
    
  3. تثبيت التبعيات:

    pip install -r requirements.txt
    
  4. إنشاء task_store.py لمخزن البيانات في الذاكرة:

    from dataclasses import dataclass, field
    from datetime import datetime, timezone
    
    
    @dataclass
    class TaskItem:
        id: int
        title: str
        description: str
        is_complete: bool = False
        created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    
        def to_dict(self) -> dict:
            return {
                "id": self.id,
                "title": self.title,
                "description": self.description,
                "is_complete": self.is_complete,
                "created_at": self.created_at.isoformat(),
            }
    
    
    class TaskStore:
        def __init__(self):
            self._tasks: list[TaskItem] = [
                TaskItem(1, "Buy groceries", "Milk, eggs, bread"),
                TaskItem(2, "Write docs", "Draft the MCP tutorial", True),
            ]
            self._next_id = 3
    
        def get_all(self) -> list[dict]:
            return [t.to_dict() for t in self._tasks]
    
        def get_by_id(self, task_id: int) -> dict | None:
            task = next((t for t in self._tasks if t.id == task_id), None)
            return task.to_dict() if task else None
    
        def create(self, title: str, description: str) -> dict:
            task = TaskItem(self._next_id, title, description)
            self._next_id += 1
            self._tasks.append(task)
            return task.to_dict()
    
        def toggle_complete(self, task_id: int) -> dict | None:
            task = next((t for t in self._tasks if t.id == task_id), None)
            if task is None:
                return None
            task.is_complete = not task.is_complete
            return task.to_dict()
    
        def delete(self, task_id: int) -> bool:
            task = next((t for t in self._tasks if t.id == task_id), None)
            if task is None:
                return False
            self._tasks.remove(task)
            return True
    
    
    # For demonstration only — not thread-safe.
    store = TaskStore()
    

    TaskItem تعرف فئة البيانات نموذج البيانات بطريقة to_dict() للتسلسل. تدير الفئة TaskStore قائمة في الذاكرة مملوءة مسبقا ببيانات عينة وتوفر طرق CRUD. يتم مشاركة الوحدة المفردة على مستوى الوحدة store عبر التطبيق من أجل البساطة.

تعريف أدوات MCP

في هذا القسم، تحدد أدوات MCP التي يمكن لنموذج الذكاء الاصطناعي استدعاؤها وتركيبها على خادم MCP في تطبيق FastAPI الخاص بك.

  1. إنشاء mcp_server.py:

    from mcp.server.fastmcp import FastMCP
    from task_store import store
    
    mcp = FastMCP("TasksMCP", stateless_http=True)
    
    
    @mcp.tool()
    async def list_tasks() -> list[dict]:
        """List all tasks with their ID, title, description, and completion status."""
        return store.get_all()
    
    
    @mcp.tool()
    async def get_task(task_id: int) -> dict | None:
        """Get a single task by its numeric ID.
    
        Args:
            task_id: The numeric ID of the task to retrieve.
        """
        return store.get_by_id(task_id)
    
    
    @mcp.tool()
    async def create_task(title: str, description: str) -> dict:
        """Create a new task with the given title and description. Returns the created task.
    
        Args:
            title: A short title for the task.
            description: A detailed description of what the task involves.
        """
        return store.create(title, description)
    
    
    @mcp.tool()
    async def toggle_task_complete(task_id: int) -> str:
        """Toggle a task's completion status between complete and incomplete.
    
        Args:
            task_id: The numeric ID of the task to toggle.
        """
        task = store.toggle_complete(task_id)
        if task:
            status = "complete" if task["is_complete"] else "incomplete"
            return f"Task {task['id']} is now {status}."
        return f"Task with ID {task_id} not found."
    
    
    @mcp.tool()
    async def delete_task(task_id: int) -> str:
        """Delete a task by its numeric ID.
    
        Args:
            task_id: The numeric ID of the task to delete.
        """
        if store.delete(task_id):
            return f"Task {task_id} deleted."
        return f"Task with ID {task_id} not found."
    

    النقاط الرئيسية:

    • FastMCP("TasksMCP", stateless_http=True) ينشئ خادم MCP باستخدام نمط HTTP بدون حالة في حزمة تطوير Python. نقطة نهاية HTTP القابلة للبث تكون افتراضيا على المسار /mcp الفرعي.
    • تصبح كل @mcp.tool() وظيفة أداة قابلة للاستدعاء. تساعد تعليقات الوثائق والمعلمات الخاصة بالوظيفة نموذج الذكاء الاصطناعي على فهم كيفية استخدام كل أداة.
  2. إنشاء app.py. يحدد هذا الملف تطبيق FastAPI الذي يثبت خادم MCP:

    from contextlib import AsyncExitStack, asynccontextmanager
    
    from fastapi import FastAPI
    from fastapi.responses import JSONResponse
    
    from mcp_server import mcp
    
    
    @asynccontextmanager
    async def lifespan(app: FastAPI):
        async with AsyncExitStack() as stack:
            await stack.enter_async_context(mcp.session_manager.run())
            yield
    
    
    app = FastAPI(lifespan=lifespan)
    app.mount("/", mcp.streamable_http_app())
    
    
    @app.get("/health")
    async def health():
        return JSONResponse({"status": "healthy"})
    

    تطبيق خادم MCP يثبت في الجذر (/). نقطة نهاية HTTP القابلة للبث في مجموعة تطوير البرمجيات تكون افتراضيا على /mcp، لذا فإن مسار نقطة النهاية الكامل هو /mcp.

    يتم استخدام نقطة نهاية منفصلة /health لاختبارات صحة تطبيقات الحاويات. نقاط نهاية MCP تتوقع طلبات JSON-RPC POST وليست مناسبة للفحوصات الصحية.

اختبار خادم MCP محليا

قبل النشر على Azure، تحقق من عمل خادم MCP عن طريق تشغيله محليا والاتصال من GitHub Copilot.

  1. ابدأ التطبيق:

    uvicorn app:app --reload --port 8080
    
  2. افتح VS Code، ثم افتح دردشة Copilot واختر وضع الوكيل .

  3. اختر زر الأدوات ، ثم اختر إضافة المزيد من الأدوات...>أضف خادم MCP.

  4. حدد HTTP (HTTP أو أحداث Server-Sent).

  5. أدخل رابط الخادم: http://localhost:8080/mcp

  6. أدخل معرف الخادم: tasks-mcp

  7. حدد إعدادات مساحة العمل.

  8. في موجه جديد لدردشة Copilot، اكتب: "أرني جميع المهام"

  9. اختر الاستمرار عندما يطلب Copilot تأكيد أداة MCP.

يجب أن ترى قائمة المهام التي تم إرجاعها من مخزن الذاكرة الخاص بك.

نصيحة

جرب تعليمات أخرى مثل "إنشاء مهمة لمراجعة الPR"، "وسم المهمة 1 كمكتملة"، أو "حذف المهمة 2".

ضع التطبيق في حاوية

قم بتغليف التطبيق كحاوية Docker حتى تتمكن من اختباره محليا قبل نشره على Azure.

  1. Dockerfileإنشاء :

    FROM python:3.12-slim
    
    WORKDIR /app
    
    COPY requirements.txt .
    RUN pip install --no-cache-dir -r requirements.txt
    
    COPY . .
    
    EXPOSE 8080
    CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8080"]
    

    يستخدم ملف Dockerfile صورة قاعدة نحيفة بإصدار بايثون 3.12، ويثبت التبعيات من requirements.txt، ثم ينسخ كود التطبيق. يخدم Uvicorn تطبيق FastAPI على المنفذ 8080.

  2. تحقق من بناء الحاوية وتشغيلها محليا:

    docker build -t tasks-mcp-server .
    docker run -p 8080:8080 tasks-mcp-server
    

    تأكد من أن نقطة النهاية الصحية تستجيب: curl http://localhost:8080/health

التوزيع إلى Azure Container Apps

بعد أن تقوم بتحويل التطبيق إلى حاوية، قم بنشره في Azure Container Apps باستخدام Azure CLI. تقوم هذه الأوامر az containerapp up ببناء صورة الحاوية في السحابة، لذا لا تحتاج إلى Docker على جهازك لهذه الخطوة.

  1. حدد متغيرات البيئة:

    RESOURCE_GROUP="mcp-tutorial-rg"
    LOCATION="eastus"
    ENVIRONMENT_NAME="mcp-env"
    APP_NAME="tasks-mcp-server-py"
    
  2. إنشاء مجموعة موارد وبيئة تطبيقات الحاويات:

    az group create --name $RESOURCE_GROUP --location $LOCATION
    
    az containerapp env create \
        --name $ENVIRONMENT_NAME \
        --resource-group $RESOURCE_GROUP \
        --location $LOCATION
    
  3. نشر تطبيق الحاوية:

    az containerapp up \
        --name $APP_NAME \
        --resource-group $RESOURCE_GROUP \
        --environment $ENVIRONMENT_NAME \
        --source . \
        --ingress external \
        --target-port 8080
    
  4. تكوين CORS:

    az containerapp ingress cors enable \
        --name $APP_NAME \
        --resource-group $RESOURCE_GROUP \
        --allowed-origins "*" \
        --allowed-methods "GET,POST,DELETE,OPTIONS" \
        --allowed-headers "*"
    

    ‏‫ملاحظة‬

    للإنتاج، استبدل الأصول الوايلدكارد بأصول موثوقة محددة. انظر خوادم MCP الآمنة على تطبيقات الحاويات.

  5. تحقق من النشر:

    APP_URL=$(az containerapp show \
        --name $APP_NAME \
        --resource-group $RESOURCE_GROUP \
        --query "properties.configuration.ingress.fqdn" -o tsv)
    
    curl https://$APP_URL/health
    

ربط GitHub Copilot بالخادم المنشور

الآن بعد أن يعمل خادم MCP في Azure، قم بتكوين VS Code لربط GitHub Copilot بنقطة النهاية المنشورة.

  1. إنشاء أو تحديث .vscode/mcp.json:

    {
        "servers": {
            "tasks-mcp-server": {
                "type": "http",
                "url": "https://<your-app-fqdn>/mcp"
            }
        }
    }
    

    استبدلها <your-app-fqdn> ب FQDN من مخرجات النشر.

  2. في VS Code، افتح دردشة Copilot في وضع الوكيل.

  3. يظهر التحقق tasks-mcp-server في قائمة الأدوات. اختر Start إذا لزم الأمر.

  4. اختبر باستخدام تنبيه مثل "إنشاء مهمة لنشر بيئة الترحيل".

تكوين التوسع للاستخدام التفاعلي

بشكل افتراضي، يمكن لتطبيقات حاويات Azure أن تتوسع إلى صفر نسخ. بالنسبة لخوادم MCP التي تخدم عملاء تفاعليين مثل Copilot، يسبب التشغيل البارد تأخيرات ملحوظة. حدد الحد الأدنى لعدد النسخ للحفاظ على تشغيل نسخة واحدة على الأقل:

az containerapp update \
    --name $APP_NAME \
    --resource-group $RESOURCE_GROUP \
    --min-replicas 1

اعتبارات الأمان

يستخدم هذا الدرس خادم MCP غير مصادق للبساطة. قبل تشغيل خادم MCP في الإنتاج، راجع التوصيات التالية. عندما يتصل وكيل مدعوم بنماذج لغوية كبيرة (LLMs) بخادم MCP الخاص بك، كن على علم بهجمات حقن الطلبات .

  • المصادقة والتفويض: قم بتأمين خادم MCP الخاص بك باستخدام Microsoft Entra ID. انظر خوادم MCP الآمنة على تطبيقات الحاويات.
  • التحقق من صحة المدخلات: تحقق دائما من معايير الأداة. استخدم Pydantic لفرض التحقق من صحة البيانات على مدخلات الأداة.
  • HTTPS: Azure Container Apps يفرض HTTPS بشكل افتراضي مع شهادات TLS تلقائية.
  • أقل امتياز: اعرض فقط الأدوات التي تتطلبها حالتك. تجنب الأدوات التي تقوم بعمليات مدمرة دون تأكيد.
  • CORS: تقييد المصادر المسموح بها على النطاقات الموثوقة في الإنتاج.
  • التسجيل والمراقبة: سجل استدعاءات أدوات MCP للتدقيق. استخدم Azure Monitor وLog Analytics.

تنظيف الموارد

إذا لم تكن تخطط للاستمرار في استخدام هذا التطبيق، احذف مجموعة الموارد لإزالة جميع الموارد التي أنشأتها في هذا الدرس:

az group delete --resource-group $RESOURCE_GROUP --yes --no-wait

الخطوة التالية