حاولت استخدام Langchain مع خوادم MCP، إليكم الخطوات:

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

💡

عند تنفيذ الاختبار لتطبيقات المعتمدة على API، يتجه المطورون والمختبرون بشكل متزايد إلى أدوات متخصصة مثل Apidog، وهو بديل شامل لـ Postman يبسط دورة تطوير API.

زر

فهم MCP وهدفه

يهدف بروتوكول سياق النموذج إلى توحيد كيفية اتصال التطبيقات القائمة على LLM بأنظمة خارجية متنوعة. اعتبر MCP كواجهة "USB-C للذكاء الاصطناعي" - واجهة عالمية تمكن من تبادل البيانات بشكل سلس وآمن وقابل للتوسع بين LLMs/وكلاء الذكاء الاصطناعي والموارد الخارجية.

يعتمد MCP على معمارية عميل-خادم:

مضيفو MCP: تطبيقات الذكاء الاصطناعي التي تحتاج إلى الوصول إلى البيانات الخارجية
خوادم MCP: مقدمو البيانات أو الأدوات الذين يقدمون المعلومات للمضيفين

يسهل البروتوكول فصل واضح للمسؤوليات، مما يسمح للمطورين بإنشاء موصلات معيارية قابلة لإعادة الاستخدام مع الحفاظ على أمان قوي من خلال التحكم في الأذونات بشكل دقيق.

المعماريات التقنية

تتكون معمارية MCP من ثلاثة مكونات رئيسية:

الخادم: خادم MCP يكشف عن الأدوات ومصادر البيانات من خلال واجهة برمجة التطبيقات الموحدة
العميل: يتواصل تطبيق العميل مع الخادم للوصول إلى الأدوات والبيانات
المحول: توفر LangChain محولات تبسط التكامل بين خوادم MCP وتطبيقات LLM

تتبع تدفق الاتصال هذا النمط:

تطبيق LangChain يطلب تنفيذ البيانات / الأداة
محول MCP يحول الطلب إلى تنسيق بروتوكول MCP
الخادم يعالج الطلب ويعيد النتائج
المحول يحول الاستجابة مرة أخرى إلى تنسيق يمكن استخدامه بواسطة LangChain

المتطلبات السابقة

قبل أن نبدأ، تأكد من أن لديك ما يلي:

تثبيت Python 3.8+
مفتاح API الخاص بـ OpenAI (لاستخدام نماذج GPT مع LangChain)
معرفة أساسية بمفاهيم LangChain
وصول إلى الطرفية (مع عرض الأمثلة على macOS)

إعداد البيئة

أولاً، دعونا ننشئ ونهيئ بيئة التطوير لدينا:

# إنشاء بيئة افتراضية
python3 -m venv MCP_Demo

# تنشيط البيئة الافتراضية
source MCP_Demo/bin/activate

# تثبيت الحزم المطلوبة
pip install langchain-mcp-adapters
pip install langchain-openai

# تعيين مفتاح API الخاص بك في OpenAI
export OPENAI_API_KEY=your_api_key

إنشاء خادم MCP بسيط

سنبدأ ببناء خادم MCP أساسي يوفر عمليات رياضية. أنشئ ملفًا باسم math_server.py:

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

mcp = FastMCP("Math")

@mcp.tool()
def add(a: int, b: int) -> int:
    """إضافة عددين"""
    return a + b

@mcp.tool()
def multiply(a: int, b: int) -> int:
    """ضرب عددين"""
    return a * b

if __name__ == "__main__":
    mcp.run(transport="stdio")

هذا الخادم يكشف عن أداتين رياضيتين: add و multiply. فئة FastMCP تبسط إنشاء الخادم، مما يتولى تفاصيل البروتوكول تلقائيًا. كل دالة مزينة بـ @mcp.tool() تصبح متاحة للعملاء، مع توثيق مستمد من docstrings.

تنفيذ عميل LangChain

بعد ذلك، أنشئ عميل LangChain للتفاعل مع خادم MCP. احفظ ذلك كـ client.py:

# إعداد معلمات الخادم لعملية الاتصال stdio
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
import asyncio

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

# تكوين معلمات الخادم
server_params = StdioServerParameters(
    command="python",
    # حدد مسار ملف الخادم الخاص بك
    args=["math_server.py"],
)

async def run_agent():
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # تهيئة الاتصال
            await session.initialize()
            
            # تحميل أدوات MCP بصيغة LangChain
            tools = await load_mcp_tools(session)
            
            # إنشاء وتشغيل الوكيل
            agent = create_react_agent(model, tools)
            agent_response = await agent.ainvoke({"messages": "ما هو (3 + 5) × 12؟"})
            return agent_response

# تشغيل الوظيفة غير المتزامنة
if __name__ == "__main__":
    result = asyncio.run(run_agent())
    print(result)

