Saya Mencoba Menggunakan Langchain dengan Server MCP, Berikut Langkah-Langkahnya:

Model Context Protocol (MCP) adalah protokol sumber terbuka yang dikembangkan oleh Anthropic yang mengatasi tantangan mendasar dalam aplikasi Large Language Model (LLM): isolasi mereka dari sumber data dan alat eksternal. Tutorial komprehensif ini akan memandu Anda dalam mengimplementasikan MCP dengan LangChain, memberi Anda pengetahuan untuk membuat aplikasi canggih yang memanfaatkan kedua teknologi secara efektif.

💡

Saat menerapkan Pengujian untuk aplikasi berbasis API, pengembang dan penguji semakin beralih ke alat khusus seperti Apidog, alternatif Postman komprehensif yang menyederhanakan siklus hidup pengembangan API.

button

Memahami MCP dan Tujuannya

Model Context Protocol bertujuan untuk menstandarisasi cara aplikasi berbasis LLM terhubung ke berbagai sistem eksternal. Anggap MCP sebagai "USB-C untuk AI" - antarmuka universal yang memungkinkan pertukaran data yang mulus, aman, dan terukur antara LLM/agen AI dan sumber daya eksternal.

MCP menggunakan arsitektur client-server:

MCP Hosts: Aplikasi AI yang perlu mengakses data eksternal
MCP Servers: Penyedia data atau alat yang memasok informasi ke host

Protokol ini memfasilitasi pemisahan tugas yang jelas, memungkinkan pengembang untuk membuat konektor modular yang dapat digunakan kembali sambil mempertahankan keamanan yang kuat melalui kontrol izin granular.

Arsitektur Teknis

Arsitektur MCP terdiri dari tiga komponen utama:

Server: Server MCP mengekspos alat dan sumber data melalui API standar
Client: Aplikasi klien berkomunikasi dengan server untuk mengakses alat dan data
Adapter: LangChain menyediakan adapter yang menyederhanakan integrasi antara server MCP dan aplikasi LLM

Alur komunikasi mengikuti pola ini:

Aplikasi LangChain meminta eksekusi data/alat
Adapter MCP mengubah permintaan menjadi format protokol MCP
Server memproses permintaan dan mengembalikan hasil
Adapter mengubah respons kembali menjadi format yang dapat digunakan oleh LangChain

Prasyarat

Sebelum kita mulai, pastikan Anda memiliki yang berikut:

Python 3.8+ terinstal
Kunci API OpenAI (untuk menggunakan model GPT dengan LangChain)
Pemahaman dasar tentang konsep LangChain
Akses terminal (contoh yang ditampilkan di macOS)

Menyiapkan Lingkungan

Pertama, mari kita buat dan konfigurasi lingkungan pengembangan kita:

# Buat lingkungan virtual
python3 -m venv MCP_Demo

# Aktifkan lingkungan virtual
source MCP_Demo/bin/activate

# Instal paket yang diperlukan
pip install langchain-mcp-adapters
pip install langchain-openai

# Atur kunci API OpenAI Anda
export OPENAI_API_KEY=your_api_key

Membuat Server MCP Sederhana

Kita akan mulai dengan membangun server MCP dasar yang menyediakan operasi matematika. Buat file bernama math_server.py:

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

mcp = FastMCP("Math")

