كيفية نشر خوادم MCP على AWS Lambda

بروتوكول سياق النموذج (MCP) يبرز بسرعة كطريقة قياسية لتمكين نماذج اللغة الكبيرة (LLMs) مثل كلود، شات جي بي تي، وغيرها للتفاعل مع العالم الخارجي. من خلال تحديد الأدوات بطريقة منظمة، يسمح بروتوكول MCP لنماذج LLM بطلب إجراءات، وجلب بيانات في الوقت الحقيقي، أو التفاعل مع واجهات برمجة التطبيقات الخارجية، متجاوزة بيانات تدريبها الثابتة.

ومع ذلك، غالبًا ما يمثل نشر هذه "الخوادم" الخاصة بـ MCP (التي تقدم الأدوات) تحديًا، خاصة في البيئات السحابية الحديثة. العديد من تنفيذات MCP الأولية صممت للتنفيذ المحلي، حيث تتواصل عبر المدخلات/المخرجات القياسية (stdio)، أو باستخدام بروتوكولات مثل الأحداث المرسلة من الخادم (SSE) لدفق البيانات. على الرغم من أنها وظيفية، فإن هذه الطرق غالبًا ما تعتمد على الاتصالات المستمرة والسلوكيات التي تعتمد على الحالة، مما يجعلها غير ملائمة للأنظمة السحابية القابلة للتوسيع وغير المعتمدة على الحالة مثل AWS Lambda.

AWS Lambda تقدم فوائد هائلة: التوسع التلقائي، وكفاءة تكلفة الدفع حسب الاستخدام، وعدم وجود تكاليف إدارة الخادم. كيف يمكننا سد الفجوة وتشغيل خوادم MCP جاهزة للإنتاج في هذه البيئة بدون خوادم؟

أدخلMCPEngine، تنفيذ مفتوح المصدر بلغة بايثون لبروتوكول MCP مصمم خصيصًا لمعالجة هذه التحديات. يدعم MCPEngine HTTP القابل للبث بالإضافة إلى SSE، مما يجعله متوافقًا تمامًا مع نموذج الطلب/الاستجابة في AWS Lambda. كما يجمع ميزات أساسية للنشر في الإنتاج، بما في ذلك دعم المصادقة المدمج والتعبئة المبسطة.

تستكشف هذه المقالة كيفية الاستفادة من MCPEngine لبناء ونشر خوادم MCP على AWS Lambda، مع تغطية الأدوات غير المعتمدة على الحالة، وإدارة الحالة، والمصادقة.

نحن متحمسون لمشاركة أن دعم MCP سيأتي قريبًا إلى Apidog! 🚀

خادم Apidog MCP يتيح لك إدخال وثائق واجهة برمجة التطبيقات مباشرة إلى الذكاء الاصطناعي، مما يعزز تجربتك في البرمجة! سواء كنت تستخدم Cursor أو Cline أو Windsurf - ستجعل عملية التطوير أسرع وأكثر سلاسة.… pic.twitter.com/ew8U38mU0K

— Apidog (@ApidogHQ) 19 مارس 2025

المفاهيم الأساسية: MCPEngine و Lambda

قبل الغوص في النشر، دعونا نفهم مكونات MCPEngine الرئيسية لتكامل Lambda:

1. MCPEngine: الفئة المركزية التي تنظم الأدوات الخاصة بك وتتعامل مع اتصالات MCP.
2. @engine.tool() الديكوراتور: يسجل دالة بايثون كأداة MCP. يصبح اسم الدالة هو اسم الأداة، وتعمل سطر الوصف الخاص بها كالوصف المقدم لنموذج LLM.
3. engine.get_lambda_handler(): هذه الطريقة تقوم بإنشاء دالة معالج متوافقة مع AWS Lambda. أنت تعرض هذه الدالة، ويتولى MCPEngine مسؤولية ترجمة الحمولة الحدثية لـ Lambda إلى طلبات MCP وتنسيق الاستجابات.

