نقل SSE لـ MCP البعيد

بينما يعمل stdio بشكل رائع للخوادم المحلية، أحداث مرسلة من الخادم (SSE) وخليفته Streamable HTTP تمكّن خوادم MCP البعيدة القابلة للوصول عبر HTTP.

مهم: اعتباراً من تحديث مواصفة MCP في 2025-03-26، تم إهمال HTTP+SSE لصالح Streamable HTTP، الذي يدمج نموذج نقطتي النهاية SSE في نقطة نهاية HTTP واحدة يمكن أن ترقى اختيارياً إلى بث SSE. خوادم SSE القديمة لا تزال تعمل، ولكن المشاريع الجديدة يجب أن تستخدم Streamable HTTP. الأنماط في هذا الدرس تنطبق على كليهما — بدائية النقل في SDK الخاص بك هي الشيء الرئيسي الذي يتغير.

اختيار وسيلة النقل

توجد ثلاثة خيارات مستخدمة فعلياً، وأحدها باقٍ للتوافق فقط. ووسائل النقل تتطور أسرع من محتوى الدورات، لذا راجع مواصفة MCP لمعرفة الوضع الحالي قبل بدء أي مشروع جديد.

الأنماط الواردة في بقية هذا الدرس — تعدد العملاء وCORS والمصادقة — تنطبق على كلتا وسيلتي النقل عبر HTTP. ما يتغير بينهما هو بدائية النقل التي يمنحك إياها الـ SDK، لا شكل كودك.

إعداد خادم SSE

from mcp.server import Server
from mcp.server.sse import sse_server
import uvicorn
from starlette.applications import Starlette
from starlette.routing import Route

server = Server(name="remote-mcp")

# تسجيل أدواتك ومواردك
@server.list_tools()
async def list_tools():
    return [...]

# إنشاء معالج SSE
async def handle_sse(request):
    async with sse_server() as (read, write):
        await server.run(read, write)

# إنشاء تطبيق Starlette
app = Starlette(
    routes=[Route("/mcp", handle_sse)],
    debug=True
)

# التشغيل مع uvicorn
if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

تكوين العميل

تكوين خادم MCP بعيد في Claude Desktop:

{
  "mcpServers": {
    "remote-kb": {
      "transport": "sse",
      "url": "https://your-server.com/mcp"
    }
  }
}

تدفق رسائل SSE

عدم التماثل هو ما ينبغي فهمه هنا: التدفق يحمل كل شيء من الخادم، بينما كل ما يذهب إلى الخادم يسافر في طلب POST منفصل. والاستجابات لا تعود عبر الـ POST، بل تصل على التدفق الذي فتحته سابقاً.

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

التعامل مع عملاء متعددين

يدعم SSE بشكل طبيعي عملاء متزامنين متعددين:

from contextlib import asynccontextmanager
from collections import defaultdict

class MultiClientServer:
    def __init__(self):
        self.clients = defaultdict(dict)

    @asynccontextmanager
    async def client_session(self, client_id: str):
        self.clients[client_id] = {"connected": True}
        try:
            yield
        finally:
            del self.clients[client_id]

    async def broadcast(self, message):
        for client_id in self.clients:
            await self.send_to_client(client_id, message)

تكوين CORS

للعملاء المستندين إلى المتصفح، كوّن CORS:

from starlette.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_methods=["GET", "POST"],
    allow_headers=["*"],
)

في القسم التالي، سنضيف المصادقة لحماية نقاط نهاية MCP الخاصة بك. :::