MCP البعيد عبر Streamable HTTP

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

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

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

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

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

إعداد خادم بعيد

from mcp.server import MCPServer

mcp = MCPServer(name="remote-mcp")


@mcp.tool()
def ping(msg: str = "hi") -> str:
    """إعادة إرسال رسالة كما هي."""
    return f"pong: {msg}"


if __name__ == "__main__":
    mcp.run("streamable-http")

هذا هو الفرق كله عن خادم stdio الذي كنت تكتبه: وسيط واحد يُمرَّر إلى run(). صار الخادم يستمع على /mcp ويتحدث البروتوكول نفسه إلى المعالِجات نفسها.

وحين تحتاج إلى تركيب MCP داخل تطبيق ويب قائم بدل تشغيله مستقلاً، اطلب تطبيق ASGI ووجّه إليه بنفسك:

app = mcp.streamable_http_app()          # تطبيق Starlette يكشف POST/GET على /mcp

تستقبل streamable_http_app() المقابض المهمة — streamable_http_path وstateless_http وjson_response وmax_request_body_size — فالجأ إليها حين تدمج، وإلى mcp.run("streamable-http") حين لا تدمج.

تكوين العميل

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

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

تدفق الرسائل

عدم التماثل هو ما ينبغي فهمه هنا: التدفق يحمل كل شيء من الخادم، بينما كل ما يذهب إلى الخادم يسافر في طلب 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 = mcp.streamable_http_app()

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://app.example.com"],   # سمِّها صراحةً ولا تنشر "*"
    allow_methods=["GET", "POST"],
    allow_headers=["*"],
    expose_headers=["Mcp-Session-Id"],
)

استخدام allow_origins=["*"] على خادم يحمل بيانات اعتماد هو الطريق الذي يتيح لمضيف يعمل في المتصفح من أي أصل أن يستدعي أدواتك. سمِّ الأصول التي تخدمها فعلاً. ولا بد من وجود Mcp-Session-Id في expose_headers وإلا تعذّر على عميل المتصفح قراءته، فينكسر استئناف الجلسة بطريقة تبدو وكأنها خلل في الخادم.

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