MCP Sunucuları ile Langchain'i Kullanmayı Denemem, İşte Adımlar:

Model Context Protocol (MCP), Anthropic tarafından geliştirilen ve Large Language Model (LLM) uygulamalarındaki temel bir zorluğa, yani harici veri kaynaklarından ve araçlardan yalıtılmış olmalarına, çözüm getiren açık kaynaklı bir protokoldür. Bu kapsamlı eğitim, MCP'yi LangChain ile uygulamanız için size rehberlik edecek ve her iki teknolojiden de etkili bir şekilde yararlanan sofistike uygulamalar oluşturma bilgisi sağlayacaktır.

💡

API tabanlı uygulamalar için Test uygulaması yaparken, geliştiriciler ve test uzmanları, API geliştirme yaşam döngüsünü kolaylaştıran kapsamlı bir Postman alternatifi olan Apidog gibi özel araçlara giderek daha fazla yöneliyor.

button

MCP'yi ve Amacını Anlamak

Model Context Protocol, LLM tabanlı uygulamaların çeşitli harici sistemlere nasıl bağlanacağını standartlaştırmayı amaçlar. MCP'yi "Yapay Zeka için USB-C" olarak düşünün; LLM'ler/Yapay Zeka aracılar ile harici kaynaklar arasında kesintisiz, güvenli ve ölçeklenebilir veri alışverişini sağlayan evrensel bir arayüz.

MCP, bir istemci-sunucu mimarisi kullanır:

MCP Ana Bilgisayarları: Harici verilere erişmesi gereken yapay zeka uygulamaları
MCP Sunucuları: Ana bilgisayarlara bilgi sağlayan veri veya araç sağlayıcıları

Protokol, geliştiricilerin modüler, yeniden kullanılabilir bağlayıcılar oluşturmasına olanak tanırken, ayrıntılı izin kontrolleri aracılığıyla sağlam güvenlik sağlayan net bir endişe ayrımı sağlar.

Teknik Mimari

MCP'nin mimarisi üç ana bileşenden oluşur:

Sunucu: MCP sunucusu, araçları ve veri kaynaklarını standartlaştırılmış bir API aracılığıyla sunar
İstemci: İstemci uygulaması, araçlara ve verilere erişmek için sunucuyla iletişim kurar
Adaptör: LangChain, MCP sunucuları ve LLM uygulamaları arasındaki entegrasyonu basitleştiren adaptörler sağlar

İletişim akışı şu deseni izler:

LangChain uygulaması veri/araç yürütme talep eder
MCP adaptörü, isteği MCP protokol formatına dönüştürür
Sunucu isteği işler ve sonuçları döndürür
Adaptör, yanıtı LangChain tarafından kullanılabilir bir formata dönüştürür

Önkoşullar

Başlamadan önce, aşağıdakilere sahip olduğunuzdan emin olun:

Python 3.8+ yüklü
OpenAI API anahtarı (GPT modellerini LangChain ile kullanmak için)
LangChain kavramlarına temel aşinalık
Terminal erişimi (macOS'ta gösterilen örnekler)

Ortamı Kurma

İlk olarak, geliştirme ortamımızı oluşturalım ve yapılandıralım:

# Sanal bir ortam oluşturun
python3 -m venv MCP_Demo

# Sanal ortamı etkinleştirin
source MCP_Demo/bin/activate

# Gerekli paketleri yükleyin
pip install langchain-mcp-adapters
pip install langchain-openai

# OpenAI API anahtarınızı ayarlayın
export OPENAI_API_KEY=your_api_key

Basit Bir MCP Sunucusu Oluşturma

Matematiksel işlemler sağlayan temel bir MCP sunucusu oluşturarak başlayacağız. math_server.py adlı bir dosya oluşturun:

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

mcp = FastMCP("Math")

@mcp.tool()
def add(a: int, b: int) -> int:
    """İki sayıyı topla"""
    return a + b

@mcp.tool()
def multiply(a: int, b: int) -> int:
    """İki sayıyı çarp"""
    return a * b

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

Bu sunucu iki matematiksel araç sunar: add ve multiply. FastMCP sınıfı, protokol ayrıntılarını otomatik olarak işleyerek sunucu oluşturmayı basitleştirir. @mcp.tool() ile dekore edilmiş her işlev, docstring'lerden türetilen dokümantasyonla birlikte istemciler için kullanılabilir hale gelir.

LangChain İstemcisini Uygulama

Ardından, MCP sunucusuyla etkileşim kurmak için bir LangChain istemcisi oluşturun. Bunu client.py olarak kaydedin:

# Sunucu parametrelerini stdio bağlantısı için oluşturun
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")

# Sunucu parametrelerini yapılandırın
server_params = StdioServerParameters(
    command="python",
    # Sunucu dosyanızın yolunu belirtin
    args=["math_server.py"],
)

async def run_agent():
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # Bağlantıyı başlatın
            await session.initialize()
            
            # MCP araçlarını LangChain formatına yükleyin
            tools = await load_mcp_tools(session)
            
            # Aracı oluşturun ve çalıştırın
            agent = create_react_agent(model, tools)
            agent_response = await agent.ainvoke({"messages": "what's (3 + 5) x 12?"})
            return agent_response

# Asenkron fonksiyonu çalıştırın
if __name__ == "__main__":
    result = asyncio.run(run_agent())
    print(result)

Bu istemci, MCP sunucusuna bir bağlantı kurar, mevcut araçları yükler ve bu araçları sorunları çözmek için kullanabilen bir LangChain aracısı oluşturur.

Örneği Çalıştırma

Bu örneği çalıştırmak için:

Bir terminal sekmesinde MCP sunucusunu başlatın:

python3 math_server.py

Başka bir terminal sekmesinde, istemciyi çalıştırın:

python3 client.py

İstemci, LangChain aracısını çağıracak ve bu da şunları yapacaktır:

"(3 + 5) x 12" sorusunu ayrıştırır
3 ve 5 argümanlarıyla add aracını çağırır
Sonucu 8 alır
8 ve 12 argümanlarıyla multiply aracını çağırır
Nihai cevabı döndürür: 96

Gelişmiş MCP Sunucu Uygulaması

Veritabanı erişimi sağlayan daha sofistike bir MCP sunucusu oluşturmak için uygulamamızı genişletelim. Bu örnek, harici sistemlere bağlayıcıların nasıl oluşturulacağını gösterir:

# 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]]:
    """Veritabanında bir SQL sorgusu çalıştırın ve sonuçları döndürün"""
    try:
        return db_connector.execute_query(query)
    except Exception as e:
        return {"error": str(e)}

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

LangChain ile Birden Fazla MCP Sunucusunu Entegre Etme

Daha karmaşık uygulamalar için, birden fazla MCP sunucusunu entegre etmeniz gerekebilir. İşte birden fazla sunucuya bağlanan bir istemci oluşturma şekli:

# 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

# Sunucu yapılandırmalarımızı tanımlayın
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):
    """Tek bir MCP sunucusuna bağlanın ve araçlarını yükleyin"""
    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():
    # Tüm sunuculara bağlanın
    connections = await asyncio.gather(
        *[connect_to_server(server) for server in servers]
    )
    
    try:
        # Tüm sunuculardan tüm araçları toplayın
        all_tools = []
        for connection in connections:
            all_tools.extend(connection["tools"])
        
        # Tüm araçlarla aracı oluşturun
        model = ChatOpenAI(model="gpt-4o")
        agent = create_react_agent(model, all_tools)
        
        # Birden fazla sunucu kullanabilecek karmaşık bir sorguyla aracı çalıştırın
        response = await agent.ainvoke({
            "messages": "Ortalama sipariş değerinden daha fazla harcama yapan müşterileri bulun ve toplam harcamalarını hesaplayın."
        })
        
        return response
    
    finally:
        # Tüm bağlantıları temizleyin
        for connection in connections:
            await connection["cleanup"]()

# Çok sunuculu aracı çalıştırın
if __name__ == "__main__":
    result = asyncio.run(run_multi_server_agent())
    print(result)

Hata İşleme ve Geri Dönüş Stratejileri

Sağlam MCP uygulamaları hata işlemeyi içermelidir. İşte hata işleme ve geri dönüş stratejilerini gösteren istemcinin geliştirilmiş bir sürümü:

# robust_client.py
async def run_agent_with_fallbacks():
    try:
        # Birincil bağlantı denemesi
        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": "what's (3 + 5) x 12?"})
                except Exception as e:
                    print(f"MCP araçları kullanılırken hata: {e}")
                    # Araçlar olmadan doğrudan model çağrısına geri dön
                    return await model.ainvoke([
                        HumanMessage(content="what's (3 + 5) x 12?")
                    ])
    except Exception as connection_error:
        print(f"Bağlantı hatası: {connection_error}")
        # Nihai geri dönüş
        return {"error": "MCP sunucusuna bağlantı kurulamadı"}

Güvenlik Hususları

MCP'yi LangChain ile uygularken, bu güvenlik en iyi uygulamalarını göz önünde bulundurun:

Girdi Doğrulama: Enjeksiyon saldırılarını önlemek için MCP araçlarına her zaman girdileri doğrulayın
Araç İzinleri: Her araç için ayrıntılı izinler uygulayın
Oran Sınırlaması: Araçların kötüye kullanılmasını önlemek için oran sınırları uygulayın
Kimlik Doğrulama: İstemciler ve sunucular arasında uygun kimlik doğrulama uygulayın

İşte araç izinlerini uygulamanın bir örneği:

from mcp.server.fastmcp import FastMCP, Permission

mcp = FastMCP("SecureTools")

# İzin seviyelerini tanımlayın
READ_PERMISSION = Permission("read", "Verileri okuyabilir")
WRITE_PERMISSION = Permission("write", "Verileri değiştirebilir")

@mcp.tool(permissions=[READ_PERMISSION])
def get_data(key: str) -> str:
    """Anahtarla veri al (okuma izni gerektirir)"""
    # Uygulama...
    return f"Veri için {key}"

@mcp.tool(permissions=[WRITE_PERMISSION])
def update_data(key: str, value: str) -> bool:
    """Verileri güncelle (yazma izni gerektirir)"""
    # Uygulama...
    return True

Performans Optimizasyonu

Üretim dağıtımları için, bu performans optimizasyonlarını göz önünde bulundurun:

Bağlantı Havuzu: Her istek için yeni bağlantılar oluşturmak yerine MCP bağlantılarını yeniden kullanın
Toplu İşleme: Mümkün olduğunda birden fazla araç çağrısını gruplayın
Asenkron İşleme: Birden fazla isteği eşzamanlı olarak işlemek için asyncio kullanın

Bağlantı havuzuna örnek:

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):
        # Önceden bazı bağlantılar oluşturun
        for _ in range(3):  # 3 bağlantıyla başlayın
            await self._create_connection()
    
    async def _create_connection(self):
        if self.current_connections >= self.max_connections:
            raise Exception("Maksimum bağlantıya ulaşıldı")
        
        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 Uygulamalarını Test Etme

Güvenilir MCP uygulamaları için kapsamlı testler çok önemlidir. İşte pytest kullanarak bir test yaklaşımı:

# 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

Sonuç

Model Context Protocol, LangChain uygulamalarını harici araçlara ve veri kaynaklarına bağlamak için güçlü bir çerçeve sağlar. Bu bağlantıları standartlaştırarak, MCP geliştiricilerin ortamlarıyla sorunsuz bir şekilde etkileşim kurabilen sofistike yapay zeka aracıları oluşturmasını sağlar.

LangChain'in aracı yeteneklerinin MCP'nin bağlantısıyla birleşimi, gerçekten güçlü, bağlam farkındalığına sahip uygulamalar oluşturmak için bir temel oluşturur. MCP ekosistemi büyümeye devam ettikçe, geliştirme sürecini daha da basitleştiren daha fazla önceden oluşturulmuş sunucu ve aracın ortaya çıkmasını bekleyebiliriz.

Bu eğitim, temel kurulumdan bağlantı havuzu ve hata işleme gibi gelişmiş desenlere kadar, MCP'yi LangChain ile kullanmanın temel kavramlarını ve uygulama ayrıntılarını ele almıştır. Bu uygulamaları izleyerek, her iki teknolojiden de en iyi şekilde yararlanan sağlam, üretime hazır uygulamalar oluşturabilirsiniz.

Daha fazla keşif için, GitHub'da bulunan büyüyen MCP sunucuları ekosistemini incelemeyi veya topluluğa kendi sunucularınızı katkıda bulunmayı düşünün. Yapay zeka aracılarının geleceği, harici araçlardan ve verilerden etkili bir şekilde yararlanma yeteneklerinde yatmaktadır ve MCP bu vizyonu gerçeğe dönüştürme yolunda önemli bir adımdır.