يقوم هذا العميل بإنشاء اتصال بخادم MCP، وتحميل الأدوات المتاحة، وإنشاء وكيل LangChain يمكنه استخدام هذه الأدوات لحل المشكلات.

تشغيل المثال

لتشغيل هذا المثال:

ابدأ خادم MCP في علامة تبويب طرفية واحدة:

python3 math_server.py

في علامة تبويب طرفية أخرى، قم بتشغيل العميل:

python3 client.py

سيقوم العميل باستدعاء الوكيل في LangChain، الذي:

يحليل السؤال "(3 + 5) × 12"
يستدعي أداة add مع الوسائط 3 و5
يحصل على النتيجة 8
يستدعي أداة multiply مع الوسائط 8 و12
يعود بالإجابة النهائية: 96

تنفيذ خادم MCP المتقدم

دعونا نوسع تنفيذنا لإنشاء خادم MCP أكثر تعقيدًا يوفر الوصول إلى قاعدة البيانات. يوضح هذا المثال كيفية بناء الموصلات إلى الأنظمة الخارجية:

# db_server.py
from mcp.server.fastmcp import FastMCP
import sqlite3
from typing import List, Dict, Any

class DatabaseConnector:
    def __init__(self, db_path):
        self.conn = sqlite3.connect(db_path)
        self.cursor = self.conn.cursor()
    
    def execute_query(self, query: str) -> List[Dict[str, Any]]:
        self.cursor.execute(query)
        columns = [desc[0] for desc in self.cursor.description]
        results = []
        for row in self.cursor.fetchall():
            results.append({columns[i]: row[i] for i in range(len(columns))})
        return results

mcp = FastMCP("DatabaseTools")
db_connector = DatabaseConnector("example.db")

@mcp.tool()
def run_sql_query(query: str) -> List[Dict[str, Any]]:
    """تنفيذ استعلام SQL على قاعدة البيانات وإرجاع النتائج"""
    try:
        return db_connector.execute_query(query)
    except Exception as e:
        return {"خطأ": str(e)}

if __name__ == "__main__":
    mcp.run(transport="stdio")

دمج عدة خوادم MCP مع LangChain

لتطبيقات أكثر تعقيدًا، قد تحتاج إلى دمج عدة خوادم MCP. إليك كيفية إنشاء عميل يتصل بخوادم متعددة:

# multi_server_client.py
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
import asyncio
from typing import List, Dict

# تعريف تكوينات خوادمنا
servers = [
    {
        "name": "math",
        "params": StdioServerParameters(
            command="python", 
            args=["math_server.py"]
        )
    },
    {
        "name": "database",
        "params": StdioServerParameters(
            command="python", 
            args=["db_server.py"]
        )
    }
]

async def connect_to_server(server_config):
    """الاتصال بخادم MCP واحد وتحميل أداةه"""
    name = server_config["name"]
    params = server_config["params"]
    
    read, write = await stdio_client(params).__aenter__()
    session = ClientSession(read, write)
    await session.__aenter__()
    await session.initialize()
    
    tools = await load_mcp_tools(session)
    return {
        "name": name,
        "session": session,
        "tools": tools,
        "cleanup": lambda: asyncio.gather(
            session.__aexit__(None, None, None),
            stdio_client(params).__aexit__(None, None, None)
        )
    }

async def run_multi_server_agent():
    # الاتصال بجميع الخوادم
    connections = await asyncio.gather(
        *[connect_to_server(server) for server in servers]
    )
    
    try:
        # جمع جميع الأدوات من جميع الخوادم
        all_tools = []
        for connection in connections:
            all_tools.extend(connection["tools"])
        
        # إنشاء الوكيل مع جميع الأدوات
        model = ChatOpenAI(model="gpt-4o")
        agent = create_react_agent(model, all_tools)
        
        # تشغيل الوكيل مع استعلام معقد قد يستخدم عدة خوادم
        response = await agent.ainvoke({
            "messages": "ابحث عن العملاء الذين أنفقوا أكثر من متوسط قيمة الطلب واحسب إنفاقهم الإجمالي."
        })
        
        return response
    
    finally:
        # تنظيف جميع الاتصالات
        for connection in connections:
            await connection["cleanup"]()

# تشغيل الوكيل متعدد الخوادم
if __name__ == "__main__":
    result = asyncio.run(run_multi_server_agent())
    print(result)

معالجة الأخطاء واستراتيجيات الاستعادة

ينبغي أن تتضمن تنفيذات MCP القوية معالجة الأخطاء. إليك نسخة مُحسّنة من العميل توضح معالجة الأخطاء واستراتيجيات الاستعادة:

# robust_client.py
async def run_agent_with_fallbacks():
    try:
        # محاولة الاتصال الأساسي
        async with stdio_client(server_params) as (read, write):
            async with ClientSession(read, write) as session:
                try:
                    await session.initialize()
                    tools = await load_mcp_tools(session)
                    agent = create_react_agent(model, tools)
                    return await agent.ainvoke({"messages": "ما هو (3 + 5) × 12؟"})
                except Exception as e:
                    print(f"خطأ في استخدام أدوات MCP: {e}")
                    # الاستعادة إلى استدعاء النموذج المباشر بدون أدوات
                    return await model.ainvoke([
                        HumanMessage(content="ما هو (3 + 5) × 12؟")
                    ])
    except Exception as connection_error:
        print(f"خطأ في الاتصال: {connection_error}")
        # الاستعادة النهائية
        return {"خطأ": "لم يكن بالإمكان إنشاء اتصال مع خادم MCP"}

اعتبارات الأمان

عند تنفيذ MCP مع LangChain، اعتبر هذه الممارسات الأمنية الجيدة:

التحقق من المدخلات: تحقق دائمًا من المدخلات التي تمرر إلى أدوات MCP لمنع هجمات الحقن
أذونات الأدوات: نفذ أذونات دقيقة لكل أداة
تحديد معدل الاستخدام: طبق حدود معدل الاستخدام لمنع إساءة استخدام الأدوات
المصادقة: نفذ مصادقة صحيحة بين العملاء والخوادم

إليك مثال على تنفيذ أذونات الأدوات:

from mcp.server.fastmcp import FastMCP, Permission

mcp = FastMCP("SecureTools")

# تعريف مستويات الأذونات
READ_PERMISSION = Permission("read", "يمكن قراءة البيانات")
WRITE_PERMISSION = Permission("write", "يمكن تعديل البيانات")

@mcp.tool(permissions=[READ_PERMISSION])
def get_data(key: str) -> str:
    """الحصول على البيانات بواسطة المفتاح (يتطلب إذن القراءة)"""
    # التنفيذ...
    return f"البيانات لـ {key}"

@mcp.tool(permissions=[WRITE_PERMISSION])
def update_data(key: str, value: str) -> bool:
    """تحديث البيانات (يتطلب إذن الكتابة)"""
    # التنفيذ...
    return True

تحسين الأداء

للنشر في الإنتاج، اعتبر تحسينات الأداء التالية:

تجميع الاتصالات: إعادة استخدام اتصالات MCP بدلاً من إنشاء اتصالات جديدة لكل طلب
معالجة دفعات: تجميع استدعاء الأدوات المتعددة عند الإمكان
المعالجة غير المتزامنة: استخدام asyncio للتعامل مع الطلبات المتعددة بالتوازي

مثال على تجميع الاتصالات:

class MCPConnectionPool:
    def __init__(self, max_connections=10):
        self.available_connections = asyncio.Queue(max_connections)
        self.max_connections = max_connections
        self.current_connections = 0
    
    async def initialize(self):
        # إنشاء بعض الاتصالات مسبقًا
        for _ in range(3):  # ابدأ بـ 3 اتصالات
            await self._create_connection()
    
    async def _create_connection(self):
        if self.current_connections >= self.max_connections:
            raise Exception("تم الوصول إلى الحد الأقصى من الاتصالات")
        
        read, write = await stdio_client(server_params).__aenter__()
        session = await ClientSession(read, write).__aenter__()
        await session.initialize()
        
        self.current_connections += 1
        await self.available_connections.put(session)
    
    async def get_connection(self):
        if self.available_connections.empty() and self.current_connections < self.max_connections:
            await self._create_connection()
        
        return await self.available_connections.get()
    
    async def release_connection(self, connection):
        await self.available_connections.put(connection)

اختبار تنفيذات MCP

يعتبر الاختبار الشامل أمرًا بالغ الأهمية لتنفيذات MCP الموثوقة. إليك نهج اختباري باستخدام pytest:

# test_mcp.py
import pytest
import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from langchain_mcp_adapters.tools import load_mcp_tools

@pytest.fixture
async def mcp_session():
    server_params = StdioServerParameters(
        command="python",
        args=["math_server.py"],
    )
    
    read, write = await stdio_client(server_params).__aenter__()
    session = ClientSession(read, write)
    await session.__aenter__()
    await session.initialize()
    
    yield session
    
    await session.__aexit__(None, None, None)
    await stdio_client(server_params).__aexit__(None, None, None)

@pytest.mark.asyncio
async def test_add_tool(mcp_session):
    tools = await load_mcp_tools(mcp_session)
    add_tool = next(tool for tool in tools if tool.name == "add")
    
    result = await add_tool.invoke({"a": 5, "b": 7})
    assert result == 12

@pytest.mark.asyncio
async def test_multiply_tool(mcp_session):
    tools = await load_mcp_tools(mcp_session)
    multiply_tool = next(tool for tool in tools if tool.name == "multiply")
    
    result = await multiply_tool.invoke({"a": 6, "b": 8})
    assert result == 48

الخاتمة

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

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

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

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