المصادقة والتفويض

الخادم العامل عبر stdio يرث أمانه من نظام التشغيل: يعمل باسمك على جهازك، ولا يصل إليه أحد سواك. وفي اللحظة التي تنتقل فيها إلى HTTP يزول ذلك كله، وتصبح لديك نقطة نهاية تُنفّذ أدوات عند الطلب وهي جالسة على الإنترنت المفتوح. هذا الدرس عن إعادة ما زال.

اختيار استراتيجية المصادقة

اختر بناءً على هوية المستدعين، لا بناءً على ما يبدو أكثر صرامة. فالمصادقة المبالغ فيها على خادم بمستخدم واحد عبء صيانة لا يشتري شيئاً:

وأياً كان الفرع الذي تصل إليه، تبقى قاعدة واحدة سارية: السر يأتي من البيئة لا من ملف الكود أبداً. وكل مثال أدناه مكتوب على هذا الأساس.

مصادقة مفتاح API

بسيطة لكن فعالة للخوادم الشخصية:

from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import JSONResponse

class APIKeyMiddleware(BaseHTTPMiddleware):
    def __init__(self, app, api_keys: set):
        super().__init__(app)
        self.api_keys = api_keys

    async def dispatch(self, request, call_next):
        api_key = request.headers.get("X-API-Key")

        if api_key not in self.api_keys:
            return JSONResponse(
                {"error": "مفتاح API غير صالح"},
                status_code=401
            )

        return await call_next(request)

# تطبيق الوسيط
app.add_middleware(APIKeyMiddleware, api_keys={"sk_live_abc123"})

مصادقة JWT

للأنظمة متعددة المستخدمين:

import os
import jwt
from datetime import datetime, timedelta, timezone

# لا تضعه كقيمة نصية في الكود أبداً. قراءته بهذه الطريقة تُفشل التشغيل
# بصوت عالٍ إن كان المتغير مفقوداً، وهذا أفضل من خادم يقلع بسلاسة
# ثم يرفض كل رمز لأسباب لا يراها أحد.
SECRET_KEY = os.environ["MCP_JWT_SECRET"]

def create_token(user_id: str) -> str:
    payload = {
        "sub": user_id,
        "exp": datetime.now(timezone.utc) + timedelta(hours=24)
    }
    return jwt.encode(payload, SECRET_KEY, algorithm="HS256")

def verify_token(token: str) -> dict:
    try:
        return jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
    except jwt.ExpiredSignatureError:
        raise ValueError("انتهت صلاحية الرمز")
    except jwt.InvalidTokenError:
        raise ValueError("رمز غير صالح")

class JWTMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        auth_header = request.headers.get("Authorization", "")

        if not auth_header.startswith("Bearer "):
            return JSONResponse({"error": "الرمز مفقود"}, status_code=401)

        token = auth_header[7:]  # إزالة "Bearer "

        try:
            payload = verify_token(token)
            request.state.user_id = payload["sub"]
        except ValueError as e:
            return JSONResponse({"error": str(e)}, status_code=401)

        return await call_next(request)

التفويض على مستوى الأداة

تقييد الأدوات بناءً على صلاحيات المستخدم:

from mcp.server import MCPServer
from mcp.shared.exceptions import MCPError

TOOL_PERMISSIONS = {
    "search_documents": ["read"],
    "add_document": ["read", "write"],
    "delete_document": ["admin"],
}


async def authz_middleware(ctx, call_next):
    """التحقق من الصلاحيات قبل تشغيل أي معالِج أداة."""
    if ctx.method != "tools/call":
        return await call_next(ctx)

    tool = (ctx.params or {}).get("name", "")
    required = TOOL_PERMISSIONS.get(tool)

    # الأداة التي لا مدخل لها تُمنع، لا تُسمح.
    if required is None:
        raise MCPError(code=-32602, message=f"أداة غير معروفة: {tool}")

    granted = permissions_for(ctx)
    if not all(p in granted for p in required):
        raise MCPError(code=-32603, message=f"صلاحية مرفوضة للأداة: {tool}")

    return await call_next(ctx)


mcp = MCPServer(name="secured-server", middleware=[authz_middleware])

ثلاثة قرارات في هذه الكتلة تستحق التصريح، لأن كل واحد منها طريق يسلكه التفويض عادةً إلى الخطأ:

• التحقق يجري في الوسيط، قبل أي معالِج. فالتفويض المكتوب داخل كل أداة تفويضٌ يمكن أن تنسى كتابته، والأداة التي تنساها هي الأداة المهمة.
• إرجاع .get(tool) للقيمة None يعني المنع. الصيغة الأسبق من هذا النمط استخدمت .get(name, [])، وقائمة المتطلبات الفارغة تجتاز كل تحقق — فتصير الأداة التي لم يضفها أحد إلى الجدول أداةً يستطيع الجميع استدعاءها. الأصل هو المنع.
• all() لا any(). تحتاج add_document إلى القراءة و الكتابة. ومع any() كانت القراءة وحدها لتسمح لقارئ بالكتابة.

تحديد المعدل

الحماية من سوء الاستخدام:

from slowapi import Limiter
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)

@app.route("/mcp")
@limiter.limit("100/minute")
async def handle_mcp(request):
    ...

في القسم التالي، سنستكشف التحديثات والإشعارات الفورية. :::