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

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

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

متى تستخدم SSE

السيناريو	النقل
أداة CLI محلية	stdio
تكامل سطح المكتب	stdio
خادم بعيد	SSE
MCP مستضاف سحابياً	SSE
وصول متعدد المستخدمين	SSE

إعداد خادم 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

يوفر SSE تدفقاً أحادي الاتجاه من الخادم إلى العميل، مع إرسال طلبات العميل عبر POST:

العميل                        الخادم
  │                            │
  │──GET /mcp (تدفق SSE)──────▶│
  │◀───event: message──────────│
  │◀───event: message──────────│
  │                            │
  │──POST /mcp (طلب)──────────▶│
  │◀───event: response─────────│

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

يدعم 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 الخاصة بك. :::