Blog

كيفية دمج LangChain مع خوادم MCP باستخدام محولات langchain-mcp

@apidog

@apidog

14 أبريل 2025

كيفية دمج LangChain مع خوادم MCP باستخدام محولات langchain-mcp

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

هنا يأتي دور مكتبة langchain-mcp-adapters. تعمل كجسر حاسم، تقوم بترجمة أدوات MCP بسلاسة إلى صيغة يمكن لـ LangChain وإطار عمل الوكيل القوي الخاص بها، LangGraph، فهمها واستخدامها. توفر هذه المكتبة ملفًا خفيف الوزن، مما يمكّن المطورين من الاستفادة من النظام البيئي المتزايد لأدوات MCP ضمن تطبيقات LangChain الخاصة بهم.

تشمل الميزات الرئيسية:

سيرشدك هذا البرنامج التعليمي خلال إعداد خوادم MCP، الاتصال بها باستخدام مكتبة المحول، ودمج الأدوات المحملة في وكيل LangGraph.

💡
هل ترغب في أداة اختبار واجهة برمجة التطبيقات رائعة تقوم بإنشاء وثائق واجهة برمجة التطبيقات جميلة؟

هل ترغب في منصة متكاملة، شاملة لفريق مطوريك للعمل معًا ب< a href="https://apidog.com/api-testing/">أقصى إنتاجية؟

يوفر Apidog جميع مطالبك، ويقوم باستبدال Postman بسعر أكثر قابلية للتحمل بكثير!
زر

ما هو خادم MCP؟ كيف يعمل؟

فهم بعض المفاهيم الأساسية أمر ضروري قبل الغوص في الأمثلة:

خادم MCP:

عميل MCP (langchain-mcp-adapters):

تحويل الأداة:

تحويل العبارات:

تثبيت langchain-mcp-adapter

أولاً، قم بتثبيت الحزم اللازمة:

pip install langchain-mcp-adapters langgraph langchain-openai # أو تكامل LLM المفضل لديك من LangChain

ستحتاج أيضًا إلى تكوين مفاتيح API لمزود نموذج اللغة الذي اخترته، عادةً عن طريق تعيين متغيرات البيئة:

export OPENAI_API_KEY=<مفتاح_openai_api_الخاص بك>
# أو export ANTHROPIC_API_KEY=<...> إلخ.

بناء خادم MCP سريع واحد باستخدام langchain-mcp-adapters

دعونا نبني مثالًا بسيطًا: خادم MCP يوفر وظائف رياضية ووكيل LangGraph يستخدم هذه الوظائف.

الخطوة 1: إنشاء خادم MCP (math_server.py)

# math_server.py
from mcp.server.fastmcp import FastMCP

# تهيئة خادم MCP مع اسم
mcp = FastMCP("الرياضيات")

@mcp.tool()
def add(a: int, b: int) -> int:
    """إضافة رقمين"""
    print(f"تنفيذ add({a}, {b})") # سجل جانب الخادم
    return a + b

@mcp.tool()
def multiply(a: int, b: int) -> int:
    """ضرب رقمين"""
    print(f"تنفيذ multiply({a}, {b})") # سجل جانب الخادم
    return a * b

# تعريف عبارة نموذجية
@mcp.prompt()
def configure_assistant(skills: str) -> list[dict]:
    """يهيئ المساعد بالمهارات المحددة."""
    return [
        {
            "role": "assistant", # يتوافق مع AIMessage
            "content": f"أنت مساعد مفيد. لديك المهارات التالية: {skills}. استخدم أداة واحدة في كل مرة.",
        }
    ]

if __name__ == "__main__":
    # تشغيل الخادم باستخدام النقل القياسي
    print("بدء خادم MCP الرياضيات عبر stdin...")
    mcp.run(transport="stdio")

احفظ هذا الكود كـ math_server.py.

الخطوة 2: إنشاء العميل والوكيل (client_app.py)

import asyncio
import os
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

from langchain_mcp_adapters.tools import load_mcp_tools
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI

# --- مهم: تحديث هذا المسار ---
# الحصول على المسار المطلق لملف math_server.py
current_dir = os.path.dirname(os.path.abspath(__file__))
math_server_script_path = os.path.join(current_dir, "math_server.py")
# ---

async def main():
    model = ChatOpenAI(model="gpt-4o") # أو نموذجك المفضل

    # ضبط المعلمات لتشغيل ملف math_server.py
    server_params = StdioServerParameters(
        command="python", # الأمر للتنفيذ
        args=[math_server_script_path], # المعاملات (المسار إلى النص)
        # cwd=..., env=... # مسار العمل المتاختاره ومتغيرات البيئة الاختيارية
    )

    print("الاتصال بخادم MCP...")
    # إنشاء اتصال باستخدام إدارة السياق stdio_client
    async with stdio_client(server_params) as (read, write):
        # إنشاء ClientSession باستخدام تدفقات القراءة/الكتابة
        async with ClientSession(read, write) as session:
            print("تهيئة الجلسة...")
            # تصافح مع الخادم
            await session.initialize()
            print("تم تهيئة الجلسة.")

            print("تحميل أدوات MCP...")
            # جلب أدوات MCP وتحويلها إلى أدوات LangChain
            tools = await load_mcp_tools(session)
            print(f"تم تحميل الأدوات: {[tool.name for tool in tools]}")

            # إنشاء وكيل LangGraph ReAct باستخدام النموذج والأدوات المحملة
            agent = create_react_agent(model, tools)

            print("استدعاء الوكيل...")
            # تشغيل الوكيل
            inputs = {"messages": [("human", "ما هو (3 + 5) * 12؟")]}
            async for event in agent.astream_events(inputs, version="v1"):
                 print(event) # بث الأحداث للمراقبة

            # أو الحصول على الاستجابة النهائية مباشرة
            # final_response = await agent.ainvoke(inputs)
            # print("استجابة الوكيل:", final_response['messages'][-1].content)

if __name__ == "__main__":
    asyncio.run(main())

احفظ هذا كـ client_app.py في نفس الدليل مثل math_server.py.

لتشغيل:

نفذ نص العميل:

python client_app.py

سيقوم نص العميل تلقائيًا ببدء math_server.py كعملية فرعية، والاتصال بها، وتحميل الأدوات add و multiply، واستخدام الوكيل LangGraph لحل مشكلة الرياضيات من خلال استدعاء تلك الأدوات عبر خادم MCP. سترى سجلات من كل من العميل والخادم.

الاتصال بعدة خوادم MCP

غالبًا، سترغب في دمج الأدوات من خوادم متخصصة مختلفة. يجعل MultiServerMCPClient هذا الأمر مباشرًا.

الخطوة 1: إنشاء خادم آخر (weather_server.py)

دعونا نقوم بإنشاء خادم الطقس الذي يعمل باستخدام نقل SSE.

# weather_server.py
from mcp.server.fastmcp import FastMCP
import uvicorn # يحتاج: pip install uvicorn

mcp = FastMCP("الطقس")

@mcp.tool()
async def get_weather(location: str) -> str:
    """الحصول على الطقس للموقع."""
    print(f"تنفيذ get_weather({location})")
    # في السيناريو الواقعي، ستقوم هذه الوظيفة باستدعاء API للطقس
    return f"الجو دائمًا مشمس في {location}"

if __name__ == "__main__":
    # تشغيل الخادم باستخدام نقل SSE (يتطلب خادم ASGI مثل uvicorn)
    # تقوم مكتبة mcp بإنشاء تطبيق FastAPI تلقائيًا لـ SSE.
    # بشكل افتراضي، يعمل على المنفذ 8000 في نقطة النهاية /sse.
    print("بدء خادم MCP للطقس عبر SSE على المنفذ 8000...")
    # uvicorn.run(mcp.app, host="0.0.0.0", port=8000) # يمكنك تشغيله يدويًا
    mcp.run(transport="sse", host="0.0.0.0", port=8000) # أو استخدم mcp.run للراحة

احفظ هذا كـ weather_server.py.

الخطوة 2: تحديث العميل لاستخدام MultiServerMCPClient (multi_client_app.py)