بناء أداة بسيطة غير معتمدة على الحالة

لنبدأ بالأساسيات: أداة غير معتمدة على الحالة تم نشرها على Lambda. هذا المثال يوفر أداة تحية بسيطة.

المتطلبات المسبقة:

• بايثون 3.8+
• حساب AWS مع أذونات لإدارة Lambda و ECR و IAM.
• Docker مثبت محلياً.
• تم تكوين AWS CLI.

1. تثبيت MCPEngine:

pip install mcpengine[cli,lambda]

2. إنشاء التطبيق (app.py):

# app.py
from mcpengine import MCPEngine, Context

# تهيئة المحرك
engine = MCPEngine()

@engine.tool()
def personalized_greeting(name: str) -> str:
    """
    يولد تحية ودية للاسم المحدد.
    استخدم هذه الأداة عند الطلب لتحية شخص ما.
    """
    # منطق غير معتمد على الحالة بسيط
    return f"مرحبًا، {name}! مرحبًا بك في عالم MCP بدون خوادم."

# احصل على دالة المعالج لـ Lambda
handler = engine.get_lambda_handler()

هذا الكود يعرف أداة واحدة، personalized_greeting، التي تأخذ name وترجع سلسلة نصية. المتغير handler هو ما ستقوم AWS Lambda باستدعائه.

سير العمل للنشر: من الكود إلى السحابة

يتطلب نشر تطبيق MCPEngine إلى Lambda حاوية باستخدام Docker، ودفعها إلى Amazon Elastic Container Registry (ECR)، وتكوين دالة Lambda.

1. Dockerize التطبيق (Dockerfile):

# استخدم صورة قاعدة بايثون الرسمية لـ AWS Lambda
FROM public.ecr.aws/lambda/python:3.12

# تعيين دليل العمل في الحاوية
WORKDIR ${LAMBDA_TASK_ROOT}

# نسخ المتطلبات أولاً للاستفادة من ذاكرة التخزين المؤقت لـ Docker
COPY requirements.txt .
# تثبيت التبعيات (على افتراض أن mcpengine مدرج في requirements.txt)
# أو تثبيت مباشرة: RUN pip install --no-cache-dir mcpengine[cli,lambda]
RUN pip install --no-cache-dir -r requirements.txt

# نسخ بقية كود التطبيق
COPY app.py .

# تعيين الأمر لتشغيل دالة المعالج (app.handler تعني المعالج في app.py)
CMD ["app.handler"]

(تأكد من أن لديك ملف requirements.txt الذي يسرد mcpengine[cli,lambda] أو تعديل الأمر RUN وفقًا لذلك).

2. بناء ودفع صورة Docker إلى ECR:

أولاً، قم بإنشاء مستودع ECR (استبدل <region> و<repo-name>):

aws ecr create-repository --repository-name <repo-name> --region <region>

دون حساب AWS ID الخاص بك و URI المستودع من الناتج (<account-id>.dkr.ecr.<region>.amazonaws.com/<repo-name>).

الآن، قم ببناء، وتوسيم، ودفع الصورة:

# مصادقة Docker مع ECR
aws ecr get-login-password --region <region> | docker login --username AWS --password-stdin <account-id>.dkr.ecr.<region>.amazonaws.com

# بناء الصورة (استخدم --platform لبناء عبر الأنظمة المعمارية إذا لزم الأمر)
docker build --platform=linux/amd64 -t <repo-name>:latest .

# تمييز الصورة لـ ECR
docker tag <repo-name>:latest <account-id>.dkr.ecr.<region>.amazonaws.com/<repo-name>:latest

# دفع الصورة إلى ECR
docker push <account-id>.dkr.ecr.<region>.amazonaws.com/<repo-name>:latest

3. إنشاء وتكوين دالة Lambda:

ستحتاج إلى دور تنفيذ IAM لـ Lambda أولاً. إذا لم يكن لديك واحد، قم بإنشاء دور أساسي:

# (مبسط - اضبط سياسة الثقة والأذونات حسب الحاجة)
aws iam create-role --role-name lambda-mcp-role --assume-role-policy-document '{"Version": "2012-10-17","Statement": [{"Effect": "Allow","Principal": {"Service": "lambda.amazonaws.com"},"Action": "sts:AssumeRole"}]}'
aws iam attach-role-policy --role-name lambda-mcp-role --policy-arn arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole

الآن، قم بإنشاء دالة Lambda باستخدام صورة ECR (استبدل العناصر النائبة):

aws lambda create-function \
  --function-name mcp-greeter-function \
  --package-type Image \
  --code ImageUri=<account-id>.dkr.ecr.<region>.amazonaws.com/<repo-name>:latest \
  --role arn:aws:iam::<account-id>:role/lambda-mcp-role \
  --timeout 30 \
  --memory-size 512 \
  --region <region>

4. الكشف عبر وظيفة URL:

لجعل Lambda callable عبر HTTP دون API Gateway، أنشئ وظيفة URL:

aws lambda create-function-url-config \
  --function-name mcp-greeter-function \
  --auth-type NONE \
  --region <region>

# أضف إذنًا للوصول العام (قم بضبطه إذا كانت المصادقة مطلوبة)
aws lambda add-permission \
  --function-name mcp-greeter-function \
  --statement-id FunctionURLAllowPublicAccess \
  --action lambda:InvokeFunctionUrl \
  --principal '*' \
  --function-url-auth-type NONE \
  --region <region>

دون وظيفة URL التي تم إرجاعها بواسطة الأمر create-function-url-config. خادم MCP غير المعتمد على الحالة الخاص بك الآن مباشر!

إدارة الحالة باستخدام lifespan السياق

Lambda غير معتمدة على الحالة، ولكن العديد من الأدوات تحتاج إلى الوصول إلى قواعد البيانات، أو تجمعات الاتصال، أو موارد أخرى تم تهيئتها عند البدء. يتعامل MCPEngine مع هذا من خلال معامل lifespan، الذي يقبل وحدة إدارة سياق غير متزامنة.

تشغل وظيفة lifespan كود الإعداد الخاص بها (قبل yield) عندما يبدأ حاوية Lambda وكود الإيقاف (بعد yield) عندما يتم إيقاف الحاوية. القيمة التي يتمyield تصبح متاحة في وظائف أدواتك عبر كائن ctx (السياق).

لنقم ببناء مسجل أحداث بسيط يقوم بتخزين الأحداث في قاعدة بيانات RDS Postgres.

1. تعديل app.py:

# app.py (مثال معتمد على الحالة)
import os
import psycopg2
from contextlib import asynccontextmanager
from mcpengine import MCPEngine, Context

# افترض أن تفاصيل اتصال DB موجودة في متغيرات البيئة
DB_HOST = os.environ.get("DB_HOST")
DB_USER = os.environ.get("DB_USER")
DB_PASS = os.environ.get("DB_PASS")
DB_NAME = os.environ.get("DB_NAME")

@asynccontextmanager
async def db_connection_manager():
    """يدير تجمع الاتصال بقاعدة البيانات."""
    conn = None
    try:
        print("يتم إنشاء اتصال DB...")
        conn = psycopg2.connect(
            host=DB_HOST,
            user=DB_USER,
            password=DB_PASS,
            dbname=DB_NAME
        )
        # إنشاء جدول إذا لم يكن موجودًا (مثال بسيط)
        with conn.cursor() as cur:
             cur.execute("""
                CREATE TABLE IF NOT EXISTS events (
                    id SERIAL PRIMARY KEY,
                    event_name TEXT NOT NULL,
                    timestamp TIMESTAMP DEFAULT now()
                );
             """)
        conn.commit()
        print("الاتصال بالقاعدة جاهز.")
        yield {"db_conn": conn} # اجعل الاتصال متاحًا عبر ctx.db_conn
    finally:
        if conn:
            print("إغلاق اتصال القاعدة.")
            conn.close()

