أنماط MCP المتقدمة

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

5 دقيقة للقراءة

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

متى تستخدم 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 الخاصة بك. :::

اختبار

اختبار الوحدة 4: أنماط MCP المتقدمة

خذ الاختبار