import asyncio
import os
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI

# --- مهم: تحديث المسارات ---
current_dir = os.path.dirname(os.path.abspath(__file__))
math_server_script_path = os.path.join(current_dir, "math_server.py")
# يعمل خادم الطقس بشكل منفصل، اتصل عبر URL
# ---

async def main():
    model = ChatOpenAI(model="gpt-4o")

    # تعريف الاتصالات لخوادم متعددة
    server_connections = {
        "math_service": { # اسم فريد لهذا الاتصال
            "transport": "stdio",
            "command": "python",
            "args": [math_server_script_path],
            # أضف معلمات StdioConnection الأخرى إذا لزم الأمر (env، cwd، إلخ.)
        },
        "weather_service": { # اسم فريد لهذا الاتصال
            "transport": "sse",
            "url": "http://localhost:8000/sse", # URL حيث يعمل weather_server
            # أضف معلمات SSEConnection الأخرى إذا لزم الأمر (headers، timeout، إلخ.)
        }
    }

    print("الاتصال بعدة خوادم MCP...")
    # استخدم إدارة سياق MultiServerMCPClient
    async with MultiServerMCPClient(server_connections) as client:
        print("تم إنشاء الاتصالات.")

        # احصل على *كل* الأدوات من *كل* الخوادم المتصلة
        all_tools = client.get_tools()
        print(f"تم تحميل الأدوات: {[tool.name for tool in all_tools]}")

        # إنشاء وكيل باستخدام قائمة الأدوات المجمعة
        agent = create_react_agent(model, all_tools)

        # --- التفاعل مع الوكيل ---
        print("\nاستدعاء الوكيل لاستفسار رياضي...")
        math_inputs = {"messages": [("human", "ما هو (3 + 5) * 12؟")]}
        math_response = await agent.ainvoke(math_inputs)
        print("استجابة الرياضيات:", math_response['messages'][-1].content)

        print("\nاستدعاء الوكيل لاستفسار عن الطقس...")
        weather_inputs = {"messages": [("human", "كيف يكون الطقس في نيويورك؟")]}
        weather_response = await agent.ainvoke(weather_inputs)
        print("استجابة الطقس:", weather_response['messages'][-1].content)

        # --- مثال: الحصول على عبارة ---
        # print("\nالحصول على عبارة من خادم الرياضيات...")
        # prompt_messages = await client.get_prompt(
        #     server_name="math_service", # استخدم الاسم المحدد في الاتصالات
        #     prompt_name="configure_assistant",
        #     arguments={"skills": "الحساب الأساسي"}
        # )
        # print("العبارة:", prompt_messages)


if __name__ == "__main__":
    # ابدأ خادم الطقس أولاً في نافذة طرفية منفصلة:
    # python weather_server.py
    # ثم قم بتشغيل نص العميل هذا:
    asyncio.run(main())

احفظ هذا كـ multi_client_app.py.

لتشغيل:

  1. ابدأ خادم الطقس في نافذة طرفية واحدة: python weather_server.py
  2. قم بتشغيل تطبيق العميل المتعدد في نافذة طرفية أخرى: python multi_client_app.py

سيقوم MultiServerMCPClient ببدء math_server.py كعملية فرعية (stdin) والاتصال بـ weather_server.py المتعمل (sse). تجمع الأدوات (add، multiply، get_weather) والتي ستكون متاحة بعد ذلك لوكيل LangGraph.

الدمج مع خادم API LangGraph

يمكنك نشر وكيل LangGraph باستخدام أدوات MCP كخدمة API مستمرة باستخدام langgraph deploy. المفتاح هو إدارة دورة حياة MultiServerMCPClient بشكل صحيح ضمن سياق تطبيق LangGraph.

أنشئ ملف graph.py:

# graph.py
from contextlib import asynccontextmanager
import os
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI # أو Anthropic، إلخ.

# --- مهم: تحديث المسارات ---
# على افتراض أن الخوادم تكون نسبية بالنسبة للمكان الذي يعمل فيه خادم LangGraph
math_server_script_path = os.path.abspath("math_server.py")
# ---