# تهيئة المحرك مع مدير فترة الحياة
engine = MCPEngine(lifespan=db_connection_manager)

@engine.tool()
def log_event(event_name: str, ctx: Context) -> str:
    """يسجل حدثًا باسم المعطى إلى قاعدة البيانات."""
    try:
        with ctx.db_conn.cursor() as cur:
            cur.execute("INSERT INTO events (event_name) VALUES (%s)", (event_name,))
        ctx.db_conn.commit()
        return f"تم تسجيل الحدث '{event_name}' بنجاح."
    except Exception as e:
        # معالجة الأخطاء الأساسية
        ctx.db_conn.rollback()
        return f"خطأ في تسجيل الحدث: {e}"

@engine.tool()
def get_latest_events(limit: int = 5, ctx: Context) -> list[str]:
    """يسترجع أحدث الأحداث المسجلة من قاعدة البيانات."""
    try:
        with ctx.db_conn.cursor() as cur:
            cur.execute("SELECT event_name, timestamp FROM events ORDER BY timestamp DESC LIMIT %s", (limit,))
            events = [f"[{row[1].strftime('%Y-%m-%d %H:%M:%S')}] {row[0]}" for row in cur.fetchall()]
            return events
    except Exception as e:
        return [f"خطأ في استرجاع الأحداث: {e}"]

# احصل على معالج Lambda 
handler = engine.get_lambda_handler()

2. اعتبارات النشر:

• قاعدة البيانات: تحتاج إلى مثيل RDS قابل للوصول (أو قاعدة بيانات أخرى).
• الشبكات: قم بتكوين إعدادات VPC الخاصة بدالة Lambda للسماح بالوصول إلى مثيل RDS (مجموعات الأمان، الشبكات الفرعية).
• متغيرات البيئة: قم بتمرير DB_HOST وDB_USER وDB_PASS وDB_NAME كمتغيرات بيئة إلى دالة Lambda.
• IAM: قد يحتاج دور تنفيذ Lambda إلى أذونات إضافية إذا كان يصل إلى خدمات AWS الأخرى (مثل Secrets Manager لبيانات الاعتماد الخاصة بقاعدة البيانات).

قم بتحديث Dockerfile إذا لزم الأمر (على سبيل المثال، لتثبيت psycopg2-binary)، أعد بناء/دفع الصورة، وقم بتحديث كود دالة Lambda وتكوينها (متغيرات البيئة، إعدادات VPC).

تأمين الأدوات باستخدام المصادقة

تحتاج أدوات الإنتاج إلى المصادقة. يدمج MCPEngine مع موفري OpenID Connect (OIDC) مثل Google وAWS Cognito وAuth0، وغيرها.

1. تكوين موفر OIDC:
قم بإعداد معرف عميل OAuth مع موفر OAuth الذي اخترته (مثل Google Cloud Console). ستحتاج إلى معرف العميل ومر可能 يمكن أن تكون أسرار العميل (اعتمادًا على السلسلة).

2. تحديث app.py للمصادقة:

# app.py (مثال مصادق عليه - مقتطفات)
import os
# ... استيرادات أخرى ...
from mcpengine import MCPEngine, Context, GoogleIdpConfig # أو أي IdpConfig آخر

# ... db_connection_manager ...

# تكوين موفر الهوية - باستخدام Google كمثال
# يفترض أن GOOGLE_CLIENT_ID مضبوط كمتغير بيئة
google_config = GoogleIdpConfig(
   client_id=os.environ.get("GOOGLE_CLIENT_ID")
   # يمكن استنتاج المصدر أو تعيينه صراحة
)