@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two numbers"""
    return a + b

@mcp.tool()
def multiply(a: int, b: int) -> int:
    """Multiply two numbers"""
    return a * b

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

Server ini mengekspos dua alat matematika: add dan multiply. Kelas FastMCP menyederhanakan pembuatan server, menangani detail protokol secara otomatis. Setiap fungsi yang didekorasi dengan @mcp.tool() menjadi tersedia untuk klien, dengan dokumentasi yang berasal dari docstring.

Mengimplementasikan Klien LangChain

Selanjutnya, buat klien LangChain untuk berinteraksi dengan server MCP. Simpan ini sebagai client.py:

# Create server parameters for stdio connection
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")

# Configure server parameters
server_params = StdioServerParameters(
    command="python",
    # Specify the path to your server file
    args=["math_server.py"],
)

async def run_agent():
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            # Initialize the connection
            await session.initialize()
            
            # Load MCP tools into LangChain format
            tools = await load_mcp_tools(session)
            
            # Create and run the agent
            agent = create_react_agent(model, tools)
            agent_response = await agent.ainvoke({"messages": "what's (3 + 5) x 12?"})
            return agent_response

# Run the async function
if __name__ == "__main__":
    result = asyncio.run(run_agent())
    print(result)

Klien ini membuat koneksi ke server MCP, memuat alat yang tersedia, dan membuat agen LangChain yang dapat menggunakan alat ini untuk memecahkan masalah.

Menjalankan Contoh

Untuk menjalankan contoh ini:

Mulai server MCP di satu tab terminal:

python3 math_server.py

Di tab terminal lain, jalankan klien:

python3 client.py

Klien akan memanggil agen LangChain, yang akan:

Mengurai pertanyaan "(3 + 5) x 12"
Memanggil alat add dengan argumen 3 dan 5
Mendapatkan hasil 8
Memanggil alat multiply dengan argumen 8 dan 12
Mengembalikan jawaban akhir: 96

Implementasi Server MCP Tingkat Lanjut

Mari kita perluas implementasi kita untuk membuat server MCP yang lebih canggih yang menyediakan akses database. Contoh ini menunjukkan cara membangun konektor ke sistem eksternal:

# 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]]:
    """Execute an SQL query on the database and return results"""
    try:
        return db_connector.execute_query(query)
    except Exception as e:
        return {"error": str(e)}

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

Mengintegrasikan Beberapa Server MCP dengan LangChain

Untuk aplikasi yang lebih kompleks, Anda mungkin perlu mengintegrasikan beberapa server MCP. Berikut cara membuat klien yang terhubung ke beberapa server:

# 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

# Define our server configurations
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):
    """Connect to a single MCP server and load its tools"""
    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():
    # Connect to all servers
    connections = await asyncio.gather(
        *[connect_to_server(server) for server in servers]
    )
    
    try:
        # Collect all tools from all servers
        all_tools = []
        for connection in connections:
            all_tools.extend(connection["tools"])
        
        # Create the agent with all tools
        model = ChatOpenAI(model="gpt-4o")
        agent = create_react_agent(model, all_tools)
        
        # Run the agent with a complex query that might use multiple servers
        response = await agent.ainvoke({
            "messages": "Find the customers who've spent more than the average order value and calculate their total spend."
        })
        
        return response
    
    finally:
        # Clean up all connections
        for connection in connections:
            await connection["cleanup"]()

# Run the multi-server agent
if __name__ == "__main__":
    result = asyncio.run(run_multi_server_agent())
    print(result)

Penanganan Kesalahan dan Strategi Cadangan

Implementasi MCP yang kuat harus menyertakan penanganan kesalahan. Berikut adalah versi klien yang ditingkatkan yang menunjukkan penanganan kesalahan dan strategi cadangan:

# robust_client.py
async def run_agent_with_fallbacks():
    try:
        # Attempt primary connection
        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"Error using MCP tools: {e}")
                    # Fallback to direct model call without tools
                    return await model.ainvoke([
                        HumanMessage(content="what's (3 + 5) x 12?")
                    ])
    except Exception as connection_error:
        print(f"Connection error: {connection_error}")
        # Ultimate fallback
        return {"error": "Could not establish connection to MCP server"}

Pertimbangan Keamanan

Saat mengimplementasikan MCP dengan LangChain, pertimbangkan praktik terbaik keamanan ini:

Validasi Input: Selalu validasi input ke alat MCP untuk mencegah serangan injeksi
Izin Alat: Implementasikan izin terperinci untuk setiap alat
Pembatasan Laju: Terapkan batasan laju untuk mencegah penyalahgunaan alat
Autentikasi: Implementasikan autentikasi yang tepat antara klien dan server

Berikut adalah contoh penerapan izin alat:

from mcp.server.fastmcp import FastMCP, Permission

mcp = FastMCP("SecureTools")

# Define permission levels
READ_PERMISSION = Permission("read", "Can read data")
WRITE_PERMISSION = Permission("write", "Can modify data")

@mcp.tool(permissions=[READ_PERMISSION])
def get_data(key: str) -> str:
    """Get data by key (requires read permission)"""
    # Implementation...
    return f"Data for {key}"

@mcp.tool(permissions=[WRITE_PERMISSION])
def update_data(key: str, value: str) -> bool:
    """Update data (requires write permission)"""
    # Implementation...
    return True

Optimasi Kinerja

Untuk penerapan produksi, pertimbangkan optimasi kinerja ini:

Penggabungan Koneksi: Gunakan kembali koneksi MCP daripada membuat yang baru untuk setiap permintaan
Pemrosesan Batch: Kelompokkan beberapa panggilan alat jika memungkinkan
Pemrosesan Asinkron: Gunakan asyncio untuk menangani beberapa permintaan secara bersamaan

Contoh penggabungan koneksi:

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):
        # Pre-create some connections
        for _ in range(3):  # Start with 3 connections
            await self._create_connection()
    
    async def _create_connection(self):
        if self.current_connections >= self.max_connections:
            raise Exception("Maximum connections reached")
        
        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)

Menguji Implementasi MCP

Pengujian menyeluruh sangat penting untuk implementasi MCP yang andal. Berikut adalah pendekatan pengujian menggunakan 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

Kesimpulan

Model Context Protocol menyediakan kerangka kerja yang kuat untuk menghubungkan aplikasi LangChain dengan alat dan sumber data eksternal. Dengan menstandarisasi koneksi ini, MCP memungkinkan pengembang untuk membuat agen AI canggih yang dapat berinteraksi dengan lingkungannya secara mulus.

Kombinasi kemampuan agen LangChain dengan konektivitas MCP menciptakan fondasi untuk membangun aplikasi sadar konteks yang benar-benar kuat. Seiring ekosistem MCP terus berkembang, kita dapat mengharapkan lebih banyak server dan alat bawaan untuk muncul, yang semakin menyederhanakan proses pengembangan.

Tutorial ini telah membahas konsep dasar dan detail implementasi penggunaan MCP dengan LangChain, dari pengaturan dasar hingga pola lanjutan seperti penggabungan koneksi dan penanganan kesalahan. Dengan mengikuti praktik ini, Anda dapat membuat aplikasi siap produksi yang kuat yang memanfaatkan yang terbaik dari kedua teknologi.

Untuk eksplorasi lebih lanjut, pertimbangkan untuk menyelidiki ekosistem server MCP yang berkembang yang tersedia di GitHub, atau kontribusikan server Anda sendiri ke komunitas. Masa depan agen AI terletak pada kemampuan mereka untuk secara efektif memanfaatkan alat dan data eksternal, dan MCP adalah langkah signifikan untuk mewujudkan visi itu.