# تعريف الاتصالات (تأكد من صحة المسارات/عناوين URL لبيئة الخادم)
server_connections = {
    "math_service": {
        "transport": "stdio",
        "command": "python",
        "args": [math_server_script_path],
    },
    "weather_service": {
        "transport": "sse",
        "url": "http://localhost:8000/sse", # يجب تشغيل خادم الطقس بشكل مستقل
    }
}

model = ChatOpenAI(model="gpt-4o")

# استخدم مدير سياق غير متزامن للتعامل مع إعداد العميل/إيقافه
@asynccontextmanager
async def lifespan(_app): # تتوقع LangGraph هذا الهيكل لإدارة الحياة
    async with MultiServerMCPClient(server_connections) as client:
        print("تم تهيئة MCP Client ضمن دورة الحياة.")
        # إنشاء الوكيل *داخل* السياق الذي يكون فيه العميل نشطًا
        agent = create_react_agent(model, client.get_tools())
        yield {"agent": agent} # اجعل الوكيل متاحًا

# لا حاجة لتعريف مخطط رئيسي منفصل إذا كانت دورة الحياة توفره

قم بتكوين langgraph.json (أو pyproject.toml تحت [tool.langgraph]) لاستخدام هذا التعريف المخطط مع مدير دورة الحياة:

// langgraph.json (مثال)
{
  "dependencies": ["."], // أو تحديد الحزم المطلوبة
  "graphs": {
    "my_mcp_agent": {
      "entrypoint": "graph:agent", // يشير إلى المفتاح الذي تم إرجاعه بواسطة دورة الحياة
      "lifespan": "graph:lifespan"
    }
  }
}

الآن، عند تشغيل langgraph up، ستقوم دالة lifespan بالتنفيذ، بدءًا من MultiServerMCPClient (وخادم الرياضيات stdin). سيتم تقديم الوكيل الذي تم إنشاؤه ضمن هذا السياق بواسطة LangGraph. تذكر أن خادم الطقس SSE لا يزال يحتاج إلى التشغيل بشكل منفصل.

نقل الخادم (stdin مقابل SSE)

stdio:

sse (أحداث السيرفر المرسلة):

اختر وسيلة النقل بناءً على احتياجات النشر الخاصة بك.

تكوين عميل متقدم لإعداد langchain-mcp-adapters

StdioConnection و SSEConnection في MultiServerMCPClient معلمات اختيارية إضافية لتحكم أكثر دقة:

راجع تعريف MultiServerMCPClient في langchain_mcp_adapters/client.py للحصول على التفاصيل.

استنتاج (100 كلمة)

تقوم مكتبة langchain-mcp-adapters بفعالية بربط الفجوة بين بروتوكول نموذج السياق القياسي والنظام البيئي المرن لـ LangChain. من خلال توفير MultiServerMCPClient وتحويل الأدوات التلقائي، يسمح للمطورين بدمج أدوات متباينة متوافقة مع MCP في وكلاء LangChain وتطبيقات LangGraph الخاصة بهم بسهولة.

تشمل سير العمل الأساسي:

  1. تحديد الأدوات (ويمكن اختياريًا العبارات) في خادم MCP باستخدام @mcp.tool().
  2. تكوين MultiServerMCPClient مع تفاصيل الاتصال (stdio أو sse) لكل خادم.
  3. استخدام مدير سياق العميل (async with ...) للاتصال واسترجاع الأدوات عبر client.get_tools().
  4. تمرير الأدوات المسترجعة المتوافقة مع LangChain إلى وكيلك (create_react_agent أو وكلاء مخصصين).

يمكن أن يؤدي ذلك إلى بناء تطبيقات AI قوية ونموذجية تستفيد من الأدوات الخارجية المتخصصة من خلال بروتوكول موحد. استكشف الأمثلة والاختبارات داخل المستودع للحصول على مزيد من الرؤى.

ممارسة تصميم API في Apidog

اكتشف طريقة أسهل لبناء واستخدام واجهات برمجة التطبيقات

التسجيل مجانًاتحميل