# تهيئة المحرك مع فترة الحياة وإعداد موفر الهوية
engine = MCPEngine(
    lifespan=db_connection_manager,
    idp_config=google_config
)

# تأمين أداة log_event
@engine.auth() # أضف هذه الزخرفة
@engine.tool()
def log_event(event_name: str, ctx: Context) -> str:
    """يسجل حدثًا باسم المعطى إلى قاعدة البيانات. يتطلب المصادقة."""
    # الوصول إلى معلومات المستخدم المصدق إذا لزم الأمر: user_email = ctx.user.email
    user_email = ctx.user.email if ctx.user else "غير معروف"
    print(f"المستخدم المصدق: {user_email}")
    try:
        # ... (تبقى منطق قاعدة البيانات كما هو) ...
         return f"تم تسجيل الحدث '{event_name}' بنجاح من قبل {user_email}."
    except Exception as e:
        # ... معالجة الأخطاء ...
        return f"خطأ في تسجيل الحدث من قبل {user_email}: {e}"

# يمكن أن تبقى get_latest_events غير مصادق عليها أو تأمينها أيضًا
@engine.tool()
def get_latest_events(limit: int = 5, ctx: Context) -> list[str]:
   # ... (يبقى المنطق كما هو) ...

# احصل على معالج Lambda
handler = engine.get_lambda_handler()

التغييرات الرئيسية:

• استيراد GoogleIdpConfig (أو المناسبة لموفر الهوية الخاص بك).
• إنشاء MCPEngine مع معلمة idp_config.
• إضافة زخرفة @engine.auth() فوق @engine.tool() للوظيفة (أو الوظائف) التي تتطلب المصادقة. سيرفض MCPEngine تلقائيًا الطلبات التي لا تحتوي على رمز JWT صالح تم التحقق منه ضد المفاتيح العامة لموفر الهوية.
• معلومات المستخدم المصدق (من مطالبات JWT) تتاح عبر ctx.user.

3. النشر:

• مرر المتغيرات البيئية اللازمة للمصادقة (مثل GOOGLE_CLIENT_ID) إلى دالة Lambda الخاصة بك.
• أعد بناء/دفع الصورة وقم بتحديث دالة Lambda.

الاتصال بعميل LLM

بمجرد نشر خادم MCP الخاص بك على Lambda مع وظيفة URL، يمكنك الاتصال بالعملاء المتوافقين. استخدام mcpengine proxy هو وسيلة مريحة لجسر العملاء مثل كلود:

mcpengine proxy <your-chosen-service-name> <your-lambda-function-url> --mode http --claude

إذا كنت تستخدم المصادقة:

mcpengine proxy <your-chosen-service-name> <your-lambda-function-url> \
  --mode http \
  --claude \
  --client-id <your-google-client-id> \
  --client-secret <your-google-client-secret> # مطلوب لتدفق التحصيل الرمزي

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

الخلاصة

يمكّن نشر خوادم MCP على AWS Lambda من تحقيق قابلية توسيع مذهلة وكفاءة تشغيلية لتوسيع قدرات LLM. غالبًا ما تكافح تنفيذات MCP التقليدية في البيئات غير المعتمدة على الحالة، ولكن MCPEngine يوفر حلاً مفتوح المصدر قويًا. من خلال دعم HTTP القابل للبث، وتقديم إدارة السياق عبر lifespan، والتكامل بسلاسة مع OIDC للمصادقة، يجعل MCPEngine استخدام MCP غير المعتمد على الخادم ليس ممكنًا فحسب، بل عملياً لحالات الاستخدام الإنتاجية. سواء كنت تبني أدوات بسيطة غير معتمدة على الحالة أو تطبيقات معقدة وموثوقة ومعتمدة، يقدم MCPEngine مع AWS Lambda منصة قوية للجيل القادم من التفاعلات المعتمدة على الذكاء الاصطناعي.